TAO/tests/Oneway_Buffering/client.cpp

   1 #include "TestC.h"
   2 #include "tao/Messaging/Messaging.h"
   3 #include "tao/AnyTypeCode/Any.h"
   4 #include "tao/TAOC.h"
   5 #include "tao/AnyTypeCode/TAOA.h"
   6 #include "ace/Get_Opt.h"
   7 #include "ace/OS_NS_sys_time.h"
   8 #include "ace/OS_NS_unistd.h"
   9
  10 const ACE_TCHAR *server_ior = ACE_TEXT("file://server.ior");
  11 const ACE_TCHAR *admin_ior = ACE_TEXT("file://admin.ior");
  12 int iterations = 20;
  13
  14 int run_message_count_test = 0;
  15 int run_timeout_test = 0;
  16 int run_timeout_reactive_test = 0;
  17 int run_buffer_size_test = 0;
  18
  19 const int PAYLOAD_LENGTH = 1024;
  20 const int BUFFERED_MESSAGES_COUNT = 10;
  21 const unsigned int TIMEOUT_MILLISECONDS = 50;
  22 const int BUFFER_SIZE = 10 * PAYLOAD_LENGTH;
  23
  24 /// Check that no more than 10% of the messages are not sent.
  25 const double LIVENESS_TOLERANCE = 0.9;
  26
  27 /// Limit the depth of the liveness test, avoid blowing up the stack
  28 /// on the server
  29 const int LIVENESS_MAX_DEPTH = 256;
  30
  31 /// Factor in GIOP overhead in the buffer size test
  32 const double GIOP_OVERHEAD = 0.9;
  33
  34 const ACE_Time_Value TRANSIENT_HOLDOFF (0, 500); // 0.5ms delay
  35 const int TRANSIENT_LIMIT = 10;
  36
  37 int
  38 parse_args (int argc, ACE_TCHAR *argv[])
  39 {
  40   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("k:a:i:ctbr"));
  41   int c;
  42
  43   while ((c = get_opts ()) != -1)
  44     switch (c)
  45       {
  46       case 'k':
  47         server_ior = get_opts.opt_arg ();
  48         break;
  49
  50       case 'a':
  51         admin_ior = get_opts.opt_arg ();
  52         break;
  53
  54       case 'i':
  55         iterations = ACE_OS::atoi (get_opts.opt_arg ());
  56         break;
  57
  58       case 'c':
  59         run_message_count_test = 1;
  60         break;
  61
  62       case 't':
  63         run_timeout_test = 1;
  64         break;
  65
  66       case 'b':
  67         run_buffer_size_test = 1;
  68         break;
  69
  70       case 'r':
  71         run_timeout_reactive_test = 1;
  72         break;
  73
  74       case '?':
  75       default:
  76         ACE_ERROR_RETURN ((LM_ERROR,
  77                            "usage:  %s "
  78                            "-k <server_ior> "
  79                            "-a <admin_ior> "
  80                            "-i <iterations> "
  81                            "<-c|-t|-b|-r> "
  82                            "\n",
  83                            argv [0]),
  84                           -1);
  85       }
  86   // Indicates successful parsing of the command line
  87   return 0;
  88 }
  89
  90 int
  91 run_message_count (CORBA::ORB_ptr orb,
  92                    Test::Oneway_Buffering_ptr oneway_buffering,
  93                    Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin);
  94
  95 int
  96 run_timeout (CORBA::ORB_ptr orb,
  97              Test::Oneway_Buffering_ptr oneway_buffering,
  98              Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin);
  99
 100 int
 101 run_timeout_reactive (CORBA::ORB_ptr orb,
 102                       Test::Oneway_Buffering_ptr oneway_buffering,
 103                       Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin);
 104
 105 int
 106 run_buffer_size (CORBA::ORB_ptr orb,
 107                  Test::Oneway_Buffering_ptr oneway_buffering,
 108                  Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin);
 109
 110 int
 111 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
 112 {
 113   int test_failed = 0;
 114   try
 115     {
 116       CORBA::ORB_var orb =
 117         CORBA::ORB_init (argc, argv);
 118
 119       if (parse_args (argc, argv) != 0)
 120         return 1;
 121
 122       CORBA::Object_var tmp =
 123         orb->string_to_object(server_ior);
 124
 125       Test::Oneway_Buffering_var oneway_buffering =
 126         Test::Oneway_Buffering::_narrow(tmp.in ());
 127
 128       if (CORBA::is_nil (oneway_buffering.in ()))
 129         {
 130           ACE_ERROR_RETURN ((LM_DEBUG,
 131                              "Nil Test::Oneway_Buffering reference <%s>\n",
 132                              server_ior),
 133                             1);
 134         }
 135
 136       tmp =
 137         orb->string_to_object(admin_ior);
 138
 139       Test::Oneway_Buffering_Admin_var oneway_buffering_admin =
 140         Test::Oneway_Buffering_Admin::_narrow(tmp.in ());
 141
 142       if (CORBA::is_nil (oneway_buffering_admin.in ()))
 143         {
 144           ACE_ERROR_RETURN ((LM_DEBUG,
 145                              "Nil Test::Oneway_Buffering_Admin reference <%s>\n",
 146                              admin_ior),
 147                             1);
 148         }
 149
 150       if (run_message_count_test)
 151         {
 152           ACE_DEBUG ((LM_DEBUG,
 153                       "Running message count flushing test\n"));
 154           test_failed =
 155             run_message_count (orb.in (),
 156                                oneway_buffering.in (),
 157                                oneway_buffering_admin.in ());
 158         }
 159       else if (run_timeout_test)
 160         {
 161           ACE_DEBUG ((LM_DEBUG,
 162                       "Running timeout flushing test\n"));
 163           test_failed =
 164             run_timeout (orb.in (),
 165                          oneway_buffering.in (),
 166                          oneway_buffering_admin.in ());
 167         }
 168       else if (run_timeout_reactive_test)
 169         {
 170           ACE_DEBUG ((LM_DEBUG,
 171                       "Running timeout (reactive) flushing test\n"));
 172           test_failed =
 173             run_timeout_reactive (orb.in (),
 174                                   oneway_buffering.in (),
 175                                   oneway_buffering_admin.in ());
 176         }
 177       else if (run_buffer_size_test)
 178         {
 179           ACE_DEBUG ((LM_DEBUG,
 180                       "Running buffer size flushing test\n"));
 181           test_failed =
 182             run_buffer_size (orb.in (),
 183                              oneway_buffering.in (),
 184                              oneway_buffering_admin.in ());
 185         }
 186       else
 187         {
 188           ACE_ERROR ((LM_ERROR,
 189                       "ERROR: No test was configured\n"));
 190         }
 191
 192       oneway_buffering->shutdown ();
 193       oneway_buffering_admin->shutdown ();
 194       ACE_OS::sleep(1);
 195       orb->destroy ();
 196     }
 197   catch (const CORBA::Exception& ex)
 198     {
 199       ACE_DEBUG ((LM_DEBUG, "(%P) Client: "));
 200       ex._tao_print_exception ("CORBA Exception caught:");
 201       return 1;
 202     }
 203   catch (...)
 204     {
 205       ACE_DEBUG ((LM_DEBUG, "(%P) Client caught unknown exception\n"));
 206       return 1;
 207     }
 208
 209   return test_failed;
 210 }
 211
 212 int
 213 configure_policies (CORBA::ORB_ptr orb,
 214                     const TAO::BufferingConstraint &buffering_constraint,
 215                     Test::Oneway_Buffering_ptr oneway_buffering,
 216                     Test::Oneway_Buffering_out flusher)
 217 {
 218   CORBA::Object_var object =
 219     orb->resolve_initial_references ("PolicyCurrent");
 220
 221   CORBA::PolicyCurrent_var policy_current =
 222     CORBA::PolicyCurrent::_narrow (object.in ());
 223
 224   if (CORBA::is_nil (policy_current.in ()))
 225     {
 226       ACE_ERROR ((LM_ERROR, "ERROR: Nil policy current\n"));
 227       return 1;
 228     }
 229   CORBA::Any scope_as_any;
 230   scope_as_any <<= Messaging::SYNC_NONE;
 231
 232   CORBA::Any buffering_as_any;
 233   buffering_as_any <<= buffering_constraint;
 234
 235   CORBA::PolicyList policies (2); policies.length (2);
 236   policies[0] =
 237     orb->create_policy (Messaging::SYNC_SCOPE_POLICY_TYPE,
 238                         scope_as_any);
 239   policies[1] =
 240     orb->create_policy (TAO::BUFFERING_CONSTRAINT_POLICY_TYPE,
 241                         buffering_as_any);
 242
 243   policy_current->set_policy_overrides (policies, CORBA::ADD_OVERRIDE);
 244
 245   policies[0]->destroy ();
 246   policies[1]->destroy ();
 247
 248   TAO::BufferingConstraint flush_constraint;
 249   flush_constraint.mode = TAO::BUFFER_FLUSH;
 250   flush_constraint.message_count = 0;
 251   flush_constraint.message_bytes = 0;
 252   flush_constraint.timeout = 0;
 253
 254   buffering_as_any <<= flush_constraint;
 255   policies.length (1);
 256   policies[0] =
 257     orb->create_policy (TAO::BUFFERING_CONSTRAINT_POLICY_TYPE,
 258                         buffering_as_any);
 259
 260   object =
 261     oneway_buffering->_set_policy_overrides (policies,
 262                                              CORBA::ADD_OVERRIDE);
 263
 264   policies[0]->destroy ();
 265
 266   flusher =
 267     Test::Oneway_Buffering::_narrow (object.in ());
 268
 269   return 0;
 270 }
 271
 272 void
 273 sync_server (Test::Oneway_Buffering_ptr flusher)
 274 {
 275   // Get back in sync with the server...
 276   int transient_count= 0;
 277   while (true)
 278     {
 279       try
 280         {
 281           flusher->flush ();
 282           break;
 283         }
 284       catch (const CORBA::TRANSIENT &)
 285         {
 286           if (++transient_count < TRANSIENT_LIMIT)
 287             {
 288               if (transient_count == TRANSIENT_LIMIT / 2)
 289                 {
 290                   ACE_DEBUG ((LM_DEBUG,
 291                     "(%P) Client large TRANSIENTS encountered calling flush().\n"));
 292                 }
 293               ACE_OS::sleep (TRANSIENT_HOLDOFF);
 294             }
 295           else
 296             {
 297               throw; // Abort the message sending.
 298             }
 299         }
 300     }
 301   while (true)
 302     {
 303       try
 304         {
 305           flusher->sync ();
 306           break;
 307         }
 308       catch (const CORBA::TRANSIENT &)
 309         {
 310           if (++transient_count < TRANSIENT_LIMIT)
 311             {
 312               if (transient_count == TRANSIENT_LIMIT / 2)
 313                 {
 314                   ACE_DEBUG ((LM_DEBUG,
 315                     "(%P) Client large TRANSIENTS encountered calling sync().\n"));
 316                 }
 317               ACE_OS::sleep (TRANSIENT_HOLDOFF);
 318             }
 319           else
 320             {
 321               throw; // Abort the message sending.
 322             }
 323         }
 324     }
 325 }
 326
 327 CORBA::ULong
 328 request_count (
 329   const Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin,
 330   const CORBA::ULong expected_request_count= static_cast<CORBA::ULong>(0))
 331 {
 332   CORBA::ULong count;
 333   int transient_count= 0;
 334   while (true)
 335     {
 336       try
 337         {
 338           count= oneway_buffering_admin->request_count (expected_request_count);
 339           break;
 340         }
 341       catch (const CORBA::TRANSIENT &)
 342         {
 343           if (++transient_count < TRANSIENT_LIMIT)
 344             {
 345               if (transient_count == TRANSIENT_LIMIT / 2)
 346                 {
 347                   ACE_DEBUG ((LM_DEBUG,
 348                     "(%P) Client large TRANSIENTS encountered calling request_count().\n"));
 349                 }
 350               ACE_OS::sleep (TRANSIENT_HOLDOFF);
 351             }
 352           else
 353             {
 354               throw; // Abort the message sending.
 355             }
 356         }
 357     }
 358
 359   return count;
 360 }
 361
 362 void
 363 receive_data (
 364   const Test::Oneway_Buffering_ptr oneway_buffering,
 365   const Test::Payload &payload)
 366 {
 367   int transient_count= 0;
 368   while (true)
 369     {
 370       try
 371         {
 372           return oneway_buffering->receive_data (payload);
 373         }
 374       catch (const CORBA::TRANSIENT &)
 375         {
 376           if (++transient_count < TRANSIENT_LIMIT)
 377             {
 378               if (transient_count == TRANSIENT_LIMIT / 2)
 379                 {
 380                   ACE_DEBUG ((LM_DEBUG,
 381                     "(%P) Client large TRANSIENTS encountered calling receive_data().\n"));
 382                 }
 383               ACE_OS::sleep (TRANSIENT_HOLDOFF);
 384             }
 385           else
 386             {
 387               throw; // Abort the message sending.
 388             }
 389         }
 390     }
 391 }
 392
 393 CORBA::ULong
 394 bytes_received_count (
 395   const Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin,
 396   const CORBA::ULong expected_bytes_received_count= static_cast<CORBA::ULong>(0))
 397 {
 398   CORBA::ULong count;
 399   int transient_count= 0;
 400   while (true)
 401     {
 402       try
 403         {
 404           count= oneway_buffering_admin->bytes_received_count (expected_bytes_received_count);
 405           break;
 406         }
 407       catch (const CORBA::TRANSIENT &)
 408         {
 409           if (++transient_count < TRANSIENT_LIMIT)
 410             {
 411               if (transient_count == TRANSIENT_LIMIT / 2)
 412                 {
 413                   ACE_DEBUG ((LM_DEBUG,
 414                     "(%P) Client large TRANSIENTS encountered calling bytes_received_count().\n"));
 415                 }
 416               ACE_OS::sleep (TRANSIENT_HOLDOFF);
 417             }
 418           else
 419             {
 420               throw; // Abort the message sending.
 421             }
 422         }
 423     }
 424
 425   return count;
 426 }
 427
 428 int
 429 run_liveness_test (CORBA::ORB_ptr orb,
 430                    Test::Oneway_Buffering_ptr oneway_buffering,
 431                    Test::Oneway_Buffering_ptr flusher,
 432                    Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin)
 433 {
 434   ACE_DEBUG ((LM_DEBUG, ".... checking for liveness\n"));
 435   int test_failed = 0;
 436
 437   sync_server (flusher);
 438
 439   CORBA::ULong send_count =
 440     request_count (oneway_buffering_admin);
 441
 442   int liveness_test_iterations = static_cast<int> (send_count);
 443
 444   Test::Payload payload (PAYLOAD_LENGTH);
 445   payload.length (PAYLOAD_LENGTH);
 446   for (int j = 0; j != PAYLOAD_LENGTH; ++j)
 447     payload[j] = CORBA::Octet(j % 256);
 448
 449   int depth = 0;
 450   for (int i = 0; i != liveness_test_iterations; ++i)
 451     {
 452       receive_data (oneway_buffering, payload);
 453       ++send_count;
 454
 455       ACE_Time_Value tv (0, 1000);
 456       orb->run (tv);
 457
 458       CORBA::ULong receive_count =
 459         request_count (oneway_buffering_admin);
 460
 461       // Once the system has sent enough messages we don't
 462       // expect it to fall too far behind, i.e. at least 90% of the
 463       // messages should be delivered....
 464       CORBA::ULong expected =
 465         CORBA::ULong (LIVENESS_TOLERANCE * send_count);
 466
 467       if (receive_count < expected)
 468         {
 469           test_failed = 1;
 470           ACE_DEBUG ((LM_DEBUG,
 471                       "DEBUG: Iteration %d "
 472                       "not enough messages received %u "
 473                       "expected %u\n",
 474                       i, receive_count, expected));
 475
 476           sync_server (flusher);
 477         }
 478
 479       if (depth++ == LIVENESS_MAX_DEPTH)
 480         {
 481           sync_server (flusher);
 482
 483           depth = 0;
 484         }
 485
 486     }
 487
 488   return test_failed;
 489 }
 490
 491 int
 492 run_message_count (CORBA::ORB_ptr orb,
 493                    Test::Oneway_Buffering_ptr oneway_buffering,
 494                    Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin)
 495 {
 496   TAO::BufferingConstraint buffering_constraint;
 497   buffering_constraint.mode = TAO::BUFFER_MESSAGE_COUNT;
 498   buffering_constraint.message_count = BUFFERED_MESSAGES_COUNT;
 499   buffering_constraint.message_bytes = 0;
 500   buffering_constraint.timeout = 0;
 501
 502   Test::Oneway_Buffering_var flusher;
 503   int test_failed =
 504     configure_policies (orb, buffering_constraint,
 505                         oneway_buffering, flusher.out ());
 506
 507   if (test_failed != 0)
 508     return test_failed;
 509
 510   Test::Payload payload (PAYLOAD_LENGTH);
 511   payload.length (PAYLOAD_LENGTH);
 512   for (int j = 0; j != PAYLOAD_LENGTH; ++j)
 513     payload[j] = CORBA::Octet(j % 256);
 514
 515   CORBA::ULong send_count = 0;
 516   for (int i = 0; i != iterations; ++i)
 517     {
 518       sync_server (flusher.in ());
 519
 520       CORBA::ULong initial_receive_count =
 521         request_count (oneway_buffering_admin, send_count);
 522
 523       if (initial_receive_count != send_count)
 524         {
 525           test_failed = 1;
 526           ACE_DEBUG ((LM_DEBUG,
 527                       "DEBUG: Iteration %d message lost (%u != %u)\n",
 528                       i, initial_receive_count, send_count));
 529         }
 530
 531       while (1)
 532         {
 533           receive_data (oneway_buffering, payload);
 534           ++send_count;
 535
 536           ACE_Time_Value tv (0, 1000);
 537           orb->run (tv);
 538
 539           CORBA::ULong iteration_count =
 540             send_count - initial_receive_count;
 541
 542           const bool
 543             too_few= (iteration_count < static_cast<CORBA::ULong> (BUFFERED_MESSAGES_COUNT));
 544
 545           CORBA::ULong receive_count =
 546             request_count (
 547               oneway_buffering_admin,
 548               static_cast<CORBA::ULong> (too_few ?
 549                                          initial_receive_count
 550                                          : (initial_receive_count
 551                                             + static_cast<CORBA::ULong> (BUFFERED_MESSAGES_COUNT))));
 552
 553           if (receive_count != initial_receive_count)
 554             {
 555               if (too_few)
 556                 {
 557                   test_failed = 1;
 558                   ACE_DEBUG ((LM_DEBUG,
 559                               "DEBUG: Iteration %d flush before "
 560                               "message count reached. "
 561                               "Iteration count = %u, Threshold = %u\n",
 562                               i,
 563                               iteration_count, BUFFERED_MESSAGES_COUNT));
 564                 }
 565               break;
 566             }
 567
 568           if (iteration_count > 3 * BUFFERED_MESSAGES_COUNT)
 569             {
 570               test_failed = 1;
 571               ACE_DEBUG ((LM_DEBUG,
 572                           "DEBUG: Iteration %d no flush past "
 573                           "message count threshold. "
 574                           "Iteration count = %u, Threshold = %u\n",
 575                           i,
 576                           iteration_count, BUFFERED_MESSAGES_COUNT));
 577               break;
 578             }
 579         }
 580     }
 581
 582   int liveness_test_failed =
 583     run_liveness_test (orb,
 584                        oneway_buffering,
 585                        flusher.in (),
 586                        oneway_buffering_admin);
 587
 588   if (liveness_test_failed)
 589     test_failed = 1;
 590
 591   return test_failed;
 592 }
 593
 594 int
 595 run_timeout (CORBA::ORB_ptr orb,
 596               Test::Oneway_Buffering_ptr oneway_buffering,
 597               Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin)
 598 {
 599   TAO::BufferingConstraint buffering_constraint;
 600   buffering_constraint.mode = TAO::BUFFER_TIMEOUT;
 601   buffering_constraint.message_count = 0;
 602   buffering_constraint.message_bytes = 0;
 603   buffering_constraint.timeout = TIMEOUT_MILLISECONDS * 10000;
 604
 605   Test::Oneway_Buffering_var flusher;
 606   int test_failed =
 607     configure_policies (orb, buffering_constraint,
 608                         oneway_buffering, flusher.out ());
 609
 610   if (test_failed != 0)
 611     return test_failed;
 612
 613   Test::Payload payload (PAYLOAD_LENGTH);
 614   payload.length (PAYLOAD_LENGTH);
 615   for (int j = 0; j != PAYLOAD_LENGTH; ++j)
 616     payload[j] = CORBA::Octet(j % 256);
 617
 618   CORBA::ULong send_count = 0;
 619   for (int i = 0; i != iterations; ++i)
 620     {
 621       sync_server (flusher.in ());
 622
 623       CORBA::ULong initial_receive_count =
 624         request_count (oneway_buffering_admin, send_count);
 625
 626       if (initial_receive_count != send_count)
 627         {
 628           test_failed = 1;
 629           ACE_DEBUG ((LM_DEBUG,
 630                       "DEBUG: Iteration %d message lost (%u != %u)\n",
 631                       i, initial_receive_count, send_count));
 632         }
 633
 634       ACE_Time_Value start = ACE_OS::gettimeofday ();
 635       while (1)
 636         {
 637           receive_data (oneway_buffering, payload);
 638           ++send_count;
 639
 640           ACE_Time_Value tv (0, 1000);
 641           orb->run (tv);
 642
 643           CORBA::ULong receive_count =
 644             request_count (oneway_buffering_admin);
 645
 646           ACE_Time_Value elapsed = ACE_OS::gettimeofday () - start;
 647           if (receive_count != initial_receive_count)
 648             {
 649               if (elapsed.msec () < TIMEOUT_MILLISECONDS)
 650                 {
 651                   test_failed = 1;
 652                   ACE_ERROR ((LM_ERROR,
 653                               "ERROR: Iteration %d flush before "
 654                               "timeout expired. "
 655                               "Elapsed = %d, Timeout = %d msecs\n",
 656                               i,
 657                               elapsed.msec (), TIMEOUT_MILLISECONDS));
 658                 }
 659               // terminate the while loop.
 660               break;
 661             }
 662
 663           if (elapsed.msec () > 3 * TIMEOUT_MILLISECONDS)
 664             {
 665               test_failed = 1;
 666               ACE_ERROR ((LM_ERROR,
 667                           "ERROR: Iteration %d no flush past "
 668                           "timeout threshold. "
 669                           "Elapsed = %d, Timeout = %d msecs\n",
 670                           i,
 671                           elapsed.msec (), TIMEOUT_MILLISECONDS));
 672               break;
 673             }
 674         }
 675     }
 676
 677   int liveness_test_failed =
 678     run_liveness_test (orb, oneway_buffering,
 679                        flusher.in (),
 680                        oneway_buffering_admin);
 681
 682   if (liveness_test_failed)
 683     test_failed = 1;
 684
 685
 686   return test_failed;
 687 }
 688
 689 int
 690 run_timeout_reactive (CORBA::ORB_ptr orb,
 691                       Test::Oneway_Buffering_ptr oneway_buffering,
 692                       Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin)
 693 {
 694   TAO::BufferingConstraint buffering_constraint;
 695   buffering_constraint.mode = TAO::BUFFER_TIMEOUT;
 696   buffering_constraint.message_count = 0;
 697   buffering_constraint.message_bytes = 0;
 698   buffering_constraint.timeout = TIMEOUT_MILLISECONDS * 10000;
 699
 700   Test::Oneway_Buffering_var flusher;
 701   int test_failed =
 702     configure_policies (orb, buffering_constraint,
 703                         oneway_buffering, flusher.out ());
 704
 705   if (test_failed != 0)
 706     return test_failed;
 707
 708   Test::Payload payload (PAYLOAD_LENGTH);
 709   payload.length (PAYLOAD_LENGTH);
 710   for (int j = 0; j != PAYLOAD_LENGTH; ++j)
 711     payload[j] = CORBA::Octet(j % 256);
 712
 713   CORBA::ULong send_count = 0;
 714   for (int i = 0; i != iterations; ++i)
 715     {
 716       sync_server (flusher.in ());
 717
 718       CORBA::ULong initial_receive_count =
 719         request_count (oneway_buffering_admin, send_count);
 720
 721       if (initial_receive_count != send_count)
 722         {
 723           test_failed = 1;
 724           ACE_DEBUG ((LM_DEBUG,
 725                       "DEBUG: Iteration %d message lost (%u != %u)\n",
 726                       i, initial_receive_count, send_count));
 727         }
 728
 729       ACE_Time_Value start = ACE_OS::gettimeofday ();
 730       for (int j = 0; j != 20; ++j)
 731         {
 732           receive_data (oneway_buffering, payload);
 733           ++send_count;
 734         }
 735
 736       ACE_Time_Value tv (0, 1000);
 737       orb->run (tv);
 738
 739       while (1)
 740         {
 741           CORBA::ULong receive_count =
 742             request_count (oneway_buffering_admin);
 743
 744           //FUZZ: disable check_for_lack_ACE_OS
 745           ACE_Time_Value sleep (0, 10000);
 746           orb->run (sleep);
 747           //FUZZ: enable check_for_lack_ACE_OS
 748
 749           ACE_Time_Value elapsed = ACE_OS::gettimeofday () - start;
 750           if (receive_count != initial_receive_count)
 751             {
 752               if (elapsed.msec () < TIMEOUT_MILLISECONDS)
 753                 {
 754                   test_failed = 1;
 755                   ACE_DEBUG ((LM_DEBUG,
 756                               "DEBUG: Iteration %d flush before "
 757                               "timeout expired. "
 758                               "Elapsed = %d, Timeout = %d msecs\n",
 759                               i,
 760                               elapsed.msec (), TIMEOUT_MILLISECONDS));
 761                 }
 762               // terminate the while loop.
 763               break;
 764             }
 765
 766           if (elapsed.msec () > 3 * TIMEOUT_MILLISECONDS)
 767             {
 768               test_failed = 1;
 769               ACE_DEBUG ((LM_DEBUG,
 770                           "DEBUG: Iteration %d no flush past "
 771                           "timeout threshold. "
 772                           "Elapsed = %d, Timeout = %d msecs\n",
 773                           i,
 774                           elapsed.msec (), TIMEOUT_MILLISECONDS));
 775               break;
 776             }
 777         }
 778     }
 779
 780   int liveness_test_failed =
 781     run_liveness_test (orb, oneway_buffering,
 782                        flusher.in (),
 783                        oneway_buffering_admin);
 784
 785   if (liveness_test_failed)
 786     test_failed = 1;
 787
 788
 789   return test_failed;
 790 }
 791
 792 int
 793 run_buffer_size (CORBA::ORB_ptr orb,
 794                  Test::Oneway_Buffering_ptr oneway_buffering,
 795                  Test::Oneway_Buffering_Admin_ptr oneway_buffering_admin)
 796 {
 797   TAO::BufferingConstraint buffering_constraint;
 798   buffering_constraint.mode = TAO::BUFFER_MESSAGE_BYTES;
 799   buffering_constraint.message_count = 0;
 800   buffering_constraint.message_bytes = BUFFER_SIZE;
 801   buffering_constraint.timeout = 0;
 802
 803   Test::Oneway_Buffering_var flusher;
 804   int test_failed =
 805     configure_policies (orb, buffering_constraint,
 806                         oneway_buffering, flusher.out ());
 807
 808   if (test_failed != 0)
 809     return test_failed;
 810
 811   Test::Payload payload (PAYLOAD_LENGTH);
 812   payload.length (PAYLOAD_LENGTH);
 813   for (int j = 0; j != PAYLOAD_LENGTH; ++j)
 814     payload[j] = CORBA::Octet(j % 256);
 815
 816   CORBA::ULong bytes_sent = 0;
 817   const CORBA::ULong
 818     expected_size= static_cast<CORBA::ULong>(GIOP_OVERHEAD * BUFFER_SIZE);
 819
 820   for (int i = 0; i != iterations; ++i)
 821     {
 822       sync_server (flusher.in ());
 823
 824       CORBA::ULong initial_bytes_received =
 825         bytes_received_count (oneway_buffering_admin, bytes_sent);
 826
 827       if (initial_bytes_received != bytes_sent)
 828         {
 829           test_failed = 1;
 830           ACE_DEBUG ((LM_DEBUG,
 831                       "DEBUG: Iteration %d data lost (%u != %u)\n",
 832                       i, initial_bytes_received, bytes_sent));
 833         }
 834
 835       while (1)
 836         {
 837           receive_data (oneway_buffering, payload);
 838           bytes_sent += PAYLOAD_LENGTH;
 839
 840           ACE_Time_Value tv (0, 1000);
 841           orb->run (tv);
 842
 843           CORBA::ULong payload_delta =
 844             bytes_sent - initial_bytes_received;
 845
 846           CORBA::ULong bytes_received =
 847             bytes_received_count (oneway_buffering_admin);
 848
 849           if (bytes_received != initial_bytes_received)
 850             {
 851               // The queue has been flushed, check that enough data
 852               // has been sent.  The check cannot be precise because
 853               // the ORB counts the GIOP message overhead, in this
 854               // test we assume the overhead to be less than 10%
 855
 856               if (payload_delta < expected_size)
 857                 {
 858                   test_failed = 1;
 859                   ACE_DEBUG ((LM_DEBUG,
 860                               "DEBUG: Iteration %d flush before "
 861                               "minimum buffer size was reached. "
 862                               "Sent = %u, Minimum buffer = %u bytes\n",
 863                               i,
 864                               payload_delta, BUFFER_SIZE));
 865                 }
 866               break;
 867             }
 868
 869           if (payload_delta > 3 * BUFFER_SIZE)
 870             {
 871               test_failed = 1;
 872               ACE_DEBUG ((LM_DEBUG,
 873                           "DEBUG: Iteration %d no flush past "
 874                           "buffer size threshold. "
 875                           "Sent = %u, Minimum buffer = %u bytes\n",
 876                           i,
 877                           payload_delta, BUFFER_SIZE));
 878               break;
 879             }
 880         }
 881     }
 882
 883   int liveness_test_failed =
 884     run_liveness_test (orb, oneway_buffering,
 885                        flusher.in (),
 886                        oneway_buffering_admin);
 887
 888   if (liveness_test_failed)
 889     test_failed = 1;
 890
 891   return test_failed;
 892 }