CODEWITHSUNDEEP
c#asynchronousakka.net
Gigi
Gigi

Reputation: 29421

Akka .NET PipeTo() with ContinueWith()

The Top 7 Mistakes Newbies Make with Akka.NET explains why using async/await within an actor is often a bad idea:

[...] we see end users develop lots of nested async / await operations inside an individual message handler. There’s a cost overlooked by most users to doing this: the actor can’t process any other messages between each await operation because those awaits are still part of the “1 message at a time” guarantee for the original message!

Yet in Unit 3 Lesson 4 of the Petabridge Akka .NET Bootcamp, this example is considered okay:

// asynchronously download the image and pipe the results to ourself
_httpClient.GetAsync(imageUrl).ContinueWith(httpRequest =>
{
    var response = httpRequest.Result;

    // successful img download
    if (response.StatusCode == HttpStatusCode.OK)
    {
        var contentStream = response.Content.ReadAsStreamAsync();
        try
        {
            contentStream.Wait(TimeSpan.FromSeconds(1));
            return new ImageDownloadResult(image,
                response.StatusCode, contentStream.Result);
        }
        catch //timeout exceptions!
        {
            return new ImageDownloadResult(image, HttpStatusCode.PartialContent);
        }
    }

    return new ImageDownloadResult(image, response.StatusCode);
},
  TaskContinuationOptions.ExecuteSynchronously)
 .PipeTo(Self);

The way I understand this, the actor will not be able to process any other messages until GetAsync() AND ContinueWith() are done, which is exactly the problem PipeTo() was trying to avoid.

Am I missing something here?

Upvotes: 2

Views: 3677

Answers (2)

RSinohara
RSinohara

Reputation: 732

TaskContinuationOptions.ExecuteSynchronously means the ContinueWith will run sinchronously with the code that transitions the task to complete (TaskContinuationOptions).

The code in the example will fire the task, setup ContinueWith() and PipeTo(), and return. The actor is free to receive new messages, and when the task completes, PipeTo() will send it the resulting message.

As Gigi mentioned, if the Task were to be awaited in the actor, then it would be blocked.

It might help to think that ContinueWith() also returns a task, and that's what PipeTo() operates on.

This is the PipeTo extention in Akka.net Github:

    public static Task PipeTo<T>(this Task<T> taskToPipe, ICanTell recipient, IActorRef sender = null, Func<T, object> success = null, Func<Exception, object> failure = null)
    {
        sender = sender ?? ActorRefs.NoSender;
        return taskToPipe.ContinueWith(tresult =>
        {
            if (tresult.IsCanceled || tresult.IsFaulted)
                recipient.Tell(failure != null
                    ? failure(tresult.Exception)
                    : new Status.Failure(tresult.Exception), sender);
            else if (tresult.IsCompleted)
                recipient.Tell(success != null
                    ? success(tresult.Result)
                    : tresult.Result, sender);
        }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
    }

Upvotes: 3

Gigi
Gigi

Reputation: 29421

What I think is happening is that since GetAsync() is not awaited, the whole thing fires off asynchronously and does not block execution.

While all this is happening, the actor is free to process additional messages.

Upvotes: 2

Related Questions