2

Ранее я уже писал ответ на подобный вопрос: Как гарантированно выполнить код в 1 потоке после await c#.net

Но возникла потребность многократно использовать прелести однопоточной синхронизации в приложении безо всяких блокировок используя асинхронный код. Известно же, что lock и прочие примитивы синхронизации в асинхронном коде использовать не получится. Ну разве что кроме семафора, но семафор - не совсем примитив.

Чтобы воспроизвести проблему, даже коллекция не нужна, можно запустить вот такой код:

static async Task Main(string[] args)
{
    await AddAsync(1000);
    Console.WriteLine(_counter);
}

static int _counter;

static async Task AddAsync(int count)
{
    Task[] tasks = new Task[count];
    for (int i = 0; i < count; i++)
        tasks[i] = IncrementAsync();
    await Task.WhenAll(tasks);
}

static async Task IncrementAsync()
{
    await Task.Delay(1);
    _counter++;
}

И получить например вот такой вывод в консоль

969

А должно быть 1000. Это значит, что код не является потокобезопасным. Если запустить этот код из UI потока в WPF или Winforms, он отработает корректно в 100% случаев и выдаст всегда 1000. В предложенном решении по ссылке сам запуск контекста выглядит очень громоздским, и на каждый запуск нужно либо создавать новый поток, либо отдавать контексту текущий.

Хочется найти более простое решение, чтобы можно было в разных частях многопоточного кода перенаправлять вызовы в какой-то свой контекст, работающий в конкретном потоке.

2
  • 1
    Interlocked.Increment :)
    – user177221
    23 фев 2022 в 17:06
  • 1
    @PashaPash спасибо, но это всего-лишь абстрактный пример того что даже обычный асинхронный код иногда надо синхронизировать :)
    – aepot
    23 фев 2022 в 17:10

1 ответ 1

2

Решение очень похоже на прошлое, немного отличается. Но использовать гораздо проще, чем предыдущее. И можно иметь один контекст и переиспользовать его из разных углов приложения.

Суть решения в создании контекста, который берет себе 1 поток из пула и в нем работает до тех пор, пока его не задиспозить. При этом обращаясь в экземпляру контекста можно из любого места кода перенаправить вызов в тот самый поток.

Закидывать в контекст можно любой код, в том числе, синхронный, но назначение класса - именно дать возможнось запускать асинхронные методы так, чтобы awaitы внутри метода не меняли поток после завершения асинхронного вызова, точно так же как это происходит в UI потоке десктопного приложения. За это отвечают 2 перегрузки метода SendAsync.

То есть появляется возможность использовать обычные коллекции и прочие прелести однопоточки без блокировок в коде. Это решение даже для UI приложений может оказаться полезным, когда хочется работать как в UI потоке с async/await, но при этом сам UI не нагружать.

public class SingleThreadedSynchronizationContext : SynchronizationContext, IDisposable
{
    private readonly BlockingCollection<(SendOrPostCallback, object)> _queue = new();
    private readonly Task _task;

    public SingleThreadedSynchronizationContext()
    {
        _task = Task.Run(RunContext);
    }

    public override void Post(SendOrPostCallback d, object state)
    {
        _queue.Add((d, state));
    }

    public override void Send(SendOrPostCallback d, object state)
    {
        if (Current == this)
            d(state);
        else
        {
            using ManualResetEventSlim mre = new();
            Post(s =>
            {
                try
                {
                    d(s);
                }
                finally
                {
                    mre.Set();
                }
            }, state);
            mre.Wait();
        }
    }

    public Task SendAsync(Func<Task> func)
    {
        if (Current == this)
            return func();
        Task task = null;
        Send(_ => task = func(), null);
        return task;
    }

    public Task<T> SendAsync<T>(Func<Task<T>> func)
    {
        if (Current == this)
            return func();
        Task<T> task = null;
        Send(_ => task = func(), null);
        return task;
    }

    private void RunContext()
    {
        SetSynchronizationContext(this);
        foreach ((SendOrPostCallback d, object state) in _queue.GetConsumingEnumerable())
        {
            d(state);
        }
    }

    public void Dispose()
    {
        _queue.CompleteAdding();
        _task.Wait();
    }
}

Проверяю.

static async Task Main(string[] args)
{
    using SingleThreadedSynchronizationContext context = new();
    await context.SendAsync(() => AddAsync(1000));
    Console.WriteLine(_counter);
}

Вывод в консоль

1000

Стоит только обратить внимание, что этот контекст не обрабатывает исключения. Вообще сам по себе контекст синхронизации не предназначен для обработчки исключений, их на самом деле обрабатывает сам await. Поэтому я не стал нагружать код обработками. Но стоит быть внимательным. Если в Post заслать вручную делегат, который сгенерит исключение, контекст упадет и не сможет больше обрабатывать вызовы. Если же вручную Send и Post не вызывать, то дорабатывать ничего не нужно.


Показываю, как обезопасить контекст от выброса исключений полностью. Для этого надо очень просто доработать пару методов.

public override void Send(SendOrPostCallback d, object state)
{
    if (Current == this)
        d(state);
    else
    {
        using ManualResetEventSlim mre = new();
        ExceptionDispatchInfo edi = null;
        Post(s =>
        {
            try
            {
                d(s);
            }
            catch (Exception ex)
            {
                edi = ExceptionDispatchInfo.Capture(ex);
            }
            mre.Set();
        }, state);
        mre.Wait();
        edi?.Throw();
    }
}

private void RunContext()
{
    SetSynchronizationContext(this);
    foreach ((SendOrPostCallback d, object state) in _queue.GetConsumingEnumerable())
    {
        try
        {
            d(state);
        }
        catch { }
    }
}

Повторюсь, такая доработка нужна только если закидывать в Send и Post свои делегаты извне контекста, которые могут выбрасывать исключения. Если использовать только SendAsync, то данная доработка не требуется.

2
  • lock вообще можно использовать в асинхронных методах. Нельзя оборачивать await в lock, но сам lock в синхронных кусках кода - пожалуйста. Впрочем, и такая реализация SingleThreadedSynchronizationContext может иногда пригодиться, тем более, что она проще в использовании будет, чем lock. Но раз уж взялся писать вопрос/ответ, имеет наверное смысл тогда и примеры с обработкой исключений привести для будущих читателей.
    – Иван
    23 фев 2022 в 19:01
  • @John всё верно. Здесь главный фокус, зачем я это написал, в том, что у меня где-то есть пошаренная между потоками коллекция, точнее нескольно, и я интенсивно с ними работаю. При этом Producer/Consumer выглядел страшновато, еще нужна была работа с индексаторами и гарантия, что пока кусок синхронного кода выполняется, индексы элементов не поедут. Локи спасали, но их было очень много. Снапшоты тоже надоели, и я решил пойти другим путем, и выкинул просто тонны кода, отвечающего за синхронизацию. Теперь всё чисто и понятно.
    – aepot
    23 фев 2022 в 19:09

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.