c# - 如何为每个异步操作创建一个新上下文并以线程安全的方式使用它来存储一些数据

Question

我正在研究诸如上下文绑定(bind)缓存之类的东西，并且有点卡在线程安全上...

假设我有以下代码:

public class AsynLocalContextualCacheAccessor : IContextualCacheAccessor
{
    private static readonly AsyncLocal<CacheScopesManager> _rCacheContextManager = new AsyncLocal<CacheScopesManager>();

    public AsynLocalContextualCacheAccessor()
    {
    }

    public CacheScope Current
    { 
        get
        {
            if (_rCacheContextManager.Value == null)
                _rCacheContextManager.Value = new CacheScopesManager();

            return _rCacheContextManager.Value.Current;
        }
    } 
}

public class CacheScopesManager
{
    private static readonly AsyncLocal<ImmutableStack<CacheScope>> _scopesStack = new AsyncLocal<ImmutableStack<CacheScope>>(OnValueChanged);

    public CacheScopesManager()
    {
        CacheScope contextualCache = _NewScope();

        _scopesStack.Value = ImmutableStack.Create<CacheScope>();
        _scopesStack.Value = _scopesStack.Value.Push(contextualCache);
    }

    public CacheScope Current
    {
        get
        {
            if (_scopesStack.Value.IsEmpty)
                return null;

            CacheScope current = _scopesStack.Value.Peek();
            if (current.IsDisposed)
            {
                _scopesStack.Value = _scopesStack.Value.Pop();
                return Current;
            }

            // Create a new scope if we entered the new physical thread in the same logical thread
            // in order to update async local stack and automatically have a new scope per every logically new operation
            int currentThreadId = Thread.CurrentThread.ManagedThreadId;
            if (currentThreadId != current.AcquiredByThread)
            {
                current = _NewScope();
                _scopesStack.Value = _scopesStack.Value.Push(current);
            }

            return current;
        }
    }

    private static void OnValueChanged(AsyncLocalValueChangedArgs<ImmutableStack<CacheScope>> args)
    {
        // Manual is not interesting to us.
        if (!args.ThreadContextChanged)
            return;

        ImmutableStack<CacheScope> currentStack = args.CurrentValue;
        ImmutableStack<CacheScope> previousStack = args.PreviousValue;

        int threadId = Thread.CurrentThread.ManagedThreadId; 

        int threadIdCurrent = args.CurrentValue?.Peek().AcquiredByThread ?? -1;
        int threadIdPrevious = args.PreviousValue?.Peek().AcquiredByThread ?? -1;

        // Be sure in disposing of the scope
        // This situation means a comeback of the previous execution context, in case if in the previous scope Current was used.
        if (currentStack != null && previousStack != null
            && currentStack.Count() > previousStack.Count())
            currentStack.Peek().Dispose();
    }
}

我正在努力满足下一个测试:

    [TestMethod]
    [TestCategory(TestCategoryCatalogs.UnitTest)]
    public async Task AsyncLocalCacheManagerAccessor_request_that_processed_by_more_than_by_one_thread_is_threadsafe()
    {
        IContextualCacheAccessor asyncLocalAccessor = new AsynLocalContextualCacheAccessor();
        Task requestAsyncFlow = Task.Run(async () =>
        {
            string key1 = "key1";
            string value1 = "value1";

            string key2 = "key2";
            string value2 = "value2";

            CacheScope scope1 = asyncLocalAccessor.Current;

            string initialKey = "k";
            object initialVal = new object();

            scope1.Put(initialKey, initialVal);
            scope1.TryGet(initialKey, out object result1).Should().BeTrue();
            result1.Should().Be(initialVal);

            var parallel1 = Task.Run(async () =>
            {
                await Task.Delay(5);
                var cache = asyncLocalAccessor.Current;
                cache.TryGet(initialKey, out object result2).Should().BeTrue();
                result2.Should().Be(initialVal);

                cache.Put(key1, value1);
                await Task.Delay(10);

                cache.Items.Count.Should().Be(1);
                cache.TryGet(key1, out string result11).Should().BeTrue();
                result11.Should().Be(value1);
            });

            var parallel2 = Task.Run(async () =>
            {
                await Task.Delay(2);
                var cache = asyncLocalAccessor.Current;

                cache.StartScope();

                cache.TryGet(initialKey, out object result3).Should().BeTrue();
                result3.Should().Be(initialVal);

                cache.Put(key2, value2);
                await Task.Delay(15);

                cache.Items.Count.Should().Be(1);
                cache.TryGet(key2, out string result21).Should().BeTrue();
                result21.Should().Be(value2);
            });

            await Task.WhenAll(parallel1, parallel2);

            // Here is an implicit dependency from Synchronization Context, and in most cases
            // the next code will be handled by a new thread, that will cause a creation of a new scope,
            // as well as for any other await inside any async operation,  which is quite bad:( 

            asyncLocalAccessor.Current.Items.Count.Should().Be(1);
            asyncLocalAccessor.Current.TryGet(initialKey, out object result4).Should().BeTrue();
            result4.Should().Be(initialVal);
        });

        await requestAsyncFlow;

        asyncLocalAccessor.Current.Items.Count.Should().Be(0);
    }

