From 3acf3c76751664619d88ff3150c6f48800e09fd3 Mon Sep 17 00:00:00 2001 From: Yunzhi Zhang <35828389+zzyunzhi@users.noreply.github.com> Date: Fri, 17 Jan 2020 15:43:56 -0800 Subject: [PATCH] [Dashboard] Add actor task counter (#6820) --- python/ray/dashboard/dashboard.py | 29 ++++++++++++++++++++--------- python/ray/state.py | 1 + python/ray/tests/test_metrics.py | 1 + src/ray/core_worker/core_worker.cc | 3 +++ src/ray/core_worker/core_worker.h | 3 +++ src/ray/protobuf/common.proto | 2 ++ src/ray/protobuf/gcs.proto | 2 ++ src/ray/raylet/node_manager.cc | 1 + 8 files changed, 33 insertions(+), 9 deletions(-) diff --git a/python/ray/dashboard/dashboard.py b/python/ray/dashboard/dashboard.py index a355cbf45..35f2f2e47 100644 --- a/python/ray/dashboard/dashboard.py +++ b/python/ray/dashboard/dashboard.py @@ -79,6 +79,10 @@ def measures_to_dict(measures): return measures_dict +def b64_decode(reply): + return b64decode(reply).decode("utf-8") + + class Dashboard(object): """A dashboard process for monitoring Ray nodes. @@ -215,7 +219,7 @@ class Dashboard(object): format_resource(resource_name, total_resource - available_resource), format_resource(resource_name, total_resource))) - data["extraInfo"] = ",".join(extra_info_strings) + "\n" + data["extraInfo"] = ", ".join(extra_info_strings) + "\n" if os.environ.get("RAY_DASHBOARD_DEBUG"): # process object store info extra_info_strings = [] @@ -307,6 +311,7 @@ class NodeStats(threading.Thread): "ipAddress": "", "isDirectCall": False, "jobId": "", + "numExecutedTasks": 0, "numLocalObjects": 0, "numObjectIdsInScope": 0, "port": 0, @@ -371,6 +376,7 @@ class NodeStats(threading.Thread): } def get_actor_tree(self, workers_info, infeasible_tasks) -> Dict: + now = time.time() # construct flattened actor tree flattened_tree = {"root": {"children": {}}} child_to_parent = {} @@ -389,13 +395,17 @@ class NodeStats(threading.Thread): addr = (core_worker_stats["ipAddress"], str(core_worker_stats["port"])) if addr in self._addr_to_actor_id: - actor_id = self._addr_to_actor_id[addr] - if "currentTaskDesc" in core_worker_stats: - core_worker_stats.pop("currentTaskDesc") - if "numPendingTasks" in core_worker_stats: - core_worker_stats.pop("numPendingTasks") + actor_info = flattened_tree[self._addr_to_actor_id[ + addr]] + if "currentTaskFuncDesc" in core_worker_stats: + core_worker_stats["currentTaskFuncDesc"] = list( + map(b64_decode, + core_worker_stats["currentTaskFuncDesc"])) format_reply(core_worker_stats) - flattened_tree[actor_id].update(core_worker_stats) + actor_info.update(core_worker_stats) + actor_info["averageTaskExecutionSpeed"] = round( + actor_info["numExecutedTasks"] / + (now - actor_info["timestamp"] / 1000), 2) for infeasible_task in infeasible_tasks: actor_id = ray.utils.binary_to_hex( @@ -407,8 +417,7 @@ class NodeStats(threading.Thread): child_to_parent[actor_id] = caller_id infeasible_task["state"] = -1 infeasible_task["functionDescriptor"] = list( - map(lambda desc: b64decode(desc).decode("utf-8"), - infeasible_task["functionDescriptor"])) + map(b64_decode, infeasible_task["functionDescriptor"])) format_reply(infeasible_tasks) flattened_tree[actor_id] = infeasible_task @@ -463,6 +472,7 @@ class NodeStats(threading.Thread): "jobId": actor_data["JobID"], "state": actor_data["State"], "isDirectCall": actor_data["IsDirectCall"], + "timestamp": actor_data["Timestamp"] } for x in p.listen(): @@ -506,6 +516,7 @@ class NodeStats(threading.Thread): actor_data.job_id), "state": actor_data.state, "isDirectCall": actor_data.is_direct_call, + "timestamp": actor_data.timestamp } else: data = json.loads(ray.utils.decode(data)) diff --git a/python/ray/state.py b/python/ray/state.py index 3c572e07e..e7b9a4acd 100644 --- a/python/ray/state.py +++ b/python/ray/state.py @@ -336,6 +336,7 @@ class GlobalState: }, "IsDirectCall": actor_table_data.is_direct_call, "State": actor_table_data.state, + "Timestamp": actor_table_data.timestamp, } return actor_info diff --git a/python/ray/tests/test_metrics.py b/python/ray/tests/test_metrics.py index 3c3d48bb6..a18c13d0a 100644 --- a/python/ray/tests/test_metrics.py +++ b/python/ray/tests/test_metrics.py @@ -168,6 +168,7 @@ def test_raylet_info_endpoint(shutdown_only): "Timed out while waiting for dashboard to start.") assert parent_actor_info["usedResources"]["CPU"] == 2 + assert parent_actor_info["numExecutedTasks"] == 3 for _, child_actor_info in children.items(): if child_actor_info["state"] == -1: assert child_actor_info["requiredResources"]["CustomResource"] == 1 diff --git a/src/ray/core_worker/core_worker.cc b/src/ray/core_worker/core_worker.cc index 66e5aafbf..5db41b211 100644 --- a/src/ray/core_worker/core_worker.cc +++ b/src/ray/core_worker/core_worker.cc @@ -84,6 +84,7 @@ CoreWorker::CoreWorker(const WorkerType worker_type, const Language language, core_worker_server_(WorkerTypeString(worker_type), 0 /* let grpc choose a port */), reference_counter_(std::make_shared()), task_queue_length_(0), + num_executed_tasks_(0), task_execution_service_work_(task_execution_service_), task_execution_callback_(task_execution_callback), resource_ids_(new ResourceMappingType()), @@ -896,6 +897,7 @@ Status CoreWorker::ExecuteTask(const TaskSpecification &task_spec, const std::shared_ptr &resource_ids, std::vector> *return_objects) { task_queue_length_ -= 1; + num_executed_tasks_ += 1; if (resource_ids != nullptr) { resource_ids_ = resource_ids; @@ -1167,6 +1169,7 @@ void CoreWorker::HandleGetCoreWorkerStats(const rpc::GetCoreWorkerStatsRequest & auto stats = reply->mutable_core_worker_stats(); stats->set_num_pending_tasks(task_manager_->NumPendingTasks()); stats->set_task_queue_length(task_queue_length_); + stats->set_num_executed_tasks(num_executed_tasks_); stats->set_num_object_ids_in_scope(reference_counter_->NumObjectIDsInScope()); if (!current_task_.TaskId().IsNil()) { stats->set_current_task_desc(current_task_.DebugString()); diff --git a/src/ray/core_worker/core_worker.h b/src/ray/core_worker/core_worker.h index 5daa374d1..68623dafd 100644 --- a/src/ray/core_worker/core_worker.h +++ b/src/ray/core_worker/core_worker.h @@ -660,6 +660,9 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler { /// Number of tasks that have been pushed to the actor but not executed. std::atomic task_queue_length_; + /// Number of executed tasks. + std::atomic num_executed_tasks_; + /// Event loop where tasks are processed. boost::asio::io_service task_execution_service_; diff --git a/src/ray/protobuf/common.proto b/src/ray/protobuf/common.proto index 2d003d634..f0860af98 100644 --- a/src/ray/protobuf/common.proto +++ b/src/ray/protobuf/common.proto @@ -212,4 +212,6 @@ message CoreWorkerStats { int64 used_object_store_memory = 12; // Length of the task queue. int32 task_queue_length = 13; + // Number of executed tasks. + int32 num_executed_tasks = 14; } diff --git a/src/ray/protobuf/gcs.proto b/src/ray/protobuf/gcs.proto index f0bc81efa..6c936dc1f 100644 --- a/src/ray/protobuf/gcs.proto +++ b/src/ray/protobuf/gcs.proto @@ -117,6 +117,8 @@ message ActorTableData { bool is_direct_call = 11; // Whether the actor is persistent. bool is_detached = 12; + // Timestamp that the actor is created or reconstructed. + double timestamp = 13; } message ErrorTableData { diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 1ece8f99a..fbc06a238 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -2445,6 +2445,7 @@ std::shared_ptr NodeManager::CreateActorTableDataFromCreationTas actor_info_ptr->mutable_address()->set_raylet_id(self_node_id_.Binary()); actor_info_ptr->mutable_address()->set_worker_id(worker_id.Binary()); actor_info_ptr->set_state(ActorTableData::ALIVE); + actor_info_ptr->set_timestamp(current_time_ms()); return actor_info_ptr; }