Как написать асинхронный конвеер на c#?

Question

[Василий]

Доброго времени суток.
Есть такой алгоритм:
1. На вход поступает массив с файлами(streams) изображений.
2. Мы эти изображения конвертируем в другой формат(ресурсоёмко),
3. А затем кладём на удалённый сервер(может быть длительный отклик).
4. Ссылки от того куда положили возвращаем в виде коллекции тому кто нас вызвал.

Количество входных файлов нам известно сразу.
Как написать алгоритм конвеера что бы мы не ждали пока все изображения сконвертируются. Надо что бы при первом готовом мы бы сразу пытались его отправлять на удалённый сервер?

Пробовал с BlockingCollection но кажется я его не правильно использую.

UPDATED
Благодаря совету mindtester был написан такой код https://dotnetfiddle.net/fHWZ9n
Он вроде работает как надо, если есть возможность посмотрите и напишите свои замечания в комментариях к этому посту. (Кому удобно гитхаб https://gist.github.com/milovidov983/e305890bfd8fb... )

Answer 1

[#]

ну так IEnumerable<> и yield вам в помощь. само собой и async/await
как раз эти четыре понятия и придназначены, для упрощения написания кода, над массивами даннх, растянутыми во времени, и не известной длинны. с их помощью, большая часть кода, выглядит так, будто все входные данные предоставлены сразу, и результат вы выдаете как будто тоже сразу

ps для практики полезны примеры из MSDN (где они есть) либо на https://metanit.com/sharp/tutorial/ придется поискать. в любом случае, однозначно придется понять обсласти применения и возможности этих понятий. а тогда уже точно сложится картина, как они лягут на ваш конкретный случай

pps

Ссылки от того куда положили возвращаем в виде коллекции тому кто нас вызвал.

вот это место не очень вписывается в концепцию конвеера на IEnumerable/yield/async/await
дело в том что результирующие ссылки, вы получаете "изредка", и явно в отрыве от момента получения исходного изображения
тут несколько путей:
- именно коллекцию ссылок (json?) формировать по окончанию обработоки
- возвращать в первоисточник ссылки по одной, и формировать коллекцию там, а привязку можно делать по имени/тегу/коду/номеру изображения.. (кадра?)
- можно и асинхронно, растянуто во времени досылать json, или любую выбранную коллекцию дотнета, если на приемнике дотнет, и вы так же правите исходниками
- как вариант, если известно количество кадров изначально, можно уведомлять первоисточнк о начале пересылки хоть массива заданной размерности, в любом понятном ему формате, если конечно он так же способенн к асинхронной обработке

ppps

Пробовал с BlockingCollection

повторюсь, на сколько я уловил вашу задачу, кроме IEnumerable<>, в подобном конвеере в принципе ни че больше не нужно. ну или разве что как вы там результат решите формировать. но и для результата, даже для накопителя, из дотнетовских конструкций, того же IEnumerable<> за уши, а из него, если кончно это требуется, можно и json формировать

Comments

[Василий]

Удивительно, на шарпе пишу несколько лет и ни разу не сталкивался.
У меня к Вам просьба посмотрите на пример это то что вы имели ввиду?
https://dotnetfiddle.net/fHWZ9n

[#]

Василий, нет. но судя по выхлопу вы где то близки к цели.
в чем критика:
- таски по моему не нужны, пока вне надо делать web api
- вы классы плодите поповоду и без, имхо
- а еще они у вас жутко зависимые друг от друга. вы же сам сказали

Как написать асинхронный конвеер на c#?

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

извините, свой код не преложу, по состоянию здоровья. возможно позднее

[#]

жуткий недосып и трудно соссредоточиться, сори... но задача затнтересовала. набросал свй прототип. пока без async/await, чисто на IEnumerable/yield.. и к конфузу, был удивлен, чо простейшая схема не работает ))

если метод не асинхронный, и получает, пусть даже не пустой, но уже IEnumerable, то он сразу завершается. мне надо самому копнуть тему глубже ))
раз ве что если очень интересно, то бесполезный, неработающий код, тут (upd! совсем уже с трудо соображаю, эксперементировал, так и кину, сейчас исправил на то что было замыслом) https://dotnetfiddle.net/wUbbLB
видимо сами методы все равно должна быть асинхронными.. и надо выспаться )) .. (сори за личные подробности)

ps сама студия все компилирует без ошибок. надо уже "завтра".. возможно онлайн IDE дает хороший намек.. но.. не в силах )) теперь в онлайне тоже нет ошибок, но не работает, надо допиливать ))

[Василий]

