Inter-Thread Communication Library

• Download demo project - 23.3 Kb
• Download source - 29.3 Kb

Introduction

I would like to start with a simplified version of the so-called multi-threaded pipeline architecture. The idea is simple: we have a pipeline of working threads. When an incoming data stream arrives at the first thread, it implements an operation over that data and forwards the stream to the next thread in the pipeline. Every thread either waits for data or implements specific operations on that data.

That architecture could be used successfully in server-oriented components. It takes off the burden from the threads handling client requests – they forward the requests to be handled by other threads on the pipeline. In that way, they can quickly accept some more requests. By using the ManagerLib assembly, one could easily set-up a multithreaded pipeline.

Using the code

1. Some ManagerLib internals:

   The library consists of four main classes (situated in files with the same names).

   C#

   public abstract class ThreadMessage

   ManagerLib threads exchange data in the form of custom messages. This is the base communication message between threads. All custom messages should be inherited from it.

   C#

   public class ThreadMessages

   ThreadMessages encapsulates a collection of ThreadMessage elements.

   C#

   public class ThreadMessageStore

   ManagerLib threads exchange data via ThreadMessageStore objects. The source thread sends data into particular stores, then the destination thread gets that data from the store (if it has been subscribed to the store). There could be more than one thread subscribed to one store. The concurrent access from multiple threads to the ThreadMessageStore has been organized via the new .NET Automatic Synchronization – expressed with a class’ attribute:

   C#

   [Synchronization(SynchronizationAttribute.REQUIRED)]

   Manual synchronization (using Monitor, Lock ..) could easily replace it.

   The CreateInstance function gives the user the ability to initialize a singleton instance of the ThreadMessageStore class – a feature that I used in the attached demo.

   C#

   public class ThreadClass

   The ThreadClass subscribes to a ThreadMessageStore for receiving ThreadMessages. Then it fires events upon receiving data. By subscribing to those events, the user could get (in his callback function) all the custom messages received from the thread. Then they could be handled and/or forwarded to another thread (via another ThreadMessageStore).

   The ThreadFunction is the actual implementation running from the working thread. It could be overwritten from a ThreadClass descendant if one intends to achieve a different thread’s behavior.

   The ThreadClass also has an ability to safely stop the working thread running the ThreadFunction.

2. How to use the ManagerLib library
   1. Add the library’s reference to your project.
   2. Include the component’s namespace.
      C#

      using nmManager;

   3. Inherit from ThreadMessage to create your own custom data message.
      C#

      public class LogMessage : ThreadMessage
      {
          public string _header;
          public string _detail;
      
          public LogMessage(string header, string detail)
          {
              _header = header;
              _detail = detail;
          }
      
          public override string GetData(string delimiter)
          {
              return _header + delimiter + _detail;
          }
      }

   4. Create a Sink class to receive events from the Thread object.
      C#

      public class SinkClass
      {
          public void OnNewMessages(object sender, 
                  ThreadMessages messages)
          {
              string logLine;
              foreach (LogMessage msg in messages)
              {
                  //one way of getting data
                  logLine = string.Format("{0}, {1}", msg._header, msg._detail);
                  //another way of getting data
                  //string logLine = msg.GetData(",");
                  Console.WriteLine("Recv - th: {0} ({1})", 
                          AppDomain.GetCurrentThreadId() ,logLine);
              }
          }
      }

   5. Instantiate ThreadMessageStore, ThreadClass, and send and receive basic messages.
      C#

      static void Main(string[] args)
      {
          //1.Create ThreadMessage Store instance.
          ThreadMessageStore msgStore = new ThreadMessageStore();
          //2.Instantiate ThreadClass object
          //  and subscribe it to a ThreadMessage Store
          ThreadClass thObj = new ThreadClass(msgStore);
          //3.Create a sink class and subscribe
          //  it to ThreadClass on new message event.
          SinkClass sink1 = new SinkClass();
          thObj._newMessagesEvent += new MessagesEvent(sink1.OnNewMessages);
          //4. Creare and run the actual working thread
          //   - listening for new messages
          //   and fireing on message events.
          Thread th = new Thread(new ThreadStart(thObj.Run));
          th.Start();
          //5. Sending sume data to the Message Store from the main thread.
          for (int i = 0;i < 1000; i++)
          {
              string msg = string.Format("log message - {0}", i);
              msgStore.Add(new LogMessage("header", msg));
              Console.WriteLine("Send - th: {0} ({1})", 
                      AppDomain.GetCurrentThreadId() ,msg);
          }
          //6. Stop the program.
          Thread.Sleep(2000);
          thObj.StopThread(100);
      }

Demo project

Although the above snippet could be used to set up a working example I have attached a bit more complicated demo – a .NET Remoting distributed application.

Basically, we have a remote server and a client. The server accepts one-way requests from the client and logs them, by using ManagerLib, into a log file or a MS SQL Server database. The client generates 500 requests to the server. By running 10 simultaneous clients on the same computer where the server is located (updating the SQL DB), I managed to achieve an average update speed of 350 requests per second (Intel Pentium 2.6 GHz. 512 MB RAM).

Bellow is a schema of how the server handles client requests:

In Remoting, every new client request has been handled by a thread from the .NET thread pool. This thread forms a new ThreadMessage from the request’s data and passes it to the ThreadMessageStore. Three working threads read messages from the store and log them into a log file / the SQL DB.

The remote clients/server connection has been implemented in the Sever Activation/Single Call mode. It also uses the TCP/Binary Serialization. The project uses Administrative Configuration - all the remote configuration settings are set in App.Config files on both the client and the server side. In that way a server’s address and remote modes could be changed without recompiling the clients.

By default, the client is configured to connect to a server located on the same computer – for connecting to different computers, the following line in the file LogClient.exe.config should be changed:

• From url="tcp://localhost:8005/LoggerUri"
• To url="tcp://192.168.1.105:8005/LoggerUri"

The server is set to log all data into a log file. For logging to an MS SQL DB the following should be done:

• Create a table tblLogger - its structure is supplied with the demo files – in the Northwind SQL DB.
• In file Logbook.cs, uncomment all references to _lDBstore and comment all references to _lFilestore.

As this article has different purposes, I am not going to describe .NET Remoting in details here. Some good references are:

• .NET Components by Juval Lowy
• Advanced .NET Remoting by Ingo Rammer

Bellow is the list of the main modules and their dependency diagram:

• LogClient: Remote client.
• Logbook: Remote server - registers and hosts remote objects.
• LoggerLib: Assembly containing remote object declaration.
• IloggerLib: Assembly containing the remote object’s interface and all serialisable objects. IloggerLib and ManagerLib are the only assemblies deployed to the client’s computer.
• ManagerLib: Inter-thread communication library.
• LogStorage: Assembly implementing the Managed Persistence Layer with access to a log file or an MS SQL Server DB.

History

• 15/11/2005 - Initial version.