Skip to main content

Running 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+):

pip install "qaboard[dask]"
# or
uv sync --extra dask

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

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 per-batch limit.
  • 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.