ChangeLogTag:Tue Feb 23 20:10:52 1999 Carlos O'Ryan <coryan@cs.wustl.edu>

10 files changed, 176 insertions, 94 deletions

diff --git a/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp b/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp
index dfac0802a25..946d4cde39a 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp
+++ b/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp
@@ -26,7 +26,8 @@ Test_Consumer::Test_Consumer (ECT_Driver *driver,
 void
 Test_Consumer::connect (RtecScheduler::Scheduler_ptr scheduler,
                         const char* name,
-                        int event_a, int event_b,
+                        int type_start,
+                        int type_count,
                         RtecEventChannelAdmin::EventChannel_ptr ec,
                         CORBA::Environment& TAO_IN_ENV)
 {
@@ -53,8 +54,10 @@ Test_Consumer::connect (RtecScheduler::Scheduler_ptr scheduler,
   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);
+  for (int i = 0; i != type_count; ++i)
+    {
+      qos.insert_type (type_start + i, rt_info);
+    }
 
   // = Connect as a consumer.
   RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
@@ -87,7 +90,7 @@ Test_Consumer::disconnect (CORBA::Environment &TAO_IN_ENV)
     RtecEventChannelAdmin::ProxyPushSupplier::_nil ();
 
   // Deactivate the servant
-  PortableServer::POA_var poa = 
+  PortableServer::POA_var poa =
     this->_default_POA (TAO_IN_ENV);
   TAO_CHECK_ENV_RETURN_VOID (TAO_IN_ENV);
   PortableServer::ObjectId_var id =
@@ -104,7 +107,7 @@ Test_Consumer::dump_results (const char* name)
   this->timer_.elapsed_time (tv);
   double f = 1.0 / (tv.sec () + tv.usec () / 1000000.0);
   double eps = this->recv_count_ * f;
-  
+
   ACE_DEBUG ((LM_DEBUG,
               "ECT_Consumer (%s):\n"
               "    Total time: %d.%08.8d (secs.usecs)\n"
@@ -148,11 +151,6 @@ Test_Consumer::push (const RtecEventComm::EventSet& events,
     {
       const RtecEventComm::Event& e = events[i];
 
-      if (e.data.payload.mb () == 0)
-        {
-          // ACE_DEBUG ((LM_DEBUG, "No data in event[%d]\n", i));
-          // continue;
-        }
       if (e.header.type == ACE_ES_EVENT_SHUTDOWN)
         {
           this->shutdown_count_++;
@@ -169,7 +167,7 @@ Test_Consumer::push (const RtecEventComm::EventSet& events,
           ACE_hrtime_t creation;
           ORBSVCS_Time::TimeT_to_hrtime (creation,
                                          e.header.creation_time);
-          
+
           ACE_hrtime_t ec_recv;
           ORBSVCS_Time::TimeT_to_hrtime (ec_recv,
                                          e.header.ec_recv_time);
diff --git a/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.h b/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.h
index c59cee0a4dd..8b33409dd5b 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.h
+++ b/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.h
@@ -26,11 +26,15 @@
 
 class Test_Consumer : public POA_RtecEventComm::PushConsumer
 {
-  //
   // = TITLE
   //   Receive the events.
   //
   // = DESCRIPTION
+  //   This class is a consumer of events. It subscribes for a
+  //   continous ranges of event types, this permits studying the
+  //   effect of the number of subscriptions for each particular kind
+  //   of event on the EC.
+  //
 public:
   Test_Consumer (ECT_Driver* driver,
                  void* cookie,
@@ -38,8 +42,8 @@ public:
 
   void connect (RtecScheduler::Scheduler_ptr scheduler,
                 const char* name,
-                int event_a,
-                int event_b,
+                int type_start,
+                int type_count,
                 RtecEventChannelAdmin::EventChannel_ptr ec,
                 CORBA::Environment& _env);
   // This method connects the consumer to the EC.
diff --git a/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.cpp b/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.cpp
index b8e68020b14..a930d680bf2 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.cpp
+++ b/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.cpp
@@ -25,8 +25,8 @@ main (int argc, char *argv [])
 ECT_Consumer_Driver::ECT_Consumer_Driver (void)
   : n_consumers_ (1),
     n_suppliers_ (1),
-    event_a_ (ACE_ES_EVENT_UNDEFINED),
-    event_b_ (ACE_ES_EVENT_UNDEFINED + 1),
+    type_start_ (ACE_ES_EVENT_UNDEFINED),
+    type_count_ (1),
     pid_file_name_ (0),
     active_count_ (0)
 {
@@ -67,14 +67,14 @@ ECT_Consumer_Driver::run (int argc, char* argv[])
                   "Execution parameters:\n"
                   "  consumers = <%d>\n"
                   "  suppliers = <%d>\n"
-                  "  supplier Event A = <%d>\n"
-                  "  supplier Event B = <%d>\n"
+                  "  type_start = <%d>\n"
+                  "  type count = <%d>\n"
                   "  pid file name = <%s>\n",
 
                   this->n_consumers_,
                   this->n_suppliers_,
-                  this->event_a_,
-                  this->event_b_,
+                  this->type_start_,
+                  this->type_start_,
 
                   this->pid_file_name_?this->pid_file_name_:"nil") );
 
@@ -131,7 +131,7 @@ ECT_Consumer_Driver::run (int argc, char* argv[])
       TAO_CHECK_ENV;
       if (CORBA::is_nil (sched_obj.in ()))
         return 1;
-      RtecScheduler::Scheduler_var scheduler = 
+      RtecScheduler::Scheduler_var scheduler =
         RtecScheduler::Scheduler::_narrow (sched_obj.in (),
                                            TAO_TRY_ENV);
       TAO_CHECK_ENV;
@@ -223,8 +223,8 @@ ECT_Consumer_Driver::connect_consumers
 
       this->consumers_[i]->connect (scheduler,
                                     buf,
-                                    this->event_a_,
-                                    this->event_b_,
+                                    this->type_start_,
+                                    this->type_count_,
                                     channel,
                                     TAO_IN_ENV);
       TAO_CHECK_ENV_RETURN_VOID (TAO_IN_ENV);
@@ -277,9 +277,9 @@ ECT_Consumer_Driver::parse_args (int argc, char *argv [])
             char* aux;
                 char* arg = ACE_OS::strtok_r (get_opt.optarg, ",", &aux);
 
-            this->event_a_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
+            this->type_start_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
                 arg = ACE_OS::strtok_r (0, ",", &aux);
-            this->event_b_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
+            this->type_count_ = ACE_OS::atoi (arg);
           }
           break;
 
@@ -294,7 +294,7 @@ ECT_Consumer_Driver::parse_args (int argc, char *argv [])
                       "[ORB options] "
                       "-c <n_consumers> "
                       "-s <n_suppliers> "
-                      "-h <event_a,event_b> "
+                      "-h <type_start,type_count> "
                       "-p <pid file name> "
                       "\n",
                       argv[0]));
@@ -319,5 +319,14 @@ ECT_Consumer_Driver::parse_args (int argc, char *argv [])
                          "suppliers out of range\n", argv[0]), -1);
     }
 
+  if (this->type_count_ <= 0)
+    {
+      ACE_ERROR_RETURN ((LM_DEBUG,
+                         "%s: number of event types "
+                         "suppliers out of range, reset to default (1)\n",
+                         argv[0]), -1);
+      this->type_count_ = 1;
+    }
+
   return 0;
 }
diff --git a/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.h b/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.h
index 36827c86218..4eaf51054c6 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.h
+++ b/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.h
@@ -74,9 +74,10 @@ private:
   // supplier sends a shutdown message after it finishes, the consumer
   // finishes when all the suppliers do.
 
-  int event_a_;
-  int event_b_;
-  // We send two types of events, with different contents.
+  int type_start_;
+  int type_count_;
+  // We receive the events whose type is in the range
+  // [type_start,type_start+type_count)
 
   const char* pid_file_name_;
   // The name of a file where the process stores its pid
diff --git a/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp b/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp
index 09c2097252f..ea94d0c59e2 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp
+++ b/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp
@@ -18,7 +18,9 @@ Test_Supplier::Test_Supplier (ECT_Driver *driver)
      burst_count_ (0),
      burst_size_ (0),
      event_size_ (0),
-     burst_pause_ (0)
+     burst_pause_ (0),
+     type_start_ (ACE_ES_EVENT_UNDEFINED),
+     type_count_ (1)
 {
 }
 
@@ -29,8 +31,8 @@ Test_Supplier::connect (RtecScheduler::Scheduler_ptr scheduler,
                         int burst_size,
                         int event_size,
                         int burst_pause,
-                        int event_a,
-                        int event_b,
+                        int type_start,
+                        int type_count,
                         RtecEventChannelAdmin::EventChannel_ptr ec,
                         CORBA::Environment &TAO_IN_ENV)
 {
@@ -38,9 +40,9 @@ Test_Supplier::connect (RtecScheduler::Scheduler_ptr scheduler,
   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;
-    
+  this->type_start_ = type_start;
+  this->type_count_ = type_count;
+
   RtecScheduler::handle_t rt_info =
     scheduler->create (name, TAO_IN_ENV);
   TAO_CHECK_ENV_RETURN_VOID (TAO_IN_ENV);
@@ -71,12 +73,12 @@ Test_Supplier::connect (RtecScheduler::Scheduler_ptr scheduler,
               this->supplier_id_));
 
   ACE_SupplierQOS_Factory qos;
-  qos.insert (this->supplier_id_,
-              event_a,
-              rt_info, 1);
-  qos.insert (this->supplier_id_,
-              event_b,
-              rt_info, 1);
+  for (int i = 0; i != type_count; ++i)
+    {
+      qos.insert (this->supplier_id_,
+                  type_start + i,
+                  rt_info, 1);
+    }
   qos.insert (this->supplier_id_,
               ACE_ES_EVENT_SHUTDOWN,
               rt_info, 1);
@@ -112,7 +114,7 @@ Test_Supplier::disconnect (CORBA::Environment &TAO_IN_ENV)
     RtecEventChannelAdmin::ProxyPushConsumer::_nil ();
 
   // Deactivate the servant
-  PortableServer::POA_var poa = 
+  PortableServer::POA_var poa =
     this->supplier_._default_POA (TAO_IN_ENV);
   TAO_CHECK_ENV_RETURN_VOID (TAO_IN_ENV);
   PortableServer::ObjectId_var id =
@@ -166,10 +168,8 @@ Test_Supplier::svc ()
         {
           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_;
+              event[0].header.type =
+                this->type_start_ + j % this->type_count_;
 
               ACE_hrtime_t now = ACE_OS::gethrtime ();
               ORBSVCS_Time::hrtime_to_TimeT (event[0].header.creation_time,
@@ -187,7 +187,7 @@ Test_Supplier::svc ()
       this->consumer_proxy ()->push(event, TAO_TRY_ENV);
       TAO_CHECK_ENV;
       this->timer_.stop ();
-      
+
     }
   TAO_CATCH (CORBA::SystemException, sys_ex)
     {
@@ -226,7 +226,7 @@ Test_Supplier::dump_results (const char* name)
   int event_count = this->burst_count_ * this->burst_size_ + 1;
   double f = 1.0 / (tv.sec () + tv.usec () / 1000000.0);
   double eps = event_count * f;
-  
+
   ACE_DEBUG ((LM_DEBUG,
               "ECT_Supplier (%s):\n"
               "    Total time: %d.%08.8d (secs.usecs)\n"
diff --git a/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.h b/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.h
index a694b13fb2e..233e270f3d7 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.h
+++ b/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.h
@@ -4,9 +4,8 @@
 // ============================================================================
 //
 // = DESCRIPTION
-//   This test to measure how many events per minute can the EC
-//   process, it also serves as an example how how to encode complex
-//   data types in a octet sequence.
+//   This is a helper class for the throughput tests of the Event
+//   Channel.
 //
 // ============================================================================
 
@@ -43,8 +42,8 @@ public:
                 int burst_size,
                 int event_size,
                 int burst_pause,
-                int event_a,
-                int event_b,
+                int type_start,
+                int type_count,
                 RtecEventChannelAdmin::EventChannel_ptr ec,
                 CORBA::Environment& _env);
   // This method connects the supplier to the EC.
@@ -89,8 +88,8 @@ private:
   int burst_size_;
   int event_size_;
   int burst_pause_;
-  int event_a_;
-  int event_b_;
+  int type_start_;
+  int type_count_;
   // The test data.
 };
 
diff --git a/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.cpp b/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.cpp
index 7444b067e0b..8423b26bc26 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.cpp
+++ b/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.cpp
@@ -28,8 +28,8 @@ ECT_Supplier_Driver::ECT_Supplier_Driver (void)
     burst_size_ (100),
     event_size_ (128),
     burst_pause_ (100),
-    event_a_ (ACE_ES_EVENT_UNDEFINED),
-    event_b_ (ACE_ES_EVENT_UNDEFINED + 1),
+    type_start_ (ACE_ES_EVENT_UNDEFINED),
+    type_count_ (1),
     pid_file_name_ (0)
 {
 }
@@ -78,8 +78,8 @@ ECT_Supplier_Driver::run (int argc, char* argv[])
                   "  burst size = <%d>\n"
                   "  event size = <%d>\n"
                   "  burst size = <%d>\n"
-                  "  supplier Event A = <%d>\n"
-                  "  supplier Event B = <%d>\n"
+                  "  type start = <%d>\n"
+                  "  type count = <%d>\n"
                   "  pid file name = <%s>\n",
 
                   this->n_suppliers_,
@@ -87,8 +87,8 @@ ECT_Supplier_Driver::run (int argc, char* argv[])
                   this->burst_size_,
                   this->event_size_,
                   this->burst_pause_,
-                  this->event_a_,
-                  this->event_b_,
+                  this->type_start_,
+                  this->type_count_,
 
                   this->pid_file_name_?this->pid_file_name_:"nil") );
 
@@ -145,7 +145,7 @@ ECT_Supplier_Driver::run (int argc, char* argv[])
       TAO_CHECK_ENV;
       if (CORBA::is_nil (sched_obj.in ()))
         return 1;
-      RtecScheduler::Scheduler_var scheduler = 
+      RtecScheduler::Scheduler_var scheduler =
         RtecScheduler::Scheduler::_narrow (sched_obj.in (),
                                            TAO_TRY_ENV);
       TAO_CHECK_ENV;
@@ -168,7 +168,7 @@ ECT_Supplier_Driver::run (int argc, char* argv[])
       poa_manager->activate (TAO_TRY_ENV);
       TAO_CHECK_ENV;
 
-      this->connect_suppliers (scheduler.in (), 
+      this->connect_suppliers (scheduler.in (),
                                channel.in (),
                                TAO_TRY_ENV);
       TAO_CHECK_ENV;
@@ -227,8 +227,8 @@ ECT_Supplier_Driver::connect_suppliers
                                     this->burst_size_,
                                     this->event_size_,
                                     this->burst_pause_,
-                                    this->event_a_,
-                                    this->event_b_,
+                                    this->type_start_,
+                                    this->type_count_,
                                     channel,
                                     TAO_IN_ENV);
       TAO_CHECK_ENV_RETURN_VOID (TAO_IN_ENV);
@@ -301,9 +301,9 @@ ECT_Supplier_Driver::parse_args (int argc, char *argv [])
             char* aux;
                 char* arg = ACE_OS::strtok_r (get_opt.optarg, ",", &aux);
 
-            this->event_a_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
+            this->type_start_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
                 arg = ACE_OS::strtok_r (0, ",", &aux);
-            this->event_b_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
+            this->type_count_ = ACE_OS::atoi (arg);
           }
           break;
 
diff --git a/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.h b/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.h
index 7db63600626..a4294e0e929 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.h
+++ b/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.h
@@ -4,9 +4,8 @@
 // ============================================================================
 //
 // = DESCRIPTION
-//   This test to measure how many events per minute can the EC
-//   process, it also serves as an example how how to encode complex
-//   data types in a octet sequence.
+//   This is a helper class for the throughput tests of the Event
+//   Channel.
 //
 // ============================================================================
 
@@ -82,8 +81,8 @@ private:
   int burst_pause_;
   // The time between each event burst, in microseconds.
 
-  int event_a_;
-  int event_b_;
+  int type_start_;
+  int type_count_;
   // We send two types of events, with different contents.
 
   const char* pid_file_name_;
diff --git a/TAO/orbsvcs/tests/EC_Throughput/ECT_Throughput.cpp b/TAO/orbsvcs/tests/EC_Throughput/ECT_Throughput.cpp
index 9fac6231604..ed0bbf768c2 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ECT_Throughput.cpp
+++ b/TAO/orbsvcs/tests/EC_Throughput/ECT_Throughput.cpp
@@ -15,6 +15,7 @@
 #include "orbsvcs/Event/Module_Factory.h"
 #include "orbsvcs/Event/EC_Event_Channel.h"
 #include "orbsvcs/Event/EC_Basic_Factory.h"
+#include "orbsvcs/Event/EC_ConsumerAdmin.h"
 #include "ECT_Throughput.h"
 
 ACE_RCSID(EC_Throughput, ECT_Throughput, "$Id$")
@@ -35,12 +36,17 @@ ECT_Throughput::ECT_Throughput (void)
     burst_size_ (100),
     event_size_ (128),
     burst_pause_ (100),
-    event_a_ (ACE_ES_EVENT_UNDEFINED),
-    event_b_ (ACE_ES_EVENT_UNDEFINED + 1),
+    consumer_type_start_ (ACE_ES_EVENT_UNDEFINED),
+    consumer_type_count_ (1),
+    consumer_type_shift_ (0),
+    supplier_type_start_ (ACE_ES_EVENT_UNDEFINED),
+    supplier_type_count_ (1),
+    supplier_type_shift_ (0),
     pid_file_name_ (0),
     active_count_ (0),
     reactive_ec_ (0),
-    new_ec_ (0)
+    new_ec_ (0),
+    ec_concurrency_hwm_ (1)
 {
 }
 
@@ -83,11 +89,16 @@ ECT_Throughput::run (int argc, char* argv[])
                   "  burst size = <%d>\n"
                   "  event size = <%d>\n"
                   "  burst size = <%d>\n"
-                  "  supplier Event A = <%d>\n"
-                  "  supplier Event B = <%d>\n"
+                  "  consumer type start = <%d>\n"
+                  "  consumer type count = <%d>\n"
+                  "  consumer type shift = <%d>\n"
+                  "  supplier type start = <%d>\n"
+                  "  supplier type count = <%d>\n"
+                  "  supplier type shift = <%d>\n"
                   "  pid file name = <%s>\n"
                   "  remote EC = <%d>\n"
-                  "  new EC = <%d>\n",
+                  "  new EC = <%d>\n"
+                  "  concurrency HWM = <%d>\n",
 
                   this->n_consumers_,
                   this->n_suppliers_,
@@ -95,12 +106,17 @@ ECT_Throughput::run (int argc, char* argv[])
                   this->burst_size_,
                   this->event_size_,
                   this->burst_pause_,
-                  this->event_a_,
-                  this->event_b_,
+                  this->consumer_type_start_,
+                  this->consumer_type_count_,
+                  this->consumer_type_shift_,
+                  this->supplier_type_start_,
+                  this->supplier_type_count_,
+                  this->supplier_type_shift_,
 
                   this->pid_file_name_?this->pid_file_name_:"nil",
                   this->reactive_ec_,
-                  this->new_ec_
+                  this->new_ec_,
+                  this->ec_concurrency_hwm_
                   ) );
 
       if (this->pid_file_name_ != 0)
