标签: executioncontext

我很难理解ExecutionContext背后的机制.

从我在线阅读的内容来看,安全性(线程主体),文化等上下文相关的项目应该跨越执行工作单元的边界内的异步线程.

我遇到了非常混乱和潜在危险的错误.我注意到我的线程的CurrentPrincipal在异步执行中丢失了.

以下是ASP.NET Web API方案的示例:

首先,让我们设置一个简单的Web API配置,其中包含两个委托处理程序,用于测试目的.

他们所做的就是写出调试信息并传递请求/响应,除了第一个"DummyHandler",它设置线程的主体以及要在整个上下文中共享的一段数据(请求的相关ID).

public static class WebApiConfig
{
    public static void Register(HttpConfiguration config)
    {
        config.MessageHandlers.Add(new DummyHandler());
        config.MessageHandlers.Add(new AnotherDummyHandler());

        config.Routes.MapHttpRoute(
            name: "DefaultApi",
            routeTemplate: "api/{controller}/{id}",
            defaults: new { id = RouteParameter.Optional }
        );
    }
}

public class DummyHandler : DelegatingHandler
{
    protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
    {
        CallContext.LogicalSetData("rcid", request.GetCorrelationId());
        Thread.CurrentPrincipal = new ClaimsPrincipal(new ClaimsPrincipal(new ClaimsIdentity(new[]{ new Claim("http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name", "dgdev") }, "myauthisthebest")));

        Debug.WriteLine("Dummy Handler Thread: {0}", Thread.CurrentThread.ManagedThreadId);
        Debug.WriteLine("User: {0}", (Object)Thread.CurrentPrincipal.Identity.Name);
        Debug.WriteLine("RCID: {0}", CallContext.LogicalGetData("rcid"));

        return base.SendAsync(request, cancellationToken) …

我正在尝试理解 C++ 中的异步模型。我正在研究 4 个旨在处理异步 I/O 的库：

1. liburing（C版本、C++版本）：提供io_uring的接口。C++ 版本使用协程。
2. libunifex：实现 C++ 发送者/接收者异步编程模型
3. Boost.Asio：用于异步网络和低级 I/O 编程的库
4. CppCoro：提供了大量通用原语来使用协程

这些库有大量的重叠。

执行上下文和io_uring

异步 I/O 的未来可能会在 Linux 机器上使用 io_uring。liburing 的存在是为了提供 io_uring 和io_service 的接口。然而，libunifex 还提供了io_uring_context。它们都明确使用io_uring，但用法与 Boost.Asio 的io_context和 CppCoro 的io_service类似。

liburing io_service、libunifex、io_uring_contextBoost.Asioio_context和 CppCoroio_service可以一起使用吗？如果我的代码包含所有这四个库，那么每个库都会有 1 个执行上下文吗？

功能：打开文件

本节包含 CppCoro 如何异步打开文件的示例。我相信这个模板类将在 Boost.Asio 中用于异步文件访问。libunifex 有一个 io_uring_text.cpp文档，其中包括对文件的异步写入。显然 liburing 是为了异步写入文件而存在的。

我应该只使用 io_uring 特定库进行文件访问吗？ …

我有一个WCF服务,它的Thread.CurrentPrincipal设置在ServiceConfiguration.ClaimsAuthorizationManager.

当我像这样异步实现服务时:

    public IAsyncResult BeginMethod1(AsyncCallback callback, object state)
    {
        // Audit log call (uses Thread.CurrentPrincipal)

        var task = Task<int>.Factory.StartNew(this.WorkerFunction, state);

        return task.ContinueWith(res => callback(task));
    }

    public string EndMethod1(IAsyncResult ar)
    {
        // Audit log result (uses Thread.CurrentPrincipal)

        return ar.AsyncState as string;
    }

    private int WorkerFunction(object state)
    {
        // perform work
    }

我发现Thread.CurrentPrincipal在Begin-method和WorkerFunction中设置为正确的ClaimsPrincipal,但是在End-method中它被设置为GenericPrincipal.

我知道我可以为服务启用ASP.NET兼容性并使用HttpContext.Current.User在所有方法中具有正确主体的服务,但我宁愿不这样做.

有没有办法强制Thread.CurrentPrincipal到正确的ClaimsPrincipal而不打开ASP.NET兼容性？

在使用ConfigureAwait(false)的引用库中使用Thread.CurrentPrincipal的声明会造成任何问题,或者ExecutionContext的逻辑调用上下文的流动会在那里照顾我吗？(到目前为止,我的阅读和测试表明它会).

示例WebAPI控制器操作:

[CustomAuthorizeThatSetsCurrentUsersClaimsToThreadCurrentContextAndHttpContextCurrentUser]
public async Task<Order> Get(int orderId)
{
    return await _orderBusinessLogicLibrary.LoadAsync(orderId); // defaults to .ConfigureAwait(true)
}

来自外部引用库的示例加载函数:

[ClaimsPrincipalPermission(
    SecurityAction.Demand,
    Operation="Read",
    Resource="Orders")]
[ClaimsPrincipalPermission(
    SecurityAction.Demand,
    Operation="Read",
    Resource="OrderItems")]
public async Task<Order> Load(int orderId)
{
    var order = await _repository.LoadOrderAsync(orderId).ConfigureAwait(false);

    // here's the key line.. assuming this lower-level function is also imposing
    // security constraints in the same way this method does, would
    // Thread.CurrentPrincipal still be correct inside the function below?
    order.Items = await _repository.LoadOrderItemsAsync(orderId).ConfigureAwait(false);
    return order;
}

此外,答案不能是"那么不要使用ConfigureAwait(false)!".这可能会导致其他问题,例如死锁(请勿在异步代码上阻止).

正如Stephen Toub在这篇文章中解释的那样，当您向ActionBlock提交消息时，可以在调用ActionBlock.Post之前先执行ExecutionContext.Capture，将包含消息和ExecutionContext的DTO传递到块中，然后在消息处理委托内部使用ExecutionContext.Run在捕获的上下文上运行委托：

public sealed class ContextFlowProcessor<T> {
    private struct MessageState {
        internal ExecutionContext Context;
        internal T Value;
    }

    private readonly ITargetBlock<MessageState> m_block;

    public ContextFlowProcessor(Action<T> action) {
        m_block = new ActionBlock<MessageState>(ms =>
        {
            if (ms.Context != null)
                using (ms.Context) ExecutionContext.Run(ms.Context, s => action((T)s), ms.Value);
            else 
                action(ms.Value);
        });
    }

    public bool Post(T item) {
        var ec = ExecutionContext.Capture();
        var rv = m_block.Post(new MessageState { Context = ec, Value = item });
        if (!rv) ec.Dispose();
        return rv;
    }

    public void Done() …

我试图发现ExecutionContext如何在.NET Framework 4.0及更高版本中实际工作.文档说当使用Thread.Start和大多数线程池操作时,托管原则,同步,区域设置和用户上下文都流向新线程.但我不能在实践中看到这一点.

这是一个简单的控制台应用程序,用于测试启动新线程时同步上下文和托管原则是否流动...

    static void Main(string[] args)
    {
        SynchronizationContext.SetSynchronizationContext(new SynchronizationContext());
        Thread.CurrentPrincipal = new GenericPrincipal(new GenericIdentity("One"), null);

        Thread t1 = new Thread(new ThreadStart(ThreadRun));
        t1.Start();
        t1.Join();

        SynchronizationContext.SetSynchronizationContext(new SynchronizationContext());
        Thread.CurrentPrincipal = new GenericPrincipal(new GenericIdentity("Two"), null);

        AsyncFlowControl aFC = ExecutionContext.SuppressFlow();
        Thread t2 = new Thread(new ThreadStart(ThreadRun));
        t2.Start();
        t2.Join();
        aFC.Undo();

        Console.Read();
    }

    static void ThreadRun()
    {
        Console.WriteLine("ThreadRun Id={0} Context={1} Principle={2}", 
            Thread.CurrentThread.ManagedThreadId, 
            (SynchronizationContext.Current != null), 
            Thread.CurrentPrincipal.Identity.Name);
    }

结果是......

    ThreadRun Id=11 Context=False Principle=One
    ThreadRun Id=12 Context=False Principle=Two

因此,同步上下文永远不会流动,并且即使在您指定它时,托管原则也始终会流动.基本上文档是完全错误的.那么是否描述了ExecutionContext在现实中的作用以及它为何有用？

当我做一个future,或者应用类似的方法onSuccess和map,我能为他们指定的ExecutionContext.

例如,

val f = future {
  // code
} executionContext

f.map(someFunction)(executionContext)

f onSuccess {
  // code
} executionContext

但是,如果我使用了对未来的理解,我该如何为该yield部分指定ExecutionContext ？

for {
  f <- future1
  g <- future2
} yield {
  // code to be executed after future1 onSuccess and future2 onSuccess
  // What ExecutionContext runs this code?
} // (executionContext) here does not work

而且,如果没有指定,ExecutionContext在yield中运行代码是什么？

好.感谢答案,我找到了一些东西.
如果我没有定义或导入隐式 ExecutionContext(如Implicits.global),则for-comprehension不会编译.这意味着,for-comprehension使用隐式ExecutionContext.

那么,如何在没有隐式ExecutionContext的情况下使用for-comprehension,即如何指定？

我正在关注akka-in-action教程,在第2章中,有一个类(https://github.com/RayRoestenburg/akka-in-action/blob/master/chapter2/src/main/scala/com/ goticks/RestInterface.scala):

trait RestApi extends HttpService with ActorLogging { actor: Actor =>
  import context.dispatcher
  import com.goticks.TicketProtocol._
  ...

在import context.dispatcher从未使用过,但它与评论定义:

  /**
   * Returns the dispatcher (MessageDispatcher) that is used for this Actor.
   * Importing this member will place an implicit ExecutionContext in scope.
   */
  implicit def dispatcher: ExecutionContextExecutor

但是,IntelliJ会将导入标记为"未使用",并在"优化导入"时将其删除,从而导致错误value pipeTo is not a member of scala.concurrent.Future[Any].

有没有办法告诉IntelliJ这个导入不打算"使用",而只是简单地提供一个上下文？

或者是否应该更新教程以不使用这种"未使用的导入"？

我刚刚开始使用Akka Stream,我正在尝试解决一些问题:

目前,我的流量,我使用mapAsync()与我的休息服务集成,为推荐这里.

我一直在想,mapAsync()应该使用什么执行上下文？它应该是我的ActorSystem的调度员吗？全球？两种情况都有不明显的后果吗？

我意识到这可能是一个愚蠢的问题,但我以前从未处理过Akka,在涉及Futures的任何scala应用程序中,我只使用过全局执行上下文.

我正在使用Scala 2.10期货创建一个异步库.库的构造函数采用一系列实现特定特征的用户定义对象,然后库类上的方法将一些数据逐个发送到用户定义的对象中.我希望用户ExecutionContext在设置主实例时提供异步操作,然后根据需要将该上下文传递给用户定义的对象.简化(伪？)代码:

case class Response(thing: String)

class LibraryObject(stack: Seq[Processor])(implicit context: ExecutionContext) {
  def entryPoint(data: String): Future[Response] = {
    val response = Future(Response(""))
    stack.foldLeft(response) { (resp, proc) => proc.process(data, resp) }
  }
}

trait Processor {
  def process(data: String, resp: Future[Response]): Future[Response]
}

可能会使用这样的东西:

class ThingProcessor extends Processor {
  override def process(data: String, response: Future[Response]) = {
    response map { _.copy(thing = "THE THING") }
  }
}

class PassThroughProcessor extends Processor {
  override def process(request: Request, response: Future[Response]) = …