Checked locking (#262)

* Lock order should be based on field order only

* Checked locking

src/scheduler.cc

@@ -3,16 +3,121 @@
 #include <random>
 #include <thread>
 #include <chrono>
+#include <sstream>
 
 #include "utils.h"
 
+// Macro used for acquiring locks. Required to pass along the field name and the line number without duplicating code.
+#define GET(FieldName) get(FieldName, #FieldName, __LINE__)
+
+#ifndef NDEBUG
+template<>
+class SchedulerService::MySynchronizedPtr<void> {
+  SchedulerService* me_;  // If NULL, then no lock is being checked
+  size_t order_delta_;
+  const char* name_;
+  unsigned int line_number_;
+  // ID returned seems to always be zero on Mac.
+  // I unfortunately can't find a workaround, so if the returned ID is zero, then the caller should not rely on it identifying the thread.
+  static unsigned long long get_thread_id() {
+    unsigned long long id = 0;
+    std::stringstream ss;
+    ss << std::this_thread::get_id();
+    ss >> id;
+    return id;
+  }
+protected:
+  MySynchronizedPtr& operator=(MySynchronizedPtr&& other) {
+    if (this != &other) {
+      me_ = std::move(other.me_);
+      order_delta_ = std::move(other.order_delta_);
+      name_ = std::move(other.name_);
+      line_number_ = std::move(other.line_number_);
+
+      other.me_ = NULL;  // Disable lock checking logic on other now that it has been moved
+    }
+    return *this;
+  }
+  ~MySynchronizedPtr() {
+    unsigned long long thread_id = get_thread_id();
+    if (thread_id != 0 && me_ != NULL) {
+      auto lock_orders = me_->lock_orders_.unchecked_get();
+      // Look for a previous lock on this thread -- it must exist, since this thread supposedly had the lock...
+      auto found = lock_orders->begin();
+      while (found != lock_orders->end() && found->first != thread_id) {
+        ++found;
+      }
+      RAY_CHECK(found != lock_orders->end() && found->second.first >= order_delta_, "Thread " << thread_id << " attempted to unlock a lock it didn't hold on line " << line_number_);
+      // Subtract back the delta
+      found->second.first -= order_delta_;
+      found->second.second = name_;
+      // If it goes to zero, then this thread no longer has locks, so remove it from the list
+      if (found->second.first == 0) {
+        using std::swap; swap(*found, lock_orders->back());
+        lock_orders->pop_back();
+      }
+      me_ = NULL;
+    }
+  }
+  MySynchronizedPtr(MySynchronizedPtr&& other) : me_() {
+    *this = std::move(other);
+  }
+  MySynchronizedPtr(SchedulerService* me, size_t order, const char* name, unsigned int line_number) : me_(me), order_delta_(order), name_(name), line_number_(line_number) {
+    unsigned long long thread_id = get_thread_id();
+    if (thread_id != 0 && me_ != NULL) {
+      auto lock_orders = me_->lock_orders_.unchecked_get();
+      auto found = lock_orders->begin();
+      // Look for a previous lock on this thread -- it shouldn't exist since these are not recursive locks
+      while (found != lock_orders->end() && found->first != thread_id) {
+        ++found;
+      }
+      if (found == lock_orders->end()) {
+        found = lock_orders->insert(found, std::make_pair(thread_id, std::make_pair(0, name_)));
+      } else if (thread_id != 0) {
+        RAY_CHECK_GE(order, found->second.first, "Thread " << thread_id << " attempted to lock " << name_ << " on line " << line_number_ << " after " << found->second.second);
+      }
+      // Store the delta between the last lock and this lock (each identified by the field offset) so we can reverse it
+      order_delta_ = order - found->second.first;
+      // Record the fact that we locked this field in the scheduler
+      found->second.first = order;
+      found->second.second = name_;
+    }
+  }
+};
+template<class T>
+class SchedulerService::MySynchronizedPtr : SynchronizedPtr<T>, MySynchronizedPtr<void> {
+  // TODO(mniknami): release(), etc. are private here -- implementing them is extra work we don't need yet
+public:
+  using SynchronizedPtr<T>::operator*;
+  using SynchronizedPtr<T>::operator->;
+  MySynchronizedPtr(SchedulerService& me, Synchronized<T>& value, const char* name, unsigned int line_number) :
+    SynchronizedPtr<T>(value.unchecked_get()),
+    MySynchronizedPtr<void>(&me, static_cast<size_t>(reinterpret_cast<unsigned char const *>(&value) - reinterpret_cast<unsigned char const *>(&me)), name, line_number) {
+  }
+  MySynchronizedPtr(MySynchronizedPtr&& other) = default;
+  MySynchronizedPtr& operator=(MySynchronizedPtr&& other) = default;
+};
+
+template<class T>
+SchedulerService::MySynchronizedPtr<T> SchedulerService::get(Synchronized<T>& my_field, const char* name, unsigned int line_number) { return MySynchronizedPtr<T>(*this, my_field, name, line_number); }
+
+template<class T>
+SchedulerService::MySynchronizedPtr<const T> SchedulerService::get(const Synchronized<T>& my_field, const char* name, unsigned int line_number) const { return MySynchronizedPtr<const T>(*this, my_field, name, line_number); }
+#else
+template<class T>
+SchedulerService::MySynchronizedPtr<T> SchedulerService::get(Synchronized<T>& my_field, const char* name, unsigned int line_number) { (void) name; (void) line_number; return my_field.unchecked_get(); }
+
+template<class T>
+SchedulerService::MySynchronizedPtr<const T> SchedulerService::get(const Synchronized<T>& my_field, const char* name, unsigned int line_number) const { (void) name; (void) line_number; return my_field.unchecked_get(); }
+#endif
+
 SchedulerService::SchedulerService(SchedulingAlgorithmType scheduling_algorithm) : scheduling_algorithm_(scheduling_algorithm) {}
 
 Status SchedulerService::SubmitTask(ServerContext* context, const SubmitTaskRequest* request, SubmitTaskReply* reply) {
   std::unique_ptr<Task> task(new Task(request->task())); // need to copy, because request is const
   size_t num_return_vals;
   {
-    auto fntable = fntable_.get();
+    auto fntable = GET(fntable_);
     FnTable::const_iterator fn = fntable->find(task->name());
     if (fn == fntable->end()) {
       num_return_vals = 0;
@@ -31,17 +136,17 @@ Status SchedulerService::SubmitTask(ServerContext* context, const SubmitTaskRequ
       result_objrefs.push_back(result);
     }
     {
-      auto reference_counts = reference_counts_.get();
+      auto reference_counts = GET(reference_counts_);
       increment_ref_count(result_objrefs, reference_counts); // We increment once so the objrefs don't go out of scope before we reply to the worker that called SubmitTask. The corresponding decrement will happen in submit_task in raylib.
       increment_ref_count(result_objrefs, reference_counts); // We increment once so the objrefs don't go out of scope before the task is scheduled on the worker. The corresponding decrement will happen in deserialize_task in raylib.
     }
 
     auto operation = std::unique_ptr<Operation>(new Operation());
     operation->set_allocated_task(task.release());
-    operation->set_creator_operationid((*workers_.get())[request->workerid()].current_task);
+    operation->set_creator_operationid((*GET(workers_))[request->workerid()].current_task);
 
-    OperationId operationid = computation_graph_.get()->add_operation(std::move(operation));
-    task_queue_.get()->push_back(operationid);
+    OperationId operationid = GET(computation_graph_)->add_operation(std::move(operation));
+    GET(task_queue_)->push_back(operationid);
     schedule();
   }
   return Status::OK;
@@ -51,22 +156,22 @@ Status SchedulerService::PutObj(ServerContext* context, const PutObjRequest* req
   ObjRef objref = register_new_object();
   auto operation = std::unique_ptr<Operation>(new Operation());
   operation->mutable_put()->set_objref(objref);
-  operation->set_creator_operationid((*workers_.get())[request->workerid()].current_task);
-  computation_graph_.get()->add_operation(std::move(operation));
+  operation->set_creator_operationid((*GET(workers_))[request->workerid()].current_task);
+  GET(computation_graph_)->add_operation(std::move(operation));
   reply->set_objref(objref);
   schedule();
   return Status::OK;
 }
 
 Status SchedulerService::RequestObj(ServerContext* context, const RequestObjRequest* request, AckReply* reply) {
-  size_t size = objtable_.get()->size();
+  size_t size = GET(objtable_)->size();
   ObjRef objref = request->objref();
   RAY_CHECK_LT(objref, size, "internal error: no object with objref " << objref << " exists");
   auto operation = std::unique_ptr<Operation>(new Operation());
   operation->mutable_get()->set_objref(objref);
-  operation->set_creator_operationid((*workers_.get())[request->workerid()].current_task);
-  computation_graph_.get()->add_operation(std::move(operation));
-  get_queue_.get()->push_back(std::make_pair(request->workerid(), objref));
+  operation->set_creator_operationid((*GET(workers_))[request->workerid()].current_task);
+  GET(computation_graph_)->add_operation(std::move(operation));
+  GET(get_queue_)->push_back(std::make_pair(request->workerid(), objref));
   schedule();
   return Status::OK;
 }
@@ -76,19 +181,19 @@ Status SchedulerService::AliasObjRefs(ServerContext* context, const AliasObjRefs
   ObjRef target_objref = request->target_objref();
   RAY_LOG(RAY_ALIAS, "Aliasing objref " << alias_objref << " with objref " << target_objref);
   RAY_CHECK_NEQ(alias_objref, target_objref, "internal error: attempting to alias objref " << alias_objref << " with itself.");
-  size_t size = objtable_.get()->size();
+  size_t size = GET(objtable_)->size();
   RAY_CHECK_LT(alias_objref, size, "internal error: no object with objref " << alias_objref << " exists");
   RAY_CHECK_LT(target_objref, size, "internal error: no object with objref " << target_objref << " exists");
   {
-    auto target_objrefs = target_objrefs_.get();
+    auto target_objrefs = GET(target_objrefs_);
     RAY_CHECK_EQ((*target_objrefs)[alias_objref], UNITIALIZED_ALIAS, "internal error: attempting to alias objref " << alias_objref << " with objref " << target_objref << ", but objref " << alias_objref << " has already been aliased with objref " << (*target_objrefs)[alias_objref]);
     (*target_objrefs)[alias_objref] = target_objref;
   }
-  (*reverse_target_objrefs_.get())[target_objref].push_back(alias_objref);
+  (*GET(reverse_target_objrefs_))[target_objref].push_back(alias_objref);
   {
     // The corresponding increment was done in register_new_object.
-    auto reference_counts = reference_counts_.get(); // we grab this lock because decrement_ref_count assumes it has been acquired
-    auto contained_objrefs = contained_objrefs_.get(); // we grab this lock because decrement_ref_count assumes it has been acquired
+    auto reference_counts = GET(reference_counts_); // we grab this lock because decrement_ref_count assumes it has been acquired
+    auto contained_objrefs = GET(contained_objrefs_); // we grab this lock because decrement_ref_count assumes it has been acquired
     decrement_ref_count(std::vector<ObjRef>({alias_objref}), reference_counts, contained_objrefs);
   }
   schedule();
@@ -96,8 +201,8 @@ Status SchedulerService::AliasObjRefs(ServerContext* context, const AliasObjRefs
 }
 
 Status SchedulerService::RegisterObjStore(ServerContext* context, const RegisterObjStoreRequest* request, RegisterObjStoreReply* reply) {
-  auto objtable = objtable_.get(); // to protect objects_in_transit_
-  auto objstores = objstores_.get();
+  auto objtable = GET(objtable_); // to protect objects_in_transit_
+  auto objstores = GET(objstores_);
   ObjStoreId objstoreid = objstores->size();
   auto channel = grpc::CreateChannel(request->objstore_address(), grpc::InsecureChannelCredentials());
   objstores->push_back(ObjStoreHandle());
@@ -137,8 +242,8 @@ Status SchedulerService::ObjReady(ServerContext* context, const ObjReadyRequest*
     // the corresponding increment was done in register_new_object in the
     // scheduler. For all subsequent calls to ObjReady, the corresponding
     // increment was done in deliver_object_if_necessary in the scheduler.
-    auto reference_counts = reference_counts_.get(); // we grab this lock because decrement_ref_count assumes it has been acquired
-    auto contained_objrefs = contained_objrefs_.get(); // we grab this lock because decrement_ref_count assumes it has been acquired
+    auto reference_counts = GET(reference_counts_); // we grab this lock because decrement_ref_count assumes it has been acquired
+    auto contained_objrefs = GET(contained_objrefs_); // we grab this lock because decrement_ref_count assumes it has been acquired
     decrement_ref_count(std::vector<ObjRef>({objref}), reference_counts, contained_objrefs);
   }
   schedule();
@@ -147,19 +252,19 @@ Status SchedulerService::ObjReady(ServerContext* context, const ObjReadyRequest*
 
 Status SchedulerService::ReadyForNewTask(ServerContext* context, const ReadyForNewTaskRequest* request, AckReply* reply) {
   WorkerId workerid = request->workerid();
-  OperationId operationid = (*workers_.get())[workerid].current_task;
+  OperationId operationid = (*GET(workers_))[workerid].current_task;
   RAY_LOG(RAY_INFO, "worker " << workerid << " is ready for a new task");
   RAY_CHECK(operationid != ROOT_OPERATION, "A driver appears to have called ReadyForNewTask.");
   {
     // Check if the worker has been initialized yet, and if not, then give it
     // all of the exported functions and all of the exported reusable variables.
-    auto workers = workers_.get();
+    auto workers = GET(workers_);
     if (!(*workers)[workerid].initialized) {
       // This should only happen once.
       // Import all remote functions on the worker.
-      export_all_functions_to_worker(workerid, workers, exported_functions_.get());
+      export_all_functions_to_worker(workerid, workers, GET(exported_functions_));
       // Import all reusable variables on the worker.
-      export_all_reusable_variables_to_worker(workerid, workers, exported_reusable_variables_.get());
+      export_all_reusable_variables_to_worker(workerid, workers, GET(exported_reusable_variables_));
       // Mark the worker as initialized.
       (*workers)[workerid].initialized = true;
     }
@@ -167,10 +272,10 @@ Status SchedulerService::ReadyForNewTask(ServerContext* context, const ReadyForN
   if (request->has_previous_task_info()) {
     RAY_CHECK(operationid != NO_OPERATION, "request->has_previous_task_info() should not be true if operationid == NO_OPERATION.");
     std::string task_name;
-    task_name = computation_graph_.get()->get_task(operationid).name();
+    task_name = GET(computation_graph_)->get_task(operationid).name();
     TaskStatus info;
     {
-      auto workers = workers_.get();
+      auto workers = GET(workers_);
       info.set_operationid(operationid);
       info.set_function_name(task_name);
       info.set_worker_address((*workers)[workerid].worker_address);
@@ -179,13 +284,13 @@ Status SchedulerService::ReadyForNewTask(ServerContext* context, const ReadyForN
     }
     if (!request->previous_task_info().task_succeeded()) {
       RAY_LOG(RAY_INFO, "Error: Task " << info.operationid() << " executing function " << info.function_name() << " on worker " << workerid << " failed with error message:\n" << info.error_message());
-      failed_tasks_.get()->push_back(info);
+      GET(failed_tasks_)->push_back(info);
     } else {
-      successful_tasks_.get()->push_back(info.operationid());
+      GET(successful_tasks_)->push_back(info.operationid());
     }
     // TODO(rkn): Handle task failure
   }
-  avail_workers_.get()->push_back(workerid);
+  GET(avail_workers_)->push_back(workerid);
   schedule();
   return Status::OK;
 }
@@ -197,7 +302,7 @@ Status SchedulerService::IncrementRefCount(ServerContext* context, const Increme
   for (int i = 0; i < num_objrefs; ++i) {
     objrefs.push_back(request->objref(i));
   }
-  auto reference_counts = reference_counts_.get();
+  auto reference_counts = GET(reference_counts_);
   increment_ref_count(objrefs, reference_counts);
   return Status::OK;
 }
@@ -209,8 +314,8 @@ Status SchedulerService::DecrementRefCount(ServerContext* context, const Decreme
   for (int i = 0; i < num_objrefs; ++i) {
     objrefs.push_back(request->objref(i));
   }
-  auto reference_counts = reference_counts_.get(); // we grab this lock, because decrement_ref_count assumes it has been acquired
-  auto contained_objrefs = contained_objrefs_.get(); // we grab this lock because decrement_ref_count assumes it has been acquired
+  auto reference_counts = GET(reference_counts_); // we grab this lock, because decrement_ref_count assumes it has been acquired
+  auto contained_objrefs = GET(contained_objrefs_); // we grab this lock because decrement_ref_count assumes it has been acquired
   decrement_ref_count(objrefs, reference_counts, contained_objrefs);
   return Status::OK;
 }
@@ -221,7 +326,7 @@ Status SchedulerService::AddContainedObjRefs(ServerContext* context, const AddCo
     // TODO(rkn): Perhaps we don't need this check. It won't work because the objstore may not have called ObjReady yet.
     // RAY_LOG(RAY_FATAL, "Attempting to add contained objrefs for non-canonical objref " << objref);
   // }
-  auto contained_objrefs = contained_objrefs_.get();
+  auto contained_objrefs = GET(contained_objrefs_);
   RAY_CHECK_EQ((*contained_objrefs)[objref].size(), 0, "Attempting to add contained objrefs for objref " << objref << ", but contained_objrefs_[objref].size() != 0.");
   for (int i = 0; i < request->contained_objref_size(); ++i) {
     (*contained_objrefs)[objref].push_back(request->contained_objref(i));
@@ -235,10 +340,10 @@ Status SchedulerService::SchedulerInfo(ServerContext* context, const SchedulerIn
 }
 
 Status SchedulerService::TaskInfo(ServerContext* context, const TaskInfoRequest* request, TaskInfoReply* reply) {
-  auto successful_tasks = successful_tasks_.get();
-  auto failed_tasks = failed_tasks_.get();
-  auto computation_graph = computation_graph_.get();
-  auto workers = workers_.get();
+  auto successful_tasks = GET(successful_tasks_);
+  auto failed_tasks = GET(failed_tasks_);
+  auto computation_graph = GET(computation_graph_);
+  auto workers = GET(workers_);
   for (int i = 0; i < failed_tasks->size(); ++i) {
     TaskStatus* info = reply->add_failed_task();
     *info = (*failed_tasks)[i];
@@ -259,13 +364,13 @@ Status SchedulerService::TaskInfo(ServerContext* context, const TaskInfoRequest*
 
 Status SchedulerService::KillWorkers(ServerContext* context, const KillWorkersRequest* request, KillWorkersReply* reply) {
   // TODO: Update reference counts
-  auto failed_tasks = failed_tasks_.get();
-  auto get_queue = get_queue_.get();
-  auto computation_graph = computation_graph_.get();
-  auto fntable = fntable_.get();
-  auto avail_workers = avail_workers_.get();
-  auto task_queue = task_queue_.get();
-  auto workers = workers_.get();
+  auto failed_tasks = GET(failed_tasks_);
+  auto get_queue = GET(get_queue_);
+  auto computation_graph = GET(computation_graph_);
+  auto fntable = GET(fntable_);
+  auto avail_workers = GET(avail_workers_);
+  auto task_queue = GET(task_queue_);
+  auto workers = GET(workers_);
   size_t busy_workers = 0;
   std::vector<WorkerHandle*> idle_workers;
   RAY_LOG(RAY_INFO, "Attempting to kill workers.");
@@ -306,8 +411,8 @@ Status SchedulerService::KillWorkers(ServerContext* context, const KillWorkersRe
 }
 
 Status SchedulerService::ExportFunction(ServerContext* context, const ExportFunctionRequest* request, ExportFunctionReply* reply) {
-  auto workers = workers_.get();
-  auto exported_functions = exported_functions_.get();
+  auto workers = GET(workers_);
+  auto exported_functions = GET(exported_functions_);
   // TODO(rkn): Does this do a deep copy?
   exported_functions->push_back(std::unique_ptr<Function>(new Function(request->function())));
   for (size_t i = 0; i < workers->size(); ++i) {
@@ -319,8 +424,8 @@ Status SchedulerService::ExportFunction(ServerContext* context, const ExportFunc
 }
 
 Status SchedulerService::ExportReusableVariable(ServerContext* context, const ExportReusableVariableRequest* request, AckReply* reply) {
-  auto workers = workers_.get();
-  auto exported_reusable_variables = exported_reusable_variables_.get();
+  auto workers = GET(workers_);
+  auto exported_reusable_variables = GET(exported_reusable_variables_);
   // TODO(rkn): Does this do a deep copy?
   exported_reusable_variables->push_back(std::unique_ptr<ReusableVar>(new ReusableVar(request->reusable_variable())));
   for (size_t i = 0; i < workers->size(); ++i) {
@@ -334,7 +439,7 @@ Status SchedulerService::ExportReusableVariable(ServerContext* context, const Ex
 void SchedulerService::deliver_object_async_if_necessary(ObjRef canonical_objref, ObjStoreId from, ObjStoreId to) {
   bool object_present_or_in_transit;
   {
-    auto objtable = objtable_.get();
+    auto objtable = GET(objtable_);
     auto &locations = (*objtable)[canonical_objref];
     bool object_present = std::binary_search(locations.begin(), locations.end(), to);
     auto &objects_in_flight = objects_in_transit_[to];
@@ -363,14 +468,14 @@ void SchedulerService::deliver_object_async(ObjRef canonical_objref, ObjStoreId
     // We increment once so the objref doesn't go out of scope before the ObjReady
     // method is called. The corresponding decrement will happen in ObjReady in
     // the scheduler.
-    auto reference_counts = reference_counts_.get(); // we grab this lock because increment_ref_count assumes it has been acquired
+    auto reference_counts = GET(reference_counts_); // we grab this lock because increment_ref_count assumes it has been acquired
     increment_ref_count(std::vector<ObjRef>({canonical_objref}), reference_counts);
   }
   ClientContext context;
   AckReply reply;
   StartDeliveryRequest request;
   request.set_objref(canonical_objref);
-  auto objstores = objstores_.get();
+  auto objstores = GET(objstores_);
   request.set_objstore_address((*objstores)[from].address);
   (*objstores)[to].objstore_stub->StartDelivery(&context, request, &reply);
 }
@@ -389,9 +494,9 @@ 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<ComputationGraph> &computation_graph) {
+void SchedulerService::assign_task(OperationId operationid, WorkerId workerid, const MySynchronizedPtr<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
+  // GET(computation_graph_), so we know that the data structure has been
   // locked.
   ObjStoreId objstoreid = get_store(workerid);
   const Task& task = computation_graph->get_task(operationid);
@@ -404,14 +509,14 @@ void SchedulerService::assign_task(OperationId operationid, WorkerId workerid, c
       ObjRef objref = task.arg(i).ref();
       ObjRef canonical_objref = get_canonical_objref(objref);
       // Notify the relevant objstore about potential aliasing when it's ready
-      alias_notification_queue_.get()->push_back(std::make_pair(objstoreid, std::make_pair(objref, canonical_objref)));
+      GET(alias_notification_queue_)->push_back(std::make_pair(objstoreid, std::make_pair(objref, canonical_objref)));
       attempt_notify_alias(objstoreid, objref, canonical_objref);
       RAY_LOG(RAY_DEBUG, "task contains object ref " << canonical_objref);
       deliver_object_async_if_necessary(canonical_objref, pick_objstore(canonical_objref), objstoreid);
     }
   }
   {
-    auto workers = workers_.get();
+    auto workers = GET(workers_);
     (*workers)[workerid].current_task = operationid;
     request.mutable_task()->CopyFrom(task); // TODO(rkn): Is ownership handled properly here?
     Status status = (*workers)[workerid].worker_stub->ExecuteTask(&context, request, &reply);
@@ -419,7 +524,7 @@ void SchedulerService::assign_task(OperationId operationid, WorkerId workerid, c
 }
 
 bool SchedulerService::can_run(const Task& task) {
-  auto objtable = objtable_.get();
+  auto objtable = GET(objtable_);
   for (int i = 0; i < task.arg_size(); ++i) {
     if (!task.arg(i).has_obj()) {
       ObjRef objref = task.arg(i).ref();
@@ -440,7 +545,7 @@ std::pair<WorkerId, ObjStoreId> SchedulerService::register_worker(const std::str
   ObjStoreId objstoreid = std::numeric_limits<size_t>::max();
   // TODO: HACK: num_attempts is a hack
   for (int num_attempts = 0; num_attempts < 5; ++num_attempts) {
-    auto objstores = objstores_.get();
+    auto objstores = GET(objstores_);
     for (size_t i = 0; i < objstores->size(); ++i) {
       if ((*objstores)[i].address == objstore_address) {
         objstoreid = i;
@@ -453,7 +558,7 @@ std::pair<WorkerId, ObjStoreId> SchedulerService::register_worker(const std::str
   RAY_CHECK_NEQ(objstoreid, std::numeric_limits<size_t>::max(), "object store with address " << objstore_address << " not yet registered");
   WorkerId workerid;
   {
-    auto workers = workers_.get();
+    auto workers = GET(workers_);
     workerid = workers->size();
     workers->push_back(WorkerHandle());
     auto channel = grpc::CreateChannel(worker_address, grpc::InsecureChannelCredentials());
@@ -474,11 +579,11 @@ std::pair<WorkerId, ObjStoreId> SchedulerService::register_worker(const std::str
 ObjRef SchedulerService::register_new_object() {
   // If we don't simultaneously lock objtable_ and target_objrefs_, we will probably get errors.
   // TODO(rkn): increment/decrement_reference_count also acquire reference_counts_lock_ and target_objrefs_lock_ (through has_canonical_objref()), which caused deadlock in the past
-  auto reference_counts = reference_counts_.get();
-  auto contained_objrefs = contained_objrefs_.get();
-  auto objtable = objtable_.get();
-  auto target_objrefs = target_objrefs_.get();
-  auto reverse_target_objrefs = reverse_target_objrefs_.get();
+  auto reference_counts = GET(reference_counts_);
+  auto contained_objrefs = GET(contained_objrefs_);
+  auto objtable = GET(objtable_);
+  auto target_objrefs = GET(target_objrefs_);
+  auto reverse_target_objrefs = GET(reverse_target_objrefs_);
   ObjRef objtable_size = objtable->size();
   ObjRef target_objrefs_size = target_objrefs->size();
   ObjRef reverse_target_objrefs_size = reverse_target_objrefs->size();
@@ -504,9 +609,9 @@ ObjRef SchedulerService::register_new_object() {
 
 void SchedulerService::add_location(ObjRef canonical_objref, ObjStoreId objstoreid) {
   // add_location must be called with a canonical objref
-  RAY_CHECK_NEQ((*reference_counts_.get())[canonical_objref], DEALLOCATED, "Calling ObjReady with canonical_objref " << canonical_objref << ", but this objref has already been deallocated");
+  RAY_CHECK_NEQ((*GET(reference_counts_))[canonical_objref], DEALLOCATED, "Calling ObjReady with canonical_objref " << canonical_objref << ", but this objref has already been deallocated");
   RAY_CHECK(is_canonical(canonical_objref), "Attempting to call add_location with a non-canonical objref (objref " << canonical_objref << ")");
-  auto objtable = objtable_.get();
+  auto objtable = GET(objtable_);
   RAY_CHECK_LT(canonical_objref, objtable->size(), "trying to put an object in the object store that was not registered with the scheduler (objref " << canonical_objref << ")");
   // do a binary search
   auto &locations = (*objtable)[canonical_objref];
@@ -519,87 +624,89 @@ void SchedulerService::add_location(ObjRef canonical_objref, ObjStoreId objstore
 }
 
 void SchedulerService::add_canonical_objref(ObjRef objref) {
-  auto target_objrefs = target_objrefs_.get();
+  auto target_objrefs = GET(target_objrefs_);
   RAY_CHECK_LT(objref, target_objrefs->size(), "internal error: attempting to insert objref " << objref << " in target_objrefs_, but target_objrefs_.size() is " << target_objrefs->size());
   RAY_CHECK((*target_objrefs)[objref] == UNITIALIZED_ALIAS || (*target_objrefs)[objref] == objref, "internal error: attempting to declare objref " << objref << " as a canonical objref, but target_objrefs_[objref] is already aliased with objref " << (*target_objrefs)[objref]);
   (*target_objrefs)[objref] = objref;
 }
 
 ObjStoreId SchedulerService::get_store(WorkerId workerid) {
-  auto workers = workers_.get();
+  auto workers = GET(workers_);
   ObjStoreId result = (*workers)[workerid].objstoreid;
   return result;
 }
 
 void SchedulerService::register_function(const std::string& name, WorkerId workerid, size_t num_return_vals) {
-  auto fntable = fntable_.get();
+  auto fntable = GET(fntable_);
   FnInfo& info = (*fntable)[name];
   info.set_num_return_vals(num_return_vals);
   info.add_worker(workerid);
 }
 
 void SchedulerService::get_info(const SchedulerInfoRequest& request, SchedulerInfoReply* reply) {
-  acquire_all_locks();
-  auto reference_counts = reference_counts_.unsafe_get();
+  auto computation_graph = GET(computation_graph_);
+  auto fntable = GET(fntable_);
+  auto avail_workers = GET(avail_workers_);
+  auto task_queue = GET(task_queue_);
+  auto reference_counts = GET(reference_counts_);
+  auto target_objrefs = GET(target_objrefs_);
+  auto function_table = reply->mutable_function_table();
   for (int i = 0; i < reference_counts->size(); ++i) {
     reply->add_reference_count((*reference_counts)[i]);
   }
-  auto target_objrefs = target_objrefs_.unsafe_get();
   for (int i = 0; i < target_objrefs->size(); ++i) {
     reply->add_target_objref((*target_objrefs)[i]);
   }
-  auto function_table = reply->mutable_function_table();
-  for (const auto& entry : *fntable_.unsafe_get()) {
+  for (const auto& entry : *fntable) {
     (*function_table)[entry.first].set_num_return_vals(entry.second.num_return_vals());
     for (const WorkerId& worker : entry.second.workers()) {
       (*function_table)[entry.first].add_workerid(worker);
     }
   }
-  for (const auto& entry : *task_queue_.unsafe_get()) {
+  for (const auto& entry : *task_queue) {
     reply->add_operationid(entry);
   }
-  for (const WorkerId& entry : *avail_workers_.unsafe_get()) {
+  for (const WorkerId& entry : *avail_workers) {
     reply->add_avail_worker(entry);
   }
-  computation_graph_.unsafe_get()->to_protobuf(reply->mutable_computation_graph());
-  release_all_locks();
+  computation_graph->to_protobuf(reply->mutable_computation_graph());
 }
 
 // pick_objstore must be called with a canonical_objref
 ObjStoreId SchedulerService::pick_objstore(ObjRef canonical_objref) {
   std::mt19937 rng;
   RAY_CHECK(is_canonical(canonical_objref), "Attempting to call pick_objstore with a non-canonical objref, (objref " << canonical_objref << ")");
-  auto objtable = objtable_.get();
+  auto objtable = GET(objtable_);
   std::uniform_int_distribution<int> uni(0, (*objtable)[canonical_objref].size() - 1);
   ObjStoreId objstoreid = (*objtable)[canonical_objref][uni(rng)];
   return objstoreid;
 }
 
 bool SchedulerService::is_canonical(ObjRef objref) {
-  auto target_objrefs = target_objrefs_.get();
+  auto target_objrefs = GET(target_objrefs_);
   RAY_CHECK_NEQ((*target_objrefs)[objref], UNITIALIZED_ALIAS, "Attempting to call is_canonical on an objref for which aliasing is not complete or the object is not ready, target_objrefs_[objref] == UNITIALIZED_ALIAS for objref " << objref << ".");
   return objref == (*target_objrefs)[objref];
 }
 
 void SchedulerService::perform_gets() {
-  auto get_queue = get_queue_.get();
+  auto get_queue = GET(get_queue_);
   // Complete all get tasks that can be completed.
   for (int i = 0; i < get_queue->size(); ++i) {
-    const std::pair<WorkerId, ObjRef>& get = (*get_queue)[i];
-    ObjRef objref = get.second;
-    WorkerId workerid = get.first;
+    const std::pair<WorkerId, ObjRef>& get_request = (*get_queue)[i];
+    ObjRef objref = get_request.second;
+    WorkerId workerid = get_request.first;
     ObjStoreId objstoreid = get_store(workerid);
     if (!has_canonical_objref(objref)) {
       RAY_LOG(RAY_ALIAS, "objref " << objref << " does not have a canonical_objref, so continuing");
       continue;
     }
     ObjRef canonical_objref = get_canonical_objref(objref);
-    RAY_LOG(RAY_DEBUG, "attempting to get objref " << get.second << " with canonical objref " << canonical_objref << " to objstore " << objstoreid);
-    int num_stores = (*objtable_.get())[canonical_objref].size();
+    RAY_LOG(RAY_DEBUG, "attempting to get objref " << get_request.second << " with canonical objref " << canonical_objref << " to objstore " << objstoreid);
+    int num_stores = (*GET(objtable_))[canonical_objref].size();
     if (num_stores > 0) {
       deliver_object_async_if_necessary(canonical_objref, pick_objstore(canonical_objref), objstoreid);
       // Notify the relevant objstore about potential aliasing when it's ready
-      alias_notification_queue_.get()->push_back(std::make_pair(objstoreid, std::make_pair(objref, canonical_objref)));
+      GET(alias_notification_queue_)->push_back(std::make_pair(objstoreid, std::make_pair(objref, canonical_objref)));
       // Remove the get task from the queue
       std::swap((*get_queue)[i], (*get_queue)[get_queue->size() - 1]);
       get_queue->pop_back();
@@ -609,10 +716,10 @@ void SchedulerService::perform_gets() {
 }
 
 void SchedulerService::schedule_tasks_naively() {
-  auto computation_graph = computation_graph_.get();
-  auto fntable = fntable_.get();
-  auto avail_workers = avail_workers_.get();
-  auto task_queue = task_queue_.get();
+  auto computation_graph = GET(computation_graph_);
+  auto fntable = GET(fntable_);
+  auto avail_workers = GET(avail_workers_);
+  auto task_queue = GET(task_queue_);
   for (int i = 0; i < avail_workers->size(); ++i) {
     // Submit all tasks whose arguments are ready.
     WorkerId workerid = (*avail_workers)[i];
@@ -636,10 +743,10 @@ void SchedulerService::schedule_tasks_naively() {
 }
 
 void SchedulerService::schedule_tasks_location_aware() {
-  auto computation_graph = computation_graph_.get();
-  auto fntable = fntable_.get();
-  auto avail_workers = avail_workers_.get();
-  auto task_queue = task_queue_.get();
+  auto computation_graph = GET(computation_graph_);
+  auto fntable = GET(fntable_);
+  auto avail_workers = GET(avail_workers_);
+  auto task_queue = GET(task_queue_);
   for (int i = 0; i < avail_workers->size(); ++i) {
     // Submit all tasks whose arguments are ready.
     WorkerId workerid = (*avail_workers)[i];
@@ -660,7 +767,7 @@ void SchedulerService::schedule_tasks_location_aware() {
             ObjRef canonical_objref = get_canonical_objref(objref);
             {
               // check if the object is already in the local object store
-              auto objtable = objtable_.get();
+              auto objtable = GET(objtable_);
               if (!std::binary_search((*objtable)[canonical_objref].begin(), (*objtable)[canonical_objref].end(), objstoreid)) {
                 num_shipped_objects += 1;
               }
@@ -685,7 +792,7 @@ void SchedulerService::schedule_tasks_location_aware() {
 }
 
 void SchedulerService::perform_notify_aliases() {
-  auto alias_notification_queue = alias_notification_queue_.get();
+  auto alias_notification_queue = GET(alias_notification_queue_);
   for (int i = 0; i < alias_notification_queue->size(); ++i) {
     const std::pair<WorkerId, std::pair<ObjRef, ObjRef> > alias_notification = (*alias_notification_queue)[i];
     ObjStoreId objstoreid = alias_notification.first;
@@ -701,7 +808,7 @@ void SchedulerService::perform_notify_aliases() {
 }
 
 bool SchedulerService::has_canonical_objref(ObjRef objref) {
-  auto target_objrefs = target_objrefs_.get();
+  auto target_objrefs = GET(target_objrefs_);
   ObjRef objref_temp = objref;
   while (true) {
     RAY_CHECK_LT(objref_temp, target_objrefs->size(), "Attempting to index target_objrefs_ with objref " << objref_temp << ", but target_objrefs_.size() = " << target_objrefs->size());
@@ -717,7 +824,7 @@ bool SchedulerService::has_canonical_objref(ObjRef objref) {
 
 ObjRef SchedulerService::get_canonical_objref(ObjRef objref) {
   // get_canonical_objref assumes that has_canonical_objref(objref) is true
-  auto target_objrefs = target_objrefs_.get();
+  auto target_objrefs = GET(target_objrefs_);
   ObjRef objref_temp = objref;
   while (true) {
     RAY_CHECK_LT(objref_temp, target_objrefs->size(), "Attempting to index target_objrefs_ with objref " << objref_temp << ", but target_objrefs_.size() = " << target_objrefs->size());
@@ -737,7 +844,7 @@ bool SchedulerService::attempt_notify_alias(ObjStoreId objstoreid, ObjRef alias_
     return true;
   }
   {
-    auto objtable = objtable_.get();
+    auto objtable = GET(objtable_);
     if (!std::binary_search((*objtable)[canonical_objref].begin(), (*objtable)[canonical_objref].end(), objstoreid)) {
       // the objstore doesn't have the object for canonical_objref yet, so it's too early to notify the objstore about the alias
       return false;
@@ -748,21 +855,21 @@ bool SchedulerService::attempt_notify_alias(ObjStoreId objstoreid, ObjRef alias_
   NotifyAliasRequest request;
   request.set_alias_objref(alias_objref);
   request.set_canonical_objref(canonical_objref);
-  (*objstores_.get())[objstoreid].objstore_stub->NotifyAlias(&context, request, &reply);
+  (*GET(objstores_))[objstoreid].objstore_stub->NotifyAlias(&context, request, &reply);
   return true;
 }
 
-void SchedulerService::deallocate_object(ObjRef canonical_objref, const SynchronizedPtr<std::vector<RefCount> > &reference_counts, const SynchronizedPtr<std::vector<std::vector<ObjRef> > > &contained_objrefs) {
+void SchedulerService::deallocate_object(ObjRef canonical_objref, const MySynchronizedPtr<std::vector<RefCount> > &reference_counts, const MySynchronizedPtr<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,
-  // which are obtained by reference_counts_.get() and contained_objrefs_.get(),
+  // which are obtained by GET(reference_counts) and GET(contained_objrefs_),
   // so we know that those data structures have been locked
   RAY_LOG(RAY_REFCOUNT, "Deallocating canonical_objref " << canonical_objref << ".");
   {
-    auto objtable = objtable_.get();
+    auto objtable = GET(objtable_);
     auto &locations = (*objtable)[canonical_objref];
-    auto objstores = objstores_.get(); // TODO(rkn): Should this be inside the for loop instead?
+    auto objstores = GET(objstores_); // TODO(rkn): Should this be inside the for loop instead?
     for (int i = 0; i < locations.size(); ++i) {
       ClientContext context;
       AckReply reply;
@@ -777,9 +884,9 @@ 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<std::vector<RefCount> > &reference_counts) {
+void SchedulerService::increment_ref_count(const std::vector<ObjRef> &objrefs, const MySynchronizedPtr<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
+  // obtained by GET(reference_counts_), so we know that the data structure has
   // been locked
   for (int i = 0; i < objrefs.size(); ++i) {
     ObjRef objref = objrefs[i];
@@ -789,10 +896,10 @@ void SchedulerService::increment_ref_count(const std::vector<ObjRef> &objrefs, c
   }
 }
 
-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) {
+void SchedulerService::decrement_ref_count(const std::vector<ObjRef> &objrefs, const MySynchronizedPtr<std::vector<RefCount> > &reference_counts, const MySynchronizedPtr<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
+  // arguments, which are obtained by GET(reference_counts_) and
+  // GET(contained_objrefs_), so we know that those data structures have been
   // locked
   for (int i = 0; i < objrefs.size(); ++i) {
     ObjRef objref = objrefs[i];
@@ -821,9 +928,9 @@ void SchedulerService::decrement_ref_count(const std::vector<ObjRef> &objrefs, c
   }
 }
 
-void SchedulerService::upstream_objrefs(ObjRef objref, std::vector<ObjRef> &objrefs, const SynchronizedPtr<std::vector<std::vector<ObjRef> > > &reverse_target_objrefs) {
+void SchedulerService::upstream_objrefs(ObjRef objref, std::vector<ObjRef> &objrefs, const MySynchronizedPtr<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
+  // obtained by GET(reverse_target_objrefs_), so we know the data structure
   // has been locked.
   objrefs.push_back(objref);
   for (int i = 0; i < (*reverse_target_objrefs)[objref].size(); ++i) {
@@ -832,17 +939,17 @@ void SchedulerService::upstream_objrefs(ObjRef objref, std::vector<ObjRef> &objr
 }
 
 void SchedulerService::get_equivalent_objrefs(ObjRef objref, std::vector<ObjRef> &equivalent_objrefs) {
-  auto target_objrefs = target_objrefs_.get();
+  auto target_objrefs = GET(target_objrefs_);
   ObjRef downstream_objref = objref;
   while ((*target_objrefs)[downstream_objref] != downstream_objref && (*target_objrefs)[downstream_objref] != UNITIALIZED_ALIAS) {
     RAY_LOG(RAY_ALIAS, "Looping in get_equivalent_objrefs");
     downstream_objref = (*target_objrefs)[downstream_objref];
   }
-  upstream_objrefs(downstream_objref, equivalent_objrefs, reverse_target_objrefs_.get());
+  upstream_objrefs(downstream_objref, equivalent_objrefs, GET(reverse_target_objrefs_));
 }
 
 
-void SchedulerService::export_function_to_worker(WorkerId workerid, int function_index, SynchronizedPtr<std::vector<WorkerHandle> > &workers, const SynchronizedPtr<std::vector<std::unique_ptr<Function> > > &exported_functions) {
+void SchedulerService::export_function_to_worker(WorkerId workerid, int function_index, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<Function> > > &exported_functions) {
   RAY_LOG(RAY_INFO, "exporting function with index " << function_index << " to worker " << workerid);
   ClientContext import_context;
   ImportFunctionRequest import_request;
@@ -851,7 +958,7 @@ void SchedulerService::export_function_to_worker(WorkerId workerid, int function
   (*workers)[workerid].worker_stub->ImportFunction(&import_context, import_request, &import_reply);
 }
 
-void SchedulerService::export_reusable_variable_to_worker(WorkerId workerid, int reusable_variable_index, SynchronizedPtr<std::vector<WorkerHandle> > &workers, const SynchronizedPtr<std::vector<std::unique_ptr<ReusableVar> > > &exported_reusable_variables) {
+void SchedulerService::export_reusable_variable_to_worker(WorkerId workerid, int reusable_variable_index, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<ReusableVar> > > &exported_reusable_variables) {
   RAY_LOG(RAY_INFO, "exporting reusable variable with index " << reusable_variable_index << " to worker " << workerid);
   ClientContext import_context;
   ImportReusableVariableRequest import_request;
@@ -860,57 +967,18 @@ void SchedulerService::export_reusable_variable_to_worker(WorkerId workerid, int
   (*workers)[workerid].worker_stub->ImportReusableVariable(&import_context, import_request, &import_reply);
 }
 
-void SchedulerService::export_all_functions_to_worker(WorkerId workerid, SynchronizedPtr<std::vector<WorkerHandle> > &workers, const SynchronizedPtr<std::vector<std::unique_ptr<Function> > > &exported_functions) {
+void SchedulerService::export_all_functions_to_worker(WorkerId workerid, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<Function> > > &exported_functions) {
   for (int i = 0; i < exported_functions->size(); ++i) {
     export_function_to_worker(workerid, i, workers, exported_functions);
   }
 }
 
-void SchedulerService::export_all_reusable_variables_to_worker(WorkerId workerid, SynchronizedPtr<std::vector<WorkerHandle> > &workers, const SynchronizedPtr<std::vector<std::unique_ptr<ReusableVar> > > &exported_reusable_variables) {
+void SchedulerService::export_all_reusable_variables_to_worker(WorkerId workerid, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<ReusableVar> > > &exported_reusable_variables) {
   for (int i = 0; i < exported_reusable_variables->size(); ++i) {
     export_reusable_variable_to_worker(workerid, i, workers, exported_reusable_variables);
   }
 }
 
-// This method defines the order in which locks should be acquired.
-void SchedulerService::do_on_locks(bool lock) {
-  std::mutex *mutexes[] = {
-    &successful_tasks_.mutex(),
-    &failed_tasks_.mutex(),
-    &get_queue_.mutex(),
-    &computation_graph_.mutex(),
-    &fntable_.mutex(),
-    &avail_workers_.mutex(),
-    &task_queue_.mutex(),
-    &reference_counts_.mutex(),
-    &contained_objrefs_.mutex(),
-    &workers_.mutex(),
-    &alias_notification_queue_.mutex(),
-    &objtable_.mutex(),
-    &objstores_.mutex(),
-    &target_objrefs_.mutex(),
-    &reverse_target_objrefs_.mutex(),
-    &exported_functions_.mutex(),
-    &exported_reusable_variables_.mutex(),
-  };
-  size_t n = sizeof(mutexes) / sizeof(*mutexes);
-  for (size_t i = 0; i != n; ++i) {
-    if (lock) {
-      mutexes[i]->lock();
-    } else {
-      mutexes[n - i - 1]->unlock();
-    }
-  }
-}
-
-void SchedulerService::acquire_all_locks() {
-  do_on_locks(true);
-}
-
-void SchedulerService::release_all_locks() {
-  do_on_locks(false);
-}
-
 void start_scheduler_service(const char* service_addr, SchedulingAlgorithmType scheduling_algorithm) {
   std::string service_address(service_addr);
   std::string::iterator split_point = split_ip_address(service_address);

src/scheduler.h

@@ -77,6 +77,16 @@ public:
   Status ExportFunction(ServerContext* context, const ExportFunctionRequest* request, ExportFunctionReply* reply) override;
   Status ExportReusableVariable(ServerContext* context, const ExportReusableVariableRequest* request, AckReply* reply) override;
 
+#ifdef NDEBUG
+  // If we've disabled assertions, then just use regular SynchronizedPtr to skip lock checking.
+  template<class T>
+  using MySynchronizedPtr = SynchronizedPtr<T>;
+#else
+  // A SynchronizedPtr specialized for this class to dynamically check that locks are obtained in the correct order (in the order of field declarations).
+  template<class T>
+  class MySynchronizedPtr;
+#endif
+
   // This will ask an object store to send an object to another object store if
   // the object is not already present in that object store and is not already
   // being transmitted.
@@ -86,7 +96,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<ComputationGraph> &computation_graph);
+  void assign_task(OperationId operationid, WorkerId workerid, const MySynchronizedPtr<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)
@@ -122,46 +132,74 @@ 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<std::vector<RefCount> > &reference_counts, const SynchronizedPtr<std::vector<std::vector<ObjRef> > > &contained_objrefs);
+  void deallocate_object(ObjRef canonical_objref, const MySynchronizedPtr<std::vector<RefCount> > &reference_counts, const MySynchronizedPtr<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<std::vector<RefCount> > &reference_count);
+  void increment_ref_count(const std::vector<ObjRef> &objrefs, const MySynchronizedPtr<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<std::vector<RefCount> > &reference_count, const SynchronizedPtr<std::vector<std::vector<ObjRef> > > &contained_objrefs);
+  void decrement_ref_count(const std::vector<ObjRef> &objrefs, const MySynchronizedPtr<std::vector<RefCount> > &reference_count, const MySynchronizedPtr<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<std::vector<std::vector<ObjRef> > > &reverse_target_objrefs);
+  void upstream_objrefs(ObjRef objref, std::vector<ObjRef> &objrefs, const MySynchronizedPtr<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);
   // Export a remote function to a worker.
-  void export_function_to_worker(WorkerId workerid, int function_index, SynchronizedPtr<std::vector<WorkerHandle> > &workers, const SynchronizedPtr<std::vector<std::unique_ptr<Function> > > &exported_functions);
+  void export_function_to_worker(WorkerId workerid, int function_index, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<Function> > > &exported_functions);
   // Export a reusable variable to a worker
-  void export_reusable_variable_to_worker(WorkerId workerid, int reusable_variable_index, SynchronizedPtr<std::vector<WorkerHandle> > &workers, const SynchronizedPtr<std::vector<std::unique_ptr<ReusableVar> > > &exported_reusable_variables);
+  void export_reusable_variable_to_worker(WorkerId workerid, int reusable_variable_index, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<ReusableVar> > > &exported_reusable_variables);
   // Export all reusable variables to a worker. This is used when a new worker
   // registers and is protected by the workers lock (which is passed in) to
   // ensure that no other reusable variables are exported to the worker while
   // this method is being called.
-  void export_all_functions_to_worker(WorkerId workerid, SynchronizedPtr<std::vector<WorkerHandle> > &workers, const SynchronizedPtr<std::vector<std::unique_ptr<Function> > > &exported_functions);
+  void export_all_functions_to_worker(WorkerId workerid, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<Function> > > &exported_functions);
   // Export all remote functions to a worker. This is used when a new worker
   // registers and is protected by the workers lock (which is passed in) to
   // ensure that no other remote functions are exported to the worker while this
   // method is being called.
-  void export_all_reusable_variables_to_worker(WorkerId workerid, SynchronizedPtr<std::vector<WorkerHandle> > &workers, const SynchronizedPtr<std::vector<std::unique_ptr<ReusableVar> > > &exported_reusable_variables);
-  // acquires all locks, this should only be used by get_info and for fault tolerance
-  void acquire_all_locks();
-  // release all locks, this should only be used by get_info and for fault tolerance
-  void release_all_locks();
-  // acquire or release all the locks. This is a single method to ensure a single canonical ordering of the locks.
-  void do_on_locks(bool lock);
+  void export_all_reusable_variables_to_worker(WorkerId workerid, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<ReusableVar> > > &exported_reusable_variables);
 
+  template<class T>
+  MySynchronizedPtr<T> get(Synchronized<T>& my_field, const char* name,unsigned int line_number);
+  template<class T>
+  MySynchronizedPtr<const T> get(const Synchronized<T>& my_field, const char* name,unsigned int line_number) const;
+
+  // Preferably keep this as the first field to distinguish it from the rest
+  // Maps every thread to an identifier of a lock it is holding, as well the name of the lock.
+  // Internally, the identifier for each lock is the offset of the field being locked.
+  // When we lock, we set the field offset and store the difference; the difference should always be positive. If not, we throw.
+  // When we unlock, we subtract back the field offset to restore it to the previous field that was locked.
+  mutable Synchronized<std::vector<std::pair<unsigned long long, std::pair<size_t, const char*> > > > lock_orders_;
+
+  // List of the IDs of successful tasks
+  Synchronized<std::vector<OperationId> > successful_tasks_; // Right now, we only use this information in the TaskInfo call.
+  // List of failed tasks
+  Synchronized<std::vector<TaskStatus> > failed_tasks_;
+  // List of pending get calls.
+  Synchronized<std::vector<std::pair<WorkerId, ObjRef> > > get_queue_;
   // The computation graph tracks the operations that have been submitted to the
   // scheduler and is mostly used for fault tolerance.
   Synchronized<ComputationGraph> computation_graph_;
+  // Hash map from function names to workers where the function is registered.
+  Synchronized<FnTable> fntable_;
+  // Vector of all workers that are currently idle.
+  Synchronized<std::vector<WorkerId> > avail_workers_;
+  // List of pending tasks.
+  Synchronized<std::deque<OperationId> > task_queue_;
+  // Reference counts. Currently, reference_counts_[objref] is the number of
+  // existing references held to objref. This is done for all objrefs, not just
+  // canonical_objrefs. This data structure completely ignores aliasing. If the
+  // object corresponding to objref has been deallocated, then
+  // reference_counts[objref] will equal DEALLOCATED.
+  Synchronized<std::vector<RefCount> > reference_counts_;
+  // contained_objrefs_[objref] is a vector of all of the objrefs contained inside the object referred to by objref
+  Synchronized<std::vector<std::vector<ObjRef> > > contained_objrefs_;
   // Vector of all workers registered in the system. Their index in this vector
   // is the workerid.
   Synchronized<std::vector<WorkerHandle> > workers_;
-  // Vector of all workers that are currently idle.
-  Synchronized<std::vector<WorkerId> > avail_workers_;
+  // List of pending alias notifications. Each element consists of (objstoreid, (alias_objref, canonical_objref)).
+  Synchronized<std::vector<std::pair<ObjStoreId, std::pair<ObjRef, ObjRef> > > > alias_notification_queue_;
+  // Mapping from canonical objref to list of object stores where the object is stored. Non-canonical (aliased) objrefs should not be used to index objtable_.
+  Synchronized<ObjTable> objtable_; // This lock protects objtable_ and objects_in_transit_
   // Vector of all object stores registered in the system. Their index in this
   // vector is the objstoreid.
   Synchronized<std::vector<ObjStoreHandle> > objstores_;
@@ -173,8 +211,6 @@ private:
   Synchronized<std::vector<ObjRef> > target_objrefs_;
   // This data structure maps an objref to all of the objrefs that alias it (there could be multiple such objrefs).
   Synchronized<std::vector<std::vector<ObjRef> > > reverse_target_objrefs_;
-  // Mapping from canonical objref to list of object stores where the object is stored. Non-canonical (aliased) objrefs should not be used to index objtable_.
-  Synchronized<ObjTable> objtable_; // This lock protects objtable_ and objects_in_transit_
   // For each object store objstoreid, objects_in_transit_[objstoreid] is a
   // vector of the canonical object references that are being streamed to that
   // object store but are not yet present. Object references are added to this
@@ -185,26 +221,6 @@ private:
   // lock (objects_lock_). // TODO(rkn): Consider making this part of the
   // objtable data structure.
   std::vector<std::vector<ObjRef> > objects_in_transit_;
-  // Hash map from function names to workers where the function is registered.
-  Synchronized<FnTable> fntable_;
-  // List of pending tasks.
-  Synchronized<std::deque<OperationId> > task_queue_;
-  // List of pending get calls.
-  Synchronized<std::vector<std::pair<WorkerId, ObjRef> > > get_queue_;
-  // List of failed tasks
-  Synchronized<std::vector<TaskStatus> > failed_tasks_;
-  // List of the IDs of successful tasks
-  Synchronized<std::vector<OperationId> > successful_tasks_; // Right now, we only use this information in the TaskInfo call.
-  // List of pending alias notifications. Each element consists of (objstoreid, (alias_objref, canonical_objref)).
-  Synchronized<std::vector<std::pair<ObjStoreId, std::pair<ObjRef, ObjRef> > > > alias_notification_queue_;
-  // Reference counts. Currently, reference_counts_[objref] is the number of
-  // existing references held to objref. This is done for all objrefs, not just
-  // canonical_objrefs. This data structure completely ignores aliasing. If the
-  // object corresponding to objref has been deallocated, then
-  // reference_counts[objref] will equal DEALLOCATED.
-  Synchronized<std::vector<RefCount> > reference_counts_;
-  // contained_objrefs_[objref] is a vector of all of the objrefs contained inside the object referred to by objref
-  Synchronized<std::vector<std::vector<ObjRef> > > contained_objrefs_;
   // All of the remote functions that have been exported to the workers.
   Synchronized<std::vector<std::unique_ptr<Function> > > exported_functions_;
   // All of the reusable variables that have been exported to the workers.

src/utils.h

@@ -4,7 +4,7 @@
 #include <mutex>
 #include <string>
 
-template<class T, class Mutex = std::mutex>
+template<class T = void, class Mutex = std::mutex>
 class Synchronized;
 
 template<class T, class Mutex>
@@ -17,6 +17,7 @@ template<class T, class Mutex> struct SynchronizedSource<const volatile T, Mutex
 
 template<class T>
 class SynchronizedPtr : public std::unique_lock<typename SynchronizedSource<T, void>::type> {
+protected:
   typedef std::unique_lock<typename SynchronizedSource<T, void>::type> base_type;
   // Make these private; they don't make much sense externally...
   using base_type::mutex;
@@ -63,20 +64,28 @@ public:
   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;
+template<class Mutex>
+class Synchronized<void, Mutex> {
   mutable Mutex mutex_;
 public:
   typedef Mutex mutex_type;
-  template<class... U>
-  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(); }
-  mutex_type& mutex() { return mutex_; }
+};
+
+template<class T, class Mutex>
+class Synchronized : public Synchronized<T, void>, public Synchronized<void, Mutex> {
+  typedef Synchronized<T, void> base1_type;
+  typedef Synchronized<void, Mutex> base2_type;
+public:
+  template<class... U>
+  Synchronized(U&&... args) : base1_type(std::forward<U>(args)...), base2_type() { }
+  SynchronizedPtr<T> unchecked_get() { return *this; }
+  SynchronizedPtr<const T> unchecked_get() const { return *this; }
+  void lock() const override { return base2_type::lock(); }
+  void unlock() const override { return base2_type::unlock(); }
+  bool try_lock() const override { return base2_type::try_lock(); }
 };
 
 std::string::iterator split_ip_address(std::string& ip_address);