ACE/tests/Future_Set_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Future_Set_Test.cpp
   5  *
   6  *  This example tests the ACE Future Set and illustrates an
   7  *  implementation of the Active Object pattern, which is available
   8  *  at <http://www.dre.vanderbilt.edu/~schmidt/PDF/Act-Obj.pdf>.  The
   9  *  Active Object itself is very simple -- it determines if numbers
  10  *  are prime.
  11  *
  12  *  @author Andres Kruse <Andres.Kruse@cern.ch>
  13  *  @author Douglas C. Schmidt <d.schmidt@vanderbilt.edu>
  14  *  @author Per Andersson <pera@ipso.se> and Johnny Tucker <jtucker@infoglide.com>
  15  */
  16 //=============================================================================
  17
  18 #include "test_config.h"
  19 #include "ace/OS_NS_string.h"
  20 #include "ace/ACE.h"
  21 #include "ace/Task.h"
  22 #include "ace/Message_Queue.h"
  23 #include "ace/Future.h"
  24 #include "ace/Future_Set.h"
  25 #include "ace/Method_Request.h"
  26 #include "ace/Activation_Queue.h"
  27 #include <memory>
  28 #include "ace/Atomic_Op.h"
  29 #include "ace/Null_Mutex.h"
  30
  31 #if defined (ACE_HAS_THREADS)
  32
  33 using ATOMIC_INT = ACE_Atomic_Op<ACE_Thread_Mutex, int>;
  34
  35 // A counter for the tasks..
  36 static ATOMIC_INT task_count (0);
  37
  38 /**
  39  * @class Prime_Scheduler
  40  *
  41  * @brief Prime number scheduler for the Active Object.
  42  *
  43  * This class also plays the role of the Proxy and the Servant
  44  * in the Active Object pattern.  Naturally, these roles could
  45  * be split apart from the Prime_Scheduler.
  46  */
  47 class Prime_Scheduler : public ACE_Task_Base
  48 {
  49   friend class Method_Request_work;
  50   friend class Method_Request_name;
  51   friend class Method_Request_end;
  52 public:
  53   /// Constructor.
  54   Prime_Scheduler (const ACE_TCHAR *,
  55                    Prime_Scheduler * = 0);
  56
  57   //FUZZ: disable check_for_lack_ACE_OS
  58   /// Initializer.
  59   int open (void *args = 0) override;
  60
  61   /// Terminator.
  62   //FUZZ: enable check_for_lack_ACE_OS
  63   virtual int shutdown ();
  64
  65   /// Destructor.
  66   ~Prime_Scheduler () override;
  67
  68   // = These methods are part of the Active Object Proxy interface.
  69   ACE_Future<u_long> work (u_long param, int count = 1);
  70   ACE_Future<const ACE_TCHAR*> name ();
  71   void end ();
  72
  73 protected:
  74   /// Runs the Prime_Scheduler's event loop, which dequeues
  75   /// <Method_Requests> and dispatches them.
  76   int svc () override;
  77
  78   // = These are the Servant methods that do the actual work.
  79   u_long work_i (u_long, int);
  80   const ACE_TCHAR *name_i ();
  81
  82 private:
  83   // = These are the <Prime_Scheduler> implementation details.
  84   ACE_TCHAR *name_;
  85   ACE_Activation_Queue activation_queue_;
  86   Prime_Scheduler *scheduler_;
  87 };
  88
  89 /**
  90  * @class Method_Request_work
  91  *
  92  * @brief Reification of the <work> method.
  93  */
  94 class Method_Request_work : public ACE_Method_Request
  95 {
  96 public:
  97   Method_Request_work (Prime_Scheduler *,
  98                        u_long,
  99                        int,
 100                        ACE_Future<u_long> &);
 101   ~Method_Request_work () override;
 102
 103   /// This is the entry point into the Active Object method.
 104   int call () override;
 105
 106 private:
 107   Prime_Scheduler *scheduler_;
 108
 109   /// Parameter to the method that's used to determine if a number if
 110   /// prime.
 111   u_long param_;
 112
 113   /// Unused.
 114   int count_;
 115
 116   /// Store the result of the Future.
 117   ACE_Future<u_long> future_result_;
 118 };
 119
 120 Method_Request_work::Method_Request_work (Prime_Scheduler *new_Prime_Scheduler,
 121                                           u_long new_param,
 122                                           int new_count,
 123                                           ACE_Future<u_long> &new_result)
 124   : scheduler_ (new_Prime_Scheduler),
 125     param_ (new_param),
 126     count_ (new_count),
 127     future_result_ (new_result)
 128 {
 129   ACE_DEBUG ((LM_DEBUG,
 130               ACE_TEXT ("(%t) Method_Request_work created\n")));
 131 }
 132
 133 Method_Request_work::~Method_Request_work ()
 134 {
 135   ACE_DEBUG ((LM_DEBUG,
 136               ACE_TEXT ("(%t) Method_Request_work will be deleted.\n")));
 137 }
 138
 139 int
 140 Method_Request_work::call ()
 141 {
 142   // Dispatch the Servant's operation and store the result into the
 143   // Future.
 144   return this->future_result_.set (this->scheduler_->work_i
 145                                    (this->param_,
 146                                     this->count_));
 147 }
 148
 149 /**
 150  * @class Method_Request_name
 151  *
 152  * @brief Reification of the <name> method.
 153  */
 154 class Method_Request_name : public ACE_Method_Request
 155 {
 156 public:
 157   Method_Request_name (Prime_Scheduler *,
 158                        ACE_Future<const ACE_TCHAR*> &);
 159   ~Method_Request_name () override;
 160
 161   /// This is the entry point into the Active Object method.
 162   int call () override;
 163
 164 private:
 165   Prime_Scheduler *scheduler_;
 166   ACE_Future<const ACE_TCHAR*> future_result_;
 167 };
 168
 169 Method_Request_name::Method_Request_name (Prime_Scheduler *new_scheduler,
 170                                           ACE_Future<const ACE_TCHAR*> &new_result)
 171   : scheduler_ (new_scheduler),
 172     future_result_ (new_result)
 173 {
 174   ACE_DEBUG ((LM_DEBUG,
 175               ACE_TEXT ("(%t) Method_Request_name created\n")));
 176 }
 177
 178 Method_Request_name::~Method_Request_name ()
 179 {
 180   ACE_DEBUG ((LM_DEBUG,
 181               ACE_TEXT ("(%t) Method_Request_name will be deleted.\n")));
 182 }
 183
 184 int
 185 Method_Request_name::call ()
 186 {
 187   // Dispatch the Servant's operation and store the result into the
 188   // Future.
 189   return future_result_.set (scheduler_->name_i ());
 190 }
 191
 192 /**
 193  * @class Method_Request_end
 194  *
 195  * @brief Reification of the <end> method.
 196  */
 197 class Method_Request_end : public ACE_Method_Request
 198 {
 199 public:
 200   Method_Request_end (Prime_Scheduler *new_Prime_Scheduler);
 201   ~Method_Request_end () override;
 202   int call () override;
 203
 204 private:
 205   Prime_Scheduler *scheduler_;
 206 };
 207
 208 Method_Request_end::Method_Request_end (Prime_Scheduler *scheduler)
 209   : scheduler_ (scheduler)
 210 {
 211 }
 212
 213 Method_Request_end::~Method_Request_end ()
 214 {
 215 }
 216
 217 int
 218 Method_Request_end::call ()
 219 {
 220   // Shut down the scheduler.
 221   this->scheduler_->shutdown ();
 222   return -1;
 223 }
 224
 225 // Constructor
 226 Prime_Scheduler::Prime_Scheduler (const ACE_TCHAR *newname,
 227                                   Prime_Scheduler *new_scheduler)
 228   : scheduler_ (new_scheduler)
 229 {
 230   ACE_NEW (this->name_,
 231            ACE_TCHAR[ACE_OS::strlen (newname) + 1]);
 232   ACE_OS::strcpy ((ACE_TCHAR *) this->name_,
 233                   newname);
 234   ACE_DEBUG ((LM_DEBUG,
 235               ACE_TEXT ("(%t) Prime_Scheduler %s created\n"),
 236               this->name_));
 237 }
 238
 239 // Destructor
 240
 241 Prime_Scheduler::~Prime_Scheduler ()
 242 {
 243   ACE_DEBUG ((LM_DEBUG,
 244               ACE_TEXT ("(%t) Prime_Scheduler %s will be destroyed\n"),
 245               this->name_));
 246   delete [] this->name_;
 247 }
 248
 249 // open
 250
 251 int
 252 Prime_Scheduler::open (void *)
 253 {
 254   task_count++;
 255   ACE_DEBUG ((LM_DEBUG,
 256               ACE_TEXT ("(%t) Prime_Scheduler %s open\n"),
 257               this->name_));
 258   // Become an Active Object.
 259   return this->activate (THR_BOUND | THR_DETACHED);
 260 }
 261
 262 // close
 263
 264 int
 265 Prime_Scheduler::shutdown ()
 266 {
 267   ACE_DEBUG ((LM_DEBUG,
 268               ACE_TEXT ("(%t) Prime_Scheduler %s shutdown\n"),
 269               this->name_));
 270   task_count--;
 271   return 0;
 272 }
 273
 274 // Service..
 275
 276 int
 277 Prime_Scheduler::svc ()
 278 {
 279   for (;;)
 280     {
 281       // Dequeue the next method request (we use an unique pointer in
 282       // case an exception is thrown in the <call>).
 283       std::unique_ptr<ACE_Method_Request> mo (this->activation_queue_.dequeue ());
 284
 285       ACE_DEBUG ((LM_DEBUG,
 286                   ACE_TEXT ("(%t) calling method request\n")));
 287       // Call it.
 288       if (mo->call () == -1)
 289         break;
 290       // Destructor automatically deletes it.
 291     }
 292
 293   /* NOTREACHED */
 294   return 0;
 295 }
 296
 297 void
 298 Prime_Scheduler::end ()
 299 {
 300   this->activation_queue_.enqueue (new Method_Request_end (this));
 301 }
 302
 303 // Here's where the Work takes place.  We compute if the parameter is
 304 // a prime number.
 305
 306 u_long
 307 Prime_Scheduler::work_i (u_long param,
 308                          int count)
 309 {
 310   ACE_UNUSED_ARG (count);
 311
 312   return ACE::is_prime (param, 2, param / 2);
 313 }
 314
 315 const ACE_TCHAR *
 316 Prime_Scheduler::name_i ()
 317 {
 318   return this->name_;
 319 }
 320
 321 ACE_Future<const ACE_TCHAR *>
 322 Prime_Scheduler::name ()
 323 {
 324   if (this->scheduler_)
 325     // Delegate to the Prime_Scheduler.
 326     return this->scheduler_->name ();
 327   else
 328     {
 329       ACE_Future<const ACE_TCHAR*> new_future;
 330
 331       // @@ What happens if new fails here?
 332       this->activation_queue_.enqueue
 333         (new Method_Request_name (this,
 334                                   new_future));
 335       return new_future;
 336     }
 337 }
 338
 339 ACE_Future<u_long>
 340 Prime_Scheduler::work (u_long newparam,
 341                        int newcount)
 342 {
 343   if (this->scheduler_) {
 344     return this->scheduler_->work (newparam, newcount);
 345   }
 346   else {
 347     ACE_Future<u_long> new_future;
 348
 349     this->activation_queue_.enqueue
 350       (new Method_Request_work (this,
 351                                 newparam,
 352                                 newcount,
 353                                 new_future));
 354     return new_future;
 355   }
 356 }
 357
 358 // @@ These values should be set by the command line options!
 359
 360 // Total number of loops.
 361 static int n_loops = 100;
 362
 363 using u_long_key = ACE_Future_Rep<u_long> *;
 364 using u_long_value = ACE_Future_Holder<u_long> *;
 365
 366 using char_star_key = ACE_Future_Rep<const ACE_TCHAR *> *;
 367 using char_star_value = ACE_Future_Holder<const ACE_TCHAR *> *;
 368
 369 #endif /* ACE_HAS_THREADS */
 370
 371 int
 372 run_main (int, ACE_TCHAR *[])
 373 {
 374   ACE_START_TEST (ACE_TEXT ("Future_Set_Test"));
 375
 376 #if defined (ACE_HAS_THREADS)
 377   // @@ Should make these be <auto_ptr>s...
 378   Prime_Scheduler *andres, *peter, *helmut, *matias;
 379
 380   // Create active objects..
 381   ACE_NEW_RETURN (andres,
 382                   Prime_Scheduler (ACE_TEXT ("andres")),
 383                   -1);
 384   int result = andres->open ();
 385   ACE_TEST_ASSERT (result != -1);
 386   ACE_NEW_RETURN (peter,
 387                   Prime_Scheduler (ACE_TEXT ("peter")),
 388                   -1);
 389   result = peter->open ();
 390   ACE_TEST_ASSERT (result != -1);
 391   ACE_NEW_RETURN (helmut,
 392                   Prime_Scheduler (ACE_TEXT ("helmut")),
 393                   -1);
 394   result = helmut->open ();
 395   ACE_TEST_ASSERT (result != -1);
 396
 397   // Matias passes all asynchronous method calls on to Andres...
 398   ACE_NEW_RETURN (matias,
 399                   Prime_Scheduler (ACE_TEXT ("matias"),
 400                                    andres),
 401                   -1);
 402   result = matias->open ();
 403   ACE_TEST_ASSERT (result != -1);
 404
 405   ACE_Future<u_long> fresulta;
 406   ACE_Future<u_long> fresultb;
 407   ACE_Future<u_long> fresultc;
 408   ACE_Future<u_long> fresultd;
 409   ACE_Future<const ACE_TCHAR *> fname;
 410
 411   ACE_Future_Set<u_long> fseta;
 412   ACE_Future_Set<u_long> fsetb;
 413   ACE_Future_Set<u_long> fsetc;
 414   ACE_Future_Set<u_long> fsetd;
 415   ACE_Future_Set<const ACE_TCHAR *> fsetname;
 416
 417   ACE_DEBUG ((LM_DEBUG,
 418               ACE_TEXT ("(%t) initializing future sets with non-blocking call\n")));
 419
 420   for (int i = 0; i < n_loops; i++)
 421     {
 422       // Spawn off the methods, which run in a separate thread as
 423       // active object invocations.
 424       fresulta = andres->work (9013);
 425       fresultb = peter->work (9013);
 426       fresultc = helmut->work (9013);
 427       fresultd = matias->work (9013);
 428       fname = andres->name ();
 429
 430       fsetname.insert (fname);
 431       fname = peter->name ();
 432       fsetname.insert (fname);
 433       fname = helmut->name ();
 434
 435       fseta.insert (fresulta);
 436       fsetb.insert (fresultb);
 437       fsetc.insert (fresultc);
 438       fsetd.insert (fresultd);
 439       fsetname.insert (fname);
 440     }
 441
 442
 443   // See if the result is available...
 444
 445   if (!fseta.is_empty ())
 446     ACE_DEBUG ((LM_DEBUG,
 447                 ACE_TEXT ("(%t) wow.. set a is not empty.....\n")));
 448
 449   if (!fsetb.is_empty ())
 450     ACE_DEBUG ((LM_DEBUG,
 451                 ACE_TEXT ("(%t) wow.. set b is not empty.....\n")));
 452
 453   if (!fsetc.is_empty ())
 454     ACE_DEBUG ((LM_DEBUG,
 455                 ACE_TEXT ("(%t) wow.. set c is not empty.....\n")));
 456
 457   if (!fsetd.is_empty ())
 458     ACE_DEBUG ((LM_DEBUG,
 459                 ACE_TEXT ("(%t) wow.. set d is not empty.....\n")));
 460
 461   if (!fsetname.is_empty ())
 462     ACE_DEBUG ((LM_DEBUG,
 463                 ACE_TEXT ("(%t) wow.. set name is not empty.....\n")));
 464
 465   ACE_DEBUG ((LM_DEBUG,
 466               ACE_TEXT ("(%t) non-blocking calls done... now blocking...\n")));
 467
 468   // Save the result of fresulta.
 469
 470   u_long resulta = 0;
 471   u_long resultb = 0;
 472   u_long resultc = 0;
 473   u_long resultd = 0;
 474
 475   u_int count = 0;
 476   while (fseta.next_readable (fresulta))
 477     {
 478       fresulta.get (resulta);
 479
 480       ACE_DEBUG ((LM_DEBUG,
 481                   ACE_TEXT ("(%t) result(%u) a %u\n"),
 482                   count,
 483                   (u_int) resulta));
 484     }
 485
 486   count = 0;
 487   while (fsetb.next_readable (fresultb))
 488     {
 489       fresultb.get (resultb);
 490
 491       ACE_DEBUG ((LM_DEBUG,
 492                   ACE_TEXT ("(%t) result(%u) b %u\n"),
 493                   count,
 494                   (u_int) resultb));
 495     }
 496
 497   count = 0;
 498   while (fsetc.next_readable (fresultc))
 499     {
 500       fresultc.get (resultc);
 501
 502       ACE_DEBUG ((LM_DEBUG,
 503                   ACE_TEXT ("(%t) result(%u) c %u\n"),
 504                   count,
 505                   (u_int) resultc));
 506     }
 507
 508   count = 0;
 509   while (fsetd.next_readable (fresultd))
 510     {
 511       fresultd.get (resultd);
 512
 513       ACE_DEBUG ((LM_DEBUG,
 514                   ACE_TEXT ("(%t) result(%u) d %u\n"),
 515                   count,
 516                   (u_int) resultd));
 517     }
 518
 519   const ACE_TCHAR *name = 0;
 520   count = 0;
 521   while (fsetname.next_readable (fname))
 522     {
 523       fname.get (name);
 524
 525       ACE_DEBUG ((LM_DEBUG,
 526                   ACE_TEXT ("(%t) result(%u) name %s\n"),
 527                   count,
 528                   name));
 529     }
 530
 531   if (fseta.is_empty ())
 532     ACE_DEBUG ((LM_DEBUG,
 533                 ACE_TEXT ("(%t) wow.. set a is empty.....\n")));
 534
 535   if (fsetb.is_empty ())
 536     ACE_DEBUG ((LM_DEBUG,
 537                 ACE_TEXT ("(%t) wow.. set b is empty.....\n")));
 538
 539   if (fsetc.is_empty ())
 540     ACE_DEBUG ((LM_DEBUG,
 541                 ACE_TEXT ("(%t) wow.. set c is empty.....\n")));
 542
 543   if (fsetd.is_empty ())
 544     ACE_DEBUG ((LM_DEBUG,
 545                 ACE_TEXT ("(%t) wow.. set d is empty.....\n")));
 546
 547   if (fsetname.is_empty ())
 548     ACE_DEBUG ((LM_DEBUG,
 549                 ACE_TEXT ("(%t) wow.. set name is empty.....\n")));
 550
 551   ACE_DEBUG ((LM_DEBUG,
 552               ACE_TEXT ("(%t) task_count %d\n"),
 553               task_count.value () ));
 554
 555   // Close things down.
 556   andres->end ();
 557   peter->end ();
 558   helmut->end ();
 559   matias->end ();
 560
 561   ACE_Thread_Manager::instance ()->wait ();
 562   ACE_DEBUG ((LM_DEBUG,
 563               ACE_TEXT ("(%t) task_count %d\n"),
 564               task_count.value () ));
 565
 566   delete andres;
 567   delete peter;
 568   delete helmut;
 569   delete matias;
 570
 571 #else
 572   ACE_ERROR ((LM_INFO,
 573               ACE_TEXT ("threads not supported on this platform\n")));
 574 #endif /* ACE_HAS_THREADS */
 575   ACE_END_TEST;
 576   return 0;
 577 }