Thread order synchronization

Question

Suppose I have a Queue of Tasks, and each Task have a locking object (syncObject) which controls shared resource access, Queue can have multiple Task which share same instances of syncObject. And I have N concurrent threads that should dequeue Tasks and proccess them in queue order, this means acquire lock on syncObject in the order of queue.

Code explanation:

abstract class Task
{
    public readonly Object SyncObject = new Object();
}

Queue<Task> taskQueue = new Queue<Task>();
Object queueLock = new Object();

void TakeItemThreadedMethod()
{
    Task task;
    lock(queueLock) task = taskQueue.Dequeue();
    //Between this lines is my problem,
    //Other thread can dequeue next task and it may share same syncObject and
    //acquire lock on it before this thread, thought this task was first in queue
    lock(task.SyncObject)
    {
        //Do some work here
    }
}

How to start proccessing Tasks (acquire Task.SyncObject lock) that share the same SyncObject in the order they were in Queue.

Answer 1

It sounds like potentially your queue shouldn't contain individual tasks - but queues of tasks, where each subqueue is "all the tasks which share a sync-lock".

Your processor would therefore:

• Take a subqueue off the main queue
• Dequeue the first task off the subqueue and process it
• When it's finished put the subqueue back at the end of the main queue (or anywhere, actually - work out how you want the scheduling to work)

This will ensure that only one task per subqueue is ever executed at a time.

You'll probably need a map from lock to subqueue, so that anything creating work can add it to the right subqueue. You'd need to atomically work out when to remove a subqueue from the map (and not put it back on the main queue), assuming you require that functionality at all.

EDIT: As an optimization for the above, you could put the subqueue itself into whatever you're using as the shared sync lock. It could have a reference to either "the single task to next execute" or "a queue of tasks" - only creating the queue lazily. You'd then put the sync lock (which wouldn't actually need to be used as a lock any more) on the queue, and each consumer would just ask it for the next task to execute. If only a single task is available, it's returned (and the "next task" variable set to null). If there are multiple tasks available, the first is dequeued.

When a producer adds a new task, either the "first task" variable is set to the task to execute if it was previously null, or a queue is created if there wasn't a queue but was already a task, or the queue is just added to if one already existed. That solves the inefficiency of unnecessary queue creation.

Again, the tricky part will be working out how to atomically throw away the shared resource lock - because you only want to do so after processing the last item, but equally you don't want to miss a task because you happened to add it at the wrong time. It shouldn't be too bad, but equally you'll need to think about it carefully.

Answer 2

How about this approach:

• use a list instead of a queue
• have each worker thread loop in order through the queue until it can find an "unlocked" task

Something like (untested):

abstract class Task
{
    public readonly Object SyncObject = new Object();
}

List<Task> taskList = new List<Task>();

void TakeItemThreadedMethod()
{
    Task task = null;
    bool found = false;

    try
    {
        // loop until found an task whose SyncObject is free
        while (!found)
        {
            lock (taskList)
            {
                for (int i = 0; i < taskList.Count; i++)
                {
                    object syncObj = taskList[i].SyncObject;
                    if (found = Monitor.TryEnter(syncObj))
                    {
                        for (int x = 0; x < taskList.Count; x++)
                        {
                            if (Object.ReferenceEquals(
                                    syncObj, taskList[x].SyncObject))
                            {
                                task = taskList[x];
                                taskList.RemoveAt(x);
                                break;
                            }
                        }
                        break;
                    }
                }
            }
        }

        // process the task...
        DoWork(task);
    }
    finally
    {
        if (found) Monitor.Exit(task.SyncObject);
    }
}

void QueueTask(Task task)
{
    lock (taskList)
    {
        taskList.Add(task);
    }
}

Answer 3

I have used QueuedLock class suggested by Matthew Brindley, with slight modification, I have split up Enter function to TakeTicket and Enter which blocks.

Now I can use TakeTicket inside shared QueueLock without blocking whole queue.

Modified code:

abstract class Task
{
    public readonly QueuedLock SyncObject = new QueuedLock();
}

Queue<Task> taskQueue = new Queue<Task>();
Object queueLock = new Object();

void TakeItemThreadedMethod()
{
    Task task;
    int ticket;
    lock(queueLock) 
    {
        task = taskQueue.Dequeue();
        ticket = task.SyncObject.TakeTicket();
    }
    task.SyncObject.Enter(ticket);
    //Do some work here
    task.SyncObject.Exit();
}