Using a Dispatcher

An RTclient dispatcher manages tasks, determining when and in what order they execute. A dispatcher does one or both of these activities:

Types of Dispatchers

Function Name	Purpose
TipcDispatcherCreate	Creates a dispatcher to manage incoming messages on an RTserver connection. The dispatcher can also manage connection, message, socket, timer, and user events.
TipcDispatcherCreateDetached	Creates a dispatcher in a detached thread, which is a thread running on its own in the background. The dispatcher runs continuously until destroyed with TipcDispatcherDestroy.
TipcDispatcherDestroy	Removes a dispatcher created with TipcDispatcherCreate or TipcDispatcherCreateDetached.
TipcDispatcherDispatch	Provides an alternate method of waiting for messages or managing events, as opposed to using TipcDispatcherMainLoop. See How a Dispatcher Executes Events for more information.
TipcDispatcherMainLoop	Controls how long the dispatcher dispatches for messages or manages events. After the defined interval of time elapses, TipcDispatcherMainLoop exits. The dispatcher does not function again until TipcDispatcherMainLoop is called again.
TipcDispatcherSrvAdd	Adds an RTserver connection to a dispatcher.
TipcDispatcherSrvRemove	Removes an RTserver connection from a dispatcher.

See the TIBCO SmartSockets Application Programming Interface for more information on these functions.

Single Threaded RTclients

When a single threaded RTclient has connections to multiple RTservers, using a dispatcher to wait for messages can greatly increase the performance of the RTclient.

What Happens Without a Dispatcher?

Without a dispatcher, an RTclient’s normal procedure of waiting for a message from an RTserver, reading the message, and processing the message, is complicated by the fact that more than one RTserver is sending messages to the RTclient. The wait-read-process cycle is multiplied by the number of RTserver connections, which can easily slow down the RTclient because the wait-read-process steps for each RTserver connection are performed sequentially.

For example, an RTclient has two RTserver connections, one for RTserver A and one for RTserver B. Here is a summary of how some of the wait-read-process steps might decrease the performance of the RTclient:

Except for reading a message, every step offers the potential to block all other processing until it is completed. If 200 messages a second flow into this RTclient from the RTservers, you can see how performance would be slowed down as time spent waiting for and processing a message accumulates.

How a Dispatcher Improves Performance

Using a dispatcher causes the wait-read-process steps to be multiplexed in such a way that long periods of time associated with the wait steps are greatly decreased.

For example, you have an RTclient connected to RTserver A and RTserver B. Here is a summary of how some of the wait-read-process steps might be managed by the dispatcher:

With a dispatcher handling the waiting steps more efficiently, the overall performance of the RTclient is increased. Further increase in performance is possible if you design your RTclient to process each message in the quickest way.

Using a Dispatcher

Adding RTserver connections to a dispatcher and running the dispatcher’s main loop is equivalent to calling the main loop function of all the RTserver connections at the same time.

The code shown in Example 20 and Example 21 result in the same behavior: waiting for a message on one RTserver connection.

 TipcSrvCreate(T_IPC_SRV_CONN_FULL); 
 
 /* Register a default callback to wait for messages. */ 
 TipcSrvDefaultCbCreate(defCbFunc, T_NULL); 
 
 /* Receive messages from the RTserver. */ 
 TipcSrvMainLoop(T_TIMEOUT_FOREVER);

 
 T_IPC_SRV srv; 
 T_IPC_DISPATCHER disp; 
 
 /* Create an RTsever connection. */ 
 TipcSrvCreate(T_IPC_SRV_CONN_FULL); 
 
 /* Register a default callback to wait for messages. */ 
 TipcSrvDefaultCbCreate(defCbFunc, T_NULL); 
 
 /* Create a dispatcher and add the RTserver connection to it. */ 
 disp = TipcDispatcherCreate(); 
 TipcSrvGetSrv(&srv); 
 TipcDispatcherSrvAdd(disp, srv); 
 
 /* Receive messages from the RTserver. */ 
 TipcDispatcherMainLoop(disp, T_TIMEOUT_FOREVER);

