summaryrefslogtreecommitdiff
path: root/ACE/tests/Bug_2740_Regression_Test.cpp
blob: 8c31bfda748070e33d04d34756cf8af92b1cf969 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
/**
 * @file Bug_2740_Regression_Test.cpp
 *
 * $Id$
 *
 * Reproduces the problems reported in bug 2740
 *   http://deuce.doc.wustl.edu/bugzilla/show_bug.cgi?id=2740
 *
 * @author paolo.volpi@tvblob.com
 */

#include "test_config.h"

ACE_RCSID (tests,
           Bug_2740_Regression_Test,
           "$Id$")

#if defined (ACE_HAS_DEV_POLL) || defined (ACE_HAS_EVENT_POLL)

#include "ace/Atomic_Op.h"
#include "ace/Reactor.h"
#include "ace/Dev_Poll_Reactor.h"
#include "ace/Connector.h"
#include "ace/SOCK_Connector.h"
#include "ace/Acceptor.h"
#include "ace/SOCK_Acceptor.h"
#include "ace/Svc_Handler.h"
#include "ace/Signal.h"
#include "ace/Time_Value.h"
#include "ace/OS_NS_string.h"

const ACE_Time_Value MAX_CLIENT_TIMEOUT(30);  // 30 secs

// The number of clients we'll run. When the ClientsLeft gets back to 0 (it's
// decremented when a ClientSvcHandler is deleted) the event loop can stop.
const long NrClients = 10;
static ACE_Atomic_Op<ACE_Thread_Mutex, long> ClientsLeft (10);


class ClientSvcHandler : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>
{
public:
  // Default ctor needed to satisfy ACE_Connector template, but not used.
  ClientSvcHandler ()
  { ACE_ERROR((LM_ERROR, ACE_TEXT ("%t: Wrong ClientSvcHandler ctor!!\n"))); }

  // Each client will send/recv 'echo_cnt' times, close/reopen the socket,
  // then echo, etc. for ACE_MAX_ITERATIONS times.
  ClientSvcHandler (const ACE_INET_Addr &addr, int echo_cnt);
  ~ClientSvcHandler ();

  int open (void* factory);
  int handle_input (ACE_HANDLE handle = ACE_INVALID_HANDLE);
  int handle_timeout (const ACE_Time_Value &now, const void *act = 0);
  int handle_close (ACE_HANDLE handle = ACE_INVALID_HANDLE,
                    ACE_Reactor_Mask mask = 0);

protected:
  static const char *send_str;

  ACE_INET_Addr server_addr_;
  // Know when we've received 'echo_cnt' echoes of the data. Don't want to
  // get into record-marking, short-reads/writes, etc. so just count the number
  // of bytes it takes to echo the data. Don't care about verifying the
  // content - this is mostly a stress test for multithreaded reactor.
  size_t echo_cnt_;
  size_t iter_;
  unsigned long timer_;
};

const char *ClientSvcHandler::send_str =
  "1234567890123456789012345678901234567890";

class ServerSvcHandler : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>
{
public:
  int open (void* factory);
  int handle_input (ACE_HANDLE handle = ACE_INVALID_HANDLE);
  int handle_close (ACE_HANDLE handle = ACE_INVALID_HANDLE,
                    ACE_Reactor_Mask mask = 0);
};


ClientSvcHandler::ClientSvcHandler (const ACE_INET_Addr &addr, int echo_cnt)
  : ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>(),
    server_addr_ (addr),
    echo_cnt_ (ACE_OS::strlen (ClientSvcHandler::send_str) * echo_cnt),
    iter_ (0)
{
}

ClientSvcHandler::~ClientSvcHandler ()
{
  if (--ClientsLeft == 0)
    {
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("%t: No more clients; ending reactor loop\n")));
      this->reactor ()->end_reactor_event_loop ();
    }
}