@@ -175,7 +191,7 @@ ECT_Throughput::run (int argc, char* argv[])
                   str.in ()));
 
       // Register the servant with the Naming Context....
-      naming_context->bind (schedule_name, scheduler.in (), TAO_TRY_ENV);
+      naming_context->rebind (schedule_name, scheduler.in (), TAO_TRY_ENV);
       TAO_CHECK_ENV;
 
       ACE_Scheduler_Factory::use_config (naming_context.in ());
@@ -219,9 +235,11 @@ ECT_Throughput::run (int argc, char* argv[])
             new TAO_EC_Event_Channel (ec_factory.get ());
           ec->activate (TAO_TRY_ENV);
           TAO_CHECK_ENV;
+          ec->consumer_admin ()->busy_hwm (this->ec_concurrency_hwm_);
 
           ec_impl =
             auto_ptr<POA_RtecEventChannelAdmin::EventChannel> (ec);
+
 #else
           ACE_ERROR_RETURN ((LM_ERROR,
                              "The new event channel is not supported "
@@ -321,7 +339,7 @@ ECT_Throughput::run (int argc, char* argv[])
 
 void
 ECT_Throughput::shutdown_consumer (void*,
-                               CORBA::Environment &)
+                                   CORBA::Environment &)
 {
   // int ID =
   //   (ACE_reinterpret_cast(Test_Consumer**,consumer_cookie)
@@ -355,10 +373,13 @@ ECT_Throughput::connect_consumers
                               this->consumers_ + i,
                               this->n_suppliers_));
 
