perf: one workspace thread by default

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.

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/<record>/<output>.h5
 ```

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'),
     )

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)

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:

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],