int
ClientSvcHandler::open (void* factory)
{
  ++iter_;
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: client h %d, open for iteration %B\n"),
              this->peer ().get_handle (), this->iter_));

  if (ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::open (factory) == 0)
    {
      this->timer_ = this->reactor ()->schedule_timer (this,
                                                       0,
                                                       MAX_CLIENT_TIMEOUT);

      size_t send_len = ACE_OS::strlen (ClientSvcHandler::send_str);
      if (-1 == this->peer().send_n (ClientSvcHandler::send_str, send_len))
        ACE_ERROR_RETURN ((LM_ERROR,
                           ACE_TEXT ("%t: client h %d, %p\n"),
                           this->peer ().get_handle(),
                           ACE_TEXT ("send")),
                          -1);
      return 0;
    }

  ACE_ERROR_RETURN ((LM_ERROR,
                     ACE_TEXT ("%t: %p\n"),
                     ACE_TEXT ("Client open")),
                    -1);
}

int
ClientSvcHandler::handle_input (ACE_HANDLE handle)
{
    // Get socket data
    char buffer[ACE_DEFAULT_MAX_SOCKET_BUFSIZ];
    ssize_t bc;
    bc = this->peer ().recv (buffer, sizeof (buffer));
    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("%t: client h %d recv %b bytes\n"),
                handle,
                bc));
    if (bc > 0)
      {
        // Reset Client timeout timer
        this->reactor ()->cancel_timer (this->timer_);

        // Reduce the echo count by what was received; if not there, send
        // more data and await more echo.
        size_t ubc = (size_t)bc;
        if (ubc <= this->echo_cnt_)
          this->echo_cnt_ -= ubc;
        if (this->echo_cnt_ == 0)
          {
            ACE_DEBUG ((LM_DEBUG,
                        ACE_TEXT ("%t: client h %d end of echo\n"),
                        handle));
            return -1;
          }

        // Not done yet; send more data and resched timer.
        size_t send_len = ACE_OS::strlen (ClientSvcHandler::send_str);
        if (-1 == this->peer().send_n (ClientSvcHandler::send_str, send_len))
          ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%t: client h %d: %p\n"),
                             ACE_TEXT ("resending")),
                            -1);
        this->timer_ = reactor ()->schedule_timer (this, 0, MAX_CLIENT_TIMEOUT);
      }
    else if (bc == 0) // Socket was closed by server
      {
        ACE_ERROR_RETURN ((LM_ERROR,
                           ACE_TEXT ("%t: client h %d: closed by server\n"),
                           handle),
                          -1);
      }
    else if (errno == EWOULDBLOCK) // no data ready on socket
      {
        ACE_ERROR_RETURN ((LM_DEBUG,
                           ACE_TEXT ("%t: client h %d: recv no data\n"),
                           handle),
                          0);
      }
    else
      ACE_ERROR_RETURN ((LM_ERROR,
                         ACE_TEXT ("%t: client h %d: %p\n"),
                         handle,
                         ACE_TEXT ("send")),
                        -1);

    return 0;
}

int
ClientSvcHandler::handle_timeout (const ACE_Time_Value &, const void*)
{
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("%t: client h %d: timeout\n"),
              this->peer ().get_handle ()));
  reactor()->remove_handler (this, ACE_Event_Handler::ALL_EVENTS_MASK);
  return 0;
}

int
ClientSvcHandler::handle_close (ACE_HANDLE handle, ACE_Reactor_Mask mask)
{
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("%t: client h %d handle_close\n"),
              handle));

  // If not done iterating, just close the socket and reopen the connection.
  // Else shut down and delete.
  if (this->iter_ >= ACE_MAX_ITERATIONS)
    return ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::handle_close(handle,
                                                                          mask);
  this->shutdown ();
  ClientSvcHandler *p = const_cast<ClientSvcHandler *>(this);
  ACE_Connector<ClientSvcHandler, ACE_SOCK_CONNECTOR>
    connector (this->reactor (), ACE_NONBLOCK);
  connector.connect (p, this->server_addr_);
  return 0;
}

int
ServerSvcHandler::open (void* factory)
{
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: server h %d open\n"),
              this->peer ().get_handle ()));

  return ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::open (factory);
}

