Do heartbeat on clients #9798
Do heartbeat on clients #9798
Conversation
CodSpeed Performance ReportMerging #9798 will not alter performanceComparing Summary
|
|
Tested with bigpoly on LSF (400 reals, 200 max running) and all was successfully reconnected.
|
xjules
force-pushed
the
fix_monitor
branch
2 times, most recently
from
887d748 to
bcd8563
Compare
| await client.send("start", retries=1) | ||
|
|
||
| await mock_server.do_ping() | ||
| await asyncio.sleep(1) |
There was a problem hiding this comment.
Do we sleep here to force the monitor to reconnect?
There was a problem hiding this comment.
Yes, that is correct. PING_TIMEOUT is 0.1 so sleeping 1 seconds guarantees re-connection.
| if ( | ||
| last_ping_time | ||
| and (asyncio.get_running_loop().time() - last_ping_time) | ||
| > 2 * PING_TIMEOUT |
There was a problem hiding this comment.
I guess the 2* here is for the ping to be received and returned, but do we need it if we are not returning the ping? Maybe in our case it should be added in the PING_TIMEOUT directly?
There was a problem hiding this comment.
This 2 * PING_TIMEOUT is to account for a potential delay, which can happen if the ee is very busy.
| @@ -93,11 +95,23 @@ async def process_message(self, msg: str) -> None: | |||
| raise NotImplementedError("Only monitor can receive messages!") | |||
|
|
|||
| async def _receiver(self) -> None: | |||
| last_beat_time: float | None = None | |||
There was a problem hiding this comment.
Maybe last_heartbeat_time?
| @@ -342,6 +360,7 @@ async def _start_running(self) -> None: | |||
| raise ValueError("no config for evaluator") | |||
| self._ee_tasks = [ | |||
| asyncio.create_task(self._server(), name="server_task"), | |||
| asyncio.create_task(self._do_heartbeat_clients(), name="beat_task"), | |||
There was a problem hiding this comment.
| asyncio.create_task(self._do_heartbeat_clients(), name="beat_task"), | |
| asyncio.create_task(self._do_heartbeat_clients(), name="heartbeat_task"), |
That has been changed, just the hearbeat_task not yet. |
| @@ -51,7 +57,7 @@ def __init__(self, ensemble: Ensemble, config: EvaluatorServerConfig): | |||
| self._ensemble: Ensemble = ensemble | |||
|
|
|||
| self._events: asyncio.Queue[Event] = asyncio.Queue() | |||
| self._events_to_send: asyncio.Queue[Event] = asyncio.Queue() | |||
| self._events_to_send: asyncio.Queue[Event | bytes] = asyncio.Queue() | |||
There was a problem hiding this comment.
Could we be more specific here?
xjules
force-pushed
the
fix_monitor
branch
2 times, most recently
from
dac0c07 to
467c276
Compare
Heartbeat task sends HEARTBEAT to all the clients (ie. Monitor) at client.HEARTBEAT_TIMEOUT intervals. Clients do not reply, just process the message. If client detects longer delay between two heartbeats, the client will send CONNECT to evaluator in addition; ie. getting the connection re-established after a break. This is to simulate re-connection. Each CONNECT_MSG will then trigger sending FullSnapshot from the ensemble evaluator. Initially HEARTBEAT_TIMEOUT is set to 5 seconds while Monitor accepts 10 seconds at max as a delay. Additionally, initial connection will now undergo same amount of retries as standard messages.
Issue
Resolves #9782
Approach
Heartbeat task sends HEARTBEAT to all the clients (ie. Monitor) at client.HEARTBEAT_TIMEOUT intervals.
Clients do not reply, just process the message. If client detects a longer delay between two heartbeats, the client will send CONNECT to evaluator in addition; ie. getting the connection re-established after a break. This is to simulate re-connection. Each CONNECT_MSG will then trigger sending FullSnapshot from the ensemble evaluator.
Initially HEARTBEAT_TIMEOUT is set to 5 seconds while Monitor accepts 10 seconds at max as a delay. Additionally, initial connection will now undergo same amount of retries as standard messages.
(Screenshot of new behavior in GUI if applicable)
git rebase -i main --exec 'pytest tests/ert/unit_tests -n auto -m "not integration_test"')When applicable