title: Parallelization of CPU bound task continuing with IO bound
body: <p>I'm trying to figure out a good way to do parallelization of code that does processing of big datasets and then imports the resulting data into RavenDb.</p>
  
  <p>The data processing is CPU bound and database import IO bound.</p>
  
  <p>I'm looking for a solution to do the processing in parallel on Environment.ProcessorCount number of threads. The resulting data should then be imported into RavenDb on x (lets say 10) pooled threads in parallel with the above process.</p>
  
  <p>The main thing here is I want the processing to continue while completed data is being imported so that processing the next dataset continues while waiting for the import to complete.</p>
  
  <p>Another issue is the memory for each batch needs to be discarded after a successful import as the private working memory can easily reach >5GB.</p>
  
  <p>The code below is what I've got so far. Do note that it does not fullfill the parallelization requirements outlined above.</p>
  
  <pre><code>datasupplier.GetDataItems()
      .Partition(batchSize)
      .AsParallel()
      .WithDegreeOfParallelism(Environment.ProcessorCount)
      .ForAll(batch =&gt;
      {
          Task.Run(() =&gt;
          {
              ...
          }
      }
  </code></pre>
  
  <p>GetDataItem yields enumerable data items that are partitioned into a batch dataset. GetDataItem will yield ~2,000,000 items each averaging around 0.3ms for processing.</p>
  
  <p>The project is running on the latest .NET 4.5 RC on a x64 platform.</p>
  
  <p><strong>Update.</strong></p>
  
  <p>My current code (seen above) will fetch items and partition them in batches. Each batch is processed in parallel on eight threads (Environment.ProcessorCount on i7). The processing is slow, cpu-bound and memory intensive.
  When processing of a single batch is complete, a task is started to asynchronously import the resulting data into RavenDb. The batch import job is itself synchronous and looks like:</p>
  
  <pre><code>using (var session = Store.OpenSession())
  {
      foreach (var data in batch)
      {
          session.Store(data);
      }
      session.SaveChanges();
  }
  </code></pre>
  
  <p>There are a few problems with this approach:</p>
  
  <ol>
  <li><p>Every time a batch is completed a task is started to run the import job. I want to limit the number of tasks that run in parallel (eg. max 10). Additionally even though many tasks are started they seem to never run in parallel.</p></li>
  <li><p>Memory allocations are a huge problem. Once a batch is processed/imported it seems to still remain in memory.</p></li>
  </ol>
  
  <p>I'm looking for ways to take care of the issues outlined above. Ideally I want:</p>
  
  <ul>
  <li>One thread per logical processor doing heavy lifting processing batches of data.</li>
  <li>Ten or so parallel threads waiting for completed batches to import into RavenDb.</li>
  <li>To keep memory allocations to a minimum which means unallocating a batch after the import task is complete.</li>
  <li>To not run import jobs on one of the threads for batch processing. Import of completed batches should run in parallel to the next batch being processed.</li>
  </ul>
  
  <p><strong>Solution</strong></p>
  
  <pre><code>var batchSize = 10000;
  var bc = new BlockingCollection&lt;List&lt;Data&gt;&gt;();
  var importTask = Task.Run(() =&gt;
  {
      bc.GetConsumingEnumerable()
          .AsParallel()
          .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
          .WithMergeOptions(ParallelMergeOptions.NotBuffered)
          .ForAll(batch =&gt;
          {
              using (var session = Store.OpenSession())
              {
                  foreach (var i in batch) session.Store(i);
                  session.SaveChanges();
              }
          });
  });
  var processTask = Task.Run(() =&gt;
  {
      datasupplier.GetDataItems()
          .Partition(batchSize)
          .AsParallel()
          .WithDegreeOfParallelism(Environment.ProcessorCount)
          .ForAll(batch =&gt;
          {
              bc.Add(batch.Select(i =&gt; new Data()
              {
                  ...
              }).ToList());
          });
  });
  
  processTask.Wait();
  bc.CompleteAdding();
  importTask.Wait();
  </code></pre>