1) Вы упоминали что если это не Web Api то Task не надо, тык вот у меня это именно web api, и в ответе я должен буду вернуть json с набором ссылок на CDN.

2) Ваш код рабочий, вы просто не запустили процесс перечисления: https://dotnetfiddle.net/zOsDSe

3) Моя основная проблема состоит в том что какой-либо процесс(конвертирование или закачка на CDN) может обгонять другой, и что бы нам не ждать завершения процессы закачки/конвертирования должны выполнятся параллельно(смотрите лог моего варианта там процессы не ждут друг друга), в вашем же случае насколько я понимаю всё выполняется последовательно и в случае когда процесс конвертирования сильно обгоняет по скорости процесс закачки, он всё ровно будет ждать когда закончится закачка что бы запустить следующий файл на конвертацию, или я ошибаюсь?

[#]

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

о последовательности и параллельности - ну та для того мы и встявляли логирорование каждой фазы (это я не буде преукрашивать - просто содрал у вас )))

раз у ваз web api, то таски ожидаемы, но на сколько я помню учебные примеры (простите не ас, тлько пробы) они та больше в возвращаемых типах

а вот такой вот упрощенный пример, как мне показалось, избавит вас от избыточной привязанности городидть классы и архитектурные замки

но повторюсь - смотрите на выхлоп - тут ни кто ни кого не ждет ;))

и, полагаю, именно упрощеные примеры, лучше высвечивают суть ))

удачи в решении вашей задачи. надеюсь помог.. хотя бы расширить кругозор ))

[#]

Василий, только проснулся, точно ждут засранцы, ну дальше просто. либо async/await допилить, лабо PLINQ )))

тогда точно ждать не будут

Answer 2

[Владислав]

Посмотри DataFlow. Может это будет overkill, но для работы с изображением подходит на ура.

BigInteger result = 0;
            var inputBlock = new BufferBlock<(string,string)>(); // Блок для входных данных

            var options = new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 200 // указываем сколько потоков будут обрабатывать наши данные
            };

            var callBalanceBlock = new TransformBlock<(string,string), BigInteger>(async x =>
            {
                BigInteger wallet = await function.CallAsync<BigInteger>(x.Item1);
                BigInteger eth = await Current.Eth.GetBalance.SendRequestAsync(x.Item1);
                TechWallets.TechEURGWallets.Add(new TechWallet
                {
                    Address = x.Item1, 
                    TokenValue = Web3.Convert.FromWei(wallet,2), 
                    EthValue = Web3.Convert.FromWei(eth, UnitConversion.EthUnit.Ether),
                    PrivateKey = x.Item2
                });
                return wallet;
            }, options); // Это главный испольнительный блок. Получает информация в виде (string,string) и отдает в формате BigInteger.

            var calc = new ActionBlock<BigInteger>(x => result += x, options); // Получает данные после работы TransformBlock, чтобы подсчитать число

            inputBlock.LinkTo(callBalanceBlock); // Прокидываем линк(соединение), что после BufferBlock, должен отрабатывать TransformBlock
            inputBlock.Completion.ContinueWith(task => callBalanceBlock.Complete());

            callBalanceBlock.LinkTo(calc); // Тоже самое, но для ActionBlock
            callBalanceBlock.Completion.ContinueWith(task => calc.Complete());

            foreach (var i in TechWallets.ListEURGTechWallets)
                inputBlock.Post(i); // Заполняем наш BufferBlock входными значениями

            inputBlock.Complete(); // Запускаем работу наших блоков

            try
            {
                calc.Completion.Wait(); // Ждем завершения работы последнего блока
            }
            catch (Exception e)
            {
                Console.WriteLine(e.Message);
            }

Код станет проще намного и быстрее.
Ты можешь построить свой конвеер по типу: BufferBlock(Массив входных данных) -> TransformBlock(Тут обрабатываешь входные картинки) -> ActionBlock(Тут заливаешь картинку на сервер)

Comments

[Василий]

А что делать если конвертация происходит быстрее чем закачка допустим на 1 закачку можно успеть сконвретировать 2 файла? В этом случае у нас будет простой по времени.

[Владислав]

Василий, нет, не будет, потому что мы блоки распаралелили на 200 потоков. Если бы мы не указывали опцию на паралель, то он бы выполнялся последовательно, а так это намного ускоряет процесс.
У меня на работе был код, который обращался к серверу. Он делал 600 однотипных запросов последовательно. Выполнение его занимало около 2 минут. Когда я с помощью данного кода сделать запросы, то время работы кода сократилось до 2 секунд

[Василий]

Владислав, спасибо попробую dataflow