Несколько итераций с многопоточностью

Question

Я переписал метод, который изначально был предназначен только для одного потока, для работы со многими потоками. Теперь этот метод принимает две одновременные коллекции: ConcurrentBag, который был List, и ConcurrentQueue, который был Queue.

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

Когда я писал это для многопоточности, случалось, что некоторые заголовки не совпадали (~ 20%), чего не было, когда они были в потоке. Только во время отладки я могу сопоставить эти заголовки, а затем присваивать значения. Должна быть некоторая проблема с повторением этих двух коллекций. Может быть, в одно и то же время многие потоки считывают значения из одного и того же элемента, но только чтение не должно быть проблемой?

Ниже код:

public void UpdateWithPercent(ref ConcurrentBag<Book> refList, ConcurrentQueue<Book> list)
{
    var size = list.Count;
    int numProcs = Environment.ProcessorCount;
    var divider = CalculatBiggestDivider(size);
    var nextIteration = 0;
    var remainingWork = numProcs;
    var internalRefList = refList;
    using (ManualResetEvent mre = new ManualResetEvent(false))
    {
        for (int i = 0; i < numProcs; i++)
        {
            ThreadPool.QueueUserWorkItem(delegate
            {
                IEnumerable<Book> partialList;
                while (-(nextIteration - Interlocked.Add(ref nextIteration, (partialList = DequeueChunk(list, divider)).Count()))> 0)
                {
                    foreach (var item in partialList)
                    {
                        foreach (var x in internalRefList)
                        {
                            if (x.Title == item.Title)
                            {
                                x.Orders += item.Orders;
                                break;
                            }
                        };
                    }
                }

                if (Interlocked.Decrement(ref remainingWork) == 0)
                {
                    mre.Set();
                }
            });
        }

        mre.WaitOne();
    }

    refList = internalRefList;
}

private int CalculatBiggestDivider(int count)
{
    var divider = 1;
    for (int i = 30; i > 0; i--)
    {
        if (count % i == 0)
        {
            divider = i;
            break;
        }
    }

    return divider;
}

private IEnumerable<T> DequeueChunk<T>(ConcurrentQueue<T> queue, int chunkSize)
{
    for (int i = 0; i < chunkSize && queue.Count > 0; i++)
    {
        T item;
        bool success = queue.TryDequeue(out item);
        if (!success)
        {
            i = chunkSize;
            continue;
        }
        yield return item;
    }
}

Answer 1

В конце концов я решил уйти из nestedloops и использовал ConcurrentDictionary. Теперь это работает.

public void UpdateWithPercent(ref ConcurrentDictionary<string, Book> refList, List<Book> list, int ticker, int maxTimes)
{
    var size = list.Count;
    int numProcs = Environment.ProcessorCount;
    var divider = CalculatBiggestDivider(size);
    var nextIteration = 0;
    var remainingWork = numProcs;
    var internalRefList = refList;
    using (ManualResetEvent mre = new ManualResetEvent(false))
    {
        for (int i = 0; i < numProcs; i++)
        {
            ThreadPool.QueueUserWorkItem(delegate
            {
                int index = 0;
                while ((index = Interlocked.Add(ref nextIteration, divider) - divider) < size)
                {
                    foreach (var item in list.GetRange(index, divider))
                    {
                        Book x;
                        if (internalRefList.TryGetValue(item.Title, out x))
                        {
                                x.Orders += item.Orders;
                        }
                    };
                }

                if (Interlocked.Decrement(ref remainingWork) == 0)
                {
                    mre.Set();
                }
            });
        }

        mre.WaitOne();
    }

    refList = internalRefList;
}