ThreadPool QueueUserWorkItem with list

Question

Hi

I would like to use the QueueUserWorkItem from the ThreadPool. When I use the following code everything works well.

private int ThreadCountSemaphore = 0;
private void (...) {

var reportingDataList = new List<LBReportingData>();
ThreadCountSemaphore = reportingDataList.Count;
using (var autoResetEvent = new AutoResetEvent(false)) {
   ThreadPool.QueueUserWorkItem((o) => this.FillReportingData(settings, reportingDataList[0], autoResetEvent));
   ThreadPool.QueueUserWorkItem((o) => this.FillReportingData(settings, reportingDataList[1], autoResetEvent));
   ThreadPool.QueueUserWorkItem((o) => this.FillReportingData(settings, reportingDataList[2], autoResetEvent));
}
}

private void FillReportingData(...) {
if (Interlocked.Decrement(ref this.ThreadCountSemaphore) == 0) {
                waitHandle.Set();
                }
}

But when I use a list instead the single method calls, then my program crash without an exception.

private void (...) {

var reportingDataList = new List<LBReportingData>();
ThreadCountSemaphore = reportingDataList.Count;
using (var autoResetEvent = new AutoResetEvent(false)) {
   ThreadPool.QueueUserWorkItem((o) => this.FillReportingData(settings, reportingDataList[i], autoResetEvent));
}
}

What do i wrong? What should I change?

Update

Sorry, I've made a fault in the code. I use .NET 2.0 with VS2010. Here's the complete code:

private int ThreadCountSemaphore = 0;

        private IList<LBReportingData> LoadReportsForBatch() {
            var reportingDataList = new List<LBReportingData>();
            var settings = OnNeedEntitySettings();

            if (settings.Settings.ReportDefinition != null) {
                var definitionList = new List<ReportDefinitionen> { ReportDefinitionen.OrgStatus, ReportDefinitionen.Mittelwerte, ReportDefinitionen.Verteilungsstatistik };
                using (var autoResetEvent = new AutoResetEvent(false)) {
                    foreach (var reportDefinition in definitionList) {
                        foreach (DataRow row in settings.Settings.ReportDefinition.Select("AuswertungsTyp = " + (int)reportDefinition)) {
                            reportingDataList.Add(new LBReportingData { SourceData = row, ReportType = reportDefinition });
                        }
                    }

                    ThreadCountSemaphore = reportingDataList.Count;

                    foreach(var reportingDataItem in reportingDataList) {                                       
                        ThreadPool.QueueUserWorkItem((o) => this.FillReportingData(settings, reportingDataItem, autoResetEvent));
                    }
                    autoResetEvent.WaitOne();
                }
            }
            return reportingDataList;
        }

private void FillReportingData(IEntitySettings<DSLBUReportDefinition> settings, LBReportingData reportingData, AutoResetEvent waitHandle){

            DoSomeWork();
            if (Interlocked.Decrement(ref this.ThreadCountSemaphore) == 0) {
                waitHandle.Set();
            }
        }

Thanks

Answer 1

As Hans pointed out, it is not clear where "i" is coming from. But also I can see your disposing block going out and disposed because you are not using WaitOne on it (or you have not copied that part of code).

Also I would prefer to use WaitAll and not using interlocked.

Answer 2

You are disposing the WaitHandle immediately after queueing the work items. There is race between the call to Dispose in the main thread and Set in the worker thread. There may be other problems, but it is difficult to guess because the code is incomplete.

Here is how the pattern is suppose to work.

using (var finished = new CountdownEvent(1))
{
  for (var item in reportingDataList)
  {  
     var captured = item;
     finished.AddCount(); 
     ThreadPool.QueueUserWorkItem(
       (state) =>
       {
         try
         {
           DoSomeWork(captured); // FillReportingData?
         }
         finally
         {
           finished.Signal();
         }
       });
  } 
  finished.Signal();
  finished.Wait();
}

