From 34191107a3c532d723ec728c1db1107870b530ae Mon Sep 17 00:00:00 2001 From: SongGuyang Date: Thu, 15 Oct 2020 07:48:34 +0800 Subject: [PATCH] [cpp worker] fix crash in empty args task (#11363) --- .../ray/runtime/task/native_task_submitter.cc | 59 ++++++++----------- cpp/src/ray/runtime/task/task_executor.cc | 17 +++--- cpp/src/ray/test/cluster/cluster_mode_test.cc | 30 ++++++++-- 3 files changed, 58 insertions(+), 48 deletions(-) diff --git a/cpp/src/ray/runtime/task/native_task_submitter.cc b/cpp/src/ray/runtime/task/native_task_submitter.cc index b7efb7ce4..4039dc76a 100644 --- a/cpp/src/ray/runtime/task/native_task_submitter.cc +++ b/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> &args, - const TaskOptions &task_options, std::vector *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(invocation.args->data()), invocation.args->size(), - true); +void BuildTaskArgs(const InvocationSpec &invocation, + std::vector> &args) { + if (invocation.args->size() > 0) { + auto buffer = std::make_shared<::ray::LocalMemoryBuffer>( + reinterpret_cast(invocation.args->data()), invocation.args->size(), + true); + auto task_arg = new TaskArgByValue( + std::make_shared<::ray::RayObject>(buffer, nullptr, std::vector())); + args.emplace_back(task_arg); + } +} + +ObjectID NativeTaskSubmitter::Submit(const InvocationSpec &invocation) { std::vector> args; - auto task_arg = new TaskArgByValue( - std::make_shared<::ray::RayObject>(buffer, nullptr, std::vector())); - args.emplace_back(task_arg); - + BuildTaskArgs(invocation, args); auto &core_worker = CoreWorkerProcess::GetCoreWorker(); std::vector 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(invocation.args->data()), invocation.args->size(), - true); std::vector> args; - auto task_arg = new TaskArgByValue( - std::make_shared<::ray::RayObject>(buffer, nullptr, std::vector())); - 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; } diff --git a/cpp/src/ray/runtime/task/task_executor.cc b/cpp/src/ray/runtime/task/task_executor.cc index b064eb14c..b07f58a7a 100644 --- a/cpp/src/ray/runtime/task/task_executor.cc +++ b/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(args_buffer->Size()); - /// TODO(Guyang Song): Avoid the memory copy. - args_sbuffer->write(reinterpret_cast(args_buffer->Data()), - args_buffer->Size()); + std::shared_ptr args_sbuffer; + if (args.size() > 0) { + auto args_buffer = args[0]->GetData(); + args_sbuffer = std::make_shared(args_buffer->Size()); + /// TODO(Guyang Song): Avoid the memory copy. + args_sbuffer->write(reinterpret_cast(args_buffer->Data()), + args_buffer->Size()); + } else { + args_sbuffer = std::make_shared(); + } auto base_addr = FunctionHelper::GetInstance().GetBaseAddress(lib_name); - std::shared_ptr data = nullptr; if (task_type == TaskType::ACTOR_CREATION_TASK) { typedef std::shared_ptr (*ExecFunction)( diff --git a/cpp/src/ray/test/cluster/cluster_mode_test.cc b/cpp/src/ray/test/cluster/cluster_mode_test.cc index be7916723..e3e0b990b 100644 --- a/cpp/src/ray/test/cluster/cluster_mode_test.cc +++ b/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 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 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 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(); }