data store: remove threading #574
Conversation
|
Nice, hope we can get this working 🚀 |
|
Ping @dwsutherland, I may well have overlooked something, does this make sense to you? (note, tests are unhappy for async reasons, but it should work for real usage) |
Will have a look, I think I wanted to do this when I was first building, but didn't manage to figure out how to have non-blocking subscription receive loops or something.. Can't remember exactly why (wasn't as simple as the sleeps..), but well done if you've figured it out 👏 |
|
The changes look quite simple .. had a play, and I did notice something I couldn't reproduce on master. Not sure why.. could be something unrelated? maybe this branch needs rebased.. Looks like it might be trying to reconnect to the disconnected workflow? |
|
Ah, think I might know why, maybe .. the subscriber resolver ( |
| except WorkflowStopped: | ||
| self.disconnect_workflow(w_id) | ||
| except Exception as exc: | ||
| self.log.error(f'Failed to connect to {w_id}: {exc}') | ||
| self.disconnect_workflow(w_id) |
There was a problem hiding this comment.
Moved error handling up a level from the _start_subscription and _entire_workflow_update methods, no more threads to interfere with error reporting.
* Closes cylc#194 * Run subscribers via asyncio rather than the ThreadPoolExecutor. * The only non-blocking operations in the code being called were: - time.sleep (has an async variant) - asyncio.sleep (async) - self.socket.recv_multipart (async) * Refactored the code so that the underlying async functions could be called via asyncio.
* Move error handling up a level into `connect_workflow` from the `_start_subscription` and `_entire_workflow_update` methods. * Simplify tests (all exceptions are now caught in the same way). * Remove the multi-workflow handling ability of `_entire_workflow_update`, this is unused and can now be achieved more easily via asyncio.gather as the threadding has been removed.
There was a problem hiding this comment.
I'm still getting the same thing happening, when stopping a workflow (from UI of CLI):
When using hubless, the UIS just disconnects and process stops:
.
.
.
[I 2024-05-06 19:14:36.301 CylcUIServer] $ cylc play --color=never --mode live five/run1
[I 2024-05-06 19:14:36.834 CylcUIServer] [data-store] connect_workflow('~sutherlander/five/run1', <dict>)
[I 2024-05-06 19:14:49.404 CylcUIServer] [data-store] disconnect_workflow('~sutherlander/five/run1')
(uiserver) sutherlander@cortex-vbox:cylc-uiserver$
So it appears the UIS stops or crashes when a workflow is stopped/disconnected.
|
Does it make sense to deal with this one: |
|
Can move away from Tornado async interfaces in due course (these are already translated into asyncio calls by the tornado library), Jupyter Server hasn't moved over yet so we may need to wait a while before changing ourselves. |
Closes #194
Run subscribers via asyncio rather than the ThreadPoolExecutor.
I think the only non-blocking operations in the code being called were:
If so, I don't think we need to use threading here so I've refactored the code so that the underlying async functions could be called via asyncio.
If so, this cuts out the need for the
ThreadPoolExecutorremoving the workflow limit.Check List
CONTRIBUTING.mdand added my name as a Code Contributor.setup.cfg(andconda-environment.ymlif present).CHANGES.mdentry included if this is a change that can affect users?.?.xbranch.