slurminade package

Module contents

slurminade allows to distribute function calls to slurm using decorators.

import slurminade

slurminade.update_default_configuration(partition="alg")  # global options for slurm

# If no slurm environment is found, the functions are called directly to make scripts
# compatible with any environment.
# You can enforce slurm with `slurminade.set_dispatcher(slurminade.SlurmDispatcher())`


# use this decorator to make a function distributable with slurm
@slurminade.slurmify(
    constraint="alggen02"
)  # function specific options can be specified
def prepare():
    print("Prepare")


@slurminade.slurmify()
def f(foobar):
    print(f"f({foobar})")


@slurminade.slurmify()
def clean_up():
    print("Clean up")


if __name__ == "__main__":
    jid = prepare.distribute()

    with slurminade.Batch(max_size=20) as batch:  # automatically bundles up to 20 tasks
        # run 100x f after `prepare` has finished
        for i in range(100):
            f.wait_for(jid).distribute(i)

        # clean up after the previous jobs have finished
        jids = batch.flush()
        clean_up.wait_for(jids).distribute()

Project structure: - batch.py: Contains code for bundling tasks, so we don’t spam slurm with too many. - conf.py: Contains code for managing the configuration of slurm. - dispatcher.py: Contains code for actually dispatching tasks to slurm. - execute.py: Contains code to execute the task on the slurm node. - function.py: Contains the code for making a function slurm-compatible. - function_map.py: Saves all the slurified functions. - guard.py: Contains code to prevent you accidentally DDoSing your infrastructure. - options.py: Contains a simple data structure to save slurm options.

class slurminade.Batch(*args, **kwargs)

Bases: JobBundling

Compatibility alias for JobBundling. This is the old name. Deprecated.

class slurminade.JobBundling(max_size: int)

Bases: Dispatcher

The logic to buffer the function calls. It wraps the original dispatcher.

You can use:

with slurminade.Batch(max_size=20) as batch:  # automatically bundles up to 20 tasks
    # run 100x f
    for i in range(100):
        f.distribute(i)

to automatically bundle up to 20 tasks and distribute them.

add(func: SlurmFunction, *args, **kwargs)

You can also add a task using add instead of distribute. :param func: The function to call :param args: The positional arguments :param kwargs: The keywords arguments. :return: None

flush() → List[JobReference]

Distribute all buffered tasks. Return the jobs used. This method is called automatically when the context is exited. However, you may want to call it manually to get the job references, for example to use them for dependency management with wait_for. :param options: Only flush tasks with specific options. :return: A list of job references.

get_all_job_ids() → List[int]

Return all job ids that have been used.

get_all_jobs() → List[JobReference]

is_sequential()

Return true if the dispatcher works sequential. In this case, the dependencies are trivially fulfilled. Slurm does not work sequentially, because this would destroy its purpose. In some cases however, you do not want to use slurm for compatibility reasons, without changing the script. In these cases, this function tells slurminade not to be too strict about dependencies. :return: True is tasks are executed sequentially, false if not.

join()

sbatch(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None) → JobReference

Define how you want to execute an sbatch command. The command is scheduled and the function return immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

srun(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None) → JobReference

Define how you want to execute an srun command. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

class slurminade.SlurmDispatcher

Bases: Dispatcher

The most important dispatcher: Distributing function calls to slurm.

join()

sbatch(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None) → SlurmJobReference

Define how you want to execute an sbatch command. The command is scheduled and the function return immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

srun(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None) → SlurmJobReference

Define how you want to execute an srun command. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

class slurminade.SubprocessDispatcher

Bases: Dispatcher

A dispatcher for debugging that distributes function calls using subprocesses. Thus, it uses the same serialization mechanisms, but without a slurm dependency. Completely useless for productive purposes. Use DirectCallDispatcher if you don’t want to use slurm. Despite using subprocesses, it does not parallelize but works sequential.

is_sequential()

Return true if the dispatcher works sequential. In this case, the dependencies are trivially fulfilled. Slurm does not work sequentially, because this would destroy its purpose. In some cases however, you do not want to use slurm for compatibility reasons, without changing the script. In these cases, this function tells slurminade not to be too strict about dependencies. :return: True is tasks are executed sequentially, false if not.

