Reverted Supplier.{h,cpp} and changed Kokyu_EC to register suppliers and consumers only once, rather than once per event type

5 files changed, 52 insertions, 93 deletions

diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_EC.cpp b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_EC.cpp
index a6aedbbd1e0..605d63e8092 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_EC.cpp
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_EC.cpp
@@ -275,8 +275,8 @@ main (int argc, char* argv[])
 
       // The schedule is returned in this variables....
 
-      consumer_ec.start(ACE_ENV_SINGLE_ARG_PARAMETER);
-      ACE_TRY_CHECK;
+      //consumer_ec.start(ACE_ENV_SINGLE_ARG_PARAMETER);
+      //ACE_TRY_CHECK;
 
       // ****************************************************************
 
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_Supplier_EC.cpp b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_Supplier_EC.cpp
index 1ad65d440f0..7130fabaffc 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_Supplier_EC.cpp
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_Supplier_EC.cpp
@@ -246,22 +246,22 @@ main (int argc, char* argv[])
 
       // ****************************************************************
 
-      Consumer_Supplier_EC supplier_ec;
-      if (supplier_ec.init(sched_type.c_str(), poa.in()) == -1)
+      Consumer_Supplier_EC cons_supp_ec;
+      if (cons_supp_ec.init(sched_type.c_str(), poa.in()) == -1)
         {
           ACE_ERROR_RETURN((LM_ERROR, "Unable to initialize Kokyu_EC"), 1);
         }
 
