Simplify SynchronizedPtr

src/scheduler.cc

@@ -349,7 +349,7 @@ void SchedulerService::schedule() {
 }
 
 // assign_task assumes that the canonical objrefs for its arguments are all ready, that is has_canonical_objref() is true for all of the call's arguments
-void SchedulerService::assign_task(OperationId operationid, WorkerId workerid, const SynchronizedPtr<Synchronized<ComputationGraph> > &computation_graph) {
+void SchedulerService::assign_task(OperationId operationid, WorkerId workerid, const SynchronizedPtr<ComputationGraph> &computation_graph) {
   // assign_task takes computation_graph as an argument, which is obtained by
   // computation_graph_.get(), so we know that the data structure has been
   // locked.
@@ -711,7 +711,7 @@ bool SchedulerService::attempt_notify_alias(ObjStoreId objstoreid, ObjRef alias_
   return true;
 }
 
-void SchedulerService::deallocate_object(ObjRef canonical_objref, const SynchronizedPtr<Synchronized<std::vector<RefCount> > > &reference_counts, const SynchronizedPtr<Synchronized<std::vector<std::vector<ObjRef> > > > &contained_objrefs) {
+void SchedulerService::deallocate_object(ObjRef canonical_objref, const SynchronizedPtr<std::vector<RefCount> > &reference_counts, const SynchronizedPtr<std::vector<std::vector<ObjRef> > > &contained_objrefs) {
   // deallocate_object should only be called from decrement_ref_count (note that
   // deallocate_object also recursively calls decrement_ref_count). Both of
   // these methods take reference_counts and contained_objrefs as argumens,
@@ -736,7 +736,7 @@ void SchedulerService::deallocate_object(ObjRef canonical_objref, const Synchron
   decrement_ref_count((*contained_objrefs)[canonical_objref], reference_counts, contained_objrefs);
 }
 
-void SchedulerService::increment_ref_count(const std::vector<ObjRef> &objrefs, const SynchronizedPtr<Synchronized<std::vector<RefCount> > > &reference_counts) {
+void SchedulerService::increment_ref_count(const std::vector<ObjRef> &objrefs, const SynchronizedPtr<std::vector<RefCount> > &reference_counts) {
   // increment_ref_count takes reference_counts as an argument, which is
   // obtained by reference_counts_.get(), so we know that the data structure has
   // been locked
@@ -748,7 +748,7 @@ void SchedulerService::increment_ref_count(const std::vector<ObjRef> &objrefs, c
   }
 }
 
-void SchedulerService::decrement_ref_count(const std::vector<ObjRef> &objrefs, const SynchronizedPtr<Synchronized<std::vector<RefCount> > > &reference_counts, const SynchronizedPtr<Synchronized<std::vector<std::vector<ObjRef> > > > &contained_objrefs) {
+void SchedulerService::decrement_ref_count(const std::vector<ObjRef> &objrefs, const SynchronizedPtr<std::vector<RefCount> > &reference_counts, const SynchronizedPtr<std::vector<std::vector<ObjRef> > > &contained_objrefs) {
   // decrement_ref_count takes reference_counts and contained_objrefs as
   // arguments, which are obtained by reference_counts_.get() and
   // contained_objrefs_.get(), so we know that those data structures have been
@@ -780,7 +780,7 @@ void SchedulerService::decrement_ref_count(const std::vector<ObjRef> &objrefs, c
   }
 }
 
-void SchedulerService::upstream_objrefs(ObjRef objref, std::vector<ObjRef> &objrefs, const SynchronizedPtr<Synchronized<std::vector<std::vector<ObjRef> > > > &reverse_target_objrefs) {
+void SchedulerService::upstream_objrefs(ObjRef objref, std::vector<ObjRef> &objrefs, const SynchronizedPtr<std::vector<std::vector<ObjRef> > > &reverse_target_objrefs) {
   // upstream_objrefs takes reverse_target_objrefs as an argument, which is
   // obtained by reverse_target_objrefs_.get(), so we know the data structure
   // has been locked.

src/scheduler.h

@@ -80,7 +80,7 @@ public:
   // assign a task to a worker
   void schedule();
   // execute a task on a worker and ship required object references
-  void assign_task(OperationId operationid, WorkerId workerid, const SynchronizedPtr<Synchronized<ComputationGraph> > &computation_graph);
+  void assign_task(OperationId operationid, WorkerId workerid, const SynchronizedPtr<ComputationGraph> &computation_graph);
   // checks if the dependencies of the task are met
   bool can_run(const Task& task);
   // register a worker and its object store (if it has not been registered yet)
@@ -116,15 +116,15 @@ private:
   bool attempt_notify_alias(ObjStoreId objstoreid, ObjRef alias_objref, ObjRef canonical_objref);
   // tell all of the objstores holding canonical_objref to deallocate it, the
   // data structures are passed into ensure that the appropriate locks are held.
-  void deallocate_object(ObjRef canonical_objref, const SynchronizedPtr<Synchronized<std::vector<RefCount> > > &reference_counts, const SynchronizedPtr<Synchronized<std::vector<std::vector<ObjRef> > > > &contained_objrefs);
+  void deallocate_object(ObjRef canonical_objref, const SynchronizedPtr<std::vector<RefCount> > &reference_counts, const SynchronizedPtr<std::vector<std::vector<ObjRef> > > &contained_objrefs);
   // increment the ref counts for the object references in objrefs, the data
   // structures are passed into ensure that the appropriate locks are held.
-  void increment_ref_count(const std::vector<ObjRef> &objrefs, const SynchronizedPtr<Synchronized<std::vector<RefCount> > > &reference_count);
+  void increment_ref_count(const std::vector<ObjRef> &objrefs, const SynchronizedPtr<std::vector<RefCount> > &reference_count);
   // decrement the ref counts for the object references in objrefs, the data
   // structures are passed into ensure that the appropriate locks are held.
-  void decrement_ref_count(const std::vector<ObjRef> &objrefs, const SynchronizedPtr<Synchronized<std::vector<RefCount> > > &reference_count, const SynchronizedPtr<Synchronized<std::vector<std::vector<ObjRef> > > > &contained_objrefs);
+  void decrement_ref_count(const std::vector<ObjRef> &objrefs, const SynchronizedPtr<std::vector<RefCount> > &reference_count, const SynchronizedPtr<std::vector<std::vector<ObjRef> > > &contained_objrefs);
   // Find all of the object references which are upstream of objref (including objref itself). That is, you can get from everything in objrefs to objref by repeatedly indexing in target_objrefs_.
-  void upstream_objrefs(ObjRef objref, std::vector<ObjRef> &objrefs, const SynchronizedPtr<Synchronized<std::vector<std::vector<ObjRef> > > > &reverse_target_objrefs);
+  void upstream_objrefs(ObjRef objref, std::vector<ObjRef> &objrefs, const SynchronizedPtr<std::vector<std::vector<ObjRef> > > &reverse_target_objrefs);
   // Find all of the object references that refer to the same object as objref (as best as we can determine at the moment). The information may be incomplete because not all of the aliases may be known.
   void get_equivalent_objrefs(ObjRef objref, std::vector<ObjRef> &equivalent_objrefs);
   // acquires all locks, this should only be used by get_info and for fault tolerance

src/utils.h

@@ -4,50 +4,78 @@
 #include <mutex>
 #include <string>
 
-template<class Synchronized>
-struct SynchronizedTarget {
-  typedef typename Synchronized::element_type type;
-};
+template<class T, class Mutex = std::mutex>
+class Synchronized;
 
-template<class Synchronized>
-struct SynchronizedTarget<const Synchronized> {
-  typedef const typename Synchronized::element_type type;
-};
+template<class T, class Mutex>
+class Synchronized<const T, Mutex>;  // Prevent use of const T; it doesn't make sense
 
-template<class Synchronized>
-class SynchronizedPtr : public std::unique_lock<Synchronized> {
-  typedef std::unique_lock<Synchronized> base_type;
+template<class T, class Mutex> struct SynchronizedSource { typedef Synchronized<T, Mutex> type; };
+template<class T, class Mutex> struct SynchronizedSource<const T, Mutex> { typedef const Synchronized<T, Mutex> type; };
+template<class T, class Mutex> struct SynchronizedSource<volatile T, Mutex> { typedef volatile Synchronized<T, Mutex> type; };
+template<class T, class Mutex> struct SynchronizedSource<const volatile T, Mutex> { typedef const Synchronized<T, Mutex> type; };
 
+template<class T>
+class SynchronizedPtr : public std::unique_lock<typename SynchronizedSource<T, void>::type> {
+  typedef std::unique_lock<typename SynchronizedSource<T, void>::type> base_type;
   // Make these private; they don't make much sense externally...
-  using typename base_type::mutex_type;
   using base_type::mutex;
 public:
-  typedef typename SynchronizedTarget<Synchronized>::type value_type;
-  SynchronizedPtr(Synchronized& value) : base_type(value) { }
+  typedef T value_type;
+  SynchronizedPtr(typename base_type::mutex_type& value) : base_type(value) { }
   value_type& operator*() const { return *mutex()->unsafe_get(); }
   value_type* operator->() const { return mutex() ? mutex()->unsafe_get() : NULL; }
 };
 
-template<class T, class Mutex = std::mutex>
-class Synchronized {
-  mutable Mutex mutex_;
+template<class T>
+class Synchronized<T, void> {
   T value_;
 public:
   typedef T element_type;
+  template<class... U>
+  Synchronized(U&&... args) : value_(std::forward<U>(args)...) { }
+  Synchronized(const Synchronized& other) : value_((std::lock_guard<Synchronized>(other), other.value_)) { }
+  Synchronized(Synchronized&& other) : value_((std::lock_guard<Synchronized>(other), std::move(other.value_))) { }
+  Synchronized& operator =(const Synchronized& other)
+  {
+    if (this != &other)
+    {
+      std::lock_guard<Synchronized> guard_this(*this);
+      std::lock_guard<Synchronized> guard_other(other);
+      value_ = other.value_;
+    }
+    return *this;
+  }
+  Synchronized& operator =(Synchronized&& other)
+  {
+    if (this != &other)
+    {
+      std::lock_guard<Synchronized> guard_this(*this);
+      std::lock_guard<Synchronized> guard_other(other);
+      value_ = std::move(other.value_);
+    }
+    return *this;
+  }
+  virtual void lock() const = 0;
+  virtual void unlock() const = 0;
+  virtual bool try_lock() const = 0;
+  element_type* unsafe_get() { return &value_; }
+  const element_type* unsafe_get() const { return &value_; }
+};
+
+template<class T, class Mutex>
+class Synchronized : public Synchronized<T, void> {
+  typedef Synchronized<T, void> base_type;
+  mutable Mutex mutex_;
+public:
   typedef Mutex mutex_type;
   template<class... U>
-  Synchronized(U&&... args) : value_(std::forward<T>(args)...) { }
-  Synchronized(const Synchronized& other) : value_(*other) { }
-  Synchronized(Synchronized&& other) : value_(std::move(*other)) { }
-  Synchronized& operator =(const Synchronized& other) { *get() = *other.get(); return *this; }
-  Synchronized& operator =(Synchronized&& other) { *get() = std::move(*other.get()); return *this; }
+  Synchronized(U&&... args) : base_type(std::forward<U>(args)...) { }
+  SynchronizedPtr<T> get() { return *this; }
+  SynchronizedPtr<const T> get() const { return *this; }
   void lock() const { return mutex_.lock(); }
   void unlock() const { return mutex_.unlock(); }
   bool try_lock() const { return mutex_.try_lock(); }
-  SynchronizedPtr<Synchronized> get() { return *this; }
-  SynchronizedPtr<const Synchronized> get() const { return *this; }
-  element_type* unsafe_get() { return &value_; }
-  const element_type* unsafe_get() const { return &value_; }
   mutex_type& mutex() { return mutex_; }
 };