diff --git a/dashboard/datacenter.py b/dashboard/datacenter.py index 357bab3c1..c6b05d9e8 100644 --- a/dashboard/datacenter.py +++ b/dashboard/datacenter.py @@ -77,7 +77,8 @@ class DataOrganizer: job_workers = {} node_workers = {} core_worker_stats = {} - for node_id in DataSource.nodes.keys(): + # await inside for loop, so we create a copy of keys(). + for node_id in list(DataSource.nodes.keys()): workers = await cls.get_node_workers(node_id) for worker in workers: job_id = worker["jobId"] diff --git a/dashboard/modules/logical_view/logical_view_head.py b/dashboard/modules/logical_view/logical_view_head.py index 674ac64e9..cf29db637 100644 --- a/dashboard/modules/logical_view/logical_view_head.py +++ b/dashboard/modules/logical_view/logical_view_head.py @@ -4,7 +4,7 @@ import ray.utils import ray.new_dashboard.utils as dashboard_utils import ray.new_dashboard.actor_utils as actor_utils from ray.new_dashboard.utils import rest_response -from ray.new_dashboard.datacenter import DataOrganizer +from ray.new_dashboard.datacenter import DataOrganizer, DataSource from ray.core.generated import core_worker_pb2 from ray.core.generated import core_worker_pb2_grpc @@ -29,6 +29,14 @@ class LogicalViewHead(dashboard_utils.DashboardHeadModule): message="Fetched actor groups.", actor_groups=actor_groups) + @routes.get("/logical/actors") + @dashboard_utils.aiohttp_cache + async def get_all_actors(self, req) -> aiohttp.web.Response: + return dashboard_utils.rest_response( + success=True, + message="All actors fetched.", + actors=DataSource.actors) + @routes.get("/logical/kill_actor") async def kill_actor(self, req) -> aiohttp.web.Response: try: diff --git a/dashboard/modules/logical_view/tests/test_logical_view_head.py b/dashboard/modules/logical_view/tests/test_logical_view_head.py index 5e4a8bb6c..ceee063dd 100644 --- a/dashboard/modules/logical_view/tests/test_logical_view_head.py +++ b/dashboard/modules/logical_view/tests/test_logical_view_head.py @@ -79,6 +79,63 @@ def test_actor_groups(ray_start_with_dashboard): raise Exception(f"Timed out while testing, {ex_stack}") +def test_actors(disable_aiohttp_cache, ray_start_with_dashboard): + @ray.remote + class Foo: + def __init__(self, num): + self.num = num + + def do_task(self): + return self.num + + @ray.remote(num_gpus=1) + class InfeasibleActor: + pass + + foo_actors = [Foo.remote(4), Foo.remote(5)] + infeasible_actor = InfeasibleActor.remote() # noqa + results = [actor.do_task.remote() for actor in foo_actors] # noqa + webui_url = ray_start_with_dashboard["webui_url"] + assert wait_until_server_available(webui_url) + webui_url = format_web_url(webui_url) + + timeout_seconds = 5 + start_time = time.time() + last_ex = None + while True: + time.sleep(1) + try: + resp = requests.get(f"{webui_url}/logical/actors") + resp_json = resp.json() + resp_data = resp_json["data"] + actors = resp_data["actors"] + assert len(actors) == 3 + one_entry = list(actors.values())[0] + assert "jobId" in one_entry + assert "taskSpec" in one_entry + assert "functionDescriptor" in one_entry["taskSpec"] + assert type(one_entry["taskSpec"]["functionDescriptor"]) is dict + assert "address" in one_entry + assert type(one_entry["address"]) is dict + assert "state" in one_entry + assert "name" in one_entry + assert "numRestarts" in one_entry + assert "pid" in one_entry + all_pids = [entry["pid"] for entry in actors.values()] + assert 0 in all_pids # The infeasible actor + assert len(all_pids) > 1 + break + except Exception as ex: + last_ex = ex + finally: + if time.time() > start_time + timeout_seconds: + ex_stack = traceback.format_exception( + type(last_ex), last_ex, + last_ex.__traceback__) if last_ex else [] + ex_stack = "".join(ex_stack) + raise Exception(f"Timed out while testing, {ex_stack}") + + def test_kill_actor(ray_start_with_dashboard): @ray.remote class Actor: diff --git a/dashboard/modules/stats_collector/stats_collector_head.py b/dashboard/modules/stats_collector/stats_collector_head.py index 1224b6d62..ae75864e5 100644 --- a/dashboard/modules/stats_collector/stats_collector_head.py +++ b/dashboard/modules/stats_collector/stats_collector_head.py @@ -246,7 +246,8 @@ class StatsCollector(dashboard_utils.DashboardHeadModule): @async_loop_forever( stats_collector_consts.NODE_STATS_UPDATE_INTERVAL_SECONDS) async def _update_node_stats(self): - for node_id, stub in self._stubs.items(): + # Copy self._stubs to avoid `dictionary changed size during iteration`. + for node_id, stub in list(self._stubs.items()): node_info = DataSource.nodes.get(node_id) if node_info["state"] != "ALIVE": continue diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.cc b/src/ray/gcs/gcs_server/gcs_actor_manager.cc index e30cf2569..17c7c0f5a 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_actor_manager.cc @@ -341,6 +341,9 @@ Status GcsActorManager::RegisterActor(const ray::rpc::RegisterActorRequest &requ // the actor state to DEAD to avoid race condition. return; } + RAY_CHECK_OK(gcs_pub_sub_->Publish(ACTOR_CHANNEL, actor->GetActorID().Hex(), + actor->GetActorTableData().SerializeAsString(), + nullptr)); // Invoke all callbacks for all registration requests of this actor (duplicated // requests are included) and remove all of them from // actor_to_register_callbacks_. diff --git a/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc b/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc index aa420f463..bbdd58856 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc +++ b/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc @@ -323,6 +323,7 @@ void GcsActorScheduler::HandleWorkerLeasedReply( .emplace(leased_worker->GetWorkerID(), leased_worker) .second); actor->UpdateAddress(leased_worker->GetAddress()); + actor->GetMutableActorTableData()->set_pid(reply.worker_pid()); // Make sure to connect to the client before persisting actor info to GCS. // Without this, there could be a possible race condition. Related issues: // https://github.com/ray-project/ray/pull/9215/files#r449469320 diff --git a/src/ray/protobuf/gcs.proto b/src/ray/protobuf/gcs.proto index fe2511bd0..82704bac4 100644 --- a/src/ray/protobuf/gcs.proto +++ b/src/ray/protobuf/gcs.proto @@ -143,6 +143,8 @@ message ActorTableData { // Resource mapping ids acquired by the leased worker. This field is only set when this // actor already has a leased worker. repeated ResourceMapEntry resource_mapping = 15; + // The process id of this actor. + uint32 pid = 16; } message ErrorTableData { diff --git a/src/ray/protobuf/node_manager.proto b/src/ray/protobuf/node_manager.proto index 087135e39..dcf6fe783 100644 --- a/src/ray/protobuf/node_manager.proto +++ b/src/ray/protobuf/node_manager.proto @@ -37,6 +37,8 @@ message RequestWorkerLeaseReply { // Whether this lease request was canceled. In this case, the // client should try again if the resources are still required. bool canceled = 4; + // PID of the worker process. + uint32 worker_pid = 5; } message PrepareBundleResourcesRequest { diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index b27a0e32c..52e9354a2 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -1711,10 +1711,14 @@ void NodeManager::HandleRequestWorkerLease(const rpc::RequestWorkerLeaseRequest [this, owner_address, reply, send_reply_callback]( const std::shared_ptr granted, const std::string &address, int port, const WorkerID &worker_id, const ResourceIdSet &resource_ids) { + auto worker = std::static_pointer_cast(granted); + uint32_t worker_pid = static_cast(worker->GetProcess().GetId()); + reply->mutable_worker_address()->set_ip_address(address); reply->mutable_worker_address()->set_port(port); reply->mutable_worker_address()->set_worker_id(worker_id.Binary()); reply->mutable_worker_address()->set_raylet_id(self_node_id_.Binary()); + reply->set_worker_pid(worker_pid); for (const auto &mapping : resource_ids.AvailableResources()) { auto resource = reply->add_resource_mapping(); resource->set_name(mapping.first); @@ -1743,7 +1747,6 @@ void NodeManager::HandleRequestWorkerLease(const rpc::RequestWorkerLeaseRequest RAY_CHECK(leased_workers_.find(worker_id) == leased_workers_.end()) << "Worker is already leased out " << worker_id; - auto worker = std::static_pointer_cast(granted); leased_workers_[worker_id] = worker; }); task.OnSpillbackInstead(