push/pull -> put/get

src/computation_graph.cc

@@ -16,6 +16,6 @@ const Task& ComputationGraph::get_task(OperationId operationid) {
   RAY_CHECK_NEQ(operationid, ROOT_OPERATION, "ComputationGraph attempting to get_task with operationid == ROOT_OPERATION");
   RAY_CHECK_NEQ(operationid, NO_OPERATION, "ComputationGraph attempting to get_task with operationid == NO_OPERATION");
   RAY_CHECK_LT(operationid, operations_.size(), "ComputationGraph attempting to get_task with operationid " << operationid << ", but operationid >= operations_.size().");
-  RAY_CHECK(operations_[operationid]->has_task(), "Calling get_task with operationid " << operationid << ", but this corresponds to a push not a task.");
+  RAY_CHECK(operations_[operationid]->has_task(), "Calling get_task with operationid " << operationid << ", but this corresponds to a put not a task.");
   return operations_[operationid]->task();
 }

src/computation_graph.h

@@ -19,10 +19,10 @@ public:
   // the new operation. This method takes ownership over operation.
   OperationId add_operation(std::unique_ptr<Operation> operation);
   // Return the task corresponding to a particular OperationId. If operationid
-  // corresponds to a push, then fail.
+  // corresponds to a put, then fail.
   const Task& get_task(OperationId operationid);
 private:
-  // maps an OperationId to the corresponding task or push
+  // maps an OperationId to the corresponding task or put
   std::vector<std::unique_ptr<Operation> > operations_;
   // spawned_operations_[operationid] is a vector of the OperationIds of the
   // operations spawned by the task with OperationId operationid

src/objstore.cc

@@ -7,8 +7,8 @@ const size_t ObjStoreService::CHUNK_SIZE = 8 * 1024;
 
 // this method needs to be protected by a objstore_lock_
 // TODO(rkn): Make sure that we do not in fact need the objstore_lock_. We want multiple deliveries to be able to happen simultaneously.
