From 1f1408e586675bca7e70adf85ba5c94cce00b99d Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Fri, 13 Nov 2026 05:43:19 +0000 Subject: [PATCH] perf: one workspace thread by default `Workspace` now runs requests on one thread unless given `workers`, and `local()` and `open_folder()` take `workers` and pass it on. A LoKI reduction is already parallel inside (dask, TBB), so more workspace threads slow it down and multiply its peak memory. Four I(Q) reductions take 14.8 s at 8 GB on 1 thread and 20.1 s at 27 GB on 4 threads; through one stage, 7.9 s at 4.7 GB on 1 thread against 7.3 s at 15 GB on 4. The best number depends on the instrument (file sizes, file counts, processing patterns), so an application sets it. Two persistence tests need a second request to run while the first waits, and now ask for two threads. Co-Authored-By: Claude Fable 5.2 --- docs/developer/README.md | 2 +- docs/developer/deployments.md | 6 +-- packages/essapps/src/ess/apps/folder.py | 5 ++- packages/essapps/src/ess/apps/workspace.py | 15 +++++-- .../essapps/tests/client/scheduling_test.py | 43 ++++++++++++++++++- packages/essapps/tests/persist_test.py | 33 +++++++++++--- 6 files changed, 87 insertions(+), 17 deletions(-) diff --git a/docs/developer/README.md b/docs/developer/README.md index 0b17bae1..49a7faab 100644 --- a/docs/developer/README.md +++ b/docs/developer/README.md @@ -148,7 +148,7 @@ In each deployment of the first release, one Python process holds the applicatio | technique application with a batch page | one per folder, started by a user on a VISA machine | in the folder | | automatic reduction | one per instrument, on a DMSC server, or on a VISA machine for the first release | one per proposal, on disk | -Only (a) lies between two processes, when a browser draws the screen; bindings run on 4 threads of the workspace's process. +Only (a) lies between two processes, when a browser draws the screen; bindings run on threads of the workspace's process, one by default. A screen never calls a workspace: what it shows is computed in Python next to the data, and the HTTP interface of a page belongs to that page. A proposal has several workspaces, each with its own history, and a reference names a record of the same history. diff --git a/docs/developer/deployments.md b/docs/developer/deployments.md index 40b3c3a5..a07e8180 100644 --- a/docs/developer/deployments.md +++ b/docs/developer/deployments.md @@ -12,7 +12,7 @@ In each deployment of the first release, one Python process holds the applicatio |---|---|---| | (a) screen to application | the application's own: HTTP for a web page, the channel of the widget library for widgets | yes, when a browser draws the screen | | (b) application to workspace | the client | no | -| (c) workspace to binding | `Binding`, `Driver` | no: 4 threads of the same process | +| (c) workspace to binding | `Binding`, `Driver` | no: threads of the same process, one by default | A screen never calls a workspace. What a screen shows is computed in Python next to the data, and only the result is sent to the screen: a plot function returns a figure, and a form is built from the spec's params model. @@ -27,7 +27,7 @@ So the client is an interface for Python code, and the HTTP interface of a page │ Client │ │ Workspace local(...): history in memory │ │ open_folder(directory): folder │ - │ 4 threads run the bindings │ + │ 1 thread runs the bindings │ └───────────────────────────────────────────────┘ ``` @@ -43,7 +43,7 @@ This deployment is implemented. │ page: │ ──────────▶ │ web server rows, edits, settings (memory) │ │ selection, │ the page's │ batch logic, Client │ │ filter, tab │ own API │ Workspace open_folder(directory) │ - └─────────────┘ │ 4 threads run the bindings │ + └─────────────┘ │ 1 thread runs the bindings │ └───────────────────────────────────────────────┘ folder: files.csv log.jsonl store//.h5 ``` diff --git a/packages/essapps/src/ess/apps/folder.py b/packages/essapps/src/ess/apps/folder.py index c032d2f9..511e2ab2 100644 --- a/packages/essapps/src/ess/apps/folder.py +++ b/packages/essapps/src/ess/apps/folder.py @@ -134,6 +134,7 @@ def open_folder( files: FileSource | None = None, proposal: str = 'local', submitter: str = 'user', + workers: int = 1, ) -> Client: """ A client of the workspace kept in ``directory`` (see the module), running the @@ -144,12 +145,14 @@ def open_folder( wrapper of that, such as :class:`~ess.apps.batch.CorrectedFiles`, to correct what ``files.csv`` says while the client is open. Raises ``FileNotFoundError`` without ``files.csv``, and ``RuntimeError`` if another - process holds the folder. + process holds the folder. ``workers`` is the number of threads that run + requests (see :class:`~ess.apps.Workspace`). """ workspace = Workspace( proposal=proposal, files=folder_files(directory) if files is None else files, bind=bind, + workers=workers, log=Log(directory / 'log.jsonl'), store=FolderStore(directory / 'store'), ) diff --git a/packages/essapps/src/ess/apps/workspace.py b/packages/essapps/src/ess/apps/workspace.py index 58bb6427..d4c40952 100644 --- a/packages/essapps/src/ess/apps/workspace.py +++ b/packages/essapps/src/ess/apps/workspace.py @@ -513,7 +513,10 @@ class Workspace(WorkspaceProtocol): spec it runs; each binding names its spec. Workflows, the steps of accumulators, and writes share ``workers`` threads. A long step or workflow holds one of them, so other work waits only once every worker is - busy; submissions are accepted meanwhile. ``log`` holds history, in + busy; submissions are accepted meanwhile. The default is one thread: a + workflow that is parallel inside gains nothing from more, and the peak + memory grows with each request run at once. The best number depends on the + instrument, so an application may set it. ``log`` holds history, in memory if not given, and ``store`` is where persisted outputs are written. A workspace started on an existing log needs the store the earlier workspace wrote to: without one, the persisted work it goes on with fails, @@ -528,7 +531,7 @@ def __init__( proposal: str, files: FileSource, bind: Sequence[Binding], - workers: int = 4, + workers: int = 1, log: Log | None = None, store: Store | None = None, ) -> None: @@ -1962,6 +1965,7 @@ def local( bind: Sequence[Binding], submitter: str = 'user', store: Store | None = None, + workers: int = 1, ) -> Client: """ A client of its own workspace in this process, running the workflows bound here. @@ -1976,7 +1980,10 @@ def local( persists nothing, and waits until what is persisted is written. A client that is never closed ends with its process. Without a ``store``, persisting is refused. The workspace keeps history in memory, so its records end with the - process. + process. ``workers`` is the number of threads that run requests (see + :class:`Workspace`). """ - workspace = Workspace(proposal=proposal, files=files, bind=bind, store=store) + workspace = Workspace( + proposal=proposal, files=files, bind=bind, workers=workers, store=store + ) return Client(workspace, submitter=submitter, on_close=workspace.close) diff --git a/packages/essapps/tests/client/scheduling_test.py b/packages/essapps/tests/client/scheduling_test.py index 9556789c..bc876cc4 100644 --- a/packages/essapps/tests/client/scheduling_test.py +++ b/packages/essapps/tests/client/scheduling_test.py @@ -7,13 +7,23 @@ import threading import time from concurrent.futures import ThreadPoolExecutor +from typing import Any import pytest -from ess.apps import Client, Record, RecordFailed, Request, Status, Workspace, file +from ess.apps import ( + Client, + Record, + RecordFailed, + Request, + Status, + Workspace, + file, + local, +) from ess.apps.log import Event, Log from ess.apps.testing import FakeFiles -from ess.spec import Binding +from ess.spec import Binding, function_binding from . import toys @@ -111,6 +121,35 @@ def test_a_workflow_that_returns_no_outputs_instance_fails(client: Client) -> No ) +@pytest.mark.parametrize(('given', 'at_once'), [({}, 1), ({'workers': 2}, 2)]) +def test_a_workspace_runs_as_many_requests_at_once_as_it_has_workers( + files: FakeFiles, given: dict[str, Any], at_once: int +) -> None: + started = 0 + running = 0 + most = 0 + changed = threading.Condition() + + def load(params: toys.Sample) -> toys.LoadOutputs: + nonlocal started, running, most + with changed: + started += 1 + running += 1 + most = max(most, running) + changed.notify_all() + # waits for the other request, which starts only on another thread + changed.wait_for(lambda: started == 2, timeout=0.5) + with changed: + running -= 1 + return toys.load_run(params) + + bind = [function_binding(toys.LOAD, load)] + with local(proposal='p1', files=files, bind=bind, **given) as client: + client.compute([toys.load_request(1), toys.load_request(2)]) + + assert most == at_once + + def test_cancel_ends_what_has_not_started( client: Client, loading: threading.Event ) -> None: diff --git a/packages/essapps/tests/persist_test.py b/packages/essapps/tests/persist_test.py index 84f4e6a1..ed033a86 100644 --- a/packages/essapps/tests/persist_test.py +++ b/packages/essapps/tests/persist_test.py @@ -230,14 +230,23 @@ def log() -> Log: return Log() +@pytest.fixture +def workers() -> int: + """The workspace's threads; a test that needs requests to run at once sets more.""" + return 1 + + @pytest.fixture def workspace( files: FakeFiles, store: GatedStore, bind: list[Binding], log: Log, + workers: int, ) -> Iterator[Workspace]: - workspace = Workspace(proposal='p1', files=files, bind=bind, store=store, log=log) + workspace = Workspace( + proposal='p1', files=files, bind=bind, workers=workers, store=store, log=log + ) yield workspace store.gate.set() workspace.close() @@ -455,12 +464,22 @@ def test_references_to_one_persisted_output_read_it_from_the_store_once( def start( files: FakeFiles, bind: list[Binding], store: GatedStore ) -> Iterator[Callable[[Path], Client]]: - """A client of a new workspace over the log at a path, with the same store.""" + """ + A client of a new workspace over the log at a path, with the same store, and + with ``workers`` threads. + """ workspaces: list[Workspace] = [] - def start(path: Path) -> Client: + def start(path: Path, workers: int = 1) -> Client: workspaces.append( - Workspace(proposal='p1', files=files, bind=bind, log=Log(path), store=store) + Workspace( + proposal='p1', + files=files, + bind=bind, + workers=workers, + log=Log(path), + store=store, + ) ) return Client(workspaces[-1], submitter='anna') @@ -646,11 +665,12 @@ def test_a_restart_fails_persisted_work_that_reads_what_is_gone( def test_a_restart_runs_a_call_through_a_stage_as_its_plain_request( tmp_path: Path, - start: Callable[[Path], Client], + start: Callable[..., Client], restart: Callable[[Path], Client], loading: threading.Event, ) -> None: - first = start(tmp_path / 'log') + # a second thread runs the stage while the first waits in LOAD + first = start(tmp_path / 'log', workers=2) kept = first.compute(LOAD, params=_sample(1)) stored = first.compute(LOAD, params=_sample(2), persist=('value',)) loading.clear() @@ -732,6 +752,7 @@ def test_reading_every_output_gives_those_persisted( assert connect().output(load) == expected +@pytest.mark.parametrize('workers', [2]) # SHIFT runs while LOAD waits def test_a_persisted_value_is_read_from_the_store_while_a_reader_holds_it( client: Client, connect: Callable[[], Client],