title: How to constraint concurrency the right way in Rx.NET
body: <p>Please, observe the following code snippet:</p>
  
  <pre><code>var result = await GetSource(1000).SelectMany(s =&gt; getResultAsync(s).ToObservable()).ToList();
  </code></pre>
  
  <p>The problem with this code is that <code>getResultAsync</code> runs concurrently in an unconstrained fashion. Which could be not what we want in certain cases. Suppose I want to restrict its concurrency to at most 10 concurrent invocations. What is the Rx.NET way to do it?</p>
  
  <p>I am enclosing a simple console application that demonstrates the subject and my lame solution of the described problem.</p>
  
  <p>There is a bit extra code, like the <code>Stats</code> class and the artificial random sleeps. They are there to ensure I truly get concurrent execution and can reliably compute the max concurrency reached during the process.</p>
  
  <p>The method <code>RunUnconstrained</code> demonstrates the naive, unconstrained run. The method <code>RunConstrained</code> shows my solution, which is not very elegant. Ideally, I would like to ease constraining the concurrency by simply applying a dedicated Rx operator to the Monad. Of course, without sacrificing the performance.</p>
  
  <pre><code>using System;
  using System.Collections.Generic;
  using System.Diagnostics;
  using System.Linq;
  using System.Reactive.Linq;
  using System.Reactive.Threading.Tasks;
  using System.Threading;
  using System.Threading.Tasks;
  
  namespace RxConstrainedConcurrency
  {
      class Program
      {
          public class Stats
          {
              public int MaxConcurrentCount;
              public int CurConcurrentCount;
              public readonly object MaxConcurrentCountGuard = new object();
          }
  
          static void Main()
          {
              RunUnconstrained().GetAwaiter().GetResult();
              RunConstrained().GetAwaiter().GetResult();
          }
          static async Task RunUnconstrained()
          {
              await Run(AsyncOp);
          }
          static async Task RunConstrained()
          {
              using (var sem = new SemaphoreSlim(10))
              {
                  await Run(async (s, pause, stats) =&gt;
                  {
                      // ReSharper disable AccessToDisposedClosure
                      await sem.WaitAsync();
                      try
                      {
                          return await AsyncOp(s, pause, stats);
                      }
                      finally
                      {
                          sem.Release();
                      }
                      // ReSharper restore AccessToDisposedClosure
                  });
              }
          }
          static async Task Run(Func&lt;string, int, Stats, Task&lt;int&gt;&gt; getResultAsync)
          {
              var stats = new Stats();
              var rnd = new Random(0x1234);
              var result = await GetSource(1000).SelectMany(s =&gt; getResultAsync(s, rnd.Next(30), stats).ToObservable()).ToList();
              Debug.Assert(stats.CurConcurrentCount == 0);
              Debug.Assert(result.Count == 1000);
              Debug.Assert(!result.Contains(0));
              Debug.WriteLine("Max concurrency = " + stats.MaxConcurrentCount);
          }
  
          static IObservable&lt;string&gt; GetSource(int count)
          {
              return Enumerable.Range(1, count).Select(i =&gt; i.ToString()).ToObservable();
          }
  
          static Task&lt;int&gt; AsyncOp(string s, int pause, Stats stats)
          {
              return Task.Run(() =&gt;
              {
                  int cur = Interlocked.Increment(ref stats.CurConcurrentCount);
                  if (stats.MaxConcurrentCount &lt; cur)
                  {
                      lock (stats.MaxConcurrentCountGuard)
                      {
                          if (stats.MaxConcurrentCount &lt; cur)
                          {
                              stats.MaxConcurrentCount = cur;
                          }
                      }
                  }
  
                  try
                  {
                      Thread.Sleep(pause);
                      return int.Parse(s);
                  }
                  finally
                  {
                      Interlocked.Decrement(ref stats.CurConcurrentCount);
                  }
              });
          }
      }
  }
  </code></pre>