Skip to main content

Run batches on a pool of Dask workers

With the LSF runner, each run is an LSF job: it pays LSF's scheduling and allocation overhead, and a batch can flood the cluster. With the dask runner, a batch's runs execute on a few long-lived Dask workers, with a bounded concurrency. By default the workers are themselves LSF jobs, started for you with dask-jobqueue's LSFCluster.

When to use it
  • Many short runs (seconds to a few minutes), where LSF overhead dominates.
  • You want to cap how many runs a batch uses at the same time.

For long runs, or to only cap concurrency, the LSF runner's concurrency option is simpler.

Setup​

Install the optional dependencies (Python 3.9 or later):

pip install "qaboard[dask]"
uv add "qaboard[dask]"   # or, in a uv project

Workers run qa run commands: they must use the same python environment as qa batch, and see the same filesystem (as for LSF).

Configuration​

qaboard.yaml
runners:
  default: dask
  lsf:
    queue: your_queue      # used by default for the worker jobs (also: project, resources, max_memory, options)
  dask:
    concurrency: 20        # max runs in flight for a `qa batch` (default: 10)
    cluster:               # arguments for dask_jobqueue.LSFCluster, they override the defaults
      cores: 4             # runs per worker job, each run is a thread in the worker (default: 1)
      memory: 32GB         # per worker job: it must cover `cores` runs (default: runners.lsf.max_memory, or 4GB)
      # walltime: "24:00"  # by default the queue decides
      # interface: ib0     # if workers can't reach the scheduler's default address
    driver:                # for --no-wait: LSF options of the job that runs the scheduler (default: runners.lsf)
      queue: some_cpu_queue          # it needs little: 1 CPU, it sleeps most of the time
      max_memory: 2000               # MB
      # resources: "select[...]"
    # no_worker_timeout: 600        # give up if all workers are gone for that long (seconds)
    # scheduler_address: tcp://my-host:8786   # use an existing cluster, see below

You can also choose the runner and concurrency on the CLI:

qa batch --runner dask --dask-concurrency 8 my-batch

A batch can have its own concurrency under dask:, e.g. for runs that need a lot of memory: at most that many of its runs are in flight. Its runs share the workers with the other batches' runs, and qa batch keeps at most the largest concurrency in flight, and starts workers for that many. The command line option wins.

How it works​

  1. qa batch starts a Dask scheduler in its own process.
  2. It bsubs ceil(min(#runs, concurrency) / cores) worker jobs, named {batch-prefix}_dask.
  3. It keeps at most concurrency runs in flight: a new run starts as soon as one finishes.
  4. When all runs are done, the worker jobs are stopped.

With --no-wait, steps 1-4 happen in a "driver" LSF job named {batch-prefix}_dask_driver, and qa batch returns as soon as it is sent. The driver uses the LSF options from runners.dask.driver, so it doesn't take e.g. a GPU slot.

Logs​

  • Each run's output is saved in its output directory as log.dask.txt (in addition to the usual log.txt).
  • Everything else is in the batch directory, under dask/{batch-prefix}/:
    • worker logs: *.err has Dask's logs, and a [qaboard] start / [qaboard] end line per run. *.out is usually empty.
    • with --no-wait: driver.log (live progress), driver.lsf.log (LSF's report, at the end) and tasks.json (the runs it was asked to execute).
  • Your environment variables are forwarded to the runs, except those specific to your host or LSF job (LSB_*, TMPDIR, CUDA_VISIBLE_DEVICES...).
  • Stopping a batch from QA-Board kills the worker and driver jobs (bkill -J "{batch-prefix}*"). Ctrl-C on qa batch kills the running runs and the workers.
  • If a worker job dies while running something (e.g. killed by LSF for using too much memory), Dask retries the run on another worker, up to 3 times.
Requirements & limitations
  • Workers open a TCP connection to the host running qa batch. It's usually fine within a cluster; otherwise set cluster.interface.
  • Without --no-wait, bsub must be available where qa batch runs: LSF "bridges" are not supported. With --no-wait, the driver is sent via the bridge like other LSF jobs, and it then starts the workers itself.
  • All workers get the same LSF options: per-input-type LSF options (e.g. more memory for some inputs) are not applied. Size memory for your heaviest runs.
  • Worker jobs are kept until the batch ends, even if idle while the last runs finish.

Using an existing Dask cluster​

If you keep a Dask cluster running (e.g. dask scheduler on a host, and dask worker tcp://host:8786 sent with bsub), point to it:

qaboard.yaml
runners:
  dask:
    scheduler_address: tcp://my-host:8786
    concurrency: 20
  • qa batch doesn't start or stop workers: the cluster size is the global limit, concurrency the limit of each qa batch.
  • With --no-wait, the driver LSF job connects to the cluster and keeps at most concurrency runs in flight. If LSF isn't available (no bsub, no QA_RUNNERS_LSF_BRIDGE), all runs are sent at once and concurrency is not enforced.
  • To stop batches from QA-Board, the QA-Board server needs distributed installed and access to the scheduler.

Troubleshooting​

  • Nothing starts: check the jobs with bjobs -J "{batch-prefix}_dask*", and the logs in {batch-dir}/dask/{batch-prefix}/.
  • Workers start but runs don't: check they reached the scheduler (Registered to: in the worker's .err), then the [qaboard] start lines, then the runs' log.dask.txt.
  • QA_BATCH_VERBOSE=1 qa batch ... prints the worker job script.