Multithreading 4-Tier Mixed Producer Consumer Pattern - Threads Not Alternating Properly

Question

Here is the problem I am trying to model: I have an incoming call center (no outbound calls). There are 3 employee levels - Freshers, Technical Lead, Project Manager. In our call center there is only one TL, one PM and multiple Freshers. Calls that the Freshers cannot handle are escalated to the TL, and calls that the TL cannot handle are escalated to the PM.

I need to take an OO C# approach.

Here is my current attempt: I used C# ConcurrentQueue since I thought it would take care of the 'locks'. I want FIFO for call center. I made a queue for each level.

I have a producer (callers) adding calls to the first queue. Mixed prod/consumer freshers check the call (take or escalate to the next queue). Mixed prod/consumer tl, then a pure consumer project manager.

The output is not what I expect. Only the first fresher runs and the project manager does not run at all. I expect a lot more alternation as the calls are added to the queue.

My code is below. Does anyone have a better approach to this problem or is there something wrong with my code?

The CallCenter Class is where most of the action takes place.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Collections.Concurrent;
namespace CallCenterThreaded
{
/// <summary>
/// 
/// MIXED CONSUMER/PRODUCER TYPE
/// </summary>
class Fresher
{
    // add a class variable for the number of Employee objects instantiated
    private static int fresherNum = 0;
    private int fresherId;
    private ConcurrentQueue<Call> incommingCalls;
    private ConcurrentQueue<Call> outgoingCalls;

    public Fresher(ConcurrentQueue<Call> calls, ConcurrentQueue<Call> outcalls)
    {
        fresherId = ++fresherNum;
        incommingCalls = calls;
        outgoingCalls = outcalls;
        //start the producer thread
        Thread thread = new Thread(new ThreadStart(HandleCalls));
        thread.Start();
    }

    public int ID
    {
        get { return fresherId; }
    }

    /// <summary>
    /// 
    /// </summary>
    public void HandleCalls() 
    {

        while (incommingCalls.Count > 0)
        {

            Call call;
            incommingCalls.TryDequeue(out call);

            if (call != null)
            {
                if (call.EscalationLevel == 0)
                {
                    TakeCalls(call);
                }
                else 
                {
                    TransferCalls(call);
                }
            }

        }

    }

    /// <summary>
    /// Transfer to the TL
    /// </summary>
    public void TransferCalls(Call call)
    {
        outgoingCalls.Enqueue(call);
        Console.WriteLine(".......Fresher {0} escalated call {1}", ID, call.CallID);
    }

    /// <summary>
    /// Consume the calls
    /// </summary>
    public void TakeCalls(Call call)
    {
        Console.WriteLine("Fresher {0} handled call {1}", ID, call.CallID);
    }



}
}
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading;
    using System.Collections.Concurrent;
   namespace CallCenterThreaded
    {
    /// <summary>
    /// MIXED CONSUMER/PRODUCER TYPE
    /// </summary>
    class TechnicalLead
    {

    // add a class variable for the number of Employee objects instantiated
    private static int tlNum = 0;
    private int tlId;

    private ConcurrentQueue<Call> incommingCalls;
    private ConcurrentQueue<Call> outgoingCalls;

    public TechnicalLead(ConcurrentQueue<Call> calls, ConcurrentQueue<Call> outcalls)
    {
        tlId = ++tlNum;
        incommingCalls = calls;
        outgoingCalls = outcalls;
        //start the producer thread
        Thread thread = new Thread(new ThreadStart(HandleCalls));
        thread.Start();
    }

    public int ID
    {
        get { return tlId; }
    }

    /// <summary>
    /// 
    /// </summary>
    public void HandleCalls() 
    {

        while (incommingCalls.Count > 0)
        {

            Call call;
            incommingCalls.TryDequeue(out call);

            if (call != null)
            {
                if (call.EscalationLevel == 1)
                {
                    TakeCalls(call);
                }
                else 
                {
                    TransferCalls(call);
                }
            }

        }

    }
    /// <summary>
    /// Transfer to the PM
    /// </summary>
    public void TransferCalls(Call call)
    {
        //Console.WriteLine(".......Technical Lead {0} escalated call {1}", ID, call.CallID);
        outgoingCalls.Enqueue(call);           
    }

    /// <summary>
    /// Consume the calls
    /// </summary>
    public void TakeCalls(Call call)
    {
        Console.WriteLine("Technical Lead {0} handled call {1}", ID, call.CallID);
    }
}
}
    using System;
   using System.Collections.Generic;
 using System.Linq;
 using System.Text;
   namespace CallCenterThreaded
   {
class Call
{
    private static int numberCalls = 0;
    private int callno;

    private int esclataion;


    public Call() 
    {
        callno = ++numberCalls;
        esclataion = 0;
        if(callno % 3 == 0)
        {
            esclataion = 1;
        }
        if(callno % 5 == 0)
        {
            esclataion = 2;
        }

    }

    public int CallID { get { return callno; } }
    public int EscalationLevel { get { return esclataion; } }
}
}
 using System;
 using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Collections.Concurrent;
