From 247a4d022ac4ca0a10553616b7c806c4429ed355 Mon Sep 17 00:00:00 2001 From: chaokunyang Date: Mon, 10 Feb 2020 12:07:29 +0800 Subject: [PATCH] Fix passing empty bytes in python tasks (#7045) * ensure data_ won't be null_ptr when size == 0 * when data_sizes[i] == 0, we should Allocate an empty buffer * work around for pyarrow.py_buffer * fix comments * add null ptr check * add test for bytes * lint --- python/ray/_raylet.pyx | 11 +++++++---- python/ray/tests/test_basic.py | 2 ++ src/ray/core_worker/core_worker.cc | 3 ++- .../core_worker/transport/direct_actor_transport.cc | 2 +- 4 files changed, 12 insertions(+), 6 deletions(-) diff --git a/python/ray/_raylet.pyx b/python/ray/_raylet.pyx index 02ea735e6..513aa903e 100644 --- a/python/ray/_raylet.pyx +++ b/python/ray/_raylet.pyx @@ -566,10 +566,13 @@ cdef write_serialized_object( from ray.serialization import Pickle5SerializedObject, RawSerializedObject if isinstance(serialized_object, RawSerializedObject): - buffer = Buffer.make(buf) - stream = pyarrow.FixedSizeBufferWriter(pyarrow.py_buffer(buffer)) - stream.set_memcopy_threads(MEMCOPY_THREADS) - stream.write(pyarrow.py_buffer(serialized_object.value)) + if buf.get() != NULL and buf.get().Size() > 0: + buffer = Buffer.make(buf) + # `Buffer` has a nullptr buffer underlying if size is 0, + # which will cause `pyarrow.py_buffer` crash + stream = pyarrow.FixedSizeBufferWriter(pyarrow.py_buffer(buffer)) + stream.set_memcopy_threads(MEMCOPY_THREADS) + stream.write(pyarrow.py_buffer(serialized_object.value)) elif isinstance(serialized_object, Pickle5SerializedObject): (serialized_object.writer).write_to( serialized_object.inband, buf, MEMCOPY_THREADS) diff --git a/python/ray/tests/test_basic.py b/python/ray/tests/test_basic.py index 0b5d027d4..f477092f5 100644 --- a/python/ray/tests/test_basic.py +++ b/python/ray/tests/test_basic.py @@ -43,6 +43,8 @@ def test_simple_serialization(ray_start_regular): 0.9, 1 << 62, 1 << 999, + b"", + b"a", "a", string.printable, "\u262F", diff --git a/src/ray/core_worker/core_worker.cc b/src/ray/core_worker/core_worker.cc index 9124378ea..dfb7dfc53 100644 --- a/src/ray/core_worker/core_worker.cc +++ b/src/ray/core_worker/core_worker.cc @@ -967,7 +967,8 @@ Status CoreWorker::ExecuteTask(const TaskSpecification &task_spec, if (!return_objects->at(i)) { continue; } - if (return_objects->at(i)->GetData()->IsPlasmaBuffer()) { + if (return_objects->at(i)->GetData() != nullptr && + return_objects->at(i)->GetData()->IsPlasmaBuffer()) { if (!Seal(return_ids[i], /*pin_object=*/false).ok()) { RAY_LOG(FATAL) << "Task " << task_spec.TaskId() << " failed to seal object " << return_ids[i] << " in store: " << status.message(); diff --git a/src/ray/core_worker/transport/direct_actor_transport.cc b/src/ray/core_worker/transport/direct_actor_transport.cc index 436bff16c..eee1a7270 100644 --- a/src/ray/core_worker/transport/direct_actor_transport.cc +++ b/src/ray/core_worker/transport/direct_actor_transport.cc @@ -265,7 +265,7 @@ void CoreWorkerDirectTaskReceiver::HandlePushTask( // The object is nullptr if it already existed in the object store. const auto &result = return_objects[i]; - if (result == nullptr || result->GetData()->IsPlasmaBuffer()) { + if (result->GetData() != nullptr && result->GetData()->IsPlasmaBuffer()) { return_object->set_in_plasma(true); plasma_return_ids.push_back(id); } else {