sbatch(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None)

Define how you want to execute an sbatch command. The command is scheduled and the function return immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

srun(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None)

Define how you want to execute an srun command. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

class slurminade.TestDispatcher

Bases: Dispatcher

A dummy dispatcher that just prints the output. Primarily for debugging and testing.

is_sequential()

Return true if the dispatcher works sequential. In this case, the dependencies are trivially fulfilled. Slurm does not work sequentially, because this would destroy its purpose. In some cases however, you do not want to use slurm for compatibility reasons, without changing the script. In these cases, this function tells slurminade not to be too strict about dependencies. :return: True is tasks are executed sequentially, false if not.

sbatch(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None)

Define how you want to execute an sbatch command. The command is scheduled and the function return immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

srun(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None)

Define how you want to execute an srun command. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

slurminade.allow_recursive_distribution() → None

Allow recursive distribution. Dangerous! :return: None

slurminade.disable_warning_on_repeated_flushes()

Disable the warning on multiple flushes. This is useful if you want to flush multiple times in a loop, without getting a warning.

slurminade.get_dispatcher() → Dispatcher

Returns the current dispatcher. Creates a dispatcher if none is available. First tries to create the slurm-dispatcher (as this is the primary purpose of slurminade). If no slurm-environment can be found, it creates a DirectCallDispatcher to allow compatibility. :return: The dispatcher.

slurminade.join()

Join all jobs that have been dispatched so far. :return: None

slurminade.node_setup(func: Callable)

Decorator: Call this function on the node before running any function calls.

slurminade.sbatch(command: str | List[str], conf: SlurmOptions | Dict | None = None, simple_slurm_kwargs: Dict | None = None) → JobReference

The command is scheduled and the function returns immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

slurminade.set_default_configuration(conf=None, **kwargs)

Replaces the default configuration. This will overwrite the default configuration with the given one. :param conf: A dictionary with the configuration. :param kwargs: Configuration parameters. (alternative to giving a dictionary)

slurminade.set_dispatch_limit(n: int | None)

Set a limit to the number of dispatches. This feature has been introduced to prevent you from accidentally DDoSing you Slurm environment due to a bug. :param n: The maximal number of dispatches. :return: None

slurminade.set_dispatcher(dispatcher: Dispatcher) → None

Replaces the dispatcher. Can be used to enforce a specific dispatcher. :param dispatcher: The dispatcher to be used. :return: None

slurminade.set_entry_point(entry_point: str | Path) → None

This function usually is not necessary for endusers. Set a manual entry point. This can allow you to use slurmify from the interactive interpreter. :param entry_point: A path to the entry point file. :return: None

slurminade.slurmify(f=None, **args) → Callable[[Callable], SlurmFunction] | SlurmFunction

Decorator: Make a function distributable to slurm. Usage:

@slurmify()
def func(a, b):
    pass

Parameters:

• f – Function

• args – Special slurm options for this function.

Returns:

A decorated function, callable with slurm.

slurminade.srun(command: str | List[str], conf: SlurmOptions | Dict | None = None, simple_slurm_kwargs: Dict | None = None) → JobReference

Call srun with the current dispatcher. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

slurminade.update_default_configuration(conf=None, **kwargs)

Adds or updates the default configuration. :param conf: A dictionary with the configuration. :param kwargs: Configuration parameters. (alternative to giving a dictionary)

Submodules

slurminade.batch module

slurminade.conf module

This file saves the default configuration for slurm.

slurminade.conf.set_default_configuration(conf=None, **kwargs)

Replaces the default configuration. This will overwrite the default configuration with the given one. :param conf: A dictionary with the configuration. :param kwargs: Configuration parameters. (alternative to giving a dictionary)

slurminade.conf.update_default_configuration(conf=None, **kwargs)

Adds or updates the default configuration. :param conf: A dictionary with the configuration. :param kwargs: Configuration parameters. (alternative to giving a dictionary)

