Как на C# реализовать асинхронную очередь задач у объекта?

Question

[More916]

Не могу понять в какую сторону копать для реализации асинхронности.
Пояснения приведу на примере:
Допустим имеется класс с методом, выполняющимся длительное время.

public class HardWorker : AbstractHardWorker
{
    public void DoSomething(int time)
    {
        Thread.Sleep(time);
    }
}

В основной программе создается пять таких объектов, и далее в цикле (например 50 раз), у каждого объекта вызывается данный метод:

static void Main(string[] args)
{
    HardWorker hardWorker1 = new HardWorker();
    ..........................................
    HardWorker hardWorker5 = new HardWorker();
    
    Random rand = new Random();
    
    for (int i=0; i < 50; i++)
    {
        hardWorker1.DoSomething(rand.Next(1,10000));
        .......................................
        hardWorker5.DoSomething(rand.Next(1,10000));
    }
}

В таком случае каждый hardWorker, прежде чем получить новый вызов метода, будет простаивать, пока предыдущий не завершит выполнение своего метода.
Я хочу сделать так, что бы у каждого hardWorker'a была своя очередь задач, которую он будет последовательно выполнять асинхронно от других hardWorker'ов. Тогда можно будет в основном потоке раздать hardWorker'ам задачи и ожидать пока они не завершат выполнение всех добавленных им задач . При этом для оконного приложения (WinForms например) элементы управления, да и все окно в целом не будет виснуть на время выполнения этих длительных задач.
Наследовать класс hardWorker от BackgroundWorker'а не возможно, так как он уже наследуется от другого класса.
Как решить данную задачу с использованием task'ов и модификаторов await/async?

Comments

[Denis Gaydak]

Наследовать класс hardWorker от BackgroundWorker'а не возможно ??? почему??

public class HardWorker : AbstractHardWorker,BackgroundWorker

уже запретили множественное наследование??

да и вообще все что вы описали, как раз BackgroundWorker и делает, без каких либо дополнительных ваших классов.
создайте парочку таких фоновых рабочих и дайте им задание.
https://msdn.microsoft.com/ru-ru/library/system.co...
вполне полный пример в документации.

[Griboks]

https://msdn.microsoft.com/ru-ru/library/7977ey2c(...

[akass]

Denis Gaydak, С каких пор в C# возможно множественное наследование?

[Denis Gaydak]

akass ну да сорян) множественное наследование только для интерфеисов. сорян))

Answer 1

[lam0x86]

class Program
    {
        static void Main(string[] args)
        {
            MainAsync(args).GetAwaiter().GetResult();

            Console.ReadKey();
        }

        static async Task MainAsync(string[] args)
        {
            HardWorker hardWorker1 = new HardWorker();
            HardWorker hardWorker2 = new HardWorker();
            HardWorker hardWorker3 = new HardWorker();
            HardWorker hardWorker4 = new HardWorker();
            HardWorker hardWorker5 = new HardWorker();

            Task t1 = Task.CompletedTask;
            Task t2 = Task.CompletedTask;
            Task t3 = Task.CompletedTask;
            Task t4 = Task.CompletedTask;
            Task t5 = Task.CompletedTask;

            Random rand = new Random();

            for (int i = 0; i < 50; i++)
            {
                t1 = t1.ContinueWith(task => hardWorker1.DoSomething(1, rand.Next(1, 10000)));
                t2 = t2.ContinueWith(task => hardWorker2.DoSomething(2, rand.Next(1, 10000)));
                t3 = t3.ContinueWith(task => hardWorker3.DoSomething(3, rand.Next(1, 10000)));
                t4 = t4.ContinueWith(task => hardWorker4.DoSomething(4, rand.Next(1, 10000)));
                t5 = t5.ContinueWith(task => hardWorker5.DoSomething(5, rand.Next(1, 10000)));
            }

            await Task.WhenAll(t1, t2, t3, t4, t5);
        }
    }

    public class HardWorker : AbstractHardWorker
    {
        public void DoSomething(int task, int time)
        {
            Console.Out.WriteLine("Start: " + task);
            Thread.Sleep(time);
            Console.Out.WriteLine("End: " + task);
        }
    }

    public interface AbstractHardWorker
    {
        void DoSomething(int task, int time);
    }

