24

Вполне простая задача - есть N упорядоченных запросов, которые надо выполнить асинхронно с ограничением на количество одновременно выполняемых, и потом достать их из них так же N упорядоченных ответов.

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

Лобовое решение выглядит очевидно.

public async Task<Tout[]> RunSemaphoreAsync<Tin, Tout>(IEnumerable<Tin> items, Func<Tin, Task<Tout>> func, int degree)
{
    using SemaphoreSlim semaphore = new(degree);
    return await Task.WhenAll(items.Select(async item => {
        await semaphore.WaitAsync();
        try
        {
            return await func(item);
        }
        finally
        {
            semaphore.Release();
        }
    }));
}

Пример использования:

Пусть это будет рабочая задача, которая что-то принимает, что-то делает и что-то отдает назад.

public async Task<int> RunJobAsync(int n)
{
    await Task.Yield(); 
    return n + 1;
}

И вот такой запуск для примера.

IEnumerable<int> numbers = Enumerable.Range(0, 100);
int[] result = await RunSemaphoreAsync(numbers, RunJobAsync, Environment.ProcessorCount * 2);
Console.WriteLine(string.Join(",", result));

Всё работает работает как часики, быстро и ожидаемо.


С тем что выше все отлично, но мне однажды, когда я в очередной раз читал про всякие реализации Producer/Consumer паттерна, пришла идея использовать воркеры вместо семафора. Собственно, почему нет.

Получился вот такой метод.

public async Task<Tout[]> RunWorkersAsync<Tin, Tout>(IEnumerable<Tin> items, Func<Tin, Task<Tout>> func, int degree)
{
    List<Task<Tout>> tasks = new();
    using (var source = items.GetEnumerator())
    {
        Task[] jobs = new Task[degree];
        for (int i = 0; i < degree; i++)
        {
            jobs[i] = ((Func<Task>)(async () =>
            {
                while (true)
                {
                    Task<Tout> task;
                    lock (source)
                    {
                        if (source.MoveNext())
                        {
                            task = func(source.Current);
                            tasks.Add(task);
                        }
                        else
                            break;
                    }
                    await task;
                }
            }))();
        }
        await Task.WhenAll(jobs);
    }
    return tasks.Select(t => t.Result).ToArray();
}

Работает точно так же красиво как и первый кандидат. Тогда что же лучше?


И решил я померять оверхед.
Я не мастер писать бенчмарки, но кого когда это останавливало? :)

class Program
{
    static void Main(string[] args)
    {
        var result = BenchmarkRunner.Run<MyBenchmarks>();
        Console.ReadKey();
    }
}

[MemoryDiagnoser]
public class MyBenchmarks
{
    private readonly List<int> numbers = Enumerable.Range(0, 2000).ToList();
    private readonly int degree = Environment.ProcessorCount * 2;

    [Benchmark]
    public Task SemaphoreTest()
    {
        return RunSemaphoreAsync(numbers, RunJobAsync, degree);
    }

    [Benchmark]
    public Task WorkersTest()
    {
        return RunWorkersAsync(numbers, RunJobAsync, degree);
    }

    public async Task<int> RunJobAsync(int n)
    {
        await Task.Yield(); 
        return n + 1;
    }

    public async Task<Tout[]> RunSemaphoreAsync<Tin, Tout>(IEnumerable<Tin> items, Func<Tin, Task<Tout>> func, int degree)
    {
        using SemaphoreSlim semaphore = new(degree);
        return await Task.WhenAll(items.Select(async item => {
            await semaphore.WaitAsync();
            try
            {
                return await func(item);
            }
            finally
            {
                semaphore.Release();
            }
        }));
    }

    public async Task<Tout[]> RunWorkersAsync<Tin, Tout>(IEnumerable<Tin> items, Func<Tin, Task<Tout>> func, int degree)
    {
        List<Task<Tout>> tasks = new();
        using (var source = items.GetEnumerator())
        {
            Task[] jobs = new Task[degree];
            for (int i = 0; i < degree; i++)
            {
                jobs[i] = ((Func<Task>)(async () =>
                {
                    while (true)
                    {
                        Task<Tout> task;
                        lock (source)
                        {
                            if (source.MoveNext())
                            {
                                task = func(source.Current);
                                tasks.Add(task);
                            }
                            else
                                break;
                        }
                        await task;
                    }
                }))();
            }
            await Task.WhenAll(jobs);
        }
        return tasks.Select(t => t.Result).ToArray();
    }
}

И тут получаю интересный результат

BenchmarkDotNet=v0.13.0, OS=Windows 10.0.19043.1081 (21H1/May2021Update)
Intel Core i7-4700HQ CPU 2.40GHz (Haswell), 1 CPU, 8 logical and 4 physical cores
.NET SDK=5.0.301
  [Host]     : .NET 5.0.7 (5.0.721.25508), X64 RyuJIT
  DefaultJob : .NET 5.0.7 (5.0.721.25508), X64 RyuJIT