slurminade.dispatcher module

The dispatcher distribute function calls to slurm or the local machine. It can be accessed with get_dispatcher and set with set_dispatcher. This allows to change the behaviour of the distribution, e.g., we use it for batch: Batch simply wraps the dispatcher by a buffered version.

class slurminade.dispatcher.DirectCallDispatcher

Bases: Dispatcher

A dispatcher that calls functions as if we would not use slurminade. This allows compatibility of scripts also on computers not integrated into the slurm network.

is_sequential()

Return true if the dispatcher works sequential. In this case, the dependencies are trivially fulfilled. Slurm does not work sequentially, because this would destroy its purpose. In some cases however, you do not want to use slurm for compatibility reasons, without changing the script. In these cases, this function tells slurminade not to be too strict about dependencies. :return: True is tasks are executed sequentially, false if not.

sbatch(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None)

Define how you want to execute an sbatch command. The command is scheduled and the function return immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

srun(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None)

Define how you want to execute an srun command. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

class slurminade.dispatcher.Dispatcher

Bases: ABC

Abstract dispatcher to be inherited by all concrete dispatchers. For implementing a dispatcher you have to implement _dispatch, srun and sbatch.

is_sequential()

Return true if the dispatcher works sequential. In this case, the dependencies are trivially fulfilled. Slurm does not work sequentially, because this would destroy its purpose. In some cases however, you do not want to use slurm for compatibility reasons, without changing the script. In these cases, this function tells slurminade not to be too strict about dependencies. :return: True is tasks are executed sequentially, false if not.

join()

abstract sbatch(command: str, conf: SlurmOptions | None = None, simple_slurm_kwargs: Dict | None = None) → JobReference

Define how you want to execute an sbatch command. The command is scheduled and the function return immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

abstract srun(command: str, conf: SlurmOptions | None = None, simple_slurm_kwargs: Dict | None = None) → JobReference

Define how you want to execute an srun command. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

class slurminade.dispatcher.LocalJobReference

Bases: JobReference

get_exit_code() → None

get_info() → Dict[str, Any]

get_job_id() → None

class slurminade.dispatcher.SlurmDispatcher

Bases: Dispatcher

The most important dispatcher: Distributing function calls to slurm.

join()

sbatch(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None) → SlurmJobReference

Define how you want to execute an sbatch command. The command is scheduled and the function return immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

srun(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None) → SlurmJobReference

Define how you want to execute an srun command. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

class slurminade.dispatcher.SlurmJobReference(job_id, exit_code, mode: str)

Bases: JobReference

get_exit_code() → int | None

get_info() → Dict[str, Any]

get_job_id() → int

class slurminade.dispatcher.SubprocessDispatcher

Bases: Dispatcher

A dispatcher for debugging that distributes function calls using subprocesses. Thus, it uses the same serialization mechanisms, but without a slurm dependency. Completely useless for productive purposes. Use DirectCallDispatcher if you don’t want to use slurm. Despite using subprocesses, it does not parallelize but works sequential.

is_sequential()

Return true if the dispatcher works sequential. In this case, the dependencies are trivially fulfilled. Slurm does not work sequentially, because this would destroy its purpose. In some cases however, you do not want to use slurm for compatibility reasons, without changing the script. In these cases, this function tells slurminade not to be too strict about dependencies. :return: True is tasks are executed sequentially, false if not.

sbatch(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None)

Define how you want to execute an sbatch command. The command is scheduled and the function return immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

srun(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None)

Define how you want to execute an srun command. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

class slurminade.dispatcher.SubprocessJobReference

Bases: JobReference

get_exit_code() → int | None

get_info() → Dict[str, Any]

get_job_id() → int | None

class slurminade.dispatcher.TestDispatcher

Bases: Dispatcher

A dummy dispatcher that just prints the output. Primarily for debugging and testing.

is_sequential()

Return true if the dispatcher works sequential. In this case, the dependencies are trivially fulfilled. Slurm does not work sequentially, because this would destroy its purpose. In some cases however, you do not want to use slurm for compatibility reasons, without changing the script. In these cases, this function tells slurminade not to be too strict about dependencies. :return: True is tasks are executed sequentially, false if not.

