Added locks to ensure no race conditions

2 files changed, 184 insertions, 143 deletions

diff --git a/TAO/orbsvcs/orbsvcs/Concurrency/CC_LockSet.cpp b/TAO/orbsvcs/orbsvcs/Concurrency/CC_LockSet.cpp
index 24742f0561f..201b4d4c695 100644
--- a/TAO/orbsvcs/orbsvcs/Concurrency/CC_LockSet.cpp
+++ b/TAO/orbsvcs/orbsvcs/Concurrency/CC_LockSet.cpp
@@ -19,10 +19,9 @@
 // Default constructor.
 
 CC_LockSet::CC_LockSet (void)
-  : strongest_held_ (CC_EM),
-    waiting_calls_ (0),
-    related_lockset_ (0)
+  : related_lockset_ (0)
 {
+  ACE_NEW(mlock_, ACE_Thread_Mutex);
   TAO_TRY
     {
       this->Init(TAO_TRY_ENV);
@@ -38,10 +37,9 @@ CC_LockSet::CC_LockSet (void)
 // Constructor used to create related lock sets.
 
 CC_LockSet::CC_LockSet (CosConcurrencyControl::LockSet_ptr related)
-  : strongest_held_ (CC_EM),
-    waiting_calls_ (0),
-    related_lockset_ (related)
+  : related_lockset_ (related)
 {
+  ACE_NEW(mlock_, ACE_Thread_Mutex);
   TAO_TRY
     {
       this->Init(TAO_TRY_ENV);
@@ -75,15 +73,27 @@ CC_LockSet::Init(CORBA::Environment &_env)
 
 CC_LockSet::~CC_LockSet (void)
 {
-  // Nothing is dynamically allocated
+  if(this->mlock_!=0)
+    delete mlock_;
 }
 
-CORBA::Boolean CC_LockSet::compatible(CC_LockModeEnum mh, CC_LockModeEnum mr)
+// Returns true if the requested lock mode is compatible with the
+// modes held. False otherwise.
+CORBA::Boolean CC_LockSet::compatible(CC_LockModeEnum mr)
 {
-  if(mh==CC_EM) // no locks held
-    return CORBA::B_TRUE;
+  int i=CC_IR;
 
-  return compatible_[mh][mr];
+  while(i<=CC_W)
+    {
+      if(this->lock_[i]>0)
+        if(this->compatible_[i][mr]==CORBA::B_FALSE)
+          {
+            return CORBA::B_FALSE;
+          }
+      i++;
+    }
+
+  return CORBA::B_TRUE;
 }
 
 // Locks the lock in the desired mode. Blocks until success.
@@ -97,30 +107,16 @@ CC_LockSet::lock (CosConcurrencyControl::lock_mode mode,
   CC_LockModeEnum lm = lmconvert(mode);
 
   // Check to see if the requested mode is compatible with the
-  // modes held so far. If not put the request on hold.  @@TAO I'm
-  // not shure whether we can run into race conditions on the
-  // variables here (waiting_calls_, lock_, and strongest_held_).
-  if(compatible(strongest_held_,lm)==CORBA::B_FALSE)
-    {
-      waiting_calls_++;
-      if(semaphore_.acquire()==-1)
-        TAO_THROW (CORBA::INTERNAL (CORBA::COMPLETED_NO));
-      lock_[lm]++;
-      strongest_held_ = lm;
-    }
-  else
-    {
-      lock_[lm]++;
-      // If the mode granted is stronger than the strongest mode
-      // held so far, we must update the strongest_held_ variable.
-      if(lm>strongest_held_)
-        strongest_held_ = lm;
-    }
+  // modes held so far. If not put the request on hold.
 
-  ACE_DEBUG ((LM_DEBUG,
-              "waiting_calls_: %i, strongest_held_: %i, IR: %i, R: %i, U: %i, IW: %i, W: %i\n",
-              waiting_calls_, strongest_held_,
-              lock_[CC_IR], lock_[CC_R], lock_[CC_U], lock_[CC_IW], lock_[CC_W]));
+  if(this->lock_d(lm)==1)
+    if(semaphore_.acquire()==-1)
+      TAO_THROW (CORBA::INTERNAL (CORBA::COMPLETED_NO));
+
+  //  ACE_DEBUG ((LM_DEBUG,
+  //              "waiting_calls_: %i, IR: %i, R: %i, U: %i, IW: %i, W: %i\n",
+  //              lock_queue_.size(),
+  //              lock_[CC_IR], lock_[CC_R], lock_[CC_U], lock_[CC_IW], lock_[CC_W]));
 }
 
 // Tries to lock. If it is not possible false is returned.
@@ -129,41 +125,26 @@ CORBA::Boolean
 CC_LockSet::try_lock (CosConcurrencyControl::lock_mode mode,
                       CORBA::Environment &_env)
 {
-  CORBA::Boolean success = CORBA::B_TRUE;
+  //  CORBA::Boolean success = CORBA::B_TRUE;
 
   CC_LockModeEnum lm = lmconvert(mode);
 
   ACE_DEBUG ((LM_DEBUG,
               "CC_LockSet::try_lock\n"));
-  TAO_TRY
-    {
-      // if the mode is compatible we lock the lock otherwise we
-      // return false because we cannot grant the lock.
-      if(compatible(strongest_held_,lm)==CORBA::B_FALSE)
-        {
-          success = CORBA::B_FALSE;
-        }
-      else
-        {
-          this->lock (mode, TAO_TRY_ENV);
-        }
-      TAO_CHECK_ENV;
-    }
-  TAO_CATCHANY
-    {
-      TAO_RETHROW_RETURN (CORBA::B_FALSE);
-    }
-  TAO_ENDTRY;
-  ACE_DEBUG ((LM_DEBUG,
-              "waiting_calls_: %i, strongest_held_: %i, IR: %i, R: %i, U: %i, IW: %i, W: %i\n",
-              waiting_calls_, strongest_held_,
-              lock_[CC_IR], lock_[CC_R], lock_[CC_U], lock_[CC_IW], lock_[CC_W]));
-  return success;
+  if(this->try_lock_d(lm)==0)
+    return CORBA::B_FALSE; // success = CORBA::B_FALSE;
+  else
+    return CORBA::B_TRUE;  // success = CORBA::B_TRUE;
+
+  //ACE_DEBUG ((LM_DEBUG,
+  //            "waiting_calls_: %i, IR: %i, R: %i, U: %i, IW: %i, W: %i\n",
+  //            lock_queue_.size(),
+  //            lock_[CC_IR], lock_[CC_R], lock_[CC_U], lock_[CC_IW], lock_[CC_W]));
+  //return success;
 }
 
 // Converts the enum from the spec to the internally (ordered)
-// enum. It could be faster to use a static array to do the conversion
-// but the switch is more portable.
+// enum.
 
 CC_LockModeEnum
 CC_LockSet::lmconvert(CosConcurrencyControl::lock_mode mode)
@@ -172,41 +153,19 @@ CC_LockSet::lmconvert(CosConcurrencyControl::lock_mode mode)
     {
     case CosConcurrencyControl::intention_read:
       return CC_IR;
-      break;
     case CosConcurrencyControl::read:
       return CC_R;
-      break;
     case CosConcurrencyControl::upgrade:
       return CC_U;
-      break;
     case CosConcurrencyControl::intention_write:
       return CC_IW;
-      break;
     case CosConcurrencyControl::write:
       return CC_W;
-      break;
     default:
       return CC_EM;
     }
 }
 
-// Loop through the locks held up to the requested mode and return
-// true if any of the locks are held
-
-CORBA::Boolean
-CC_LockSet::weaker_held(CC_LockModeEnum mode)
-{
-  int li = CC_IR;
-
-  while(li<=mode)
-    {
-      if(lock_[li]>0)
-        return CORBA::B_TRUE;
-      li++;
-    }
-  return CORBA::B_FALSE;
-}
-
 // Unlock the lock
 
 void
@@ -218,9 +177,11 @@ CC_LockSet::unlock (CosConcurrencyControl::lock_mode mode,
 
   CC_LockModeEnum lm = lmconvert(mode);
 
+  ACE_GUARD (ACE_Thread_Mutex, ace_mon, *this->mlock_);
+
   TAO_TRY
     {
-      if(lock_[lm]==0) // This lock is not held
+      if(lock_[lm]==0) // This lock is not held.
         TAO_THROW (CosConcurrencyControl::LockNotHeld);
       else
         lock_[lm]--;
@@ -228,34 +189,21 @@ CC_LockSet::unlock (CosConcurrencyControl::lock_mode mode,
       // If we do not have a lock held in a weaker mode than the
       // strongest held and we have requests on the semaphore signal
       // the semaphore.
-      if(weaker_held(strongest_held_)==CORBA::B_FALSE &&
-         waiting_calls_>0)
+      while(lock_queue_.size()>0)
         {
-          waiting_calls_--;
-          if(semaphore_.release()==-1)
-            TAO_THROW (CORBA::INTERNAL (CORBA::COMPLETED_NO));
-          // Here's a problem. We do not know the mode of the lock
-          // request on the semaphore. That turns out to be handled in
-          // the lock method and thus isn't a problem.
-        }
-      else
-        {
-          // check if any locks are held. if not set strongest held to CC_EM.
-          int locks_held = 0;
-          int strongest_so_far = 0;
-          for(int i=CC_IR; i<=CC_W; i++)
+          CC_LockModeEnum lock_on_queue = CC_EM;
+          lock_queue_.dequeue_head(lock_on_queue);
+          if(compatible(lock_on_queue)==CORBA::B_TRUE)
             {
-              locks_held+=lock_[i];
-              if(i<=strongest_held_ &&
-                 lock_[i]>0)
-                strongest_so_far = i;
+              if(semaphore_.release()==-1)
+                TAO_THROW (CORBA::INTERNAL (CORBA::COMPLETED_NO));
+              lock_[lock_on_queue]++;
             }
-          // We need to make sure, that the strongest_held_ is
-          // decremented to the actual strongest held mode.
-          if(locks_held==0)
-            strongest_held_ = CC_EM;
           else
-            strongest_held_ = (CC_LockModeEnum) strongest_so_far;
+            {
+              lock_queue_.enqueue_head(lock_on_queue);
+              break;
+            }
         }
     }
   TAO_CATCHANY
@@ -264,8 +212,8 @@ CC_LockSet::unlock (CosConcurrencyControl::lock_mode mode,
     }
   TAO_ENDTRY;
   ACE_DEBUG ((LM_DEBUG,
-              "waiting_calls_: %i, strongest_held_: %i, IR: %i, R: %i, U: %i, IW: %i, W: %i\n",
-              waiting_calls_, strongest_held_,
+              "waiting_calls_: %i, IR: %i, R: %i, U: %i, IW: %i, W: %i\n",
+              lock_queue_.size(),
               lock_[CC_IR], lock_[CC_R], lock_[CC_U], lock_[CC_IW], lock_[CC_W]));
 }
 
@@ -283,24 +231,15 @@ CC_LockSet::change_mode (CosConcurrencyControl::lock_mode held_mode,
 
   TAO_TRY
     {
-      if(lock_[lm_held] == 0) // This lock is not held
+      if(this->lock_held(lm_held) == 0) // This lock is not held
         TAO_THROW (CosConcurrencyControl::LockNotHeld);
       else
         {
-          // If the new mode is compatible with the strongest mode
-          // held, we just grant the new mode. Otherwise we must
-          // unlock the old mode and lock the new.
-          if(compatible(strongest_held_, lm_new)==CORBA::B_TRUE)
+          if(this->change_mode_d(lm_held, lm_new)==1)
             {
-              lock_[lm_held]--;
-              lock_[lm_new]++;
-            }
-          else
-            {
-              this->unlock(held_mode, TAO_TRY_ENV);
-              TAO_CHECK_ENV;
-              this->lock(new_mode, TAO_TRY_ENV);
-              TAO_CHECK_ENV;
+              this->unlock(held_mode, _env);
+              if(semaphore_.acquire()==-1)
+                TAO_THROW (CORBA::INTERNAL (CORBA::COMPLETED_NO));
             }
         }
     }
@@ -311,16 +250,92 @@ CC_LockSet::change_mode (CosConcurrencyControl::lock_mode held_mode,
   TAO_ENDTRY;
 
   ACE_DEBUG ((LM_DEBUG,
-              "waiting_calls_: %i, strongest_held_: %i, IR: %i, R: %i, U: %i, IW: %i, W: %i\n",
-              waiting_calls_, strongest_held_,
+              "waiting_calls_: %i, IR: %i, R: %i, U: %i, IW: %i, W: %i\n",
+              lock_queue_.size(),
               lock_[CC_IR], lock_[CC_R], lock_[CC_U], lock_[CC_IW], lock_[CC_W]));
 }
 
+int
+CC_LockSet::lock_d(CC_LockModeEnum lm)
+{
+  ACE_GUARD_RETURN (ACE_Thread_Mutex, ace_mon, *this->mlock_, 1);
+  // If the lock is not compatible with the locks we hold allready or
+  // there is lock requests in the queue we cannot grant the lock and
+  // thus we queue the request. Otherwise update the lock count.
+  if(compatible(lm)==CORBA::B_FALSE || lock_queue_.size()>0)
+    {
+      // Put the lock mode in the queue
+      lock_queue_.enqueue_tail(lm);
+      this->dump();
+      return 1; // Lock the semaphore
+    }
+  else
+    {
+      lock_[lm]++;
+    }
+  this->dump();
+  return 0;
+}
+
+int
+CC_LockSet::try_lock_d(CC_LockModeEnum lm)
+{
+  ACE_GUARD_RETURN (ACE_Thread_Mutex, ace_mon, *this->mlock_, 1);
+  // If the lock we try is compatible with the locks we hold we just
+  // opdates the count. Otherwise we return false.
+  if(compatible(lm)==CORBA::B_FALSE)
+    {
+      this->dump();
+      return 0;
+    }
+  else
+    {
+      lock_[lm]++;
+    }
+  this->dump();
+  return 1;
+}
+
+int
+CC_LockSet::change_mode_d(CC_LockModeEnum lm_held,
+                          CC_LockModeEnum lm_new)
+{
+  ACE_GUARD_RETURN (ACE_Thread_Mutex, ace_mon, *this->mlock_, 1);
+  // If the new mode is compatible with the modes we hold we change
+  // the counts for the two locks. If not we must queue the new
+  // request. We can decrement the count for the old mode without
+  // signalling the semaphore because we know we only check modes
+  // granted this far.
+  lock_[lm_held]--;
+  if(compatible(lm_new)==CORBA::B_TRUE)
+    {
+      lock_[lm_new]++;
+      this->dump();
+      return 0;
+    }
+  else
+    {
+      lock_[lm_held]++;
+      lock_queue_.enqueue_tail(lm_new);
+      this->dump();
+      return 1;
+    }
+}
+
+int
+CC_LockSet::lock_held(CC_LockModeEnum lm)
+{
+  ACE_GUARD_RETURN (ACE_Thread_Mutex, ace_mon, *this->mlock_, 1);
+  if(lock_[lm]>0)
+    return 1;
+  else
+    return 0;
+}
+
 void
 CC_LockSet::dump(void)
 {
-  printf("waiting_calls_: %i, strongest_held_: %i, ",
-         waiting_calls_, strongest_held_);
+  printf("waiting_calls_: %i, ", lock_queue_.size());
   for(int i=CC_IR; i<=CC_W; i+=1)
     printf("%i ", lock_[i]);
   printf("\n");
@@ -330,5 +345,5 @@ CORBA::Boolean CC_LockSet::compatible_[NUMBER_OF_LOCK_MODES][NUMBER_OF_LOCK_MODE
   {CORBA::B_TRUE, CORBA::B_TRUE, CORBA::B_TRUE, CORBA::B_TRUE, CORBA::B_FALSE},
   {CORBA::B_TRUE, CORBA::B_TRUE, CORBA::B_TRUE, CORBA::B_FALSE, CORBA::B_FALSE},
   {CORBA::B_TRUE, CORBA::B_TRUE, CORBA::B_FALSE, CORBA::B_FALSE, CORBA::B_FALSE},
-  {CORBA::B_TRUE, CORBA::B_FALSE, CORBA::B_FALSE, CORBA::B_FALSE, CORBA::B_FALSE},
+  {CORBA::B_TRUE, CORBA::B_FALSE, CORBA::B_FALSE, CORBA::B_TRUE, CORBA::B_FALSE},
   {CORBA::B_FALSE, CORBA::B_FALSE, CORBA::B_FALSE, CORBA::B_FALSE, CORBA::B_FALSE}};
diff --git a/TAO/orbsvcs/orbsvcs/Concurrency/CC_LockSet.h b/TAO/orbsvcs/orbsvcs/Concurrency/CC_LockSet.h
index 2e36267e71f..73179519b72 100644
--- a/TAO/orbsvcs/orbsvcs/Concurrency/CC_LockSet.h
+++ b/TAO/orbsvcs/orbsvcs/Concurrency/CC_LockSet.h
@@ -36,6 +36,7 @@
 #define _CC_LOCKSET_H
 
 #include "ace/Synch.h"
+#include "ace/Token.h"
 #include "orbsvcs/CosConcurrencyControlS.h"
 
 #define NUMBER_OF_LOCK_MODES 5
@@ -45,7 +46,10 @@
 
 typedef enum {CC_EM=-1, CC_IR=0, CC_R, CC_U, CC_IW, CC_W} CC_LockModeEnum;
 // Enummeration representing the lock modes. The incomming request is
-// always converted to this representation
+// always converted to this representation. There are two reasons for
+// this: Firstly the lock modes are ordered from weakest to strongest
+// in the internal representation, and secondly it is possible to
+// indicate a 'non-mode' (CC_EM)
 
 class TAO_ORBSVCS_Export CC_LockSet :  public POA_CosConcurrencyControl::LockSet
   // = TITLE
@@ -99,28 +103,46 @@ private:
   // Initiatlizes the lock set array and acquires the initial
   // semaphore.
 
-  CORBA::Boolean weaker_held(CC_LockModeEnum mode);
-  // Checks if the lock set contains locked locks of weaker mode than
-  // mode
-
-  CORBA::Boolean compatible(CC_LockModeEnum mh, CC_LockModeEnum mr);
+  CORBA::Boolean compatible(CC_LockModeEnum mr);
   // Returns true if the held lock and the requested lock are compatible
 
+  // The _d functions below ensures atomical access the the state data
+  // for the lock set. The functions acquires a thread lock in order
+  // to insure consistency within the lock set. The return value
+  // typically indicates whether the current thread should be
+  // suspended or not (by locking the semaphore.
+
+  int lock_d(CC_LockModeEnum lm);
+  // Locks the access to the data and decides whether to lock or
+  // not. Returns 1 if the semaphore should be locked.
+
+  // int unlock_d(CosConcurrencyControl::lock_mode lm);
+  // This function is not necessary because we lock access to the data
+  // and unlocks the semaphore until an invalid lock mode is first on
+  // the queue. Thereafter we release the lock.
+
+  int try_lock_d(CC_LockModeEnum lm);
+  // Locks the access to the data and determines whether to return
+  // true or false. Returns 1 if true should be returned.
+
+  int change_mode_d(CC_LockModeEnum lm_held,
+                    CC_LockModeEnum lm_new);
+  // Locks access to the data and determines if the semaphore should
+  // be locked. Returns 1 if the semaphore should be locked.
+
+  int lock_held(CC_LockModeEnum lm);
+  // Locks access ti the data and checks whether the lock is held.
+
   int lock_[NUMBER_OF_LOCK_MODES];
   // An array of lock counters that counts how many locks of that type
   // that the lock set holds.
 
-  CC_LockModeEnum strongest_held_;
-  // The mode of the strongest lock held
-
-  ACE_Thread_Semaphore semaphore_;
+    ACE_Thread_Semaphore semaphore_;
+  //  ACE_Token semaphore_;
   // This is the semaphore for the lock set. The semaphore is used to
   // queue requests for locks in modes stronger than currently
   // possible to grant.
 
-  int waiting_calls_;
-  // The number of calls waiting on the saemaphore
-
   CosConcurrencyControl::LockSet_ptr related_lockset_;
   // If this lock set is related to another lock set, this is the
   // pointer to the related lock set. This is a really simple
@@ -134,7 +156,11 @@ private:
   // Uses the internal enumeration as indices.
 
   //  ACE_Thread_Mutex mlock_;
-  // Mutex to ensure that race conditions does not occur
+  ACE_Thread_Mutex *mlock_;
+  // Lock to ensure that race conditions does not occur
+
+  ACE_Unbounded_Queue <CC_LockModeEnum> lock_queue_;
+  // Queue to hold the requested locks not yet granted.
 };
 
 #endif /* _CC_LOCKSET_H */