Orleans: Await tasks in DedicatedAsynchAgent.Run()

[csyszf]

Source description in gitter:

I'm discovering the internal of orleans' runtime, and I found that DedicatedAsynchAgent.Run()'s return type is void so its implements all can not be awaited. So is it safe to make DedicatedAsynchAgent.Run() awaitable(return a Task)? I want to try System.IO.Pipelines and System.Threading.Channels in Orleans which are designed to reactivate by async/wait pattern

There are some ambiguities in my description, so I'll try to explain it more clearly

Intention

I'm learning about orleans' internal and wanna try out some new .NET Core features like System.IO.Pipelines and System.Threading.Channels in orleans.
Pipelines for sockets communication, and Channels for internal message queue.
(There are already some great work by @JanEggers in https://github.com/dotnet/orleans/issues/3617#issuecomment-352587339)

What's The Problem

The readers/writers in Pipelines/Channels are reactivate, which means when pipes/channels are empty, they will suspend readers and awaken them after data arrived in without blocking the current thread. This is implemented by Tasks and async/await.
I want to achieve these "awaitable" behavior in IncomingMessageAgent.Run()(For polling/posting message from/to channels) but couldn't because DedicatedAsynchAgent.Run() returns void.

At first I simply change DedicatedAsynchAgent.Run() to return a task and enqueue async delegate to thread executor instead:

internal abstract class DedicatedAsynchAgent : AsynchAgent
{
    public override void OnStart()
    {
        executor.QueueWorkItem(async _ => await Run());
    }
    protected abstract Task Run();
}

Or start a new task like what IncomingMessageAcceptor.ProcessAccept does:

Task.Factory.StartNew(async () =>
{
    await DoWork();
});

BUT:
a) What DedicatedAsynchAgent.Run()s do is to enqueue a delegate to a specific Thread, so it seems there's no meaning for "un-blocking" this thread.
b) await in async delegate will lead to the remaining code to enter a .Net thread pool thread.

What I Want To Know

• Why DedicatedAsynchAgent uses the custom ThreadPoolExecutor rather than just start a job/task in the .NET thread pool.
• If it is safe to execute DedicatedAsynchAgent logic out of its ThreadPoolExecutor's thread.

PS.
There may be some misunderstanding about ThreadPoolExecutor/AsynchAgent/Async-Await ..., I'm glad to be corrected.
It's so much fun to read Orleans' source code, what you people did is fantastic

[sergeybykov]

Thank you, @csyszf! :-)

[JanEggers]

maybe I can investigate further now that channels are released, I already did some experiments but postponded them because there is no (official) client side story for System.IO.Pipelines + Sockets yet. server works great with kestrel

[JanEggers]

have a look at https://github.com/JanEggers/orleans/tree/channelsV2

current master is at 78K with the changes we are at 85K msgs/sec

maybe I should port IncomingMessageAccepter to kestrel next....

[JanEggers]

with kestrel im kind of stuck the orleanshostbuilder is not compatible with webhostbuilder. and there are a bunch of options related classes that are mirrored from microsoft.extensions.options

maybe I should wait for #4702 arg always have to wait for the good stuff :(

[ReubenBond]

Regarding options, @JanEggers, that's tracked by #4703. The reason that those classes are duplicated is that they existed in Orleans first and were then upstreamed to Microsoft.Extensions.Options. We can make the switch, but it would be considered a breaking change. The question is which Orleans version those changes should appear in.

Regarding Hosting, the ASP.NET folks aren't going to move to the generic host until 3.0, AFAIK. We can add official support earlier than that. Note that @galvesribeiro has been using generic host with Orleans already, by adding some adapters.

System.IO.Pipelines/Bedrock is definitely on our radar. Currently there is no 'official' support for creating a client connection on Bedrock, so we would likely have to pursue a route similar to what @mgravell did with Pipelines.Sockets.Unofficial, at least until support is available upstream.

I have a work item on my plate to rework our messaging layer to make it more pluggable and I will consider Pipelines/Channels/etc when designing that.

Thank you, @csyszf 😊!

Regarding the reasoning behind dedicated threads for AsyncAgent, that's largely due to the choice to use BlockingCollection<T> as the underlying queue data structure. In future we ought to be able to move away from that and potentially share threads using the .NET ThreadPool instead. Actually, @dVakulen's scheduler includes a fork of parts of the .NET thread pool for its work stealing queues. We can likely #if some of those classes out on .NET Core 2.1 and use https://github.com/dotnet/corefx/issues/12442 instead. That will require us to multi-target, but we will likely want to start doing that at some point anyway so that we can take advantage of .NET Core advancements.

Regarding your questions:

Why DedicatedAsynchAgent uses the custom ThreadPoolExecutor rather than just start a job/task in the .NET thread pool.

Is answered by @dVakulen here: https://github.com/dotnet/orleans/pull/3792#issuecomment-390277172

If it is safe to execute DedicatedAsynchAgent logic out of its ThreadPoolExecutor's thread.

Start assumes it's being executed from a 'dedicated' thread (eg, it logs the thread id). I don't believe it has a negative effect if you call from the wrong thread, though. What kind of logic do you want to execute from another thread?

For now, perhaps a single-threaded TaskScheduler could be used to enable experimentation?

[csyszf]

@JanEggers 🎉 I'll keep tracking and experiment it.

@ReubenBond Thanks for the guidance, found many valuable commits/discuss in those Issues/PRs, now most of my puzzles have been solved.

What kind of logic do you want to execute from another thread?

Take messages from queue and enqueue and handle them (IncomingMessageAgent)
Actually there's no need to to run the work "from another thread" at all, it's basically about some misunderstanding of mine 😂.

https://github.com/dotnet/orleans/pull/2060 https://github.com/dotnet/orleans/pull/3792 https://github.com/dotnet/orleans/issues/3498 help a lot

BTW, System.IO.Pipelines seems really helpful, not only Pipelines it self, it can also introduce many useful feature - ReadOnlySequence<byte> as message buffer, span-based serialization, etc.
I've tried some of them, and it reduces considerable complexity of current logic.

As for now, this issue can be closed, thanks again.

[ReubenBond]

I've been experimenting with span-based serialization. See my repo here: https://github.com/ReubenBond/Hagar. My intention is to bring the benefits to Orleans, so I've been putting thought into how that should look.

Very interested to see your experiments.