misc/423: Problem when using AIO sockets

[vkarasik]

>Number:         423
>Category:       misc
>Synopsis:       Problem when using AIO sockets
>Confidential:   no
>Severity:       critical
>Priority:       high
>Responsible:    unassigned
>State:          open
>Class:          sw-bug
>Submitter-Id:   net
>Arrival-Date:   Sun Oct 06 03:26:54 -0400 2002
>Originator:     Vitaly Karasik
>Release:        glibc  2.2.4-13,  2.2.5-37,  2.2.93-5
>Organization:
NDS
>Environment:
different HP Vectra's with RH7.2,7.3,8.0
>Description:
i've attached a program [written by Steve Huston address@hidden which is 
only C/C++ (no ACE) and
reproduces the
problem with AIO on Linux. It has compilation
instructions in it.
This program opens a socket back to itself - there's a client
and server.
The client sends some data; the server reads it and sends it
back. When the
client receives the echo, it starts the process over again,
counting how
many times it goes around.

There is one command line argument - the number of
iterations to run
through. On each successful iteration, the client prints a
"." to the
terminal. If it completes all of the iterations, it exits. If
it goes for
more than 2 seconds with no AIO activity, it will print an
error message and
sleep for 5 minutes. This gives you an opportunity to attach
to it with gdb
and examine the threads. One will be stuck in libc_read, and
if you also do
a "netstat" you'll see there is data available to read, but
it's not being
read. This is the same condition we see in ACE.

It won't run correctly in gdb - it runs the machine out of
threads and runs
very slowly. This is why you need to wait for it to lock up
and then attach
to it.
>How-To-Repeat:
1. g++ -g -o aiotest -lrt aiotest.cpp
2. run  "./aiotest 1000000" few times
        
Actual Results:  many times I get "Resorce temporarily unavailable"