The difference between the two examples is that more connections to more RTservers can be added to the dispatcher disp in Example 21 with TipcDispatcherSrvAdd. The dispatcher could also manage various events affecting the RTclient. Waiting for messages from all the RTservers and dealing with events only requires a single main loop, the dispatcher’s.

Events

How a Dispatcher Executes Events

It is the dispatcher’s role to recognize when an event’s execution criteria is met. When the criteria is met, it triggers the dispatcher to execute the callback function associated with the event. A callback function always executes in the thread that dispatches the event. You specify the thread in which you want the callback function executed by passing a dispatcher’s identifier in the TipcEventCreate* function at the time an event is registered. An event’s callback function should not run for a long time because this delays the execution of other events waiting to be dispatched.

For example, a connection event is triggered whenever a connection is available for a read operation. When the dispatcher determines a connection is available for reading, it executes the connection event’s callback function to do the actual work of reading the incoming message from the connection.

The typical way to run a dispatcher involves calling TipcDispatcherMainLoop to run the dispatcher for a specified period of time. During the time period, the dispatcher manages the wait-read-process steps for messages from RTservers, and determines if any of its events were triggered for execution. After the time period elapses, the dispatcher exits from its main loop and does not run until TipcDispatcherMainLoop is called again.

Another way to run a dispatcher is to call TipcDispatcherDispatch. This function also defines a time period for the dispatcher to run, but the dispatcher exits its main loop as soon as it handles one or more events. It only runs for its entire time period if no messages arrive or there are no events to execute.

Connection Events

Connection events are a special method of communicating between SmartSockets connections without routing the messages through an RTserver. This kind of connection is referred to as the peer-to-peer model.

Event Type	Description
Connection	A connection event executes when an RTclient connection is available for reading or writing a message to another RTclient. A connection event is similar to a socket event except the sockets are SmartSockets connections between RTclients. See Connection Events for more information.
Message	A message event executes when an incoming message matches the message type or subject defined for the event. See Message Events for more information.
Socket	Socket events are used when integrating other socket-based software into a dispatcher’s main loop. A socket event executes when a socket is available for sending or receiving a message. See Socket Events for more information.
Timer	A timer event executes at a time interval you define. The event usually repeats but can be executed only once. See Timer Events for more information.
User	A user event is useful for inter-thread communication. By running a dispatcher for each thread, user events can be sent between dispatchers in a manner that does not disrupt a thread’s processing. See User Events for more information.

A connection event is executed by a dispatcher when the connection associated with the event is ready for a read or write operation. When you register a connection event, you must specify which operation triggers the event with a check mode value of T_IO_CHECK_READ or T_IO_CHECK_WRITE.

These steps take place to implement a connection event for read-mode in an RTclient:

These steps take place to implement a connection event for write-mode in an RTclient:

This table summarizes the SmartSockets functions used for SmartSockets connection events:

Function Name	Purpose
TipcEventCreateConn	Registers a SmartSockets connection event with a dispatcher. The event’s execution is triggered when a message can be written to a connection (`T_IO_CHECK_WRITE` mode), or a message can be read from a connection (`T_IO_CHECK_READ` mode).
TipcEventDestroy	Removes a connection event from a dispatcher.
TipcEventGetCheckMode	Returns the triggering mode for a connection event (`T_IO_CHECK_READ` or `T_IO_CHECK_WRITE`).
TipcEventGetConn	Returns the identifier of the connection for a connection event.
TipcEventGetDispatcher	Returns the identifier of the dispatcher where a connection event is registered.
TipcEventGetType	Returns the type of an event. When used for a connection event, the returned value is T_IPC_EVENT_CONN.

See the TIBCO SmartSockets Application Programming Interface for more information on these functions.

Message Events

With message events, the processing of a message can easily be spread across multiple threads based on the message’s type or subject.

When using message events, a message travels from the sending RTclient through the RTserver to the receiving RTclient’s connection or dispatcher. The RTclient’s connection or dispatcher, whichever is waiting for messages, sends the message to the appropriate thread’s dispatcher.

