ChangeLogTag:Wed Sep 27 08:23:58 2000 Carlos O'Ryan <coryan@uci.edu>

2 files changed, 633 insertions, 12 deletions

diff --git a/tests/RMCast/Makefile b/tests/RMCast/Makefile
index 38dbce49c4d..c1949350a25 100644
--- a/tests/RMCast/Makefile
+++ b/tests/RMCast/Makefile
@@ -11,6 +11,7 @@
 BIN =   RMCast_Fragment_Test \
 	RMCast_Reassembly_Test \
 	RMCast_UDP_Best_Effort_Test \
+	RMCast_Membership_Test
 
 PSRC=$(addsuffix .cpp,$(BIN))
 LDLIBS = -lACE_RMCast
@@ -71,16 +72,16 @@ endif
  $(ACE_ROOT)/ace/Synch.h \
  $(ACE_ROOT)/ace/ACE.h \
  $(ACE_ROOT)/ace/ACE.i \
- $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \
- $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \
- $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \
- $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \
  $(ACE_ROOT)/ace/Synch.i \
  $(ACE_ROOT)/ace/Synch_T.h \
  $(ACE_ROOT)/ace/Event_Handler.h \
  $(ACE_ROOT)/ace/Event_Handler.i \
  $(ACE_ROOT)/ace/Synch_T.i \
  $(ACE_ROOT)/ace/Thread.h \
+ $(ACE_ROOT)/ace/Thread_Adapter.h \
+ $(ACE_ROOT)/ace/Base_Thread_Adapter.h \
+ $(ACE_ROOT)/ace/Base_Thread_Adapter.inl \
+ $(ACE_ROOT)/ace/Thread_Adapter.inl \
  $(ACE_ROOT)/ace/Thread.i \
  $(ACE_ROOT)/ace/Atomic_Op.i \
  $(ACE_ROOT)/ace/Synch_T.cpp \
@@ -124,6 +125,10 @@ endif
  $(ACE_ROOT)/ace/Signal.i \
  $(ACE_ROOT)/ace/Mem_Map.h \
  $(ACE_ROOT)/ace/Mem_Map.i \
+ $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \
+ $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \
+ $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \
+ $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \
  $(ACE_ROOT)/ace/Memory_Pool.i \
  $(ACE_ROOT)/ace/Thread_Manager.i \
  $(ACE_ROOT)/ace/Task.i \
@@ -172,6 +177,8 @@ endif
  $(ACE_ROOT)/ace/Service_Types.i \
  $(ACE_ROOT)/ace/Service_Repository.i \
  $(ACE_ROOT)/ace/WFMO_Reactor.h \
+ $(ACE_ROOT)/ace/Process_Mutex.h \
+ $(ACE_ROOT)/ace/Process_Mutex.inl \
  $(ACE_ROOT)/ace/WFMO_Reactor.i \
  $(ACE_ROOT)/ace/Strategies.i \
  $(ACE_ROOT)/ace/Message_Queue.i \
@@ -216,16 +223,16 @@ endif
  $(ACE_ROOT)/ace/Synch.h \
  $(ACE_ROOT)/ace/ACE.h \
  $(ACE_ROOT)/ace/ACE.i \
- $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \
- $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \
- $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \
- $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \
  $(ACE_ROOT)/ace/Synch.i \
  $(ACE_ROOT)/ace/Synch_T.h \
  $(ACE_ROOT)/ace/Event_Handler.h \
  $(ACE_ROOT)/ace/Event_Handler.i \
  $(ACE_ROOT)/ace/Synch_T.i \
  $(ACE_ROOT)/ace/Thread.h \
+ $(ACE_ROOT)/ace/Thread_Adapter.h \
+ $(ACE_ROOT)/ace/Base_Thread_Adapter.h \
+ $(ACE_ROOT)/ace/Base_Thread_Adapter.inl \
+ $(ACE_ROOT)/ace/Thread_Adapter.inl \
  $(ACE_ROOT)/ace/Thread.i \
  $(ACE_ROOT)/ace/Atomic_Op.i \
  $(ACE_ROOT)/ace/Synch_T.cpp \
