ACE/examples/Threads/future1.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    future1.cpp
   5  *
   6  *  This example tests the ACE Future.
   7  *
   8  *  @author Andres Kruse <Andres.Kruse@cern.ch> and Douglas C. Schmidt <d.schmidt@vanderbilt.edu>
   9  */
  10 //=============================================================================
  11
  12
  13 #include "ace/OS_NS_string.h"
  14 #include "ace/OS_NS_unistd.h"
  15 #include "ace/OS_main.h"
  16 #include "ace/ACE.h"
  17 #include "ace/Task.h"
  18 #include "ace/Thread_Mutex.h"
  19 #include "ace/Message_Queue.h"
  20 #include "ace/Future.h"
  21 #include "ace/Method_Request.h"
  22 #include "ace/Activation_Queue.h"
  23 #include "ace/Atomic_Op.h"
  24 #include <memory>
  25
  26 #if defined (ACE_HAS_THREADS)
  27
  28 typedef ACE_Atomic_Op<ACE_Thread_Mutex, int> ATOMIC_INT;
  29
  30 // a counter for the tasks..
  31 static ATOMIC_INT task_count (0);
  32
  33 // a counter for the futures..
  34 static ATOMIC_INT future_count (0);
  35 static ATOMIC_INT future_no (0);
  36
  37 // a counter for the capsules..
  38 static ATOMIC_INT capsule_count (0);
  39 static ATOMIC_INT capsule_no (0);
  40
  41 // a counter for the method objects...
  42 static ATOMIC_INT methodobject_count (0);
  43 static ATOMIC_INT methodobject_no (0);
  44
  45 /**
  46  * @class Scheduler
  47  *
  48  * @brief Active Object Scheduler.
  49  */
  50 class Scheduler : public ACE_Task_Base
  51 {
  52   friend class Method_RequestWork;
  53 public:
  54   Scheduler (const char *, Scheduler * = 0);
  55   virtual ~Scheduler ();
  56
  57   //FUZZ: disable check_for_lack_ACE_OS
  58   ///FUZZ: enable check_for_lack_ACE_OS
  59   virtual int open (void *args = 0);
  60   virtual int close (u_long flags = 0);
  61
  62   virtual int svc ();
  63
  64   ACE_Future<u_long> work (u_long param, int count = 1);
  65   ACE_Future<const char*> name ();
  66   void end ();
  67
  68   u_long work_i (u_long, int);
  69   const char *name_i ();
  70
  71 private:
  72   char *name_;
  73   ACE_Activation_Queue activation_queue_;
  74   Scheduler *scheduler_;
  75 };
  76
  77 /**
  78  * @class Method_Request_work
  79  *
  80  * @brief Reification of the <work> method.
  81  */
  82 class Method_Request_work : public ACE_Method_Request
  83 {
  84 public:
  85   Method_Request_work (Scheduler *, u_long, int, ACE_Future<u_long> &);
  86   virtual ~Method_Request_work ();
  87   virtual int call ();
  88
  89 private:
  90   Scheduler *scheduler_;
  91   u_long param_;
  92   int count_;
  93   ACE_Future<u_long> future_result_;
  94 };
  95
  96 Method_Request_work::Method_Request_work (Scheduler* new_Scheduler,
  97                                         u_long new_param,
  98                                         int new_count,
  99                                         ACE_Future<u_long> &new_result)
 100   :   scheduler_ (new_Scheduler),
 101       param_ (new_param),
 102       count_ (new_count),
 103       future_result_ (new_result)
 104 {
 105   ACE_DEBUG ((LM_DEBUG,
 106               "(%t) Method_Request_work created\n"));
 107 }
 108
 109 Method_Request_work::~Method_Request_work ()
 110 {
 111   ACE_DEBUG ((LM_DEBUG, "(%t) Method_Request_work will be deleted.\n"));
 112 }
 113
 114
 115 int
 116 Method_Request_work::call ()
 117 {
 118   return this->future_result_.set (this->scheduler_->work_i (this->param_, this->count_));
 119 }
 120
 121 /**
 122  * @class Method_Request_name
 123  *
 124  * @brief Reification of the <name> method.
 125  */
 126 class Method_Request_name : public ACE_Method_Request
 127 {
 128 public:
 129   Method_Request_name (Scheduler *, ACE_Future<const char*> &);
 130   virtual ~Method_Request_name ();
 131   virtual int call ();
 132
 133 private:
 134   Scheduler *scheduler_;
 135   ACE_Future<const char *> future_result_;
 136 };
 137
 138 Method_Request_name::Method_Request_name (Scheduler *new_scheduler,
 139                                         ACE_Future<const char *> &new_result)
 140   : scheduler_ (new_scheduler),
 141     future_result_ (new_result)
 142 {
 143   ACE_DEBUG ((LM_DEBUG,
 144               "(%t) Method_Request_name created\n"));
 145 }
 146
 147 Method_Request_name::~Method_Request_name ()
 148 {
 149   ACE_DEBUG ((LM_DEBUG,
 150               "(%t) Method_Request_name will be deleted.\n"));
 151 }
 152
 153 int
 154 Method_Request_name::call ()
 155 {
 156   return future_result_.set (scheduler_->name_i ());
 157 }
 158
 159 /**
 160  * @class Method_Request_end
 161  *
 162  * @brief Reification of the <end> method.
 163  */
 164 class Method_Request_end : public ACE_Method_Request
 165 {
 166 public:
 167   Method_Request_end (Scheduler *new_scheduler): scheduler_ (new_scheduler) {}
 168   virtual ~Method_Request_end () {}
 169   virtual int call () { return -1; }
 170
 171 private:
 172   /// Keep track of our scheduler.
 173   Scheduler *scheduler_;
 174 };
 175
 176 // Constructor.
 177 Scheduler::Scheduler (const char *newname,
 178                       Scheduler *new_scheduler)
 179 {
 180   ACE_NEW (this->name_, char[ACE_OS::strlen (newname) + 1]);
 181   ACE_OS::strcpy (this->name_, newname);
 182   this->scheduler_ = new_scheduler;
 183   ACE_DEBUG ((LM_DEBUG, "(%t) Scheduler %s created\n", this->name_));
 184 }
 185
 186 // Destructor
 187 Scheduler::~Scheduler ()
 188 {
 189   ACE_DEBUG ((LM_DEBUG, "(%t) Scheduler %s will be destroyed\n", this->name_));
 190   delete [] this->name_;
 191 }
 192
 193 // open
 194 int
 195 Scheduler::open (void *)
 196 {
 197   task_count++;
 198   ACE_DEBUG ((LM_DEBUG, "(%t) Scheduler %s open\n", this->name_));
 199   return this->activate (THR_BOUND);
 200 }
 201
 202 // close
 203 int
 204 Scheduler::close (u_long)
 205 {
 206   ACE_DEBUG ((LM_DEBUG, "(%t) Scheduler %s close\n", this->name_));
 207   task_count--;
 208   return 0;
 209 }
 210
 211 // service..
 212 int
 213 Scheduler::svc ()
 214 {
 215   for (;;)
 216     {
 217       // Dequeue the next method object (we use an unique pointer in
 218       // case an exception is thrown in the <call>).
 219       std::unique_ptr<ACE_Method_Request> mo (this->activation_queue_.dequeue ());
 220
 221       ACE_DEBUG ((LM_DEBUG, "(%t) calling method object\n"));
 222       // Call it.
 223       if (mo->call () == -1)
 224         break;
 225       // Destructor automatically deletes it.
 226     }
 227
 228   /* NOTREACHED */
 229   return 0;
 230 }
 231
 232 void
 233 Scheduler::end ()
 234 {
 235   this->activation_queue_.enqueue (new Method_Request_end (this));
 236 }
 237
 238
 239 // Here's where the Work takes place.
 240 u_long
 241 Scheduler::work_i (u_long param,
 242                    int count)
 243 {
 244   ACE_UNUSED_ARG (count);
 245
 246   return ACE::is_prime (param, 2, param / 2);
 247 }
 248
 249 const char *
 250 Scheduler::name_i ()
 251 {
 252   char *the_name;
 253
 254   ACE_NEW_RETURN (the_name, char[ACE_OS::strlen (this->name_) + 1], 0);
 255   ACE_OS::strcpy (the_name, this->name_);
 256
 257   return the_name;
 258 }
 259
 260 ACE_Future<const char *>
 261 Scheduler::name ()
 262 {
 263   if (this->scheduler_)
 264     // Delegate to the Scheduler.
 265     return this->scheduler_->name ();
 266   else
 267     {
 268       ACE_Future<const char*> new_future;
 269
 270       // @@ What happens if new fails here?
 271       this->activation_queue_.enqueue
 272         (new Method_Request_name (this, new_future));
 273
 274       return new_future;
 275     }
 276 }
 277
 278 ACE_Future<u_long>
 279 Scheduler::work (u_long newparam,
 280                  int newcount)
 281 {
 282   if (this->scheduler_) {
 283     return this->scheduler_->work (newparam, newcount);
 284   }
 285   else {
 286     ACE_Future<u_long> new_future;
 287
 288     this->activation_queue_.enqueue
 289       (new Method_Request_work (this, newparam, newcount, new_future));
 290     return new_future;
 291   }
 292 }
 293
 294 // @@ These values should be set by the command line options!
 295
 296 // Total number of loops.
 297 static size_t n_loops = 100;
 298
 299 int
 300 ACE_TMAIN (int, ACE_TCHAR *[])
 301 {
 302   Scheduler *andres, *peter, *helmut, *matias;
 303
 304   // Create active objects..
 305   // @@ Should "open" be subsumed within the constructor of
 306   // Scheduler()?
 307   ACE_NEW_RETURN (andres, Scheduler ("andres"), -1);
 308   andres->open ();
 309   ACE_NEW_RETURN (peter, Scheduler ("peter"), -1);
 310   peter->open ();
 311   ACE_NEW_RETURN (helmut, Scheduler ("helmut"), -1);
 312   helmut->open ();
 313
 314   // Matias passes all asynchronous method calls on to Andres...
 315   ACE_NEW_RETURN (matias, Scheduler ("matias", andres), -1);
 316   matias->open ();
 317
 318   for (size_t i = 0; i < n_loops; i++)
 319     {
 320       {
 321         ACE_Future<u_long> fresulta, fresultb, fresultc, fresultd, fresulte;
 322         ACE_Future<const char *> fname;
 323
 324         ACE_DEBUG ((LM_DEBUG, "(%t) going to do a non-blocking call\n"));
 325
 326         fresulta = andres->work (9013);
 327         fresultb = peter->work (9013);
 328         fresultc = helmut->work (9013);
 329         fresultd = matias->work (9013);
 330         fname = andres->name ();
 331
 332         // see if the result is available...
 333         if (fresulta.ready ())
 334           ACE_DEBUG ((LM_DEBUG, "(%t) wow.. work is ready.....\n"));
 335
 336         ACE_DEBUG ((LM_DEBUG, "(%t) non-blocking call done... now blocking...\n"));
 337
 338         // Save the result of fresulta.
 339
 340         fresulte = fresulta;
 341
 342         if (i % 3 == 0)
 343           {
 344             // Every 3rd time... disconnect the futures...
 345             // but "fresulte" should still contain the result...
 346             fresulta.cancel (10);
 347             fresultb.cancel (20);
 348             fresultc.cancel (30);
 349             fresultd.cancel (40);
 350           }
 351
 352         u_long resulta = 0, resultb = 0, resultc = 0, resultd = 0, resulte = 0;
 353
 354         fresulta.get (resulta);
 355         fresultb.get (resultb);
 356         fresultc.get (resultc);
 357         fresultd.get (resultd);
 358         fresulte.get (resulte);
 359
 360         ACE_DEBUG ((LM_DEBUG, "(%t) result a %u\n", (u_int) resulte));
 361         ACE_DEBUG ((LM_DEBUG, "(%t) result b %u\n", (u_int) resulta));
 362         ACE_DEBUG ((LM_DEBUG, "(%t) result c %u\n", (u_int) resultb));
 363         ACE_DEBUG ((LM_DEBUG, "(%t) result d %u\n", (u_int) resultc));
 364         ACE_DEBUG ((LM_DEBUG, "(%t) result e %u\n", (u_int) resultd));
 365
 366         const char *name = 0;
 367
 368         fname.get (name);
 369
 370         ACE_DEBUG ((LM_DEBUG, "(%t) name %s\n", name));
 371         delete [] (char *) name;
 372       }
 373
 374       ACE_DEBUG ((LM_DEBUG,
 375                   "(%t) task_count %d future_count %d capsule_count %d methodobject_count %d\n",
 376                   task_count.value (),
 377                   future_count.value (),
 378                   capsule_count.value (),
 379                   methodobject_count.value ()));
 380     }
 381
 382   // Close things down.
 383   andres->end ();
 384   peter->end ();
 385   helmut->end ();
 386   matias->end ();
 387
 388   ACE_OS::sleep (2);
 389
 390   ACE_DEBUG ((LM_DEBUG,
 391               "(%t) task_count %d future_count %d capsule_count %d methodobject_count %d\n",
 392               task_count.value (),
 393               future_count.value (),
 394               capsule_count.value (),
 395               methodobject_count.value ()));
 396
 397   ACE_DEBUG ((LM_DEBUG,"(%t) th' that's all folks!\n"));
 398
 399   ACE_OS::sleep (5);
 400   return 0;
 401 }
 402
 403 #else
 404 int
 405 ACE_TMAIN (int, ACE_TCHAR *[])
 406 {
 407   ACE_ERROR ((LM_ERROR, "threads not supported on this platform\n"));
 408   return 0;
 409 }
 410 #endif /* ACE_HAS_THREADS */