Block concurrent access ThreadPool

Question

So I'm running a query and processing the rows that return concurrently using a function called StartJob that will work on my job:

ThreadPool.QueueUserWorkItem(StartJob, job);

Works great and is very fast. But now I'm told that when the query returns, some rows may have the same value for job.UserID and that we can't be running the StartJob function concurrently for identical values of job.UserID. The question is: how can I have StartJob block execution until any other instances of StartJob with the same UserID have completed?

I'm sure there's a some way to obtain a per-UserID lock, but I don't know how to do it. Thanks for the help.

Answer 1

HashSet<int> hs = new HashSet<int>(); // In common with all the threads

int id = 1; // Your id

// This is the body of your Thread. You pass it the id as you want. 
// A closure on it, or as a parameter of the thread.

// This will begin with short spins, every time trying to add the id to the hashset.
// SpinUntil stops when the lambda function returns true.
SpinWait.SpinUntil(() =>
{
    lock (cd)
    {
        return hs.Add(id);
    }
});

// OR, if you know the operation is slow, or < .NET 4.0

// This is clearer. The thread yields until it can add the id to the hashset.
while (true)
{
    lock (hs)
    {
        if (hs.Add(id))
        {
            break;
        }
    }

    Thread.Yield();
}

// End of the variant

// Remember the try/finally! It's important in case of exceptions!!!
try
{
    // Put here your code
    // Put here your code
    // Put here your code
}
finally
{
    lock (hs)
    {
        hs.Remove(id);
    }
}

Two versions, one that is good for short StartJob and works only on .NET 4.0, one that works with .NET >= 3.5.

Clearly hs is in common between all the threads, and id is the job.UserID.

I'll add that under .NET 4.0 you could use SpinLock instead of lock. It is a little faster, but its syntax is a little tricky.

Answer 2

use task parallel library

var tasks = new Dictionary<int, Task>();

QueueJob(Job job)
{
    lock(tasks)
      if (tasks.ContainsKey(job.UserID))
      {
         var newTask = tasks[job.UserID].ContinueWith(_=>StartJob(job));
         tasks[job.UserID] = newTask;
      }
      else
            tasks[job.UserID] = Task.Factory.StartNew(()=>StartJob(job));                  
}