TAO/orbsvcs/performance-tests/RTEvent/TCP_Baseline/server.cpp

   1 #include "RT_Class.h"
   2
   3 #include "ace/Task.h"
   4 #include "ace/SOCK_Acceptor.h"
   5 #include "ace/SOCK_Stream.h"
   6 #include "ace/Acceptor.h"
   7 #include "ace/Svc_Handler.h"
   8 #include "ace/Reactor.h"
   9 #include "ace/TP_Reactor.h"
  10 #include "ace/Get_Opt.h"
  11
  12 int use_rt   = 0;
  13 int nthreads = 1;
  14 ACE_TCHAR const * hi_endpoint = ACE_TEXT ("localhost:12345");
  15 ACE_TCHAR const * lo_endpoint = ACE_TEXT ("localhost:23456");
  16
  17 int
  18 parse_args (int argc, ACE_TCHAR *argv[]);
  19
  20 class Task : public ACE_Task_Base
  21 {
  22 public:
  23   Task (ACE_TCHAR const *endpoint);
  24
  25   virtual int svc();
  26
  27 private:
  28   ACE_TCHAR const *endpoint_;
  29
  30   ACE_Reactor reactor_;
  31 };
  32
  33 int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
  34 {
  35   /// Move the test to the real-time class if it is possible.
  36   RT_Class rt_class;
  37
  38   if (parse_args (argc, argv) != 0)
  39     return 1;
  40
  41   Task hi_task (hi_endpoint);
  42   Task lo_task (lo_endpoint);
  43
  44   hi_task.activate(rt_class.thr_sched_class() | THR_NEW_LWP | THR_JOINABLE,
  45                    1, 1, rt_class.priority_high());
  46
  47   lo_task.activate(rt_class.thr_sched_class() | THR_NEW_LWP | THR_JOINABLE,
  48                    nthreads, 1, rt_class.priority_low());
  49   lo_task.wait();
  50
  51   hi_task.wait();
  52
  53   return 0;
  54 }
  55
  56 // ****************************************************************
  57
  58 class Svc_Handler : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_SYNCH>
  59 {
  60 public:
  61   Svc_Handler(ACE_Reactor * reactor = 0);
  62
  63   virtual int handle_input(ACE_HANDLE);
  64   virtual int handle_close (ACE_HANDLE, ACE_Reactor_Mask);
  65 };
  66
  67 typedef ACE_Acceptor<Svc_Handler,ACE_SOCK_ACCEPTOR> Acceptor;
  68
  69 Svc_Handler::Svc_Handler (ACE_Reactor * reactor)
  70   : ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_SYNCH> (0, 0, reactor)
  71 {
  72   ACE_DEBUG ((LM_DEBUG, "(%P|%t) Accepted connection\n"));
  73 }
  74
  75 int
  76 Svc_Handler::handle_input(ACE_HANDLE h)
  77 {
  78   const size_t bufsize = BUFSIZ;
  79   char buf[bufsize];
  80
  81   while(1)
  82     {
  83       ssize_t n = this->peer().recv(buf, bufsize);
  84       if (n == 0) {
  85         ACE_ERROR((LM_ERROR,
  86                    "Connection %d closed while reading data\n",
  87                    h, ""));
  88         break;
  89       } else if (n == -1) {
  90         if(errno == EWOULDBLOCK)
  91           return 0;
  92
  93         ACE_ERROR((LM_ERROR,
  94                    "Error on <%d> while reading %p\n",
  95                    h, ""));
  96         break;
  97       }
  98
  99       ssize_t k = this->peer().send(buf, n);
 100       if(k == 0) {
 101         ACE_ERROR((LM_ERROR,
 102                    "Connection <%d> closed while writing\n",
 103                    h, ""));
 104         break;
 105       } else if (k == -1) {
 106         if(errno == EWOULDBLOCK)
 107           return 0;
 108
 109         ACE_ERROR((LM_ERROR,
 110                    "Error on <%d> while writing %p\n",
 111                    h, ""));
 112         break;
 113       } else if (k != n) {
 114         ACE_ERROR((LM_ERROR,
 115                    "Short write on <%d>\n",
 116                    h, ""));
 117         break;
 118       }
 119     }
 120
 121   return -1;
 122 }
 123
 124 int
 125 Svc_Handler::handle_close(ACE_HANDLE h, ACE_Reactor_Mask m)
 126 {
 127   ACE_DEBUG ((LM_DEBUG, "(%P|%t) Closed connection\n"));
 128   this->peer().close();
 129   int r =
 130     this->ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_SYNCH>::handle_close (h,m);
 131   return r;
 132 }
 133
 134 // ****************************************************************
 135
 136 Task::Task(ACE_TCHAR const * endpoint)
 137   : endpoint_ (endpoint)
 138   , reactor_ (new ACE_TP_Reactor)
 139 {
 140   ACE_INET_Addr local_sap (endpoint_);
 141   Acceptor * acceptor = new Acceptor;
 142
 143   if(acceptor->open(local_sap, &reactor_, ACE_NONBLOCK) == -1)
 144     {
 145       ACE_ERROR((LM_ERROR, "Cannot open endpoint <%s>\n", endpoint_));
 146     }
 147 }
 148
 149 int
 150 Task::svc()
 151 {
 152   ACE_DEBUG ((LM_DEBUG, "(%P|%t) Starting thread for %s\n", endpoint_));
 153   (void) reactor_.run_reactor_event_loop();
 154   return 0;
 155 }
 156
 157 // ****************************************************************
 158
 159 int
 160 parse_args (int argc, ACE_TCHAR *argv[])
 161 {
 162   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("h:l:n:r"));
 163   int c;
 164
 165   while ((c = get_opts ()) != -1)
 166     switch (c)
 167       {
 168       case 'h':
 169         hi_endpoint = get_opts.opt_arg ();
 170         break;
 171
 172       case 'l':
 173         lo_endpoint = get_opts.opt_arg ();
 174         break;
 175
 176       case 'n':
 177         nthreads = ACE_OS::atoi (get_opts.opt_arg ());
 178         break;
 179
 180       case 'r':
 181         use_rt = 1;
 182         break;
 183
 184       case '?':
 185       default:
 186         ACE_ERROR_RETURN ((LM_ERROR,
 187                            "usage:  %s "
 188                            "-o <iorfile> "
 189                            "-r (use RT-CORBA) "
 190                            "-n nthreads "
 191                            "\n",
 192                            argv [0]),
 193                           -1);
 194       }
 195   // Indicates successful parsing of the command line
 196   return 0;
 197 }