ACE/examples/IPC_SAP/SOCK_SAP/CPP-inserver-poll.cpp

   1 // IPC_SAP/poll server, which illustrates how to integrate the ACE
   2 // socket wrappers with the SVR4 <poll> system call to create a
   3 // single-threaded concurrent server.  This server program can be
   4 // driven by the oneway test mode of CPP-inclient.cpp.
   5
   6 #include "ace/OS_main.h"
   7 #include "ace/SOCK_Acceptor.h"
   8 #include "ace/SOCK_Stream.h"
   9 #include "ace/INET_Addr.h"
  10 #include "ace/Log_Msg.h"
  11 #include "ace/OS_NS_poll.h"
  12 #include "ace/OS_NS_stdio.h"
  13
  14
  15 #if defined (ACE_HAS_POLL)
  16
  17 // Should we be verbose?
  18 static int verbose = 0;
  19
  20 // Max number of open handles.
  21 static const int MAX_HANDLES = 200;
  22
  23 struct Buffer_Info
  24 {
  25   void *buf_;
  26   // Pointer to the buffer.
  27
  28   size_t len_;
  29   // Length of the buffer.
  30 };
  31
  32 // Array of <pollfd>'s.
  33 static struct pollfd poll_array[MAX_HANDLES];
  34
  35 // Array of <Buffer_Info>.
  36 static Buffer_Info buffer_array[MAX_HANDLES];
  37
  38 static void
  39 init_poll_array ()
  40 {
  41   int i;
  42
  43   for (i = 0; i < MAX_HANDLES; i++)
  44     {
  45       poll_array[i].fd = ACE_INVALID_HANDLE;
  46       poll_array[i].events = POLLIN;
  47     }
  48 }
  49
  50 static int
  51 init_buffer (size_t index)
  52 {
  53   ACE_INT32 len;
  54
  55  if (ACE::recv_n (poll_array[index].fd,
  56                   (void *) &len,
  57                   sizeof (ACE_INT32)) != sizeof (ACE_INT32))
  58     ACE_ERROR_RETURN ((LM_ERROR,
  59                        "(%P|%t) %p\n",
  60                        "recv_n failed"),
  61                       -1);
  62   else
  63     {
  64       len = ntohl (len);
  65       ACE_DEBUG ((LM_DEBUG,
  66                   "(%P|%t) reading messages of size %d from handle %d\n",
  67                   len,
  68                   poll_array[index].fd));
  69
  70       ACE_ALLOCATOR_RETURN (buffer_array[index].buf_,
  71                             ACE_OS::malloc (len),
  72                             -1);
  73       buffer_array[index].len_ = len;
  74     }
  75   return 0;
  76 }
  77
  78 static void
  79 handle_data (size_t &n_handles)
  80 {
  81   // Handle pending logging messages first (s_handle + 1 is guaranteed
  82   // to be lowest client descriptor).
  83
  84   for (size_t index = 1; index < n_handles; index++)
  85     {
  86       if (ACE_BIT_ENABLED (poll_array[index].revents, POLLIN))
  87         {
  88           // First time in, we need to initialize the buffer.
  89           if (buffer_array[index].buf_ == 0
  90               && init_buffer (index) == -1)
  91             {
  92               ACE_ERROR ((LM_ERROR,
  93                           "(%P|%t) %p\n",
  94                           "init_buffer"));
  95               continue;
  96             }
  97
  98           // Read data from client (terminate on error).
  99
 100           ssize_t n = ACE::recv (poll_array[index].fd,
 101                                  buffer_array[index].buf_,
 102                                  buffer_array[index].len_);
 103           // <recv> will not block in this case!
 104
 105           if (n == -1)
 106             ACE_ERROR ((LM_ERROR,
 107                         "%p\n",
 108                         "read failed"));
 109           else if (n == 0)
 110             {
 111               ACE_DEBUG ((LM_DEBUG,
 112                           "(%P|%t) closing oneway server at handle %d\n",
 113                           poll_array[index].fd));
 114
 115               // Handle client connection shutdown.
 116               ACE_OS::close (poll_array[index].fd);
 117               poll_array[index].fd = poll_array[--n_handles].fd;
 118
 119               ACE_OS::free ((void *) buffer_array[index].buf_);
 120               buffer_array[index].buf_ = 0;
 121               buffer_array[index].len_ = 0;
 122             }
 123           else if (verbose)
 124             ACE_DEBUG ((LM_DEBUG,
 125                         "(%P|%t) %*s",
 126                         n,
 127                         buffer_array[index].buf_));
 128         }
 129     }
 130 }
 131
 132 static void
 133 handle_connections (ACE_SOCK_Acceptor &peer_acceptor,
 134                     size_t &n_handles)
 135 {
 136   if (ACE_BIT_ENABLED (poll_array[0].revents, POLLIN))
 137     {
 138       ACE_SOCK_Stream new_stream;
 139
 140       ACE_INET_Addr client;
 141       ACE_Time_Value nonblock (0, 0);
 142
 143       // Handle all pending connection requests (note use of "polling"
 144       // feature that doesn't block).
 145
 146       while (ACE_OS::poll (poll_array, 1, nonblock) > 0)
 147         if (peer_acceptor.accept (new_stream, &client) == -1)
 148           ACE_OS::perror ("accept");
 149         else
 150           {
 151             const char *s = client.get_host_name ();
 152
 153             ACE_ASSERT (s != 0);
 154             ACE_DEBUG ((LM_DEBUG,
 155                         "(%P|%t) client %s\n",
 156                         s));
 157             poll_array[n_handles++].fd = new_stream.get_handle ();
 158           }
 159     }
 160 }
 161
 162 int
 163 ACE_TMAIN (int, ACE_TCHAR *[])
 164 {
 165   u_short port = ACE_DEFAULT_SERVER_PORT + 1;
 166
 167   // Create a server end-point.
 168   ACE_INET_Addr addr (port);
 169   ACE_SOCK_Acceptor peer_acceptor (addr);
 170
 171   ACE_HANDLE s_handle = peer_acceptor.get_handle ();
 172
 173   init_poll_array ();
 174
 175   poll_array[0].fd = s_handle;
 176
 177   ACE_DEBUG ((LM_DEBUG,
 178               "(%P|%t) starting oneway server at port %d\n",
 179               port));
 180
 181   for (size_t n_handles = 1;;)
 182     {
 183       // Wait for client I/O events (handle interrupts).
 184       while (ACE_OS::poll (poll_array, n_handles) == -1
 185              && errno == EINTR)
 186         continue;
 187
 188       handle_data (n_handles);
 189       handle_connections (peer_acceptor, n_handles);
 190     }
 191
 192   /* NOTREACHED */
 193   return 0;
 194 }
 195 #else
 196 #include <stdio.h>
 197 int ACE_TMAIN (int, ACE_TCHAR *[])
 198 {
 199   ACE_OS::fprintf (stderr, "This feature is not supported\n");
 200   return 0;
 201 }
 202 #endif /* ACE_HAS_POLL */