Block running of client and server until both fully configured. Client does not shutdown the ORB because it would do so immediately (having no suppliers to block the acquire_write)

5 files changed, 50 insertions, 8 deletions

diff --git a/TAO/orbsvcs/examples/RtEC/test_driver/Consumer.cpp b/TAO/orbsvcs/examples/RtEC/test_driver/Consumer.cpp
index 20b3262f6ef..e5087d85037 100644
--- a/TAO/orbsvcs/examples/RtEC/test_driver/Consumer.cpp
+++ b/TAO/orbsvcs/examples/RtEC/test_driver/Consumer.cpp
@@ -149,7 +149,8 @@ Consumer::connect_impl (bool set_rtinfo, //true if should set RT_Info
   ACE_DEBUG((LM_DEBUG,"\tEvent type: %d\n",event_type));
 
   //calibrate
-  this->_work.calibrate();
+  //TODO: calibrate seems to cause Consumer's connect_push_consumer() to hang!
+  //this->_work.calibrate();
 }
 
 void
@@ -203,6 +204,7 @@ Consumer::push (const RtecEventComm::EventSet& events
   //when work triggered by event starts.
 
   //do work
+  //what happens when run w/o calibratin?
   timeval load = (timeval) this->_work_time;
   this->_work.run(load);
 
diff --git a/TAO/orbsvcs/examples/RtEC/test_driver/ECConfig.cpp b/TAO/orbsvcs/examples/RtEC/test_driver/ECConfig.cpp
index ce50bad3e54..6ed00a5b387 100644
--- a/TAO/orbsvcs/examples/RtEC/test_driver/ECConfig.cpp
+++ b/TAO/orbsvcs/examples/RtEC/test_driver/ECConfig.cpp
@@ -35,6 +35,9 @@ namespace TestConfig {
   struct spawn_data_t {
     ACE_RW_Mutex *lock;
     CORBA::ORB_var *orb;
+    int *ready;
+    int use_federated;
+    int is_server;
   };
 
 //TODO: Obviously, we can't just hardcode these!
@@ -67,6 +70,7 @@ ECConfig<SCHED_STRAT>::ECConfig (void)
   , configured (0) //false
   , use_federated (1) //TODO Check whether or not FEDERATED; default to true
   //, use_federated (0) //TODO Check whether or not FEDERATED; default to false
+  , ready (0)
 {
   this->test_done = new ACE_RW_Mutex();
 }
@@ -146,6 +150,7 @@ ECConfig<SCHED_STRAT>::reset (ACE_ENV_SINGLE_ARG_DECL)
   //TODO clear RT_Infos from scheduler?
 
   configured = 0; //false
+  this->ready = 0;
 }
 
 template <class SCHED_STRAT> int
@@ -395,6 +400,8 @@ ECConfig<SCHED_STRAT>::run (void)
     return 1;
   }
 