namespace CallCenterThreaded
{
/// <summary>
/// 
/// </summary>
class CallCenter
{

    private ConcurrentQueue<Call> fresherCalls;
    private ConcurrentQueue<Call> tlCalls;
    private ConcurrentQueue<Call> pmCalls;

    private List<Caller> myCallers;
    private List<Fresher> myFreshers;

    private TechnicalLead tl;
    private ProjectManager pm;

    public CallCenter() 
    {
        //initial incomming calls to the fresher queue
        fresherCalls = new ConcurrentQueue<Call>();
        tlCalls = new ConcurrentQueue<Call>();
        pmCalls = new ConcurrentQueue<Call>();

        myFreshers = new List<Fresher>();
        myCallers = new List<Caller>();

        generate_callers();
        generate_freshers();

        tl = new TechnicalLead(tlCalls, pmCalls);
        pm = new ProjectManager(pmCalls);
    }

    /// <summary>
    /// 
    /// </summary>
    private void generate_freshers() 
    {
        for (int i = 0; i < 5; i++ )
        {
            myFreshers.Add(new Fresher(fresherCalls, tlCalls));
        }
    }

    /// <summary>
    /// 
    /// </summary>
    private void generate_callers() 
    {
        for (int i = 0; i < 5; i++ )
        {
            myCallers.Add(new Caller(fresherCalls));
        }
    }

   }
}
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Collections.Concurrent;
namespace CallCenterThreaded
{
/// <summary>
/// PURE CONSUMER
/// </summary>
class ProjectManager
{

     // add a class variable for the number of Employee objects instantiated
    private static int pmNum = 0;
    private int pmId;

    private ConcurrentQueue<Call> incommingCalls;

    public ProjectManager(ConcurrentQueue<Call> calls)
    {
        pmId = ++pmNum;
        incommingCalls = calls;

        //start the producer thread
        Thread thread = new Thread(new ThreadStart(HandleCalls));
        thread.Start();
    }

    public int ID
    {
        get { return pmId; }
    }

    /// <summary>
    /// 
    /// </summary>
    public void HandleCalls() 
    {

        while (incommingCalls.Count > 0)
        {

            Call call;
            incommingCalls.TryDequeue(out call);

            if (call != null && call.EscalationLevel == 2)
            {
                TakeCalls(call); 
            }

        }

    }
    /// <summary>
    /// Consume the calls
    /// </summary>
    public void TakeCalls(Call call)
    {
        Console.WriteLine("Project Manager {0} handled call {1}", ID, call.CallID);
    }
}
}
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Collections.Concurrent;
namespace CallCenterThreaded
{
class MainClass
{
    static void Main(string[] args)
    {

        CallCenter myCenter = new CallCenter();

        Console.ReadLine();
    }
}
}
    using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections.Concurrent;
using System.Threading;
namespace CallCenterThreaded
{
/// <summary>
/// This is a producer. It produces calls and adds them to the queue.
/// PURE PRODUCER TYPE
/// </summary>
class Caller
{

    private ConcurrentQueue<Call> incommingCalls;

    public Caller(ConcurrentQueue<Call> calls)
    {
        incommingCalls = calls;
        //start the producer thread
        Thread thread = new Thread(new ThreadStart(placeCalls));
        thread.Start();
    }

    public void placeCalls()
    {

        //place 10 calls
        for (int callno = 0; callno < 4; callno++)
        {
            //Console.WriteLine("Call {0} was added to the queue", callno);
            incommingCalls.Enqueue(new Call());
        }

    }

}
}

Answer 1

The problem is in the HandleCalls() methods:

public void HandleCalls() 
{
    while (incommingCalls.Count > 0)
    {
        …
    }
}

What this code does is to check incomingCalls and if there are none, it immediately exits the thread. It's as if the worker came to work, looked at his queue of calls, found nothing there (because he came to work at the same time as the freshers who are supposed to redirect the calls) and so he left and went home.

What you need to do is to wait if there is no work at the moment. Probably the best way to do that is to use BlockingCollection instead of ConcurrentQueue.

Also, your design doesn't seem to that good. For example, there is lots of repeated code.