c# – 强制一个任务继续当前的线程？

我正在建立一个用于.NET的AKKA框架的端口(现在不要这么认真,现在是周末黑客的Actor部分)

我在“未来”的支持中遇到一些问题.
在Java / Scala Akka中,期待与Await电话同步等待.
很像.NET Task.Wait()

我的目标是支持真正的异步等待.
它现在可以工作,但是在我当前的解决方案中,延续是在错误的线程上执行的.

这是将消息传递给我的一个包含等待未来块的演员的结果.
正如你所看到的,演员总是在同一个线程上执行,而等待块在随机线程池线程上执行.

actor thread: 6
await thread 10
actor thread: 6
await thread 12
actor thread: 6
actor thread: 6
await thread 13
...

演员使用DataFlow BufferBlock< Message>获取消息.
或者,我使用RX缓冲区来订阅消息.
它是这样配置的：

var messages = new BufferBlock<Message>()
{
        BoundedCapacity = 100,TaskScheduler = TaskScheduler.Default,};
messages.AsObservable().Subscribe(this);

到现在为止还挺好.

但是,当我等待未来的结果.
像这样：

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"));
     ...

我知道这是异步等待的正常行为,但我真的必须确保只有一个线程可以访问我的演员.

我不希望将来能够同步运行,我想象正常一样运行异步,但是我希望继续在与消息处理器/ actor相同的线程上运行.

我的未来支持代码如下所示：

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;
}

任何想法我可以做什么来强制继续运行在启动上述代码的同一个线程？

解决方法

I’m making a port of the AKKA framework for .NET

甜.尽管从未触及过Java / Scala / Akka,但我还是在CodeMash ’13上进行了Akka演讲.我看到很多潜在的.NET库/框架.我希望最终能够提供微软is working on something similar(currently in a limited preview).

我怀疑尽可能多地停留在Dataflow / Rx世界中;当异步操作(每个操作具有单个启动和单个结果)时,异步最好,而Dataflow和Rx在流和订阅(具有单个起始和多个结果)时可以更好地工作.所以我的第一个反应是将缓冲区链接到具有特定调度器的ActionBlock,或者使用ObserveOn将Rx通知移动到特定的调度程序,而不是尝试在异步端执行.当然,我不太熟悉Akka API设计,所以用一粒盐就可以了.

无论如何,我的async intro描述了调度等待延续的唯一两个可靠选项：SynchronizationContext.Current和TaskScheduler.Current.如果您的Akka端口更多是一个框架(您的代码执行托管,最终用户代码始终由代码执行),那么一个SynchronizationContext可能是有意义的.如果您的端口更多是一个库(最终用户代码根据需要托管并调用您的代码),那么TaskScheduler会更有意义.

自定义SynchronizationContext没有很多示例,因为这很罕见.我的AsyncEx library中有一个 AsyncContextThread type,它定义了该线程的SynchronizationContext和TaskScheduler.有几个自定义TaskSchedulers的例子,例如Parallel Extensions Extras和“current thread” scheduler.

c# – 强制一个任务继续当前的线程？

解决方法

相关文章