4

Имеется такой код.

// Коллекция буферов на запись
List<List<int>> writeReadyList = new List<List<int>>();

private void backgroundWorker1_DoWork(object sender, DoWorkEventArgs e)     
{
    // Локальный буфер данного потока
    List<int> localBufer = new List<int>();
    Random random = new Random();

    while (!backgroundWorker1.CancellationPending)
    {
        // Забиваем локальный буфер значениями
        localBufer.Add(random.Next(0, 100));

        if (localBufer.Count > 10)            
        {
            // Передаем буфер в очередь на запись и обнуляем его
            writeReadyList.Add(localBufer);
            localBufer = new List<int>();
        }

        Thread.Sleep(50);
    }
}

private void backgroundWorker2_DoWork(object sender, DoWorkEventArgs e)
{
    while (!backgroundWorker2.CancellationPending) 
    {
        // Коллекция буферов, которые не удалось записать
        List<List<int>> failList = new List<List<int>>(); 

        foreach (List<int> currentBufer in writeReadyList)
        {
            try
            {
                // Пишем значения из перебираемого буфера в базу данных
            }
            catch { failList.Add(currentBufer); }
        }

        writeReadyList = new List<List<int>>(failList);

        Thread.Sleep(450);
    }
}

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

Насколько понимаю, при таком подходе возможны несколько вариантов некорректной работы.

  1. в момент работы foreach во втором потоке она будет изменена из первого потока - получаем исключение?

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

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

______UPD:

По совету VladD использовал мьютексы.

    Mutex writeReadyAccess = new Mutex();

    private void backgroundWorker1_DoWork(object sender, DoWorkEventArgs e)
    {
        if (localBufer.Count > 10)
            // Таймаут в 1 секунду тк нельзя тормозить первый поток на время обработки очереди
            if (writeReadyAccess.WaitOne(1))
            {
                // Если второй поток не обрабатывает очередь, то добавляем в нее локальный буфер и обнуляем его
                writeReadyAccess.ReleaseMutex();
            }
        // Если очередь была в обработке на момент запроса, то продолжаем писать в локальный буфер        
    }

    private void backgroundWorker2_DoWork(object sender, DoWorkEventArgs e)
    {
        // Этот поток можно тормозить по времени, поэтому ждем без таймаута
        if (writeReadyAccess.WaitOne())
        {
            // Обработка очереди, в том числе ее перезапись
            writeReadyAccess.ReleaseMutex();
        }
    }
5
  • 2
    берете System.Collections.Concurrent.BlockingCollection. и в одном потоке пишете в эту коллекцию. а в другом потоке читатете GetConsumingEnumerable()
    – Stack
    Commented 19 янв. 2016 в 14:00
  • 1
    если будет использовать System.Collections.Concurrent.BlockingCollection, то в первом потоке, когда больше нечего будет бодавлять, то вызовите CompleteAdding() -- для того, чтобы второй поток тоже перестал ждать элементы.
    – Stack
    Commented 19 янв. 2016 в 14:22
  • 1
    Возможный дубликат вопроса: очередь задач на WinRT
    – andreycha
    Commented 19 янв. 2016 в 16:37
  • Комментарии не предназначены для расширенной дискуссии; разговор перемещен в чат.
    – Nick Volynkin
    Commented 21 янв. 2016 в 6:20
  • "использовал мьютексы" -- можно сделать проще, если использовать TPL DataFlow
    – Stack
    Commented 21 янв. 2016 в 8:39

1 ответ 1

3
using System.Collections.Concurrent;
using System.Threading;

class Dto {
    public int State;
    public int Index;
}

class Test {
    BlockingCollection<Dto> list = new BlockingCollection<Dto>();
    public void Run() {
        Task.Run(() => backgroundWorker2());
        Task.Run(() => backgroundWorker1());
    }
    void backgroundWorker1() {
        for(int i=0; i < 3; i++) 
            list.Add(new Dto() { Index = i });
    }
    void backgroundWorker2() {
        foreach (var v in list.GetConsumingEnumerable()) {
            // сохраняем данные
            Console.WriteLine("index=" + v.Index + " state=" + v.State);
            // если сохранить не удалось, то возвращаем в list
            if (v.Index == 0 && v.State++ < 2) list.Add(v);
            if(list.Count == 0) 
              break;
        }
        Console.WriteLine("completed"); 
    }
}

var t = new Test();
t.Run();

Результат

index=0 state=0
index=1 state=0
index=2 state=0
index=0 state=1     вторая попытка записи
index=0 state=2     еще одна 
completed

Если надо передавать данные порциями, то определите коллекцию, например, так BlockingCollection<Dto[]>


UPDATE

Другая версия, в которой backgroundWorker1 работает медленно

class Dto {
    public int Index;
}

class Test {
    BlockingCollection<Dto> list = new BlockingCollection<Dto>();
    public void Run() {
        Task.Run(() => backgroundWorker2());
        Task.Run(() => backgroundWorker1());
    }
    void backgroundWorker1() {
        for (int i = 0; i < 3; i++) {
            list.Add(new Dto() { Index = i });
            // Thread.Sleep(1000);
        }
        list.CompleteAdding();
    }
    void backgroundWorker2() {
        var retry = new Queue<Dto>();
        foreach (var v in list.GetConsumingEnumerable()) {
            // сохраняем данные
            Console.WriteLine("index=" + v.Index);
            // если сохранить не удалось, то ставим в очередь
            if (v.Index == 0) retry.Enqueue(v);
        }
        // повторяем попытку записи 
        while(retry.Count > 0) Console.WriteLine("retry=" + retry.Dequeue().Index);
        Console.WriteLine("completed");
    }
}

var t = new Test();
t.Run();

Результат

index=0
index=1
index=2
retry=0          повторная попытка записи
completed

UPDATE

Для конвейеризации обработки данных в разных потоках предназначена библиотека потоков данных - TPL Dataflow. Описание с примерами есть в MSDN и есть nuget-пакет Microsoft TPL Dataflow.

11
  • 1
    (1) CompleteAdding? (2) Вы всё же не синхронизируете list.
    – VladD
    Commented 19 янв. 2016 в 20:07
  • 1
    Ну... ideone.com/WLhzBG // По поводу синхронизации list, я вас не убедил, а жаль.
    – VladD
    Commented 19 янв. 2016 в 20:41
  • 1
    Ну да. Правильность многопоточного кода не должна зависеть от порядка, в котором scheduler выбирает потоки. Sleep там просто чтобы спровоцировать race condition пойти по тому пути, где не всё так гладко.
    – VladD
    Commented 19 янв. 2016 в 20:55
  • 1
    Ну, раз на миллион scheduler задержит выполнение одного из потоков, так что аргумент про «неспешно» не катит. Любой поток может в любое время остановиться (например, на JIT или сборку мусора) на сколько угодно.
    – VladD
    Commented 19 янв. 2016 в 22:03
  • 2
    Ну, завтра запустят на сервере. Или внутри цикла добавят вызов другой функции. Ну и даже без этого всего, scheduler вовсе не обязан давать потоку квант, это не гарантировано ничем. Честно. Он может отложить активацию потока на неопределённое время. Нечасто, раз на миллион случаев или и того реже. Поверьте уж человеку, написавшему за свою жизнь (и отладившему) много многопоточного кода.
    – VladD
    Commented 19 янв. 2016 в 22:14

Ваш ответ

Нажимая «Отправить ответ», вы соглашаетесь с условиями пользования и подтверждаете, что прочитали политику конфиденциальности.

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