The receiving dispatcher determines if this message matches any of its registered message types or subjects. Upon a match, the dispatcher immediately triggers the execution of a callback function to process the message.

	Messages must not be destroyed within a message event’s callback function. SmartSockets automatically destroys a message once all the events have received it.

Function Name	Purpose
TipcEventCreateMsg	Registers a message event with a dispatcher. The event’s execution is triggered by the subject of the incoming message.
TipcEventCreateMsgType	Registers a message event with a dispatcher. The event’s execution is triggered by the type of incoming message.
TipcEventDestroy	Removes a message event from a dispatcher.
TipcEventGetDispatcher	Returns the identifier of the dispatcher where a message event is registered.
TipcEventGetType	Returns the type of an event. When used for a message event, the returned value is: T_IPC_EVENT_MSG if the message event is triggered by a subject, or T_IPC_EVENT_MSG_TYPE if the message event is triggered by a message’s type

See the TIBCO SmartSockets Application Programming Interface for more information on these functions.

Socket Events

Socket events, similar to connection events, are events that signal a message can be sent or received between the sockets of a client and another vendor’s server. The client can be an RTclient connected to an RTserver and another vendor’s server, or it can be a client only connected to another vendor’s server. In either case, the client uses the SmartSockets dispatcher to handle the execution of the socket event.

A socket event is executed by the dispatcher when the vendor’s socket is ready for a read or write operation. When you register a socket event, you must specify which operation triggers the event with a check mode value of T_IO_CHECK_READ or T_IO_CHECK_WRITE.

Any vendor’s software that supports the ANSI standard for socket architecture can be used in a socket event. However, the socket must be in non-blocking mode to support the asynchronous model of SmartSockets. A write event is triggered when one or more bytes can be written to the vendor’s socket. A read event is triggered for these conditions:

Function Name	Purpose
TipcEventCreateSocket	Registers a socket event with a dispatcher. The event’s execution is triggered when a message can be written to the socket (`T_IO_CHECK_WRITE` mode), or a message can be read from the socket (`T_IO_CHECK_READ` mode).
TipcEventDestroy	Removes a socket event from a dispatcher.
TipcEventGetCheckMode	Returns the triggering mode for a socket event (`T_IO_CHECK_READ` or `T_IO_CHECK_WRITE`).
TipcEventGetDispatcher	Returns the identifier of the dispatcher where a socket event is registered.
TipcEventGetSocket	Returns the identifier of the socket for a socket event.
TipcEventGetType	Returns the type of an event. When used for a socket event, the returned value is T_IPC_EVENT_SOCKET.

See the TIBCO SmartSockets Application Programming Interface for more information on these functions.

Timer Events

Timer events originate within an RTclient. After a defined number of seconds has elapsed, the dispatcher triggers the execution of the timer event. A timer event is executed repeatedly unless it is destroyed within the timer event’s callback function after the first time it executes.

Function Name	Purpose
TipcEventCreateTimer	Registers a timer event with a dispatcher. The event’s execution is triggered every time a set number of seconds, defined by TipcEventSetInterval, has elapsed.
TipcEventDestroy	Removes a timer event from a dispatcher.
TipcEventGetDispatcher	Returns the identifier of the dispatcher where a timer event is registered.
TipcEventGetInterval	Returns the number of seconds set for a timer event.
TipcEventGetType	Returns the type of an event. When used for a timer event, the returned value is T_IPC_EVENT_TIMER.
TipcEventSetInterval	Defines the number of seconds to wait between executing the timer event.

See the TIBCO SmartSockets Application Programming Interface for more information on these functions.

User Events

An RTclient communicates between its threads with user events. A dispatcher always executes a user event immediately when the event becomes the next in line for the dispatcher’s attention. Unlike the other kinds of events, there is no triggering condition to be met before it is executed.

Unlike connection, message, socket, and timer events, which persist until removed with TipcEventDestroy, a user event is temporary. A user event must be registered each time it is required because after the user event executes, it is automatically destroyed.