+  this->ready = 0; //ensure suppliers don't start when orb runs yet
+
   ACE_TRY
     {
       ACE_Thread_Manager *inst = ACE_Thread_Manager::instance();
@@ -410,6 +417,9 @@ ECConfig<SCHED_STRAT>::run (void)
       printf("data->lock = %p\n",(void*)(data->lock));
       data->orb = &(this->orb);
       printf("data->orb = %p\n",(void*)(data->orb));
+      data->ready = &(this->ready);
+      data->use_federated = this->use_federated;
+      data->is_server = this->suppliers.size()>0; //assume client if no suppliers
       int ret = inst->spawn(ECConfig<SCHED_STRAT>::run_orb,data);
       //int ret = inst->spawn(ECConfig<SCHED_STRAT>::run_orb,reactor);
       //no need for getting tid?
@@ -615,7 +625,8 @@ ECConfig<SCHED_STRAT>::connect_suppliers (ACE_ENV_SINGLE_ARG_DECL)
           entry_prefix << "Supplier " << supp_idx;
 
           ACE_DEBUG((LM_DEBUG,"Supplier.connect() for %s\n",entry_prefix.str().c_str()));
-          this->suppliers[supp_idx]->connect (this->test_done,
+          this->suppliers[supp_idx]->connect (&(this->ready),
+                                              this->test_done,
                                               this->scheduler.in(),
                                               entry_prefix.str().c_str(),
                                               this->periods[i], //period
@@ -821,6 +832,18 @@ ECConfig<SCHED_STRAT>::run_orb(void *data)
   //printf("acquire_write(): %d\n",test_done->acquire_write());
   //std::exit(0);
 
+  if (data_ptr->use_federated)
+    {
+      //assume if have no suppliers that we are client
+      ACE_DEBUG((LM_DEBUG,"Barrier to wait until both apps configured and orbs running\n"));
+      ECConfig<SCHED_STRAT>::barrier(data_ptr->is_server); //wait until both apps are ready to run
+      *(data_ptr->ready) = 1; //checked by suppliers to start reacting to timeouts
+    }
+  else
+    {
+      ACE_DEBUG((LM_DEBUG,"No barrier after configured&orbrun because not federated\n"));
+    }
+
   //      Block waiting for consumers to finish
   //when can acquire write lock, all Suppliers are finished
   int ret = data_ptr->lock->acquire_write();
@@ -835,8 +858,14 @@ ECConfig<SCHED_STRAT>::run_orb(void *data)
 
   //REACTOR CHANGE
   // Shutdown ORB
-  (*(data_ptr->orb))->shutdown(1); //argument is TRUE so orb waits until work
-                              //done before shutting down
+  //TODO: Client stops immediately when it should wait for gateway to finish
+  if (data_ptr->is_server)
+    {
+      //HACK: client doesn't shutdown!
+      ACE_DEBUG((LM_DEBUG,"Supplier shutting down ORB\n"));
+      (*(data_ptr->orb))->shutdown(1); //argument is TRUE so orb waits until work
+      //done before shutting down
+    }
   /*
   //orb->orb_core()->reactor()->end_reactor_event_loop();
   ACE_DEBUG((LM_DEBUG,"DONE; stopping reactor event loop\n"));
diff --git a/TAO/orbsvcs/examples/RtEC/test_driver/ECConfig.h b/TAO/orbsvcs/examples/RtEC/test_driver/ECConfig.h
index 5930d13fc1d..2a322b81448 100644
--- a/TAO/orbsvcs/examples/RtEC/test_driver/ECConfig.h
+++ b/TAO/orbsvcs/examples/RtEC/test_driver/ECConfig.h
@@ -130,10 +130,13 @@ private:
 
   RtecEventChannelAdmin::Observer_var gateway_obs;
 
+  //Becomes true when client signals (through barrier) that it is ready to run
+  int ready; //boolean
+
   ///Blocks the application using ACE_SOCK_Connector or
   ///ACE_SOCK_Acceptor.  Used for barrier synchronization between the
   ///application and others in the system.
-  void barrier(bool is_supplier);
+  static void barrier(bool is_supplier);
 };
 
 } /* namespace TestConfig */
diff --git a/TAO/orbsvcs/examples/RtEC/test_driver/Supplier.cpp b/TAO/orbsvcs/examples/RtEC/test_driver/Supplier.cpp
index 9698626f935..660e31e428a 100644
--- a/TAO/orbsvcs/examples/RtEC/test_driver/Supplier.cpp
+++ b/TAO/orbsvcs/examples/RtEC/test_driver/Supplier.cpp
@@ -9,6 +9,7 @@ ACE_RCSID(EC_Examples, Supplier, "$Id$")
 Supplier::Supplier (void)
   : timeoutconsumer(this)
   , _supplier(this)
+  , _ready (0)
 {
 }
 
@@ -21,7 +22,8 @@ Supplier::update(ACE_ENV_SINGLE_ARG_DECL)
 {
   ACE_DEBUG((LM_DEBUG,"Supplier %d (%P|%t) received update\n",this->_supplier_id));
 
-  if (this->_num_sent < this->_to_send)
+  //only react to update if ready=1
+  if (*this->_ready == 1 && this->_num_sent < this->_to_send)
     {
       //@BT INSTRUMENT with event ID: EVENT_PUSH Measure time
       //when event is pushed by client.
@@ -47,7 +49,8 @@ Supplier::update(ACE_ENV_SINGLE_ARG_DECL)
 }
 
 void
-Supplier::connect (ACE_RW_Mutex* done,
+Supplier::connect (int *ready,
+                   ACE_RW_Mutex* done,
                    RtecScheduler::Scheduler_ptr scheduler,
                    const char *entry_prefix,
                    TimeBase::TimeT period,
@@ -59,6 +62,7 @@ Supplier::connect (ACE_RW_Mutex* done,
                    RtecEventChannelAdmin::EventChannel_ptr ec
                    ACE_ENV_ARG_DECL)
 {
+  this->_ready = ready;
   this->_supplier_id = supplier_id;
   this->_to_send = to_send;
   this->_num_sent = 0;
diff --git a/TAO/orbsvcs/examples/RtEC/test_driver/Supplier.h b/TAO/orbsvcs/examples/RtEC/test_driver/Supplier.h
index 3d31dce16d3..0c5887c3931 100644
--- a/TAO/orbsvcs/examples/RtEC/test_driver/Supplier.h
+++ b/TAO/orbsvcs/examples/RtEC/test_driver/Supplier.h
@@ -51,7 +51,8 @@ public:
 
   virtual void update(ACE_ENV_SINGLE_ARG_DECL);
 
-  void connect (ACE_RW_Mutex* done,
+  void connect (int *ready, //variable to check for readiness
+                ACE_RW_Mutex* done,
                 RtecScheduler::Scheduler_ptr scheduler,
                 const char *entry_prefix,
                 TimeBase::TimeT period,
@@ -93,6 +94,9 @@ private:
 
   RtecEventComm::EventSet _events;
   // set of events to push when a timeout event is received.
+
+  int *_ready; //do not respond to timeouts until this variable = 1
+
 };
 
 #endif /* SUPPLIER_H */