Is there a better way to wait for queued threads?

Question

Is there a better way to wait for queued threads before execute another process?

Currently I'm doing:

this.workerLocker = new object(); // Global variable
this.RunningWorkers = arrayStrings.Length; // Global variable

// Initiate process

foreach (string someString in arrayStrings)
{
     ThreadPool.QueueUserWorkItem(this.DoSomething, someString);
     Thread.Sleep(100);
}

// Waiting execution for all queued threads
lock (this.workerLocker)  // Global variable (object)
{
     while (this.RunningWorkers > 0)
     {
          Monitor.Wait(this.workerLocker);
     }
}

// Do anything else    
Console.WriteLine("END");

---

// Method DoSomething() definition
public void DoSomething(object data)
{
    // Do a slow process...
    .
    .
    .

    lock (this.workerLocker)
    {
        this.RunningWorkers--;
        Monitor.Pulse(this.workerLocker);
    }
}

Answer 1

I'm not sure there is really, I've recently done something similar to scan each IP of a subnet for accepting a particular port.

A couple of things I can suggest that may improve performance:

• Use the SetMaxThreads method of ThreadPool to tweak performance (i.e. balancing having lots of threads running at once, against the locking time on such a large number of threads).

• Don't sleep when setting up all the work items, there is no real need (that I am immediately aware of. Do however sleep inside the DoSomething method, perhaps only for a millisecond, to allow other threads to jump in there if need be.

I'm sure you could implement a more customised method yourself, but I doubt it would be more efficient than using the ThreadPool.

P.S I'm not 100% clear on the reason for using monitor, since you are locking anyway? Note that the question is asked only because I haven't used the Monitor class previously, not because I'm actually doubting it's use.

Answer 2

You likely want to take a look at AutoResetEvent and ManualResetEvent.

These are meant for exactly this situation (waiting for a ThreadPool thread to finish, prior to doing "something").

You'd do something like this:

static void Main(string[] args)
{
    List<ManualResetEvent> resetEvents = new List<ManualResetEvent>();
    foreach (var x in Enumerable.Range(1, WORKER_COUNT))
    {
     ManualResetEvent resetEvent = new ManualResetEvent();
     ThreadPool.QueueUserWorkItem(DoSomething, resetEvent);
        resetEvents.Add(resetEvent);
    }

    // wait for all ManualResetEvents
    WaitHandle.WaitAll(resetEvents.ToArray()); // You probably want to use an array instead of a List, a list was just easier for the example :-)
}

public static void DoSomething(object data)
{
    ManualResetEvent resetEvent = data as ManualResetEvent;

    // Do something

    resetEvent.Set();
}

Edit: Forgot to mention you can wait for a single thread, any thread and so forth as well. Also depending on your situation, AutoResetEvent can simplify things a bit, since it (as the name implies) can signal events automatically :-)

Answer 3

I really like the Begin- End- Async Pattern when I have to wait for the tasks to finish.

I would advice you to wrap the BeginEnd in a worker class:

public class StringWorker
{
    private string m_someString;
    private IAsyncResult m_result;

    private Action DoSomethingDelegate;

    public StringWorker(string someString)
    {
        DoSomethingDelegate = DoSomething;
    }

    private void DoSomething()
    {
        throw new NotImplementedException();
    }

    public IAsyncResult BeginDoSomething()
    {
        if (m_result != null) { throw new InvalidOperationException(); }
        m_result = DoSomethingDelegate.BeginInvoke(null, null);
        return m_result;
    }

    public void EndDoSomething()
    {
        DoSomethingDelegate.EndInvoke(m_result);
    }
}

To do your starting and working use this code snippet:

List<StringWorker> workers = new List<StringWorker>();

foreach (var someString in arrayStrings)
{
    StringWorker worker = new StringWorker(someString);
    worker.BeginDoSomething();
    workers.Add(worker);
}

foreach (var worker in workers)
{
    worker.EndDoSomething();
}

Console.WriteLine("END");

And that's it.

Sidenote: If you want to get a result back from the BeginEnd then change the "Action" to Func and change the EndDoSomething to return a type.

public class StringWorker
{
    private string m_someString;
    private IAsyncResult m_result;

    private Func<string> DoSomethingDelegate;

    public StringWorker(string someString)
    {
        DoSomethingDelegate = DoSomething;
    }

    private string DoSomething()
    {
        throw new NotImplementedException();
    }

    public IAsyncResult BeginDoSomething()
    {
        if (m_result != null) { throw new InvalidOperationException(); }
        m_result = DoSomethingDelegate.BeginInvoke(null, null);
        return m_result;
    }

    public string EndDoSomething()
    {
        return DoSomethingDelegate.EndInvoke(m_result);
    }
}

Answer 4

How about a Fork and Join that uses just Monitor ;-p

Forker p = new Forker();
foreach (var obj in collection)
{
    var tmp = obj;
    p.Fork(delegate { DoSomeWork(tmp); });
}
p.Join();

Full code shown on this earlier answer.

Or for a producer/consumer queue of capped size (thread-safe etc), here.

Answer 5

Yes, there is.

Suggested approach

1) a counter and a wait handle

int ActiveCount = 1; // 1 (!) is important
EventWaitHandle ewhAllDone = new EventWaitHandle(false, ResetMode.Manual);

2) adding loop

foreach (string someString in arrayStrings)
{
     Interlocked.Increment(ref ActiveCount);

     ThreadPool.QueueUserWorkItem(this.DoSomething, someString);
     // Thread.Sleep(100); // you really need this sleep ?
}

PostActionCheck();
ewhAllDone.Wait();

3) DoSomething should look like

{
    try
    {
        // some long executing code
    }
    finally
    {
        // ....
        PostActionCheck();
    }
}

4) where PostActionCheck is

void PostActionCheck()
{
    if (Interlocked.Decrement(ref ActiveCount) == 0)
        ewhAllDone.Set();
}

Idea

ActiveCount is initialized with 1, and then get incremented n times.

PostActionCheck is called n + 1 times. The last one will trigger the event.

The benefit of this solution is that is uses a single kernel object (which is an event), and 2 * n + 1 calls of lightweight API's. (Can there be less ?)

P.S.

I wrote the code here, I might have misspelled some class names.

Answer 6

In addition to Barrier, pointed out by Henk Holterman (BTW his is a very bad usage of Barrier, see my comment to his answer), .NET 4.0 provides whole bunch of other options (to use them in .NET 3.5 you need to download an extra DLL from Microsoft). I blogged a post that lists them all, but my favorite is definitely Parallel.ForEach:

Parallel.ForEach<string>(arrayStrings, someString =>
{
    DoSomething(someString);
});

Behind the scenes, Parallel.ForEach queues to the new and improved thread pool and waits until all threads are done.

Answer 7

Use Spring Threading. It has Barrier implementations built in.

http://www.springsource.org/extensions/se-threading-net