1
\$\begingroup\$

I am trying to implement producer/consumer pattern with multiple or parallel consumers.

I did an implementation but I would like to know how good it is. Can somebody do better? Can any of you spot any errors?

Unfortunately I can not use TPL dataflow, because we are at the end of our project and to put in an extra library in our package would take to much paperwork and we do not have that time.

What I am trying to do is to speed up the following portion:

anIntermediaryList = StepOne(anInputList); // I will put StepOne as Producer :-) Step one is remote call.

aResultList = StepTwo(anIntermediaryList); // I will put StepTwo as Consumer, however he also produces result. Step two is also a remote call.
// StepOne is way faster than StepTwo.

For this I came up with the idea that I will chunk the input list (anInputList)

StepOne will be inside of a Producer and will put the intermediary chunks into a queue. There will be multiple Producers and they will take the intermediary results and process it with StepTwo.

Here is a simplified version of of the implementation later:

Task.Run(() => {
     aChunkinputList = Split(anInputList)
     foreach(aChunk in aChunkinputList)
     {
          anIntermediaryResult = StepOne(aChunk)
          intermediaryQueue.Add(anIntermediaryResult)
     }
})

while(intermediaryQueue.HasItems)
{
     anItermediaryResult = intermediaryQueue.Dequeue()
     Task.Run(() => {
         aResultList = StepTwo(anItermediaryResult);
         resultQueue.Add(aResultList)
     }
}

I also thought that the best number for the parallel running Consumers would be: "Environment.ProcessorCount / 2". I would like to know if this also is a good idea.

Now here is my mock implementation and the question is can somebody do better or spot any error?

class Example
{
    protected static readonly int ParameterCount_ = 1000;
    protected static readonly int ChunkSize_ = 100;
    // This might be a good number for the parallel consumers.
    protected static readonly int ConsumerCount_ = Environment.ProcessorCount / 2;
    protected Semaphore mySemaphore_ = new Semaphore(Example.ConsumerCount_, Example.ConsumerCount_);

    protected ConcurrentQueue<List<int>> myIntermediaryQueue_ = new ConcurrentQueue<List<int>>();
    protected ConcurrentQueue<List<int>> myResultQueue_ = new ConcurrentQueue<List<int>>();

    public void Main()
    {
        List<int> aListToProcess = new List<int>(Example.ParameterCount_ + 1);
        aListToProcess.AddRange(Enumerable.Range(0, Example.ParameterCount_));

        Task aProducerTask = Task.Run(() => Producer(aListToProcess));

        List<Task> aTaskList = new List<Task>();            
        while(!aProducerTask.IsCompleted || myIntermediaryQueue_.Count > 0)
        {
            List<int> aChunkToProcess;
            if (myIntermediaryQueue_.TryDequeue(out aChunkToProcess))
            {
                mySemaphore_.WaitOne();
                aTaskList.Add(Task.Run(() => Consumer(aChunkToProcess)));
            }
        }

        Task.WaitAll(aTaskList.ToArray());

        List<int> aResultList = new List<int>();
        foreach(List<int> aChunk in myResultQueue_)
        {
            aResultList.AddRange(aChunk);
        }
        aResultList.Sort();
        if (aListToProcess.SequenceEqual(aResultList))
        {
            Console.WriteLine("All good!");
        }
        else
        {
            Console.WriteLine("Bad, very bad!");
        }
    }

    protected void Producer(List<int> elements_in)
    {
        List<List<int>> aChunkList = Example.SplitList(elements_in, Example.ChunkSize_);

        foreach(List<int> aChunk in aChunkList)
        {
            Console.WriteLine("Thread Id: {0} Producing from: ({1}-{2})", 
                Thread.CurrentThread.ManagedThreadId,
                aChunk.First(),
                aChunk.Last());

            myIntermediaryQueue_.Enqueue(ProduceItemsRemoteCall(aChunk));
        }
    }

    protected void Consumer(List<int> elements_in)
    {
        Console.WriteLine("Thread Id: {0} Consuming from: ({1}-{2})",
                Thread.CurrentThread.ManagedThreadId,
                Convert.ToInt32(Math.Sqrt(elements_in.First())),
                Convert.ToInt32(Math.Sqrt(elements_in.Last())));

        myResultQueue_.Enqueue(ConsumeItemsRemoteCall(elements_in));
        mySemaphore_.Release();
    }

    // Dummy Remote Call
    protected List<int> ProduceItemsRemoteCall(List<int> elements_in)
    {
        return elements_in.Select(x => x * x).ToList();
    }

    // Dummy Remote Call
    protected List<int> ConsumeItemsRemoteCall(List<int> elements_in)
    {
        return elements_in.Select(x => Convert.ToInt32(Math.Sqrt(x))).ToList();
    }

    public static List<List<int>> SplitList(List<int> masterList_in, int chunkSize_in)
    {
        List<List<int>> aReturnList = new List<List<int>>();
        for (int i = 0; i < masterList_in.Count; i += chunkSize_in)
        {
            aReturnList.Add(masterList_in.GetRange(i, Math.Min(chunkSize_in, masterList_in.Count - i)));
        }
        return aReturnList;
    }
}

Main function:

class Program
{
    static void Main(string[] args)
    {
        Example anExample = new Example();
        anExample.Main();
    }
}
\$\endgroup\$

closed as off-topic by Nick Udell, RubberDuck, Vogel612, syb0rg, ferada Dec 2 '14 at 19:35

This question appears to be off-topic. The users who voted to close gave this specific reason:

If this question can be reworded to fit the rules in the help center, please edit the question.

  • \$\begingroup\$ It contained only introductory pseudo code, but the rest was code. It can be closed, I am almost finished with it. \$\endgroup\$ – Laszlo Dec 14 '14 at 21:14
1
\$\begingroup\$

Based on comments I also released a third version too. I used this: http://www.albahari.com/threading/part5.aspx#_BlockingCollectionT But over here there is a problem, the producer and the consumers are not running parallel. Need to look into that. :-(

Here is the code in that case:

class Example
{
    protected static readonly int ParameterCount_ = 1000;
    protected static readonly int ChunkSize_ = 100;
    protected static readonly int DefaultTimeoutInMilliseconds_ = 5000;
    // This might be a good number for the parallel consumers.
    protected static readonly int ConsumerCount_ = Environment.ProcessorCount / 2;

    protected ConcurrentQueue<List<int>> myResultQueue_ = null;

    public void Main()
    {
        List<int> aListToProcess = new List<int>(Example.ParameterCount_ + 1);
        aListToProcess.AddRange(Enumerable.Range(0, Example.ParameterCount_));

        myResultQueue_ = new ConcurrentQueue<List<int>>();

        using (PCQueue aPCQueue = new PCQueue(Example.ConsumerCount_))
        {
            List<Task> aTaskList = new List<Task>();

            List<List<int>> aChunkList = Example.SplitList(aListToProcess, Example.ChunkSize_);
            foreach (List<int> aChunk in aChunkList)
            {
                aTaskList.Add(aPCQueue.EnqueueTask(() =>
                {
                    List<int> anIntermediaryChunk = ProduceItemsRemoteCall(aChunk);
                    Consumer(anIntermediaryChunk);
                }));
            }                

            Task.WaitAll(aTaskList.ToArray());
        }

        List<int> aResultList = new List<int>();
        foreach (List<int> aChunk in myResultQueue_)
        {
            aResultList.AddRange(aChunk);
        }
        aResultList.Sort();
        if (aListToProcess.SequenceEqual(aResultList))
        {
            Console.WriteLine("All good!");
        }
        else
        {
            Console.WriteLine("Bad, very bad!");
        }
    }

    protected void Consumer(List<int> elements_in)
    {
        int anIntervalStart = Convert.ToInt32(Math.Sqrt(elements_in.First()));
        int anIntervalEnd = Convert.ToInt32(Math.Sqrt(elements_in.Last()));
        WriteOutInfo("Consuming", anIntervalStart, anIntervalEnd);

        myResultQueue_.Enqueue(ConsumeItemsRemoteCall(elements_in));

        WriteOutInfo("Consumed ", anIntervalStart, anIntervalEnd);
    }

    // Dummy Remote Call
    protected List<int> ProduceItemsRemoteCall(List<int> elements_in)
    {
        return elements_in.Select(x => x * x).ToList();
    }

    // Dummy Remote Call
    protected List<int> ConsumeItemsRemoteCall(List<int> elements_in)
    {
        return elements_in.Select(x => Convert.ToInt32(Math.Sqrt(x))).ToList();
    }

    protected void WriteOutInfo(string actionName_in, int intervalStart_in, int intervalEnd_in)
    {
        Console.WriteLine("Thread Id: {0} {1} from: ({2}-{3})",
            Thread.CurrentThread.ManagedThreadId,
            actionName_in,
            intervalStart_in,
            intervalEnd_in);
    }

    public static List<List<int>> SplitList(List<int> masterList_in, int chunkSize_in)
    {
        List<List<int>> aReturnList = new List<List<int>>();
        for (int i = 0; i < masterList_in.Count; i += chunkSize_in)
        {
            aReturnList.Add(masterList_in.GetRange(i, Math.Min(chunkSize_in, masterList_in.Count - i)));
        }
        return aReturnList;
    }
}

And here is my slightly modified version of PCQueue:

public class PCQueue : IDisposable
{
    class WorkItem
    {
        public readonly TaskCompletionSource<object> TaskSource;
        public readonly Action Action;
        public readonly CancellationToken? CancelToken;

        public WorkItem(TaskCompletionSource<object> taskSource_in, Action action_in, CancellationToken? cancelToken_in)
        {
            TaskSource = taskSource_in;
            Action = action_in;
            CancelToken = cancelToken_in;
        }
    }

    BlockingCollection<WorkItem> myBlockingCollection_ = new BlockingCollection<WorkItem>();

    public PCQueue(int workerCount_in)
    {
        // Create and start a separate Task for each consumer:
        for (int i = 0; i < workerCount_in; i++)
            Task.Run(() => Consume());
    }

    public void Dispose() { myBlockingCollection_.CompleteAdding(); }

    public Task EnqueueTask(Action action_in)
    {
        return EnqueueTask(action_in, null);
    }

    public Task EnqueueTask(Action action_in, CancellationToken? cancelToken_in)
    {
        TaskCompletionSource<object> aTCS = new TaskCompletionSource<object>();
        myBlockingCollection_.Add(new WorkItem(aTCS, action_in, cancelToken_in));
        return aTCS.Task;
    }

    void Consume()
    {
        foreach (WorkItem aWorkItem in myBlockingCollection_.GetConsumingEnumerable())
        {
            if (aWorkItem.CancelToken.HasValue && aWorkItem.CancelToken.Value.IsCancellationRequested)
            {
                aWorkItem.TaskSource.SetCanceled();
            }
            else
            {
                try
                {
                    aWorkItem.Action();
                    aWorkItem.TaskSource.SetResult(null);   // Indicate completion
                }
                catch (OperationCanceledException ex)
                {
                    if (ex.CancellationToken == aWorkItem.CancelToken)
                        aWorkItem.TaskSource.SetCanceled();
                    else
                        aWorkItem.TaskSource.SetException(ex);
                }
                catch (Exception ex)
                {
                    aWorkItem.TaskSource.SetException(ex);
                }
            }
        }
    }
}
\$\endgroup\$
0
\$\begingroup\$

Based on some tips I would like to release a second version of the code. :-) First of all I introduced BlockingCollection. I can not get rid of Semaphore because I want to control the number of Consumer/Threads. I also made the code more robust (and complicated) and added some basic error handling.

