Added boost header

test/external/boost/graph/distributed/detail/remote_update_set.hpp

@@ -0,0 +1,259 @@
+// Copyright (C) 2005-2006 The Trustees of Indiana University.
+
+// Use, modification and distribution is subject to the Boost Software
+// License, Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at
+// http://www.boost.org/LICENSE_1_0.txt)
+
+//  Authors: Douglas Gregor
+//           Andrew Lumsdaine
+#ifndef BOOST_GRAPH_DETAIL_REMOTE_UPDATE_SET_HPP
+#define BOOST_GRAPH_DETAIL_REMOTE_UPDATE_SET_HPP
+
+#ifndef BOOST_GRAPH_USE_MPI
+#error "Parallel BGL files should not be included unless <boost/graph/use_mpi.hpp> has been included"
+#endif
+
+#include <boost/graph/parallel/process_group.hpp>
+#include <boost/type_traits/is_convertible.hpp>
+#include <vector>
+#include <boost/assert.hpp>
+#include <boost/optional.hpp>
+#include <queue>
+
+namespace boost { namespace graph { namespace detail {
+
+template<typename ProcessGroup>
+void do_synchronize(ProcessGroup& pg)
+{ 
+  using boost::parallel::synchronize;
+  synchronize(pg);
+}
+
+struct remote_set_queued {};
+struct remote_set_immediate {};
+
+template<typename ProcessGroup>
+class remote_set_semantics
+{
+  BOOST_STATIC_CONSTANT
+    (bool, 
+     queued = (is_convertible<
+                 typename ProcessGroup::communication_category,
+                 parallel::bsp_process_group_tag>::value));
+
+ public:
+  typedef typename mpl::if_c<queued, 
+                             remote_set_queued, 
+                             remote_set_immediate>::type type;
+};
+
+
+template<typename Derived, typename ProcessGroup, typename Value,
+         typename OwnerMap,
+         typename Semantics = typename remote_set_semantics<ProcessGroup>::type>
+class remote_update_set;
+
+/**********************************************************************
+ * Remote updating set that queues messages until synchronization     *
+ **********************************************************************/
+template<typename Derived, typename ProcessGroup, typename Value,
+         typename OwnerMap>
+class remote_update_set<Derived, ProcessGroup, Value, OwnerMap,
+                        remote_set_queued>
+{
+  typedef typename property_traits<OwnerMap>::key_type Key;
+  typedef std::vector<std::pair<Key, Value> > Updates;
+  typedef typename Updates::size_type   updates_size_type;
+  typedef typename Updates::value_type  updates_pair_type;
+
+public:
+
+private:
+  typedef typename ProcessGroup::process_id_type process_id_type;
+
+  enum message_kind {
+    /** Message containing the number of updates that will be sent in
+     *  a msg_updates message that will immediately follow. This
+     *  message will contain a single value of type
+     *  updates_size_type. 
+     */
+    msg_num_updates,
+
+    /** Contains (key, value) pairs with all of the updates from a
+     *  particular source. The number of updates is variable, but will
+     *  be provided in a msg_num_updates message that immediately
+     *  preceeds this message.
+     *
+     */
+    msg_updates
+  };
+
+  struct handle_messages
+  {
+    explicit 
+    handle_messages(remote_update_set* self, const ProcessGroup& pg)
+      : self(self), update_sizes(num_processes(pg), 0) { }
+
+    void operator()(process_id_type source, int tag) 
+    { 
+      switch(tag) {
+      case msg_num_updates:
+        {
+          // Receive the # of updates
+          updates_size_type num_updates;
+          receive(self->process_group, source, tag, num_updates);
+
+          update_sizes[source] = num_updates;
+        }
+        break;
+
+      case msg_updates:
+        {
+          updates_size_type num_updates = update_sizes[source];
+          BOOST_ASSERT(num_updates);
+
+          // Receive the actual updates
+          std::vector<updates_pair_type> updates(num_updates);
+          receive(self->process_group, source, msg_updates, &updates[0],
+                  num_updates);
+          
+          // Send updates to derived "receive_update" member
+          Derived* derived = static_cast<Derived*>(self);
+          for (updates_size_type u = 0; u < num_updates; ++u)
+            derived->receive_update(source, updates[u].first, updates[u].second);
+
+          update_sizes[source] = 0;
+        }
+        break;
+      };
+    }
+
+  private:
+    remote_update_set* self;
+    std::vector<updates_size_type> update_sizes;
+  };
+  friend struct handle_messages;
+
+ protected:
+  remote_update_set(const ProcessGroup& pg, const OwnerMap& owner)
+    : process_group(pg, handle_messages(this, pg)),
+      updates(num_processes(pg)), owner(owner) { 
+    }
+
+
+  void update(const Key& key, const Value& value)
+  { 
+    if (get(owner, key) == process_id(process_group)) {
+      Derived* derived = static_cast<Derived*>(this);
+      derived->receive_update(get(owner, key), key, value);
+    }
+    else {
+      updates[get(owner, key)].push_back(std::make_pair(key, value));
+    }
+  }
+
+  void collect() { }
+
+  void synchronize()
+  {
+    // Emit all updates and then remove them
+    process_id_type num_processes = updates.size();
+    for (process_id_type p = 0; p < num_processes; ++p) {
+      if (!updates[p].empty()) {
+        send(process_group, p, msg_num_updates, updates[p].size());
+        send(process_group, p, msg_updates, 
+             &updates[p].front(), updates[p].size());
+        updates[p].clear();
+      }
+    }
+    
+    do_synchronize(process_group);
+  }
+
+  ProcessGroup process_group;
+
+ private:
+  std::vector<Updates> updates;
+  OwnerMap owner;
+};
+
+/**********************************************************************
+ * Remote updating set that sends messages immediately                *
+ **********************************************************************/
+template<typename Derived, typename ProcessGroup, typename Value,
+         typename OwnerMap>
+class remote_update_set<Derived, ProcessGroup, Value, OwnerMap,
+                        remote_set_immediate>
+{
+  typedef typename property_traits<OwnerMap>::key_type Key;
+  typedef std::pair<Key, Value> update_pair_type;
+  typedef typename std::vector<update_pair_type>::size_type updates_size_type;
+
+public:
+  typedef typename ProcessGroup::process_id_type process_id_type;
+
+private:
+  enum message_kind {
+    /** Contains a (key, value) pair that will be updated. */
+    msg_update
+  };
+
+  struct handle_messages
+  {
+    explicit handle_messages(remote_update_set* self, const ProcessGroup& pg) 
+      : self(self)
+    { update_sizes.resize(num_processes(pg), 0); }
+
+    void operator()(process_id_type source, int tag) 
+    { 
+      // Receive the # of updates
+      BOOST_ASSERT(tag == msg_update);
+      update_pair_type update;
+      receive(self->process_group, source, tag, update);
+      
+      // Send update to derived "receive_update" member
+      Derived* derived = static_cast<Derived*>(self);
+      derived->receive_update(source, update.first, update.second);
+    }
+
+  private:
+    std::vector<updates_size_type> update_sizes;
+    remote_update_set* self;
+  };
+  friend struct handle_messages;
+
+ protected:
+  remote_update_set(const ProcessGroup& pg, const OwnerMap& owner)
+    : process_group(pg, handle_messages(this, pg)), owner(owner) { }
+
+  void update(const Key& key, const Value& value)
+  { 
+    if (get(owner, key) == process_id(process_group)) {
+      Derived* derived = static_cast<Derived*>(this);
+      derived->receive_update(get(owner, key), key, value);
+    }
+    else
+      send(process_group, get(owner, key), msg_update, 
+           update_pair_type(key, value));
+  }
+
+  void collect() 
+  { 
+    typedef std::pair<process_id_type, int> probe_type;
+    handle_messages handler(this, process_group);
+    while (optional<probe_type> stp = probe(process_group))
+      if (stp->second == msg_update) handler(stp->first, stp->second);
+  }
+
+  void synchronize()
+  {
+    do_synchronize(process_group);
+  }
+
+  ProcessGroup process_group;
+  OwnerMap owner;
+};
+
+} } } // end namespace boost::graph::detail
+
+#endif // BOOST_GRAPH_DETAIL_REMOTE_UPDATE_SET_HPP