Comments

[lam0x86]

Но! Если мы говорим про асинхронность, то Thread.Sleep - ооочень плохо. По хорошему, DoSomething должен быть async, и в теле содержать вызовы асинхронных операций с префиксом await.

[More916]

Вроде то что нужно (По крайней мере выглядит так как хотелось). Буду пробовать.

По хорошему, DoSomething должен быть async, и в теле содержать вызовы асинхронных операций с префиксом await.

Вот как раз таки и хотелось, что бы очередь тасков была именно в HardWorker'е. Тогда получается, я у самого HardWorker'а добавлю поле t типа Tusk и добавление задачей будет происходить уже непосредственно в методе DoSomethingAsync:

class Program
    {
        static void Main(string[] args)
        {
            MainAsync(args).GetAwaiter().GetResult();

            Console.ReadKey();
        }

        static async Task MainAsync(string[] args)
        {
            HardWorker hardWorker1 = new HardWorker();
            HardWorker hardWorker2 = new HardWorker();
            HardWorker hardWorker3 = new HardWorker();
            HardWorker hardWorker4 = new HardWorker();
            HardWorker hardWorker5 = new HardWorker();

            Task t2 = Task.CompletedTask;
            Task t3 = Task.CompletedTask;
            Task t4 = Task.CompletedTask;
            Task t5 = Task.CompletedTask;

            Random rand = new Random();

            for (int i = 0; i < 50; i++)
            {
                hardWorker1.DoSomethingAsync(1, rand.Next(1, 10000)));
                hardWorker2.DoSomethingAsync(1, rand.Next(1, 10000)));
                hardWorker3.DoSomethingAsync(1, rand.Next(1, 10000)));
                hardWorker4.DoSomethingAsync(1, rand.Next(1, 10000)));
                hardWorker5.DoSomethingAsync(1, rand.Next(1, 10000)));
            }

            await Task.WhenAll(t1, t2, t3, t4, t5);
        }
    }

    public class HardWorker : AbstractHardWorker
    {
        public void DoSomething(int task, int time)
        {
            Console.Out.WriteLine("Start: " + task);
            Thread.Sleep(time);
            Console.Out.WriteLine("End: " + task);
        }
		
    }

    public abstract class AbstractHardWorker
    {
        Task t;
		
		public AbstractHardWorker()
		{
			t = Task.CompletedTask;
		}
		
        public abstract void DoSomething(int task, int time);
		
        public void DoSomethingAsync(int task, int time)
        {
            t = t.ContinueWith(task => this.DoSomething(int task, int time));
        }
    }

Правда это не совсем то, о чем вы писали в комментарии к своему ответу)

[lam0x86]

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

