[xray] Use pubsub instead of timeout for ObjectManager Pull. #2079

elibol · 2018-05-16T23:29:48Z

This PR removes the ObjectManager timeout for Pull. The ObjectDirectory interface method Lookup has been removed and replaced with methods Subscribe and Unsubscribe, which implement the existing GCS Lookup call followed by RequestNotifications whenever GCS Lookup fails to obtain object locations. This implementation eliminates the need to poll the GCS for object locations, which improves the performance of Pull.

robertnishihara · 2018-05-16T23:52:27Z

src/ray/object_manager/object_directory.cc

+  ray::Status status = ray::Status::OK();
+  if (listeners_.find(object_id) != listeners_.end()) {
+    RAY_LOG(ERROR) << "Duplicate calls to SubscribeObjectLocations for " << object_id;
+    return ray::Status::Invalid("Cannot do things.");


I don't understand this message.

AmplabJenkins · 2018-05-17T00:02:16Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5439/
Test FAILed.

AmplabJenkins · 2018-05-17T01:11:42Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5442/
Test PASSed.

eric-jj · 2018-05-17T01:53:25Z

src/ray/object_manager/object_directory.cc

+void ObjectDirectory::RegisterBackend() {
+  auto object_notification_callback = [this](gcs::AsyncGcsClient *client,
+                                             const ObjectID &object_id,
+                                             const std::vector<ObjectTableDataT> data) {


const std::vector& data？

eric-jj · 2018-05-17T01:54:15Z

src/ray/object_manager/object_directory.cc

+    }
+    // Obtain reported client ids.
+    std::vector<ClientID> client_ids;
+    for (auto item : data) {


for (auto& item ：data)

eric-jj · 2018-05-17T01:59:54Z

src/ray/object_manager/object_directory.cc

+                             const std::vector<ObjectTableDataT> &location_entries) {
+        // Build the set of current locations based on the entries in the log.
+        std::unordered_set<ClientID> locations;
+        for (auto entry : location_entries) {


for (auto& entry : location_entries) {

stephanie-wang

In general, let's try to think carefully about the places where we return ray::Status and what it means if we return a status that is not OK. If there are methods where we should always return OK, I think it's better to just return void and add fatal checks for non-OK situations. If there are places where we should return a status, we should document in the method what it means if the status is not OK.

stephanie-wang · 2018-05-17T18:40:19Z

src/ray/object_manager/object_directory.cc

+      if (!item.is_eviction) {
+        ClientID client_id = ClientID::from_binary(item.manager);
+        client_ids.push_back(client_id);
+      }


Since the object table is a log, you have to apply the eviction entries as well. Sorry there's no documentation for this, but for example, if an object was evicted on node A, then added again, then its log will look like: add A, evict A, add A. The easiest way to apply the entries is probably to build up an unordered_set of client IDs instead of a vector.

Actually, it would be great if you could add that documentation to gcs/tables.h!

stephanie-wang · 2018-05-17T18:49:11Z

src/ray/object_manager/object_directory.cc

+                             const std::vector<ObjectTableDataT> &location_entries) {
+        // Build the set of current locations based on the entries in the log.
+        std::unordered_set<ClientID> locations;
+        for (auto &entry : location_entries) {


This is the logic that you want for the handler for pubsub notifications. I would probably just separate this out as a helper function.

stephanie-wang · 2018-05-17T18:52:15Z

src/ray/object_manager/object_directory.cc

+        // Invoke the callback.
+        std::vector<ClientID> locations_vector(locations.begin(), locations.end());
+        if (locations_vector.empty()) {
+          fail_callback(object_id);


Since GetLocations is now private, can we remove the success and fail callbacks from the signature and just call the callbacks that are registered in SubscribeObjectLocations directly?

stephanie-wang · 2018-05-17T18:54:44Z

src/ray/object_manager/object_directory.h

+  struct LocationListenerState {
+    LocationListenerState(const OnLocationsFound &locations_found_callback)
+        : locations_found_callback(locations_found_callback), listening(false) {}
+    OnLocationsFound locations_found_callback;


Document the struct members.

stephanie-wang · 2018-05-17T18:54:55Z

src/ray/object_manager/object_directory.h

-  /// Invoked when call to ExecuteGetLocations completes.
-  void GetLocationsComplete(const ObjectID &object_id,
-                            const std::vector<ObjectTableDataT> &location_entries);
+  ray::Status GetLocations(const ObjectID &object_id,


stephanie-wang · 2018-05-17T18:59:34Z

src/ray/object_manager/object_manager.cc

@@ -299,8 +285,7 @@ ray::Status ObjectManager::SendObjectData(const ObjectID &object_id,
 }

 ray::Status ObjectManager::Cancel(const ObjectID &object_id) {
-  // TODO(hme): Account for pull timers.
-  ray::Status status = object_directory_->Cancel(object_id);
+  ray::Status status = object_directory_->UnsubscribeObjectLocations(object_id);


This status is unused. Return it?

stephanie-wang · 2018-05-17T19:00:48Z

src/ray/object_manager/object_directory.cc

-    } else {
-      locations.erase(client_id);
-    }
+ray::Status ObjectDirectory::UnsubscribeObjectLocations(const ObjectID &object_id) {


We should be careful about having too many signatures return ray::Status. This one always seems to return OK. Can we either change the signature to void, or have it return the status from the CancelNotifications call?

This now returns status of CancelNotifications.

stephanie-wang · 2018-05-17T19:02:24Z

src/ray/object_manager/object_manager.cc

-        RAY_CHECK_OK(PullGetLocations(object_id));
-      });
-}
-
 ray::Status ObjectManager::PullGetLocations(const ObjectID &object_id) {


PullGetLocations is now only called by the Pull method. Can we squash this method into Pull?

stephanie-wang · 2018-05-17T19:08:51Z

src/ray/object_manager/object_directory.cc

+  ray::Status status = ray::Status::OK();
+  if (listeners_.find(object_id) != listeners_.end()) {
+    RAY_LOG(ERROR) << "Duplicate calls to SubscribeObjectLocations for " << object_id;
+    return ray::Status::Invalid("Unable to subscribe to the same object twice.");


Returning a non-OK error here will break the node manager, which always checks that a Pull returns OK and currently does not check if it has already called Pull on the object before. We should be clear on the contracts with the callers of SubscribeObjectLocations and Pull. For instance, I'm not sure that it makes sense to make the caller of SubscribeObjectLocations keep track of the calls it's made so far.

stephanie-wang · 2018-05-17T19:09:08Z

src/ray/object_manager/object_directory.cc

-                                const std::vector<ObjectTableDataT> &data) {
-        GetLocationsComplete(object_id, data);
+  listeners_.emplace(object_id, LocationListenerState(callback));
+  GetLocations(


GetLocations returns a status. We should probably return that here.

AmplabJenkins · 2018-05-17T19:17:55Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5451/
Test PASSed.

AmplabJenkins · 2018-05-17T23:52:58Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5457/
Test PASSed.

stephanie-wang · 2018-05-18T18:02:34Z

src/ray/object_manager/object_directory.cc

+    std::vector<ClientID> client_id_vec(client_id_set.begin(), client_id_set.end());
+    entry->second.locations_found_callback(client_id_vec, object_id);
+  };
+  gcs_client_->object_table().Subscribe(UniqueID::nil(),


There is an unused return status here. Either return it or RAY_CHECK_OK.

stephanie-wang · 2018-05-18T18:04:27Z

src/ray/object_manager/object_directory.cc

+      }
+    }
+    std::vector<ClientID> client_id_vec(client_id_set.begin(), client_id_set.end());
+    entry->second.locations_found_callback(client_id_vec, object_id);


Do you want to call the callback if client_id_vec is empty?

AmplabJenkins · 2018-05-18T18:38:50Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5481/
Test PASSed.

AmplabJenkins · 2018-05-18T20:40:16Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5482/
Test PASSed.

AmplabJenkins · 2018-05-19T00:59:46Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5486/
Test PASSed.

elibol · 2018-05-19T04:34:45Z

Tests look good: https://travis-ci.com/robertnishihara/ray-private-travis/builds/73884240

* master: (24 commits) Performance fix (ray-project#2110) Use flake8-comprehensions (ray-project#1976) Improve error message printing and suppression. (ray-project#2104) [rllib] [doc] Broken link in ddpg doc YAPF, take 3 (ray-project#2098) [rllib] rename async -> _async (ray-project#2097) fix unused lambda capture (ray-project#2102) [xray] Use pubsub instead of timeout for ObjectManager Pull. (ray-project#2079) [DataFrame] Update _inherit_docstrings (ray-project#2085) [JavaWorker] Changes to the build system for support java worker (ray-project#2092) [xray] Fix bug in updating actor execution dependencies (ray-project#2064) [DataFrame] Refactor __delitem__ (ray-project#2080) [xray] Better error messaging when pulling from self. (ray-project#2068) Use source code in hash where possible (fix ray-project#2089) (ray-project#2090) Functions for flushing done tasks and evicted objects. (ray-project#2033) Fix compilation error for RAY_USE_NEW_GCS with latest clang. (ray-project#2086) [xray] Corrects Error Handling During Push and Pull. (ray-project#2059) [xray] Sophisticated task dependency management (ray-project#2035) Support calling positional arguments by keyword (fix ray-project#998) (ray-project#2081) [DataFrame] Improve performance of iteration methods (ray-project#2026) ...

* fix-a3c-torch: (37 commits) Add missing channel major Use correct filter size Add TODO Fix shape errors fmt Performance fix (ray-project#2110) Use flake8-comprehensions (ray-project#1976) Improve error message printing and suppression. (ray-project#2104) [rllib] [doc] Broken link in ddpg doc YAPF, take 3 (ray-project#2098) [rllib] rename async -> _async (ray-project#2097) fix unused lambda capture (ray-project#2102) [xray] Use pubsub instead of timeout for ObjectManager Pull. (ray-project#2079) [DataFrame] Update _inherit_docstrings (ray-project#2085) [JavaWorker] Changes to the build system for support java worker (ray-project#2092) [xray] Fix bug in updating actor execution dependencies (ray-project#2064) [DataFrame] Refactor __delitem__ (ray-project#2080) [xray] Better error messaging when pulling from self. (ray-project#2068) Use source code in hash where possible (fix ray-project#2089) (ray-project#2090) Functions for flushing done tasks and evicted objects. (ray-project#2033) ...

Use pubsub instead of timeout.

0e18ca7

elibol requested review from pcmoritz, robertnishihara, atumanov and stephanie-wang May 16, 2018 23:29

robertnishihara reviewed May 16, 2018

View reviewed changes

Correct status message.

fc2572c

elibol changed the title ~~Use pubsub instead of timeout for ObjectManager Pull.~~ [xray] Use pubsub instead of timeout for ObjectManager Pull. May 17, 2018

eric-jj reviewed May 17, 2018

View reviewed changes

eric's feedback!

2e8af60

stephanie-wang requested changes May 17, 2018

View reviewed changes

Changes from Stephanie's review.

b57a548

elibol added 2 commits May 18, 2018 10:28

Merge branch 'master' into om_pubsub

fa5c32d

Merge branch 'master' into om_pubsub

0ccf46b

stephanie-wang reviewed May 18, 2018

View reviewed changes

Stephanie round 2 review.

15fe758

elibol added 2 commits May 18, 2018 15:28

copy callback

2268202

check return value

717b7dc

stephanie-wang approved these changes May 18, 2018

View reviewed changes

elibol merged commit f1da721 into ray-project:master May 19, 2018

[xray] Use pubsub instead of timeout for ObjectManager Pull. #2079

[xray] Use pubsub instead of timeout for ObjectManager Pull. #2079

Uh oh!

Conversation

elibol commented May 16, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AmplabJenkins commented May 17, 2018

Uh oh!

AmplabJenkins commented May 17, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stephanie-wang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AmplabJenkins commented May 17, 2018

Uh oh!

AmplabJenkins commented May 17, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AmplabJenkins commented May 18, 2018

Uh oh!

AmplabJenkins commented May 18, 2018

Uh oh!

AmplabJenkins commented May 19, 2018

Uh oh!

elibol commented May 19, 2018

Uh oh!

Uh oh!