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