What is the best way to wait in a network packet using the new async C # function

Question

I recently played with the new Async CTP, and I came across a situation where I am not sure how to proceed.

In my current code base, I use the concepts of "jobs" and "job manager". Tasks exist solely to process the original message, send a response, and wait for a response.

I already have existing code based on synchronous sockets, where the network stream waits for data to arrive, and then passes it along with the event handler and, ultimately, to the task manager.

The task manager looks for what work will process the message, and passes it.

So the scenario is this:

• The task manager receives a new message and starts the task.
• The job begins, processes the message, and sends a response message.
• At this point, the task will wait for a response to the answer.

Here is an example of pseudo code:

class MyJob : Job { public override void RunJob( IPacketMsg packet ) { // handle packet var myReply = new Packet(); SendReply( myReply ); await GetResponse(); } } 

But I'm not quite sure how to proceed in step 3. The task manager will receive an answer, and then transfer it to the current work. But I'm not sure how to make the work wait for an answer.

I considered creating an expected task that simply blocks WaitHandle, but is this a better solution?

Are there any other things that I could do in this case?

Edit With regard to Async CTP, what happens when the user interface is not in use. I read Eric Lippert on the Async blog, but I don't think it ever touched on how everything works in the background without a user interface thread (does it distract the background worker or ...?)

Answer 1

• The task manager receives a new message and starts the task.
• The job begins, processes the message, and sends a response message.
• At this point, the task will wait for a response to the answer.

First, I have to mention that Async CTP handles asynchronous operations very well, but there are not many asynchronous events. You might want to consider an Rx-based approach. But let him continue to work with Async CTP for the moment.

You have two main options for creating tasks:

• With a delegate. for example, Task.Factory.StartNew will run a delegate in the thread pool. Custom task factories and schedulers provide you with more options for task delegates (for example, delegations must be specified in the STA stream).
• Without a delegate. for example, TaskFactory.FromAsync wraps an existing pair of Begin / End methods, TaskEx.FromResult returns a "future constant", and TaskCompletionSource can be used to explicitly control Task (both FromAsync and FromResult use TCS internally).

If the processing of the task is related to the CPU, it makes sense to transfer it to Task.Factory.StartNew . I'm going to assume that job processing is CPU related.

Pseudocode of work manager:

 // Responds to a new message by starting a new job on the thread pool. private void RespondToNewMessage(IPacketMsg message) { IJob job = ..; Task.Factory.StartNew(job.RunJob(message)); } // Holds tasks waiting for a response. private ConcurrentDictionary<int, TaskCompletionSource<IResponse>> responseTasks = ..; // Asynchronously gets a response for the specified reply. public Task<IResponse> GetResponseForReplyAsync(int replyId) { var tcs = new TaskCompletionSource<IResponse>(); responseTasks.Add(replyId, tcs); return tcs.Task; } // Responds to a new response by completing and removing its task. private void RespondToResponse(IResponse response) { var tcs = responseTasks[response.ReplyId]; responseTasks.Remove(response.ReplyId); tcs.TrySetComplete(response); } 

The idea is that the job manager also manages a list of outstanding answers. To make this happen, I introduced a simple int response identifier, which the task manager can use to determine the answer with which answer.

Now tasks can work as follows:

 public override void RunJob(IPacketMsg packet) { // handle packet var myReply = new Packet(); var response = jobManager.GetResponseForReplyAsync(myReply.ReplyId); SendReply(myReply); await response; } 

There are a couple of tricky things as we host jobs in a thread pool thread:

• GetResponseForReplyAsync must be called (register the task) before sending a response and then await ed later. This is done in order to avoid a situation where a response can be sent and a response is received before we have the opportunity to register for it.
• RespondToResponse deletes the registration of the task before its completion, just in case the completion of the task will send another response with the same identifier.

If the tasks are short enough so that they do not need to be placed in the thread pool thread, then the solution can be simplified.

Answer 2

As for Async CTP, what happens when the user interface is not in use. I re-read Eric Lippert Asink’s blog, but I don’t think that the question of how everything works in the background without a user interface thread has ever been raised (does it work in the background or ...?)

await will return to its synchronization context. In the user interface process, this is a user interface message loop. In ASP.NET, this is an ASP.NET thread pool. In other situations (console applications and Win32 services), there is no context, so extensions are queued in ThreadPool . This is usually an undesirable behavior, so I wrote an AsyncContext class that can be used in such situations.

BackgroundWorker not used. In a server-side script such as yours, it is not unusual to have a background thread at all.

Answer 3

You simply hook up the rest of the event handler using the wait pattern as follows:

  public async void RunJob(IPacketMsg msg) { // Do Stuff var response = await GetResponse(); // response is "string", not "Task<string>" // Do More Stuff } public Task<string> GetResponse() { return Task.Factory.StartNew(() => { _networkThingy.WaitForDataAvailable(); return _networkThingy.ResponseString; }); } 

When your response task completes, the rest of the method executes in the current synchronization context. Until then, however, the method is executed (therefore, any code after waiting does not run until the task completes in GetResponse)