title: ZeroMQ PUB/SUB Pattern with Multi-Threaded Poller Cancellation
body: <p>I have two applications, a C++ server, and a C# WPF UI. The C++ code takes requests (from anywhere/anyone) via a ZeroMQ messaging [PUB/SUB] service. I use my C# code for back testing and to create "back tests" and execute them. These back tests can be made up of many "unit tests" and each of these sending/receiving thousands of messages from the C++ server.</p>
  
  <p>Currently individual back tests work well can send off N unit tests each with thousands of requests and captures. My problem is architecture; when I dispatch another back test (following the first) I get a problem with event subscription being done a second time due to the polling thread not being cancelled and disposed. This results in erroneous output. This may seem like a trivial problem (perhaps it is for some of you), but the cancellation of this polling Task under my current configuration is proving troublesome. Some code...</p>
  
  <p>My message broker class is simple and looks like</p>
  
  <pre><code>public class MessageBroker : IMessageBroker&lt;Taurus.FeedMux&gt;, IDisposable
  {
      private Task pollingTask;
      private NetMQContext context;
      private PublisherSocket pubSocket;
  
      private CancellationTokenSource source;
      private CancellationToken token;
      private ManualResetEvent pollerCancelled;
  
      public MessageBroker()
      {
          this.source = new CancellationTokenSource();
          this.token = source.Token;
  
          StartPolling();
          context = NetMQContext.Create();
          pubSocket = context.CreatePublisherSocket();
          pubSocket.Connect(PublisherAddress);
      }
  
      public void Dispatch(Taurus.FeedMux message)
      {
          pubSocket.Send(message.ToByteArray&lt;Taurus.FeedMux&gt;());
      }
  
      private void StartPolling()
      {
          pollerCancelled = new ManualResetEvent(false);
          pollingTask = Task.Run(() =&gt;
          {
              try
              {
                  using (var context = NetMQContext.Create())
                  using (var subSocket = context.CreateSubscriberSocket())
                  {
                      byte[] buffer = null;
                      subSocket.Options.ReceiveHighWatermark = 1000;
                      subSocket.Connect(SubscriberAddress);
                      subSocket.Subscribe(String.Empty);
                      while (true)
                      {
                          buffer = subSocket.Receive();
                          MessageRecieved.Report(buffer.ToObject&lt;Taurus.FeedMux&gt;());
                          if (this.token.IsCancellationRequested)
                              this.token.ThrowIfCancellationRequested();
                      }
                  }
              }
              catch (OperationCanceledException)
              {
                  pollerCancelled.Set();
              }
          }, this.token);
      }
  
      private void CancelPolling()
      {
          source.Cancel();
          pollerCancelled.WaitOne();
          pollerCancelled.Close();
      }
  
      public IProgress&lt;Taurus.FeedMux&gt; MessageRecieved { get; set; }
      public string PublisherAddress { get { return "tcp://127.X.X.X:6500"; } }
      public string SubscriberAddress { get { return "tcp://127.X.X.X:6501"; } }
  
      private bool disposed = false;
  
      protected virtual void Dispose(bool disposing)
      {
          if (!disposed)
          {
              if (disposing)
              {
                  if (this.pollingTask != null)
                  {
                      CancelPolling();
                      if (this.pollingTask.Status == TaskStatus.RanToCompletion ||
                           this.pollingTask.Status == TaskStatus.Faulted ||
                           this.pollingTask.Status == TaskStatus.Canceled)
                      {
                          this.pollingTask.Dispose();
                          this.pollingTask = null;
                      }
                  }
                  if (this.context != null)
                  {
                      this.context.Dispose();
                      this.context = null;
                  }
                  if (this.pubSocket != null)
                  {
                      this.pubSocket.Dispose();
                      this.pubSocket = null;
                  }
                  if (this.source != null)
                  {
                    this.source.Dispose();
                    this.source = null;
                  }
              }
              disposed = true;
          }
      }
  
      public void Dispose()
      {
          Dispose(true);
          GC.SuppressFinalize(this);
      }
  
      ~MessageBroker()
      {
          Dispose(false);
      }
  }
  </code></pre>
  
  <p>The backtesting "engine" use to execute each back test, first constructs a <code>Dictionary</code> containing each <code>Test</code> (unit test) and the messages to dispatch to the C++ application for each test. </p>
  
  <p>The <code>DispatchTests</code> method, here it is</p>
  
  <pre><code>private void DispatchTests(ConcurrentDictionary&lt;Test, List&lt;Taurus.FeedMux&gt;&gt; feedMuxCollection)
  {
      broker = new MessageBroker();
      broker.MessageRecieved = new Progress&lt;Taurus.FeedMux&gt;(OnMessageRecieved);
      testCompleted = new ManualResetEvent(false);
  
      try
      {
          // Loop through the tests. 
          foreach (var kvp in feedMuxCollection)
          {
              testCompleted.Reset();
              Test t = kvp.Key;
              t.Bets = new List&lt;Taurus.Bet&gt;();
              foreach (Taurus.FeedMux mux in kvp.Value)
              {
                  token.ThrowIfCancellationRequested();
                  broker.Dispatch(mux);
              }
              broker.Dispatch(new Taurus.FeedMux()
              {
                  type = Taurus.FeedMux.Type.PING,
                  ping = new Taurus.Ping() { event_id = t.EventID }
              });
              testCompleted.WaitOne(); // Wait until all messages are received for this test. 
          }
          testCompleted.Close();
      }
      finally
      {
          broker.Dispose(); // Dispose the broker.
      }
  }
  </code></pre>
  
  <p>The <code>PING</code> message at the end, it to tell the C++ that we are finished. We then force a wait, so that the next [unit] test is not dispatched before all of the returns are received from the C++ code - we do this using a <code>ManualResetEvent</code>. </p>
  
  <p>When the C++ receives the PING message, it sends the message straight back. We handle the received messages via <code>OnMessageRecieved</code> and the PING tells us to set the <code>ManualResetEvent.Set()</code> so that we can continue the unit testing; "Next Please"...</p>
  
  <pre><code>private async void OnMessageRecieved(Taurus.FeedMux mux)
  {
      string errorMsg = String.Empty;
      if (mux.type == Taurus.FeedMux.Type.MSG)
      {
          // Do stuff.
      }
      else if (mux.type == Taurus.FeedMux.Type.PING)
      {
          // Do stuff.
  
          // We are finished reciving messages for this "unit test"
          testCompleted.Set(); 
      }
  }
  </code></pre>
  
  <p><s>My problem is that, <code>broker.Dispose()</code> in the finally above is never hit. I appreciate that finally blocks that are executed on background threads are not guaranteed to get executed</s>.</p>
  
  <p>The crossed out text above was due to me messing about with the code; I was stopping a parent thread before the child had completed. However, there are still problems...</p>
  
  <p>Now <code>broker.Dispose()</code> is called correctly, and <code>broker.Dispose()</code> is called, in this method I attempt to cancell the poller thread and dispose of the <code>Task</code> correctly to avoid any multiple subscriptions.</p>
  
  <p>To cancel the thread I use the <code>CancelPolling()</code> method </p>
  
  <pre><code>private void CancelPolling()
  {
      source.Cancel();
      pollerCancelled.WaitOne(); &lt;- Blocks here waiting for cancellation.
      pollerCancelled.Close();
  }
  </code></pre>
  
  <p>but in the <code>StartPolling()</code> method </p>
  
  <pre><code>while (true)
  {
      buffer = subSocket.Receive();
      MessageRecieved.Report(buffer.ToObject&lt;Taurus.FeedMux&gt;());
      if (this.token.IsCancellationRequested)
          this.token.ThrowIfCancellationRequested();
  }
  </code></pre>
  
  <p><code>ThrowIfCancellationRequested()</code> is never called and the thread is never cancelled, thus never properly disposed. The poller thread is being blocked by the <code>subSocket.Receive()</code> method. </p>
  
  <p>Now, it is not clear to me how to achieve what I want, I need to invoke the <code>broker.Dispose()</code>/<code>PollerCancel()</code> on a thread other than that used to poll for messages and some how force the cancellation. Thread abort is not what I want to get into at any cost.</p>
  
  <p>Essentially, I want to properly dispose of the <code>broker</code> before executing the next back test, how do I correctly handle this, split out the polling and run it in a separate Application Domain?</p>
  
  <p>I have tried, disposing inside the <code>OnMessageRecived</code> handler, but this is clearly executed on the same thread as the poller and is not the way to do this, without invoking additional threads, it blocks. </p>
  
  <p><strong>What is the best way to achieve what I want</strong> and <strong>is there a pattern</strong> for this sort of case that I can follow?</p>
  
  <p>Thanks for your time.</p>