TAO/orbsvcs/tests/AVStreams/Latency/pong.cpp

   1 #include "pong.h"
   2 #include "orbsvcs/AV/Protocol_Factory.h"
   3 #include "tao/PortableServer/PortableServer.h"
   4 #include "tao/Strategies/advanced_resource.h"
   5 #include "tao/ORB.h"
   6 #include "tao/debug.h"
   7 #include "ace/Get_Opt.h"
   8 #include "ace/High_Res_Timer.h"
   9 #include "ace/Stats.h"
  10 #include "ace/Throughput_Stats.h"
  11
  12 const ACE_TCHAR *ior_output_file = ACE_TEXT ("pong.ior");
  13 const char *protocol = "RTP/UDP";
  14 int milliseconds = 100;
  15 size_t message_size = 64;
  16 int respond = 1;
  17 AVStreams::protocolSpec pong_protocols;
  18 AVStreams::protocolSpec ping_protocols;
  19
  20 ACE_hrtime_t recv_throughput_base = 0;
  21 ACE_Throughput_Stats recv_latency;
  22
  23 ACE_hrtime_t send_throughput_base = 0;
  24 ACE_Throughput_Stats send_latency;
  25
  26
  27 int
  28 parse_args (int argc, ACE_TCHAR *argv[])
  29 {
  30   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("xo:s:r:t:b:d"));
  31   int c;
  32
  33   while ((c = get_opts ()) != -1)
  34     switch (c)
  35       {
  36       case 'o':
  37         ior_output_file = get_opts.opt_arg ();
  38         break;
  39
  40       case 'r':
  41         {
  42           CORBA::ULong l = ping_protocols.length ();
  43           ping_protocols.length (l + 1);
  44           ping_protocols[l] = CORBA::string_dup (ACE_TEXT_ALWAYS_CHAR (get_opts.opt_arg ()));
  45         }
  46         break;
  47
  48       case 's':
  49         {
  50           CORBA::ULong l = pong_protocols.length ();
  51           pong_protocols.length (l + 1);
  52           pong_protocols[l] = CORBA::string_dup (ACE_TEXT_ALWAYS_CHAR (get_opts.opt_arg ()));
  53         }
  54         break;
  55
  56       case 't':
  57         milliseconds = ACE_OS::atoi (get_opts.opt_arg ());
  58         break;
  59
  60       case 'b':
  61         message_size = ACE_OS::atoi (get_opts.opt_arg ());
  62         if (message_size < sizeof(ACE_hrtime_t))
  63           {
  64             ACE_DEBUG ((LM_DEBUG, "Invalid message size\n"));
  65             message_size = 64;
  66           }
  67         break;
  68
  69       case 'x':
  70         respond = 0;
  71         break;
  72       case 'd':
  73         TAO_debug_level++;
  74         break;
  75
  76       case '?':
  77       default:
  78         ACE_ERROR_RETURN ((LM_ERROR,
  79                            "usage:  %s "
  80                            "-o <iorfile> "
  81                            "-r <protocol=addr> "
  82                            "-s <protocol=addr> "
  83                            "-t <milliseconds> "
  84                            "\n",
  85                            argv [0]),
  86                           -1);
  87       }
  88
  89
  90   // If no protocols are specified use the default...
  91   if (pong_protocols.length () == 0)
  92     {
  93       pong_protocols.length (1);
  94       pong_protocols[0] = CORBA::string_dup ("UDP=localhost:23456");
  95     }
  96
  97   if (ping_protocols.length () == 0)
  98     {
  99       ping_protocols.length (1);
 100       ping_protocols[0] = CORBA::string_dup ("UDP=localhost:12345");
 101     }
 102
 103   // Indicates successful parsing of the command line
 104   return 0;
 105 }
 106
 107 int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 108 {
 109   try
 110     {
 111       CORBA::ORB_var orb = CORBA::ORB_init (argc, argv);
 112
 113       parse_args (argc, argv);
 114
 115       CORBA::Object_var obj
 116         = orb->resolve_initial_references ("RootPOA");
 117
 118       PortableServer::POA_var poa
 119         = PortableServer::POA::_narrow (obj.in ());
 120
 121       PortableServer::POAManager_var mgr
 122         = poa->the_POAManager ();
 123
 124       mgr->activate ();
 125
 126       TAO_AV_CORE::instance ()->init (orb.in (),
 127                                       poa.in ());
 128
 129       Reactive_Strategy *reactive_strategy;
 130       ACE_NEW_RETURN (reactive_strategy,
 131                       Reactive_Strategy,
 132                       1);
 133       reactive_strategy->init (orb.in (), poa.in ());
 134       TAO_MMDevice *mmdevice_impl;
 135       ACE_NEW_RETURN (mmdevice_impl,
 136                       TAO_MMDevice (reactive_strategy),
 137                       1);
 138
 139       AVStreams::MMDevice_var mmdevice =
 140         mmdevice_impl->_this ();
 141
 142       CORBA::String_var ior =
 143         orb->object_to_string (mmdevice.in ());
 144
 145       ACE_DEBUG ((LM_DEBUG, "Activated as <%C>\n", ior.in () ));
 146
 147       // If the ior_output_file exists, output the ior to it
 148       if (ior_output_file != 0)
 149         {
 150           FILE *output_file= ACE_OS::fopen (ior_output_file, "w");
 151           if (output_file == 0)
 152             ACE_ERROR_RETURN ((LM_ERROR,
 153                                "Cannot open output file %s for writing IOR: %C",
 154                                ior_output_file,
 155                                ior.in ()),
 156                               1);
 157           ACE_OS::fprintf (output_file, "%s", ior.in ());
 158           ACE_OS::fclose (output_file);
 159         }
 160
 161       Pong_Recv_FDev* pong_fdev_impl;
 162       ACE_NEW_RETURN (pong_fdev_impl,
 163                       Pong_Recv_FDev ("Pong"),
 164                       1);
 165       Ping_Send_FDev* ping_fdev_impl;
 166       ACE_NEW_RETURN (ping_fdev_impl,
 167                       Ping_Send_FDev ("Ping"),
 168                       1);
 169
 170       AVStreams::FDev_var ping_fdev =
 171         ping_fdev_impl->_this ();
 172       AVStreams::FDev_var pong_fdev =
 173         pong_fdev_impl->_this ();
 174
 175       mmdevice->add_fdev (ping_fdev.in ());
 176       if (respond == 1)
 177         {
 178           mmdevice->add_fdev (pong_fdev.in ());
 179         }
 180
 181       orb->run ();
 182
 183       ACE_DEBUG ((LM_DEBUG, "event loop finished\n"));
 184
 185       ACE_DEBUG ((LM_DEBUG, "Calibrating scale factory . . . "));
 186       ACE_High_Res_Timer::global_scale_factor_type gsf =
 187         ACE_High_Res_Timer::global_scale_factor ();
 188       ACE_DEBUG ((LM_DEBUG, "done\n"));
 189
 190       recv_latency.dump_results (ACE_TEXT("Receive"), gsf);
 191
 192       send_latency.dump_results (ACE_TEXT("Send"), gsf);
 193     }
 194   catch (const CORBA::Exception& ex)
 195     {
 196       ex._tao_print_exception ("Caught exception:");
 197       return 1;
 198     }
 199
 200   return 0;
 201 }
 202
 203 // ****************************************************************
 204
 205 Pong_Recv::Pong_Recv ()
 206   :  TAO_FlowConsumer ("Pong",
 207                        pong_protocols,
 208                        "UNS:pong")
 209 {
 210 }
 211
 212 int
 213 Pong_Recv::get_callback (const char *,
 214                          TAO_AV_Callback *&callback)
 215 {
 216   // ACE_DEBUG ((LM_DEBUG,"Pong_Recv::get_callback\n"));
 217   callback = &this->callback_;
 218   return 0;
 219 }
 220
 221 int
 222 Pong_Recv_Callback::handle_stop ()
 223 {
 224   // ACE_DEBUG ((LM_DEBUG,"Pong_Recv_Callback::stop"));
 225   TAO_AV_CORE::instance ()->orb ()->shutdown ();
 226   return 0;
 227 }
 228
 229 int
 230 Pong_Recv_Callback::receive_frame (ACE_Message_Block *frame,
 231                                    TAO_AV_frame_info *,
 232                                    const ACE_Addr &)
 233 {
 234   // ACE_DEBUG ((LM_DEBUG,"Pong_Recv_Callback::receive_frame\n"));
 235
 236   ACE_hrtime_t now = ACE_OS::gethrtime ();
 237   for (const ACE_Message_Block *i = frame;
 238        i != 0;
 239        i = frame->cont ())
 240     {
 241       ACE_hrtime_t buf[2];
 242
 243       if (frame->length () < sizeof(buf))
 244         {
 245           ACE_DEBUG ((LM_DEBUG, "Unexpected message size\n"));
 246           return 0;
 247         }
 248
 249       ACE_OS::memcpy (buf, i->rd_ptr (), sizeof(buf));
 250
 251       if (recv_throughput_base == 0)
 252         {
 253           recv_throughput_base = now;
 254         }
 255       recv_latency.sample (now - recv_throughput_base,
 256                            now - buf[0]);
 257     }
 258   return 0;
 259 }
 260
 261 int
 262 Pong_Recv_Callback::handle_destroy ()
 263 {
 264   ACE_DEBUG ((LM_DEBUG,"Pong_Recv_Callback::destroy\n"));
 265   return 0;
 266 }
 267
 268 // ****************************************************************
 269
 270 Ping_Send::Ping_Send ()
 271   :  TAO_FlowProducer ("Ping",
 272                        ping_protocols,
 273                        "UNS:ping")
 274 {
 275 }
 276
 277 int
 278 Ping_Send::get_callback (const char *,
 279                          TAO_AV_Callback *&callback)
 280 {
 281   // ACE_DEBUG ((LM_DEBUG,"Ping_Send::get_callback\n"));
 282   callback = &this->callback_;
 283   return 0;
 284 }
 285
 286 Ping_Send_Callback::Ping_Send_Callback ()
 287   :count_ (0)
 288 {
 289   this->timeout_ = ACE_Time_Value (2);
 290
 291   this->frame_.size (message_size);
 292   this->frame_.wr_ptr (message_size);
 293 }
 294
 295 void
 296 Ping_Send_Callback::get_timeout (ACE_Time_Value *&tv,
 297                                  void *&)
 298 {
 299   tv = &this->timeout_;
 300 }
 301
 302 int
 303 Ping_Send_Callback::handle_timeout (void *)
 304 {
 305   this->count_++;
 306
 307   ACE_DEBUG ((LM_DEBUG, "Ping timeout frame %d\n", this->count_));
 308
 309   if (this->count_ > 10)
 310     {
 311       TAO_AV_CORE::instance ()->orb ()->shutdown ();
 312       return 0;
 313     }
 314
 315   ACE_hrtime_t stamp = ACE_OS::gethrtime ();
 316   ACE_OS::memcpy (this->frame_.rd_ptr (), &stamp, sizeof(stamp));
 317
 318   int result = this->protocol_object_->send_frame (&this->frame_);
 319   if (result < 0)
 320     ACE_ERROR_RETURN ((LM_ERROR,
 321                        "Ping_Send_Callback::handle_timeout - send_frame - %p\n",
 322                        ""),
 323                       -1);
 324
 325   if (send_throughput_base == 0)
 326     {
 327       send_throughput_base = stamp;
 328     }
 329   ACE_hrtime_t now = ACE_OS::gethrtime ();
 330   send_latency.sample (now - send_throughput_base,
 331                        now - stamp);
 332
 333   return 0;
 334 }
 335
 336 int
 337 Ping_Send_Callback::handle_end_stream ()
 338 {
 339   return 0;
 340 }