实际上这个测试是绿色的，但是有一个(或多个)问题。所以，我想要实现的是为每个新的异步操作创建一个范围堆栈(如果访问了当前范围)，并且当这个操作完成时，我需要成功返回到前一个堆栈。我已经根据当前线程 ID 完成了此操作(因为我没有找到任何其他方法来自动执行此操作，但我不喜欢我的解决方案)，但如果继续执行异步操作不在初始线程(来自当前的隐式依赖 SynchronizationContext )，但在任何其他情况下，这都会导致创建一个新的范围，这对我来说非常糟糕。

如果有人能建议如何正确地做到这一点，我会很高兴，非常感谢! :)

UPD 1. 更新代码以添加 static对于每个 AsyncLocal字段，因为每个 AsyncLocal 的值从 ExecutionContext.GetLocalValue() 获得这是静态的，所以非静态的 AsyncLocal 只是一个冗余的内存压力。

UPD 2. 谢谢@weichch 的回答，因为评论可能很大，我只是直接在问题中添加了附加信息。所以，在我的案例逻辑中， AsyncLocal封装的东西，以及我的代码的客户端可以做什么 - 它只调用 Current在 IContextualCacheAccessor , 这将获得 AsyncLocal<CacheScopesManager> 下的对象的实例, AsyncLocal此处仅用于拥有 CacheScopesManager 的一个实例每个逻辑请求并在此请求中共享它，类似于 IoC-Container 范围的生命周期，但此类对象的生命周期是从创建对象到创建该对象的异步流程结束定义的。或者让我们考虑一下我们拥有 IHttpContext 的 ASP NET Core , IHttpContext似乎不是一成不变的，但仍用作 AsyncLocal通过 IHttpContextAccessor ，不是吗？类似这种方式 CacheScopesManager被设计。

所以，如果是客户端代码，获取当前 CacheScope , 只能调用 Current在 IContextualCacheAccessor , 那么在 AsyncLocal 的情况下 IContextualCacheAccessor的实现调用堆栈将落入下一个代码:

public CacheScope Current
{
    get
    {
        if (_scopesStack.Value.IsEmpty)
            return null;

        CacheScope current = _scopesStack.Value.Peek();
        if (current.IsDisposed)
        {
            _scopesStack.Value = _scopesStack.Value.Pop();
            return Current;
        }

        // Create a new scope if we entered the new physical thread in the same logical thread
        // in order to update async local stack and automatically have a new scope per every logically new operation
        int currentThreadId = Thread.CurrentThread.ManagedThreadId;
        if (currentThreadId != current.AcquiredByThread)
        {
            current = _NewScope();
            _scopesStack.Value = _scopesStack.Value.Push(current);
        }

        return current;
    }
}

如果另一个线程决定使用 Current ，这将导致新范围的创建，并且由于 ImmutableStack<CacheScope>是“AsyncLocal”，我们正在保存先前异步流的任何更改的堆栈，这意味着当我们返回它时，堆栈将没有任何损坏(当然，如果没有使用黑客)。所有这些都是为了使范围堆栈线程安全，而不是真正的“AsyncLocal”。所以，你的代码