Expected Results:  "AIOCB test successful"
>Fix:
Unknown
>Unformatted:
 ----gnatsweb-attachment----
 Content-Type: text/plain; name="aiotest.cpp"
 Content-Disposition: inline; filename="aiotest.cpp"
 
 // ============================================================================
 //
 // = LIBRARY
 //    proactor
 //
 // = FILENAME
 //    aiotest.cpp
 //
 // = DESCRIPTION
 //    Checkout $ACE_ROOT/examples/Reactor/Proactor/test_aiocb_ace.cpp,
 //    which is the ACE'ified version of this program. 
 // 
 // = COMPILE and RUN
 //    % g++ -g -o aiotest -lrt aiotest.cpp
 //    % ./aiotest
 // 
 // = AUTHOR
 //    Steve Huston <address@hidden>
 //
 // ============================================================================
 
 #include <stdio.h>
 #include <iostream.h>
 #include <aio.h>
 #include <errno.h>
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
 
 class Test_Aio
 {
 public:
  Test_Aio (void) : nr_recvd_ (0) {};
 
  // Set up the socket and AIOCBs
  int init (void);
 
  // Doing the testing stuff.
  int do_aio (int iterations);
 
  ~Test_Aio (void);
 
 private:
  //  Server socket.
  int server_;
 
  // Server receive and echo operations
  struct aiocb server_in_;
  struct aiocb server_out_;
  char server_in_buff_[128];
  char server_out_buff_[128];
 
  // Client socket
  int client_;
 
  // Client send and receive operations
  struct aiocb client_in_;
  struct aiocb client_out_;
  char client_in_buff_[128];
  char client_out_buff_[128];
  int nr_recvd_;
 };
 
 Test_Aio::~Test_Aio (void)
 {
  close (this->server_);
  close (this->client_);
 }
 
 // Set up the sockets.
 int
 Test_Aio::init (void)
 {
  sockaddr_in pick_a_port;
 
  // Helpers to avoid lots of casting
  sockaddr * addr = (sockaddr *)&pick_a_port;
  socklen_t addrlen = (socklen_t)(sizeof pick_a_port);
 
  int listener = socket (PF_INET, SOCK_STREAM, 0);
 
  memset (&pick_a_port, 0, sizeof pick_a_port);
  if (bind (listener, addr, addrlen) == -1)
    { perror ("bind"); return -1; }
  getsockname (listener, addr, &addrlen);
  if (listen (listener, 1) == -1)
    { perror ("listen"); return -1; }
 
  this->client_ = socket (PF_INET, SOCK_STREAM, 0);
  if (connect (this->client_, addr, addrlen) == -1)
    { perror ("connect"); return -1; }
 
  if ((this->server_ = accept (listener, 0, 0)) == -1)
    { perror ("accept"); return -1; }
 
  memset (this->server_in_buff_, 0, sizeof (this->server_in_buff_));
  memset (this->server_out_buff_, 0, sizeof (this->server_out_buff_));
  memset (this->client_in_buff_, 0, sizeof (this->client_in_buff_));
  memset (this->client_out_buff_, 0, sizeof (this->client_out_buff_));
  close (listener);
  return 0;
 }
 
 // Start the initial client send and server receive. Whatever the server
 // receives, it sends back. Continue until <iterations> have completed
 // or 2 seconds goes by without any activity.
 int
 Test_Aio::do_aio (int iterations)
 {
  struct aiocb *list_aiocb[4];
  enum { SRV_IN = 0, SRV_OUT, CLI_IN, CLI_OUT };
  list_aiocb[SRV_IN] = 0;
  list_aiocb[SRV_OUT] = 0;
  list_aiocb[CLI_IN] = 0;
  list_aiocb[CLI_OUT] = 0;
 
  this->server_in_.aio_fildes = this->server_;
  this->server_in_.aio_offset = 0;
  this->server_in_.aio_buf = this->server_in_buff_;
  this->server_in_.aio_nbytes = sizeof (this->server_in_buff_);
  this->server_in_.aio_reqprio = 0;
  this->server_in_.aio_sigevent.sigev_notify = SIGEV_NONE;
  this->server_in_.aio_sigevent.sigev_signo = 0;
  this->server_in_.aio_sigevent.sigev_value.sival_ptr = 0;
 
  this->server_out_.aio_fildes = this->server_;
  this->server_out_.aio_offset = 0;
  this->server_out_.aio_buf = this->server_out_buff_;
  this->server_out_.aio_nbytes = 0;    // Fill in when receive length known
  this->server_out_.aio_reqprio = 0;
  this->server_out_.aio_sigevent.sigev_notify = SIGEV_NONE;
  this->server_out_.aio_sigevent.sigev_signo = 0;
  this->server_out_.aio_sigevent.sigev_value.sival_ptr = 0;
 
  this->client_in_.aio_fildes = this->client_;
  this->client_in_.aio_offset = 0;
  this->client_in_.aio_buf = this->client_in_buff_;
  this->client_in_.aio_nbytes = sizeof (this->client_in_buff_);
  this->client_in_.aio_reqprio = 0;
  this->client_in_.aio_sigevent.sigev_notify = SIGEV_NONE;
  this->client_in_.aio_sigevent.sigev_signo = 0;
  this->client_in_.aio_sigevent.sigev_value.sival_ptr = 0;
 
  this->client_out_.aio_fildes = this->client_;
  this->client_out_.aio_offset = 0;
  this->client_out_.aio_buf = this->client_out_buff_;
  strcpy (this->client_out_buff_, "This is a test of Linux AIO\n");
  this->client_out_.aio_nbytes = strlen (this->client_out_buff_);
  this->client_out_.aio_reqprio = 0;
  this->client_out_.aio_sigevent.sigev_notify = SIGEV_NONE;
  this->client_out_.aio_sigevent.sigev_signo = 0;
  this->client_out_.aio_sigevent.sigev_value.sival_ptr = 0;
 
  // Initiate the client's first write and server's first read
  if (aio_write (&this->client_out_) == -1)
    { perror ("client initial write"); return -1; }
  list_aiocb[CLI_OUT] = &this->client_out_;
 
  if (aio_read (&this->server_in_) == -1)
    { perror ("server initial read"); return -1; }
  list_aiocb[SRV_IN] = &this->server_in_;
 
  // Wait for the completion on aio_suspend.
  // The structure of send/receive for client and server are different
  // to try and poke holes in AIO.
  // When:
  //    the server completes a write, note it.
  //    the server completes a read, start a write and the next read.
  //    the client completes a write, start another read.
  //    the client completes a read, start a write.
  int return_val = 0;
  while (this->nr_recvd_ < iterations)
    {
      struct timespec timeout;
      timeout.tv_sec = 2;
      timeout.tv_nsec = 0;
 
      return_val = aio_suspend (list_aiocb, 4, &timeout);
      if (return_val == -1)
        { perror ("aio_suspend"); sleep(300); break; }
 
      // Analyze return and error values. As soon as something completes,
      // start the corresponding operation.
      if (list_aiocb[SRV_OUT] &&
          aio_error (list_aiocb [SRV_OUT]) != EINPROGRESS)
        {
          return_val = aio_return (list_aiocb [SRV_OUT]);
          if (return_val <= 0)    // Socket close or error
            {
              perror ("server write");
              break;
            }
          list_aiocb[SRV_OUT] = 0;   // Done; remove from dispatch
        }
 
      if (list_aiocb[SRV_IN] &&
          aio_error (list_aiocb [SRV_IN]) != EINPROGRESS)
        {
          return_val = aio_return (list_aiocb [SRV_IN]);
          if (return_val <= 0)    // Socket close or error
            {
              perror ("server read");
              break;
            }
          // Data read; send it back and restart the read.
          strncpy (this->server_out_buff_, this->server_in_buff_, return_val);
          this->server_out_.aio_nbytes = return_val;
          if (aio_write (&this->server_out_) == -1)
            { perror ("server aio_write"); break; }
          list_aiocb[SRV_OUT] = &this->server_out_;  // Let aio_suspend see it
 
          if (aio_read (&this->server_in_) == -1)
            { perror ("server aio_read"); break; }
        }
 
      if (list_aiocb[CLI_OUT] &&
          aio_error (list_aiocb [CLI_OUT]) != EINPROGRESS)
        {
          return_val = aio_return (list_aiocb [CLI_OUT]);
          if (return_val <= 0)    // Socket close or error
            {
              perror ("client write");
              break;
            }
          list_aiocb[CLI_OUT] = 0;   // Done; remove from dispatch
 
          // Data sent; prepare to receive the echo
          if (aio_read (&this->client_in_) == -1)
            { perror ("client aio_read"); break; }
          list_aiocb[CLI_IN] = &this->client_in_;  // Let aio_suspend see it
        }
 
      if (list_aiocb[CLI_IN] &&
          aio_error (list_aiocb [CLI_IN]) != EINPROGRESS)
        {
          return_val = aio_return (list_aiocb [CLI_IN]);
          if (return_val <= 0)    // Socket close or error
            {
              perror ("client read");
              break;
            }
          cout << ".";
          ++this->nr_recvd_;
          list_aiocb[CLI_IN] = 0;   // Done; remove from dispatch
 
          // Data received; start another pass
          if (aio_write (&this->client_out_) == -1)
            { perror ("client aio_write"); break; }
          list_aiocb[CLI_OUT] = &this->client_out_;  // Let aio_suspend see it
        }
    }
 
  if (this->nr_recvd_ < iterations)
    {
      cerr << "Oops; only did " << this->nr_recvd_ << " of " << iterations
           << " iterations." << endl;
      return -1;
    }
  cout << iterations << " iterations done." << endl;
  return 0;
 }
 
 int
 main (int argc, char **argv)
 {
  Test_Aio test_aio;
 
  int iterations = 1000;
  if (argc > 1)
    iterations = atoi (argv[1]);
 
  if (test_aio.init () != 0)
    {
      printf ("AIOCB test failed:\n"
              "ACE_POSIX_AIOCB_PROACTOR may not work in this platform\n");
      return -1;
    }
  
  if (test_aio.do_aio (iterations) != 0)
    {
      printf ("AIOCB test failed:\n"
              "ACE_POSIX_AIOCB_PROACTOR may not work in this platform\n"); 
      return -1;
    }
  printf ("AIOCB test successful:\n"
          "ACE_POSIX_AIOCB_PROACTOR should work in this platform\n");
  return 0;
 }