The code uses the CountdownEvent class. It is available in .NET 4.0 or as part of the Reactive Extensions download.

Answer 3

Sorry, I've made a fault in the code. I use .NET 2.0 with VS2010. Here's the complete code:

private int ThreadCountSemaphore = 0;

        private IList<LBReportingData> LoadReportsForBatch() {
            var reportingDataList = new List<LBReportingData>();
            var settings = OnNeedEntitySettings();

            if (settings.Settings.ReportDefinition != null) {
                var definitionList = new List<ReportDefinitionen> { ReportDefinitionen.OrgStatus, ReportDefinitionen.Mittelwerte, ReportDefinitionen.Verteilungsstatistik };
                using (var autoResetEvent = new AutoResetEvent(false)) {
                    foreach (var reportDefinition in definitionList) {
                        foreach (DataRow row in settings.Settings.ReportDefinition.Select("AuswertungsTyp = " + (int)reportDefinition)) {
                            reportingDataList.Add(new LBReportingData { SourceData = row, ReportType = reportDefinition });
                        }
                    }

                    ThreadCountSemaphore = reportingDataList.Count;

                    foreach(var reportingDataItem in reportingDataList) {                                       
                        ThreadPool.QueueUserWorkItem((o) => this.FillReportingData(settings, reportingDataItem, autoResetEvent));
                    }
                    autoResetEvent.WaitOne();
                }
            }
            return reportingDataList;
        }

private void FillReportingData(IEntitySettings<DSLBUReportDefinition> settings, LBReportingData reportingData, AutoResetEvent waitHandle){

            DoSomeWork();
            if (Interlocked.Decrement(ref this.ThreadCountSemaphore) == 0) {
                waitHandle.Set();
            }
        }

Answer 4

The updated code looks better. It should not dispose of the event prematurely.

Consider what happens if there are no rows - the autoResetEvent.WaitOne() will hang indefinitely.

Just to be on the safe side, also consider putting a Thread.MemoryBarrier() after ThreadCountSemaphore = reportingDataList.Count, so that the store does not move later in "time".

It would look something like this:

private int ThreadCountSemaphore = 0;

private IList<LBReportingData> LoadReportsForBatch()
{
    var reportingDataList = new List<LBReportingData>();
    var settings = OnNeedEntitySettings();

    if (settings.Settings.ReportDefinition != null)
    {
        var definitionList = new List<ReportDefinitionen> { ReportDefinitionen.OrgStatus, ReportDefinitionen.Mittelwerte, ReportDefinitionen.Verteilungsstatistik };
        using (var autoResetEvent = new AutoResetEvent(false))
        {
            foreach (var reportDefinition in definitionList)
            {
                foreach (DataRow row in settings.Settings.ReportDefinition.Select("AuswertungsTyp = " + (int)reportDefinition))
                {
                    reportingDataList.Add(new LBReportingData { SourceData = row, ReportType = reportDefinition });
                }
            }
            ThreadCountSemaphore = reportingDataList.Count;
            //Make sure that the stores in the code above this 
            //(construction of rows, initializing threadcount, etc.)
            //also happen before and do not float below this line.
            Thread.MemoryBarrier();
            foreach (var reportingDataItem in reportingDataList)
            {
                ThreadPool.QueueUserWorkItem((o) => this.FillReportingData(settings, reportingDataItem, autoResetEvent));
            }
            //Do not wait for the event if there was no work at all.
            if (reportingDataList.Count > 0)
            {
                autoResetEvent.WaitOne();
            }
        }
    }
    return reportingDataList;
}

private void FillReportingData(IEntitySettings<DSLBUReportDefinition> settings, LBReportingData reportingData, AutoResetEvent waitHandle)
{
    //DoSomeWork() ought to contain or be wrapped in a try block, 
    //so that the main thread does not wait indefinitely.
    DoSomeWork();
    if (Interlocked.Decrement(ref this.ThreadCountSemaphore) == 0)
    {
        waitHandle.Set();
    }
}