molforge.parallel

parallel

Run a callable over many inputs, in parallel, with order and error control.

Every user of an engine ends up writing the same multiprocessing.Pool loop to fold / dock / score a batch of inputs. :func:map_parallel is that loop, once:

from molforge.parallel import map_parallel
structures = map_parallel(engine.predict, sequences, backend="process")

The backend is the one real decision:

• "process" — CPU-bound work that releases no GIL: subprocess engine wrappers (Vina, fpocket), NumPy-heavy analysis. One OS process per worker. func and each item must be picklable.
• "thread" — I/O-bound work: remote fetches (fetch_many), reading files. Cheap, shares memory, no pickling.
• "serial" — one at a time. The right choice for GPU engines, where running several models at once just fights over one device.

Results come back in input order. on_error="skip" drops the inputs that raised (returning fewer results) instead of aborting the whole batch, which is what you want when one bad input shouldn't sink a long run.

map_parallel

map_parallel(
    func: Callable[[_T], _R],
    items: Iterable[_T],
    *,
    workers: int | None = None,
    backend: Backend = "process",
    on_error: OnError = "raise",
) -> list[_R]

Apply func to every item, in parallel, preserving input order.

Parameters:

Name	Type	Description	Default
func	Callable[[_T], _R]	A single-argument callable. For an engine method that takes options, bind them first with :func:functools.partial so what's left is one positional input.	required
items	Iterable[_T]	The inputs. Consumed once into a list.	required
workers	int | None	Number of parallel workers. None uses os.cpu_count(); 1 runs serially regardless of backend.	None
backend	Backend	"process" (default; CPU-bound, needs picklable func/items), "thread" (I/O-bound), or "serial".	'process'
on_error	OnError	"raise" (default) propagates the first failure; "skip" drops failing inputs and returns the rest, logging a warning per failure.	'raise'

Returns:

Type	Description
list[_R]	The results in input order. With on_error="skip" the list is
list[_R]	shorter than items by the number of failures.

fold_many

fold_many(
    engine: FoldingEngine,
    sequences: Sequence[str],
    *,
    workers: int | None = None,
    backend: Backend | None = None,
    on_error: OnError = "raise",
    **kwargs: object,
) -> list[Protein]

Fold many sequences with a folding engine.

Runs engine.predict(sequence, **kwargs) for each sequence. backend defaults to the engine's parallelism hint ("serial" for GPU engines, "process" for CPU ones). See :func:map_parallel for workers / on_error semantics.

dock_many

dock_many(
    engine: DockingEngine,
    receptor: Protein,
    ligands: Iterable[object],
    *,
    workers: int | None = None,
    backend: Backend | None = None,
    on_error: OnError = "raise",
    **kwargs: object,
) -> list[DockingResult]

Dock many ligands against one receptor.

Runs engine.dock(receptor, ligand, **kwargs) for each ligand — the common virtual-screening shape. backend defaults to the engine's parallelism hint (Vina and other CPU engines set "process").

run_many

run_many(
    engine: object,
    items: Iterable[object],
    *,
    method: str,
    workers: int | None = None,
    backend: Backend | None = None,
    on_error: OnError = "raise",
    **kwargs: object,
) -> list[Any]

Generic engine batch: run engine.<method>(item, **kwargs) over items.

The escape hatch for modalities without a dedicated wrapper (e.g. an MD engine's run): run_many(md_engine, systems, method="run"). Each item is passed as the first positional argument. backend defaults to the engine's parallelism hint.