From 9bdde6941fc537b79070fd678dd62d710e954b9e Mon Sep 17 00:00:00 2001 From: Mehrdad Date: Sat, 2 Jul 2016 17:45:38 -0700 Subject: [PATCH] Simplify SynchronizedPtr --- src/scheduler.cc | 10 +++--- src/scheduler.h | 10 +++--- src/utils.h | 80 ++++++++++++++++++++++++++++++++---------------- 3 files changed, 64 insertions(+), 36 deletions(-) diff --git a/src/scheduler.cc b/src/scheduler.cc index 09c642a72..a7275dc3e 100644 --- a/src/scheduler.cc +++ b/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 > &computation_graph) { +void SchedulerService::assign_task(OperationId operationid, WorkerId workerid, const SynchronizedPtr &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 > > &reference_counts, const SynchronizedPtr > > > &contained_objrefs) { +void SchedulerService::deallocate_object(ObjRef canonical_objref, const SynchronizedPtr > &reference_counts, const SynchronizedPtr > > &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 &objrefs, const SynchronizedPtr > > &reference_counts) { +void SchedulerService::increment_ref_count(const std::vector &objrefs, const SynchronizedPtr > &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 &objrefs, c } } -void SchedulerService::decrement_ref_count(const std::vector &objrefs, const SynchronizedPtr > > &reference_counts, const SynchronizedPtr > > > &contained_objrefs) { +void SchedulerService::decrement_ref_count(const std::vector &objrefs, const SynchronizedPtr > &reference_counts, const SynchronizedPtr > > &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 &objrefs, c } } -void SchedulerService::upstream_objrefs(ObjRef objref, std::vector &objrefs, const SynchronizedPtr > > > &reverse_target_objrefs) { +void SchedulerService::upstream_objrefs(ObjRef objref, std::vector &objrefs, const SynchronizedPtr > > &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. diff --git a/src/scheduler.h b/src/scheduler.h index f084f3bea..f0289d783 100644 --- a/src/scheduler.h +++ b/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 > &computation_graph); + void assign_task(OperationId operationid, WorkerId workerid, const SynchronizedPtr &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 > > &reference_counts, const SynchronizedPtr > > > &contained_objrefs); + void deallocate_object(ObjRef canonical_objref, const SynchronizedPtr > &reference_counts, const SynchronizedPtr > > &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 &objrefs, const SynchronizedPtr > > &reference_count); + void increment_ref_count(const std::vector &objrefs, const SynchronizedPtr > &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 &objrefs, const SynchronizedPtr > > &reference_count, const SynchronizedPtr > > > &contained_objrefs); + void decrement_ref_count(const std::vector &objrefs, const SynchronizedPtr > &reference_count, const SynchronizedPtr > > &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 &objrefs, const SynchronizedPtr > > > &reverse_target_objrefs); + void upstream_objrefs(ObjRef objref, std::vector &objrefs, const SynchronizedPtr > > &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 &equivalent_objrefs); // acquires all locks, this should only be used by get_info and for fault tolerance diff --git a/src/utils.h b/src/utils.h index fd432535b..7dad63499 100644 --- a/src/utils.h +++ b/src/utils.h @@ -4,50 +4,78 @@ #include #include -template -struct SynchronizedTarget { - typedef typename Synchronized::element_type type; -}; +template +class Synchronized; -template -struct SynchronizedTarget { - typedef const typename Synchronized::element_type type; -}; +template +class Synchronized; // Prevent use of const T; it doesn't make sense -template -class SynchronizedPtr : public std::unique_lock { - typedef std::unique_lock base_type; +template struct SynchronizedSource { typedef Synchronized type; }; +template struct SynchronizedSource { typedef const Synchronized type; }; +template struct SynchronizedSource { typedef volatile Synchronized type; }; +template struct SynchronizedSource { typedef const Synchronized type; }; +template +class SynchronizedPtr : public std::unique_lock::type> { + typedef std::unique_lock::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::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 Synchronized { - mutable Mutex mutex_; +template +class Synchronized { T value_; public: typedef T element_type; + template + Synchronized(U&&... args) : value_(std::forward(args)...) { } + Synchronized(const Synchronized& other) : value_((std::lock_guard(other), other.value_)) { } + Synchronized(Synchronized&& other) : value_((std::lock_guard(other), std::move(other.value_))) { } + Synchronized& operator =(const Synchronized& other) + { + if (this != &other) + { + std::lock_guard guard_this(*this); + std::lock_guard guard_other(other); + value_ = other.value_; + } + return *this; + } + Synchronized& operator =(Synchronized&& other) + { + if (this != &other) + { + std::lock_guard guard_this(*this); + std::lock_guard 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 Synchronized : public Synchronized { + typedef Synchronized base_type; + mutable Mutex mutex_; +public: typedef Mutex mutex_type; template - Synchronized(U&&... args) : value_(std::forward(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(args)...) { } + SynchronizedPtr get() { return *this; } + SynchronizedPtr 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 get() { return *this; } - SynchronizedPtr get() const { return *this; } - element_type* unsafe_get() { return &value_; } - const element_type* unsafe_get() const { return &value_; } mutex_type& mutex() { return mutex_; } };