Support metadata for passing by value task arguments (#5527)

src/ray/common/buffer.h

@@ -39,6 +39,11 @@ class LocalMemoryBuffer : public Buffer {
  public:
   /// Constructor.
   ///
+  /// By default when initializing a LocalMemoryBuffer with a data pointer and a length,
+  /// it just assigns the pointer and length without coping the data content. This is
+  /// for performance reasons. In this case the buffer cannot ensure data validity. It
+  /// instead relies on the lifetime passed in data pointer.
+  ///
   /// \param data The data pointer to the passed-in buffer.
   /// \param size The size of the passed in buffer.
   /// \param copy_data If true, data will be copied and owned by this buffer,

src/ray/common/ray_object.h

@@ -0,0 +1,77 @@
+#ifndef RAY_COMMON_RAY_OBJECT_H
+#define RAY_COMMON_RAY_OBJECT_H
+
+#include "ray/common/buffer.h"
+#include "ray/util/logging.h"
+
+namespace ray {
+
+/// Binary representation of a ray object, consisting of buffer pointers to data and
+/// metadata. A ray object may have both data and metadata, or only one of them.
+class RayObject {
+ public:
+  /// Create a ray object instance.
+  ///
+  /// Set `copy_data` to `false` is fine for most cases - for example when putting
+  /// an object into store with a temporary RayObject, and we don't want to do an extra
+  /// copy. But in some cases we do want to always hold a valid data - for example, memory
+  /// store uses RayObject to represent objects, in this case we actually want the object
+  /// data to remain valid after user puts it into store.
+  ///
+  /// \param[in] data Data of the ray object.
+  /// \param[in] metadata Metadata of the ray object.
+  /// \param[in] copy_data Whether this class should hold a copy of data.
+  RayObject(const std::shared_ptr<Buffer> &data, const std::shared_ptr<Buffer> &metadata,
+            bool copy_data = false)
+      : data_(data), metadata_(metadata), has_data_copy_(copy_data) {
+    RAY_CHECK(!data || data_->Size())
+        << "Zero-length buffers are not allowed when constructing a RayObject.";
+    RAY_CHECK(!metadata || metadata->Size())
+        << "Zero-length buffers are not allowed when constructing a RayObject.";
+
+    if (has_data_copy_) {
+      // If this object is required to hold a copy of the data,
+      // make a copy if the passed in buffers don't already have a copy.
+      if (data_ && !data_->OwnsData()) {
+        data_ = std::make_shared<LocalMemoryBuffer>(data_->Data(), data_->Size(),
+                                                    /*copy_data=*/true);
+      }
+
+      if (metadata_ && !metadata_->OwnsData()) {
+        metadata_ = std::make_shared<LocalMemoryBuffer>(
+            metadata_->Data(), metadata_->Size(), /*copy_data=*/true);
+      }
+    }
+
+    RAY_CHECK(data_ || metadata_) << "Data and metadata cannot both be empty.";
+  }
+
+  /// Return the data of the ray object.
+  const std::shared_ptr<Buffer> &GetData() const { return data_; };
+
+  /// Return the metadata of the ray object.
+  const std::shared_ptr<Buffer> &GetMetadata() const { return metadata_; };
+
+  uint64_t GetSize() const {
+    uint64_t size = 0;
+    size += (data_ != nullptr) ? data_->Size() : 0;
+    size += (metadata_ != nullptr) ? metadata_->Size() : 0;
+    return size;
+  }
+
+  /// Whether this object has data.
+  bool HasData() const { return data_ != nullptr; }
+
+  /// Whether this object has metadata.
+  bool HasMetadata() const { return metadata_ != nullptr; }
+
+ private:
+  std::shared_ptr<Buffer> data_;
+  std::shared_ptr<Buffer> metadata_;
+  /// Whether this class holds a data copy.
+  bool has_data_copy_;
+};
+
+}  // namespace ray
+
+#endif  // RAY_COMMON_BUFFER_H

src/ray/common/task/task_spec.cc

@@ -53,14 +53,22 @@ ObjectID TaskSpecification::ArgId(size_t arg_index, size_t id_index) const {
   return ObjectID::FromBinary(message_->args(arg_index).object_ids(id_index));
 }
 
-const uint8_t *TaskSpecification::ArgVal(size_t arg_index) const {
+const uint8_t *TaskSpecification::ArgData(size_t arg_index) const {
   return reinterpret_cast<const uint8_t *>(message_->args(arg_index).data().data());
 }
 
-size_t TaskSpecification::ArgValLength(size_t arg_index) const {
+size_t TaskSpecification::ArgDataSize(size_t arg_index) const {
   return message_->args(arg_index).data().size();
 }
 
+const uint8_t *TaskSpecification::ArgMetadata(size_t arg_index) const {
+  return reinterpret_cast<const uint8_t *>(message_->args(arg_index).metadata().data());
+}
+
+size_t TaskSpecification::ArgMetadataSize(size_t arg_index) const {
+  return message_->args(arg_index).metadata().size();
+}
+
 const ResourceSet TaskSpecification::GetRequiredResources() const {
   return required_resources_;
 }

src/ray/common/task/task_spec.h

@@ -70,9 +70,13 @@ class TaskSpecification : public MessageWrapper<rpc::TaskSpec> {
 
   ObjectID ReturnId(size_t return_index) const;
 
-  const uint8_t *ArgVal(size_t arg_index) const;
+  const uint8_t *ArgData(size_t arg_index) const;
 
-  size_t ArgValLength(size_t arg_index) const;
+  size_t ArgDataSize(size_t arg_index) const;
+
+  const uint8_t *ArgMetadata(size_t arg_index) const;
+
+  size_t ArgMetadataSize(size_t arg_index) const;
 
   /// Return the resources that are to be acquired during the execution of this
   /// task.

src/ray/common/task/task_util.h

@@ -1,6 +1,8 @@
 #ifndef RAY_COMMON_TASK_TASK_UTIL_H
 #define RAY_COMMON_TASK_TASK_UTIL_H
 
+#include "ray/common/buffer.h"
+#include "ray/common/ray_object.h"
 #include "ray/common/task/task_spec.h"
 #include "ray/protobuf/common.pb.h"
 
@@ -56,19 +58,29 @@ class TaskSpecBuilder {
   /// Add a by-value argument to the task.
   ///
   /// \param data String object that contains the data.
+  /// \param metadata String object that contains the metadata.
   /// \return Reference to the builder object itself.
-  TaskSpecBuilder &AddByValueArg(const std::string &data) {
-    message_->add_args()->set_data(data);
+  TaskSpecBuilder &AddByValueArg(const std::string &data, const std::string &metadata) {
+    auto arg = message_->add_args();
+    arg->set_data(data);
+    arg->set_metadata(metadata);
     return *this;
   }
 
   /// Add a by-value argument to the task.
   ///
-  /// \param data Pointer to the data.
-  /// \param size Size of the data.
+  /// \param value the RayObject instance that contains the data and the metadata.
   /// \return Reference to the builder object itself.
-  TaskSpecBuilder &AddByValueArg(const void *data, size_t size) {
-    message_->add_args()->set_data(data, size);
+  TaskSpecBuilder &AddByValueArg(const RayObject &value) {
+    auto arg = message_->add_args();
+    if (value.HasData()) {
+      const auto &data = value.GetData();
+      arg->set_data(data->Data(), data->Size());
+    }
+    if (value.HasMetadata()) {
+      const auto &metadata = value.GetMetadata();
+      arg->set_metadata(metadata->Data(), metadata->Size());
+    }
     return *this;
   }
 

src/ray/core_worker/common.h

@@ -3,8 +3,8 @@
 
 #include <string>
 
-#include "ray/common/buffer.h"
 #include "ray/common/id.h"
+#include "ray/common/ray_object.h"
 #include "ray/common/task/task_spec.h"
 #include "ray/raylet/raylet_client.h"
 #include "ray/util/util.h"
@@ -31,12 +31,13 @@ class TaskArg {
     return TaskArg(std::make_shared<ObjectID>(object_id), nullptr);
   }
 
-  /// Create a pass-by-reference task argument.
+  /// Create a pass-by-value task argument.
   ///
-  /// \param[in] object_id Id of the argument.
+  /// \param[in] value Value of the argument.
   /// \return The task argument.
-  static TaskArg PassByValue(const std::shared_ptr<Buffer> &data) {
-    return TaskArg(nullptr, data);
+  static TaskArg PassByValue(const std::shared_ptr<RayObject> &value) {
+    RAY_CHECK(value) << "Value can't be null.";
+    return TaskArg(nullptr, value);
   }
 
   /// Return true if this argument is passed by reference, false if passed by value.
@@ -49,19 +50,19 @@ class TaskArg {
   }
 
   /// Get the value.
-  std::shared_ptr<Buffer> GetValue() const {
-    RAY_CHECK(data_ != nullptr) << "This argument isn't passed by value.";
-    return data_;
+  const RayObject &GetValue() const {
+    RAY_CHECK(value_ != nullptr) << "This argument isn't passed by value.";
+    return *value_;
   }
 
  private:
-  TaskArg(const std::shared_ptr<ObjectID> id, const std::shared_ptr<Buffer> data)
-      : id_(id), data_(data) {}
+  TaskArg(const std::shared_ptr<ObjectID> id, const std::shared_ptr<RayObject> value)
+      : id_(id), value_(value) {}
 
-  /// Id of the argument, if passed by reference, otherwise nullptr.
+  /// Id of the argument if passed by reference, otherwise nullptr.
   const std::shared_ptr<ObjectID> id_;
-  /// Data of the argument, if passed by value, otherwise nullptr.
-  const std::shared_ptr<Buffer> data_;
+  /// Value of the argument if passed by value, otherwise nullptr.
+  const std::shared_ptr<RayObject> value_;
 };
 
 enum class StoreProviderType { PLASMA, MEMORY };

src/ray/core_worker/lib/java/jni_init.cc

@@ -43,7 +43,7 @@ jmethodID java_language_get_number;
 
 jclass java_function_arg_class;
 jfieldID java_function_arg_id;
-jfieldID java_function_arg_data;
+jfieldID java_function_arg_value;
 
 jclass java_base_task_options_class;
 jfieldID java_base_task_options_resources;
@@ -137,7 +137,8 @@ jint JNI_OnLoad(JavaVM *vm, void *reserved) {
   java_function_arg_class = LoadClass(env, "org/ray/runtime/task/FunctionArg");
   java_function_arg_id =
       env->GetFieldID(java_function_arg_class, "id", "Lorg/ray/api/id/ObjectId;");
-  java_function_arg_data = env->GetFieldID(java_function_arg_class, "data", "[B");
+  java_function_arg_value = env->GetFieldID(java_function_arg_class, "value",
+                                            "Lorg/ray/runtime/object/NativeRayObject;");
 
   java_base_task_options_class = LoadClass(env, "org/ray/api/options/BaseTaskOptions");
   java_base_task_options_resources =

src/ray/core_worker/lib/java/jni_utils.h

@@ -4,6 +4,7 @@
 #include <jni.h>
 #include "ray/common/buffer.h"
 #include "ray/common/id.h"
+#include "ray/common/ray_object.h"
 #include "ray/common/status.h"
 #include "ray/core_worker/store_provider/store_provider.h"
 
@@ -77,12 +78,12 @@ extern jclass java_language_class;
 /// getNumber of Language class
 extern jmethodID java_language_get_number;
 
-/// NativeTaskArg class
+/// FunctionArg class
 extern jclass java_function_arg_class;
-/// id field of NativeTaskArg class
+/// id field of FunctionArg class
 extern jfieldID java_function_arg_id;
-/// data field of NativeTaskArg class
-extern jfieldID java_function_arg_data;
+/// value field of FunctionArg class
+extern jfieldID java_function_arg_value;
 
 /// BaseTaskOptions class
 extern jclass java_base_task_options_class;
@@ -279,11 +280,11 @@ inline std::shared_ptr<ray::RayObject> JavaNativeRayObjectToNativeRayObject(
   std::shared_ptr<ray::Buffer> data_buffer = JavaByteArrayToNativeBuffer(env, java_data);
   std::shared_ptr<ray::Buffer> metadata_buffer =
       JavaByteArrayToNativeBuffer(env, java_metadata);
-  if (!data_buffer) {
-    data_buffer = std::make_shared<ray::LocalMemoryBuffer>(nullptr, 0);
+  if (data_buffer && data_buffer->Size() == 0) {
+    data_buffer = nullptr;
   }
-  if (!metadata_buffer) {
-    metadata_buffer = std::make_shared<ray::LocalMemoryBuffer>(nullptr, 0);
+  if (metadata_buffer && metadata_buffer->Size() == 0) {
+    metadata_buffer = nullptr;
   }
   return std::make_shared<ray::RayObject>(data_buffer, metadata_buffer);
 }

src/ray/core_worker/lib/java/org_ray_runtime_task_NativeTaskSubmitter.cc

@@ -34,10 +34,11 @@ inline std::vector<ray::TaskArg> ToTaskArgs(JNIEnv *env, jobject args) {
           return ray::TaskArg::PassByReference(
               JavaByteArrayToId<ray::ObjectID>(env, java_id_bytes));
         }
-        auto java_data =
-            static_cast<jbyteArray>(env->GetObjectField(arg, java_function_arg_data));
-        RAY_CHECK(java_data) << "Both id and data of FunctionArg are null.";
-        return ray::TaskArg::PassByValue(JavaByteArrayToNativeBuffer(env, java_data));
+        auto java_value =
+            static_cast<jbyteArray>(env->GetObjectField(arg, java_function_arg_value));
+        RAY_CHECK(java_value) << "Both id and value of FunctionArg are null.";
+        auto value = JavaNativeRayObjectToNativeRayObject(env, java_value);
+        return ray::TaskArg::PassByValue(value);
       });
   return task_args;
 }

src/ray/core_worker/store_provider/plasma_store_provider.cc

@@ -69,9 +69,15 @@ Status CoreWorkerPlasmaStoreProvider::FetchAndGetFromPlasmaStore(
   for (size_t i = 0; i < plasma_results.size(); i++) {
     if (plasma_results[i].data != nullptr || plasma_results[i].metadata != nullptr) {
       const auto &object_id = batch_ids[i];
-      const auto result_object = std::make_shared<RayObject>(
-          std::make_shared<PlasmaBuffer>(plasma_results[i].data),
-          std::make_shared<PlasmaBuffer>(plasma_results[i].metadata));
+      std::shared_ptr<PlasmaBuffer> data = nullptr;
+      std::shared_ptr<PlasmaBuffer> metadata = nullptr;
+      if (plasma_results[i].data && plasma_results[i].data->size()) {
+        data = std::make_shared<PlasmaBuffer>(plasma_results[i].data);
+      }
+      if (plasma_results[i].metadata && plasma_results[i].metadata->size()) {
+        metadata = std::make_shared<PlasmaBuffer>(plasma_results[i].metadata);
+      }
+      const auto result_object = std::make_shared<RayObject>(data, metadata);
       (*results)[object_id] = result_object;
       remaining.erase(object_id);
       if (IsException(*result_object)) {
@@ -174,6 +180,9 @@ Status CoreWorkerPlasmaStoreProvider::Delete(const std::vector<ObjectID> &object
 
 bool CoreWorkerPlasmaStoreProvider::IsException(const RayObject &object) {
   // TODO (kfstorm): metadata should be structured.
+  if (!object.HasMetadata()) {
+    return false;
+  }
   const std::string metadata(reinterpret_cast<const char *>(object.GetMetadata()->Data()),
                              object.GetMetadata()->Size());
   const auto error_type_descriptor = ray::rpc::ErrorType_descriptor();

src/ray/core_worker/store_provider/store_provider.h

@@ -8,56 +8,6 @@
 
 namespace ray {
 
-/// Binary representation of a ray object.
-class RayObject {
- public:
-  /// Create a ray object instance.
-  ///
-  /// \param[in] data Data of the ray object.
-  /// \param[in] metadata Metadata of the ray object.
-  /// \param[in] copy_data Whether this class should hold a copy of data.
-  RayObject(const std::shared_ptr<Buffer> &data, const std::shared_ptr<Buffer> &metadata,
-            bool copy_data = false)
-      : data_(data), metadata_(metadata), has_data_copy_(copy_data) {
-    if (has_data_copy_) {
-      // If this object is required to hold a copy of the data,
-      // make a copy if the passed in buffers don't already have a copy.
-      if (data_ && !data_->OwnsData()) {
-        data_ = std::make_shared<LocalMemoryBuffer>(data_->Data(), data_->Size(), true);
-      }
-
-      if (metadata_ && !metadata_->OwnsData()) {
-        metadata_ = std::make_shared<LocalMemoryBuffer>(metadata_->Data(),
-                                                        metadata_->Size(), true);
-      }
-    }
-  }
-
-  /// Return the data of the ray object.
-  const std::shared_ptr<Buffer> &GetData() const { return data_; };
-
-  /// Return the metadata of the ray object.
-  const std::shared_ptr<Buffer> &GetMetadata() const { return metadata_; };
-
-  uint64_t GetSize() const {
-    uint64_t size = 0;
-    size += (data_ != nullptr) ? data_->Size() : 0;
-    size += (metadata_ != nullptr) ? metadata_->Size() : 0;
-    return size;
-  }
-
-  /// Whether this object has metadata.
-  bool HasMetadata() const { return metadata_ != nullptr && metadata_->Size() > 0; }
-
- private:
-  /// Data of the ray object.
-  std::shared_ptr<Buffer> data_;
-  /// Metadata of the ray object.
-  std::shared_ptr<Buffer> metadata_;
-  /// Whether this class holds a data copy.
-  bool has_data_copy_;
-};
-
 /// Provider interface for store access. Store provider should inherit from this class and
 /// provide implementions for the methods. The actual store provider may use a plasma
 /// store or local memory store in worker process, or possibly other types of storage.

src/ray/core_worker/task_execution.cc

@@ -89,10 +89,17 @@ Status CoreWorkerTaskExecutionInterface::BuildArgsForExecutor(
       indices.push_back(i);
     } else {
       // pass by value.
-      (*args)[i] = std::make_shared<RayObject>(
-          std::make_shared<LocalMemoryBuffer>(const_cast<uint8_t *>(task.ArgVal(i)),
-                                              task.ArgValLength(i)),
-          nullptr);
+      std::shared_ptr<LocalMemoryBuffer> data = nullptr;
+      if (task.ArgDataSize(i)) {
+        data = std::make_shared<LocalMemoryBuffer>(const_cast<uint8_t *>(task.ArgData(i)),
+                                                   task.ArgDataSize(i));
+      }
+      std::shared_ptr<LocalMemoryBuffer> metadata = nullptr;
+      if (task.ArgMetadataSize(i)) {
+        metadata = std::make_shared<LocalMemoryBuffer>(
+            const_cast<uint8_t *>(task.ArgMetadata(i)), task.ArgMetadataSize(i));
+      }
+      (*args)[i] = std::make_shared<RayObject>(data, metadata);
     }
   }
 

src/ray/core_worker/task_interface.cc

@@ -130,7 +130,7 @@ void CoreWorkerTaskInterface::BuildCommonTaskSpec(
     if (arg.IsPassedByReference()) {
       builder.AddByRefArg(arg.GetReference());
     } else {
-      builder.AddByValueArg(arg.GetValue()->Data(), arg.GetValue()->Size());
+      builder.AddByValueArg(arg.GetValue());
     }
   }
 

src/ray/core_worker/test/core_worker_test.cc

@@ -3,6 +3,7 @@
 #include "gtest/gtest.h"
 
 #include "ray/common/buffer.h"
+#include "ray/common/ray_object.h"
 #include "ray/core_worker/context.h"
 #include "ray/core_worker/core_worker.h"
 #include "ray/core_worker/transport/direct_actor_transport.h"
@@ -59,7 +60,7 @@ std::unique_ptr<ActorHandle> CreateActorHelper(
 
   RayFunction func{ray::Language::PYTHON, {"actor creation task"}};
   std::vector<TaskArg> args;
-  args.emplace_back(TaskArg::PassByValue(buffer));
+  args.emplace_back(TaskArg::PassByValue(std::make_shared<RayObject>(buffer, nullptr)));
 
   ActorCreationOptions actor_options{max_reconstructions, is_direct_call, resources, {}};
 
@@ -232,7 +233,8 @@ void CoreWorkerTest::TestNormalTask(
       RAY_CHECK_OK(driver.Objects().Put(RayObject(buffer2, nullptr), &object_id));
 
       std::vector<TaskArg> args;
-      args.emplace_back(TaskArg::PassByValue(buffer1));
+      args.emplace_back(
+          TaskArg::PassByValue(std::make_shared<RayObject>(buffer1, nullptr)));
       args.emplace_back(TaskArg::PassByReference(object_id));
 
       RayFunction func{ray::Language::PYTHON, {}};
@@ -273,8 +275,10 @@ void CoreWorkerTest::TestActorTask(
 
       // Create arguments with PassByRef and PassByValue.
       std::vector<TaskArg> args;
-      args.emplace_back(TaskArg::PassByValue(buffer1));
-      args.emplace_back(TaskArg::PassByValue(buffer2));
+      args.emplace_back(
+          TaskArg::PassByValue(std::make_shared<RayObject>(buffer1, nullptr)));
+      args.emplace_back(
+          TaskArg::PassByValue(std::make_shared<RayObject>(buffer2, nullptr)));
 
       TaskOptions options{1, resources};
       std::vector<ObjectID> return_ids;
@@ -315,7 +319,8 @@ void CoreWorkerTest::TestActorTask(
     // Create arguments with PassByRef and PassByValue.
     std::vector<TaskArg> args;
     args.emplace_back(TaskArg::PassByReference(object_id));
-    args.emplace_back(TaskArg::PassByValue(buffer2));
+    args.emplace_back(
+        TaskArg::PassByValue(std::make_shared<RayObject>(buffer2, nullptr)));
 
     TaskOptions options{1, resources};
     std::vector<ObjectID> return_ids;
@@ -380,7 +385,8 @@ void CoreWorkerTest::TestActorReconstruction(
 
       // Create arguments with PassByValue.
       std::vector<TaskArg> args;
-      args.emplace_back(TaskArg::PassByValue(buffer1));
+      args.emplace_back(
+          TaskArg::PassByValue(std::make_shared<RayObject>(buffer1, nullptr)));
 
       TaskOptions options{1, resources};
       std::vector<ObjectID> return_ids;
@@ -425,7 +431,8 @@ void CoreWorkerTest::TestActorFailure(
 
       // Create arguments with PassByRef and PassByValue.
       std::vector<TaskArg> args;
-      args.emplace_back(TaskArg::PassByValue(buffer1));
+      args.emplace_back(
+          TaskArg::PassByValue(std::make_shared<RayObject>(buffer1, nullptr)));
 
       TaskOptions options{1, resources};
       std::vector<ObjectID> return_ids;
@@ -618,11 +625,10 @@ TEST_F(ZeroNodeTest, TestTaskArg) {
   ASSERT_TRUE(by_ref.IsPassedByReference());
   ASSERT_EQ(by_ref.GetReference(), id);
   // Test by-value argument.
-  std::shared_ptr<LocalMemoryBuffer> buffer =
-      std::make_shared<LocalMemoryBuffer>(static_cast<uint8_t *>(0), 0);
-  TaskArg by_value = TaskArg::PassByValue(buffer);
+  auto buffer = GenerateRandomBuffer();
+  TaskArg by_value = TaskArg::PassByValue(std::make_shared<RayObject>(buffer, nullptr));
   ASSERT_FALSE(by_value.IsPassedByReference());
-  auto data = by_value.GetValue();
+  auto data = by_value.GetValue().GetData();
   ASSERT_TRUE(data != nullptr);
   ASSERT_EQ(*data, *buffer);
 }
@@ -635,7 +641,7 @@ TEST_F(ZeroNodeTest, TestTaskSpecPerf) {
   auto buffer = std::make_shared<LocalMemoryBuffer>(array, sizeof(array));
   RayFunction function{ray::Language::PYTHON, {}};
   std::vector<TaskArg> args;
-  args.emplace_back(TaskArg::PassByValue(buffer));
+  args.emplace_back(TaskArg::PassByValue(std::make_shared<RayObject>(buffer, nullptr)));
 
   std::unordered_map<std::string, double> resources;
   ActorCreationOptions actor_options{0, /*is_direct_call*/ true, resources, {}};
@@ -664,7 +670,7 @@ TEST_F(ZeroNodeTest, TestTaskSpecPerf) {
       if (arg.IsPassedByReference()) {
         builder.AddByRefArg(arg.GetReference());
       } else {
-        builder.AddByValueArg(arg.GetValue()->Data(), arg.GetValue()->Size());
+        builder.AddByValueArg(arg.GetValue());
       }
     }
 
@@ -696,7 +702,7 @@ TEST_F(SingleNodeTest, TestDirectActorTaskSubmissionPerf) {
   auto buffer = std::make_shared<LocalMemoryBuffer>(array, sizeof(array));
   RayFunction func{ray::Language::PYTHON, {}};
   std::vector<TaskArg> args;
-  args.emplace_back(TaskArg::PassByValue(buffer));
+  args.emplace_back(TaskArg::PassByValue(std::make_shared<RayObject>(buffer, nullptr)));
 
   std::unordered_map<std::string, double> resources;
   ActorCreationOptions actor_options{0, /*is_direct_call*/ true, resources, {}};
@@ -712,7 +718,7 @@ TEST_F(SingleNodeTest, TestDirectActorTaskSubmissionPerf) {
   for (int i = 0; i < num_tasks; i++) {
     // Create arguments with PassByValue.
     std::vector<TaskArg> args;
-    args.emplace_back(TaskArg::PassByValue(buffer));
+    args.emplace_back(TaskArg::PassByValue(std::make_shared<RayObject>(buffer, nullptr)));
 
     TaskOptions options{1, resources};
     std::vector<ObjectID> return_ids;

src/ray/core_worker/transport/direct_actor_transport.cc

@@ -224,10 +224,10 @@ void CoreWorkerDirectActorTaskReceiver::HandlePushTask(
         /*transport_type=*/static_cast<int>(TaskTransportType::DIRECT_ACTOR));
     return_object->set_object_id(id.Binary());
     const auto &result = results[i];
-    if (result->GetData() != nullptr) {
+    if (result->HasData()) {
       return_object->set_data(result->GetData()->Data(), result->GetData()->Size());
     }
-    if (result->GetMetadata() != nullptr) {
+    if (result->HasMetadata()) {
       return_object->set_metadata(result->GetMetadata()->Data(),
                                   result->GetMetadata()->Size());
     }

src/ray/protobuf/common.proto

@@ -74,6 +74,8 @@ message TaskArg {
   repeated bytes object_ids = 1;
   // Data for pass-by-value arguments.
   bytes data = 2;
+  // Metadata for pass-by-value arguments.
+  bytes metadata = 3;
 }
 
 // Task spec of an actor creation task.