sbatch(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None)

Define how you want to execute an sbatch command. The command is scheduled and the function return immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

srun(command: str, conf: Dict | None = None, simple_slurm_kwargs: Dict | None = None)

Define how you want to execute an srun command. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

class slurminade.dispatcher.TestJobReference

Bases: JobReference

get_exit_code() → None

get_info() → Dict[str, Any]

get_job_id() → None

slurminade.dispatcher.dispatch(funcs: FunctionCall | Iterable[FunctionCall], options: SlurmOptions, entry_point: Path, block: bool = False) → JobReference

Distribute function calls with the current dispatcher. :param funcs: The functions calls to be distributed. :param options: The slurm options to be used. :return: The job id.

slurminade.dispatcher.get_dispatcher() → Dispatcher

Returns the current dispatcher. Creates a dispatcher if none is available. First tries to create the slurm-dispatcher (as this is the primary purpose of slurminade). If no slurm-environment can be found, it creates a DirectCallDispatcher to allow compatibility. :return: The dispatcher.

slurminade.dispatcher.join()

Join all jobs that have been dispatched so far. :return: None

slurminade.dispatcher.sbatch(command: str | List[str], conf: SlurmOptions | Dict | None = None, simple_slurm_kwargs: Dict | None = None) → JobReference

The command is scheduled and the function returns immediately. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id.

slurminade.dispatcher.set_dispatcher(dispatcher: Dispatcher) → None

Replaces the dispatcher. Can be used to enforce a specific dispatcher. :param dispatcher: The dispatcher to be used. :return: None

slurminade.dispatcher.srun(command: str | List[str], conf: SlurmOptions | Dict | None = None, simple_slurm_kwargs: Dict | None = None) → JobReference

Call srun with the current dispatcher. This command is directly executed and only terminates after completion. :param command: A system command, e.g. echo hello world > foobar.txt. :param conf: The slurm configuration. :param simple_slurm_kwargs: Additional options for simple_slurm. :return: Job id

slurminade.execute module

This module provides the starting point for the slurm node. You do not have to call anything of this file yourself.

slurminade.function module