async Task Method1()
{
    Cache.Push(new CacheScope { Value = "Method1" });

    await Task.WhenAll(Method2(), Method3());

    Cache.Pop();
}

async Task Method2()
{
    await Task.Delay(10);

    var scope = Cache.CurrentStack.Peek();
    scope.Value = "Method2";

    Console.WriteLine($"Method2 - {scope.Value}");
}

async Task Method3()
{
    await Task.Delay(10);

    var scope = Cache.CurrentStack.Peek();

    Console.WriteLine($"Method3 - {scope.Value}");
}

如果使用我的访问器，不会导致异步流中的突变，这将反射(reflect)在另一个流中(并且在线程切换之前将数据添加到前一个异步流的范围内——对我来说很好)。但是有一个问题，其实是这个 CacheScope的目的是有一些存储跨越逻辑请求并缓存一些数据，这些数据的范围为 CacheScope一旦作用域结束，就会从可引用的内存中弹出。而且我想尽量减少此类范围的创建，这意味着如果代码是按顺序执行的，则不应有任何理由创建新范围，即使某些异步操作的继续发生在另一个线程上，因为逻辑上代码仍然是'顺序的”，并且可以按照这样的“顺序”代码共享相同的范围。如果我在某个地方错了，请纠正我。

但是您的回答和解释确实很有用，并且肯定会保护其他人免于犯错。此外，它帮助我理解了斯蒂芬的意思:

If you do go down this route, I recommend writing lots and lots of unit tests.

我的英语很差，我认为“路线”的意思是“链接到文章”，现在明白在这种情况下它是相当“方式”。

UPD 3. 添加了 CacheScope的一些代码以获得更好的画面。

public class CacheScope : IDisposableExtended
{
    private ICacheScopesManager _scopeManager;
    private CacheScope _parentScope;

    private Dictionary<string, object> _storage = new Dictionary<string, object>();


    internal CacheScope(Guid id, int boundThreadId, ICacheScopesManager scopeManager, 
        CacheScope parentScope)
    {
        _scopeManager = scopeManager.ThrowIfArgumentIsNull(nameof(scopeManager));

        Id = id;
        AcquiredByThread = boundThreadId;

        _parentScope = parentScope;
    }

    public Guid Id { get; }

    public int AcquiredByThread { get; }

    public IReadOnlyCollection<object> Items => _storage?.Values;

    public bool IsDisposed { get; private set; } = false;

    public bool TryExpire<TItem>(string key, out TItem expiredItem)
    {
        _AssertInstanceIsDisposed();

        key.ThrowIfArgumentIsNull(nameof(key));

        expiredItem = default(TItem);

        try
        {
            expiredItem = (TItem)_storage[key];
        }
        catch (KeyNotFoundException)
        {
            // Even if item is present in parent scope it cannot be expired from inner scope.
            return false;
        }

        _storage.Remove(key);

        return true;
    }


    public TItem GetOrPut<TItem>(string key, Func<string, TItem> putFactory)
    {
        _AssertInstanceIsDisposed();

        key.ThrowIfArgumentIsNull(nameof(key));
        putFactory.ThrowIfArgumentIsNull(nameof(putFactory));

        TItem result;

        try
        {
            result = (TItem)_storage[key];
        }
        catch (KeyNotFoundException)
        {
            if (_parentScope != null && _parentScope.TryGet(key, out result))
                return result;

            result = putFactory(key);

            _storage.Add(key, result);
        }

        return result;
    }

    public void Put<TItem>(string key, TItem item)
    {
        _AssertInstanceIsDisposed();

        key.ThrowIfArgumentIsNull(nameof(key));

        _storage[key] = item;

        // We are not even thinking about to change the parent scope here,
        // because parent scope should be considered by current as immutable.
    }

