C# AsyncEnumerable, выполняющий/ожидающий несколько задач, никогда не завершается

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

Принцип работы AsyncEnumerable заключается в том, чтобы дождаться, пока производитель произведет значение, затем дождаться, пока потребитель использует это значение, а затем повторить. Он не поддерживает производитель и потребитель, работающие с разной скоростью, поэтому я говорю, что этот шаблон является последовательным. У него нет очереди произведенных предметов, только текущее значение. ReturnAsync не ждет, чтобы потребитель использовал значение, вместо этого вы должны ждать задачи, которую он возвращает, что дает вам сигнал о том, что она готова. Поэтому мы можем сделать вывод, что это не потокобезопасно.

Однако RetryComponent.RunTasks выполняет 10 задач параллельно, и этот код вызывает yield.ReturnAsync, не проверяя, не вызывал ли его кто-либо еще, и если да, то завершена ли эта задача. Поскольку класс Yield хранит только текущее значение, ваши 10 одновременных задач перезаписывают текущее значение, не дожидаясь готовности объекта Yield для нового значения, поэтому 9 задач теряются и никогда не ожидаются. Поскольку эти 9 задач никогда не ожидаются, методы никогда не завершатся, и Task.WhenAll никогда не вернется, как и другие методы во всем стеке вызовов.

Я создал вопрос на github предлагает им улучшить свою библиотеку, чтобы в таких случаях возникали исключения. Если они это реализуют, ваш блок catch выведет сообщение на консоль и повторно выдаст ошибку, поставив задачу в состояние сбоя, что позволит task.WhenAll завершиться и, следовательно, ваша программа не зависнет.

Вы можете использовать многопоточные API-интерфейсы синхронизации, чтобы гарантировать, что только одна задача за раз вызывает yield.ReturnAsync и ожидает возврата задачи. Или вы можете избежать использования шаблона с несколькими производителями, поскольку один производитель может легко быть перечислителем. В противном случае вам придется полностью переосмыслить, как вы хотите реализовать шаблон с несколькими производителями. Я предлагаю Поток данных TPL, который встроен в .NET Core и доступен в .NET Framework в виде пакета NuGet.

@zivkan абсолютно прав насчет последовательного шаблона производителя. Если вы хотите иметь параллельных производителей для одного потока, это все еще можно реализовать с помощью библиотеки AsyncEnumerable, но для этого потребуется дополнительный код.

Вот пример решения проблемы с параллельными производителями и потребителями (в данном случае только с одним потребителем):

        static void Main(string[] args)
        {
            var e = new AsyncEnumerable<int>(async yield =>
            {
                var threadCount = 10;
                var maxItemsOnQueue = 20;

                var queue = new ConcurrentQueue<int>();
                var consumerLimiter = new SemaphoreSlim(initialCount: 0, maxCount: maxItemsOnQueue + 1);
                var produceLimiter = new SemaphoreSlim(initialCount: maxItemsOnQueue, maxCount: maxItemsOnQueue);

                // Kick off producers
                var producerTasks = Enumerable.Range(0, threadCount)
                    .Select(index => Task.Run(() => ProduceAsync(queue, produceLimiter, consumerLimiter)));

                // When production ends, send a termination signal to the consumer.
                var endOfProductionTask = Task.WhenAll(producerTasks).ContinueWith(_ => consumerLimiter.Release());

                // The consumer loop.
                while (true)
                {
                    // Wait for an item to be produced, or a signal for the end of production.
                    await consumerLimiter.WaitAsync();

                    // Get a produced item.
                    if (queue.TryDequeue(out var item))
                    {
                        // Tell producers that they can keep producing.
                        produceLimiter.Release();
                        // Yield a produced item.
                        await yield.ReturnAsync(item);
                    }
                    else
                    {
                        // If the queue is empty, the production is over.
                        break;
                    }
                }
            });

            e.ForEachAsync((item, index) => Console.WriteLine($"{index + 1}: {item}")).Wait();
        }

        static async Task ProduceAsync(ConcurrentQueue<int> queue, SemaphoreSlim produceLimiter, SemaphoreSlim consumerLimiter)
        {
            var rnd = new Random();
            for (var i = 0; i < 10; i++)
            {
                await Task.Delay(10);
                var value = rnd.Next();

                await produceLimiter.WaitAsync(); // Wait for the next production slot
                queue.Enqueue(value); // Produce item on the queue
                consumerLimiter.Release(); // Notify the consumer
            }
        }