1 files changed, 153 insertions, 0 deletions
diff --git a/ACE/TAO/orbsvcs/orbsvcs/ESF/ESF_Delayed_Changes.h b/ACE/TAO/orbsvcs/orbsvcs/ESF/ESF_Delayed_Changes.h
new file mode 100644
index 00000000000..36742c93d80
--- /dev/null
+++ b/ACE/TAO/orbsvcs/orbsvcs/ESF/ESF_Delayed_Changes.h
@@ -0,0 +1,153 @@
+// -*- C++ -*-
+/**
+ *  @file   ESF_Delayed_Changes.h
+ *
+ *  $Id$
+ *
+ *  @author Carlos O'Ryan (coryan@cs.wustl.edu)
+ *
+ *  http://doc.ece.uci.edu/~coryan/EC/index.html
+ */
+
+#ifndef TAO_ESF_DELAYED_CHANGES_H
+#define TAO_ESF_DELAYED_CHANGES_H
+
+#include "orbsvcs/ESF/ESF_Proxy_Collection.h"
+
+#if !defined (ACE_LACKS_PRAGMA_ONCE)
+# pragma once
+#endif /* ACE_LACKS_PRAGMA_ONCE */
+
+#include "orbsvcs/ESF/ESF_Busy_Lock.h"
+
+#include "ace/Containers.h"
+
+ACE_BEGIN_VERSIONED_NAMESPACE_DECL
+class ACE_Command_Base;
+ACE_END_VERSIONED_NAMESPACE_DECL
+
+TAO_BEGIN_VERSIONED_NAMESPACE_DECL
+
+template<class Target,class Object> class TAO_ESF_Connected_Command;
+template<class Target,class Object> class TAO_ESF_Disconnected_Command;
+template<class Target,class Object> class TAO_ESF_Reconnected_Command;
+template<class Target> class TAO_ESF_Shutdown_Command;
+
+/**
+ * @class TAO_ESF_Delayed_Changes
+ *
+ * @brief TAO_ESF_Delayed_Operations
+ *
+ * This class implements the Delayed Operations protocol to solve
+ * the concurrency challenges outlined in the documentation of
+ * TAO_ESF_Proxy_Collection.
+ * In short the class delays changes by putting them on an
+ * "operation queue", the operations are stored as command objects
+ * in this queue and executed once the system is quiescent
+ * (i.e. no threads are iterating over the collection).
+ * The algorithm implemented so far is:
+ * - If a thread is using the set then it increases the busy
+ * count, this is done by calling the busy() method. Once the
+ * thread has stopped using the collection the idle() method is
+ * invoked and the busy count is decreased.
+ * A helper class (Busy_Lock) is used to hide this protocol
+ * behind the familiar GUARD idiom.
+ * - If the busy count reaches the busy_hwm then the thread must
+ * wait until the count reaches 0 again.
+ * This can be used to control the maximum concurrency in the
+ * EC, matching it (for example) with the number of
+ * processors. Setting the concurrency to a high value (say one
+ * million) allows for an arbitrary number of threads to execute
+ * concurrently.
+ * - If a modification is posted to the collection we need to
+ * execute it at some point.
+ * Just using the busy_hwm would not work, the HWM may not be
+ * reached ever, so another form of control is needed.
+ * Instead we use another counter, that keeps track of how many
+ * threads have used the set since the modification was
+ * posted. If this number of threads reaches max_write_delay then
+ * we don't allow any more threads to go in, eventually the
+ * thread count reaches 0 and we can proceed with the operations.
+ * - There is one aspect of concurrency that can be problematic: if
+ * thread pushes events as part of an upcall then the same thread
+ * could be counted twice, we need to keep track of the threads
+ * that are dispatching events and not increase (or decrease) the
+ * reference count when a thread iterates twice over the same
+ * set.
+ * This solves the major problems, but there are other issues to
+ * be addressed:
+ * + How do we ensure that the operations are eventually executed?
+ * + How do we simplify the execution of the locking protocol for
+ * clients of this class?
+ * + How do we minimize overhead for single threaded execution?
+ * + How do we minimize the overhead for the cases where the
+ * threads dispatching events don't post changes to the
+ * collection?
+ */
+template<class PROXY, class COLLECTION, class ITERATOR, ACE_SYNCH_DECL>
+class TAO_ESF_Delayed_Changes : public TAO_ESF_Proxy_Collection<PROXY>
+{
+public:
+  TAO_ESF_Delayed_Changes (void);
+  TAO_ESF_Delayed_Changes (const COLLECTION &collection);
+
+  int busy (void);
+  int idle (void);
+  int execute_delayed_operations (void);
+
+  void connected_i (PROXY *proxy);
+  void reconnected_i (PROXY *proxy);
+  void disconnected_i (PROXY *proxy);
+  void shutdown_i (void);
+
+  typedef TAO_ESF_Connected_Command<TAO_ESF_Delayed_Changes<PROXY,COLLECTION,ITERATOR,ACE_SYNCH_USE>,PROXY> Connected_Command;
+  typedef TAO_ESF_Reconnected_Command<TAO_ESF_Delayed_Changes<PROXY,COLLECTION,ITERATOR,ACE_SYNCH_USE>,PROXY> Reconnected_Command;
+  typedef TAO_ESF_Disconnected_Command<TAO_ESF_Delayed_Changes<PROXY,COLLECTION,ITERATOR,ACE_SYNCH_USE>,PROXY> Disconnected_Command;
+  typedef TAO_ESF_Shutdown_Command<TAO_ESF_Delayed_Changes<PROXY,COLLECTION,ITERATOR,ACE_SYNCH_USE> > Shutdown_Command;
+
+  // = The TAO_ESF_Proxy methods
+  virtual void for_each (TAO_ESF_Worker<PROXY> *worker);
+  virtual void connected (PROXY *proxy);
+  virtual void reconnected (PROXY *proxy);
+  virtual void disconnected (PROXY *proxy);
+  virtual void shutdown (void);
+
+private:
+  COLLECTION collection_;
+
+  typedef TAO_ESF_Busy_Lock_Adapter<TAO_ESF_Delayed_Changes<PROXY,COLLECTION,ITERATOR,ACE_SYNCH_USE> > Busy_Lock;
+
+  Busy_Lock lock_;
+
+  ACE_SYNCH_MUTEX_T busy_lock_;
+
+  ACE_SYNCH_CONDITION_T busy_cond_;
+
+  CORBA::ULong busy_count_;
+
+  CORBA::ULong write_delay_count_;
+
+  /// Control variables for the concurrency policies.
+  CORBA::ULong busy_hwm_;
+  CORBA::ULong max_write_delay_;
+
+  ACE_Unbounded_Queue<ACE_Command_Base*> command_queue_;
+};
+
+// ****************************************************************
+
+TAO_END_VERSIONED_NAMESPACE_DECL
+
+#if defined (__ACE_INLINE__)
+#include "orbsvcs/ESF/ESF_Delayed_Changes.inl"
+#endif /* __ACE_INLINE__ */
+
+#if defined (ACE_TEMPLATES_REQUIRE_SOURCE)
+#include "orbsvcs/ESF/ESF_Delayed_Changes.cpp"
+#endif /* ACE_TEMPLATES_REQUIRE_SOURCE */
+
+#if defined (ACE_TEMPLATES_REQUIRE_PRAGMA)
+#pragma implementation ("ESF_Delayed_Changes.cpp")
+#endif /* ACE_TEMPLATES_REQUIRE_PRAGMA */
+
+#endif /* TAO_ESF_DELAYED_CHANGES_H */