From e0a340ee7efcad97b73924cfe651876b7ccf5493 Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Wed, 22 Nov 2017 13:38:01 -0800 Subject: [PATCH] Allow actors to pin at most 1000 dummy objects at a time. (#1241) * Allow actors to pin at most 1000 dummy objects at a time. * Fix linting. --- python/ray/actor.py | 5 ++++- python/ray/worker.py | 2 +- src/common/lib/python/config_extension.cc | 7 +++++++ src/common/lib/python/config_extension.h | 1 + src/common/state/ray_config.h | 7 +++++++ 5 files changed, 20 insertions(+), 2 deletions(-) diff --git a/python/ray/actor.py b/python/ray/actor.py index 2facccede..06cdc3f94 100644 --- a/python/ray/actor.py +++ b/python/ray/actor.py @@ -134,7 +134,10 @@ def put_dummy_object(worker, dummy_object_id): # actor, to prevent eviction from the object store. dummy_object = worker.get_object([dummy_object_id]) dummy_object = dummy_object[0] - worker.actor_pinned_objects[dummy_object_id] = dummy_object + worker.actor_pinned_objects.append(dummy_object) + if (len(worker.actor_pinned_objects) > + ray._config.actor_max_dummy_objects()): + worker.actor_pinned_objects.pop(0) def make_actor_method_executor(worker, method_name, method): diff --git a/python/ray/worker.py b/python/ray/worker.py index 926dc0730..e987ecbb7 100644 --- a/python/ray/worker.py +++ b/python/ray/worker.py @@ -1836,7 +1836,7 @@ def connect(info, object_id_seed=None, mode=WORKER_MODE, worker=global_worker, worker.class_id = class_id # Store a list of the dummy outputs produced by actor tasks, to pin the # dummy outputs in the object store. - worker.actor_pinned_objects = {} + worker.actor_pinned_objects = [] # Initialize the serialization library. This registers some classes, and so # it must be run before we export all of the cached remote functions. diff --git a/src/common/lib/python/config_extension.cc b/src/common/lib/python/config_extension.cc index 131443430..05508ef28 100644 --- a/src/common/lib/python/config_extension.cc +++ b/src/common/lib/python/config_extension.cc @@ -35,6 +35,10 @@ PyObject *PyRayConfig_worker_fetch_request_size(PyObject *self) { return PyLong_FromLongLong(RayConfig::instance().worker_fetch_request_size()); } +PyObject *PyRayConfig_actor_max_dummy_objects(PyObject *self) { + return PyLong_FromLongLong(RayConfig::instance().actor_max_dummy_objects()); +} + PyObject *PyRayConfig_num_connect_attempts(PyObject *self) { return PyLong_FromLongLong(RayConfig::instance().num_connect_attempts()); } @@ -145,6 +149,9 @@ static PyMethodDef PyRayConfig_methods[] = { {"worker_fetch_request_size", (PyCFunction) PyRayConfig_worker_fetch_request_size, METH_NOARGS, "Return worker_fetch_request_size"}, + {"actor_max_dummy_objects", + (PyCFunction) PyRayConfig_actor_max_dummy_objects, METH_NOARGS, + "Return actor_max_dummy_objects"}, {"num_connect_attempts", (PyCFunction) PyRayConfig_num_connect_attempts, METH_NOARGS, "Return num_connect_attempts"}, {"connect_timeout_milliseconds", diff --git a/src/common/lib/python/config_extension.h b/src/common/lib/python/config_extension.h index 3dc3b13de..746be54d1 100644 --- a/src/common/lib/python/config_extension.h +++ b/src/common/lib/python/config_extension.h @@ -22,6 +22,7 @@ PyObject *PyRayConfig_num_heartbeats_timeout(PyObject *self); PyObject *PyRayConfig_get_timeout_milliseconds(PyObject *self); PyObject *PyRayConfig_worker_get_request_size(PyObject *self); PyObject *PyRayConfig_worker_fetch_request_size(PyObject *self); +PyObject *PyRayConfig_actor_max_dummy_objects(PyObject *self); PyObject *PyRayConfig_num_connect_attempts(PyObject *self); PyObject *PyRayConfig_connect_timeout_milliseconds(PyObject *self); PyObject *PyRayConfig_local_scheduler_fetch_timeout_milliseconds( diff --git a/src/common/state/ray_config.h b/src/common/state/ray_config.h index fb8cb23b8..58e710149 100644 --- a/src/common/state/ray_config.h +++ b/src/common/state/ray_config.h @@ -27,6 +27,8 @@ class RayConfig { return worker_fetch_request_size_; } + int64_t actor_max_dummy_objects() const { return actor_max_dummy_objects_; } + int64_t num_connect_attempts() const { return num_connect_attempts_; } int64_t connect_timeout_milliseconds() const { @@ -95,6 +97,7 @@ class RayConfig { get_timeout_milliseconds_(1000), worker_get_request_size_(10000), worker_fetch_request_size_(10000), + actor_max_dummy_objects_(1000), num_connect_attempts_(50), connect_timeout_milliseconds_(100), local_scheduler_fetch_timeout_milliseconds_(1000), @@ -135,6 +138,10 @@ class RayConfig { int64_t worker_get_request_size_; int64_t worker_fetch_request_size_; + /// This is a temporary constant used by actors to determine how many dummy + /// objects to store. + int64_t actor_max_dummy_objects_; + /// Number of times we try connecting to a socket. int64_t num_connect_attempts_; int64_t connect_timeout_milliseconds_;