ACE/examples/Reactor/Proactor/test_cancel.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    test_cancel.cpp
   5  *
   6  *  This program tests cancelling an Asynchronous Operation in the
   7  *  Proactor framework.
   8  *
   9  *  This tests accepts a connection and issues an Asynchronous Read
  10  *  Stream. It reads <read_size> (option -s) number of bytes and
  11  *  when this operation completes, it issues another Asynchronous
  12  *  Read Stream to <read_size> and immediately calls <cancel> to
  13  *  cancel the operation and so the program exits closing the
  14  *  connection.
  15  *
  16  *  Works fine on NT. LynxOS has a good <aio_cancel> implementation. It works
  17  *  fine.
  18  *
  19  * = RUN
  20  * ./test_cancel -p <port_number>
  21  * Then telnet to this port and send <read_size> bytes and your
  22  * connection should get closed down.
  23  *
  24  *  @author Irfan Pyarali (irfan@cs.wustl.edu)
  25  */
  26 //=============================================================================
  27
  28
  29 #include "ace/OS_main.h"
  30 #include "ace/Service_Config.h"
  31 #include "ace/Proactor.h"
  32 #include "ace/Asynch_IO.h"
  33 #include "ace/Asynch_IO_Impl.h"
  34 #include "ace/Asynch_Acceptor.h"
  35 #include "ace/INET_Addr.h"
  36 #include "ace/SOCK_Connector.h"
  37 #include "ace/SOCK_Acceptor.h"
  38 #include "ace/SOCK_Stream.h"
  39 #include "ace/Message_Block.h"
  40 #include "ace/Get_Opt.h"
  41 #include "ace/Log_Msg.h"
  42 #include "ace/OS_NS_sys_socket.h"
  43
  44
  45 #if defined (ACE_HAS_WIN32_OVERLAPPED_IO) || defined (ACE_HAS_AIO_CALLS)
  46   // This only works on Win32 platforms and on Unix platforms supporting
  47   // POSIX aio calls.
  48
  49 #include "test_cancel.h"
  50
  51 static u_short port = ACE_DEFAULT_SERVER_PORT;
  52 static int done = 0;
  53 static int read_size = 2;
  54
  55
  56 Receiver::Receiver ()
  57   : mb_ (read_size + 1),
  58     handle_ (ACE_INVALID_HANDLE)
  59 {
  60 }
  61
  62 Receiver::~Receiver ()
  63 {
  64   ACE_DEBUG ((LM_DEBUG,
  65               "Receiver: Closing down Remote connection:%d\n",
  66               this->handle_));
  67
  68   ACE_OS::closesocket (this->handle_);
  69 }
  70
  71 void
  72 Receiver::open (ACE_HANDLE handle,
  73                 ACE_Message_Block &)
  74 {
  75   // New connection, initiate stuff
  76
  77   ACE_DEBUG ((LM_DEBUG, "%N:%l:Receiver::open called\n"));
  78
  79   // Cache the new connection
  80   this->handle_ = handle;
  81
  82   // Initiate ACE_Asynch_Read_Stream
  83   if (this->rs_.open (*this, this->handle_) == -1)
  84     {
  85       ACE_ERROR ((LM_ERROR, "%p\n", "ACE_Asynch_Read_Stream::open"));
  86       return;
  87     }
  88
  89   // Try to read <n> bytes from the stream.
  90
  91   ACE_DEBUG ((LM_DEBUG,
  92               "Receiver::open: Issuing Asynch Read of (%d) bytes from the stream\n",
  93               read_size));
  94
  95   if (this->rs_.read (this->mb_,
  96                       read_size) == -1)
  97     ACE_ERROR ((LM_ERROR,
  98                 "%p\n",
  99                 "Receiver::open: Failed to issue the read"));
 100 }
 101
 102 void
 103 Receiver::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
 104 {
 105   ACE_DEBUG ((LM_DEBUG, "handle_read_stream called\n"));
 106
 107   // Reset pointers
 108   result.message_block ().rd_ptr ()[result.bytes_transferred ()] = '\0';
 109
 110   ACE_DEBUG ((LM_DEBUG, "********************\n"));
 111   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_to_read", result.bytes_to_read ()));
 112   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "handle", result.handle ()));
 113   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_transfered", result.bytes_transferred ()));
 114   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "act", (uintptr_t) result.act ()));
 115   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "success", result.success ()));
 116   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "completion_key", (uintptr_t) result.completion_key ()));
 117   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "error", result.error ()));
 118   ACE_DEBUG ((LM_DEBUG, "********************\n"));
 119   ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "message_block", result.message_block ().rd_ptr ()));
 120
 121   if (result.success () && !result.error ())
 122     {
 123       // Successful read: No error.
 124
 125       // Set the pointers back in the message block.
 126       result.message_block ().wr_ptr (result.message_block ().rd_ptr ());
 127
 128       // Issue another read, but immediately cancel it.
 129
 130       // Issue the read.
 131
 132       ACE_DEBUG ((LM_DEBUG,
 133                   "Issuing Asynch Read of (%d) bytes from the stream\n",
 134                   read_size));
 135
 136       if (this->rs_.read (this->mb_,
 137                           read_size) == -1)
 138         ACE_ERROR ((LM_ERROR,
 139                     "%p\n",
 140                     "Receiver::handle_read_stream: Failed to issue the read"));
 141
 142       // Cancel the read.
 143
 144       ACE_DEBUG ((LM_DEBUG,
 145                   "Cacelling Asynch Read "));
 146
 147       int ret_val = this->rs_.cancel ();
 148       if (ret_val == -1)
 149         ACE_ERROR ((LM_ERROR,
 150                     "%p\n",
 151                     "Receiver::handle_read_stream: Failed to cancel the read"));
 152
 153       ACE_DEBUG ((LM_DEBUG, "Asynch IO : Cancel : Result = %d\n",
 154                   ret_val));
 155     }
 156   else
 157     {
 158       done = 1;
 159
 160       ACE_DEBUG ((LM_DEBUG, "Receiver completed\n"));
 161
 162       // Print the error message if any.
 163       if (result.error () != 0)
 164         {
 165           errno = result.error ();
 166
 167           ACE_ERROR ((LM_ERROR,
 168                       "%p\n",
 169                       "Asynch Read Stream Error: "));
 170         }
 171     }
 172 }
 173
 174 static int
 175 parse_args (int argc, ACE_TCHAR *argv[])
 176 {
 177   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT("p:s:"));
 178   int c;
 179
 180   while ((c = get_opt ()) != EOF)
 181     switch (c)
 182       {
 183       case 'p':
 184         port = ACE_OS::atoi (get_opt.opt_arg ());
 185         break;
 186       case 's':
 187         read_size = ACE_OS::atoi (get_opt.opt_arg ());
 188         break;
 189       default:
 190         ACE_ERROR ((LM_ERROR, "%p.\n",
 191                     "usage :\n"
 192                     "-p <port>\n"
 193                     "-s <read_size>\n"));
 194         return -1;
 195       }
 196
 197   return 0;
 198 }
 199
 200 int
 201 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 202 {
 203   if (parse_args (argc, argv) == -1)
 204     return -1;
 205
 206   // Note: acceptor parameterized by the Receiver
 207   ACE_Asynch_Acceptor<Receiver> acceptor;
 208
 209   // Listening passively.
 210   if (acceptor.open (ACE_INET_Addr (port),
 211                      read_size,
 212                      1) == -1)
 213     ACE_ERROR_RETURN ((LM_ERROR,
 214                        "ACE:acceptor::open failed\n"),
 215                       1);
 216
 217   int success = 1;
 218
 219   while (success > 0  && !done)
 220     // dispatch events
 221     success = ACE_Proactor::instance ()->handle_events ();
 222
 223   return 0;
 224 }
 225
 226 #else /* ACE_HAS_WIN32_OVERLAPPED_IO || ACE_HAS_AIO_CALLS */
 227
 228 int
 229 ACE_TMAIN (int, ACE_TCHAR *[])
 230 {
 231   ACE_DEBUG ((LM_DEBUG,
 232               "This example does not work on this platform.\n"));
 233   return 1;
 234 }
 235
 236 #endif /* ACE_HAS_WIN32_OVERLAPPED_IO || ACE_HAS_AIO_CALLS */