I'm making a port of the AKKA framework for .NET (don't take this too serious now, it is a weekend hack of the Actor part of it right now)
I'm having some problems with the "Future" support in it. In Java/Scala Akka, Futures are to be awaited synchronously with an Await call. Much like the .NET Task.Wait()
My goal is to support true async await for this. It works right now, but the continuation is executed on the wrong thread in my current solution.
This is the result when passing a message to one of my actors that contain an await block for a future. As you can see, the actor always executes on the same thread, while the await block executes on a random threadpool thread.
actor thread: 6
await thread 10
actor thread: 6
await thread 12
actor thread: 6
actor thread: 6
await thread 13
...
The actor gets a message using a DataFlow BufferBlock<Message>
Or rather, I use RX over the bufferblock to subscribe to messages.
It is configured like this:
var messages = new BufferBlock<Message>()
{
BoundedCapacity = 100,
TaskScheduler = TaskScheduler.Default,
};
messages.AsObservable().Subscribe(this);
So far so good.
However, when I await on a future result. like so:
protected override void OnReceive(IMessage message)
{
....
var result = await Ask(logger, m);
// This is not executed on the same thread as the above code
result.Match()
.With<SomeMessage>(t => {
Console.WriteLine("await thread {0}",
System.Threading.Thread.CurrentThread.GetHashCode());
})
.Default(_ => Console.WriteLine("Unknown message"));
...
I know this is normal behavior of async await, but I really must ensure that only one thread has access to my actor.
I don't want the future to run synchronously, I want to to run async just like normal, but I want the continuation to run on the same thread as the message processor/actor does.
My code for the future support looks like this:
public Task<IMessage> Ask(ActorRef actor, IMessage message)
{
TaskCompletionSource<IMessage> result =
new TaskCompletionSource<IMessage>();
var future = Context.ActorOf<FutureActor>(name : Guid.NewGuid().ToString());
// once this object gets a response,
// we set the result for the task completion source
var futureActorRef = new FutureActorRef(result);
future.Tell(new SetRespondTo(), futureActorRef);
actor.Tell(message, future);
return result.Task;
}
Any ideas what I can do to force the continuation to run on the same thread that started the above code?
while(true);. How could other code possibly execute on that thread? It is impossible to interrupt code (at arbitrary locations). If your thread looked like this:while(true) ExecuteNextItemForSyncContext();then someone could submit an item and inject code. Look into how sync contexts work. – Eviaevict