+      int start = this->consumer_type_start_
+        + i * this->consumer_type_shift_;
+
       this->consumers_[i]->connect (scheduler,
                                     buf,
-                                    this->event_a_,
-                                    this->event_b_,
+                                    start,
+                                    this->consumer_type_count_,
                                     channel,
                                     TAO_IN_ENV);
       if (TAO_IN_ENV.exception () != 0) return;
@@ -378,14 +399,15 @@ ECT_Throughput::connect_suppliers
 
       ACE_NEW (this->suppliers_[i], Test_Supplier (this));
 
+      int start = this->supplier_type_start_ + i*this->supplier_type_shift_;
       this->suppliers_[i]->connect (scheduler,
                                     buf,
                                     this->burst_count_,
                                     this->burst_size_,
                                     this->event_size_,
                                     this->burst_pause_,
-                                    this->event_a_,
-                                    this->event_b_,
+                                    start,
+                                    this->supplier_type_count_,
                                     channel,
                                     TAO_IN_ENV);
       if (TAO_IN_ENV.exception () != 0) return;
@@ -444,7 +466,7 @@ ECT_Throughput::dump_results (void)
 int
 ECT_Throughput::parse_args (int argc, char *argv [])
 {
-  ACE_Get_Opt get_opt (argc, argv, "dc:s:u:n:t:b:h:p:m:r");
+  ACE_Get_Opt get_opt (argc, argv, "rdc:s:u:n:t:b:h:l:p:m:w:");
   int opt;
 
   while ((opt = get_opt ()) != EOF)
@@ -512,9 +534,24 @@ ECT_Throughput::parse_args (int argc, char *argv [])
             char* aux;
                 char* arg = ACE_OS::strtok_r (get_opt.optarg, ",", &aux);
 
-            this->event_a_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
+            this->consumer_type_start_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
+                arg = ACE_OS::strtok_r (0, ",", &aux);
+            this->consumer_type_count_ = ACE_OS::atoi (arg);
+                arg = ACE_OS::strtok_r (0, ",", &aux);
+            this->consumer_type_shift_ = ACE_OS::atoi (arg);
+          }
+          break;
+
+        case 'l':
+          {
+            char* aux;
+                char* arg = ACE_OS::strtok_r (get_opt.optarg, ",", &aux);
+
+            this->supplier_type_start_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
+                arg = ACE_OS::strtok_r (0, ",", &aux);
+            this->supplier_type_count_ = ACE_OS::atoi (arg);
                 arg = ACE_OS::strtok_r (0, ",", &aux);
-            this->event_b_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
+            this->supplier_type_shift_ = ACE_OS::atoi (arg);
           }
           break;
 