class slurminade.function.CallPolicy(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Policy for the call of a function.

DISTRIBUTED = 1

DISTRIBUTED_BLOCKING = 2

LOCALLY = 0

class slurminade.function.SlurmFunction(special_slurm_opts: Dict, func: Callable, func_id: str, call_policy: CallPolicy = CallPolicy.LOCALLY)

Bases: object

A wrapper around a function that allows it to be distributed to slurm.

@slurmify(function specific slurm options)
def f(foobar):
    print(foobar)

if __name__=="__main__":
    assert isinstance(f, SlurmFunction), "f has become a SlurmFunction"
    jid = f.distribute("hello")
    f.wait_for(jid).distribute("bye")

static call(func_id, *args, **kwargs)

distribute(*args, **kwargs) → JobReference

Try to distribute function call. If slurm is not available, a direct function call will be performed. f.distribute(“hello”) Call with function arguments. :return: Job id. Not necessarily valid (usually -1 in this case).

distribute_and_wait(*args, **kwargs) → JobReference

Distribute the function and wait for it to finish. :param args: The positional arguments. :param kwargs: The keyword arguments. :return: The job id.

get_entry_point() → Path

Returns the entry point for the function. Either it is defined in the FunctionMap, or the defining file is used.

run_locally(*args, **kwargs)

Call the function locally. f.local(“hello”) Call with function arguments. :return: The return value of the function.

update_options(conf: Dict[str, Any])

wait_for(job_ids: JobReference | Iterable[JobReference], method: str = 'afterany') → SlurmFunction

Add a dependency to a distribution. f_jid = f.wait_for(job_ids).distribute(“hello”) f will only be executed once all jobs within job_ids have finished (or failed). Its own job id can then also be used as dependency for other distributions. Note that this method does not manipulate the original function but returns a new function object, so you have to use chaining. Not every dispatcher returns valid job ids on distribute. For example with Batch, you get the job ids with job_ids = batch.flush(). :param job_ids: A single job id or an iterable of job ids. :param method: ‘after’|’afterany’|’afternotok’|’afterok’|’singleton’ :return: Chainable slurm function object.

with_options(**kwargs) → SlurmFunction

Add slurm options to the function. :param kwargs: The slurm options. :return: The modified function.

slurminade.function.slurmify(f=None, **args) → Callable[[Callable], SlurmFunction] | SlurmFunction

Decorator: Make a function distributable to slurm. Usage:

@slurmify()
def func(a, b):
    pass

Parameters:

• f – Function

• args – Special slurm options for this function.

Returns:

A decorated function, callable with slurm.

slurminade.function_map module

The internal datastructure to save all the slurmified functions. Not relevant for endusers.

class slurminade.function_map.FunctionMap

Bases: object

The function map assigns functions an id and stores them to be called later. This id is reproducible such that the slurm node can retrieve the function it is supposed to call just by the id.

static call(func_id: str, args: Iterable, kwargs: Dict[str, Any]) → Any

Calls a function by its id. :param func_id: The id of the function to be called. :param args: The positional arguments. :param kwargs: The keyword arguments. :return: The return value of the function.

static check_compatibility(func: Callable)

Throw if the function cannot be assigned an id. :param func: The function to be checked. :return: None

static check_id(func_id: str, entry_point: Path) → bool

entry_point: str | None = None

static exists(func_id: str) → bool

static get_all_ids() → List[str]

static get_id(func: Callable) → str

Returns the unique id of a function. Necessary to slurmify it. Probably not needed by the enduser. It uses the function name and its file, which should be sufficient unless you do not overwrite a function (which is bad anyway). :param func: The function you want the id of. :return: The id as string.

static get_readable_name(func_id: str) → str

static register(func: Callable, allow_overwrite: bool = False) → str

Register a function, allowing it to be called just by its id. :param func: The function to be stored. Needs to be a proper function. :return: The function’s id.

slurminade.function_map.get_entry_point() → Path

slurminade.function_map.set_entry_point(entry_point: str | Path) → None

This function usually is not necessary for endusers. Set a manual entry point. This can allow you to use slurmify from the interactive interpreter. :param entry_point: A path to the entry point file. :return: None

slurminade.guard module

Some security measures to warn you about common mistakes and prevent you from accidentally DDoSing your slurm environment.

1. Preventing recursive distributions, i.e., slurm nodes also distributing tasks.

2. Limiting the number of distributed tasks.

3. Warn about multiple flushes of batches, often caused by wrong indentation.

You can disable these security mechanisms by allow_recursive_distribution, set_dispatch_limit(None), and disable_warning_for_multiple_flushes.

class slurminade.guard.BatchGuard

Bases: object

Warns you if you flush more than once, as putting the flush call in a loop is a common mistake, compared to the intended use of flushing once at the end of your context, to get the job ids for dependency management.

already_warned = False

report_flush(num_tasks: int) → None

exception slurminade.guard.TooManyDispatchesError(n_calls)

Bases: RuntimeError

slurminade.guard.allow_recursive_distribution() → None

Allow recursive distribution. Dangerous! :return: None

slurminade.guard.disable_warning_on_repeated_flushes()

Disable the warning on multiple flushes. This is useful if you want to flush multiple times in a loop, without getting a warning.

slurminade.guard.guard_recursive_distribution()

slurminade.guard.on_slurm_node()

slurminade.guard.prevent_distribution()

slurminade.guard.set_dispatch_limit(n: int | None)

Set a limit to the number of dispatches. This feature has been introduced to prevent you from accidentally DDoSing you Slurm environment due to a bug. :param n: The maximal number of dispatches. :return: None

slurminade.options module

class slurminade.options.SlurmOptions

Bases: dict

Primarily just a wrapper to allow using the options in a dict as key. Necessary for batching function calls, because only function calls with the same options can be bundled.

add_dependencies(job_ids, method: str = 'afterany')

as_dict() → Dict