-      supplier_ec.init_gateway(orb.in(),
+      cons_supp_ec.init_gateway(orb.in(),
                                poa.in(),
                                "file://consumer_ec.ior" ACE_ENV_ARG_PARAMETER);
       ACE_TRY_CHECK;
       // ****************************************************************
-      RtEventChannelAdmin::RtSchedEventChannel_var supplier_ec_ior =
-        supplier_ec._this(ACE_ENV_SINGLE_ARG_PARAMETER);
+      RtEventChannelAdmin::RtSchedEventChannel_var cons_supp_ec_ior =
+        cons_supp_ec._this(ACE_ENV_SINGLE_ARG_PARAMETER);
       ACE_TRY_CHECK;
 
-      CORBA::String_var ior = orb->object_to_string(supplier_ec_ior.in()
+      CORBA::String_var ior = orb->object_to_string(cons_supp_ec_ior.in()
                                                     ACE_ENV_ARG_PARAMETER);
 
       ACE_OS::fprintf(ior_output_file, ior.in());
@@ -274,9 +274,9 @@ main (int argc, char* argv[])
       int prio = ACE_Sched_Params::priority_max (ACE_SCHED_FIFO);
       ACE_OS::thr_setprio (thr_handle, prio);
 
-      supplier_ec.set_up_supp_and_cons(ACE_ENV_SINGLE_ARG_PARAMETER);
+      cons_supp_ec.set_up_supp_and_cons(ACE_ENV_SINGLE_ARG_PARAMETER);
       ACE_TRY_CHECK;
-      //supplier_ec.start(ACE_ENV_SINGLE_ARG_PARAMETER);
+      //cons_supp_ec.start(ACE_ENV_SINGLE_ARG_PARAMETER);
       //ACE_TRY_CHECK;
 
       //@BT: Timeouts start when orb starts, similar to starting the DT worker thread
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Kokyu_EC.cpp b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Kokyu_EC.cpp
index 4d519a88ef2..9d66d8d7a97 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Kokyu_EC.cpp
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Kokyu_EC.cpp
@@ -134,13 +134,14 @@ Kokyu_EC::register_consumer (
   ACE_CHECK;
 
   ACE_ConsumerQOS_Factory consumer_qos1;
+  consumer_qos1.start_disjunction_group(types.size());
 
   for(EventType_Vector::Iterator iter(types);
       !iter.done(); iter.advance())
     {
       EventType_Vector::TYPE *type; //would rather const to ensure we don't change it, but not supported!
       iter.next(type);
-      //ACE_DEBUG((LM_DEBUG,"Kokyu_EC register_consumer() registering event type %d\n",*type));
+      ACE_DEBUG((LM_DEBUG,"Kokyu_EC register_consumer() registering event type %d\n",*type));
       if (*type != ACE_ES_EVENT_INTERVAL_TIMEOUT)
         {
           consumer_qos1.insert_type (*type,  consumer1_rt_info);
@@ -151,7 +152,7 @@ Kokyu_EC::register_consumer (
                                      info.period, //in 100s of nanosec
                                      consumer1_rt_info);
         }
-      //ACE_DEBUG((LM_DEBUG,"Kokyu_EC register_consumer() registered event type\n"));
+      ACE_DEBUG((LM_DEBUG,"Kokyu_EC register_consumer() registered event type\n"));
     }
 
   proxy_supplier =
@@ -263,6 +264,12 @@ Kokyu_EC::start (ACE_ENV_SINGLE_ARG_DECL)
       ec_impl_->activate (ACE_ENV_SINGLE_ARG_PARAMETER);
       ACE_CHECK;
 
+      ACE_Scheduler_Factory::dump_schedule (infos.in (),
+                                            dependencies.in (),
+                                            configs.in (),
+                                            anomalies.in (),
+                                            "schedule.out");
+
       //@BT: EC activated is roughly equivalent to having the DT scheduler ready to run
       //DSTRM_EVENT (MAIN_GROUP_FAM, SCHEDULER_STARTED, 1, 0, NULL);
       ACE_DEBUG((LM_DEBUG,"Kokyu_EC thread %t SCHEDULER_STARTED at %u\n",ACE_OS::gettimeofday().msec()));
@@ -351,7 +358,7 @@ Kokyu_EC::add_timeout_consumer(
   ACE_CHECK;
   //don't need to save supplier_timeout_consumer_rt_info because only used to set dependency here:
 
-  for(Supplier::RT_Info_Vector::Iterator iter(supplier_impl->all_rt_infos());
+  for(Supplier::RT_Info_Vector::Iterator iter(supplier_impl->rt_info());
       !iter.done(); iter.advance())
     {
       Supplier::RT_Info_Vector::TYPE *info; //would rather const to ensure we don't change it, but not supported!
@@ -394,9 +401,10 @@ Kokyu_EC::add_supplier(
   supplier = supplier_impl->_this(ACE_ENV_SINGLE_ARG_PARAMETER);
   ACE_CHECK;
 
-  EventType_Vector one_type(1);
   Supplier::RT_Info_Vector rt_infos(types.size());
   Supplier::PushConsumer_Vector proxies(types.size());
+  /*
+  EventType_Vector one_type(1);
   for(EventType_Vector::Iterator iter(types);
       !iter.done(); iter.advance())
     {
@@ -409,25 +417,26 @@ Kokyu_EC::add_supplier(
       entry << entry_point << ":" << *type;
 
       ACE_DEBUG((LM_DEBUG,"Kokyu_EC add_supplier() registering supplier for type %d\n",*type));
-
+  */
       RtecScheduler::handle_t supplier_rt_info =
-        this->register_supplier(entry.str().c_str(),
+        this->register_supplier(entry_point,//entry.str().c_str(),
                                 supplier_id,
-                                one_type,
+                                types,//one_type,
                                 supplier.in(),
                                 consumer_proxy.out()
                                 ACE_ENV_ARG_PARAMETER);
       ACE_CHECK;
 
-      ACE_DEBUG((LM_DEBUG,"Kokyu_EC add_supplier() consumer_proxy: %@ with RT_Info %s [%d]\n",consumer_proxy.in(),entry.str().c_str(),supplier_rt_info));
+      //ACE_DEBUG((LM_DEBUG,"Kokyu_EC add_supplier() consumer_proxy: %@ with RT_Info %s [%d]\n",consumer_proxy.in(),entry.str().c_str(),supplier_rt_info));
 
       rt_infos.push_back(supplier_rt_info);
       proxies.push_back(consumer_proxy); //proxies has a _var ref to  Consumer_Proxy so won't be deleted on return
 
-      supplier_impl->set_consumer_proxies(*type,rt_infos,proxies);
-    }
+      //supplier_impl->set_consumer_proxies(*type,rt_infos,proxies);
+      supplier_impl->set_consumer_proxy(proxies);
+      //}
 
-  //supplier_impl->rt_info(rt_infos);
+  supplier_impl->rt_info(rt_infos);
 
   this->suppliers_.push_back(supplier_impl);
 } //add_supplier()
@@ -465,7 +474,7 @@ Kokyu_EC::add_consumer_with_supplier(
       Consumer::RT_Info_Vector::TYPE *cons_info;
       citer.next(cons_info);
 
-      for(Supplier::RT_Info_Vector::Iterator iter(supplier_impl->all_rt_infos());
+      for(Supplier::RT_Info_Vector::Iterator iter(supplier_impl->rt_info());
           !iter.done(); iter.advance())
         {
           Supplier::RT_Info_Vector::TYPE *supp_info;
@@ -514,8 +523,9 @@ Kokyu_EC::add_consumer(
   info.threads = 0;
   info.info_type = RtecScheduler::OPERATION;
 
-  EventType_Vector one_type(1);
   Consumer::RT_Info_Vector rt_infos(types.size());
+  /*
+  EventType_Vector one_type(1);
   for(EventType_Vector::Iterator iter(types);
       !iter.done(); iter.advance())
     {
@@ -528,20 +538,20 @@ Kokyu_EC::add_consumer(
       entry << entry_point << ":" << *type;
 
       ACE_DEBUG((LM_DEBUG,"Kokyu_EC add_consumer() registering consumer for type %d\n",*type));
-
+  */
       RtecScheduler::handle_t consumer_rt_info =
-        this->register_consumer(entry.str().c_str(),
+        this->register_consumer(entry_point,//entry.str().c_str(),
                                 info,
-                                one_type,
+                                types,//one_type,
                                 consumer.in(),
                                 proxy_supplier.out()
                                 ACE_ENV_ARG_PARAMETER);
       ACE_CHECK;
 
-      ACE_DEBUG((LM_DEBUG,"Kokyu_EC add_consumer() registered consumer with RT_Info %s [%d]\n",entry.str().c_str(),consumer_rt_info));
+      //ACE_DEBUG((LM_DEBUG,"Kokyu_EC add_consumer() registered consumer with RT_Info %s [%d]\n",entry.str().c_str(),consumer_rt_info));
 
       rt_infos.push_back(consumer_rt_info);
-    }
+      //}
 
   //ACE_DEBUG((LM_DEBUG,"Kokyu_EC add_consumer() setting RT_Infos and storing consumer_impl\n"));
 
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.cpp b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.cpp
index 8d7d5406d2c..d182e1e9062 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.cpp
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.cpp
@@ -33,29 +33,10 @@ Supplier::~Supplier (void)
 {
 }
 
-//void
-//Supplier::set_consumer_proxy(PushConsumer_Vector consumer_proxies)
 void
-Supplier::set_consumer_proxies(RtecEventComm::EventType type, RT_Info_Vector& rt_infos,
-                               PushConsumer_Vector& consumer_proxies)
+Supplier::set_consumer_proxy(PushConsumer_Vector consumer_proxies)
 {
-  PushConsumer_Vector *push_consumers;
-  RT_Info_Vector *infos;
-
-  if (type == this->ft_type_)
-    {
-      push_consumers = &(this->ft_consumer_proxies_);
-      infos = &(this->ft_rt_infos_);
-    }
-  else //NORMAL
-    {
-      push_consumers = &(this->normal_consumer_proxies_);
-      infos = &(this->normal_rt_infos_);
-    }
-
-  this->all_rt_infos_.clear();
-  infos->clear();
-  push_consumers->clear();
+  this->consumer_proxy_.clear();
 
   for(PushConsumer_Vector::Iterator iter(consumer_proxies);
       !iter.done(); iter.advance())
@@ -63,41 +44,20 @@ Supplier::set_consumer_proxies(RtecEventComm::EventType type, RT_Info_Vector& rt
       PushConsumer_Vector::TYPE *proxy; //would rather const to ensure we don't change it, but not supported!
       iter.next(proxy);
 
-      push_consumers->push_back(*proxy);
-    }
-  for(RT_Info_Vector::Iterator iter(rt_infos);
-      !iter.done(); iter.advance())
-    {
-      RT_Info_Vector::TYPE *info; //would rather const to ensure we don't change it, but not supported!
-      iter.next(info);
-
-      infos->push_back(*info);
-      this->all_rt_infos_.push_back(*info);
+      this->consumer_proxy_.push_back(*proxy);
     }
 }
-/*
+
 void
 Supplier::rt_info(RT_Info_Vector& supplier_rt_info)
 {
   this->rt_info_ = supplier_rt_info;
 }
-*/
-Supplier::RT_Info_Vector&
-Supplier::normal_rt_infos(void)
-{
-  return this->normal_rt_infos_;
-}
 
 Supplier::RT_Info_Vector&
-Supplier::ft_rt_infos(void)
+Supplier::rt_info(void)
 {
-  return this->ft_rt_infos_;
-}
-
-Supplier::RT_Info_Vector&
-Supplier::all_rt_infos(void)
-{
-  return this->all_rt_infos_;
+  return this->rt_info_;
 }
 
 void
@@ -110,8 +70,6 @@ Supplier::timeout_occured (ACE_ENV_SINGLE_ARG_DECL)
       ACE_DEBUG((LM_DEBUG,"Supplier (%P|%t) handle_service_start() DONE\n"));
     }
 
-  PushConsumer_Vector *proxies = 0;
-
   RtecEventComm::EventSet event (1);
   event.length (1);
   event[0].header.source = id_;
@@ -120,13 +78,11 @@ Supplier::timeout_occured (ACE_ENV_SINGLE_ARG_DECL)
   case FAULT_TOLERANT:
     {
       event[0].header.type   = this->ft_type_;
-      proxies = &(this->ft_consumer_proxies_);
       break;
     }
   default: //NORMAL
     {
       event[0].header.type   = this->norm_type_;
-      proxies = &(this->normal_consumer_proxies_);
       break;
     }
   }
@@ -147,7 +103,7 @@ Supplier::timeout_occured (ACE_ENV_SINGLE_ARG_DECL)
   ACE_DEBUG((LM_DEBUG,"Supplier (id %d) in thread %t ONE_WAY_CALL_START at %u\n",this->id_,ACE_OS::gettimeofday().msec()));
   DSTRM_EVENT (WORKER_GROUP_FAM, ONE_WAY_CALL_START, 0, sizeof(Object_ID), (char*)&oid);
 
-  for(PushConsumer_Vector::Iterator iter(*proxies);
+  for(PushConsumer_Vector::Iterator iter(this->consumer_proxy_);
       !iter.done(); iter.advance())
     {
       PushConsumer_Vector::TYPE *proxy; //would rather const to ensure we don't change it, but not supported!
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.h b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.h
index be26a548d6c..44adcaa76b5 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.h
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.h
@@ -52,7 +52,6 @@ public:
   };
 
   typedef ACE_Vector<RtecEventChannelAdmin::ProxyPushConsumer_var> PushConsumer_Vector;
-  typedef ACE_Vector<RtecScheduler::handle_t> RT_Info_Vector;
 
   Supplier (RtecEventComm::EventSourceID id, RtecEventComm::EventType norm_type, RtecEventComm::EventType ft_type,
             Service_Handler * handler = 0);
@@ -68,15 +67,12 @@ public:
 
   virtual void timeout_occured (ACE_ENV_SINGLE_ARG_DECL);
 
-  void set_consumer_proxies(RtecEventComm::EventType type, RT_Info_Vector& rt_infos,
-                            PushConsumer_Vector& consumer_proxies);
-  //void set_consumer_proxies(RtecScheduler::handle_t rt_info, PushConsumer_Vector consumer_proxies);
+  void set_consumer_proxy(PushConsumer_Vector consumer_proxies);
+
+  typedef ACE_Vector<RtecScheduler::handle_t> RT_Info_Vector;
 
-  //void normal_rt_info(RT_Info_Vector& supplier_rt_info);
-  //void ft_rt_info(RT_Info_Vector& supplier_rt_info);
-  RT_Info_Vector& normal_rt_infos(void);
-  RT_Info_Vector& ft_rt_infos(void);
-  RT_Info_Vector& all_rt_infos(void);
+  void rt_info(RT_Info_Vector& supplier_rt_info);
+  RT_Info_Vector& rt_info(void);
 
   RtecEventComm::EventSourceID get_id(void) const;
 
@@ -90,14 +86,11 @@ protected:
   RtecEventComm::EventType norm_type_;
   RtecEventComm::EventType ft_type_;
 
-  PushConsumer_Vector normal_consumer_proxies_;
-  PushConsumer_Vector ft_consumer_proxies_;
+  PushConsumer_Vector consumer_proxy_;
 
   mode_t mode_;
 
-  RT_Info_Vector all_rt_infos_;
-  RT_Info_Vector normal_rt_infos_;
-  RT_Info_Vector ft_rt_infos_;
+  RT_Info_Vector rt_info_;
 
   Service_Handler *handler_;
 }; //class Supplier