TAO/orbsvcs/LoadBalancer/LoadMonitor.cpp

   1 #include "orbsvcs/Log_Macros.h"
   2 #include "Push_Handler.h"
   3 #include "Monitor_Signal_Handler.h"
   4
   5 #include "orbsvcs/LoadBalancing/LB_CPU_Load_Average_Monitor.h"
   6 #include "orbsvcs/LoadBalancing/LB_conf.h"
   7
   8 #include "tao/ORB_Core.h"
   9
  10 #include "ace/Reactor.h"
  11 #include "ace/Get_Opt.h"
  12 #include "ace/OS_main.h"
  13 #include "ace/OS_NS_strings.h"
  14
  15 static const ACE_TCHAR * location_id = 0;
  16 static const ACE_TCHAR * location_kind = 0;
  17 static const ACE_TCHAR * mtype = ACE_TEXT("CPU");
  18 static const ACE_TCHAR * mstyle = ACE_TEXT("PUSH");
  19 static const ACE_TCHAR * custom_monitor_ior = 0;
  20
  21 // For the sake of consistency, make default push monitoring interval
  22 // the same as the pull monitoring interval.
  23 static long push_interval = TAO_LB_PULL_HANDLER_INTERVAL;
  24
  25 void
  26 usage (const ACE_TCHAR * cmd)
  27 {
  28   ORBSVCS_DEBUG ((LM_INFO,
  29               ACE_TEXT ("Usage:\n")
  30               ACE_TEXT ("  %s\n")
  31               ACE_TEXT ("    -l <location_id>\n")
  32               ACE_TEXT ("    -k <location_kind>\n")
  33               ACE_TEXT ("    -t <CPU | Disk | Memory | Network>\n")
  34               ACE_TEXT ("    -s <PULL | PUSH>\n")
  35               ACE_TEXT ("    -i <push_interval> (in seconds,")
  36               ACE_TEXT (" and requires \"PUSH\" style monitoring)\n")
  37               ACE_TEXT ("    -m <custom_monitor_ior>")
  38               ACE_TEXT (" (overrides \"-t\", \"-l\" and \"-k\")\n")
  39               ACE_TEXT ("    -h\n")
  40               ACE_TEXT ("\n"),
  41               cmd));
  42 }
  43
  44 void
  45 parse_args (int argc, ACE_TCHAR *argv[])
  46 {
  47   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT ("l:k:t:s:i:m:h"));
  48
  49   int c = 0;
  50   const ACE_TCHAR * s = 0;
  51
  52   while ((c = get_opts ()) != -1)
  53     {
  54       switch (c)
  55         {
  56         case 'm':
  57           ::custom_monitor_ior = get_opts.opt_arg ();
  58           break;
  59
  60         case 'l':
  61           ::location_id = get_opts.opt_arg ();
  62           break;
  63
  64         case 'k':
  65           ::location_kind = get_opts.opt_arg ();
  66           break;
  67
  68         case 't':
  69           ::mtype = get_opts.opt_arg ();
  70           break;
  71
  72         case 's':
  73           ::mstyle = get_opts.opt_arg ();
  74           break;
  75
  76         case 'i':
  77           s = get_opts.opt_arg ();
  78           push_interval = ACE_OS::atoi (s);
  79           if (push_interval < 1)
  80             {
  81               ORBSVCS_ERROR ((LM_ERROR,
  82                            ACE_TEXT ("ERROR: Invalid push interval: %s\n"),
  83                            s));
  84
  85               throw CORBA::BAD_PARAM ();
  86             }
  87           break;
  88
  89         case 'h':
  90           ::usage (argv[0]);
  91           ACE_OS::exit (0);
  92           break;
  93
  94         default:
  95           ::usage (argv[0]);
  96           throw CORBA::BAD_PARAM ();
  97         }
  98     }
  99 }
 100
 101 #if defined (ACE_LINUX) && defined (ACE_HAS_THREADS)
 102 // Only the main thread can handle signals in Linux.  Run the
 103 // LoadManager in thread other than main().
 104 extern "C"
 105 void *
 106 TAO_LB_run_load_monitor (void * orb_arg)
 107 {
 108   CORBA::ORB_ptr orb = static_cast<CORBA::ORB_ptr> (orb_arg);
 109
 110   // Only the main thread should handle signals.
 111   //
 112   // @@ This is probably unnecessary since no signals should be
 113   //    delivered to this thread on Linux.
 114   ACE_Sig_Guard signal_guard;
 115
 116   try
 117     {
 118       orb->run ();
 119     }
 120   catch (const CORBA::Exception& ex)
 121     {
 122       ex._tao_print_exception ("TAO Load Monitor");
 123
 124       return reinterpret_cast<void *> (-1);
 125     }
 126
 127   return 0;
 128 }
 129 #endif  /* linux && ACE_HAS_THREADS */
 130
 131
 132 CosLoadBalancing::LoadMonitor_ptr
 133 get_load_monitor (CORBA::ORB_ptr orb,
 134                   PortableServer::POA_ptr root_poa)
 135 {
 136   if (::custom_monitor_ior != 0)
 137     {
 138       CORBA::Object_var obj =
 139         orb->string_to_object (::custom_monitor_ior);
 140
 141       return CosLoadBalancing::LoadMonitor::_narrow (obj.in ());
 142     }
 143   else
 144     {
 145       // The POA is only needed for the built-in load monitors since
 146       // they must be activated.
 147       PortableServer::POAManager_var poa_manager =
 148         root_poa->the_POAManager ();
 149
 150       poa_manager->activate ();
 151
 152       if (ACE_OS::strcasecmp (::mtype, ACE_TEXT("CPU")) == 0)
 153         {
 154           TAO_LB_CPU_Load_Average_Monitor * monitor = 0;
 155           ACE_NEW_THROW_EX (monitor,
 156                             TAO_LB_CPU_Load_Average_Monitor (::location_id,
 157                                                              ::location_kind),
 158                             CORBA::NO_MEMORY ());
 159
 160           // Transfer ownership to the POA.
 161           PortableServer::ServantBase_var s = monitor;
 162
 163           return monitor->_this ();
 164         }
 165       else if (ACE_OS::strcasecmp (::mtype, ACE_TEXT("Disk")) == 0
 166                || ACE_OS::strcasecmp (::mtype, ACE_TEXT("Memory")) == 0
 167                || ACE_OS::strcasecmp (::mtype, ACE_TEXT("Network")) == 0)
 168         {
 169           ORBSVCS_ERROR ((LM_ERROR,
 170                       ACE_TEXT ("ERROR: \"%s\" load monitor currently ")
 171                       ACE_TEXT ("unimplemented.\n"),
 172                       ::mtype));
 173
 174           throw CORBA::NO_IMPLEMENT ();
 175         }
 176       else
 177         {
 178           ORBSVCS_ERROR ((LM_ERROR,
 179                       ACE_TEXT ("ERROR: Unrecognized built-in load monitor ")
 180                       ACE_TEXT ("type: <%s>.\n"),
 181                       ::mtype));
 182
 183           throw CORBA::BAD_PARAM ();
 184         }
 185     }
 186 }
 187
 188 void
 189 register_load_monitor (CosLoadBalancing::LoadManager_ptr manager,
 190                        CosLoadBalancing::LoadMonitor_ptr monitor,
 191                        TAO_LB_Push_Handler * handler,
 192                        ACE_Reactor * reactor,
 193                        long & timer_id)
 194 {
 195   if (ACE_OS::strcasecmp (::mstyle, ACE_TEXT("PULL")) == 0)
 196     {
 197       PortableGroup::Location_var location =
 198         monitor->the_location ();
 199
 200       manager->register_load_monitor (location.in (),
 201                                       monitor);
 202     }
 203   else if (ACE_OS::strcasecmp (::mstyle, ACE_TEXT("PUSH")) == 0)
 204     {
 205       ACE_Time_Value interval (::push_interval, 0);
 206       ACE_Time_Value restart (::push_interval, 0);
 207       timer_id = reactor->schedule_timer (handler,
 208                                           0,
 209                                           interval,
 210                                           restart);
 211
 212       if (timer_id == -1)
 213         {
 214           ORBSVCS_ERROR ((LM_ERROR,
 215                       ACE_TEXT ("ERROR: Unable to schedule timer for ")
 216                       ACE_TEXT ("\"PUSH\" style load monitoring.\n")));
 217
 218           throw CORBA::INTERNAL ();
 219         }
 220     }
 221   else
 222     {
 223       ORBSVCS_ERROR ((LM_ERROR,
 224                   ACE_TEXT ("ERROR: Unrecognized load monitoring ")
 225                   ACE_TEXT ("style: <%s>.\n"),
 226                   ::mstyle));
 227
 228       throw CORBA::BAD_PARAM ();
 229     }
 230 }
 231
 232 int
 233 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 234 {
 235   try
 236     {
 237       // The usual server side boilerplate code.
 238
 239       CORBA::ORB_var orb = CORBA::ORB_init (argc, argv);
 240
 241       // Check the non-ORB arguments.
 242       ::parse_args (argc,
 243                     argv);
 244
 245       CORBA::Object_var obj =
 246         orb->resolve_initial_references ("RootPOA");
 247
 248       PortableServer::POA_var root_poa =
 249         PortableServer::POA::_narrow (obj.in ());
 250
 251       CosLoadBalancing::LoadMonitor_var load_monitor =
 252         ::get_load_monitor (orb.in (),
 253                             root_poa.in ());
 254
 255       PortableGroup::Location_var location =
 256         load_monitor->the_location ();
 257
 258       // The "LoadManager" reference should have already been
 259       // registered with the ORB by its ORBInitializer.
 260       obj = orb->resolve_initial_references ("LoadManager");
 261
 262       CosLoadBalancing::LoadManager_var load_manager =
 263         CosLoadBalancing::LoadManager::_narrow (obj.in ());
 264
 265       // This "push" handler will only be used if the load monitor
 266       // style is "PUSH".
 267       TAO_LB_Push_Handler push_handler (load_monitor.in (),
 268                                         location.in (),
 269                                         load_manager.in ());
 270
 271       ACE_Reactor * reactor = orb->orb_core ()->reactor ();
 272
 273       long timer_id = -1;
 274
 275       ::register_load_monitor (load_manager.in (),
 276                                load_monitor.in (),
 277                                &push_handler,
 278                                reactor,
 279                                timer_id);
 280
 281 #if defined (ACE_LINUX) && defined (ACE_HAS_THREADS)
 282       if (ACE_Thread_Manager::instance ()->spawn (::TAO_LB_run_load_monitor,
 283                                                   orb.in ()) == -1)
 284         {
 285           ORBSVCS_ERROR_RETURN ((LM_ERROR,
 286                              "ERROR:  Unable to spawn TAO LoadMonitor's "
 287                              "ORB thread.\n"),
 288                             -1);
 289         }
 290
 291       ACE_Sig_Set sigset;
 292       sigset.sig_add (SIGINT);
 293       sigset.sig_add (SIGTERM);
 294
 295       int signum = -1;
 296
 297       // Block waiting for the registered signals.
 298       if (ACE_OS::sigwait (sigset, &signum) == -1)
 299         {
 300           ORBSVCS_ERROR_RETURN ((LM_ERROR,
 301                              "(%P|%t) %p\n",
 302                              "ERROR waiting on signal"),
 303                             -1);
 304         }
 305
 306       ACE_ASSERT (signum == SIGINT || signum == SIGTERM);
 307
 308       // Deregister the LoadMonitor from the LoadManager in the PULL
 309       // load monitoring case.
 310       if (timer_id == -1)
 311         {
 312           load_manager->remove_load_monitor (location.in ());
 313         }
 314 #else
 315       // Activate/register the signal handler that (attempts) to
 316       // ensure graceful shutdown of the LoadMonitor so that
 317       // LoadMonitors registered with the LoadManager can be
 318       // deregistered.
 319       CosLoadBalancing::LoadManager_ptr tmp;
 320
 321       if (timer_id == -1)
 322         tmp = load_manager.in ();   // PULL monitoring
 323       else
 324         tmp = CosLoadBalancing::LoadManager::_nil ();  // PUSH
 325                                                        // monitoring
 326       TAO_LB_Monitor_Signal_Handler signal_handler (
 327          orb.in (),
 328          root_poa.in (),
 329          tmp,
 330          location.in ());
 331
 332       if (signal_handler.activate () != 0)
 333         return -1;
 334
 335       // @@ There is a subtle race condition here.  If the signal
 336       //    handler thread shuts down the ORB before it is run, the
 337       //    below call to ORB::run() will throw a CORBA::BAD_INV_ORDER
 338       //    exception.
 339       orb->run ();
 340
 341       // Wait for the signal handler thread to finish
 342       // before the process exits.
 343       signal_handler.wait ();
 344 #endif  /* linux && ACE_HAS_THREADS */
 345
 346       if (timer_id != -1 && reactor->cancel_timer (timer_id) == 0)
 347         {
 348           ORBSVCS_ERROR ((LM_ERROR,
 349                       ACE_TEXT ("ERROR: Unable to cancel \"push\" load ")
 350                       ACE_TEXT ("monitoring timer.\n")));
 351
 352           // Just keep going.  We're shutting down anyway.
 353         }
 354
 355       orb->destroy ();
 356     }
 357   catch (const CORBA::Exception& ex)
 358     {
 359       ex._tao_print_exception ("TAO Load Monitor");
 360
 361       return -1;
 362     }
 363
 364   return 0;
 365 }