@@ -269,6 +276,10 @@ endif
  $(ACE_ROOT)/ace/Signal.i \
  $(ACE_ROOT)/ace/Mem_Map.h \
  $(ACE_ROOT)/ace/Mem_Map.i \
+ $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \
+ $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \
+ $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \
+ $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \
  $(ACE_ROOT)/ace/Memory_Pool.i \
  $(ACE_ROOT)/ace/Thread_Manager.i \
  $(ACE_ROOT)/ace/Task.i \
@@ -317,6 +328,8 @@ endif
  $(ACE_ROOT)/ace/Service_Types.i \
  $(ACE_ROOT)/ace/Service_Repository.i \
  $(ACE_ROOT)/ace/WFMO_Reactor.h \
+ $(ACE_ROOT)/ace/Process_Mutex.h \
+ $(ACE_ROOT)/ace/Process_Mutex.inl \
  $(ACE_ROOT)/ace/WFMO_Reactor.i \
  $(ACE_ROOT)/ace/Strategies.i \
  $(ACE_ROOT)/ace/Message_Queue.i \
@@ -362,16 +375,16 @@ endif
  $(ACE_ROOT)/ace/Synch.h \
  $(ACE_ROOT)/ace/ACE.h \
  $(ACE_ROOT)/ace/ACE.i \
- $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \
- $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \
- $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \
- $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \
  $(ACE_ROOT)/ace/Synch.i \
  $(ACE_ROOT)/ace/Synch_T.h \
  $(ACE_ROOT)/ace/Event_Handler.h \
  $(ACE_ROOT)/ace/Event_Handler.i \
  $(ACE_ROOT)/ace/Synch_T.i \
  $(ACE_ROOT)/ace/Thread.h \
+ $(ACE_ROOT)/ace/Thread_Adapter.h \
+ $(ACE_ROOT)/ace/Base_Thread_Adapter.h \
+ $(ACE_ROOT)/ace/Base_Thread_Adapter.inl \
+ $(ACE_ROOT)/ace/Thread_Adapter.inl \
  $(ACE_ROOT)/ace/Thread.i \
  $(ACE_ROOT)/ace/Atomic_Op.i \
  $(ACE_ROOT)/ace/Synch_T.cpp \
@@ -443,6 +456,10 @@ endif
  $(ACE_ROOT)/ace/Memory_Pool.h \
  $(ACE_ROOT)/ace/Mem_Map.h \
  $(ACE_ROOT)/ace/Mem_Map.i \
+ $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \
+ $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \
+ $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \
+ $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \
  $(ACE_ROOT)/ace/Memory_Pool.i \
  $(ACE_ROOT)/ace/Signal.i \
  $(ACE_ROOT)/ace/SString.h \
@@ -491,6 +508,161 @@ endif
  $(ACE_ROOT)/ace/Service_Types.i \
  $(ACE_ROOT)/ace/Service_Repository.i \
  $(ACE_ROOT)/ace/WFMO_Reactor.h \
