Using the Real-Time Event Service

Chris Gill, Tim Harrison, and Carlos O'Ryan

cdgill@cs.wustl.edu, harrison@cs.wustl.edu, and coryan@cs.wustl.edu




Overview



Contents


  1. Overview
  2. The COS Event Model
  3. Real-Time Event Service Enhancements
  4. The Real-Time Event Service
  5. Developing Consumers
  6. Developing Suppliers
  7. Caring For Your Event Channel
  8. Sample Applications
  9. Reference Materials


Introduction


This material is intended to provide an introduction to the COS Event Model, the Real-Time extensions to that model in TAO, and a set of examples that illustrate the techniques used to build systems using these models. The CORBA Event Service provides a flexible model for asynchronous communication among objects. However, the standard CORBAEvent Service specification lacks important features required by real-time applications. These features include event filtering, event correlation, and periodic event processing.

The standard CORBA operation invocation model supports twoway, oneway, and deferred synchronous interactions between clients and servers. The primary strength of the twoway model is its intuitive mapping onto the object->operation() paradigm supported by OO languages. In principle, twoway invocations simplify the development of distributed applications by supporting an implicit request/response protocol that makes remote operation invocations transparent to the client.

In practice, however, the standard CORBA operation invocation models are too restrictive for real-time applications. In particular, these models lack asynchronous message delivery, do not support timed invocations or group communication, and can lead to excessive polling by clients. Moreover, standard oneway invocations might not implement reliable delivery and deferred synchronous invocations require the use of the CORBA Dynamic Invocation Interface (DII), which yields excessive overhead for most real-time applications.

The Event Service is a CORBA Object Service (COS) that is designed to alleviate some of the restrictions with standard CORBA invocation models. In particular, the COS Event Service supports asynchronous message delivery and allows one or more suppliers to send messages to one or more consumers. Event data can be delivered from suppliers to consumers without requiring these participants to know about each other explicitly.

There are two models (i.e., push vs. pull) of participant collaborations in the COS Event Service architecture. This material focuses on real-time enhancements to the push model, which allows suppliers of events to initiate the transfer of event data to consumers. Suppliers push events to the Event Channel, which in turn pushes the events to consumers.

Suppliers use Event Channels to push data to consumers. Likewise, consumers can explicitly pull data from suppliers. The push and pull semantics of event propagation help to free consumers and suppliers from the overly restrictive synchronous semantics of the standard CORBA twoway communication model. In addition, Event Channels can implement group communication by serving as a replicator, broadcaster, or multicaster that forward events from one or more suppliers to multiple consumers.


The COS Event Model

Relationships Between Modules

The role of each component in the COS Event Model is outlined below:




The Push Model

  • Consumers - Ultimate targets of events
  • Suppliers - Generate events
  • Event Channel - Decouple consumers and suppliers by propagating events to consumers on behalf of suppliers


Real-Time Event Service Enhancements

  • Prioritized dispatching within preemption classes -- The current implementation can dispatch events in the same queue by their order of importance, which is necessary to support priorities within a rate group.

  • Suspend/resume -- If a consumer's event dependencies change at run-time, it can utilize the suspend/resume functionality through the new ProxyPushSupplier::suspend and ProxyPushSupplier::resume methods. When a consumer invokes ProxyPushSupplier::suspend, the dependencies registered with that proxy will be disabled until the resume methods is called. These enhancements allow run-time flexibility of event dependencies, but maintains the determinism required by the system scheduling policy (i.e., consumers can not add and remove dependencies at run-time - just suspend and resume them).

  • Event data model -- The data model may use unions, untyped buffers, or type anys.

  • Event filtering -- Consumers may register for events based on event type, or supplier id. The event channel filters events based on these registrations, to ensure efficient event delivery.

  • Event correlation -- Consumers may register for event delivery based on conjunctive or disjunctive sets of events. Conjunctive registrations cause the event channel to notify the consumer when all events in the set have arrived. Disjunctive registrations cause the event channel to notify the consumer when any event in the set has arrived.

  • Periodic event processing -- Consumers may register for suppliers based on timed events. Periodic suppliers push events into the channel at well defined intervals.

  • Active consumers and suppliers -- See The Real-Time Event Service

    .


The Real-Time Event Service

Real-Time ORB and Services


  • Real-time event dispatching

    • Priority-based queueing and preemption mechanisms

  • Centralized event filtering and correlation

    • Source/type-based filtering

    • Conjunction/Disjunction (AND/OR) correlations

  • Periodic and Aperiodic processing

    • Canonical timeouts

    • Dependency timeouts


Real-Time Event Service Internals




Characteristics of Real-Time Push Event Service Participants

Two major roles are played by the participants in a real-time push event service.

The first role is that of an event consumer, which receives events from the event channel. A consumer specifies the type and/or source id for each event it is interested in receiving.

In hard real-time applications, a consumer must also specify RT_Info data for each event it is interested in receiving, and any other events on which that event depends. The RT_Info structure resides in the Scheduler, and is accessed through the scheduler interface. A unique handle is returned to the consumer when an RT_Info is created, which can then be used to set the information in the RT_Info. An RT_Info handle may also be obtained via the Scheduler's lookup method.

