Mon Mar 26 23:29:47 2001 Krishnakumar B <kitty@cs.wustl.edu>

	* Leader_Follower.{i,h,cpp}
	* Wait_On_Leader_Follower.cpp

	Implemented the Leader_Follower component of the ORD to use a
	stack instead of a heap, using the same strategy as in ACE_Token.
	This fixes Bug 721.

4 files changed, 148 insertions, 39 deletions

diff --git a/TAO/tao/Leader_Follower.cpp b/TAO/tao/Leader_Follower.cpp
index 54b04613a25..13d9e88fbef 100644
--- a/TAO/tao/Leader_Follower.cpp
+++ b/TAO/tao/Leader_Follower.cpp
@@ -18,17 +18,89 @@ TAO_Leader_Follower::~TAO_Leader_Follower (void)
   this->reactor_ = 0;
 }
 
+TAO_Leader_Follower::TAO_Follower_Node::TAO_Follower_Node (TAO_SYNCH_CONDITION* follower_ptr)
+  : follower_ (follower_ptr),
+    next_ (0)
+{
+
+}
+
+TAO_Leader_Follower::TAO_Follower_Queue::TAO_Follower_Queue (void)
+  : head_ (0),
+    tail_ (0)
+{
+
+}
+
+int
+TAO_Leader_Follower::TAO_Follower_Queue::insert (TAO_Follower_Node* node)
+{
+  TAO_Follower_Node* curr = 0;
+
+  // Check if there is already an existing entry
+  for (curr = this->head_;
+       curr != 0 && curr != node;
+       curr = curr->next_)
+    ;
+  // An entry exists already in the queue
+  if (curr == node)
+    return 1;
+  else
+    {
+      // Add the node to the tail and modify the pointers
+      TAO_Follower_Node* temp = this->tail_;
+      this->tail_ = node;
+      if (this->head_ == 0)
+        this->head_ = this->tail_;
+      else
+        temp->next_ = this->tail_;
+    }
+  return 0;
+}
+
+int
+TAO_Leader_Follower::TAO_Follower_Queue::remove (TAO_Follower_Node* node)
+{
+  TAO_Follower_Node* prev = 0;
+  TAO_Follower_Node* curr = 0;
+
+  // No followers in queue, return
+  if (this->head_ == 0)
+    return -1;
+
+  for (curr = this->head_;
+       curr != 0 && curr != node;
+       curr = curr->next_)
+    {
+      prev = curr;
+    }
+
+  // Entry not found in the queue
+  if (curr == 0)
+    return -1;
+  // Entry found at the head of the queue
+  else if (prev == 0)
+    this->head_ = this->head_->next_;
+  else
+    prev->next_ = curr->next_;
+  // Entry at the tail
+  if (curr->next_ == 0)
+    this->tail_ = prev;
+
+  return 0;
+}
+
+
 TAO_SYNCH_CONDITION*
 TAO_Leader_Follower::get_next_follower (void)
 {
-  ACE_Unbounded_Set_Iterator<TAO_SYNCH_CONDITION *> iterator (
-    this->follower_set_);
-
-  if (iterator.first () == 0)
-    // means set is empty
+  // If the queue is empty return
+  if (this->follower_set_.is_empty())
     return 0;
 
-  TAO_SYNCH_CONDITION *cond = *iterator;
+  TAO_Follower_Node* next_follower = this->follower_set_.head_;
+
+  TAO_SYNCH_CONDITION *cond = next_follower->follower_;
 
 #if defined (TAO_DEBUG_LEADER_FOLLOWER)
   ACE_DEBUG ((LM_DEBUG,
@@ -43,7 +115,7 @@ TAO_Leader_Follower::get_next_follower (void)
   // The follower may not be there if the reply is received while the
   // consumer is not yet waiting for it (i.e. it send the request but
   // has not blocked to receive the reply yet)
-  (void) this->remove_follower (cond); // Ignore errors..
+  (void) this->remove_follower (next_follower); // Ignore errors..
 
   return cond;
 }
@@ -153,17 +225,3 @@ TAO_Leader_Follower::reset_client_thread (void)
       this->orb_core_->reactor ()->end_reactor_event_loop ();
     }
 }