+ $(ACE_ROOT)/ace/Process_Mutex.h \
+ $(ACE_ROOT)/ace/Process_Mutex.inl \
+ $(ACE_ROOT)/ace/WFMO_Reactor.i \
+ $(ACE_ROOT)/ace/Strategies.i \
+ $(ACE_ROOT)/ace/Message_Queue.i \
+ $(ACE_ROOT)/ace/Task_T.i \
+ $(ACE_ROOT)/ace/Task_T.cpp \
+ $(ACE_ROOT)/ace/Module.h \
+ $(ACE_ROOT)/ace/Module.i \
+ $(ACE_ROOT)/ace/Module.cpp \
+ $(ACE_ROOT)/ace/Stream_Modules.h \
+ $(ACE_ROOT)/ace/Stream_Modules.cpp
+
+.obj/RMCast_Membership_Test.o .obj/RMCast_Membership_Test.so .shobj/RMCast_Membership_Test.o .shobj/RMCast_Membership_Test.so: RMCast_Membership_Test.cpp ../test_config.h \
+ $(ACE_ROOT)/ace/pre.h \
+ $(ACE_ROOT)/ace/post.h \
+ $(ACE_ROOT)/ace/ACE_export.h \
+ $(ACE_ROOT)/ace/svc_export.h \
+ $(ACE_ROOT)/ace/ace_wchar.h \
+ $(ACE_ROOT)/ace/OS.h \
+ $(ACE_ROOT)/ace/OS_Dirent.h \
+ $(ACE_ROOT)/ace/OS_Export.h \
+ $(ACE_ROOT)/ace/OS_Dirent.inl \
+ $(ACE_ROOT)/ace/OS_String.h \
+ $(ACE_ROOT)/ace/OS_String.inl \
+ $(ACE_ROOT)/ace/OS_Memory.h \
+ $(ACE_ROOT)/ace/OS_Memory.inl \
+ $(ACE_ROOT)/ace/OS_TLI.h \
+ $(ACE_ROOT)/ace/OS_TLI.inl \
+ $(ACE_ROOT)/ace/Min_Max.h \
+ $(ACE_ROOT)/ace/streams.h \
+ $(ACE_ROOT)/ace/Basic_Types.h \
+ $(ACE_ROOT)/ace/Basic_Types.i \
+ $(ACE_ROOT)/ace/Trace.h \
+ $(ACE_ROOT)/ace/OS.i \
+ $(ACE_ROOT)/ace/Singleton.h \
+ $(ACE_ROOT)/ace/Synch.h \
+ $(ACE_ROOT)/ace/ACE.h \
+ $(ACE_ROOT)/ace/ACE.i \
+ $(ACE_ROOT)/ace/Synch.i \
+ $(ACE_ROOT)/ace/Synch_T.h \
+ $(ACE_ROOT)/ace/Event_Handler.h \
+ $(ACE_ROOT)/ace/Event_Handler.i \
+ $(ACE_ROOT)/ace/Synch_T.i \
+ $(ACE_ROOT)/ace/Thread.h \
+ $(ACE_ROOT)/ace/Thread_Adapter.h \
+ $(ACE_ROOT)/ace/Base_Thread_Adapter.h \
+ $(ACE_ROOT)/ace/Base_Thread_Adapter.inl \
+ $(ACE_ROOT)/ace/Thread_Adapter.inl \
+ $(ACE_ROOT)/ace/Thread.i \
+ $(ACE_ROOT)/ace/Atomic_Op.i \
+ $(ACE_ROOT)/ace/Synch_T.cpp \
+ $(ACE_ROOT)/ace/Log_Msg.h \
+ $(ACE_ROOT)/ace/Log_Record.h \
+ $(ACE_ROOT)/ace/Log_Priority.h \
+ $(ACE_ROOT)/ace/Log_Record.i \
+ $(ACE_ROOT)/ace/Singleton.i \
+ $(ACE_ROOT)/ace/Singleton.cpp \
+ $(ACE_ROOT)/ace/Object_Manager.h \
+ $(ACE_ROOT)/ace/Object_Manager.i \
+ $(ACE_ROOT)/ace/Managed_Object.h \
+ $(ACE_ROOT)/ace/Managed_Object.i \
+ $(ACE_ROOT)/ace/Managed_Object.cpp \
+ $(ACE_ROOT)/ace/RMCast/RMCast_Proxy.h \
+ $(ACE_ROOT)/ace/RMCast/RMCast_Module.h \
+ $(ACE_ROOT)/ace/RMCast/RMCast.h \
+ $(ACE_ROOT)/ace/RMCast/RMCast_Export.h \
+ $(ACE_ROOT)/ace/RMCast/RMCast.i \
+ $(ACE_ROOT)/ace/RMCast/RMCast_Module.i \
+ $(ACE_ROOT)/ace/RMCast/RMCast_Proxy.i \
+ $(ACE_ROOT)/ace/RMCast/RMCast_Membership.h \
+ $(ACE_ROOT)/ace/Containers.h \
+ $(ACE_ROOT)/ace/Malloc_Base.h \
+ $(ACE_ROOT)/ace/Containers.i \
+ $(ACE_ROOT)/ace/Containers_T.h \
+ $(ACE_ROOT)/ace/Containers_T.i \
+ $(ACE_ROOT)/ace/Containers_T.cpp \
+ $(ACE_ROOT)/ace/Malloc.h \
+ $(ACE_ROOT)/ace/Based_Pointer_T.h \
+ $(ACE_ROOT)/ace/Based_Pointer_T.i \
+ $(ACE_ROOT)/ace/Based_Pointer_T.cpp \
+ $(ACE_ROOT)/ace/Based_Pointer_Repository.h \
+ $(ACE_ROOT)/ace/Malloc.i \
+ $(ACE_ROOT)/ace/Malloc_T.h \
+ $(ACE_ROOT)/ace/Free_List.h \
+ $(ACE_ROOT)/ace/Free_List.i \
+ $(ACE_ROOT)/ace/Free_List.cpp \
+ $(ACE_ROOT)/ace/Malloc_T.i \
+ $(ACE_ROOT)/ace/Malloc_T.cpp \
+ $(ACE_ROOT)/ace/Memory_Pool.h \
+ $(ACE_ROOT)/ace/Signal.h \
+ $(ACE_ROOT)/ace/Signal.i \
+ $(ACE_ROOT)/ace/Mem_Map.h \
+ $(ACE_ROOT)/ace/Mem_Map.i \
+ $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \
+ $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \
+ $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \
+ $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \
+ $(ACE_ROOT)/ace/Memory_Pool.i \
+ $(ACE_ROOT)/ace/RMCast/RMCast_Membership.i \
+ $(ACE_ROOT)/ace/Task.h \
+ $(ACE_ROOT)/ace/Service_Object.h \
+ $(ACE_ROOT)/ace/Shared_Object.h \
+ $(ACE_ROOT)/ace/Shared_Object.i \
+ $(ACE_ROOT)/ace/Service_Object.i \
+ $(ACE_ROOT)/ace/Thread_Manager.h \
+ $(ACE_ROOT)/ace/Thread_Manager.i \
+ $(ACE_ROOT)/ace/Task.i \
+ $(ACE_ROOT)/ace/Task_T.h \
+ $(ACE_ROOT)/ace/Message_Queue.h \
+ $(ACE_ROOT)/ace/Message_Block.h \
+ $(ACE_ROOT)/ace/Message_Block.i \
+ $(ACE_ROOT)/ace/Message_Block_T.h \
+ $(ACE_ROOT)/ace/Message_Block_T.i \
+ $(ACE_ROOT)/ace/Message_Block_T.cpp \
+ $(ACE_ROOT)/ace/IO_Cntl_Msg.h \
+ $(ACE_ROOT)/ace/Message_Queue_T.h \
+ $(ACE_ROOT)/ace/Message_Queue_T.i \
+ $(ACE_ROOT)/ace/Message_Queue_T.cpp \
+ $(ACE_ROOT)/ace/Strategies.h \
+ $(ACE_ROOT)/ace/Strategies_T.h \
+ $(ACE_ROOT)/ace/Service_Config.h \
+ $(ACE_ROOT)/ace/SString.h \
+ $(ACE_ROOT)/ace/SString.i \
+ $(ACE_ROOT)/ace/Service_Config.i \
+ $(ACE_ROOT)/ace/Reactor.h \
+ $(ACE_ROOT)/ace/Handle_Set.h \
+ $(ACE_ROOT)/ace/Handle_Set.i \
+ $(ACE_ROOT)/ace/Timer_Queue.h \
+ $(ACE_ROOT)/ace/Timer_Queue_T.h \
+ $(ACE_ROOT)/ace/Timer_Queue_T.i \
+ $(ACE_ROOT)/ace/Timer_Queue_T.cpp \
+ $(ACE_ROOT)/ace/Reactor.i \
+ $(ACE_ROOT)/ace/Reactor_Impl.h \
+ $(ACE_ROOT)/ace/Svc_Conf_Tokens.h \
+ $(ACE_ROOT)/ace/Synch_Options.h \
+ $(ACE_ROOT)/ace/Synch_Options.i \
+ $(ACE_ROOT)/ace/Hash_Map_Manager.h \
+ $(ACE_ROOT)/ace/Functor.h \
+ $(ACE_ROOT)/ace/Functor.i \
+ $(ACE_ROOT)/ace/Functor_T.h \
+ $(ACE_ROOT)/ace/Functor_T.i \
+ $(ACE_ROOT)/ace/Functor_T.cpp \
+ $(ACE_ROOT)/ace/Hash_Map_Manager_T.h \
+ $(ACE_ROOT)/ace/Hash_Map_Manager_T.i \
+ $(ACE_ROOT)/ace/Hash_Map_Manager_T.cpp \
+ $(ACE_ROOT)/ace/Strategies_T.i \
+ $(ACE_ROOT)/ace/Strategies_T.cpp \
+ $(ACE_ROOT)/ace/Service_Repository.h \
+ $(ACE_ROOT)/ace/Service_Types.h \
+ $(ACE_ROOT)/ace/Service_Types.i \
+ $(ACE_ROOT)/ace/Service_Repository.i \
+ $(ACE_ROOT)/ace/WFMO_Reactor.h \
+ $(ACE_ROOT)/ace/Process_Mutex.h \
+ $(ACE_ROOT)/ace/Process_Mutex.inl \
  $(ACE_ROOT)/ace/WFMO_Reactor.i \
  $(ACE_ROOT)/ace/Strategies.i \
  $(ACE_ROOT)/ace/Message_Queue.i \