-void ObjStoreService::pull_data_from(ObjRef objref, ObjStore::Stub& stub) {
-  RAY_LOG(RAY_DEBUG, "Objstore " << objstoreid_ << " is beginning to pull objref " << objref);
+void ObjStoreService::get_data_from(ObjRef objref, ObjStore::Stub& stub) {
+  RAY_LOG(RAY_DEBUG, "Objstore " << objstoreid_ << " is beginning to get objref " << objref);
   ObjChunk chunk;
   ClientContext context;
   StreamObjToRequest stream_request;
@@ -76,7 +76,7 @@ Status ObjStoreService::StartDelivery(ServerContext* context, const StartDeliver
     }
     else {
       RAY_CHECK_NEQ(memory_[objref].second, MemoryStatusType::DEALLOCATED, "Objstore " << objstoreid_ << " is attempting to get objref " << objref << ", but memory_[objref] == DEALLOCATED.");
-      RAY_LOG(RAY_DEBUG, "Objstore " << objstoreid_ << " already has objref " << objref << " or it is already being shipped, so no need to pull it again.");
+      RAY_LOG(RAY_DEBUG, "Objstore " << objstoreid_ << " already has objref " << objref << " or it is already being shipped, so no need to get it again.");
       return Status::OK;
     }
     memory_[objref].second = MemoryStatusType::PRE_ALLOCED;
@@ -84,7 +84,7 @@ Status ObjStoreService::StartDelivery(ServerContext* context, const StartDeliver
   delivery_threads_.push_back(std::make_shared<std::thread>([this, address, objref]() {
     std::lock_guard<std::mutex> objstores_lock(objstores_lock_);
     ObjStore::Stub& stub = get_objstore_stub(address);
-    pull_data_from(objref, stub);
+    get_data_from(objref, stub);
   }));
   return Status::OK;
 }
@@ -173,14 +173,14 @@ Status ObjStoreService::DeallocateObject(ServerContext* context, const Deallocat
 // -------------+-------------+------------------+----------------------------
 // NOT_PRESENT  | ALLOC       | NOT_READY        | allocate object
 // NOT_READY    | WORKER_DONE | READY            | send ObjReady to scheduler
-// NOT_READY    | GET         | NOT_READY        | add to pull queue
+// NOT_READY    | GET         | NOT_READY        | add to get queue
 // READY        | GET         | READY            | return handle
 // READY        | DEALLOC     | DEALLOCATED      | deallocate
 // -------------+-------------+------------------+----------------------------
 void ObjStoreService::process_objstore_request(const ObjRequest request) {
   switch (request.type) {
     case ObjRequestType::ALIAS_DONE: {
-        process_pulls_for_objref(request.objref);
+        process_gets_for_objref(request.objref);
       }
       break;
     default: {
@@ -216,8 +216,8 @@ void ObjStoreService::process_worker_request(const ObjRequest request) {
           RAY_LOG(RAY_DEBUG, "Responding to GET request: returning objref " << request.objref);
           send_queues_[request.workerid].send(&item.first);
         } else if (item.second == MemoryStatusType::NOT_READY || item.second == MemoryStatusType::NOT_PRESENT || item.second == MemoryStatusType::PRE_ALLOCED) {
-          std::lock_guard<std::mutex> lock(pull_queue_lock_);
-          pull_queue_.push_back(std::make_pair(request.workerid, request.objref));
+          std::lock_guard<std::mutex> lock(get_queue_lock_);
+          get_queue_.push_back(std::make_pair(request.workerid, request.objref));
         } else {
           RAY_CHECK(false, "A worker requested objref " << request.objref << ", but memory_[objref].second = " << memory_[request.objref].second);
         }
@@ -265,16 +265,16 @@ void ObjStoreService::process_requests() {
   }
 }
 
-void ObjStoreService::process_pulls_for_objref(ObjRef objref) {
+void ObjStoreService::process_gets_for_objref(ObjRef objref) {
   std::pair<ObjHandle, MemoryStatusType>& item = memory_[objref];
-  std::lock_guard<std::mutex> pull_queue_lock(pull_queue_lock_);
-  for (size_t i = 0; i < pull_queue_.size(); ++i) {
-    if (pull_queue_[i].second == objref) {
+  std::lock_guard<std::mutex> get_queue_lock(get_queue_lock_);
+  for (size_t i = 0; i < get_queue_.size(); ++i) {
+    if (get_queue_[i].second == objref) {
       ObjHandle& elem = memory_[objref].first;
-      send_queues_[pull_queue_[i].first].send(&item.first);
-      // Remove the pull task from the queue
-      std::swap(pull_queue_[i], pull_queue_[pull_queue_.size() - 1]);
-      pull_queue_.pop_back();
+      send_queues_[get_queue_[i].first].send(&item.first);
+      // Remove the get task from the queue
+      std::swap(get_queue_[i], get_queue_[get_queue_.size() - 1]);
+      get_queue_.pop_back();
       i -= 1;
     }
   }
@@ -300,7 +300,7 @@ void ObjStoreService::object_ready(ObjRef objref, size_t metadata_offset) {
     item.first.set_metadata_offset(metadata_offset);
     item.second = MemoryStatusType::READY;
   }
-  process_pulls_for_objref(objref);
+  process_gets_for_objref(objref);
   // Tell the scheduler that the object arrived
   // TODO(pcm): put this in a separate thread so we don't have to pay the latency here
   ClientContext objready_context;

src/objstore.h

@@ -46,13 +46,13 @@ public:
   Status ObjStoreInfo(ServerContext* context, const ObjStoreInfoRequest* request, ObjStoreInfoReply* reply) override;
   void start_objstore_service();
 private:
-  void pull_data_from(ObjRef objref, ObjStore::Stub& stub);
+  void get_data_from(ObjRef objref, ObjStore::Stub& stub);
   // check if we already connected to the other objstore, if yes, return reference to connection, otherwise connect
   ObjStore::Stub& get_objstore_stub(const std::string& objstore_address);
   void process_worker_request(const ObjRequest request);
   void process_objstore_request(const ObjRequest request);
   void process_requests();
-  void process_pulls_for_objref(ObjRef objref);
+  void process_gets_for_objref(ObjRef objref);
   ObjHandle alloc(ObjRef objref, size_t size);
   void object_ready(ObjRef objref, size_t metadata_offset);
 
@@ -66,8 +66,8 @@ private:
   std::unordered_map<std::string, std::unique_ptr<ObjStore::Stub>> objstores_;
   std::mutex objstores_lock_;
   std::unique_ptr<Scheduler::Stub> scheduler_stub_;
-  std::vector<std::pair<WorkerId, ObjRef> > pull_queue_;
-  std::mutex pull_queue_lock_;
+  std::vector<std::pair<WorkerId, ObjRef> > get_queue_;
+  std::mutex get_queue_lock_;
   MessageQueue<ObjRequest> recv_queue_; // This queue is used by workers to send tasks to the object store.
   std::vector<MessageQueue<ObjHandle> > send_queues_; // This maps workerid -> queue. The object store uses these queues to send replies to the relevant workers.
   std::thread communicator_thread_;

src/scheduler.cc

@@ -57,7 +57,7 @@ Status SchedulerService::SubmitTask(ServerContext* context, const SubmitTaskRequ
   return Status::OK;
 }
 
-Status SchedulerService::PushObj(ServerContext* context, const PushObjRequest* request, PushObjReply* reply) {
+Status SchedulerService::PutObj(ServerContext* context, const PutObjRequest* request, PutObjReply* reply) {
   ObjRef objref = register_new_object();
   ObjStoreId objstoreid = get_store(request->workerid());
   reply->set_objref(objref);
@@ -74,8 +74,8 @@ Status SchedulerService::RequestObj(ServerContext* context, const RequestObjRequ
   ObjRef objref = request->objref();
   RAY_CHECK_LT(objref, size, "internal error: no object with objref " << objref << " exists");
   {
-    std::lock_guard<std::mutex> pull_queue_lock(pull_queue_lock_);
-    pull_queue_.push_back(std::make_pair(request->workerid(), objref));
+    std::lock_guard<std::mutex> get_queue_lock(get_queue_lock_);
+    get_queue_.push_back(std::make_pair(request->workerid(), objref));
   }
   schedule();
   return Status::OK;
@@ -313,7 +313,7 @@ void SchedulerService::deliver_object(ObjRef canonical_objref, ObjStoreId from,
 
 void SchedulerService::schedule() {
   // TODO(rkn): Do this more intelligently.
-  perform_pulls(); // See what we can do in pull_queue_
+  perform_gets(); // See what we can do in get_queue_
   if (scheduling_algorithm_ == SCHEDULING_ALGORITHM_NAIVE) {
     schedule_tasks_naively(); // See what we can do in task_queue_
   } else if (scheduling_algorithm_ == SCHEDULING_ALGORITHM_LOCALITY_AWARE) {
@@ -513,20 +513,20 @@ bool SchedulerService::is_canonical(ObjRef objref) {
   return objref == target_objrefs_[objref];
 }
 
-void SchedulerService::perform_pulls() {
-  std::lock_guard<std::mutex> pull_queue_lock(pull_queue_lock_);
-  // Complete all pull tasks that can be completed.
-  for (int i = 0; i < pull_queue_.size(); ++i) {
-    const std::pair<WorkerId, ObjRef>& pull = pull_queue_[i];
-    ObjRef objref = pull.second;
-    WorkerId workerid = pull.first;
+void SchedulerService::perform_gets() {
+  std::lock_guard<std::mutex> get_queue_lock(get_queue_lock_);
+  // 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;
     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 pull objref " << pull.second << " with canonical objref " << canonical_objref << " to objstore " << get_store(workerid));
+    RAY_LOG(RAY_DEBUG, "attempting to get objref " << get.second << " with canonical objref " << canonical_objref << " to objstore " << get_store(workerid));
     int num_stores;
     {
       std::lock_guard<std::mutex> objects_lock(objects_lock_);
@@ -539,9 +539,9 @@ void SchedulerService::perform_pulls() {
         std::lock_guard<std::mutex> alias_notification_queue_lock(alias_notification_queue_lock_);
         alias_notification_queue_.push_back(std::make_pair(get_store(workerid), std::make_pair(objref, canonical_objref)));
       }
-      // Remove the pull task from the queue
-      std::swap(pull_queue_[i], pull_queue_[pull_queue_.size() - 1]);
-      pull_queue_.pop_back();
+      // Remove the get task from the queue
+      std::swap(get_queue_[i], get_queue_[get_queue_.size() - 1]);
+      get_queue_.pop_back();
       i -= 1;
     }
   }
@@ -782,7 +782,7 @@ void SchedulerService::do_on_locks(bool lock) {
   std::mutex *mutexes[] = {
     &successful_tasks_lock_,
     &failed_tasks_lock_,
-    &pull_queue_lock_,
+    &get_queue_lock_,
     &computation_graph_lock_,
     &fntable_lock_,
     &avail_workers_lock_,

src/scheduler.h

@@ -55,7 +55,7 @@ public:
   SchedulerService(SchedulingAlgorithmType scheduling_algorithm);
 
   Status SubmitTask(ServerContext* context, const SubmitTaskRequest* request, SubmitTaskReply* reply) override;
-  Status PushObj(ServerContext* context, const PushObjRequest* request, PushObjReply* reply) override;
+  Status PutObj(ServerContext* context, const PutObjRequest* request, PutObjReply* reply) override;
   Status RequestObj(ServerContext* context, const RequestObjRequest* request, AckReply* reply) override;
   Status AliasObjRefs(ServerContext* context, const AliasObjRefsRequest* request, AckReply* reply) override;
   Status RegisterObjStore(ServerContext* context, const RegisterObjStoreRequest* request, RegisterObjStoreReply* reply) override;
@@ -101,7 +101,7 @@ private:
   // checks if objref is a canonical objref
   bool is_canonical(ObjRef objref);
 
-  void perform_pulls();
+  void perform_gets();
   // schedule tasks using the naive algorithm
   void schedule_tasks_naively();
   // schedule tasks using a scheduling algorithm that takes into account data locality
@@ -174,9 +174,9 @@ private:
   // List of pending tasks.
   std::deque<OperationId> task_queue_;
   std::mutex task_queue_lock_;
-  // List of pending pull calls.
-  std::vector<std::pair<WorkerId, ObjRef> > pull_queue_;
-  std::mutex pull_queue_lock_;
+  // List of pending get calls.
+  std::vector<std::pair<WorkerId, ObjRef> > get_queue_;
+  std::mutex get_queue_lock_;
   // List of failed tasks
   std::vector<TaskStatus> failed_tasks_;
   std::mutex failed_tasks_lock_;

src/worker.cc

@@ -73,11 +73,11 @@ void Worker::request_object(ObjRef objref) {
 ObjRef Worker::get_objref() {
   // first get objref for the new object
   RAY_CHECK(connected_, "Attempted to perform get_objref but failed.");
-  PushObjRequest push_request;
-  PushObjReply push_reply;
-  ClientContext push_context;
-  Status push_status = scheduler_stub_->PushObj(&push_context, push_request, &push_reply);
-  return push_reply.objref();
+  PutObjRequest request;
+  PutObjReply reply;
+  ClientContext context;
+  Status status = scheduler_stub_->PutObj(&context, request, &reply);
+  return reply.objref();
 }
 
 slice Worker::get_object(ObjRef objref) {