TAO/tests/Big_Twoways/Session.cpp

   1 #include "Session.h"
   2 #include "tao/debug.h"
   3
   4 Session::Session (Test::Session_Control_ptr control,
   5                   CORBA::ULong payload_size,
   6                   CORBA::ULong thread_count,
   7                   CORBA::ULong message_count,
   8                   CORBA::ULong peer_count)
   9   : control_ (Test::Session_Control::_duplicate (control))
  10   , running_ (0)
  11   , payload_size_ (payload_size)
  12   , thread_count_ (thread_count)
  13   , message_count_ (message_count)
  14   , active_thread_count_ (0)
  15   , expected_messages_ (thread_count * message_count * (peer_count - 1))
  16   , task_ (this)
  17   , barrier_ (thread_count + 1)
  18 {
  19 }
  20
  21 Session::~Session ()
  22 {
  23 }
  24
  25 int
  26 Session::svc ()
  27 {
  28   this->barrier_.wait ();
  29
  30   /// Automatically decrease the reference count at the end of the
  31   /// thread
  32   PortableServer::ServantBase_var auto_decrement (this);
  33   CORBA::ULong i = 0;
  34
  35   try
  36     {
  37       // Use the same payload over and over
  38       Test::Payload payload (this->payload_size_);
  39       payload.length (this->payload_size_);
  40
  41       // Get the number of peers just once.
  42       CORBA::ULong session_count =
  43         this->other_sessions_.length ();
  44
  45       for (; i != this->message_count_; ++i)
  46         {
  47 #if 0
  48           if (i % 500 == 0)
  49             {
  50               ACE_DEBUG ((LM_DEBUG,
  51                           "(%P|%t) Session::svc, "
  52                           "sending message %d\n",
  53                           i));
  54             }
  55 #endif /* 0 */
  56           for (CORBA::ULong j = 0; j != session_count; ++j)
  57             {
  58               Test::Payload_var received =
  59                 this->other_sessions_[j]->echo_payload (payload);
  60             }
  61         }
  62
  63       {
  64         ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->mutex_, -1);
  65         this->active_thread_count_--;
  66         if (this->more_work ())
  67           {
  68             return 0;
  69           }
  70       }
  71       this->terminate (1);
  72     }
  73   catch (const CORBA::Exception& ex)
  74     {
  75       ACE_ERROR ((LM_ERROR,
  76                           "(%P|%t) ERROR: Session::svc, "
  77                           "send %d messages out of %d\n",
  78                           i, message_count_));
  79       ex._tao_print_exception ("Session::svc - ");
  80       return -1;
  81     }
  82
  83   return 0;
  84 }
  85
  86 void
  87 Session::start (const Test::Session_List &other_sessions)
  88 {
  89   if (other_sessions.length () == 0)
  90     throw Test::No_Peers ();
  91
  92   {
  93     ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->mutex_);
  94     if (this->running_)
  95       throw Test::Already_Running ();
  96
  97     this->other_sessions_ = other_sessions;
  98
  99     for (CORBA::ULong i = 0; i != this->thread_count_; ++i)
 100       {
 101         // Increase the reference count because the new thread will have
 102         // access to this object....
 103         try
 104           {
 105             this->_add_ref ();
 106
 107             if (this->task_.activate (
 108                     THR_NEW_LWP | THR_JOINABLE, 1, 1) == -1)
 109               {
 110                 this->_remove_ref ();
 111               }
 112             else
 113               {
 114                 this->running_ = 1;
 115                 this->active_thread_count_++;
 116               }
 117           }
 118         catch (const CORBA::Exception& ex)
 119           {
 120             ex._tao_print_exception ("Session::start, ignored");
 121           }
 122       }
 123
 124     if (this->active_thread_count_ != this->thread_count_)
 125       return;
 126   }
 127
 128   this->validate_connections ();
 129
 130   this->barrier_.wait ();
 131
 132   if (this->running_ != 0)
 133     return;
 134
 135   /// None of the threads are running, this session is useless at
 136   /// this point, report the problem and destroy the local objects
 137   this->terminate (0);
 138 }
 139
 140 Test::Payload *
 141 Session::echo_payload (const Test::Payload &the_payload)
 142 {
 143   if (the_payload.length () != this->payload_size_)
 144     {
 145       ACE_ERROR ((LM_ERROR,
 146                   "ERROR: (%P|%t) Session::echo_payload, "
 147                   "unexpected payload size (%d != %d)\n",
 148                   the_payload.length (), this->payload_size_));
 149     }
 150
 151   Test::Payload_var retval (new Test::Payload (the_payload));
 152
 153   {
 154     ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->mutex_,
 155                       retval._retn ());
 156     this->expected_messages_--;
 157
 158     int verbose = 0;
 159 #if 0
 160     verbose = this->expected_messages_ % 500 == 0;
 161     if (this->expected_messages_ < 500)
 162       verbose = (this->expected_messages_ % 100 == 0);
 163     if (this->expected_messages_ < 100)
 164     verbose = (this->expected_messages_ % 10 == 0);
 165 #endif /* 0 */
 166     if (this->expected_messages_ < 5)
 167       verbose = 1;
 168
 169     if (verbose)
 170       {
 171         ACE_DEBUG ((LM_DEBUG,
 172                     "(%P|%t) Session::echo_payload, "
 173                     "%d messages to go\n",
 174                     this->expected_messages_));
 175       }
 176     if (this->more_work ())
 177       return retval._retn ();
 178   }
 179   this->terminate (1);
 180
 181   return retval._retn ();
 182 }
 183
 184
 185 void
 186 Session::destroy ()
 187 {
 188   // Make sure local resources are released
 189
 190   PortableServer::POA_var poa =
 191     this->_default_POA ();
 192   PortableServer::ObjectId_var oid =
 193     poa->servant_to_id (this);
 194   poa->deactivate_object (oid.in ());
 195 }
 196
 197 int
 198 Session::more_work () const
 199 {
 200   if (this->expected_messages_ > 0
 201       || this->active_thread_count_ > 0
 202       || this->running_ == 0)
 203     return 1;
 204
 205   return 0;
 206 }
 207
 208 void
 209 Session::validate_connections ()
 210 {
 211   const CORBA::ULong session_count =
 212     this->other_sessions_.length ();
 213   for (CORBA::ULong j = 0; j != session_count; ++j)
 214     {
 215       try
 216         {
 217 #if (TAO_HAS_CORBA_MESSAGING == 1)
 218           CORBA::PolicyList_var unused;
 219           this->other_sessions_[j]->_validate_connection (unused);
 220 #else
 221           (void) this->other_sessions_[j]->_is_a ("Not_An_IDL_Type");
 222 #endif /* TAO_HAS_MESSAGING == 1 */
 223         }
 224       catch (const CORBA::Exception&)
 225         {
 226         }
 227     }
 228 }
 229
 230 void
 231 Session::terminate (CORBA::Boolean success)
 232 {
 233   // Make sure that global resources are released
 234   try
 235     {
 236       this->control_->session_finished (success);
 237     }
 238   catch (const CORBA::Exception& ex)
 239     {
 240       ex._tao_print_exception ("Session::terminate, ignored");
 241     }
 242
 243 }