ACE/ace/Token.cpp

   1 #include "ace/Token.h"
   2
   3 #if !defined (__ACE_INLINE__)
   4 # include "ace/Token.inl"
   5 #endif /* __ACE_INLINE__ */
   6
   7
   8
   9 #if defined (ACE_HAS_THREADS)
  10
  11 #include "ace/Thread.h"
  12 #include "ace/Log_Category.h"
  13
  14 #if defined (ACE_TOKEN_DEBUGGING)
  15 // FUZZ: disable check_for_streams_include
  16 #include "ace/streams.h"
  17 #endif /* ACE_TOKEN_DEBUGGING */
  18
  19 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
  20
  21 ACE_ALLOC_HOOK_DEFINE(ACE_Token)
  22
  23 void
  24 ACE_Token::dump (void) const
  25 {
  26 #if defined (ACE_HAS_DUMP)
  27   ACE_TRACE ("ACE_Token::dump");
  28
  29   ACELIB_DEBUG ((LM_DEBUG, ACE_BEGIN_DUMP, this));
  30
  31   ACELIB_DEBUG ((LM_DEBUG, ACE_TEXT ("\nthread = %d"), ACE_Thread::self ()));
  32   // @@ Is there a portable way to do this?
  33   // ACELIB_DEBUG ((LM_DEBUG, "\nowner_ = %d", (long) this->owner_));
  34   ACELIB_DEBUG ((LM_DEBUG, ACE_TEXT ("\nowner_ addr = %x"), &this->owner_));
  35   ACELIB_DEBUG ((LM_DEBUG, ACE_TEXT ("\nwaiters_ = %d"), this->waiters_));
  36   ACELIB_DEBUG ((LM_DEBUG, ACE_TEXT ("\nin_use_ = %d"), this->in_use_));
  37   ACELIB_DEBUG ((LM_DEBUG, ACE_TEXT ("\nnesting level = %d"), this->nesting_level_));
  38   ACELIB_DEBUG ((LM_DEBUG, ACE_END_DUMP));
  39 #endif /* ACE_HAS_DUMP */
  40 }
  41
  42 ACE_Token::ACE_Token_Queue_Entry::ACE_Token_Queue_Entry (ACE_Thread_Mutex &m,
  43                                                          ACE_thread_t t_id)
  44   : next_ (0),
  45     thread_id_ (t_id),
  46 #if defined (ACE_TOKEN_USES_SEMAPHORE)
  47     cv_ (0),
  48 #else
  49     cv_ (m),
  50 #endif /* ACE_TOKEN_USES_SEMAPHORE */
  51     runable_ (0)
  52 {
  53 #if defined (ACE_TOKEN_USES_SEMAPHORE)
  54   ACE_UNUSED_ARG (m);
  55 #endif /* ACE_TOKEN_USES_SEMAPHORE */
  56
  57   ACE_TRACE ("ACE_Token::ACE_Token_Queue_Entry::ACE_Token_Queue_Entry");
  58 }
  59
  60 ACE_Token::ACE_Token_Queue_Entry::ACE_Token_Queue_Entry (ACE_Thread_Mutex &m,
  61                                                          ACE_thread_t t_id,
  62                                                          ACE_Condition_Attributes &attributes)
  63   : next_ (0),
  64     thread_id_ (t_id),
  65 #if defined (ACE_TOKEN_USES_SEMAPHORE)
  66     cv_ (0),
  67 #else
  68     cv_ (m, attributes),
  69 #endif /* ACE_TOKEN_USES_SEMAPHORE */
  70     runable_ (0)
  71 {
  72 #if defined (ACE_TOKEN_USES_SEMAPHORE)
  73   ACE_UNUSED_ARG (m);
  74   ACE_UNUSED_ARG (attributes);
  75 #endif /* ACE_TOKEN_USES_SEMAPHORE */
  76
  77   ACE_TRACE ("ACE_Token::ACE_Token_Queue_Entry::ACE_Token_Queue_Entry");
  78 }
  79
  80 ACE_Token::ACE_Token_Queue::ACE_Token_Queue (void)
  81   : head_ (0),
  82     tail_ (0)
  83 {
  84   ACE_TRACE ("ACE_Token::ACE_Token_Queue::ACE_Token_Queue");
  85 }
  86
  87 //
  88 // Remove an entry from the list.  Must be called with locks held.
  89 //
  90 void
  91 ACE_Token::ACE_Token_Queue::remove_entry (ACE_Token::ACE_Token_Queue_Entry *entry)
  92 {
  93   ACE_TRACE ("ACE_Token::ACE_Token_Queue::remove_entry");
  94   ACE_Token_Queue_Entry *curr = 0;
  95   ACE_Token_Queue_Entry *prev = 0;
  96
  97   if (this->head_ == 0)
  98     return;
  99
 100   for (curr = this->head_;
 101        curr != 0 && curr != entry;
 102        curr = curr->next_)
 103     prev = curr;
 104
 105   if (curr == 0)
 106     // Didn't find the entry...
 107     return;
 108   else if (prev == 0)
 109     // Delete at the head.
 110     this->head_ = this->head_->next_;
 111   else
 112     // Delete in the middle.
 113     prev->next_ = curr->next_;
 114
 115   // We need to update the tail of the list if we've deleted the last
 116   // entry.
 117   if (curr->next_ == 0)
 118     this->tail_ = prev;
 119 }
 120
 121 //
 122 // Add an entry into the list.  Must be called with locks held.
 123 //
 124 void
 125 ACE_Token::ACE_Token_Queue::insert_entry (ACE_Token::ACE_Token_Queue_Entry &entry,
 126                                           int requeue_position)
 127 {
 128   if (this->head_ == 0)
 129     {
 130       // No other threads - just add me
 131       this->head_ = &entry;
 132       this->tail_ = &entry;
 133     }
 134   else if (requeue_position == -1)
 135     {
 136       // Insert at the end of the queue.
 137       this->tail_->next_ = &entry;
 138       this->tail_ = &entry;
 139     }
 140   else if (requeue_position == 0)
 141     {
 142       // Insert at head of queue.
 143       entry.next_ = this->head_;
 144       this->head_ = &entry;
 145     }
 146   else
 147     // Insert in the middle of the queue somewhere.
 148     {
 149       // Determine where our thread should go in the queue of waiters.
 150
 151       ACE_Token::ACE_Token_Queue_Entry *insert_after = this->head_;
 152       while (requeue_position-- && insert_after->next_ != 0)
 153         insert_after = insert_after->next_;
 154
 155       entry.next_ = insert_after->next_;
 156
 157       if (entry.next_ == 0)
 158         this->tail_ = &entry;
 159
 160       insert_after->next_ = &entry;
 161     }
 162 }
 163
 164 ACE_Token::ACE_Token (const ACE_TCHAR *name, void *any)
 165   : lock_ (name, (ACE_mutexattr_t *) any),
 166     owner_ (ACE_OS::NULL_thread),
 167     in_use_ (0),
 168     waiters_ (0),
 169     nesting_level_ (0),
 170     attributes_ (USYNC_THREAD),
 171     queueing_strategy_ (FIFO)
 172 {
 173 //  ACE_TRACE ("ACE_Token::ACE_Token");
 174 }
 175
 176 ACE_Token::~ACE_Token (void)
 177 {
 178   ACE_TRACE ("ACE_Token::~ACE_Token");
 179 }
 180
 181 int
 182 ACE_Token::shared_acquire (void (*sleep_hook_func)(void *),
 183                            void *arg,
 184                            ACE_Time_Value *timeout,
 185                            ACE_Token_Op_Type op_type)
 186 {
 187   ACE_TRACE ("ACE_Token::shared_acquire");
 188   ACE_GUARD_RETURN (ACE_Thread_Mutex, ace_mon, this->lock_, -1);
 189
 190 #if defined (ACE_TOKEN_DEBUGGING)
 191   this->dump ();
 192 #endif /* ACE_TOKEN_DEBUGGING */
 193
 194   ACE_thread_t const thr_id = ACE_Thread::self ();
 195
 196   // Nobody holds the token.
 197   if (!this->in_use_)
 198     {
 199       // Its mine!
 200       this->in_use_ = op_type;
 201       this->owner_ = thr_id;
 202       return 0;
 203     }
 204
 205   // Someone already holds the token.
 206
 207   // Check if it is us.
 208   if (ACE_OS::thr_equal (thr_id, this->owner_))
 209     {
 210       ++this->nesting_level_;
 211       return 0;
 212     }
 213
 214   // Do a quick check for "polling" behavior.
 215   if (timeout != 0 && *timeout == ACE_Time_Value::zero)
 216     {
 217       errno = ETIME;
 218       return -1;
 219     }
 220
 221   //
 222   // We've got to sleep until we get the token.
 223   //
 224
 225   // Which queue we should end up in...
 226   ACE_Token_Queue *queue = (op_type == ACE_Token::READ_TOKEN
 227                             ? &this->readers_
 228                             : &this->writers_);
 229
 230   // Allocate queue entry on stack.  This works since we don't exit
 231   // this method's activation record until we've got the token.
 232   ACE_Token::ACE_Token_Queue_Entry my_entry (this->lock_,
 233                                              thr_id,
 234                                              this->attributes_);
 235   queue->insert_entry (my_entry, this->queueing_strategy_);
 236   ++this->waiters_;
 237
 238   // Execute appropriate <sleep_hook> callback.  (@@ should these
 239   // methods return a success/failure status, and if so, what should
 240   // we do with it?)
 241   int ret = 0;
 242   if (sleep_hook_func)
 243     {
 244       (*sleep_hook_func) (arg);
 245       ++ret;
 246     }
 247   else
 248     {
 249       // Execute virtual method.
 250       this->sleep_hook ();
 251       ++ret;
 252     }
 253
 254   bool timed_out = false;
 255   bool error = false;
 256
 257   // Sleep until we've got the token (ignore signals).
 258   do
 259     {
 260       int const result = my_entry.wait (timeout, this->lock_);
 261
 262       if (result == -1)
 263         {
 264           // Note, this should obey whatever thread-specific interrupt
 265           // policy is currently in place...
 266           if (errno == EINTR)
 267             continue;
 268
 269 #if defined (ACE_TOKEN_DEBUGGING)
 270           cerr << '(' << ACE_Thread::self () << ')'
 271                << " acquire: "
 272                << (errno == ETIME ? "timed out" : "error occurred")
 273                << endl;
 274 #endif /* ACE_TOKEN_DEBUGGING */
 275
 276           // We come here if a timeout occurs or some serious
 277           // ACE_Condition object error.
 278           if (errno == ETIME)
 279             timed_out = true;
 280           else
 281             error = true;
 282
 283           // Stop the loop.
 284           break;
 285         }
 286     }
 287   while (!ACE_OS::thr_equal (thr_id, this->owner_));
 288
 289   // Do this always and irrespective of the result of wait().
 290   --this->waiters_;
 291   queue->remove_entry (&my_entry);
 292
 293 #if defined (ACE_TOKEN_DEBUGGING)
 294   ACELIB_DEBUG ((LM_DEBUG, "(%t) ACE_Token::shared_acquire (UNBLOCKED)\n"));
 295 #endif /* ACE_TOKEN_DEBUGGING */
 296
 297   // If timeout occurred
 298   if (timed_out)
 299     {
 300       // This thread was still selected to own the token.
 301       if (my_entry.runable_)
 302         {
 303           // Wakeup next waiter since this thread timed out.
 304           this->wakeup_next_waiter ();
 305         }
 306
 307       // Return error.
 308      return -1;
 309     }
 310   else if (error)
 311     {
 312       // Return error.
 313       return -1;
 314     }
 315
 316   // If this is a normal wakeup, this thread should be runnable.
 317   ACE_ASSERT (my_entry.runable_);
 318
 319   return ret;
 320 }
 321
 322 // By default this is a no-op.
 323
 324 /* virtual */
 325 void
 326 ACE_Token::sleep_hook (void)
 327 {
 328   ACE_TRACE ("ACE_Token::sleep_hook");
 329 }
 330
 331 int
 332 ACE_Token::acquire (ACE_Time_Value *timeout)
 333 {
 334   ACE_TRACE ("ACE_Token::acquire");
 335   return this->shared_acquire (0, 0, timeout, ACE_Token::WRITE_TOKEN);
 336 }
 337
 338 // Acquire the token, sleeping until it is obtained or until <timeout>
 339 // expires.
 340
 341 int
 342 ACE_Token::acquire (void (*sleep_hook_func)(void *),
 343                     void *arg,
 344                     ACE_Time_Value *timeout)
 345 {
 346   ACE_TRACE ("ACE_Token::acquire");
 347   return this->shared_acquire (sleep_hook_func, arg, timeout, ACE_Token::WRITE_TOKEN);
 348 }
 349
 350 // Try to renew the token.
 351
 352 int
 353 ACE_Token::renew (int requeue_position,
 354                   ACE_Time_Value *timeout)
 355 {
 356   ACE_TRACE ("ACE_Token::renew");
 357   ACE_GUARD_RETURN (ACE_Thread_Mutex, ace_mon, this->lock_, -1);
 358
 359 #if defined (ACE_TOKEN_DEBUGGING)
 360   this->dump ();
 361 #endif /* ACE_TOKEN_DEBUGGING */
 362   // ACE_ASSERT (ACE_OS::thr_equal (ACE_Thread::self (), this->owner_));
 363
 364   // Check to see if there are any waiters worth giving up the lock
 365   // for.
 366
 367   // If no writers and either we are a writer or there are no readers.
 368   if (this->writers_.head_ == 0 &&
 369       (this->in_use_ == ACE_Token::WRITE_TOKEN ||
 370        this->readers_.head_ == 0))
 371     // Immediate return.
 372     return 0;
 373
 374   // We've got to sleep until we get the token again.
 375
 376   // Determine which queue should this thread go to.
 377   ACE_Token::ACE_Token_Queue *this_threads_queue =
 378     this->in_use_ == ACE_Token::READ_TOKEN ?
 379     &this->readers_ : &this->writers_;
 380
 381   ACE_Token::ACE_Token_Queue_Entry my_entry (this->lock_,
 382                                              this->owner_);
 383
 384   this_threads_queue->insert_entry (my_entry,
 385                                     // if requeue_position == 0 then we want to go next,
 386                                     // otherwise use the queueing strategy, which might also
 387                                     // happen to be 0.
 388                                     requeue_position == 0 ? 0 : this->queueing_strategy_);
 389   ++this->waiters_;
 390
 391   // Remember nesting level...
 392   int const save_nesting_level_ = this->nesting_level_;
 393
 394   // Reset state for new owner.
 395   this->nesting_level_ = 0;
 396
 397   // Wakeup waiter.
 398   this->wakeup_next_waiter ();
 399
 400   bool timed_out = false;
 401   bool error = false;
 402
 403   // Sleep until we've got the token (ignore signals).
 404   do
 405     {
 406       int const result = my_entry.wait (timeout, this->lock_);
 407
 408       if (result == -1)
 409         {
 410           // Note, this should obey whatever thread-specific interrupt
 411           // policy is currently in place...
 412           if (errno == EINTR)
 413             continue;
 414
 415 #if defined (ACE_TOKEN_DEBUGGING)
 416           cerr << '(' << ACE_Thread::self () << ')'
 417                << " renew: "
 418                << (errno == ETIME ? "timed out" : "error occurred")
 419                << endl;
 420 #endif /* ACE_TOKEN_DEBUGGING */
 421
 422           // We come here if a timeout occurs or some serious
 423           // ACE_Condition object error.
 424           if (errno == ETIME)
 425             timed_out = true;
 426           else
 427             error = true;
 428
 429           // Stop the loop.
 430           break;
 431         }
 432     }
 433   while (!ACE_OS::thr_equal (my_entry.thread_id_, this->owner_));
 434
 435   // Do this always and irrespective of the result of wait().
 436   --this->waiters_;
 437   this_threads_queue->remove_entry (&my_entry);
 438
 439 #if defined (ACE_TOKEN_DEBUGGING)
 440   ACELIB_DEBUG ((LM_DEBUG, "(%t) ACE_Token::renew (UNBLOCKED)\n"));
 441 #endif /* ACE_TOKEN_DEBUGGING */
 442
 443   // If timeout occurred
 444   if (timed_out)
 445     {
 446       // This thread was still selected to own the token.
 447       if (my_entry.runable_)
 448         {
 449           // Wakeup next waiter since this thread timed out.
 450           this->wakeup_next_waiter ();
 451         }
 452
 453       // Return error.
 454      return -1;
 455     }
 456   else if (error)
 457     {
 458       // Return error.
 459       return -1;
 460     }
 461
 462   // If this is a normal wakeup, this thread should be runnable.
 463   ACE_ASSERT (my_entry.runable_);
 464
 465   // Reinstate nesting level.
 466   this->nesting_level_ = save_nesting_level_;
 467
 468   return 0;
 469 }
 470
 471 // Release the current holder of the token (which had
 472 // better be the caller's thread!).
 473
 474 int
 475 ACE_Token::release (void)
 476 {
 477   ACE_TRACE ("ACE_Token::release");
 478   ACE_GUARD_RETURN (ACE_Thread_Mutex, ace_mon, this->lock_, -1);
 479
 480 #if defined (ACE_TOKEN_DEBUGGING)
 481   this->dump ();
 482 #endif /* ACE_TOKEN_DEBUGGING */
 483
 484   // Nested release...
 485   if (this->nesting_level_ > 0)
 486     --this->nesting_level_;
 487   else
 488     {
 489       //
 490       // Regular release...
 491       //
 492
 493       // Wakeup waiter.
 494       this->wakeup_next_waiter ();
 495     }
 496
 497   return 0;
 498 }
 499
 500 void
 501 ACE_Token::wakeup_next_waiter (void)
 502 {
 503   ACE_TRACE ("ACE_Token::wakeup_next_waiter");
 504
 505   // Reset state for new owner.
 506   this->owner_ = ACE_OS::NULL_thread;
 507   this->in_use_ = 0;
 508
 509   // Any waiters...
 510   if (this->writers_.head_ == 0 &&
 511       this->readers_.head_ == 0)
 512     {
 513       // No more waiters...
 514       return;
 515     }
 516
 517   // Wakeup next waiter.
 518   ACE_Token_Queue *queue = 0;
 519
 520   // Writer threads get priority to run first.
 521   if (this->writers_.head_ != 0)
 522     {
 523       this->in_use_ = ACE_Token::WRITE_TOKEN;
 524       queue = &this->writers_;
 525     }
 526   else
 527     {
 528       this->in_use_ = ACE_Token::READ_TOKEN;
 529       queue = &this->readers_;
 530     }
 531
 532   // Wake up waiter and make it runable.
 533   queue->head_->runable_ = 1;
 534   queue->head_->signal ();
 535   this->owner_ = queue->head_->thread_id_;
 536 }
 537
 538 ACE_END_VERSIONED_NAMESPACE_DECL
 539
 540 #endif /* ACE_HAS_THREADS */