    public bool TryGet<TItem>(string key, out TItem item)
    {
        _AssertInstanceIsDisposed();

        key.ThrowIfArgumentIsNull(nameof(key));

        item = default(TItem);

        try
        {
            item = (TItem)_storage[key];
        }
        catch (KeyNotFoundException)
        {
            return _parentScope != null && _parentScope.TryGet(key, out item);
        }

        return true;
    }

    public void Dispose()
    {
        if (IsDisposed)
            return;

        Dictionary<string, object> localStorage = Interlocked.Exchange(ref _storage, null);
        if (localStorage == null)
        {
            // that should never happen but Dispose in general is expected to be safe to call so... let's obey the rules
            return;
        }

        foreach (var item in localStorage.Values)
            if (item is IDisposable disposable)
                disposable.Dispose();

        _parentScope = null;
        _scopeManager = null;

        IsDisposed = true;
    }

    public CacheScope StartScope() => _scopeManager.CreateScope(this);
}

Answer 1

你的代码真的在争吵 AsyncLocal<T>作品。在 getter 中设置、尝试手动管理范围、为异步本地类型设置异步本地管理器以及使用更改处理程序的代码都是有问题的。

我相信这一切真的是为了尝试处理 CacheScope 的事实。不是一成不变的。解决这个问题的最好方法是制作 CacheScope一个适当的不可变对象(immutable对象)。然后其他一切都会或多或少自然地到位。

我发现写一个单独的 static 通常更容易。更“异步本地友好”的不可变对象(immutable对象)的 API。例如。:

public class ImplicitCache
{
  private static readonly AsyncLocal<ImmutableStack<(string, object)>> _asyncLocal = new AsyncLocal<ImmutableStack<(string, object)>>();

  private static ImmutableStack<(string, object)> CurrentStack
  {
    get => _asyncLocal.Current ?? ImmutableStack.Create<ImmutableDictionary<string, object>>();
    set => _asyncLocal.Current = value.IsEmpty ? null : value;
  }

  // Separate API:

  public static IDisposable Put(string key, object value)
  {
    if (key == null)
      throw new InvalidOperationException();
    CurrentStack = CurrentStack.Push((key, value));
    return new Disposable(() => CurrentStack = CurrentStack.Pop());
  }

  public static bool TryGet(string key, out object value)
  {
    var result = CurrentStack.Reverse().FirstOrDefault(x => x.Item1 == key);
    value = result.Item2;
    return result.Item1 != null;
  }
}

用法:

public async Task AsyncLocalCacheManagerAccessor_request_that_processed_by_more_than_by_one_thread_is_threadsafe()
{
  Task requestAsyncFlow = Task.Run(async () =>
  {
    string key1 = "key1";
    string value1 = "value1";

    string key2 = "key2";
    string value2 = "value2";

    string initialKey = "k";
    object initialVal = new object();

    using var dispose1 = ImplicitCache.Put(initialKey, initialVal);
    ImplicitCache.TryGet(initialKey, out object result1).Should().BeTrue();
    result1.Should().Be(initialVal);

    var parallel1 = Task.Run(async () =>
    {
      await Task.Delay(5);
      ImplicitCache.TryGet(initialKey, out object result2).Should().BeTrue();
      result2.Should().Be(initialVal);

      using var dispose2 = ImplicitCache.Put(key1, value1);
      await Task.Delay(10);

      ImplicitCache.TryGet(key1, out string result11).Should().BeTrue();
      result11.Should().Be(value1);
    });

    var parallel2 = Task.Run(async () =>
    {
      await Task.Delay(2);

      ImplicitCache.TryGet(initialKey, out object result3).Should().BeTrue();
      result3.Should().Be(initialVal);

      using var disose3 = ImplicitCache.Put(key2, value2);
      await Task.Delay(15);

      ImplicitCache.TryGet(key2, out string result21).Should().BeTrue();
      result21.Should().Be(value2);
    });

    await Task.WhenAll(parallel1, parallel2);

    ImplicitCache.TryGet(initialKey, out object result4).Should().BeTrue();
    result4.Should().Be(initialVal);
  });

  await requestAsyncFlow;

  ImplicitCache.TryGet(initialKey, out _).Should().BeFalse();
}