-
-#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
-
-template class ACE_Node<TAO_SYNCH_CONDITION*>;
-template class ACE_Unbounded_Set<TAO_SYNCH_CONDITION*>;
-template class ACE_Unbounded_Set_Iterator<TAO_SYNCH_CONDITION*>;
-
-#elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)
-
-#pragma instantiate ACE_Node<TAO_SYNCH_CONDITION*>
-#pragma instantiate ACE_Unbounded_Set<TAO_SYNCH_CONDITION*>
-#pragma instantiate ACE_Unbounded_Set_Iterator<TAO_SYNCH_CONDITION*>
-
-#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */
diff --git a/TAO/tao/Leader_Follower.h b/TAO/tao/Leader_Follower.h
index 570b4802b17..7d17e7d447c 100644
--- a/TAO/tao/Leader_Follower.h
+++ b/TAO/tao/Leader_Follower.h
@@ -20,8 +20,6 @@
 # pragma once
 #endif /* ACE_LACKS_PRAGMA_ONCE */
 
-#include "ace/Unbounded_Queue.h"
-
 class TAO_Export TAO_Leader_Follower
 {
 public:
@@ -74,13 +72,25 @@ public:
    */
   int elect_new_leader (void);
 
+  /// Node structure for the queue of followers
+  struct TAO_Follower_Node
+  {
+    /// Constructor
+    TAO_Follower_Node (TAO_SYNCH_CONDITION* follower_ptr);
+
+    /// Follower
+    TAO_SYNCH_CONDITION *follower_;
+
+    /// Pointer to the next follower
+    TAO_Follower_Node  *next_;
+  };
   /**
    * adds the a follower to the set of followers in the leader-
    * follower model
    * returns 0 on success, -1 on failure and 1 if the element is
    * already there.
    */
-  int add_follower (TAO_SYNCH_CONDITION *follower_ptr);
+  int add_follower (TAO_Follower_Node *follower_ptr);
 
   /// checks for the availablity of a follower
   /// returns 1 on available, 0 else
@@ -88,7 +98,7 @@ public:
 
   /// removes a follower from the leader-follower set
   /// returns 0 on success, -1 on failure
-  int remove_follower (TAO_SYNCH_CONDITION *follower_ptr);
+  int remove_follower (TAO_Follower_Node *follower_ptr);
 
   /// returns randomly a follower from the leader-follower set
   /// returns follower on success, else 0
@@ -128,8 +138,32 @@ private:
   /// do protect the access to the following three members
   ACE_Reverse_Lock<TAO_SYNCH_MUTEX> reverse_lock_;
 
-  ACE_Unbounded_Set<TAO_SYNCH_CONDITION *> follower_set_;
-  // keep a set of followers around (protected)
+  /// Queue to store the followers.
+  struct TAO_Follower_Queue
+  {
+    /// Constructor
+    TAO_Follower_Queue (void);
+
+    /// Checks if the queue is empty.
+    int is_empty (void) const;
+
+    /// Removes a follower from the queue.
+    int remove (TAO_Follower_Node *);
+
+    /// Inserts a follower into the queue.
+    /// Returns 0 on success, -1 for failure, 1 if the element is already
+    /// present.
+    int insert (TAO_Follower_Node *);
+
+    /// Pointer to the head of the queue.
+    TAO_Follower_Node *head_;
+
+    /// Pointer to the tail of the queue.
+    TAO_Follower_Node *tail_;
+  };
+
+  /// Queue to keep the followers on the stack.
+  TAO_Follower_Queue follower_set_;
 
   /**
    * Count the number of active leaders.
diff --git a/TAO/tao/Leader_Follower.i b/TAO/tao/Leader_Follower.i
index 08a0a301058..db9324ae84d 100644
--- a/TAO/tao/Leader_Follower.i
+++ b/TAO/tao/Leader_Follower.i
@@ -156,15 +156,15 @@ TAO_Leader_Follower::is_client_leader_thread (void) const
 }
 
 ACE_INLINE int
-TAO_Leader_Follower::add_follower (TAO_SYNCH_CONDITION *follower_ptr)
+TAO_Leader_Follower::add_follower (TAO_Follower_Node *follower_node)
 {
-  return this->follower_set_.insert (follower_ptr);
+  return this->follower_set_.insert (follower_node);
 }
 
 ACE_INLINE int
-TAO_Leader_Follower::remove_follower (TAO_SYNCH_CONDITION *follower_ptr)
+TAO_Leader_Follower::remove_follower (TAO_Follower_Node *follower_node)
 {
-  return this->follower_set_.remove (follower_ptr);
+  return this->follower_set_.remove (follower_node);
 }
 
 ACE_INLINE ACE_Reverse_Lock<TAO_SYNCH_MUTEX> &
@@ -179,6 +179,12 @@ TAO_Leader_Follower::has_clients (void) const
   return this->clients_;
 }
 
+ACE_INLINE int
+TAO_Leader_Follower::TAO_Follower_Queue::is_empty (void) const
+{
+  return this->head_ == 0;
+}
+
 ACE_INLINE
 TAO_LF_Client_Thread_Helper::TAO_LF_Client_Thread_Helper (TAO_Leader_Follower &leader_follower)
   : leader_follower_ (leader_follower)
diff --git a/TAO/tao/Wait_On_Leader_Follower.cpp b/TAO/tao/Wait_On_Leader_Follower.cpp
index 1039158bfc2..4cda5f6a4c2 100644
--- a/TAO/tao/Wait_On_Leader_Follower.cpp
+++ b/TAO/tao/Wait_On_Leader_Follower.cpp
@@ -91,6 +91,9 @@ TAO_Wait_On_Leader_Follower::wait (ACE_Time_Value *max_wait_time,
                       this->transport_,
                       cond));
 
+        // Keep the entry on the stack
+        ACE_NESTED_CLASS (TAO_Leader_Follower, TAO_Follower_Node) node(cond);
+
         while (!reply_received &&
                leader_follower.leader_available ())
           {
@@ -118,7 +121,7 @@ TAO_Wait_On_Leader_Follower::wait (ACE_Time_Value *max_wait_time,
             // lost.
             //
 
-            (void) leader_follower.add_follower (cond);
+            (void) leader_follower.add_follower (&node);
 
             if (max_wait_time == 0)
               {
@@ -145,10 +148,10 @@ TAO_Wait_On_Leader_Follower::wait (ACE_Time_Value *max_wait_time,
                                   ACE_TEXT ("cond == 0 || cond->wait (tv) == -1\n"),
                                   this->transport_));
 
-                    if (leader_follower.remove_follower (cond) == -1)
+                    if (leader_follower.remove_follower (&node) == -1)
                       ACE_ERROR ((LM_ERROR,
                                   "TAO (%P|%t) TAO_Wait_On_Leader_Follower::wait - "
-                                  "remove_follower failed for <%x>\n", cond));
+                                  "remove_follower failed for <%x>\n", node.follower_));
 
                     return -1;
                   }
@@ -161,10 +164,10 @@ TAO_Wait_On_Leader_Follower::wait (ACE_Time_Value *max_wait_time,
         // Cannot remove the follower here, we *must* remove it when
         // we signal it so the same condition is not signalled for
         // both wake up as a follower and as the next leader.
-        if (leader_follower.remove_follower (cond) == -1)
+        if (leader_follower.remove_follower (&node) == -1)
           ACE_ERROR ((LM_ERROR,
                       "TAO (%P|%t) TAO_Wait_On_Leader_Follower::wait - "
-                      "remove_follower failed for <%x>\n", cond));
+                      "remove_follower failed for <%x>\n", node.follower));
 #endif /* 0 */
 
         if (TAO_debug_level >= 5)
@@ -274,7 +277,7 @@ TAO_Wait_On_Leader_Follower::wait (ACE_Time_Value *max_wait_time,
           result = -1;
           errno = ETIME;
         }
-      else if (reply_received == -1)        
+      else if (reply_received == -1)
         {
           // If the time did not expire yet, but we get a failure,
           // e.g. the connections closed, we should still return an error.
@@ -316,6 +319,10 @@ TAO_Wait_On_Leader_Follower::reply_dispatched (int &reply_received_flag,
 
   reply_received_flag = 1;
 
+  // The following works as the node is assumed to be on the stack
+  // till the thread is alive.
+  ACE_NESTED_CLASS (TAO_Leader_Follower, TAO_Follower_Node) node (condition);
+
   // We *must* remove it when we signal it so the same condition
   // is not signalled for both wake up as a follower and as the
   // next leader.
@@ -323,7 +330,7 @@ TAO_Wait_On_Leader_Follower::reply_dispatched (int &reply_received_flag,
   // the consumer is not yet waiting for it (i.e. it send the
   // request but has not blocked to receive the reply yet).
   // Ignore errors.
-  (void) leader_follower.remove_follower (condition);
+  (void) leader_follower.remove_follower (&node);
 
   if (condition->signal () == -1)
     return -1;
@@ -346,7 +353,11 @@ TAO_Wait_On_Leader_Follower::connection_closed (int &reply_received_flag,
 
   reply_received_flag = -1;
 
-  (void) leader_follower.remove_follower (condition);
+  // The following works as the node is assumed to be on the stack
+  // till the thread is alive.
+  ACE_NESTED_CLASS (TAO_Leader_Follower, TAO_Follower_Node) node(condition);
+
+  (void) leader_follower.remove_follower (&node);
 
   (void) condition->signal ();
 }