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