Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[dashboard][perf] Offload reporter_head CPU heavy JSON parsing to a thread. #45048

Merged
merged 2 commits into from
May 2, 2024
Merged

[dashboard][perf] Offload reporter_head CPU heavy JSON parsing to a thread. #45048

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d8287b6
Offload reporter_head CPU heavy JSON parsing to a thread.
rynewang Apr 30, 2024
917c360
rename
rynewang Apr 30, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 13 additions & 6 deletions dashboard/modules/reporter/reporter_head.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,11 @@
import logging
import asyncio
import aiohttp.web
from concurrent.futures import ThreadPoolExecutor
from typing import Optional, Tuple, List


import ray
import ray._private.services
import ray._private.utils
from ray._private.utils import get_or_create_event_loop, init_grpc_channel
import ray.dashboard.optional_utils as dashboard_optional_utils
from ray.dashboard.consts import GCS_RPC_TIMEOUT_SECONDS
import ray.dashboard.utils as dashboard_utils
Expand Down Expand Up @@ -69,6 +68,9 @@ def __init__(self, dashboard_head):
)
self._gcs_aio_client = dashboard_head.gcs_aio_client
self._state_api = None
self.thread_pool_executor = ThreadPoolExecutor(
max_workers=1, thread_name_prefix="reporter_head_worker"
)

async def _update_stubs(self, change):
if change.old:
Expand All @@ -79,7 +81,7 @@ async def _update_stubs(self, change):
node_id, ports = change.new
ip = DataSource.node_id_to_ip[node_id]
options = GLOBAL_GRPC_OPTIONS
channel = ray._private.utils.init_grpc_channel(
channel = init_grpc_channel(
f"{ip}:{ports[1]}", options=options, asynchronous=True
)
stub = reporter_pb2_grpc.ReporterServiceStub(channel)
Expand Down Expand Up @@ -637,9 +639,14 @@ async def run(self, server):
key, data = await subscriber.poll()
if key is None:
continue
data = json.loads(data)
# The JSON Parsing can be CPU heavy. Offload to another thread to avoid
# blocking the event loop.
loop = get_or_create_event_loop()
parsed_data = await loop.run_in_executor(
self.thread_pool_executor, json.loads, data
)
node_id = key.split(":")[-1]
DataSource.node_physical_stats[node_id] = data
DataSource.node_physical_stats[node_id] = parsed_data
except Exception:
logger.exception(
"Error receiving node physical stats from reporter agent."
Expand Down
Loading