wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-08-11 05:51:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

[multi-language part 1] add a 'language' field to task specification (#2639)

Browse Source
This commit is contained in:
Hao Chen
2018-08-16 21:26:42 -07:00
committed by Robert Nishihara
parent 6670880f03
commit a719e089b0
13 changed files with 200 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/format/common.fbs
+10
View File
@@ -20,6 +20,14 @@ table ResourcePair {
value: double;
}
// NOTE: This enum is duplicate with the `Language` enum in `gcs.fbs`,
// because we cannot include this file in `gcs.fbs` due to cyclic dependency.
// TODO(raulchen): remove it once we get rid of legacy ray.
enum TaskLanguage:int {
PYTHON = 0,
JAVA = 1
}
table TaskInfo {
// ID of the driver that created this task.
driver_id: string;
@@ -52,6 +60,8 @@ table TaskInfo {
// The required_resources vector indicates the quantities of the different
// resources required by this task.
required_resources: [ResourcePair];
// The language that this task belongs to
language: TaskLanguage;
}
// Object information data structure.
src/common/lib/python/common_extension.cc
+2 -1
View File
@@ -462,7 +462,8 @@ static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) {
self->task_spec = new ray::raylet::TaskSpecification(
driver_id, parent_task_id, parent_counter, actor_creation_id,
actor_creation_dummy_object_id, actor_id, actor_handle_id,
actor_counter, function_id, args, num_returns, required_resources);
actor_counter, function_id, args, num_returns, required_resources,
Language::PYTHON);
}
/* Set the task's execution dependencies. */
src/ray/raylet/lineage_cache_test.cc
+1 -1
View File
@@ -114,7 +114,7 @@ static inline Task ExampleTask(const std::vector<ObjectID> &arguments,
}
auto spec = TaskSpecification(UniqueID::nil(), UniqueID::from_random(), 0,
UniqueID::from_random(), task_arguments, num_returns,
required_resources);
required_resources, Language::PYTHON);
auto execution_spec = TaskExecutionSpecification(std::vector<ObjectID>());
execution_spec.IncrementNumForwards();
Task task = Task(execution_spec, spec);
src/ray/raylet/task_dependency_manager_test.cc
+1 -1
View File
@@ -76,7 +76,7 @@ static inline Task ExampleTask(const std::vector<ObjectID> &arguments,
}
auto spec = TaskSpecification(UniqueID::nil(), UniqueID::from_random(), 0,
UniqueID::from_random(), task_arguments, num_returns,
required_resources);
required_resources, Language::PYTHON);
auto execution_spec = TaskExecutionSpecification(std::vector<ObjectID>());
execution_spec.IncrementNumForwards();
Task task = Task(execution_spec, spec);
src/ray/raylet/task_spec.cc
+37 -4
View File
@@ -45,10 +45,12 @@ TaskSpecification::TaskSpecification(
const UniqueID &driver_id, const TaskID &parent_task_id, int64_t parent_counter,
const FunctionID &function_id,
const std::vector<std::shared_ptr<TaskArgument>> &task_arguments, int64_t num_returns,
const std::unordered_map<std::string, double> &required_resources)
const std::unordered_map<std::string, double> &required_resources,
const Language &language)
: TaskSpecification(driver_id, parent_task_id, parent_counter, ActorID::nil(),
ObjectID::nil(), ActorID::nil(), ActorHandleID::nil(), -1,
function_id, task_arguments, num_returns, required_resources) {}
function_id, task_arguments, num_returns, required_resources,
language) {}
TaskSpecification::TaskSpecification(
const UniqueID &driver_id, const TaskID &parent_task_id, int64_t parent_counter,
@@ -56,7 +58,8 @@ TaskSpecification::TaskSpecification(
const ActorID &actor_id, const ActorHandleID &actor_handle_id, int64_t actor_counter,
const FunctionID &function_id,
const std::vector<std::shared_ptr<TaskArgument>> &task_arguments, int64_t num_returns,
const std::unordered_map<std::string, double> &required_resources)
const std::unordered_map<std::string, double> &required_resources,
const Language &language)
: spec_() {
flatbuffers::FlatBufferBuilder fbb;
@@ -87,6 +90,20 @@ TaskSpecification::TaskSpecification(
returns.push_back(to_flatbuf(fbb, return_id));
}
// convert Language to TaskLanguage
// TODO(raulchen): remove this once we get rid of legacy ray.
TaskLanguage task_language = TaskLanguage::PYTHON;
switch (language) {
case Language::PYTHON:
task_language = TaskLanguage::PYTHON;
break;
case Language::JAVA:
task_language = TaskLanguage::JAVA;
break;
default:
RAY_LOG(FATAL) << "Unknown language: " << static_cast<int32_t>(language);
}
// Serialize the TaskSpecification.
auto spec = CreateTaskInfo(
fbb, to_flatbuf(fbb, driver_id), to_flatbuf(fbb, task_id),
@@ -94,7 +111,7 @@ TaskSpecification::TaskSpecification(
to_flatbuf(fbb, actor_creation_dummy_object_id), to_flatbuf(fbb, actor_id),
to_flatbuf(fbb, actor_handle_id), actor_counter, false,
to_flatbuf(fbb, function_id), fbb.CreateVector(arguments),
fbb.CreateVector(returns), map_to_flatbuf(fbb, required_resources));
fbb.CreateVector(returns), map_to_flatbuf(fbb, required_resources), task_language);
fbb.Finish(spec);
AssignSpecification(fbb.GetBufferPointer(), fbb.GetSize());
}
@@ -179,6 +196,22 @@ bool TaskSpecification::IsDriverTask() const {
return FunctionId().is_nil();
}
Language TaskSpecification::GetLanguage() const {
auto message = flatbuffers::GetRoot<TaskInfo>(spec_.data());
// TODO(raulchen): remove this once we get rid of legacy ray.
auto language = message->language();
switch (language) {
case TaskLanguage::PYTHON:
return Language::PYTHON;
case TaskLanguage::JAVA:
return Language::JAVA;
default:
// This shouldn't be reachable.
RAY_LOG(FATAL) << "Unknown task language: " << static_cast<int32_t>(language);
return Language::PYTHON;
}
}
bool TaskSpecification::IsActorCreationTask() const {
return !ActorCreationId().is_nil();
}
src/ray/raylet/task_spec.h
+5 -2
View File
@@ -98,7 +98,8 @@ class TaskSpecification {
int64_t parent_counter, const FunctionID &function_id,
const std::vector<std::shared_ptr<TaskArgument>> &arguments,
int64_t num_returns,
const std::unordered_map<std::string, double> &required_resources);
const std::unordered_map<std::string, double> &required_resources,
const Language &language);
TaskSpecification(const UniqueID &driver_id, const TaskID &parent_task_id,
int64_t parent_counter, const ActorID &actor_creation_id,
@@ -107,7 +108,8 @@ class TaskSpecification {
int64_t actor_counter, const FunctionID &function_id,
const std::vector<std::shared_ptr<TaskArgument>> &task_arguments,
int64_t num_returns,
const std::unordered_map<std::string, double> &required_resources);
const std::unordered_map<std::string, double> &required_resources,
const Language &language);
/// Deserialize a task specification from a flatbuffer's string data.
///
@@ -141,6 +143,7 @@ class TaskSpecification {
double GetRequiredResource(const std::string &resource_name) const;
const ResourceSet GetRequiredResources() const;
bool IsDriverTask() const;
Language GetLanguage() const;
// Methods specific to actor tasks.
bool IsActorCreationTask() const;