Function Name	Purpose
TipcEventCreate	Registers a user event with a dispatcher.
TipcEventGetData	The user-defined data associated with the user event when it was registered with the dispatcher.
TipcEventGetDispatcher	Returns the identifier of the dispatcher where a user event was registered.
TipcEventGetType	Returns the type of an event. When used for a user event, the returned value is T_IPC_EVENT_USER.

See the TIBCO SmartSockets Application Programming Interface for more information on these functions.

Multiple Thread Example with Timer and Message Events

 
#include <rtworks/ipc.h> 
 
#define T_MT_CLIENT_REQUEST   101 
#define T_MT_CLIENT_RESPONSE  201 
 
#define T_MT_SERVER_REQUEST   102 
#define T_MT_SERVER_RESPONSE  202 
 
/* -----------------------------------------------------------------------------------*/ 
static void T_ENTRY clientRequestEventTimerFunc 
( 
 T_IPC_EVENT event, 
 T_IPC_EVENT_DATA data, 
 T_PTR arg 
) 
{ 
 /*  -------------------------------------------------------------------------------------------------------------------------------  
   *  This is the callback function invoked whenever the associated timer event is triggered. The timer event sends  
   *   a type T_MT_CLIENT_REQUEST message to RTclient B. A message event in RTclient A processes the     
   *   type T_MT_CLIENT_RESPONSE message returned from RTclient B in response to the  T_MT_CLIENT_ 
   *  REQUEST message. 
   */ ----------------------------------------------------------------------------------------------------------------------------------  
 
  TutOut("%s: CLIENT REQUEST - sending\n", TutGetWallTimeStr()); 
 
  /*  ------------------------------------------  
     * Send the RTclient request message.  
     */ ------------------------------------------  
  TipcSrvMsgWrite("/request", TipcMtLookupByNum(T_MT_CLIENT_REQUEST), 
                  T_TRUE, T_NULL); 
  TipcSrvFlush(); 
 
} /* clientRequestEventTimerFunc */ 
 
/* --------------------------------------------------------------------------------------*/ 
static void T_ENTRY serverRequestEventTimerFunc 
( 
 T_IPC_EVENT event, 
 T_IPC_EVENT_DATA data, 
 T_PTR arg 
) 
{ 
  /*  ------------------------------------------------------------------------------------------------------------------------- 
    *  This is the callback function invoked whenever the  associated timer event is triggered. It sends a type       
    *  T_MT_SERVER_REQUEST message to RTclient B. A message event in RTclient A processes the      
    *  type T_MT_SERVER_RESPONSE message returned from RTclient B in response to the T_MT_SERVER_ 
    *  REQUEST message.   
    */ --------------------------------------------------------------------------------------------------------------------------  
 
  TutOut("%s: SERVER REQUEST - sending\n", TutGetWallTimeStr()); 
 
 /* ----------------------------------------  
   * Send the server request message.   
   */ ----------------------------------------  
  TipcSrvMsgWrite("/request", TipcMtLookupByNum(T_MT_SERVER_REQUEST), 
                  T_TRUE, T_NULL); 
  TipcSrvFlush(); 
 
} /* serverRequestEventTimerFunc */ 
/* -------------------------------------------------------------------------------------------*/ 
static void T_ENTRY clientResponseEventMsgTypeFunc 
( 
 T_IPC_EVENT event, 
 T_IPC_EVENT_DATA data, 
 T_PTR arg 
) 
{ 
  /*  ----------------------------------------------------------------------------------------------------------------------------  
     *   This is the callback function invoked whenever the associated message event is triggered. It processes a type            
     *   T_MT_CLIENT_RESPONSE message from RTclient B.                                                            
     */ ------------------------------------------------------------------------------------------------------------------------------  
 
  TutOut("%s: CLIENT RESPONSE - processing\n", TutGetWallTimeStr()); 
 
} /* clientResponseEventMsgTypeFunc */ 
 