Here is the second version:

class Example
{
    protected static readonly int ParameterCount_ = 1000;
    protected static readonly int ChunkSize_ = 100;
    protected static readonly int DefaultTimeoutInMilliseconds_ = 5000;
    // This might be a good number for the parallel consumers.
    protected static readonly int ConsumerCount_ = Environment.ProcessorCount / 2;

    protected ConcurrentQueue<List<int>> myResultQueue_ = null;

    public void Main()
    {
        List<int> aListToProcess = new List<int>(Example.ParameterCount_ + 1);
        aListToProcess.AddRange(Enumerable.Range(0, Example.ParameterCount_));

        using (BlockingCollection<List<int>> anIntermediaryQueue = new BlockingCollection<List<int>>())            
        using (Semaphore aSemaphore = new Semaphore(Example.ConsumerCount_, Example.ConsumerCount_))
        {
            myResultQueue_ = new ConcurrentQueue<List<int>>();

            Task aProducerTask = Task.Run(() => Producer(aListToProcess, anIntermediaryQueue));

            List<Task> aTaskList = new List<Task>();
            while (!anIntermediaryQueue.IsCompleted)
            {
                List<int> aChunkToProcess;
                if (anIntermediaryQueue.TryTake(out aChunkToProcess, Example.DefaultTimeoutInMilliseconds_))
                {
                    if (aSemaphore.WaitOne(Example.DefaultTimeoutInMilliseconds_))
                    {
                        aTaskList.Add(Task.Run(() =>
                        {
                            try
                            {
                                Consumer(aChunkToProcess);
                            }
                            finally
                            {
                                aSemaphore.Release();
                            }
                        }));
                    }
                    else
                    {
                        throw new Exception("Semaphore timed out, threads/consumers did not release the Semaphores.");
                    }
                }
                else
                {
                    throw new Exception("anIntermediaryQueue timed out, the producer did not produce in time or did not close the queue.");
                }
            }

            Task.WaitAll(aTaskList.ToArray());
        }

        List<int> aResultList = new List<int>();
        foreach(List<int> aChunk in myResultQueue_)
        {
            aResultList.AddRange(aChunk);
        }
        aResultList.Sort();
        if (aListToProcess.SequenceEqual(aResultList))
        {
            Console.WriteLine("All good!");
        }
        else
        {
            Console.WriteLine("Bad, very bad!");
        }
    }

