Parallel.ForEach: Best way to save off a collection when its record count gets high?

Question

So I'm running a Parallel.ForEach that basically generates a bunch of data which is ultimately going to be saved to a database. However, since collection of data can get quite large I need to be able to occasionally save/clear the collection so as to not run into an OutOfMemoryException.

I'm new to using Parallel.ForEach, concurrent collections, and locks, so I'm a little fuzzy on what exactly needs to be done to make sure everything works correctly (i.e. we don't get any records added to the collection between the Save and Clear operations).

Currently I'm saying, if the record count is above a certain threshold, save the data in the current collection, within a lock block.

ConcurrentStack<OutRecord> OutRecs = new ConcurrentStack<OutRecord>();
object StackLock = new object();

Parallel.ForEach(inputrecords, input =>
{
  lock(StackLock)
  {  
    if (OutRecs.Count >= 50000)
    {
       Save(OutRecs);
       OutRecs.Clear();
     }
   }

  OutRecs.Push(CreateOutputRecord(input);
});

if (OutRecs.Count > 0) Save(OutRecs);

I'm not 100% certain whether or not this works the way I think it does. Does the lock stop other instances of the loop from writing to output collection? If not is there a better way to do this?

Answer 1

Your lock will work correctly but it will not be very efficient because all your worker threads will be forced to pause for the entire duration of each save operation. Also, locks tends to be (relatively) expensive, so performing a lock in each iteration of each thread is a bit wasteful.

One of your comments mentioned giving each worker thread its own data storage: yes, you can do this. Here's an example that you could tailor to your needs:

Parallel.ForEach(
    // collection of objects to iterate over
    inputrecords,

    // delegate to initialize thread-local data
    () => new List<OutRecord>(),

    // body of loop
    (inputrecord, loopstate, localstorage) =>
    {
        localstorage.Add(CreateOutputRecord(inputrecord));
        if (localstorage.Count > 1000)
        {
            // Save() must be thread-safe, or you'll need to wrap it in a lock
            Save(localstorage);
            localstorage.Clear();
        }
        return localstorage;
    },

    // finally block gets executed after each thread exits
    localstorage =>
    {
        if (localstorage.Count > 0)
        {
            // Save() must be thread-safe, or you'll need to wrap it in a lock
            Save(localstorage);
            localstorage.Clear();
        }
    });