Building Responsive .NET Applications with Microsoft Concurrency and Coordination Runtime (CCR)

Ports and Arbiters

Most of how the CCR is configured to execute is housed in the GatherPingRepliesModifyState function in the CCRMediator class. The code appears below:

  private void GatherPingRepliesModifyState()
  {
      Port<PINGCOMPLETEDEVENTARGS> portPing = new Port<PINGCOMPLETEDEVENTARGS>();
      Port<PINGEXCEPTIONINFO> portException = new Port<PINGEXCEPTIONINFO>();
      PingCompletedEventHandler cbfunct = null;
  
      //Get a callback function with the ports wired to it
      cbfunct = AsyncCallbackFactory.Create(portPing,portException);
  
      Arbiter.Activate(_dq, Arbiter.Interleave(
         new TeardownReceiverGroup(),
  
         // Move Receive to here if code touches shared data to ensure thread safety
         new ExclusiveReceiverGroup(
             Arbiter.Receive(true, portPing, PingCompletedCB)
         ),
  
         new ConcurrentReceiverGroup()
         )
      );
  
      Arbiter.Activate(_dqExceptions, Arbiter.Interleave(
          new TeardownReceiverGroup(),
  
          // Move Receive to here if code touches shared data to ensure thread safety
          new ExclusiveReceiverGroup(
              Arbiter.Receive(true, portException,
                  delegate(PingExceptionInfo e)
                  {
                      string resource = e.resource.ToString();
  
                      _pingState.UpdateState(resource, "FAILURE");
  
                      PingReplyWriter.WriteReply(_pingState, resource);
                  }
              )
          ),
  
          new ConcurrentReceiverGroup()
      )
     );
  
    SendPings(cbfunct);
  }

As I stated earlier Ports are generic. Ports are also relatively simple lightweight classes. Later in the article I'll discuss the AsyncCallbackFactory class. For now, it's just important to note that it returns the APM PingCompletedEventFunction delegate with the Exception and Success Port wired to the appropriate response.

Often an executing function will modify some data in the application, earlier I referred to this as shared state. In the sample, I modify the PingState class, which stores the results of the Ping. Normally an application gets a lock on a data structure before modifying it. Instead I used an Interleave. In an Interleave, when code associated with the Port assigned to an ExclusiveReceiverGroup executes, all threads in the Dispatcher are suspended until the execution completes.

The sample also uses two techniques for associating an executing function to the Arbiter. First I use a private function in the CCRMediator class matching the PingCompleteEventFunction signature. Then I use an inline delegate function. A third option would be a lambda (=>) expression. Also important to point out is, when the code executes, it may execute on another thread. Like the ThreadPool, CCR handles all the details seamlessly for you.

Page 3 of 4