    protected void Producer(List<int> elements_in, BlockingCollection<List<int>> intermediaryQueue_in)
    {
        try
        {
            List<List<int>> aChunkList = Example.SplitList(elements_in, Example.ChunkSize_);
            foreach (List<int> aChunk in aChunkList)
            {
                WriteOutInfo("Producing", aChunk.First(), aChunk.Last());
                intermediaryQueue_in.Add(ProduceItemsRemoteCall(aChunk));
            }
        }
        finally
        {
            intermediaryQueue_in.CompleteAdding();
        }            
    }

    protected void Consumer(List<int> elements_in)
    {
        int anIntervalStart = Convert.ToInt32(Math.Sqrt(elements_in.First()));
        int anIntervalEnd = Convert.ToInt32(Math.Sqrt(elements_in.Last()));
        WriteOutInfo("Consuming", anIntervalStart, anIntervalEnd );

        myResultQueue_.Enqueue(ConsumeItemsRemoteCall(elements_in));

        WriteOutInfo("Consumed ", anIntervalStart, anIntervalEnd);
    }

    // Dummy Remote Call
    protected List<int> ProduceItemsRemoteCall(List<int> elements_in)
    {
        return elements_in.Select(x => x * x).ToList();
    }

    // Dummy Remote Call
    protected List<int> ConsumeItemsRemoteCall(List<int> elements_in)
    {
        return elements_in.Select(x => Convert.ToInt32(Math.Sqrt(x))).ToList();
    }

    protected void WriteOutInfo(string actionName_in, int intervalStart_in, int intervalEnd_in)
    {
        Console.WriteLine("Thread Id: {0} {1} from: ({2}-{3})",
            Thread.CurrentThread.ManagedThreadId,
            actionName_in,
            intervalStart_in,
            intervalEnd_in);
    }

    public static List<List<int>> SplitList(List<int> masterList_in, int chunkSize_in)
    {
        List<List<int>> aReturnList = new List<List<int>>();
        for (int i = 0; i < masterList_in.Count; i += chunkSize_in)
        {
            aReturnList.Add(masterList_in.GetRange(i, Math.Min(chunkSize_in, masterList_in.Count - i)));
        }
        return aReturnList;
    }
}
\$\endgroup\$

Not the answer you're looking for? Browse other questions tagged or ask your own question.