Run parallel workers#

Parallel workers need the hosted service, which serializes concurrent asks and observations. The local driver keeps its state in one process and is not thread-safe (plain-dict state, no locks), so run one thread per Client(driver="local"); without outbound HTTPS to the hosted service, one local-driver process per experiment is the only shape, because sharing one experiment across processes or nodes is hosted-only. The hosted procedure on this page is verified by reading python/examples/parallel_workers.py and python/tauopt/_http.py rather than by execution; the last section runs locally.

Closed beta (nothing installed): the same loop over plain HTTP is parallel_workers_rest.py: with TAUOPT_BASE_URL and TAUOPT_TOKEN exported, python parallel_workers_rest.py --workers 4 runs N threads on one experiment created with parallel_bandwidth N, each looping POST .../suggestions -> evaluate -> POST .../observations; a 429 bandwidth_full is waited out as backpressure and every wait ends at --deadline-s.

Parallel workers on one experiment. Three workers share one experiment id, each with its own Client and its own worker tag (1). Every ask passes the token check and the quotas (2) and then the slot check: with both of the experiment’s parallel_bandwidth slots held by open suggestions, the third worker’s ask is refused with 429 bandwidth_full and Retry-After: 1, backpressure rather than a fault, and the worker waits a second and asks again (3). The experiment store takes asks and observations one at a time and keeps the open set, one slot per open suggestion, each carrying its worker’s tag (4). A slot returns when its suggestion is observed, dismissed with close_suggestion or close_all_suggestions, or expired by suggestion_ttl_s at the experiment’s next ask; worker 2 died between suggest and observe, so its suggestion stays open and its slot stays held until a dismissal frees it (5).#

Run from a batch job array#

A job array is N tasks that start and die on the scheduler’s clock, and the hosted experiment outlives all of them.

Create the experiment once, outside the array, with parallel_bandwidth equal to the array width, and record exp.id. Hand the id to every task (an exported variable in the job script does it) and pass it to the example as --experiment-id <id>, so each task runs the worker loop against the shared experiment instead of creating its own.
Keep the token out of argv and out of the job script’s text: put the two export lines in a file outside the repository with mode 0600 and source it from the job script, or use the scheduler’s secret mechanism. Queue-listing tools show every task’s command line.
Before resubmitting an array whose tasks were killed, free the slots their open suggestions still hold: client.close_all_suggestions(exp.id) dismisses every open suggestion and returns the count, or list client.suggestions(exp.id, state="open") and evaluate the ones you can still finish. Otherwise the new tasks wait on 429 bandwidth_full until their deadline.
Give the compute nodes outbound HTTPS to the service’s base URL. The client never retries a connection failure: httpx.ConnectError raises raw.
Export OMP_NUM_THREADS, MKL_NUM_THREADS, and OPENBLAS_NUM_THREADS at the task’s core count for your own evaluation code. The hosted client is a thin HTTP client and needs none of them. The local driver’s Gaussian-process (GP) strategies pin torch to one thread for the duration of a suggest and restore the previous count afterwards, so a suggest never oversubscribes a shared node; the variables govern only what your evaluation does.
Bound every wait with a deadline inside the task’s walltime and exit nonzero naming the error type when it expires, so a stuck task is visible in the scheduler rather than in a silent loop.

The local driver is the wrong tool for an array: its state lives in process memory, nothing is written to disk, and a killed task loses the experiment with it. The recovery recipe for a local run, replaying a log of every observe, is in Troubleshooting.

Recovering after a worker dies#

A worker that dies between suggest and observe leaves its suggestion open: the slot stays held until something observes or dismisses it, and the service cannot tell which worker died. Give it what it needs to tell you:

Tag every suggestion with the worker’s id, keyed by the job as well as the rank: client.suggest(exp.id, tag=f"{job_id}-worker-{rank}"), so a resubmitted array never finds the previous run’s rows under its own tag. The tag is a label of 1 to 128 printable characters, served back as Suggestion.tag on every read and never a key for anything (no quota, no admission).
On restart, list the open suggestions that carry your tag: client.suggestions(exp.id, state="open", tag=f"{job_id}-worker-{rank}") (on the hosted plane the list is a history-sized read: it draws the credential’s heavy-read sub-bucket, 15 a second with a burst of 30, so a restart loop polls it once, never in a tight loop). Each row carries its created time; evaluate and observe the ones still worth finishing. The listing is a snapshot: expiry is lazy, so a listed row past the experiment’s TTL stays open until a sibling’s next ask closes it, and an observe that loses that race answers 409 suggestion_expired whose details.assignments carry the point – re-post the result as an unsolicited observation (client.observe(exp.id, assignments=..., value=...)); on the hosted plane that answer is part of the recipe, not a fault.
Dismiss the ones you cannot finish: client.close_suggestion(exp.id, s.id) frees one slot; client.close_all_suggestions(exp.id) is the blunt form that frees every open slot, the other workers’ included.
Set suggestion_ttl_s on the experiment to the job’s horizon (create_experiment(..., suggestion_ttl_s=...) or update_experiment), so an orphan nobody picks up expires on its own at the experiment’s next ask (the hosted plane; the local driver records the value and never expires).