/* ------------------------------------------------------------------*/ 
static void T_ENTRY serverResponseEventMsgTypeFunc 
( 
 T_IPC_EVENT event, 
 T_IPC_EVENT_DATA data, 
 T_PTR arg 
) 
{ 
  /*  -----------------------------------------------------------------------------------------------------------------------  
    *  This is the callback function invoked whenever the associated message event is triggered. It processes a     
    *  type T_MT_SERVER_RESPONSE message from RTclient B.                                                      
    */  ------------------------------------------------------------------------------------------------------------------------  
 
  TutOut("%s: SERVER RESPONSE - processing\n", TutGetWallTimeStr()); 
 
} /* serverResponseEventMsgTypeFunc */ 
 
/* ------------------------------------------------------------------ */ 
int main 
( 
 int argc, 
 char **argv 
) 
{ 
  T_IPC_SRV srv; 
  T_IPC_DISPATCHER main_disp; 
  T_IPC_DISPATCHER client_response_disp; 
  T_IPC_DISPATCHER server_response_disp; 
 
  /*  -------------------------------------------------------  
     *  Make the SmartSockets libraries thread-safe.  
     */ -------------------------------------------------------  
  TipcInitThreads(); 
 
  /*  ------------------------------------------  
     *  Parse the command file if available.  
     */ ------------------------------------------  
  TutCommandParseFile("request.cm"); 
 
  /*  ---------------------------------  
     *  Create the message types.  
     */ ---------------------------------  
  TipcMtCreate("client request", T_MT_CLIENT_REQUEST, "verbose"); 
  TipcMtCreate("client response", T_MT_CLIENT_RESPONSE, "verbose"); 
 
  TipcMtCreate("server request", T_MT_SERVER_REQUEST, "verbose"); 
  TipcMtCreate("server response", T_MT_SERVER_RESPONSE, "verbose"); 
 
  /*  --------------------------------------------  
     *  Create a connection to the RTserver.  
     */  --------------------------------------------  
  TipcSrvCreate(T_IPC_SRV_CONN_FULL); 
 
  /*  ---------------------------------------------------  
    *  Get the connection's T_IPC_SRV object.  
    */  --------------------------------------------------  
  TipcSrvGetSrv(&srv); 
 
  /*  -----------------------  
    *  Create a dispatcher.  
    */  -----------------------  
  main_disp = TipcDispatcherCreate(); 
 
  /*   -----------------------------------------------------------------  
     *   Add the connection for the RTserver to the dispatcher.  
     */  ------------------------------------------------------------------  
  TipcDispatcherSrvAdd(main_disp, srv); 
 
  /*  --------------------------------------------------------------------------------------------  
     *  Add two timer events to the dispatcher, one for each type of request message  
     */ ---------------------------------------------------------------------------------------------  
  TipcEventCreateTimer(main_disp,  
                       1.0,  
                       clientRequestEventTimerFunc,  
                       T_NULL); 
  TipcEventCreateTimer(main_disp,  
                       5.0,  
                       serverRequestEventTimerFunc,  
                       T_NULL); 
 
  /*  ------------------------------------------------------------------------------------------------------------------------------------  
    *  Create two dispatchers, each running in their own thread, to process two types of response messages. 
    *  Response messages are read from the RTserver connection in the main thread and handed off to one of the two  
    *  background threads for processing. This allows for concurrent processing of messages based on message  
    *  type, T_MT_CLIENT_RESPONSE or T_MT_SERVER_RESPONSE. A similar setup could be used  
    *  to processes messages based on subject.                                                                                               
    */  ------------------------------------------------------------------------------------------------------------------------------------  
  client_response_disp = TipcDispatcherCreateDetached(); 
  TipcEventCreateMsgType(client_response_disp, 
                         srv, 
                         TipcMtLookupByNum(T_MT_CLIENT_RESPONSE), 
                         clientResponseEventMsgTypeFunc, 
                         T_NULL); 
 
  server_response_disp = TipcDispatcherCreateDetached(); 
  TipcEventCreateMsgType(server_response_disp, 
                         srv, 
                         TipcMtLookupByNum(T_MT_SERVER_RESPONSE), 
                         serverResponseEventMsgTypeFunc, 
                         T_NULL); 
 
  /*   --------------------------------------------------------------------------------  
     *   Dispatch the timer events and wait for messages from the RTserver. 
     */  ---------------------------------------------------------------------------------  
  TipcDispatcherMainLoop(main_disp, T_TIMEOUT_FOREVER); 
 
  TutExit(T_EXIT_SUCCESS); 
} /* main */

 
#include <rtworks/ipc.h> 
 