The second role is that of an event supplier, which generates events and passes them into the event channel. A supplier must specify its source id, and the type of each event it will generate.
In hard real-time applications, a supplier must also specify RT_Info data for the events it will generate. In particular, it must specify the maximum rate at which it will generate each event. This information is used by a real-time scheduler to assign appropriate dispatch priorities.

Note that the event channel may also be configured to use a null scheduling service. This will cause all operations to be dispatched at the same priority, and will not require the application to specify worst case execution times, periods, etc.
The consumer and supplier roles may be combined, as illustrated in the tables below. There are two main cases in which the roles are combined: a passive one termed Consumer/Supplier which borrows a thread of execution to produce events, and an active one termed Supplier/Consumer which produces events in its own thread. Both consume events and produce events.



EC Roles and Specified RT_Info Contents

EC Roles RT_Info Contents Domain Examples
Consumer dependencies (and optionally, importance) Display, Exception & Maintenance Logs
Consumer/Supplier dependencies (and optionally, importance) Navigation Component (NAV)
Supplier/Consumer rate, dependencies (and optionally, importance) Kalman Filter
Supplier rate Operator Control Panel, EC Reactor Threads


EC Roles and Scheduler Dependency Chain

EC Roles Scheduler Dependency Chain
Pure Consumer root node
Consumer/Supplier internal node
Supplier/Consumer internal node
Pure Supplier leaf node


EC Roles, Threading, and CORBA Roles

EC Roles Activity Thread Behavior CORBA Roles
Pure Consumer Passive Threads optional, "internal", wait for an event to occur Servant
Consumer/Supplier Passive Threads optional, "internal", wait for an event to occur Client and/or Servant
Supplier/Consumer Active Threads required and visible to EC: consume events and actively produce other events Client and/or Servant
Pure Supplier Active Threads required and visible to EC: actively produce events Client







Developing Consumers



Providing QoS Information

The following steps are only necessary for applications that make use of the Event Service's hard real-time features. Applications that do not need these features and are configured with a null scheduler may skip the following operations on the scheduling server.
For each operation, a Consumer should provide the worst case, expected, and cached execution time for that operation. It must also specify criticality and importance values for each operation. A real-time scheduler uses this information to order dispatches within a set of operations whose dependencies have been met.
If it is a Consumer/Supplier (one which consumes an event and passively generates one or more events from the thread in which it was called, as illustrated in the tables above), it must provide dependencies on one or more other events to the scheduler.
If it is a Supplier/Consumer (one which consumes an event and actively generates one or more events from its own thread, as illustrated in the tables above), it must also specify the rate at which it will generate the new events by passing a positive value in the period argument to the scheduler set method. It may also indicate a positive number of threads in which the dispatch will be made. If the number of threads given is zero, but a period is specified, the number of threads defaults to 1.

  // Obtain a reference to the scheduler server.
  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  // Create new RT_Info descriptors for three events.

  RtecScheduler::handle_t handle1 = 
    server->create ("event_1",              // Name of entry point
                    TAO_TRY_ENV             // Environment
                   );
      
  RtecScheduler::handle_t handle2 = 
    server->create ("event_2",              // Name of entry point
                    TAO_TRY_ENV             // Environment
                   );

  RtecScheduler::handle_t handle3 = 
    server->create ("event_3",              // Name of entry point
                    TAO_TRY_ENV             // Environment
                   );


  // Register as a consumer/supplier: act as a supplier of event_1 but with
  // a consumer dependency on event_3.  Therefore, the actual period and 
  // number of threads for event_1 depends on the characteristics of event_3.
  server->set (handle1,                     // RT_Info handle
               RtecScheduler::HIGH_CRITICALITY,   // Criticality
               500,                         // Worst case time (in 100 nanosecs)
               500,                         // Typical time (in 100 nanosecs)
               500,                         // Cached time (in 100 nanosecs)
               0,                           // Period - will depend on event_3
               RtecScheduler::LOW_IMPORTANCE,     // Importance
               0,                           // Quantum (unused)
               0,                           // Threads - will depend on event_3
               RtecScheduler::OPERATION,          // Info type
               TAO_TRY_ENV);

  // Register as a producer of event_2.
  server->set (handle2,                     // RT_Info handle
               RtecScheduler::HIGH_CRITICALITY,   // Criticality
               500,                         // Worst case time (in 100 nanosecs)
               500,                         // Typical time (in 100 nanosecs)
               500,                         // Cached time (in 100 nanosecs)
               50000 * 10,                  // Period in 100 nsec (= 20 Hz)
               RtecScheduler::LOW_IMPORTANCE,     // Importance
               0,                           // Quantum (unused)
               1,                           // Threads
               RtecScheduler::OPERATION,          // Info type
               TAO_TRY_ENV);

  // Register as a consumer of event_3.
  server->set (handle3,                     // RT_Info handle
               RtecScheduler::HIGH_CRITICALITY,   // Criticality
               500,                         // Worst case time (in 100 nanosecs)
               500,                         // Typical time (in 100 nanosecs)
               500,                         // Cached time (in 100 nanosecs)
               0,                           // Period - will depend on supplier
               RtecScheduler::LOW_IMPORTANCE,     // Importance
               0,                           // Quantum (unused)
               0,                           // Threads - will depend on supplier
               RtecScheduler::OPERATION,          // Info type
               TAO_TRY_ENV);


  // Establish a dependency of event_1 on event_3.
  server->add_dependency (handle1,          // handle that depends
                          handle3,          // handle that is depended on
                          1,                // number of calls per event occurance
                          TAO_TRY_ENV       // environment
                         );



