Multithreading BlockingCollection same value

Question

I am using two threads in a C# application that access the same BlockingCollection. This works fine, but I want to retrieve the first value twice so the two threads retrieve the same value *.

After a few seconds I want to poll the currentIndex of both threads and delete every value < index. So for example the lowest currentIndex of a thread is 5, the application deletes the items at index 0 -5 in the queue. Another solution is to delete the value in the queue if all threads processed the value.

How can I accomplish this? I think I need another type of buffer..?

Thank you in advance!

*If .Take() is called by thread1, the item is removed in the collection and thread2 can't get the same item again.

---

Update:

I want to store data in a buffer, so for example thread1 saves the data to a HDD and thread2 analyzes the (same) data (concurrent).

Answer 1

Use a producer-consumer to add Value1 to two separate ConcurrentQueues. Have the threads dequeue then process them from their own queue.

Edit 7/4/14: Here's a, hazy, hacky, and half thought out solution: Create a custom object that is buffered. It could include space for both the information you're trying to buffer in thread 1 and the analysis results in thread 2.

Add the objects to a buffer in thread 1 and a BlockingCollection. Use thread 2 to analyse the results and update the objects with the results. The blocking collection shouldn't get too big, and since it's only dealing with references shouldn't hit your memory. This assumes that you won't be modifying the info in the buffer at the same time on both threads.

Another, also half thought out solution is to feed the info into the buffer and a blocking collection simultaneously. Analyse the data from the BlockingCollection, feed it into an output collection and match them up with the buffer again. This option can handle concurrent modification if you do it right, but is probably more work.

I think option one is better. As I've pointed out, these are only half-formed, but they might help you find something that suits your specific needs. Good luck.

Answer 2

I would suggest to rethink your design.

When you have a list of items which have to processed then give each thread a queue of items which he have to work on.

With such a solution it wouldn't be a problem to give both or more threads the same value to process.

Something like this, not tested just typed.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Collections.Concurrent;

namespace ConsoleApplication2
{

  class Item
  {
    private int _value;
    public int Value
    {
      get
      {
        return _value;
      }
    }

    // all you need
    public Item(int i)
    {
      _value = i;
    }
  }

  class WorkerParameters
  {
    public ConcurrentQueue<Item> Items = new ConcurrentQueue<Item>();
  }

  class Worker
  {
    private Thread _thread;
    private WorkerParameters _params = new WorkerParameters();

    public void EnqueueItem(Item item)
    {
      _params.Items.Enqueue(item);
    }

    public void Start()
    {
      _thread = new Thread(new ParameterizedThreadStart(ThreadProc));
      _thread.Start();
    }

    public void Stop()
    {
      // build somthing to stop your thread
    }

    public static void ThreadProc(object threadParams)
    {
      WorkerParameters p = (WorkerParameters)threadParams;
      while (true)
      {
        while (p.Items.Count > 0)
        {
          Item item = null;
          p.Items.TryDequeue(out item);

          if (item != null)
          {
            // do something
          }

        }
        System.Threading.Thread.Sleep(50);
      }
    }
  }

  class Program
  {
    static void Main(string[] args)
    {

      Worker w1 = new Worker();
      Worker w2 = new Worker();
      w1.Start();
      w2.Start();

      List<Item> itemsToProcess = new List<Item>();
      for (int i = 1; i < 1000; i++)
      {
        itemsToProcess.Add(new Item(i));
      }

      for (int i = 1; i < 1000; i++)
      {
        w1.EnqueueItem(itemsToProcess[i]);
        w2.EnqueueItem(itemsToProcess[i]);
      }


    }
  }
}