@@ -522,6 +559,10 @@ ECT_Throughput::parse_args (int argc, char *argv [])
           this->pid_file_name_ = get_opt.optarg;
           break;
 
+        case 'w':
+          this->ec_concurrency_hwm_ = ACE_OS::atoi (get_opt.optarg);
+          break;
+
         case '?':
         default:
           ACE_DEBUG ((LM_DEBUG,
@@ -533,8 +574,10 @@ ECT_Throughput::parse_args (int argc, char *argv [])
                       "-n <burst size> "
                       "-b <event payload size> "
                       "-t <burst pause (usecs)> "
-                      "-h <eventa,eventb> "
+                      "-h <consumer_start,consumer_count,consumer_shift> "
+                      "-l <supplier_start,supplier_count,supplier_shift> "
                       "-p <pid file name> "
+                      "-w <concurrency HWM> "
                       "-r "
                       "\n",
                       argv[0]));
@@ -603,6 +646,15 @@ ECT_Throughput::parse_args (int argc, char *argv [])
                          argv[0], 1), -1);
     }
 
+  if (this->ec_concurrency_hwm_ <= 0)
+    {
+      ACE_ERROR_RETURN ((LM_DEBUG,
+                         "%s: invalid concurrency HWM, "
+                         "reset to default (%d)\n",
+                         argv[0], 1), -1);
+      this->ec_concurrency_hwm_ = 1;
+    }
+
   return 0;
 }
 
