Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Priority based DFS #2

Closed
wants to merge 1 commit into from
Closed

Priority based DFS #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions python/ray/_raylet.pyx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -303,8 +303,8 @@ cdef void prepare_args(
else:
args_vector.push_back(
CTaskArg.PassByReference(
(<ObjectID>core_worker.put_serialized_object(
serialized_arg)).native()))
(CObjectID.FromBinary(core_worker.put_serialized_cobject(
serialized_arg)))))

cdef deserialize_args(
const c_vector[shared_ptr[CRayObject]] &c_args,
Expand Down Expand Up @@ -689,6 +689,11 @@ cdef class CoreWorker:
def put_serialized_object(self, serialized_object,
ObjectID object_id=None,
c_bool pin_object=True):
return ObjectID(self.put_serialized_cobject(serialized_object, object_id, pin_object))

def put_serialized_cobject(self, serialized_object,
ObjectID object_id=None,
c_bool pin_object=True):
cdef:
CObjectID c_object_id
shared_ptr[CBuffer] data
Expand All @@ -710,7 +715,7 @@ cdef class CoreWorker:
self.core_worker.get().Seal(
c_object_id, pin_object and object_id is None))

return ObjectID(c_object_id.Binary())
return c_object_id.Binary()

def wait(self, object_ids, int num_returns, int64_t timeout_ms,
TaskID current_task_id):
Expand Down
6 changes: 1 addition & 5 deletions python/ray/tests/test_reference_counting.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,17 +195,13 @@ def test_pending_task_dependency_pinning(shutdown_only):
def pending(input1, input2):
return

@ray.remote
def slow(dep):
pass

# The object that is ray.put here will go out of scope immediately, so if
# pending task dependencies aren't considered, it will be evicted before
# the ray.get below due to the subsequent ray.puts that fill up the object
# store.
np_array = np.zeros(40 * 1024 * 1024, dtype=np.uint8)
random_id = ray.ObjectID.from_random()
oid = pending.remote(np_array, slow.remote(random_id))
oid = pending.remote(np_array, random_id)

for _ in range(2):
ray.put(np_array)
Expand Down