The hosted plane serves the tag (every suggestion the service answers carries "tag", null when the ask set none). The same loop against the local driver – an in-process simulation: the local driver never expires a suggestion, so the race above cannot happen here, and the assert lines are what the fence proves:

from tauopt import Client

client = Client(driver="local")
exp = client.create_experiment(
    name="tagged-workers",
    parameters=[{"name": "x", "type": "double", "bounds": (0.0, 1.0)}],
    metric={"name": "y", "objective": "minimize"},
    parallel_bandwidth=2,
)
mine = client.suggest(exp.id, tag="worker-1")    # this worker's
other = client.suggest(exp.id, tag="worker-2")   # a sibling's
# worker-1 dies and restarts: its own open work, by its tag (the sibling's row is not listed)
open_mine = client.suggestions(exp.id, state="open", tag="worker-1")
assert [s.id for s in open_mine] == [mine.id] and open_mine[0].tag == "worker-1"
recovered = open_mine[0]
client.observe(exp.id, suggestion_id=recovered.id, value=0.5)   # finish the recovered row
assert client.suggestions(exp.id, state="open", tag="worker-1") == []
print(recovered.tag, len(client.suggestions(exp.id, state="open")))

worker-1 1

Know what the transport retries for you#

The hosted transport makes at most 5 attempts per request:

429 is retried on every method: the server rejected the request and nothing was committed.
500, 502, 503, and 504 are retried on non-POST methods only; a POST that answers 5xx raises after its first attempt.
Each delay honors Retry-After when present, capped at 8 seconds; otherwise it doubles from 0.5 seconds (0.5, 1, 2, 4), capped at 8.
Connection-level failures (httpx.ConnectError, httpx.ReadTimeout) raise raw and are never retried; redirects are never followed.

After the last attempt, the raised ApiError carries the final response’s headers as e.headers and the parsed delay as e.retry_after (None when absent or malformed) for your own backoff loop. A hosted quota 429 (1,000 suggestions per rolling 24 hours by default) carries Retry-After; the active-experiment cap 429 does not, and waiting on it does not help (see Archive experiments).

Check for duplicate assignments#

Within one experiment the engine never re-suggests an observed or still-open point, so concurrent workers receive distinct assignments; the example records each assignment as a sorted tuple in a Counter, asserts that none repeats and that no more than N were ever open at once, and exits nonzero otherwise. A suggestion closed without an observation leaves the dedup universe (the observed and open points the engine refuses to repeat) and may come back later; see Dedup and completion.

Reproduce the backpressure signal without a service#

The local driver mirrors the bandwidth check, so the exact error shape, and a worker loop around it, run in-process. Its ApiError carries no headers, so retry_after is None and the loop falls back to its fixed delay.

import time

from tauopt import ApiError, Client


def suggest_with_backoff(client, exp_id, deadline_s):
    deadline = time.monotonic() + deadline_s
    while True:
        try:
            return client.suggest(exp_id)
        except ApiError as e:
            if e.status_code != 429:
                raise
            # a held slot frees within seconds; a quota or rate limit takes longer
            wait = e.retry_after or (0.5 if e.error_type == "bandwidth_full" else 5.0)
            if time.monotonic() + wait > deadline:
                raise RuntimeError(f"gave up on {e.error_type} after {deadline_s} s") from e
            time.sleep(wait)


client = Client(driver="local")
exp = client.create_experiment(
    name="one-slot",
    parameters=[{"name": "x", "type": "double", "bounds": (0.0, 1.0)}],
    metric={"name": "y", "objective": "minimize"},
    parallel_bandwidth=1,
)
held = client.suggest(exp.id)          # holds the only slot
try:
    client.suggest(exp.id)             # a second ask while the first is open
except ApiError as e:
    print(e.status_code, e.error_type, "-", e.message)
    print("retry_after:", e.retry_after, "headers:", dict(e.headers))
try:
    suggest_with_backoff(client, exp.id, deadline_s=1.0)
except RuntimeError as e:
    print(e, "| cause:", e.__cause__.error_type)
client.observe(exp.id, suggestion_id=held.id, value=0.5)   # frees the slot
print("after observe:", suggest_with_backoff(client, exp.id, deadline_s=1.0).state)

429 bandwidth_full - 1 open suggestions >= parallel_bandwidth=1; observe or close one first
retry_after: None headers: {}
gave up on bandwidth_full after 1.0 s | cause: bandwidth_full
after observe: open