Паттерн WaitAllOneByOne с semaphoreSlim, безопасен ли этот поток реализации? - PullRequest
Новая
       38

Паттерн WaitAllOneByOne с semaphoreSlim, безопасен ли этот поток реализации?

0 голосов
/ 21 сентября 2018

У меня есть две задачи, одна из которых получает данные и занимает около 2 секунд, а другая обрабатывает данные и занимает около 3 секунд.Я должен выполнить эти две задачи для около 13.000 различных элементов.Если я буду выполнять эти задачи последовательно, это займет около 18 часов.

Теперь я думал об использовании шаблона WaitAllOneByOne, но вызов Task.Run для каждого элемента приведет к свертыванию моей базы данных, поэтому я подумал о реализацииSemaphoreSlim для контроля того, сколько потоков должно выполняться в каждой задаче.

До сих пор мне удавалось создать реализацию шаблона, но Я не знаю, является ли он потокобезопасным, и у меня нетидеи о том, как его протестировать .

Код выглядит следующим образом: 

class Program
{
    static async Task Main(string[] args)
    {
        List<Guid> idsCombinacion = GetIdsCombinacion(10);
        List<Task<string>> todo = new List<Task<string>>(idsCombinacion.Count);
        List<Task> processing = new List<Task>(idsCombinacion.Count);

        //this starts callings to GetData, up to 2 threads.
        SemaphoreSlim semaphoreGetData = new SemaphoreSlim(2);
        var dataTask = GetDataAsync(idsCombinacion, todo, semaphoreGetData);

        SemaphoreSlim semaphoreProcessData = new SemaphoreSlim(3);
        while (todo.Count > 0)
        {
            Task<string>[] tasks = todo.ToArray();
            int idx = Task.WaitAny(tasks);
            todo.RemoveAt(idx);

            //once the tasks in the to-do list goes finishing, we start a new task to perform the Processing, up to 3 threads
            semaphoreProcessData.Wait();
            var result = tasks[idx].Result;
            tasks[idx].Dispose();
            var t = Task.Factory.StartNew(() =>
            {
                Task.Delay(3000).Wait();
                Console.WriteLine($"todo: {todo.Count}; processing: {processing.Count}; guid: {result}");
                semaphoreProcessData.Release();
            }, TaskCreationOptions.LongRunning);
            processing.Add(t);

        }

        var aux = processing.ToArray();
        //we wait for all the jobs to complete
        Task.WaitAll(aux);

        await dataTask;

        semaphoreGetData.Dispose();
        semaphoreProcessData.Dispose();
    }

    private static async Task GetDataAsync(List<Guid> idsCombinacion, List<Task<string>> todo, SemaphoreSlim semaphoreGetData)
    {
        foreach (var idComb in idsCombinacion)
        {
            await semaphoreGetData.WaitAsync();
            var t = Task<string>.Factory.StartNew(() =>
            {
                Task.Delay(2000).Wait();
                semaphoreGetData.Release();
                return "Guid: " + idComb;
            }, TaskCreationOptions.LongRunning);
            todo.Add(t);
        }
    }


    private static List<Guid> GetIdsCombinacion(int howMany)
    {
        var idsCombinacion = new List<Guid>(howMany);
        for (int i = 0; i < howMany; i++)
            idsCombinacion.Add(Guid.NewGuid());
        return idsCombinacion;
    }
}

Шаблон WaitAllOneByOne от доктора Джо Хаммела в этом множественном курсе .

...