#define T_MT_CLIENT_REQUEST   101 
#define T_MT_CLIENT_RESPONSE  201 
 
#define T_MT_SERVER_REQUEST   102 
#define T_MT_SERVER_RESPONSE  202 
 
/* ------------------------------------------------------------------ */ 
static void T_ENTRY clientRequestCbFunc 
( 
 T_IPC_CONN conn, 
 T_IPC_CONN_PROCESS_CB_DATA data, 
 T_CB_ARG arg 
) 
{ 
  T_STR sender; 
 
  TipcMsgGetSender(data->msg, &sender); 
 
  TutOut("%s: CLIENT REQUEST - received\n", TutGetWallTimeStr()); 
 
  /* ----------------------------------------------------------------------------------------------  
     * Respond to the RTclient request message with an RTclient response message.   
     */  ---------------------------------------------------------------------------------------------  
  TipcSrvMsgWrite(sender, TipcMtLookupByNum(T_MT_CLIENT_RESPONSE), 
                  T_TRUE, T_NULL); 
  TipcSrvFlush(); 
} /* clientRequestCbFunc */ 
 
/* ------------------------------------------------------------ */ 
static void T_ENTRY serverRequestCbFunc 
( 
 T_IPC_CONN conn, 
 T_IPC_CONN_PROCESS_CB_DATA data, 
 T_CB_ARG arg 
) 
{ 
  T_STR sender; 
 
  TipcMsgGetSender(data->msg, &sender); 
 
  TutOut("%s: SERVER REQUEST - received\n", TutGetWallTimeStr()); 
 
  /* ----------------------------------------------------------------------------------------------  
    * Respond to the RTserver request message with an RTserver response message.  
   */ -----------------------------------------------------------------------------------------------  
  TipcSrvMsgWrite(sender, TipcMtLookupByNum(T_MT_SERVER_RESPONSE), 
                  T_TRUE, T_NULL); 
  TipcSrvFlush(); 
} /* serverRequestCbFunc */ 
/* ------------------------------------------------------------------ */ 
int main 
( 
 int argc, 
 char **argv 
) 
{ 
  /* -------------------------------------------  
    * Parse the command file if available.  
   */ --------------------------------------------  
  TutCommandParseFile("response.cm"); 
 
  /* --------------------------------  
    * Create the message types.    
    */ ---------------------------------  
  TipcMtCreate("client request", T_MT_CLIENT_REQUEST, "verbose"); 
  TipcMtCreate("client response", T_MT_CLIENT_RESPONSE, "verbose"); 
 
  TipcMtCreate("server request", T_MT_SERVER_REQUEST, "verbose"); 
  TipcMtCreate("server response", T_MT_SERVER_RESPONSE, "verbose"); 
 
  /*  --------------------------------------------  
     *  Create a connection to the RTserver.  
     */ --------------------------------------------  
  TipcSrvCreate(T_IPC_SRV_CONN_FULL); 
 
  /*  --------------------------------------  
     * Subscribe to /request messages.  
     */ --------------------------------------  
  TipcSrvSubjectSetSubscribe("/request", T_TRUE); 
  TipcSrvFlush(); 
 
  /*  -------------------------------------------------------------------  
     *  Add two callbacks, one for each type of request messages.  
     */ -------------------------------------------------------------------  
  TipcSrvProcessCbCreate(TipcMtLookupByNum(T_MT_CLIENT_REQUEST), 
                         clientRequestCbFunc, 
                         T_NULL); 
  TipcSrvProcessCbCreate(TipcMtLookupByNum(T_MT_SERVER_REQUEST), 
                         serverRequestCbFunc, 
                         T_NULL); 
 
  /* --------------------------------------------  
    * Process messages from the RTserver.  
    */ --------------------------------------------  
  TipcSrvMainLoop(T_TIMEOUT_FOREVER); 
 
  TutExit(T_EXIT_SUCCESS); 
} /* main */