[cpp worker] fix crash in empty args task (#11363)

cpp/src/ray/runtime/task/native_task_submitter.cc

@@ -8,36 +8,39 @@
 namespace ray {
 namespace api {
 
-void SubmitActorTask(const ActorID &actor_id, const RayFunction &function,
-                     const std::vector<std::unique_ptr<TaskArg>> &args,
-                     const TaskOptions &task_options, std::vector<ObjectID> *return_ids);
-
-ObjectID NativeTaskSubmitter::Submit(const InvocationSpec &invocation) {
+RayFunction BuildRayFunction(const InvocationSpec &invocation) {
   auto base_addr =
       GetBaseAddressOfLibraryFromAddr((void *)invocation.fptr.function_pointer);
-
   auto func_offset = (size_t)(invocation.fptr.function_pointer - base_addr);
   auto exec_func_offset = (size_t)(invocation.fptr.exec_function_pointer - base_addr);
   auto function_descriptor = FunctionDescriptorBuilder::BuildCpp(
       invocation.lib_name, std::to_string(func_offset), std::to_string(exec_func_offset));
-  auto ray_function = RayFunction(Language::CPP, function_descriptor);
+  return RayFunction(Language::CPP, function_descriptor);
+}
 
-  auto buffer = std::make_shared<::ray::LocalMemoryBuffer>(
-      reinterpret_cast<uint8_t *>(invocation.args->data()), invocation.args->size(),
-      true);
+void BuildTaskArgs(const InvocationSpec &invocation,
+                   std::vector<std::unique_ptr<ray::TaskArg>> &args) {
+  if (invocation.args->size() > 0) {
+    auto buffer = std::make_shared<::ray::LocalMemoryBuffer>(
+        reinterpret_cast<uint8_t *>(invocation.args->data()), invocation.args->size(),
+        true);
+    auto task_arg = new TaskArgByValue(
+        std::make_shared<::ray::RayObject>(buffer, nullptr, std::vector<ObjectID>()));
+    args.emplace_back(task_arg);
+  }
+}
+
+ObjectID NativeTaskSubmitter::Submit(const InvocationSpec &invocation) {
   std::vector<std::unique_ptr<ray::TaskArg>> args;
-  auto task_arg = new TaskArgByValue(
-      std::make_shared<::ray::RayObject>(buffer, nullptr, std::vector<ObjectID>()));
-  args.emplace_back(task_arg);
-
+  BuildTaskArgs(invocation, args);
   auto &core_worker = CoreWorkerProcess::GetCoreWorker();
   std::vector<ObjectID> return_ids;
   if (invocation.task_type == TaskType::ACTOR_TASK) {
-    core_worker.SubmitActorTask(invocation.actor_id, ray_function, args, TaskOptions(),
-                                &return_ids);
+    core_worker.SubmitActorTask(invocation.actor_id, BuildRayFunction(invocation), args,
+                                TaskOptions(), &return_ids);
   } else {
-    core_worker.SubmitTask(ray_function, args, TaskOptions(), &return_ids, 1,
-                           std::make_pair(PlacementGroupID::Nil(), -1), true);
+    core_worker.SubmitTask(BuildRayFunction(invocation), args, TaskOptions(), &return_ids,
+                           1, std::make_pair(PlacementGroupID::Nil(), -1), true);
   }
   return return_ids[0];
 }
@@ -47,22 +50,8 @@ ObjectID NativeTaskSubmitter::SubmitTask(const InvocationSpec &invocation) {
 }
 
 ActorID NativeTaskSubmitter::CreateActor(const InvocationSpec &invocation) {
-  auto base_addr =
-      GetBaseAddressOfLibraryFromAddr((void *)invocation.fptr.function_pointer);
-
-  auto func_offset = (size_t)(invocation.fptr.function_pointer - base_addr);
-  auto exec_func_offset = (size_t)(invocation.fptr.exec_function_pointer - base_addr);
-  auto function_descriptor = FunctionDescriptorBuilder::BuildCpp(
-      invocation.lib_name, std::to_string(func_offset), std::to_string(exec_func_offset));
-  auto ray_function = RayFunction(Language::CPP, function_descriptor);
-
-  auto buffer = std::make_shared<::ray::LocalMemoryBuffer>(
-      reinterpret_cast<uint8_t *>(invocation.args->data()), invocation.args->size(),
-      true);
   std::vector<std::unique_ptr<ray::TaskArg>> args;
-  auto task_arg = new TaskArgByValue(
-      std::make_shared<::ray::RayObject>(buffer, nullptr, std::vector<ObjectID>()));
-  args.emplace_back(task_arg);
+  BuildTaskArgs(invocation, args);
 
   auto &core_worker = CoreWorkerProcess::GetCoreWorker();
 
@@ -78,11 +67,11 @@ ActorID NativeTaskSubmitter::CreateActor(const InvocationSpec &invocation) {
                                      name,
                                      /*is_asyncio=*/false};
   ActorID actor_id;
-  auto status = core_worker.CreateActor(ray_function, args, actor_options, "", &actor_id);
+  auto status = core_worker.CreateActor(BuildRayFunction(invocation), args, actor_options,
+                                        "", &actor_id);
   if (!status.ok()) {
     throw RayException("Create actor error");
   }
-
   return actor_id;
 }
 

cpp/src/ray/runtime/task/task_executor.cc

@@ -38,14 +38,17 @@ Status TaskExecutor::ExecuteTask(
   std::string lib_name = typed_descriptor->LibName();
   std::string func_offset = typed_descriptor->FunctionOffset();
   std::string exec_func_offset = typed_descriptor->ExecFunctionOffset();
-
-  auto args_buffer = args[0]->GetData();
-  auto args_sbuffer = std::make_shared<msgpack::sbuffer>(args_buffer->Size());
-  /// TODO(Guyang Song): Avoid the memory copy.
-  args_sbuffer->write(reinterpret_cast<const char *>(args_buffer->Data()),
-                      args_buffer->Size());
+  std::shared_ptr<msgpack::sbuffer> args_sbuffer;
+  if (args.size() > 0) {
+    auto args_buffer = args[0]->GetData();
+    args_sbuffer = std::make_shared<msgpack::sbuffer>(args_buffer->Size());
+    /// TODO(Guyang Song): Avoid the memory copy.
+    args_sbuffer->write(reinterpret_cast<const char *>(args_buffer->Data()),
+                        args_buffer->Size());
+  } else {
+    args_sbuffer = std::make_shared<msgpack::sbuffer>();
+  }
   auto base_addr = FunctionHelper::GetInstance().GetBaseAddress(lib_name);
-
   std::shared_ptr<msgpack::sbuffer> data = nullptr;
   if (task_type == TaskType::ACTOR_CREATION_TASK) {
     typedef std::shared_ptr<msgpack::sbuffer> (*ExecFunction)(

cpp/src/ray/test/cluster/cluster_mode_test.cc

@@ -7,6 +7,7 @@
 using namespace ray::api;
 
 /// general function of user code
+int Return1() { return 1; }
 int Plus1(int x) { return x + 1; }
 
 /// a class of user code
@@ -15,9 +16,13 @@ class Counter {
   int count;
 
   Counter(int init) { count = init; }
-
+  static Counter *FactoryCreate() { return new Counter(0); }
   static Counter *FactoryCreate(int init) { return new Counter(init); }
   /// non static function
+  int Plus1() {
+    count += 1;
+    return count;
+  }
   int Add(int x) {
     count += x;
     return count;
@@ -36,14 +41,27 @@ TEST(RayClusterModeTest, FullTest) {
   auto get_result = *(Ray::Get(obj));
   EXPECT_EQ(12345, get_result);
 
-  auto task_obj = Ray::Task(Plus1, 5).Remote();
+  /// common task without args
+  auto task_obj = Ray::Task(Return1).Remote();
   int task_result = *(Ray::Get(task_obj));
+  EXPECT_EQ(1, task_result);
+
+  /// common task with args
+  task_obj = Ray::Task(Plus1, 5).Remote();
+  task_result = *(Ray::Get(task_obj));
   EXPECT_EQ(6, task_result);
 
-  ActorHandle<Counter> actor = Ray::Actor(Counter::FactoryCreate, 1).Remote();
-  auto actor_object = actor.Task(&Counter::Add, 5).Remote();
-  int actor_task_result = *(Ray::Get(actor_object));
-  EXPECT_EQ(6, actor_task_result);
+  /// actor task without args
+  ActorHandle<Counter> actor1 = Ray::Actor(Counter::FactoryCreate).Remote();
+  auto actor_object1 = actor1.Task(&Counter::Plus1).Remote();
+  int actor_task_result1 = *(Ray::Get(actor_object1));
+  EXPECT_EQ(1, actor_task_result1);
+
+  /// actor task with args
+  ActorHandle<Counter> actor2 = Ray::Actor(Counter::FactoryCreate, 1).Remote();
+  auto actor_object2 = actor2.Task(&Counter::Add, 5).Remote();
+  int actor_task_result2 = *(Ray::Get(actor_object2));
+  EXPECT_EQ(6, actor_task_result2);
 
   Ray::Shutdown();
 }