title: How to do async operations in a TPL Dataflow for best performance?
body: <p>I wrote the following method to batch process a huge CSV file. The idea is to read a chunk of lines from the file into memory, then partition these chunk of lines into batches of fixed size. Once we get the partitions, send these partitions to a server (sync or async) which might take a while.</p>
  
  <pre><code>private static void BatchProcess(string filePath, int chunkSize, int batchSize)
  {
      List&lt;string&gt; chunk = new List&lt;string&gt;(chunkSize);
  
      foreach (var line in File.ReadLines(filePath))
      {
          if (chunk.Count == chunk.Capacity)
          {
              // Partition each chunk into smaller chunks grouped on column 1
              var partitions = chunk.GroupBy(c =&gt; c.Split(',')[0], (key, g) =&gt; g);
  
              // Further breakdown the chunks into batch size groups
              var groups = partitions.Select(x =&gt; x.Select((i, index) =&gt;
                  new { i, index }).GroupBy(g =&gt; g.index / batchSize, e =&gt; e.i));
  
              // Get batches from groups
              var batches = groups.SelectMany(x =&gt; x)
                  .Select(y =&gt; y.Select(z =&gt; z)).ToList();
  
              // Process all batches asynchronously
              batches.AsParallel().ForAll(async b =&gt;
              {
                  WebClient client = new WebClient();
                  byte[] bytes = System.Text.Encoding.ASCII
                      .GetBytes(b.SelectMany(x =&gt; x).ToString());
                  await client.UploadDataTaskAsync("myserver.com", bytes);
              });
  
              // clear the chunk
              chunk.Clear();
          }
  
          chunk.Add(line);
      }
  }
  </code></pre>
  
  <p>This piece of code doesn't seem to be very efficient because of 2 reasons.</p>
  
  <ol>
  <li><p>The main thread that reads from the CSV file is blocked, till all the partitions are processed.</p></li>
  <li><p>The AsParallel blocks till all the tasks are finished. So if there are more threads available in the thread pool to do work, I'm not using them coz the no of tasks are bound by no of partitions. </p></li>
  </ol>
  
  <p>The batchSize is fixed so can't be changed but chunkSize is tuneable for performance. I can choose a large enough chunkSize, such that no of batches created are >> no of threads available in the system, but it still means that the Parallel.ForEach method blocks till all the tasks are finished.</p>
  
  <p>How can I change the code such that all available threads in the system are utilized to do the work w/o sitting idle. I'm thinking I could use a BlockingCollection to store the batches but not sure what capacity size to give it since no of batches are dynamic in each chunk. </p>
  
  <p>Any ideas on how to use TPL to maximize thread utilization so that most available threads on the system are always doing stuff?</p>
  
  <p><strong>UPDATE:</strong>
  This is what I got so far using TPL dataflow. Is this correct?</p>
  
  <pre><code>private static void UploadData(string filePath, int chunkSize, int batchSize)
  {
      var buffer1 = new BatchBlock&lt;string&gt;(chunkSize);
      var buffer2 = new BufferBlock&lt;IEnumerable&lt;string&gt;&gt;();
  
      var action1 = new ActionBlock&lt;string[]&gt;(t =&gt;
      {
          Console.WriteLine("Got a chunk of lines " + t.Count());
  
          // Partition each chunk into smaller chunks grouped on column 1
          var partitions = t.GroupBy(c =&gt; c.Split(',')[0], (key, g) =&gt; g);
  
          // Further breakdown the chunks into batch size groups
          var groups = partitions.Select(x =&gt; x.Select((i, index) =&gt;
              new { i, index }).GroupBy(g =&gt; g.index / batchSize, e =&gt; e.i));
  
          // Get batches from groups
          var batches = groups.SelectMany(x =&gt; x).Select(y =&gt; y.Select(z =&gt; z));
  
          foreach (var batch in batches)
          {
              buffer2.Post(batch);
          }
  
      }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 });
  
      buffer1.LinkTo(action1, new DataflowLinkOptions
          { PropagateCompletion = true });
  
      var action2 = new TransformBlock&lt;IEnumerable&lt;string&gt;,
          IEnumerable&lt;string&gt;&gt;(async b =&gt;
      {
          await ExecuteBatch(b);
          return b;
  
      }, new ExecutionDataflowBlockOptions
          { MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded });
  
      buffer2.LinkTo(action2, new DataflowLinkOptions
          { PropagateCompletion = true });
  
      var action3 = new ActionBlock&lt;IEnumerable&lt;string&gt;&gt;(b =&gt;
      {
          Console.WriteLine("Finised executing a batch");
      }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 });
  
      action2.LinkTo(action3, new DataflowLinkOptions
          { PropagateCompletion = true });
  
      Task produceTask = Task.Factory.StartNew(() =&gt;
      {
          foreach (var line in File.ReadLines(filePath))
          {
              buffer1.Post(line);
          }
  
          //Once marked complete your entire data flow will signal a stop for
          // all new items
          Console.WriteLine("Finished producing");
          buffer1.Complete();
      });
  
      Task.WaitAll(produceTask);
      Console.WriteLine("Produced complete");
  
      action1.Completion.Wait();//Will add all the items to buffer2
      Console.WriteLine("Action1 complete");
  
      buffer2.Complete();//will not get any new items
      action2.Completion.Wait();//Process the batch of 5 and then complete
  
      Task.Wait(action3.Completion);
  
      Console.WriteLine("Process complete");
      Console.ReadLine();
  }
  </code></pre>