Connecting Consumers to the Event Channel

The following code is derived from the EC_Throughput consumer code, which can be found in TAO in the file: $TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp

void
Test_Consumer::connect (const char* name,
                        int event_a, int event_b,
                        RtecEventChannelAdmin::EventChannel_ptr ec,
                        CORBA::Environment& _env)
{

  // Register operations with the scheduling service.  The following steps are
  // only necessary for applications that make use of the Event Service's hard
  // real-time features.  Applications that do not need these features and are
  // configured with a null scheduler may skip the following operations on the
  // scheduling server.

  // Obtain a reference to the scheduler from the ACE_Scheduler_Factory. 
  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  // Create a new RT_Info entry for the function identifier
  // we were passed, and hang onto the handle to the RT_Info.
  RtecScheduler::handle_t rt_info =
    server->create (name, _env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Set the attributes for the RT_Info.  
  ACE_Time_Value tv (0, 2000);
  TimeBase::TimeT time;
  ORBSVCS_Time::Time_Value_to_TimeT (time, tv);
  server->set (rt_info,
               RtecScheduler::VERY_HIGH_CRITICALITY,
               time, time, time,
               0,
               RtecScheduler::VERY_LOW_IMPORTANCE,
               time,
               0,
               RtecScheduler::OPERATION,
               _env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Specify a disjunctive dependency on the arrival of event_a, the arrival
  // of event b, OR the arrival of an event service shutdown event.  Note that 
  // the same RT_Info is used for each event.  This can be used to simplify
  // code in applications using a null scheduler, or to consolidate events
  // with identical characteristics in hard real-time applications.
  ACE_ConsumerQOS_Factory qos;
  qos.start_disjunction_group ();
  qos.insert_type (ACE_ES_EVENT_SHUTDOWN, rt_info);
  qos.insert_type (event_a, rt_info);
  qos.insert_type (event_b, rt_info);

  // = Connect as a consumer.

  // Obtain a reference to the consumer administration object.
  RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
    ec->for_consumers (_env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Obtain a reference to the push supplier proxy.
  this->supplier_proxy_ =
    consumer_admin->obtain_push_supplier (_env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Obtain a reference to this object.
  RtecEventComm::PushConsumer_var objref = this->_this (_env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Connect as a consumer.
  this->supplier_proxy_->connect_push_consumer (objref.in (),
                                                qos.get_ConsumerQOS (),
                                                _env);
  TAO_CHECK_ENV_RETURN_VOID(_env);
}

The following code is derived from the EC_Throughput consumer driver code, which can be found in TAO in the file: $TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.cpp

int
ECT_Consumer_Driver::run (int argc, char* argv[])
{
  // argc/argv are used to initialize the ORB and the options
  // for this particular test. Other applications may hard-code
  // the ORB options, obtain them from another source, etc.

  TAO_TRY
    {
      // The use of TAO_TRY macros isolate us from the differences
      // between platforms with and without native C++ exceptions.
      // This is work in progress and may change in the future!

      // Below is some boiler plate code to initialize the ORB and
      // the POA. Notice that applications that connect to the Event
      // Channel play the server role in some instances, because
      // they receive push() requests (as Consumers) or
      // disconnect_push_supplier() requests (as Suppliers).

      // Initialize the ORB reference.
      this->orb_ =
        CORBA::ORB_init (argc, argv, "", TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Initialize the root POA reference.
      CORBA::Object_var poa_object =
        this->orb_->resolve_initial_references("RootPOA");
      if (CORBA::is_nil (poa_object.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to initialize the POA.\n"),
                          1);

      // Obtain the narrowed root POA reference.
      PortableServer::POA_var root_poa =
        PortableServer::POA::_narrow (poa_object.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Obtain a reference to the POA manager.
      PortableServer::POAManager_var poa_manager =
        root_poa->the_POAManager (TAO_TRY_ENV);
      TAO_CHECK_ENV;

       // Now some boiler plate code to obtain a reference to the
       // naming service.....

      // Resolve a reference to the naming service.
      CORBA::Object_var naming_obj =
        this->orb_->resolve_initial_references ("NameService");
      if (CORBA::is_nil (naming_obj.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to get the Naming Service.\n"),
                          1);

      // Narrow the naming service reference.
      CosNaming::NamingContext_var naming_context =
        CosNaming::NamingContext::_narrow (naming_obj.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Use the Naming Service to locate the Scheduling Service and
      // use the Scheduler_Factory to keep a global pointer to the
      // latter. 

      // Initialize the scheduler factory to operate in configuration mode.
      if (ACE_Scheduler_Factory::use_config (naming_context.in ()) == -1)
        return -1;

      // Use the Naming Service to locate the Event Service....

      // Set up the event service lookup name.
      CosNaming::Name name (1);
      name.length (1);
      name[0].id = CORBA::string_dup ("EventService");

      // Resolve a reference to the event service.
      CORBA::Object_var ec_obj =
        naming_context->resolve (name, TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Narrow the reference to the event service.
      RtecEventChannelAdmin::EventChannel_var channel;
      if (CORBA::is_nil (ec_obj.in ()))
        channel = RtecEventChannelAdmin::EventChannel::_nil ();
      else
        channel = RtecEventChannelAdmin::EventChannel::_narrow (ec_obj.in (),
                                                                TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Activate the POA so we can start receiving requests...

      // Activate the POA manager.
      poa_manager->activate (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Connect consumers to the event service.
      this->connect_consumers (channel.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      ACE_DEBUG ((LM_DEBUG, "connected consumer(s)\n"));
      ACE_DEBUG ((LM_DEBUG, "running the test\n"));

      // Run the event loop.
      if (this->orb_->run () == -1)
        ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "orb->run"), -1);
      ACE_DEBUG ((LM_DEBUG, "event loop finished\n"));

      this->dump_results ();

      // Disconnect consumers from the event service.
      this->disconnect_consumers (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Destroy the event service.
      channel->destroy (TAO_TRY_ENV);
      TAO_CHECK_ENV;
    }
  TAO_CATCH (CORBA::SystemException, sys_ex)
    {
      TAO_TRY_ENV.print_exception ("SYS_EX");
    }
  TAO_CATCHANY
    {
      TAO_TRY_ENV.print_exception ("NON SYS EX");
    }
  TAO_ENDTRY;
  return 0;
}


Receiving Events

The following code is derived from the EC_Throughput consumer code, which can be found in TAO in the file: $TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp

void
Test_Consumer::push (const RtecEventComm::EventSet& events,
                     CORBA::Environment &_env)
{
  // Make sure at least one event was pushed.
  if (events.length () == 0)
    {
      // ACE_DEBUG ((LM_DEBUG, "no events\n"));
      return;
    }

  // Make sure only one thread has access.
  ACE_GUARD (ACE_SYNCH_MUTEX, ace_mon, this->lock_);

  // We start the timer as soon as we receive the first event.
  if (this->recv_count_ == 0)
    this->timer_.start ();

  // Update the count of received events.
  this->recv_count_ += events.length ();

  if (TAO_debug_level > 0
      && this->recv_count_ % 1000 == 0)
    {
      ACE_DEBUG ((LM_DEBUG,
                  "ECT_Consumer (%P|%t): %d events received\n",
		  this->recv_count_));
    }

  // Loop through the events, looking for shutdown events.
  for (u_int i = 0; i < events.length (); ++i)
    {
      if (events[i].header.type == ACE_ES_EVENT_SHUTDOWN)
        {
          this->shutdown_count_++;
          if (this->shutdown_count_ >= this->n_suppliers_)
            {
              // We stop the timer as soon as we realize it is time to
              // do so.
              this->timer_.stop ();
              this->driver_->shutdown_consumer (this->cookie_, _env);
            }
        }
    }
}


Disconnecting Consumers from the Event Channel

The following code is derived from the EC_Throughput consumer code, which can be found in TAO in the file: $TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp

void
Test_Consumer::disconnect (CORBA::Environment &_env)
{
  // Make sure the supplier proxy reference is valid.
  if (CORBA::is_nil (this->supplier_proxy_.in ()))
    return;

  // Disconnect from further communication with the push
  // supplier(s).  Each consumer is represented by a unique
  // ACE_ES_ConsumerModule instance. Which connection to
  // disconnect is determined by the instance for the consumer.
  this->supplier_proxy_->disconnect_push_supplier (_env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Mark the supplier proxy reference invalid.
  this->supplier_proxy_ =
    RtecEventChannelAdmin::ProxyPushSupplier::_nil ();

  // We want to stop processing events for this consumer. Above,
  // we disconnected the consumer from the Event Channel, so no
  // more events will be sent, but we could have some events in
  // transit.

  // Without a flushing protocol we need to deactivate the
  // servant to stop accepting push () requests for any 
  // incoming events.

  // Deactivate the servant
  PortableServer::POA_var poa = 
    this->_default_POA (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
  PortableServer::ObjectId_var id =
    poa->servant_to_id (this, _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
  poa->deactivate_object (id.in (), _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
}


Developing Suppliers

Providing QoS Information

In applications that use hard real-time scheduling, a Supplier should provide the worst case, expected, and cached execution time for each operation on the supplier side. Even if these values are small and highly deterministic, it is generally better to specify them in the supplier's RT_Info rather than folding them into the RT_Info of each consumer.
Such a supplier must also specify criticality and importance values, a period, and the number of threads for each operation. A real-time scheduler propagates this information to consumer RT_Infos along the graph of dependencies. The scheduler then uses the propagated information to order dispatches within a set of operations whose dependencies have been met.
The Event Service matches supplier publications with consumer subscriptions to provide efficient event filtering. Providing incorrect publications or subscriptions will result in missed events. The Event Service also uses the subscription information to create additional dependencies between registered RT_Infos. Thus, providing correct supplier publication and consumer subscription information is also critical for correct scheduling in hard real-time applications.
As noted before in the discussion of consumers, the following steps are only necessary for applications that make use of the Event Service's hard real-time features. Applications that do not need these features and are configured with a null scheduler may skip the following operations on the scheduling server.

  // Obtain a reference to the scheduler server.
  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  // Create new RT_Info descriptors for two events.

  RtecScheduler::handle_t handle0 = 
    server->create ("event_0",              // Name of entry point
                    TAO_TRY_ENV             // Environment
                   );

  RtecScheduler::handle_t handle1 = 
    server->create ("event_1",              // Name of entry point
                    TAO_TRY_ENV             // Environment
                   );

  // Register as a producer of event_0.
  server->set (handle0,                     // RT_Info handle
               RtecScheduler::HIGH_CRITICALITY,   // Criticality
               10,                          // Worst case time (in 100 nanosecs)
               10,                          // Typical time (in 100 nanosecs)
               10,                          // Cached time (in 100 nanosecs)
               50000 * 10,                  // Period in 100 nanosecs (= 20 Hz)
               RtecScheduler::LOW_IMPORTANCE,     // Importance
               0,                           // Quantum (unused)
               1,                           // Threads
               RtecScheduler::OPERATION,          // Info type
               TAO_TRY_ENV);

  // Register as a producer of event_1.
  server->set (handle1,                     // RT_Info handle
               RtecScheduler::HIGH_CRITICALITY,   // Criticality
               10,                          // Worst case time (in 100 nanosecs)
               10,                          // Typical time (in 100 nanosecs)
               10,                          // Cached time (in 100 nanosecs)
               50000 * 10,                  // Period in 100 nanosecs (= 20 Hz)
               RtecScheduler::LOW_IMPORTANCE,     // Importance
               0,                           // Quantum (unused)
               1,                           // Threads
               RtecScheduler::OPERATION,          // Info type
               TAO_TRY_ENV);


Connecting Suppliers to Event Channel

The following code is derived from the EC_Throughput supplier code, which can be found in TAO in the file: $TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp

void
Test_Supplier::connect (const char* name,
                        int burst_count,
                        int burst_size,
                        int event_size,
                        int burst_pause,
                        int event_a,
                        int event_b,
                        RtecEventChannelAdmin::EventChannel_ptr ec,
                        CORBA::Environment &_env)
{
  // Some application-specific setup code.

  // Store the passed parameters in the object.
  this->burst_count_ = burst_count;
  this->burst_size_ = burst_size;
  this->event_size_ = event_size;
  this->burst_pause_ = burst_pause;
  this->event_a_ = event_a;
  this->event_b_ = event_b;
   
  // Register operations with the scheduling service.  The following steps are
  // only necessary for applications that make use of the Event Service's hard
  // real-time features.  Applications that do not need these features and are
  // configured with a null scheduler may skip the following operations on the
  // scheduling server.

  // Obtain a reference to the scheduling service. 
  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  // Create an RT_Info descriptor for the passed operation name. 
  RtecScheduler::handle_t rt_info =
    server->create (name, _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Calculate the period at which to supply events.
  ACE_Time_Value tv (0, burst_pause);
  RtecScheduler::Period_t rate = tv.usec () * 10;

  // Set the information in the RT_Info descriptor.
  tv.set (0, 2000);
  TimeBase::TimeT time;
  ORBSVCS_Time::Time_Value_to_TimeT (time, tv);
  server->set (rt_info,
               RtecScheduler::VERY_HIGH_CRITICALITY,
               time, time, time,
               rate,
               RtecScheduler::VERY_LOW_IMPORTANCE,
               time,
               1,
               RtecScheduler::OPERATION,
               _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Now, create a supplier id, and publish the events 
  // that will be supplied under this id.

  // Create a supplier id from the passed name
  this->supplier_id_ = ACE::crc32 (name);
  ACE_DEBUG ((LM_DEBUG, "ID for <%s> is %04.4x\n", name,
              this->supplier_id_));

  // Publish the events the supplier provides.
  ACE_SupplierQOS_Factory qos;
  qos.insert (this->supplier_id_,
              event_a,
              rt_info, 1);
  qos.insert (this->supplier_id_,
              event_b,
              rt_info, 1);
  qos.insert (this->supplier_id_,
              ACE_ES_EVENT_SHUTDOWN,
              rt_info, 1);

  // And finally, some boiler plate code to connect a supplier 
  // to the Event Service.  This is where the connection is
  // actually made.

  // Obtain a reference to the supplier administration object.
  RtecEventChannelAdmin::SupplierAdmin_var supplier_admin =
    ec->for_suppliers (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Obtain a reference to the consumer proxy object.
  this->consumer_proxy_ =
    supplier_admin->obtain_push_consumer (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Obtain a reference to this supplier object.
  RtecEventComm::PushSupplier_var objref =
    this->supplier_._this (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Connect as a supplier of the published events.
  this->consumer_proxy_->connect_push_supplier (objref.in (),
                                                qos.get_SupplierQOS (),
                                                _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
}

The following code is derived from the EC_Throughput supplier driver code, which can be found in TAO in the file: $TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.cpp

int
ECT_Supplier_Driver::run (int argc, char* argv[])
{
  // argc/argv are used to initialize the ORB and the options
  // for this particular test. Other applications may hard-code
  // the ORB options, obtain them from another source, etc.

  TAO_TRY
    {
      // The use of TAO_TRY macros isolate us from the differences
      // between platforms with and without native C++ exceptions.
      // This is work in progress and may change in the future!

      // Below is some boiler plate code to initialize the ORB and
      // the POA. Notice that applications that connect to the Event
      // Channel play the server role in some instances, because
      // they receive push() requests (as Consumers) or
      // disconnect_push_supplier() requests (as Suppliers).

      // Initialize the ORB reference.
      CORBA::ORB_var orb =
        CORBA::ORB_init (argc, argv, "", TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Initialize the root POA reference.
      CORBA::Object_var poa_object =
        orb->resolve_initial_references("RootPOA");
      if (CORBA::is_nil (poa_object.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to initialize the POA.\n"),
                          1);

      // Obtain the narrowed root POA reference.
      PortableServer::POA_var root_poa =
        PortableServer::POA::_narrow (poa_object.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Obtain a reference to the POA manager.
      PortableServer::POAManager_var poa_manager =
        root_poa->the_POAManager (TAO_TRY_ENV);
      TAO_CHECK_ENV;


       // Now some boiler plate code to obtain a reference to the
       // naming service.....

      // Resolve a reference to the naming service.
      CORBA::Object_var naming_obj =
        orb->resolve_initial_references ("NameService");
      if (CORBA::is_nil (naming_obj.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to get the Naming Service.\n"),
                          1);

      // Narrow the naming service reference.
      CosNaming::NamingContext_var naming_context =
        CosNaming::NamingContext::_narrow (naming_obj.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Use the Naming Service to locate the Scheduling Service and
      // use the Scheduler_Factory to keep a global pointer to the
      // latter. 

      // Initialize the scheduler factory to operate in configuration mode.
      if (ACE_Scheduler_Factory::use_config (naming_context.in ()) == -1)
        return -1;

      // Use the Naming Service to locate the Event Service....

      // Set up the event service lookup name.
      CosNaming::Name name (1);
      name.length (1);
      name[0].id = CORBA::string_dup ("EventService");

      // Resolve a reference to the event service.
      CORBA::Object_var ec_obj =
        naming_context->resolve (name, TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Narrow the reference to the event service.
      RtecEventChannelAdmin::EventChannel_var channel;
      if (CORBA::is_nil (ec_obj.in ()))
        channel = RtecEventChannelAdmin::EventChannel::_nil ();
      else
        channel = RtecEventChannelAdmin::EventChannel::_narrow (ec_obj.in (),
                                                                TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Activate the POA so we can start receiving requests...

      // Activate the POA manager.
      poa_manager->activate (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Connect suppliers to the event service.
      this->connect_suppliers (channel.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      ACE_DEBUG ((LM_DEBUG, "connected supplier(s)\n"));

      // Activate the supplier objects
      this->activate_suppliers (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      ACE_DEBUG ((LM_DEBUG, "suppliers are active\n"));

      // Wait for the supplier threads.
      if (ACE_Thread_Manager::instance ()->wait () == -1)
        {
          ACE_ERROR ((LM_ERROR, "Thread_Manager wait failed\n"));
          return 1;
        }

      ACE_DEBUG ((LM_DEBUG, "suppliers finished\n"));

      this->dump_results ();

      // Disconnect suppliers from the event service.
      this->disconnect_suppliers (TAO_TRY_ENV);
      TAO_CHECK_ENV;
    }
  TAO_CATCH (CORBA::SystemException, sys_ex)
    {
      TAO_TRY_ENV.print_exception ("SYS_EX");
    }
  TAO_CATCHANY
    {
      TAO_TRY_ENV.print_exception ("NON SYS EX");
    }
  TAO_ENDTRY;
  return 0;
}


Generating Events

The following code is derived from the EC_Throughput supplier code, which can be found in TAO in the file: $TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp

int
Test_Supplier::svc ()
{
  TAO_TRY
    {
      // First, a bunch of code that is specific to this test.

      // Set pause (sleep) value between message bursts.
      ACE_Time_Value tv (0, this->burst_pause_);

      // Set up message block for event data.
      ACE_Message_Block mb (this->event_size_);
      mb.wr_ptr (this->event_size_);

      // Create an event set for one event, initialize event header.
      RtecEventComm::EventSet event (1);
      event.length (1);
      event[0].header.source = this->supplier_id ();
      event[0].header.ttl = 1;

      // Set up time stamps in event header.  This is for performance
      // measurements, so this step can be omitted at will.
      ACE_hrtime_t t = ACE_OS::gethrtime ();
      ORBSVCS_Time::hrtime_to_TimeT (event[0].header.creation_time, t);
      event[0].header.ec_recv_time = ORBSVCS_Time::zero;
      event[0].header.ec_send_time = ORBSVCS_Time::zero;

      // Initialize data fields in event.
      event[0].data.x = 0;
      event[0].data.y = 0;

      // We use replace to minimize the copies. This should result
      // in just one memory allocation;
      event[0].data.payload.replace (this->event_size_,
                                     &mb);

      // This is where the events are actually pushed into
      // the event channel.  The test pushes bursts of events,
      // pausing a specified interval between bursts.

      // Start the timer, and begin pushing events.
      this->timer_.start ();
      for (int i = 0; i < this->burst_count_; ++i)
        {
          // Send a burst of events.
          for (int j = 0; j < this->burst_size_; ++j)
            {
              if (j % 2 == 0)
                event[0].header.type = this->event_a_;
              else
                event[0].header.type = this->event_b_;

              // ACE_DEBUG ((LM_DEBUG, "(%t) supplier push event\n"));
              this->consumer_proxy ()->push (event, TAO_TRY_ENV);

              TAO_CHECK_ENV;
            }
 
          // Sleep until it's time to send the next burst.
          ACE_OS::sleep (tv);
        }

      // Send a "magic" type of event to inform the consumer that we are 
      // not sending anything else...

      // Send one event shutdown from each supplier
      event[0].header.type = ACE_ES_EVENT_SHUTDOWN;
      this->consumer_proxy ()->push(event, TAO_TRY_ENV);
      TAO_CHECK_ENV;
      this->timer_.stop ();
      
    }
  TAO_CATCH (CORBA::SystemException, sys_ex)
    {
      TAO_TRY_ENV.print_exception ("SYS_EX");
    }
  TAO_CATCHANY
    {
      TAO_TRY_ENV.print_exception ("NON SYS EX");
    }
  TAO_ENDTRY;
  return 0;
}


Disconnecting Suppliers from the Event Channel

The following code is derived from the EC_Throughput supplier code, which can be found in TAO in the file: $TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp

void
Test_Supplier::disconnect (CORBA::Environment &_env)
{
  // Make sure the consumer proxy reference is valid.
  if (CORBA::is_nil (this->consumer_proxy_.in ()))
    return;

  // Disconnect communication with the push consumer(s).
  this->consumer_proxy_->disconnect_push_consumer (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Mark the consumer proxy reference invalid.
  this->consumer_proxy_ =
    RtecEventChannelAdmin::ProxyPushConsumer::_nil ();

  // We need to stop accepting disconnect_push_supplier () requests 
  // for this supplier, before it is safe to destroy the supplier.
  // As required by the CORBA spec, you must explicitly deactivate
  // a servant before destroying it.

  // Deactivate the servant
  PortableServer::POA_var poa = 
    this->supplier_._default_POA (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
  PortableServer::ObjectId_var id =
    poa->servant_to_id (&this->supplier_, _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
  poa->deactivate_object (id.in (), _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
    RtecEventChannelAdmin::ProxyPushConsumer::_nil ();
}


Caring for your Event Channel


Creation and Name Service Registration

The following code is derived from the Event_Service executable, which can be found in TAO in the file: $TAO_ROOT/orbsvcs/Event_Service/Event_Service.cpp

int main (int argc, char *argv[])
{
  TAO_TRY
    {
      // argc/argv are used to initialize the ORB and the options
      // for the Event Service executable. Other applications may
      // hard code the ORB options, obtain them from another source, etc.

      // Again the boiler plate code for ORB and POA initialization.

      // Initialize ORB.
      CORBA::ORB_var orb =
        CORBA::ORB_init (argc, argv, "internet", TAO_TRY_ENV);
      TAO_CHECK_ENV;

      if (parse_args (argc, argv) == -1)
        return 1;

      CORBA::Object_var poa_object =
        orb->resolve_initial_references("RootPOA");
      if (CORBA::is_nil (poa_object.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to initialize the POA.\n"),
                          1);

      PortableServer::POA_var root_poa =
        PortableServer::POA::_narrow (poa_object.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      PortableServer::POAManager_var poa_manager =
        root_poa->the_POAManager (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      CORBA::Object_var naming_obj =
        orb->resolve_initial_references ("NameService");
      if (CORBA::is_nil (naming_obj.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to initialize the Naming Service.\n"),
                          1);

      CosNaming::NamingContext_var naming_context =
        CosNaming::NamingContext::_narrow (naming_obj.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Notice the use of auto_ptr<> to automagically manage the
      // destruction of the servant.  When the auto_ptr goes out
      // of scope, its destructor is called, which in turn destroys
      // the servant.

      auto_ptr scheduler_impl;
      RtecScheduler::Scheduler_var scheduler;


      // Create a new servant to implement the Scheduling Service, 
      // register it with the Naming Service, and use the 
      // Scheduler_Factory to keep a global pointer to the new
      // Scheduling Service.

      // This is the name we (potentially) use to register the Scheduling
      // Service in the Naming Service.
      CosNaming::Name schedule_name (1);
      schedule_name.length (1);
      schedule_name[0].id = CORBA::string_dup ("ScheduleService");

      if (global_scheduler == 0)
        {
          scheduler_impl =
            auto_ptr(new ACE_Config_Scheduler);
          if (scheduler_impl.get () == 0)
	    return 1;
          scheduler = scheduler_impl->_this (TAO_TRY_ENV);
          TAO_CHECK_ENV;

	  CORBA::String_var str =
	    orb->object_to_string (scheduler.in (), TAO_TRY_ENV);
	  TAO_CHECK_ENV;
	  ACE_DEBUG ((LM_DEBUG, "The (local) scheduler IOR is <%s>\n",
		      str.in ()));

	  // Register the servant with the Naming Context....
	  naming_context->bind (schedule_name, scheduler.in (), TAO_TRY_ENV);
	  TAO_CHECK_ENV;
        }

      ACE_Scheduler_Factory::use_config (naming_context.in ());

      // The Event Service can be configured to support priority based
      // dispatching (the "default_Module_Factory") or best effort (the
      // "Reactive_Module_Factory"). We pick the right module factory
      // according to the command line options processed above.

      TAO_Default_Module_Factory default_module_factory;
      TAO_Reactive_Module_Factory reactive_module_factory;

      TAO_Module_Factory* module_factory = &default_module_factory;
      if (reactive)
        module_factory = &reactive_module_factory;

      // Now, create a new event channel servant to implement the
      // Event Service, and register it with Naming Service.

      // Construct the event channel using the given module factory.
      ACE_EventChannel ec_impl (1,
                                ACE_DEFAULT_EVENT_CHANNEL_TYPE,
                                module_factory);

      // Obtain an object reference to the new channel.
      RtecEventChannelAdmin::EventChannel_var ec =
        ec_impl._this (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Convert the EC object reference to a string.
      CORBA::String_var str =
        orb->object_to_string (ec.in (), TAO_TRY_ENV);

      // Output the EC object reference string (debug only).
      ACE_DEBUG ((LM_DEBUG,
		  "The EC IOR is <%s>\n", str.in ()));

      // Register the EC with the Naming Service.
      CosNaming::Name channel_name (1);
      channel_name.length (1);
      channel_name[0].id = CORBA::string_dup (service_name);
      naming_context->bind (channel_name, ec.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Activate the POA so we can start receiving requests...

      // Activate the POA manager.
      poa_manager->activate (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Run the ORB event loop
      ACE_DEBUG ((LM_DEBUG, "%s; running event service\n", __FILE__));
      if (orb->run () == -1)
        ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "run"), 1);

      // Now the Event Service is finished.  We could deactivate the 
      // EC and SS here, but we don't need to, as the server is 
      // going down anyway.


      // Remove the event service and the scheduling service from
      // the Naming Service.

      naming_context->unbind (channel_name, TAO_TRY_ENV);
      TAO_CHECK_ENV;

      if (global_scheduler == 0)
	{
	  naming_context->unbind (schedule_name, TAO_TRY_ENV);
	  TAO_CHECK_ENV;
	}

    }
  TAO_CATCHANY
    {
      TAO_TRY_ENV.print_exception ("EC");
    }
  TAO_ENDTRY;


  return 0;
}


Register Suppliers and Consumers




Sample Applications

A number of sample applications are available in the directories under TAO's ORB Services tests.
In particular, much of the code shown in this tutorial was drawn from the EC_Throughput test. This test exercises the Event Service and measures its throughput capabilities.
A similar test, Event_Latency, measures the latency of events through the Event Service.
The EC_Basic test demonstrates the basic use the Event Service.
The EC_Multiple test shows a number of ways to connect multiple Event Channels.
For the IDL source for the various interfaces, please see RtecScheduler.idl, CosEventChannelAdmin.idl, CosEventComm.idl and CosNaming.idl.


Reference Materials

The following materials were used in developing this tutorial: please refer to them for further information.

Books

Mowbray, T. and Zahavi, R. The Essential CORBA, Systems Integration Using Distributed Objects. Wiley, 1995. ISBN 0-471-10611-9

Baker, S. CORBA Distributed Objects Using Orbix. Addison-Wesley, 1997. ISBN 0-201-92475-7

Papers


Last modified 10:50:30 CST 22 December 1998 by Chris Gill