Method Mean Error StdDev Gen 0 Gen 1 Gen 2 Allocated
SemaphoreTest 1,780.2 us 4.23 us 3.95 us 140.6250 41.0156 - 519 KB
WorkersTest 943.2 us 18.37 us 26.92 us 74.2188 19.5313 - 262 KB

Глаза отказываются в это верить. Почему метод с семафором просто вот так вот слился, а быть может кривой тест или кривая реализация? Рассудите пожалуйста.

P.S. Я же не сразу побежал на SO, а первым делом поискал жалобы на медленный SemaphoreSlim...и не нашел, но нашел вот что: https://github.com/dotnet/runtime/pull/55262. Другими словами, светофорчик немного полечат в .NET 6.

1
  • 1
    Комментарии не предназначены для расширенной дискуссии; разговор перемещён в чат.
    – Grundy
    10 июл 2021 в 18:12

2 ответа 2

8
+500

Код для RunWorkersAsync и для RunSemaphoreAsync по разному распределяет работу между потоками.

В RunWorkersAsync в первоначальном потоке отрабатывает только первая операция цикла while каждого worker-a, после этого перебор происходит в несколько потоков, т.к. все после вызова await отрабатывает на потоках из пула. Больше никакой завязки на какой-то общий поток в коде нет.

В RunSemaphoreAsync же в первоначальном потоке для каждого элемента отрабатывает

items.Select(async item => {
        semaphore.WaitAsync(); // await и продолжение - на потоке из пула

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

Это можно решить принудительным yield прямо в начале цикла:

public async Task<Tout[]> RunSemaphoreAsync<Tin, Tout>(IEnumerable<Tin> items, Func<Tin, Task<Tout>> func, int degree)
{
    using SemaphoreSlim semaphore = new(degree);
    return await Task.WhenAll(items.Select(async item =>
    {
        await Task.Yield();
        await semaphore.WaitAsync();
        try
        {
            return await func(item);
        }
        finally
        {
            semaphore.Release();
        }
    }));
}

Результаты станут чуть лучше, но сам боттлнек в виде одного потока все равно не исчезнет.

4

Я решил попробовать применить каналы - Channel. Код получился такой:

public async Task<Tout[]> RunChannelAsync<Tin, Tout>(IEnumerable<Tin> items, Func<Tin, Task<Tout>> func, int degree)
{
    var channel = Channel.CreateBounded<Task<Tout>>(degree);

    var writer = channel.Writer;
    var reader = channel.Reader;

    Task.Run(async () =>
    {
        foreach (Tin item in items)
            await writer.WriteAsync(func(item));

        writer.Complete();
    });

    var tasks = new List<Tout>();

    await foreach (Task<Tout> item in reader.ReadAllAsync())
        tasks.Add(await item);

    return tasks.ToArray();
}

Количество одновременно выполняемых задач задаётся в методе CreateBounded.

Производительность получилась средняя.
Причина фактически та же, что и с семафором: задачи создаёт один поток, как объяснил PashaPash.

Method Mean Error StdDev Gen 0 Gen 1 Gen 2 Allocated
SemaphoreTest 1,148.9 us 15.50 us 14.50 us 156.2500 48.8281 - 527 KB
SemaphoreYieldTest 1,154.4 us 22.51 us 40.59 us 169.9219 68.3594 - 711 KB
WorkersTest 763.3 us 5.61 us 5.25 us 87.8906 27.3438 - 260 KB
ChannelTest 991.8 us 6.64 us 6.21 us 125.0000 - - 246 KB

Разве что мусора генерируется мало. Даже до первого поколения не доходит.

Ещё я бы хотел отметить, что вариант на семафорах с await Task.Yield(); на моей машине работает как и без него, в пределах погрешности и даже хуже.


Основная причина, по которой я решил написать, такова. Тестируемые методы принимают IEnumerable - это хорошо, значения могут генерироваться или получаться откуда-либо на лету.
А вот результат возвращается сразу весь целиком, в виде массива. Таким образом, отчасти теряется асинхронность, т. к. получается блокировка на создании/получении этого массива.

7
  • Очень странно, что нет разницы между yield и семафором. У меня разница в 1,5-2 раза в лучщую сторону в пользу yield.
    – aepot
    18 июл 2021 в 15:29
  • Если что я бенчмарки гонял на чистом x64. Без каких-либо контекстов синхронизации.
    – aepot
    18 июл 2021 в 15:39
  • @aepot - Win10, x64, .net-5, console 18 июл 2021 в 15:42
  • прогнал у себя. Каналы в 2 раза медленнее воркеров, то есть еще хуже семафора получились. Быть может у меня меньше физических ядер (4), чем у вас, а оно вроде-как зависит. У меня degree = 16.
    – aepot
    18 июл 2021 в 16:29
  • Последнюю проблему можно решить, используя IAsyncEnumerable, возвращая пачками работу, не придется дожидаться всей её работы целиком. На моей машине вышло чуть медленнее чем воркеры
    – Vein
    18 июл 2021 в 16:41

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

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