diff --git a/TAO/orbsvcs/tests/EC_Throughput/ECT_Throughput.h b/TAO/orbsvcs/tests/EC_Throughput/ECT_Throughput.h
index a37826e2543..47d1ff7dacd 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ECT_Throughput.h
+++ b/TAO/orbsvcs/tests/EC_Throughput/ECT_Throughput.h
@@ -4,6 +4,9 @@
 // ============================================================================
 //
 // = DESCRIPTION
+//   This is a helper class for the throughput tests of the Event
+//   Channel.
+//   Used for the collocated test.
 //
 // ============================================================================
 
@@ -93,9 +96,23 @@ private:
   int burst_pause_;
   // The time between each event burst, in microseconds.
 
-  int event_a_;
-  int event_b_;
-  // We send two types of events, with different contents.
+  int consumer_type_start_;
+  int consumer_type_count_;
+  int consumer_type_shift_;
+  // The consumers subscribe to different sets of events, as follows:
+  // Consumer0: [start          , start           + count)
+  // Consumer1: [start + 1*shift, start + 1*shift + count)
+  // Consumer2: [start + 2*shift, start + 2*shift + count)
+  // And so on.
+
+  int supplier_type_start_;
+  int supplier_type_count_;
+  int supplier_type_shift_;
+  // The suppliers generate different sets of events, as follows:
+  // Supplier0: [start          , start           + count)
+  // Supplier1: [start + 1*shift, start + 1*shift + count)
+  // Supplier2: [start + 2*shift, start + 2*shift + count)
+  // And so on.
 
   const char* pid_file_name_;
   // The name of a file where the process stores its pid
@@ -112,6 +129,9 @@ private:
 
   int new_ec_;
   // If not zero then we use the new EC implementation
+
+  int ec_concurrency_hwm_;
+  // Set the HWM for the concurrency in the EC.
 };
 
 #endif /* ECT_TRHOUGHPUT_H */