TAO/orbsvcs/tests/Notify/lib/Periodic_Consumer.cpp

   1 #include "Periodic_Consumer.h"
   2
   3 #include "ace/Arg_Shifter.h"
   4 #include "ace/High_Res_Timer.h"
   5 #include "ace/Task.h"
   6 #include "tao/debug.h"
   7 #include "orbsvcs/Time_Utilities.h"
   8 #include "StructuredEvent.h"
   9 #include "Task_Stats.h"
  10 #include "Task_Callback.h"
  11 #include "LookupManager.h"
  12 #include "Priority_Mapping.h"
  13
  14 int WARMUP_COUNT = 10;
  15
  16 TAO_Notify_Tests_Periodic_Consumer::TAO_Notify_Tests_Periodic_Consumer ()
  17   : count_ (-2)
  18   , warmup_countdown_ (WARMUP_COUNT)
  19   , max_count_ (-1)
  20   , load_ (0)
  21   , client_ (0)
  22   , check_priority_ (0)
  23   , stop_received_ (0)
  24 {
  25 }
  26
  27 TAO_Notify_Tests_Periodic_Consumer::~TAO_Notify_Tests_Periodic_Consumer ()
  28 {
  29 }
  30
  31 void
  32 TAO_Notify_Tests_Periodic_Consumer::task_callback (TAO_Notify_Tests_Task_Callback* client)
  33 {
  34   this->client_ = client;
  35 }
  36
  37 int
  38 TAO_Notify_Tests_Periodic_Consumer::init_state (ACE_Arg_Shifter& arg_shifter)
  39 {
  40   // First, let the base class look for options.
  41   if (TAO_Notify_Tests_StructuredPushConsumer::init_state (arg_shifter) == -1)
  42     return -1;
  43
  44   const ACE_TCHAR *current_arg = 0;
  45
  46   while (arg_shifter.is_anything_left ())
  47     {
  48       if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-MaxCount"))))
  49         {
  50           this->max_count_ = ACE_OS::atoi (current_arg);
  51           arg_shifter.consume_arg ();
  52
  53           if (max_count_ == 0)
  54           {
  55             if (this->client_)
  56               this->client_->done (this);
  57           }
  58         }
  59        else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-Check_Priority")) == 0)
  60          {
  61           this->check_priority_ = 1;
  62
  63           arg_shifter.consume_arg ();
  64          }
  65        else
  66          {
  67            break;
  68          }
  69     } /* while */
  70
  71   return 0;
  72 }
  73
  74 void
  75 TAO_Notify_Tests_Periodic_Consumer::handle_start_event (const CosNotification::PropertySeq& prop_seq)
  76 {
  77   if (TAO_debug_level > 0)
  78     ACE_DEBUG ((LM_DEBUG, "(%P, %t)Consumer %s received inital (-1)th event\n", this->name_.c_str ()));
  79
  80   for (CORBA::ULong i = 0; i < prop_seq.length (); ++i)
  81     {
  82       if (ACE_OS::strcmp (prop_seq[i].name.in (), "BaseTime") == 0)
  83         {
  84           TimeBase::TimeT base_time;
  85           ACE_hrtime_t base_time_hrtime;
  86
  87           prop_seq[i].value >>= base_time;
  88
  89           ORBSVCS_Time::TimeT_to_hrtime (base_time_hrtime, base_time);
  90           stats_.base_time (base_time_hrtime);
  91         }
  92       // if max_count has not been already specified, get it from the supplier.
  93       else if (this->max_count_ == -1 &&
  94                ACE_OS::strcmp (prop_seq[i].name.in (), "MaxCount") == 0)
  95         {
  96           prop_seq[i].value >>= this->max_count_;
  97         }
  98       else if (ACE_OS::strcmp (prop_seq[i].name.in (), "Load") == 0)
  99         {
 100           prop_seq[i].value >>= this->load_;
 101         }
 102     }
 103
 104   if (TAO_debug_level > 0)
 105     {
 106       ACE_DEBUG ((LM_DEBUG, "(%P, %t) Maxcount = %d, Load = %d\n",
 107                   this->max_count_, this->load_));
 108     }
 109 }
 110
 111 void
 112 TAO_Notify_Tests_Periodic_Consumer::check_priority (const CosNotification::PropertySeq& prop_seq)
 113 {
 114   // Check if the event carries a Priority.
 115   int event_has_priority_set = 0;
 116   CORBA::Short event_priority = 0;
 117
 118   for (CORBA::ULong i = 0; i < prop_seq.length (); ++i)
 119     {
 120       if (ACE_OS::strcmp (prop_seq[i].name.in (), CosNotification::Priority) == 0)
 121         {
 122           prop_seq[i].value >>= event_priority;
 123
 124           event_has_priority_set = 1;
 125               break;
 126         }
 127     }
 128
 129   if (event_has_priority_set == 1)
 130     {
 131       // Confirm that the current thread is at the priority set in the event
 132       ACE_hthread_t current;
 133       ACE_Thread::self (current);
 134
 135       int priority;
 136       if (ACE_Thread::getprio (current, priority) == -1)
 137         {
 138           ACE_DEBUG ((LM_DEBUG,
 139                       ACE_TEXT ("TAO (%P|%t) - ")
 140                       ACE_TEXT (" ACE_Thread::get_prio\n")));
 141
 142           return;
 143         }
 144
 145       CORBA::Short native_priority = CORBA::Short (priority);
 146
 147       TAO_Notify_Tests_Priority_Mapping* priority_mapping;
 148       LOOKUP_MANAGER->resolve (priority_mapping);
 149
 150       CORBA::Short corba_priority;
 151
 152       priority_mapping->to_CORBA (native_priority, corba_priority);
 153
 154       if (TAO_debug_level > 0)
 155         ACE_DEBUG ((LM_DEBUG,
 156                     "Periodic Consumer expected priority = %d, received priority = %d/%d (native/corba)\n",
 157                     event_priority, native_priority, corba_priority));
 158
 159       if (corba_priority != event_priority)
 160         ACE_DEBUG ((LM_DEBUG,
 161                     "Error: Periodic Consumer expected priority = %d, received priority = %d\n",
 162                     event_priority, corba_priority));
 163     }
 164 }
 165
 166 void
 167 TAO_Notify_Tests_Periodic_Consumer::push_structured_event (const CosNotification::StructuredEvent & notification)
 168 {
 169   ACE_GUARD_THROW_EX (TAO_SYNCH_MUTEX, ace_mon, this->lock_,
 170                       CORBA::INTERNAL ());
 171
 172   const CosNotification::PropertySeq& prop_seq =
 173     notification.header.variable_header;
 174
 175   if (this->count_ == -2)
 176     {
 177       if (--warmup_countdown_ == 0)
 178         this->count_ = -1;
 179
 180       return;
 181     }
 182   else if (this->count_ == -1)
 183     {
 184       this->handle_start_event (prop_seq);
 185
 186       if (this->max_count_ > 0)
 187         this->stats_.init (this->max_count_);
 188
 189       this->count_ = 0;
 190       return;
 191     }
 192
 193   if (this->check_priority_)
 194     {
 195       this->check_priority (prop_seq);
 196     }
 197
 198   if (TAO_debug_level > 0)
 199     {
 200       ACE_DEBUG ((LM_DEBUG, "(%P, %t)Consumer %s received %d event type (%s,%s)\n", this->name_.c_str (), this->count_,
 201                   notification.header.fixed_header.event_type.domain_name.in(),
 202                   notification.header.fixed_header.event_type.type_name.in()));
 203     }
 204
 205   for (CORBA::ULong i = 0; i < prop_seq.length (); ++i)
 206     {
 207       if (ACE_OS::strcmp (prop_seq[i].name.in (), "Stop") == 0)
 208         {
 209           this->stop_received_ = 1;
 210         }
 211     }
 212
 213   TimeBase::TimeT send_time, now;
 214   ACE_hrtime_t send_time_hrtime;
 215
 216   notification.remainder_of_body >>= send_time;
 217
 218   ORBSVCS_Time::TimeT_to_hrtime (send_time_hrtime, send_time);
 219
 220   now = ACE_OS::gethrtime ();
 221
 222   stats_.sample (send_time_hrtime, now);
 223
 224   // Eat CPU
 225   static CORBA::ULong prime_number = 9619;
 226
 227   (void)ACE::gcd (prime_number, prime_number/2 -1);
 228
 229   for (CORBA::ULong load = this->load_; load != 0; --load)
 230     ACE::is_prime (prime_number,
 231                    2,
 232                    prime_number / 2);
 233
 234   // ---
 235
 236   if (++this->count_ >= this->max_count_ || this->stop_received_ == 1)
 237     {
 238       stats_.end_time (ACE_OS::gethrtime ());
 239
 240       if (this->client_)
 241         this->client_->done (this);
 242
 243       if (TAO_debug_level > 0)
 244         ACE_DEBUG ((LM_DEBUG, "(%P, %t)Consumer %s done\n", this->name_.c_str ()));
 245     }
 246 }
 247
 248 void
 249 TAO_Notify_Tests_Periodic_Consumer::dump_stats (ACE_TCHAR* msg, int dump_samples)
 250 {
 251   ACE_TCHAR buf[BUFSIZ];
 252   ACE_OS::sprintf (buf, ACE_TEXT("%s.dat"), this->name_.c_str ());
 253
 254   ACE_TString fname (buf);
 255
 256   ACE_OS::sprintf (buf,
 257                    ACE_TEXT("%s# Consumer Name = %s, Proxy ID = %d Load = %u\n"),
 258                    msg,
 259                    this->name_.c_str (), this->proxy_id_, this->load_);
 260
 261   stats_.dump_samples (fname.c_str (), buf, dump_samples);
 262 }