class Program
    {
        // Вычисление суммы байт из 5-ти разных файлов размером по 50000 байт

        static int Main(string[] args)
        {
            // пример, как можно использовать в синхронном коде
            var result = CalcByteSum().GetAwaiter().GetResult();

            Console.Out.WriteLine(result);

            return result;
        }

        static async Task<int> MainAsync(string[] args)
        {
            // а тут - как можно использовать в асинхронном коде
            var result = await CalcByteSum();

            Console.Out.WriteLine(result);

            return result;
        }

        static async Task<int> CalcByteSum()
        {
            using (var hardWorker1 = new SumFileBytesProcessor("file1"))
            using (var hardWorker2 = new SumFileBytesProcessor("file2"))
            using (var hardWorker3 = new SumFileBytesProcessor("file3"))
            using (var hardWorker4 = new SumFileBytesProcessor("file4"))
            using (var hardWorker5 = new SumFileBytesProcessor("file5"))
            {
                for (int i = 0; i < 50; i++)
                {
                    hardWorker1.ProcessFilePartition(i * 1000, 1000);
                    hardWorker2.ProcessFilePartition(i * 1000, 1000);
                    hardWorker3.ProcessFilePartition(i * 1000, 1000);
                    hardWorker4.ProcessFilePartition(i * 1000, 1000);
                    hardWorker5.ProcessFilePartition(i * 1000, 1000);
                }

                var r1 = await hardWorker1.GetResultAsync();
                var r2 = await hardWorker2.GetResultAsync();
                var r3 = await hardWorker3.GetResultAsync();
                var r4 = await hardWorker4.GetResultAsync();
                var r5 = await hardWorker5.GetResultAsync();

                return r1 + r2 + r3 + r4 + r5;
            }
        }
    }

    public class SumFileBytesProcessor : AbstractFileProcessor<int>
    {
        private readonly FileStream _fileStream;
        private readonly List<int> _interimResults = new List<int>();

        public SumFileBytesProcessor(string fileName)
        {
            _fileStream = new FileStream(fileName, FileMode.Open);
        }

        protected override async Task ProcessFilePartitionAsync(int offset, int length)
        {
            var buffer = new byte[length];
            await _fileStream.ReadAsync(buffer, offset, length);
            _interimResults.Add(GetSum(buffer));
        }

        protected override int GetResult()
        {
            return _interimResults.Sum();
        }

        private int GetSum(IEnumerable<byte> buffer)
        {
            return buffer.Sum(b => (int) b);
        }

        public override void Dispose()
        {
            _fileStream?.Dispose();
        }
    }

    public abstract class AbstractFileProcessor<TResult> : IDisposable
    {
        private Task _task;

        protected AbstractFileProcessor()
        {
            _task = Task.CompletedTask;
        }

        protected abstract Task ProcessFilePartitionAsync(int offset, int length);

        protected abstract TResult GetResult();

        public void ProcessFilePartition(int offset, int length)
        {
            _task = _task.ContinueWith(_ => ProcessFilePartitionAsync(offset, length));
        }

        public Task<TResult> GetResultAsync()
        {
            return _task.ContinueWith(_ => GetResult());
        }

        public abstract void Dispose();
    }

Answer 2

[cicatrix]

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace TasksSample
{
    public class Worker
    {
        private int id;
        public Worker(int id)
        {
            this.id = id;
        }
        public void DoHardWork()
        {
            Console.WriteLine($"Task {this.id} started.");
            Random rng = new Random();
            int delay = rng.Next(1000, 10000);
            Thread.Sleep(delay);
            Console.WriteLine($"Task {this.id} finished after {delay} ms.");
        } // DoHardWork
    } // class Worker

    class Program
    {
        static void Main(string[] args)
        {
            TaskFactory fact = new TaskFactory();
            int TaskCount = 10;
            List<Task> myTasks = new List<Task>();
            for (int id = 0; id < TaskCount; ++id)
            {
                Worker w = new Worker(id);
                myTasks.Add(fact.StartNew(() => w.DoHardWork()));
            }
            Task.WaitAll(myTasks.ToArray());
        } // void Main
    } // class Program
} // namespace TasksSample

Comments

[More916]

Приведенный мною пример был упрощенной моделью реальной задачи.
В принципе, это близко к решению, которое я пытался получить, но:
1. В методе DoHardWork необходим параметр, так как он выполняет определенные операции на основе переданного в него значения. Как передать в него параметр при таком подходе?
2. Поскольку переданный в DoHardWork параметр преобразуется и сохраняется в массив (причем важно сохранить последовательность поступления значений), то необходимо, что бы все вызовы метода DoHardWork у конкретного Worker'а, были выполнены в той же последовательности, в которой они вызывались (добавлялись в список). Будут ли задачи в листе выполнятся последовательно или асинхронно друг от друга?

[cicatrix]

More916, параметров передавайте сколько угодно, но...

были выполнены в той же последовательности

Сразу забудьте. Это многопоточность, потому - никаких гарантий. Громоздите семафоры, мутексы и прочие элементы синхронизации... в общем, Добро пожаловать

Answer 3

[gibiw]

async и await, не?
Тут

Comments

[More916]

Ну так я же так в вопросе и написал, не?

Как решить данную задачу с использованием task'ов и модификаторов await/async?

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

Answer 4

[stcmd04236]

Так создайте несколько потоков в котором задачи будет выполнятся асинхронно. В чем трудность?

Comments

[More916]

Все задачи являются операциями считывания из файла очередного байта и выполнения с ним определенных преобразований (с последующим добавление в массив). Разве могут несколько потоков считывать один и тот же файл?

[lam0x86]

More916, конечно, могут!

[stcmd04236]

Потоки могут обращаться одному и тому же потоку входных данных. Только тут возникает другая проблема связанная с синхронизацией потоков.