int
ServerSvcHandler::handle_input (ACE_HANDLE handle)
{
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: server h %d input\n"), handle));

  // Receive whatever is here and send it back.
  char buffer[ACE_DEFAULT_MAX_SOCKET_BUFSIZ];
  ssize_t bc;
  bc = this->peer ().recv (buffer, sizeof (buffer));
  if (bc > 0)
    {
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("%t: server h %d recv %b bytes\n"),
                  handle,
                  bc));
      if (-1 == this->peer ().send_n (buffer, bc))
        ACE_ERROR_RETURN ((LM_ERROR,
                           ACE_TEXT ("%t: server h %d: %p\n"),
                           handle, ACE_TEXT ("send")),
                          -1);
    }
  else if (bc == 0 || errno == ECONNRESET) // Socket was closed by client
    {
      ACE_ERROR_RETURN ((LM_DEBUG,
                         ACE_TEXT ("%t: server h %d: closed by client\n"),
                         handle),
                        -1);
    }
  else if (errno == EWOULDBLOCK) // no data ready on socket
    {
      // The reactor shouldn't call back for input if there's no data, so
      // log an error, but keep running.
      ACE_ERROR_RETURN ((LM_ERROR,
                         ACE_TEXT ("%t: server h %d: recv no data\n"),
                         handle),
                        0);
    }
  else
    ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%t: server h %d: %p\n"),
                       handle,
                       ACE_TEXT ("recv")),
                      -1);

  return 0;
}

int
ServerSvcHandler::handle_close(ACE_HANDLE handle, ACE_Reactor_Mask mask)
{
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: server h %d: handle_close\n"), handle));
  return ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::handle_close (handle,
                                                                         mask);
}


ACE_THR_FUNC_RETURN
event_loop(void *arg)
{
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: Starting reactor event loop\n")));

  ACE_Reactor *reactor = static_cast<ACE_Reactor*>(arg);
  int s = reactor->run_reactor_event_loop();

  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: reactor loop done; status %d\n"), s));
  return 0;
}


int run_main(int, ACE_TCHAR *[])
{
  ACE_START_TEST (ACE_TEXT ("Bug_2740_Regression_Test"));

  // Make sure we ignore SIGPIPE
  sigset_t sigsetNew[1];
  sigset_t sigsetOld[1];
  ACE_OS::sigemptyset (sigsetNew);
  ACE_OS::sigaddset (sigsetNew, SIGPIPE);
  ACE_OS::sigprocmask (SIG_BLOCK, sigsetNew, sigsetOld);

  ACE_Dev_Poll_Reactor dp_reactor;
  dp_reactor.restart (1);          // Restart on EINTR
  ACE_Reactor reactor (&dp_reactor);

  // Bind acceptor to any port and then find out what the port was.
  ACE_Acceptor<ServerSvcHandler, ACE_SOCK_Acceptor> acceptor (&reactor,
                                                              ACE_NONBLOCK);
  ACE_SOCK_Acceptor::PEER_ADDR server_addr;
  if (acceptor.open (ACE_sap_any_cast (const ACE_INET_Addr &), &reactor) == -1
      || acceptor.acceptor ().get_local_addr (server_addr) == -1)
    {
      ACE_ERROR_RETURN ((LM_ERROR,
                         ACE_TEXT ("(%P|%t) %p\n"),
                         ACE_TEXT ("acceptor open")),
                        1);
    }

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("(%P|%t) starting server at port %d\n"),
              server_addr.get_port_number ()));

  ACE_Thread_Manager::instance ()->spawn_n (ACE_MAX_THREADS,
                                            event_loop,
                                            &reactor);

  ACE_Connector<ClientSvcHandler, ACE_SOCK_CONNECTOR> connector (&reactor,
                                                                 ACE_NONBLOCK);
  // Spin up clients to connect and iterate
  long i;
  for (i = 0; i < NrClients; ++i)
    {
      ClientSvcHandler *c = new ClientSvcHandler (server_addr, i % 4);
      connector.connect(c, server_addr);
    }

  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%d clients connected\n"), (int)i));
  ACE_Thread_Manager::instance ()->wait ();
  ACE_END_TEST;
  return 0;
}

#else

int
run_main (int, ACE_TCHAR *[])
{
  ACE_START_TEST (ACE_TEXT ("Bug_2740_Regression_Test"));
  ACE_ERROR ((LM_INFO,
              ACE_TEXT ("Dev Poll and Event Poll are not supported ")
              ACE_TEXT ("on this platform\n")));
  ACE_END_TEST;
  return 0;
}

#endif  /* ACE_HAS_DEV_POLL || ACE_HAS_EVENT_POLL */