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

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