TAO/tests/LongUpcalls/AMI_Manager.cpp

   1
   2 #include "AMI_Manager.h"
   3 #include "ace/OS_NS_unistd.h"
   4
   5 static void
   6 validate_connection (Test::Controller_ptr controller)
   7 {
   8   try
   9     {
  10 #if (TAO_HAS_CORBA_MESSAGING == 1)
  11       CORBA::PolicyList_var unused;
  12       controller->_validate_connection (unused);
  13 #else
  14       controller->_is_a ("Not_an_IDL_Type");
  15 #endif
  16     }
  17   catch (const CORBA::Exception&)
  18     {
  19     }
  20 }
  21
  22 AMI_Manager::AMI_Manager (CORBA::ORB_ptr orb)
  23   : orb_ (CORBA::ORB::_duplicate (orb))
  24 {
  25 }
  26
  27 void
  28 AMI_Manager::start_workers (CORBA::Short worker_count,
  29                             CORBA::Long milliseconds,
  30                             Test::Controller_ptr controller)
  31 {
  32   ACE_Thread_Manager thread_manager;
  33
  34   validate_connection(controller);
  35
  36   // ACE_DEBUG ((LM_DEBUG, "Starting %d workers\n", worker_count));
  37   Worker worker (&thread_manager,
  38                  controller, milliseconds,
  39                  this->orb_.in ());
  40
  41   worker.activate (THR_NEW_LWP | THR_JOINABLE, worker_count);
  42
  43   thread_manager.wait ();
  44 }
  45
  46 void
  47 AMI_Manager::shutdown ()
  48 {
  49   this->orb_->shutdown (false);
  50 }
  51
  52 // ****************************************************************
  53
  54 Worker::Worker (ACE_Thread_Manager *thr_mgr,
  55                 Test::Controller_ptr controller,
  56                 CORBA::Long milliseconds,
  57                 CORBA::ORB_ptr orb)
  58   :  ACE_Task_Base (thr_mgr)
  59   ,  controller_ (Test::Controller::_duplicate (controller))
  60   ,  milliseconds_ (milliseconds)
  61   ,  orb_ (CORBA::ORB::_duplicate (orb))
  62 {
  63 }
  64
  65 int
  66 Worker::svc ()
  67 {
  68   // ACE_DEBUG ((LM_DEBUG, "Worker starts\n"));
  69   try
  70     {
  71       TAO_SYNCH_MUTEX mutex;
  72       int pending_requests = 2;
  73       Test::AMI_ControllerHandler_var handler;
  74
  75       {
  76         Controller_Handler *handler_impl;
  77         ACE_NEW_RETURN (handler_impl,
  78                         Controller_Handler (&mutex,
  79                                             &pending_requests),
  80                         -1);
  81
  82         CORBA::Object_var poa_object =
  83           this->orb_->resolve_initial_references("RootPOA");
  84
  85         PortableServer::POA_var root_poa =
  86           PortableServer::POA::_narrow (poa_object.in ());
  87
  88         PortableServer::ObjectId_var id =
  89           root_poa->activate_object (handler_impl);
  90
  91         CORBA::Object_var object = root_poa->id_to_reference (id.in ());
  92
  93         PortableServer::ServantBase_var auto_destroy (handler_impl);
  94         handler = Test::AMI_ControllerHandler::_narrow (object.in ());
  95       }
  96
  97       validate_connection(this->controller_.in());
  98
  99       this->controller_->sendc_worker_started (handler.in ());
 100
 101       // ACE_DEBUG ((LM_DEBUG, "Worker start reported\n"));
 102
 103       ACE_Time_Value tv (0, 1000 * this->milliseconds_);
 104       ACE_OS::sleep (tv);
 105
 106       this->controller_->sendc_worker_finished (handler.in ());
 107
 108       // ACE_DEBUG ((LM_DEBUG, "Worker completion reported\n"));
 109
 110       for (;;)
 111         {
 112           ACE_Time_Value tv (0, 1000 * this->milliseconds_);
 113           this->orb_->run (tv);
 114
 115           ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, mutex, -1);
 116           if (pending_requests == 0)
 117             break;
 118         }
 119     }
 120   catch (const CORBA::Exception& ex)
 121     {
 122       ex._tao_print_exception ("Exception in svc() method\n");
 123     }
 124   return 0;
 125 }
 126
 127 // ****************************************************************
 128
 129 Controller_Handler::Controller_Handler (TAO_SYNCH_MUTEX *mutex,
 130                                         int *pending_replies)
 131   :  mutex_ (mutex)
 132   ,  pending_replies_ (pending_replies)
 133 {
 134 }
 135
 136 void
 137 Controller_Handler::worker_started ()
 138 {
 139   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, *this->mutex_);
 140   (*this->pending_replies_)--;
 141 }
 142
 143 void
 144 Controller_Handler::worker_started_excep
 145     (::Messaging::ExceptionHolder* h)
 146 {
 147   try
 148     {
 149       h->raise_exception ();
 150     }
 151   catch (const CORBA::Exception& ex)
 152     {
 153       ex._tao_print_exception (
 154         "Controller_Handler exception raised in"" worker_started");
 155     }
 156 }
 157
 158 void
 159 Controller_Handler::worker_finished ()
 160 {
 161   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, *this->mutex_);
 162   (*this->pending_replies_)--;
 163 }
 164
 165 void
 166 Controller_Handler::worker_finished_excep
 167     (::Messaging::ExceptionHolder *h)
 168 {
 169   try
 170     {
 171       h->raise_exception ();
 172     }
 173   catch (const CORBA::Exception& ex)
 174     {
 175       ex._tao_print_exception (
 176         "Controller_Handler exception raised in"" worker_finished");
 177     }
 178 }