Chris Gill,
Tim Harrison, and
Carlos O'Ryan
cdgill@cs.wustl.edu,
harrison@cs.wustl.edu, and
coryan@cs.wustl.edu
Overview |
Contents
|
IntroductionThis 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 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 |
![]() |
![]() |
|
Real-Time Event Service Enhancements |
|
The Real-Time Event Service |
![]()
|
|
Real-Time Event Service Internals |
![]() |
|
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 | 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 | Scheduler Dependency Chain |
|---|---|
| Pure Consumer | root node |
| Consumer/Supplier | internal node |
| Supplier/Consumer | internal node |
| Pure Supplier | leaf node |
| 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 |
| 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
);
|
| 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;
}
|
| 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 (TAO_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);
}
}
}
}
|
| 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 |
| 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;
}
|
| 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 ChannelThe 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
|
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 |