diff --git a/tests/RMCast/RMCast_Membership_Test.cpp b/tests/RMCast/RMCast_Membership_Test.cpp
new file mode 100644
index 00000000000..54b5f9a5aa8
--- /dev/null
+++ b/tests/RMCast/RMCast_Membership_Test.cpp
@@ -0,0 +1,449 @@
+// $Id$
+
+// ============================================================================
+//
+// = DESCRIPTION
+//    Unit test for the UDP sending module of the RMCast library.
+//
+// = AUTHORS
+//    Carlos O'Ryan <coryan@uci.edu>
+//
+// ============================================================================
+
+#include "test_config.h"
+#include "ace/RMCast/RMCast_Proxy.h"
+#include "ace/RMCast/RMCast_Membership.h"
+
+#include "ace/Task.h"
+
+ACE_RCSID(tests, RMCast_Membership_Test, "$Id$")
+
+const size_t message_size = 8 * 1024;
+const int total_message_count = 40;
+
+// ****************************************************************
+
+//! Simple proxy for the ACE_RMCast_Membership test harness
+/*!
+ * Implement a simple version of the ACE_RMCast_Proxy class used in
+ * the ACE_RMCast_Membership test harness.
+ */
+class Test_Proxy : public ACE_RMCast_Proxy
+{
+public:
+  Test_Proxy (void);
+
+  //! Get the flag to remember if this proxy has joined the group or
+  //! not.
+  int joined (void) const
+  {
+    return this->joined_;
+  }
+  //! Set the flag to remember if this proxy has joined the group or
+  //! not. 
+  void joined (int j)
+  {
+    this->joined_ = j;
+  }
+
+  //@{
+  //! All the reply_ methods just return 0, there is no real remote
+  //! peer, this is just a test harness
+  virtual int reply_data (ACE_RMCast::Data &)
+  {
+    return 0;
+  }
+  virtual int reply_poll (ACE_RMCast::Poll &)
+  {
+    return 0;
+  }
+  virtual int reply_ack_join (ACE_RMCast::Ack_Join &)
+  {
+    return 0;
+  }
+  virtual int reply_ack_leave (ACE_RMCast::Ack_Leave &)
+  {
+    return 0;
+  }
+  virtual int reply_ack (ACE_RMCast::Ack &)
+  {
+    return 0;
+  }
+  virtual int reply_join (ACE_RMCast::Join &)
+  {
+    return 0;
+  }
+  virtual int reply_leave (ACE_RMCast::Leave &)
+  {
+    return 0;
+  }
+  //@}
+
+private:
+  //! Remember if we joined the group already.
+  int joined_;
+};
+
+// ****************************************************************
+
+//! The number of proxies used in the test
+/*!
+ * Not all member will be present in the group at the same time. But
+ * this variable controls the maximum number
+ */
+const size_t nproxy = 16;
+
+//! A simple module to receive the messages from ACE_RMCast_Membership
+/*!
+ * The ACE_RMCast_Membership layer pushes messages to its next module
+ * when all the members have acked a message, when a new member joins,
+ * when a member leaves, etc.
+ * This class will verify that the messages are exactly what we
+ * expect.
+ */
+class Tester : public ACE_RMCast_Module
+{
+public:
+  Tester (void);
+
+  //! Run the test for \iterations times
+  void run (int iterations);
+
+  virtual int join (ACE_RMCast::Join &join);
+  virtual int leave (ACE_RMCast::Leave &leave);
+  virtual int ack (ACE_RMCast::Ack &ack);
+
+private:
+  //! Add a few proxies to the group
+  void join_random (void);
+
+  //! Remove a few proxies from the group
+  void leave_random (void);
+
+  //! Generate a few ack messages from all the proxies currently in
+  //! the group
+  void generate_acks (int iterations);
+
+private:
+  //! The array of proxies
+  Test_Proxy proxy_[nproxy];
+
+  //! The Membership layer
+  ACE_RMCast_Membership membership_;
+
+  //! Synchronize internal data structures
+  ACE_SYNCH_MUTEX lock_;
+
+  //! The test is randomized to get better coverage.  This is the seed
+  //! variable for the test
+  ACE_RANDR_TYPE seed_;
+};
+
+// ****************************************************************
+
+//! An Adapter to run Tester::run the test is a separate thread
+class Task : public ACE_Task_Base
+{
+public:
+  Task (Tester *tester);
+
+  // = Read the documentation in "ace/Task.h"
+  int svc (void);
+
+private:
+  //! The tester object.
+  Tester *tester_;
+};
+
+// ****************************************************************
+
+int
+main (int, ACE_TCHAR *[])
+{
+  ACE_START_TEST (ACE_TEXT ("RMCast_Membership_Test"));
+
+  ACE_DEBUG ((LM_DEBUG,
+              ACE_TEXT ("This is ACE Version %u.%u.%u\n\n"),
+              ACE::major_version(),
+              ACE::minor_version(),
+              ACE::beta_version()));
+
+  {
+    ACE_DEBUG ((LM_DEBUG, "Running single threaded test\n"));
+    //! Run the test in single threaded mode
+    Tester tester;
+    tester.run (100);
+  }
+  {
+    ACE_DEBUG ((LM_DEBUG, "Running multi threaded test\n"));
+    //! Run the test in multi-threaded mode
+    Tester tester;
+    Task task (&tester);
+    if (task.activate (THR_NEW_LWP|THR_JOINABLE, 4) == -1)
+      ACE_ERROR_RETURN ((LM_ERROR, "Cannot activate the threads\n"), 1);
+    ACE_Thread_Manager::instance ()->wait ();
+  }
+
+  ACE_END_TEST;
+  return 0;
+}
+
+// ****************************************************************
+
+Test_Proxy::Test_Proxy (void)
+  :  joined_ (0)
+{
+}
+
+// ****************************************************************
+
+Tester::Tester (void)
+  : seed_ (ACE_OS::gethrtime ())
+{
+  // Initialize the stack...
+  this->membership_.next (this);
+  for (size_t i = 0; i != nproxy; ++i)
+    this->proxy_[i].next (&this->membership_);
+}
+
+void
+Tester::run (int iterations)
+{
+  for (int i = 0; i < iterations; ++i)
+    {
+      // Connect a few....
+      this->join_random ();
+
+      // Push acks....
+      this->generate_acks (iterations);
+
+      // Disconnect a few
+      this->leave_random ();
+
+      // Push acks...
+      this->generate_acks (iterations / 10);
+    }
+}
+
+int
+Tester::join (ACE_RMCast::Join &join)
+{
+  if (join.source == 0)
+    {
+      ACE_ERROR_RETURN ((LM_ERROR,
+                         "Invalid join message in Tester\n"),
+                        -1);
+    }
+  for (size_t i = 0; i != nproxy; ++i)
+    {
+      if (&this->proxy_[i] != join.source)
+        continue;
+      if (this->proxy_[i].joined () != 1)
+        ACE_ERROR_RETURN ((LM_ERROR,
+                           "Invalid state for proxy %d "
+                           "in Tester::join\n"),
+                          -1);
+      return 0;
+    }
+  // Not found
+  ACE_ERROR_RETURN ((LM_ERROR, "Unknown proxy in Tester::join\n"), -1);
+}
+
+int
+Tester::leave (ACE_RMCast::Leave &leave)
+{
+  if (leave.source == 0)
+    {
+      ACE_ERROR_RETURN ((LM_ERROR,
+                         "Invalid leave message in Tester\n"),
+                        -1);
+    }
+  for (size_t i = 0; i != nproxy; ++i)
+    {
+      if (&this->proxy_[i] != leave.source)
+        continue;
+      if (this->proxy_[i].joined () != 0)
+        ACE_ERROR_RETURN ((LM_ERROR,
+                           "Invalid state for proxy %d "
+                           "in Tester::leave\n"),
+                          -1);
+      return 0;
+    }
+  // Not found
+  ACE_ERROR_RETURN ((LM_ERROR, "Unknown proxy in Tester::leave\n"), -1);
+}
+
+int
+Tester::ack (ACE_RMCast::Ack &ack)
+{
+  // After the membership layer the source makes no sense....
+  if (ack.source == 0)
+    {
+      ACE_ERROR_RETURN ((LM_ERROR,
+                         "Invalid ack message in Tester\n"),
+                        -1);
+    }
+
+  //  ACE_DEBUG ((LM_DEBUG,
+  //              "Received ack in Tester %d,%d\n",
+  //              ack.highest_in_sequence,
+  //              ack.highest_received));
+
+  // Assume the lock is held, verify that the ack message satisfy the
+  // invariants...
+  ACE_UINT32 highest_in_sequence;
+  ACE_UINT32 highest_received;
+  int set = 0;
+  for (size_t i = 0; i != nproxy; ++i)
+    {
+      if (!this->proxy_[i].joined ())
+        continue;
+      if (!set)
+        {
+          set = 1;
+          highest_in_sequence = this->proxy_[i].highest_in_sequence ();
+          highest_received = this->proxy_[i].highest_received ();
+        }
+      else
+        {
+          if (highest_in_sequence >
+              this->proxy_[i].highest_in_sequence ())
+            {
+              highest_in_sequence =
+                this->proxy_[i].highest_in_sequence ();
+            }
+          if (highest_received < 
+              this->proxy_[i].highest_received ())
+            {
+              highest_received =
+                this->proxy_[i].highest_received ();
+            }
+        }
+    }
+  // No local proxy just return...
+  if (!set)
+    return 0;
+
+  // Check the invariants
+  if (ack.highest_in_sequence != highest_in_sequence)
+    {
+      ACE_ERROR_RETURN ((LM_ERROR,
+                         "Invalid highest_in_sequence in Ack\n"),
+                        -1);
+    }
+  if (ack.highest_received != highest_received)
+    {
+      ACE_ERROR_RETURN ((LM_ERROR,
+                         "Invalid highest_received in Ack\n"),
+                        -1);
+    }
+  return 0;
+}
+
+void
+Tester::join_random (void)
+{
+  for (size_t i = 0; i != nproxy; ++i)
+    {
+      ACE_GUARD (ACE_SYNCH_MUTEX, ace_mon, this->lock_);
+      int r = ACE_OS::rand_r (this->seed_) % 100;
+      if (this->proxy_[i].joined () == 0 && r > 25)
+        {
+          this->proxy_[i].joined (1);
+
+          ACE_RMCast::Join join;
+          join.source = &this->proxy_[i];
+          //          ACE_DEBUG ((LM_DEBUG,
+          //                      "Sending join mesage for proxy %d\n",
+          //                      i));
+          this->proxy_[i].join (join);
+        }
+    }
+}
+
+void
+Tester::leave_random (void)
+{
+  for (size_t i = 0; i != nproxy; ++i)
+    {
+      ACE_GUARD (ACE_SYNCH_MUTEX, ace_mon, this->lock_);
+      int r = ACE_OS::rand_r (this->seed_) % 100;
+      if (this->proxy_[i].joined () == 1 && r > 75)
+        {
+          this->proxy_[i].joined (0);
+
+          ACE_RMCast::Leave leave;
+          leave.source = &this->proxy_[i];
+          //          ACE_DEBUG ((LM_DEBUG,
+          //                      "Sending leave mesage for proxy %d\n",
+          //                      i));
+          this->proxy_[i].leave (leave);
+        }
+    }
+}
+
+void
+Tester::generate_acks (int iterations)
+{
+  int n = 0;
+  for (size_t i = 0; n < iterations && i != nproxy; ++i, ++n)
+    {
+      ACE_GUARD (ACE_SYNCH_MUTEX, ace_mon, this->lock_);
+      int r = ACE_OS::rand_r (this->seed_) % 10;
+      if (this->proxy_[i].joined () == 0)
+        continue;
+
+      ACE_RMCast::Ack ack;
+      ack.source = &this->proxy_[i];
+      ack.highest_in_sequence =
+        this->proxy_[i].highest_in_sequence ();
+      ack.highest_received =
+        this->proxy_[i].highest_received ();
+
+      // we randomly perform one of 3 actions, with different
+      // probabilities
+      switch (r)
+        {
+        case 0:
+          // Ack the same data that we already have:
+          break;
+        case 1:
+        case 2:
+        case 3:
+        case 4:
+          // Simulate out of sequence messages...
+          ack.highest_received++;
+          break;
+        default:
+          if (ack.highest_received > ack.highest_in_sequence)
+            ack.highest_in_sequence++;
+          break;
+        }
+      //      ACE_DEBUG ((LM_DEBUG,
+      //                  "Sending ack message (%d,%d) through proxy %d\n",
+      //                  ack.highest_in_sequence,
+      //                  ack.highest_received,
+      //                  i));
+      int result = this->proxy_[i].ack (ack);
+      if (result != 0)
+        {
+          ACE_ERROR ((LM_ERROR,
+                      "Returned %d in proxy %d\n",
+                      result, i));
+        }
+    }
+}
+
+// ****************************************************************
+
+Task::Task (Tester *tester)
+  :  tester_ (tester)
+{
+}
+
+int
+Task::svc (void)
+{
+  this->tester_->run (100);
+  return 0;
+}