7
\$\begingroup\$

I have multiple threads calling a method, passing in an ID and value.

I have two constraints I need to place on this method:

  • Only one of the same ID can be processed at a time. Additional threads with the same ID must wait.
  • Subsequent calls with the same ID must be processed in the order they were received.

Here's an Example.

  • Method is called by four different threads with the parameters

    • ID 1 Value a
    • ID 2 Value b
    • ID 1 Value c
    • ID 1 Value d
  • Program should be able to process [ID 1 Value a] and [ID 2 Value b] immediately, * while [ID 1 Value c] and [ID 1 Value d] wait for [ID 1 Value a] to finish.

  • Once [ID 1 Value a] finishes, then the method runs for [ID 1 Value c] while [ID 1 Value d] waits.
  • When [ID 1 Value c] finishes the method runs for [ID 1 Value d].
  • If the method wrote the value to the console then I'd expect the output to be either a, b, c, d or b, a, c, d.

I've had a quick go at it but it's messy. I'm writing this post in hopes of getting an improved and abstracted code snippet for doing this.

Here is my attempt:

static object _lockObj = new object();
static Dictionary<int, Queue<ManualResetEvent>> runningIds = new Dictionary<int, Queue<ManualResetEvent>>();

public async Task TheMethod(int id, string val)
{
    ManualResetEvent enquedResetEvent = null;
    lock (_lockObj)
    {
        if (runningIds.ContainsKey(id))
        {
            enquedResetEvent = new ManualResetEvent(false);
            runningIds[id].Enqueue(enquedResetEvent);
        }
        else
        {
            runningIds.Add(id, new Queue<ManualResetEvent>());
        }
    }
    if (enquedResetEvent != null)
        enquedResetEvent.WaitOne();
    try
    {
        // DO LONG RUNNING TASK
        await Task.Delay(3000);
        Console.WriteLine(id + " " + val + " at time " + DateTime.Now.ToString("m:ss"));
    }
    finally
    {
        lock (_lockObj)
        {
            if (runningIds[id].Count > 0)
            {
                var dequedResetEvent = runningIds[id].Dequeue();
                dequedResetEvent.Set();
            }
            else
            {
                runningIds.Remove(id);
            }
        }
    }
}

Sample input and output would be:

List<Task> ts = new List<Task>();
ts.Add(Task.Run(() => new Test().TheMethod(1, "a")));
Thread.Sleep(10);
ts.Add(Task.Run(() => new Test().TheMethod(2, "b")));
Thread.Sleep(10);
ts.Add(Task.Run(() => new Test().TheMethod(1, "c")));
Thread.Sleep(10);
ts.Add(Task.Run(() => new Test().TheMethod(1, "d")));
Task.WaitAll(ts.ToArray());
// Output:
// 1 a at time 25:53
// 2 b at time 25:53
// 1 c at time 25:56 // note 3 second delay
// 1 d at time 25:59
// 1 e at time 26:02
// 1 f at time 26:05
\$\endgroup\$
  • \$\begingroup\$ After closer inspection, I've reopened this question. Thanks for your patience. \$\endgroup\$ – 200_success Feb 12 '15 at 7:44
4
\$\begingroup\$

Here are my thoughts on the code that you provided:

  1. You should mark your lockObj and runningIds as readonly
  2. You are using the presence of the queue in the dictionary as a signal that another thread is currently working on specific ID which is not quite obvious
  3. You mix non-blocking (await) approach along with blocking (ManualResetEvent.WaitOne). It would be better to stick to one of them (preferably non-blocking).
  4. If you change the approach from "notify the guy after me that he can start working" to "I'm done", then you would be able to capture only the last "guy" against specific ID. "I'm done" is easily modelled by Task.

As a result, my first refactoring of your code looks like this:

private static readonly object _lockObj = new object();
private static readonly Dictionary<int, Task> _runningIds = new Dictionary<int, Task>();

public async Task TheMethod(int id, string val)
{
    //Will be used to signal others that the work has finished
    var completionSource = new TaskCompletionSource<bool>();

    Task currentTask;

    lock (_lockObj)
    {
        _runningIds.TryGetValue(id, out currentTask);
        _runningIds[id] = completionSource.Task;
    }

    if (currentTask != null)
        await currentTask; 

    // DO LONG RUNNING TASK
    await Task.Delay(3000);
    Console.WriteLine(id + " " + val + " at time " + DateTime.Now.ToString("m:ss"));

    completionSource.SetResult(true);

    //cleanup
    lock (_lockObj)
    {
        if (_runningIds.TryGetValue(id, out currentTask) && currentTask == completionSource.Task)
            _runningIds.Remove(id);
    }
}

Second take was mostly as an exercise to get rid of lock at all by using the ConcurrentDictionary.

private static readonly ConcurrentDictionary<int, Task> _runningIds = new ConcurrentDictionary<int, Task>();

private static Task RegisterAsLastScheduledTask(int id, Task task)
{
    Task previousTask = null;

    //Either adding a new task to the dictionary, or capturing and replacing previous task
    while (!_runningIds.TryAdd(id, task) &&
        !(_runningIds.TryGetValue(id, out previousTask) && _runningIds.TryUpdate(id, task, previousTask)))
    {
        previousTask = null;
    }

    return previousTask;
}

public async Task TheMethod(int id, string val)
{
    //Will be used to signal others that the work has finished
    var completionSource = new TaskCompletionSource<bool>();
    Task previousTask = RegisterAsLastScheduledTask(id, completionSource.Task);

    if (previousTask != null)
        await previousTask;

    // DO LONG RUNNING TASK
    await Task.Delay(3000);
    Console.WriteLine(id + " " + val + " at time " + DateTime.Now.ToString("m:ss"));

    completionSource.SetResult(true);

    //"hack" to atomically remove the element only if both key and value match, see http://blogs.msdn.com/b/pfxteam/archive/2011/04/02/10149222.aspx
    ((ICollection<KeyValuePair<int, Task>>)_runningIds).Remove(new KeyValuePair<int, Task>(id, completionSource.Task));
}
\$\endgroup\$
  • 1
    \$\begingroup\$ Why didn't you come earlier with this review ? It would have saved me from installing a VM with VS2012 ;-) Great job. \$\endgroup\$ – Heslacher Feb 12 '15 at 15:59
  • \$\begingroup\$ Thanks almaz! I'm going to replace my code with your first attempt for the reasons you mentioned and because it took me less time to work out that the 2nd one. \$\endgroup\$ – Steve Feb 13 '15 at 2:10
  • \$\begingroup\$ @Heslacher I appreciate the thought - installing a VM for a post! \$\endgroup\$ – Steve Feb 13 '15 at 2:11

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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