ACE/tests/Cached_Conn_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Cached_Conn_Test.cpp
   5  *
   6  *  The test illustrates how the <ACE_Strategy_Connector> works by
   7  *  showing how you can cache connections on the client using
   8  *  different caching strategies. Also how connections can be purged
   9  *  explicitly and implicitly if needed from the connection cache
  10  *  maintained by the connector.
  11  *
  12  *  @author Kirthika Parameswaran <kirthika@cs.wustl.edu>
  13  */
  14 //=============================================================================
  15
  16
  17 #ifndef CACHED_CONNECT_TEST
  18 #define CACHED_CONNECT_TEST
  19
  20 #include "test_config.h"
  21
  22 #include "Cached_Conn_Test.h"
  23
  24 #include "ace/OS_NS_string.h"
  25 #include "ace/INET_Addr.h"
  26 #include "ace/SOCK_Connector.h"
  27 #include "ace/SOCK_Acceptor.h"
  28 #include "ace/Svc_Handler.h"
  29 #include "ace/SOCK_Stream.h"
  30 #include "ace/Acceptor.h"
  31 #include "ace/Connector.h"
  32 #include "ace/Get_Opt.h"
  33 #include "ace/Caching_Utility_T.h"
  34 #include "ace/Cached_Connect_Strategy_T.h"
  35 #include "ace/Handle_Gobbler.h"
  36
  37 #if defined(_MSC_VER)
  38 #pragma warning(disable:4503)
  39 #endif /* _MSC_VER */
  40
  41
  42 using ATTRIBUTES = size_t;
  43 using CACHED_HANDLER = std::pair<Svc_Handler *, ATTRIBUTES>;
  44 using ACE_ADDR = ARHR<ACE_INET_Addr>;
  45 using H_KEY = ACE_Hash<ACE_ADDR>;
  46 using C_KEYS = ACE_Equal_To<ACE_ADDR>;
  47
  48 using HASH_MAP = ACE_Hash_Map_Manager_Ex<ACE_ADDR, CACHED_HANDLER, H_KEY, C_KEYS, ACE_Null_Mutex>;
  49 using HASH_MAP_ITERATOR = ACE_Hash_Map_Iterator_Ex<ACE_ADDR, CACHED_HANDLER, H_KEY, C_KEYS, ACE_Null_Mutex>;
  50 using HASH_MAP_REVERSE_ITERATOR = ACE_Hash_Map_Reverse_Iterator_Ex<ACE_ADDR, CACHED_HANDLER, H_KEY, C_KEYS, ACE_Null_Mutex>;
  51
  52 using CLEANUP_STRATEGY = ARHCLE<ACE_ADDR, CACHED_HANDLER, HASH_MAP>;
  53 using CACHING_UTILITY = ACE_Recyclable_Handler_Caching_Utility<ACE_ADDR, CACHED_HANDLER, HASH_MAP, HASH_MAP_ITERATOR, ATTRIBUTES>;
  54
  55 using LRU_CACHING_STRATEGY = ALRU<ATTRIBUTES, CACHING_UTILITY>;
  56
  57 using LFU_CACHING_STRATEGY = ALFU<ATTRIBUTES, CACHING_UTILITY>;
  58 using FIFO_CACHING_STRATEGY = AFIFO<ATTRIBUTES, CACHING_UTILITY>;
  59 using NULL_CACHING_STRATEGY = ANULL<ATTRIBUTES, CACHING_UTILITY>;
  60 using LRU_CACHING_STRATEGY_ADAPTER = ACSA<ATTRIBUTES, CACHING_UTILITY, LRU_CACHING_STRATEGY>;
  61 using LFU_CACHING_STRATEGY_ADAPTER = ACSA<ATTRIBUTES, CACHING_UTILITY, LFU_CACHING_STRATEGY>;
  62 using FIFO_CACHING_STRATEGY_ADAPTER = ACSA<ATTRIBUTES, CACHING_UTILITY, FIFO_CACHING_STRATEGY>;
  63 using NULL_CACHING_STRATEGY_ADAPTER = ACSA<ATTRIBUTES, CACHING_UTILITY, NULL_CACHING_STRATEGY>;
  64 using CACHING_STRATEGY = ACS<ATTRIBUTES, CACHING_UTILITY>;
  65
  66 using ACCEPTOR = ACE_Oneshot_Acceptor<Svc_Handler, ACE_SOCK_Acceptor>;
  67
  68 using STRATEGY_CONNECTOR = ACE_Strategy_Connector<Svc_Handler, ACE_SOCK_Connector>;
  69
  70 using NULL_CREATION_STRATEGY = ACE_NOOP_Creation_Strategy<Svc_Handler>;
  71
  72 using NULL_ACTIVATION_STRATEGY = ACE_NOOP_Concurrency_Strategy<Svc_Handler>;
  73
  74 using CACHED_CONNECT_STRATEGY = ACCSE<Svc_Handler, ACE_SOCK_Connector, CACHING_STRATEGY, ATTRIBUTES, ACE_MT_SYNCH::NULL_MUTEX>;
  75
  76 #endif /* CACHED_CONNECT_TEST */
  77
  78 static int debug = 0;
  79
  80 Svc_Handler::Svc_Handler (ACE_Thread_Manager *t)
  81   : ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH> (t)
  82 {
  83 }
  84
  85 int
  86 Svc_Handler::open (void *)
  87 {
  88   if (debug)
  89     ACE_DEBUG ((LM_DEBUG,
  90                 ACE_TEXT ("opening Svc_Handler %@ with handle %d\n"),
  91                 this,
  92                 this->peer ().get_handle ()));
  93
  94   return 0;
  95 }
  96
  97 int
  98 Svc_Handler::close (u_long flags)
  99 {
 100   ACE_DEBUG ((LM_DEBUG,
 101               ACE_TEXT ("Closing Svc_Handler %@ with handle %d\n"),
 102               this,
 103               this->peer ().get_handle ()));
 104   return ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::close (flags);
 105 }
 106
 107 enum Caching_Strategy_Type
 108 {
 109   ACE_LFU,
 110   ACE_FIFO,
 111   ACE_LRU,
 112   ACE_NULL,
 113   ACE_ALL
 114 };
 115
 116 // Default number of clients/servers.
 117 static int listen_once = 1;
 118 static int user_has_specified_iterations = 0;
 119 static size_t keep_handles_available = 100;
 120 static double purge_percentage = 20;
 121 static Caching_Strategy_Type caching_strategy_type = ACE_ALL;
 122 static CACHED_CONNECT_STRATEGY *connect_strategy = 0;
 123
 124 // On Win32, the handle gobbling doesn't work.  Therefore, we need
 125 // more iterations to get to the handle limit.
 126 #if defined (ACE_WIN32)
 127 static int iterations = 2000;
 128 #else
 129 static int iterations = 200;
 130 #endif /* ACE_WIN32 */
 131
 132 //====================================================================
 133
 134 static void
 135 out_of_sockets_handler ()
 136 {
 137   if (ACE::out_of_handles (errno))
 138     {
 139       // Close connections which are cached by explicitly purging the
 140       // connection cache maintained by the connector.
 141       ACE_DEBUG
 142         ((LM_DEBUG,
 143           ACE_TEXT ("Purging connections from Connection Cache...\n")));
 144
 145       if (-1 == connect_strategy->purge_connections ())
 146         ACE_ERROR ((LM_ERROR,
 147                     ACE_TEXT ("%p\n"),
 148                     ACE_TEXT ("purge_connections")));
 149     }
 150   else
 151     {
 152       ACE_ERROR ((LM_ERROR,
 153                   ACE_TEXT ("%p\n"),
 154                   ACE_TEXT ("in out_of_sockets_handler, ")
 155                   ACE_TEXT ("but out_of_handles said no")));
 156     }
 157 }
 158
 159 static int
 160 cached_connect (STRATEGY_CONNECTOR &con,
 161                 const ACE_INET_Addr &server_addr)
 162 {
 163   // This will make sure we get the host information correct.
 164   ACE_INET_Addr remote_addr (server_addr.get_port_number (),
 165                              ACE_LOCALHOST);
 166
 167   // Perform a blocking connect to the server using the Strategy
 168   // Connector with a connection caching strategy.
 169   Svc_Handler *svc_handler = 0;
 170   int result = con.connect (svc_handler,
 171                             remote_addr);
 172   if (result == -1)
 173     ACE_ERROR_RETURN ((LM_ERROR,
 174                        ACE_TEXT ("%p\n"),
 175                        ACE_TEXT ("connection failed")),
 176                       -1);
 177
 178   // Reset Svc_Handler state.
 179   svc_handler->recycle_state (ACE_RECYCLABLE_PURGABLE_BUT_NOT_IDLE);
 180
 181   return 0;
 182 }
 183
 184 static int
 185 server (ACCEPTOR *acceptor)
 186 {
 187   ACE_INET_Addr cli_addr;
 188
 189   // Create a new <Svc_Handler> to consume the data.
 190   Svc_Handler svc_handler;
 191
 192   int result = acceptor->accept (&svc_handler,
 193                                  &cli_addr);
 194   if (result == -1)
 195     return -1;
 196
 197   if (debug)
 198     ACE_DEBUG ((LM_DEBUG,
 199                 ACE_TEXT ("client %s connected from %d\n"),
 200                 cli_addr.get_host_name (),
 201                 cli_addr.get_port_number ()));
 202
 203   //
 204   // Svc_Handler dies here, closing the server side socket.
 205   //
 206   return 0;
 207 }
 208
 209 static void
 210 test_connection_management (CACHING_STRATEGY &caching_strategy)
 211 {
 212   // Configure the Strategy Connector with a strategy that caches
 213   // connection.
 214   CACHED_CONNECT_STRATEGY caching_connect_strategy (caching_strategy);
 215
 216   // This is required by the <out_of_sockets_handler>.
 217   connect_strategy = &caching_connect_strategy;
 218
 219   NULL_CREATION_STRATEGY creation_strategy;
 220   NULL_ACTIVATION_STRATEGY activation_strategy;
 221
 222   STRATEGY_CONNECTOR strategy_connector (0,
 223                                          &creation_strategy,
 224                                          &caching_connect_strategy,
 225                                          &activation_strategy);
 226
 227   // If <listen_once> is true, only one Acceptor is used for the test.
 228   ACCEPTOR listen_one_time_acceptor;
 229   ACE_INET_Addr server_addr;
 230
 231   if (0 != listen_one_time_acceptor.open (ACE_sap_any_cast (const ACE_INET_Addr &)))
 232     {
 233       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("acceptor open")));
 234       return;
 235     }
 236
 237   if (0 != listen_one_time_acceptor.acceptor ().get_local_addr (server_addr))
 238     {
 239       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("acceptor getaddr")));
 240       listen_one_time_acceptor.close ();
 241       return;
 242     }
 243
 244   for (int i = 1; i <= iterations; ++i)
 245     {
 246       ACE_DEBUG ((LM_DEBUG,
 247                   ACE_TEXT ("%T iteration %d\n"),
 248                   i));
 249
 250       // If <listen_once> is false, one Acceptor is used for every
 251       // iteration.
 252       ACCEPTOR listen_multiple_times_acceptor;
 253
 254       ACCEPTOR &acceptor = listen_once ?
 255         listen_one_time_acceptor :
 256         listen_multiple_times_acceptor;
 257
 258       if (!listen_once)
 259         {
 260           // Bind acceptor to any port and then find out what the port
 261           // was.
 262           if (acceptor.open (ACE_sap_any_cast (const ACE_INET_Addr &)) == -1)
 263             {
 264               out_of_sockets_handler ();
 265               continue;
 266             }
 267
 268           if (acceptor.acceptor ().get_local_addr (server_addr) == -1)
 269             {
 270               ACE_ERROR ((LM_ERROR,
 271                           ACE_TEXT ("%p\n"),
 272                           ACE_TEXT ("get_local_addr")));
 273               break;
 274             }
 275
 276           if (debug)
 277             ACE_DEBUG ((LM_DEBUG,
 278                         ACE_TEXT ("starting server at port %d\n"),
 279                         server_addr.get_port_number ()));
 280         }
 281
 282       // Run the cached blocking test.
 283       if (-1 == cached_connect (strategy_connector, server_addr))
 284         ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("cached_connect")));
 285
 286       if (-1 == server (&acceptor))
 287         out_of_sockets_handler ();
 288     }
 289 }
 290
 291 void
 292 test_caching_strategy_type ()
 293 {
 294   CACHING_STRATEGY *caching_strategy = 0;
 295
 296   switch (caching_strategy_type)
 297     {
 298     case ACE_NULL:
 299       ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("\nNull_Caching_Strategy\n\n")));
 300       ACE_NEW (caching_strategy,
 301                NULL_CACHING_STRATEGY_ADAPTER);
 302       break;
 303
 304     case ACE_LRU:
 305       ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("\nLRU_Caching_Strategy\n\n")));
 306       ACE_NEW (caching_strategy,
 307                LRU_CACHING_STRATEGY_ADAPTER);
 308       break;
 309
 310     case ACE_LFU:
 311       ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("\nLFU_Caching_Strategy\n\n")));
 312       ACE_NEW (caching_strategy,
 313                LFU_CACHING_STRATEGY_ADAPTER);
 314       break;
 315
 316     case ACE_FIFO:
 317       ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("\nFIFO_Caching_Strategy\n\n")));
 318       ACE_NEW (caching_strategy,
 319                FIFO_CACHING_STRATEGY_ADAPTER);
 320       break;
 321
 322     case ACE_ALL: // Just to remove warnings!
 323       break;
 324     }
 325
 326   caching_strategy->purge_percent (purge_percentage);
 327   test_connection_management (*caching_strategy);
 328   delete caching_strategy;
 329 }
 330
 331 int
 332 parse_args (int argc, ACE_TCHAR *argv[])
 333 {
 334   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT ("l:i:p:c:a:d"));
 335
 336   int cc;
 337
 338   while ((cc = get_opt ()) != -1)
 339     switch (cc)
 340       {
 341       case 'd':
 342         debug = 1;
 343         break;
 344       case 'l':
 345         listen_once = ACE_OS::atoi (get_opt.opt_arg ());
 346         break;
 347       case 'i':
 348         iterations = ACE_OS::atoi (get_opt.opt_arg ());
 349         user_has_specified_iterations = 1;
 350         break;
 351       case 'p':
 352         purge_percentage = ACE_OS::atoi (get_opt.opt_arg ());
 353         break;
 354       case 'c':
 355         // Note that if null caching strategy is used then this test
 356         // will fail if the number of servers exceed number of open
 357         // files allowed for the process.
 358         if (ACE_OS::strcmp (get_opt.opt_arg (), ACE_TEXT ("null")) == 0)
 359           caching_strategy_type = ACE_NULL;
 360         if (ACE_OS::strcmp (get_opt.opt_arg (), ACE_TEXT ("lru")) == 0)
 361           caching_strategy_type = ACE_LRU;
 362         if (ACE_OS::strcmp (get_opt.opt_arg (), ACE_TEXT ("lfu")) == 0)
 363           caching_strategy_type = ACE_LFU;
 364         if (ACE_OS::strcmp (get_opt.opt_arg (), ACE_TEXT ("fifo")) == 0)
 365           caching_strategy_type = ACE_FIFO;
 366         break;
 367       case 'a':
 368         keep_handles_available = ACE_OS::atoi (get_opt.opt_arg ());
 369         break;
 370       case '?':
 371       case 'h':
 372       default:
 373         ACE_ERROR ((LM_ERROR,
 374                     ACE_TEXT ("usage: %s ")
 375                     ACE_TEXT ("[-c (caching strategy: lru / lfu / fifo / null [default = all])] ")
 376                     ACE_TEXT ("[-i (iterations)] ")
 377                     ACE_TEXT ("[-l (listen once)] ")
 378                     ACE_TEXT ("[-d (addition debugging output)] ")
 379                     ACE_TEXT ("[-p (purge percent)] ")
 380                     ACE_TEXT ("[-a (keep handles available)] "),
 381                     ACE_TEXT ("Cached_Conn_Test")));
 382         return -1;
 383       }
 384
 385   return 0;
 386 }
 387
 388 int
 389 run_main (int argc, ACE_TCHAR *argv[])
 390 {
 391   // Validate options.
 392   int result = parse_args (argc, argv);
 393   if (result != 0)
 394     return result;
 395
 396 #if defined (ACE_WIN32)
 397   // Somehow, on Win32, the <listen once> option allows us to create
 398   // more handles.
 399   if (!user_has_specified_iterations && listen_once)
 400     iterations *= 2;
 401 #endif /* ACE_WIN32 */
 402
 403   // Start the test only if options are valid.
 404   ACE_START_TEST (ACE_TEXT ("Cached_Conn_Test"));
 405
 406   // Remove the extra debugging attributes from Log_Msg output.
 407   ACE_LOG_MSG->clr_flags (ACE_Log_Msg::VERBOSE_LITE);
 408
 409   // The reactor's constructor changes the handle limit for the
 410   // process.
 411   ACE_Reactor::instance ();
 412
 413   // Consume all handles in the process, leaving us
 414   // <keep_handles_available> to play with.
 415   ACE_Handle_Gobbler handle_gobbler;
 416   if (0 != handle_gobbler.consume_handles (keep_handles_available))
 417     ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("handle_gobbler")));
 418
 419   // Do we need to test all the strategies.  Note, that the less
 420   // useful null strategy is ignored in this case.
 421   if (caching_strategy_type == ACE_ALL)
 422     {
 423       caching_strategy_type = ACE_LRU;
 424       test_caching_strategy_type ();
 425
 426       // Default iterations are too many; if the user hasn't specified
 427       // otherwise, we'll shrink the iterations for LFU and FIFO.
 428       if (!user_has_specified_iterations)
 429         iterations /= 100;
 430
 431       caching_strategy_type = ACE_LFU;
 432       test_caching_strategy_type ();
 433
 434       caching_strategy_type = ACE_FIFO;
 435       test_caching_strategy_type ();
 436     }
 437   else
 438     {
 439       test_caching_strategy_type ();
 440     }
 441
 442   ACE_LOG_MSG->set_flags (ACE_Log_Msg::VERBOSE_LITE);
 443
 444   ACE_END_TEST;
 445   return 0;
 446 }