# aiida.engine.processes.calcjobs package¶

Module for the CalcJob process and related utilities.

class aiida.engine.processes.calcjobs.CalcJob(*args: Any, **kwargs: Any)[source]

Implementation of the CalcJob process.

__abstractmethods__ = frozenset({})
__annotations__ = {'link_label_retrieved': <class 'str'>}
__init__(*args, **kwargs)None[source]

Construct a CalcJob instance.

Construct the instance only if it is a sub class of CalcJob, otherwise raise InvalidOperation.

See documentation of aiida.engine.Process.

__module__ = 'aiida.engine.processes.calcjobs.calcjob'
_abc_impl = <_abc_data object>
_node_class
_spec = <aiida.engine.processes.process_spec.CalcJobProcessSpec object>
_spec_class
classmethod define(spec: aiida.engine.processes.process_spec.CalcJobProcessSpec)None[source]

Define the process specification, including its inputs, outputs and known exit codes.

Ports are added to the metadata input namespace (inherited from the base Process), and a code input Port, a remote_folder output Port and retrieved folder output Port are added.

Parameters

spec – the calculation job process spec to define.

classmethod get_state_classes() → Dict[Hashable, Type[plumpy.process_states.State]][source]

A mapping of the State constants to the corresponding state class.

Overrides the waiting state with the Calcjob specific version.

on_terminated()None[source]

Cleanup the node by deleting the calulation job state.

Note

This has to be done before calling the super because that will seal the node after we cannot change it

property options

Return the options of the metadata that were specified when this process instance was launched.

Returns

options dictionary

parse(retrieved_temporary_folder: Optional[str] = None)aiida.engine.processes.exit_code.ExitCode[source]

Parse a retrieved job calculation.

This is called once it’s finished waiting for the calculation to be finished and the data has been retrieved.

Parameters

retrieved_temporary_folder – The path to the temporary folder

parse_retrieved_output(retrieved_temporary_folder: Optional[str] = None) → Optional[aiida.engine.processes.exit_code.ExitCode][source]

Parse the retrieved data by calling the parser plugin if it was defined in the inputs.

parse_scheduler_output(retrieved: aiida.orm.nodes.node.Node) → Optional[aiida.engine.processes.exit_code.ExitCode][source]

Parse the output of the scheduler if that functionality has been implemented for the plugin.

prepare_for_submission(folder: aiida.common.folders.Folder)aiida.common.datastructures.CalcInfo[source]

Prepare the calculation for submission.

Convert the input nodes into the corresponding input files in the format that the code will expect. In addition, define and return a CalcInfo instance, which is a simple data structure that contains information for the engine, for example, on what files to copy to the remote machine, what files to retrieve once it has completed, specific scheduler settings and more.

Parameters

folder – a temporary folder on the local file system.

Returns

the CalcInfo instance

presubmit(folder: aiida.common.folders.Folder)aiida.common.datastructures.CalcInfo[source]

Prepares the calculation folder with all inputs, ready to be copied to the cluster.

Parameters

folder – a SandboxFolder that can be used to write calculation input files and the scheduling script.

Return calcinfo

the CalcInfo object containing the information needed by the daemon to handle operations.

run() → Union[plumpy.process_states.Stop, int, plumpy.process_states.Wait][source]

Run the calculation job.

This means invoking the presubmit and storing the temporary folder in the node’s repository. Then we move the process in the Wait state, waiting for the UPLOAD transport task to be started.

Returns

the Stop command if a dry run, int if the process has an exit status, Wait command if the calcjob is to be uploaded

spec_options = <aiida.engine.processes.ports.PortNamespace object>

## Submodules¶

Implementation of the CalcJob process.

class aiida.engine.processes.calcjobs.calcjob.CalcJob(*args: Any, **kwargs: Any)[source]

Implementation of the CalcJob process.

__abstractmethods__ = frozenset({})
__annotations__ = {'link_label_retrieved': <class 'str'>}
__init__(*args, **kwargs)None[source]

Construct a CalcJob instance.

Construct the instance only if it is a sub class of CalcJob, otherwise raise InvalidOperation.

See documentation of aiida.engine.Process.

__module__ = 'aiida.engine.processes.calcjobs.calcjob'
_abc_impl = <_abc_data object>
_node_class
_spec = <aiida.engine.processes.process_spec.CalcJobProcessSpec object>
_spec_class
classmethod define(spec: aiida.engine.processes.process_spec.CalcJobProcessSpec)None[source]

Define the process specification, including its inputs, outputs and known exit codes.

Ports are added to the metadata input namespace (inherited from the base Process), and a code input Port, a remote_folder output Port and retrieved folder output Port are added.

Parameters

spec – the calculation job process spec to define.

classmethod get_state_classes() → Dict[Hashable, Type[plumpy.process_states.State]][source]

A mapping of the State constants to the corresponding state class.

Overrides the waiting state with the Calcjob specific version.

on_terminated()None[source]

Cleanup the node by deleting the calulation job state.

Note

This has to be done before calling the super because that will seal the node after we cannot change it

property options

Return the options of the metadata that were specified when this process instance was launched.

Returns

options dictionary

parse(retrieved_temporary_folder: Optional[str] = None)aiida.engine.processes.exit_code.ExitCode[source]

Parse a retrieved job calculation.

This is called once it’s finished waiting for the calculation to be finished and the data has been retrieved.

Parameters

retrieved_temporary_folder – The path to the temporary folder

parse_retrieved_output(retrieved_temporary_folder: Optional[str] = None) → Optional[aiida.engine.processes.exit_code.ExitCode][source]

Parse the retrieved data by calling the parser plugin if it was defined in the inputs.

parse_scheduler_output(retrieved: aiida.orm.nodes.node.Node) → Optional[aiida.engine.processes.exit_code.ExitCode][source]

Parse the output of the scheduler if that functionality has been implemented for the plugin.

prepare_for_submission(folder: aiida.common.folders.Folder)aiida.common.datastructures.CalcInfo[source]

Prepare the calculation for submission.

Convert the input nodes into the corresponding input files in the format that the code will expect. In addition, define and return a CalcInfo instance, which is a simple data structure that contains information for the engine, for example, on what files to copy to the remote machine, what files to retrieve once it has completed, specific scheduler settings and more.

Parameters

folder – a temporary folder on the local file system.

Returns

the CalcInfo instance

presubmit(folder: aiida.common.folders.Folder)aiida.common.datastructures.CalcInfo[source]

Prepares the calculation folder with all inputs, ready to be copied to the cluster.

Parameters

folder – a SandboxFolder that can be used to write calculation input files and the scheduling script.

Return calcinfo

the CalcInfo object containing the information needed by the daemon to handle operations.

run() → Union[plumpy.process_states.Stop, int, plumpy.process_states.Wait][source]

Run the calculation job.

This means invoking the presubmit and storing the temporary folder in the node’s repository. Then we move the process in the Wait state, waiting for the UPLOAD transport task to be started.

Returns

the Stop command if a dry run, int if the process has an exit status, Wait command if the calcjob is to be uploaded

spec_options = <aiida.engine.processes.ports.PortNamespace object>

Module containing utilities and classes relating to job calculations running on systems that require transport.

class aiida.engine.processes.calcjobs.manager.JobManager(transport_queue: TransportQueue)[source]

Bases: object

A manager for CalcJob submitted to Computer instances.

When a calculation job is submitted to a Computer, it actually uses a specific AuthInfo, which is a computer configured for a User. The JobManager maintains a mapping of JobsList instances for each authinfo that has active calculation jobs. These jobslist instances are then responsible for bundling scheduler updates for all the jobs they maintain (i.e. that all share the same authinfo) and update their status.

As long as a Runner will create a single JobManager instance and use that for its lifetime, the guarantees made by the JobsList about respecting the minimum polling interval of the scheduler will be maintained. Note, however, that since each Runner will create its own job manager, these guarantees only hold per runner.

__dict__ = mappingproxy({'__module__': 'aiida.engine.processes.calcjobs.manager', '__doc__': 'A manager for :py:class:~aiida.engine.processes.calcjobs.calcjob.CalcJob submitted to Computer instances.\n\n When a calculation job is submitted to a :py:class:~aiida.orm.computers.Computer, it actually uses a specific\n :py:class:~aiida.orm.authinfos.AuthInfo, which is a computer configured for a :py:class:~aiida.orm.users.User.\n The JobManager maintains a mapping of :py:class:~aiida.engine.processes.calcjobs.manager.JobsList instances\n for each authinfo that has active calculation jobs. These jobslist instances are then responsible for bundling\n scheduler updates for all the jobs they maintain (i.e. that all share the same authinfo) and update their status.\n\n As long as a :py:class:~aiida.engine.runners.Runner will create a single JobManager instance and use that for\n its lifetime, the guarantees made by the JobsList about respecting the minimum polling interval of the scheduler\n will be maintained. Note, however, that since each Runner will create its own job manager, these guarantees\n only hold per runner.\n ', '__init__': <function JobManager.__init__>, 'get_jobs_list': <function JobManager.get_jobs_list>, 'request_job_info_update': <function JobManager.request_job_info_update>, '__dict__': <attribute '__dict__' of 'JobManager' objects>, '__weakref__': <attribute '__weakref__' of 'JobManager' objects>})
__init__(transport_queue: TransportQueue)None[source]

Initialize self. See help(type(self)) for accurate signature.

__module__ = 'aiida.engine.processes.calcjobs.manager'
__weakref__

list of weak references to the object (if defined)

get_jobs_list(authinfo: aiida.orm.authinfos.AuthInfo)aiida.engine.processes.calcjobs.manager.JobsList[source]

Get or create a new JobLists instance for the given authinfo.

Parameters

authinfo – the AuthInfo

Returns

a JobsList instance

request_job_info_update(authinfo: aiida.orm.authinfos.AuthInfo, job_id: Hashable) → Iterator[asyncio.Future[JobInfo]][source]

Get a future that will resolve to information about a given job.

This is a context manager so that if the user leaves the context the request is automatically cancelled.

class aiida.engine.processes.calcjobs.manager.JobsList(authinfo: aiida.orm.authinfos.AuthInfo, transport_queue: TransportQueue, last_updated: Optional[float] = None)[source]

Bases: object

Manager of calculation jobs submitted with a specific AuthInfo, i.e. computer configured for a specific user.

This container of active calculation jobs is used to update their status periodically in batches, ensuring that even when a lot of jobs are running, the scheduler update command is not triggered for each job individually.

In addition, the Computer for which the AuthInfo is configured, can define a minimum polling interval. This class will guarantee that the time between update calls to the scheduler is larger or equal to that minimum interval.

Note that since each instance operates on a specific authinfo, the guarantees of batching scheduler update calls and the limiting of number of calls per unit time, through the minimum polling interval, is only applicable for jobs launched with that particular authinfo. If multiple authinfo instances with the same computer, have active jobs these limitations are not respected between them, since there is no communication between JobsList instances. See the JobManager for example usage.

__dict__ = mappingproxy({'__module__': 'aiida.engine.processes.calcjobs.manager', '__doc__': 'Manager of calculation jobs submitted with a specific AuthInfo, i.e. computer configured for a specific user.\n\n This container of active calculation jobs is used to update their status periodically in batches, ensuring that\n even when a lot of jobs are running, the scheduler update command is not triggered for each job individually.\n\n In addition, the :py:class:~aiida.orm.computers.Computer for which the :py:class:~aiida.orm.authinfos.AuthInfo\n is configured, can define a minimum polling interval. This class will guarantee that the time between update calls\n to the scheduler is larger or equal to that minimum interval.\n\n Note that since each instance operates on a specific authinfo, the guarantees of batching scheduler update calls\n and the limiting of number of calls per unit time, through the minimum polling interval, is only applicable for jobs\n launched with that particular authinfo. If multiple authinfo instances with the same computer, have active jobs\n these limitations are not respected between them, since there is no communication between JobsList instances.\n See the :py:class:~aiida.engine.processes.calcjobs.manager.JobManager for example usage.\n ', '__init__': <function JobsList.__init__>, 'logger': <property object>, 'get_minimum_update_interval': <function JobsList.get_minimum_update_interval>, 'last_updated': <property object>, '_get_jobs_from_scheduler': <function JobsList._get_jobs_from_scheduler>, '_update_job_info': <function JobsList._update_job_info>, 'request_job_info_update': <function JobsList.request_job_info_update>, '_ensure_updating': <function JobsList._ensure_updating>, '_has_job_state_changed': <staticmethod object>, '_get_next_update_delay': <function JobsList._get_next_update_delay>, '_update_requests_outstanding': <function JobsList._update_requests_outstanding>, '_get_jobs_with_scheduler': <function JobsList._get_jobs_with_scheduler>, '__dict__': <attribute '__dict__' of 'JobsList' objects>, '__weakref__': <attribute '__weakref__' of 'JobsList' objects>})
__init__(authinfo: aiida.orm.authinfos.AuthInfo, transport_queue: TransportQueue, last_updated: Optional[float] = None)[source]

Construct an instance for the given authinfo and transport queue.

Parameters
• authinfo – The authinfo used to check the jobs list

• transport_queue – A transport queue

• last_updated – initialize the last updated timestamp

__module__ = 'aiida.engine.processes.calcjobs.manager'
__weakref__

list of weak references to the object (if defined)

_ensure_updating()None[source]

Ensure that we are updating the job list from the remote resource.

This will automatically stop if there are no outstanding requests.

async _get_jobs_from_scheduler() → Dict[Hashable, JobInfo][source]

Get the current jobs list from the scheduler.

Returns

a mapping of job ids to JobInfo instances

_get_jobs_with_scheduler() → List[str][source]

Get all the jobs that are currently with scheduler.

Returns

the list of jobs with the scheduler

Return type

list

_get_next_update_delay()float[source]

Calculate when we are next allowed to poll the scheduler.

This delay is calculated as the minimum polling interval defined by the authentication info for this instance, minus time elapsed since the last update.

Returns

delay (in seconds) after which the scheduler may be polled again

static _has_job_state_changed(old: Optional[JobInfo], new: Optional[JobInfo])bool[source]

Return whether the states old and new are different.

async _update_job_info()None[source]

Update all of the job information objects.

This will set the futures for all pending update requests where the corresponding job has a new status compared to the last update.

_update_requests_outstanding()bool[source]
get_minimum_update_interval()float[source]

Get the minimum interval that should be respected between updates of the list.

Returns

the minimum interval

property last_updated

Get the timestamp of when the list was last updated as produced by time.time()

Returns

The last update point

property logger

Return the logger configured for this instance.

Returns

the logger

request_job_info_update(job_id: Hashable) → Iterator[asyncio.Future[JobInfo]][source]

Request job info about a job when the job next changes state.

If the job is not found in the jobs list at the update, the future will resolve to None.

Parameters

job_id – job identifier

Returns

future that will resolve to a JobInfo object when the job changes state

Transport tasks for calculation jobs.

exception aiida.engine.processes.calcjobs.tasks.PreSubmitException[source]

Bases: Exception

Raise in the do_upload coroutine when an exception is raised in CalcJob.presubmit.

__module__ = 'aiida.engine.processes.calcjobs.tasks'
__weakref__

list of weak references to the object (if defined)

class aiida.engine.processes.calcjobs.tasks.Waiting(process: CalcJob, done_callback: Optional[Callable[[], Any]], msg: Optional[str] = None, data: Optional[Any] = None)[source]

The waiting state for the CalcJob process.

__init__(process: CalcJob, done_callback: Optional[Callable[[], Any]], msg: Optional[str] = None, data: Optional[Any] = None)[source]
Parameters

process – The process this state belongs to

__module__ = 'aiida.engine.processes.calcjobs.tasks'
async _launch_task(coro, *args, **kwargs)[source]

Launch a coroutine as a task, making sure to make it interruptable.

async execute() → plumpy.process_states.State[source]

Override the execute coroutine of the base Waiting state.

interrupt(reason: Any) → Optional[_asyncio.Future][source]

Interrupt the Waiting state by calling interrupt on the transport task InterruptableFuture.

load_instance_state(saved_state, load_context)[source]
parse(retrieved_temporary_folder: str)plumpy.process_states.Running[source]

Return the Running state that will parse the CalcJob.

Parameters

retrieved_temporary_folder – temporary folder used in retrieving that can be used during parsing.

property process
Returns

The process

retrieve()aiida.engine.processes.calcjobs.tasks.Waiting[source]

Return the Waiting state that will retrieve the CalcJob.

stash()[source]

Return the Waiting state that will stash the CalcJob.

submit()aiida.engine.processes.calcjobs.tasks.Waiting[source]

Return the Waiting state that will submit the CalcJob.

update()aiida.engine.processes.calcjobs.tasks.Waiting[source]

Return the Waiting state that will update the CalcJob.

upload()aiida.engine.processes.calcjobs.tasks.Waiting[source]

Return the Waiting state that will upload the CalcJob.

async aiida.engine.processes.calcjobs.tasks.task_kill_job(node: aiida.orm.nodes.process.calculation.calcjob.CalcJobNode, transport_queue: aiida.engine.transports.TransportQueue, cancellable: aiida.engine.utils.InterruptableFuture)[source]

Transport task that will attempt to kill a job calculation.

The task will first request a transport from the queue. Once the transport is yielded, the relevant execmanager function is called, wrapped in the exponential_backoff_retry coroutine, which, in case of a caught exception, will retry after an interval that increases exponentially with the number of retries, for a maximum number of retries. If all retries fail, the task will raise a TransportTaskException

Parameters
• node – the node that represents the job calculation

• transport_queue – the TransportQueue from which to request a Transport

• cancellable – the cancelled flag that will be queried to determine whether the task was cancelled

Raises

TransportTaskException if after the maximum number of retries the transport task still excepted

async aiida.engine.processes.calcjobs.tasks.task_retrieve_job(node: aiida.orm.nodes.process.calculation.calcjob.CalcJobNode, transport_queue: aiida.engine.transports.TransportQueue, retrieved_temporary_folder: str, cancellable: aiida.engine.utils.InterruptableFuture)[source]

Transport task that will attempt to retrieve all files of a completed job calculation.

The task will first request a transport from the queue. Once the transport is yielded, the relevant execmanager function is called, wrapped in the exponential_backoff_retry coroutine, which, in case of a caught exception, will retry after an interval that increases exponentially with the number of retries, for a maximum number of retries. If all retries fail, the task will raise a TransportTaskException

Parameters
• node – the node that represents the job calculation

• transport_queue – the TransportQueue from which to request a Transport

• retrieved_temporary_folder – the absolute path to a directory to store files

• cancellable – the cancelled flag that will be queried to determine whether the task was cancelled

Raises

TransportTaskException if after the maximum number of retries the transport task still excepted

async aiida.engine.processes.calcjobs.tasks.task_stash_job(node: aiida.orm.nodes.process.calculation.calcjob.CalcJobNode, transport_queue: aiida.engine.transports.TransportQueue, cancellable: aiida.engine.utils.InterruptableFuture)[source]

Transport task that will optionally stash files of a completed job calculation on the remote.

The task will first request a transport from the queue. Once the transport is yielded, the relevant execmanager function is called, wrapped in the exponential_backoff_retry coroutine, which, in case of a caught exception, will retry after an interval that increases exponentially with the number of retries, for a maximum number of retries. If all retries fail, the task will raise a TransportTaskException

Parameters
• node – the node that represents the job calculation

• transport_queue – the TransportQueue from which to request a Transport

• cancellable (aiida.engine.utils.InterruptableFuture) – the cancelled flag that will be queried to determine whether the task was cancelled

Raises

Return if the tasks was successfully completed

Raises

TransportTaskException if after the maximum number of retries the transport task still excepted

async aiida.engine.processes.calcjobs.tasks.task_submit_job(node: aiida.orm.nodes.process.calculation.calcjob.CalcJobNode, transport_queue: aiida.engine.transports.TransportQueue, cancellable: aiida.engine.utils.InterruptableFuture)[source]

Transport task that will attempt to submit a job calculation.

The task will first request a transport from the queue. Once the transport is yielded, the relevant execmanager function is called, wrapped in the exponential_backoff_retry coroutine, which, in case of a caught exception, will retry after an interval that increases exponentially with the number of retries, for a maximum number of retries. If all retries fail, the task will raise a TransportTaskException

Parameters
• node – the node that represents the job calculation

• transport_queue – the TransportQueue from which to request a Transport

• cancellable – the cancelled flag that will be queried to determine whether the task was cancelled

Raises

TransportTaskException if after the maximum number of retries the transport task still excepted

async aiida.engine.processes.calcjobs.tasks.task_update_job(node: aiida.orm.nodes.process.calculation.calcjob.CalcJobNode, job_manager, cancellable: aiida.engine.utils.InterruptableFuture)[source]

Transport task that will attempt to update the scheduler status of the job calculation.

The task will first request a transport from the queue. Once the transport is yielded, the relevant execmanager function is called, wrapped in the exponential_backoff_retry coroutine, which, in case of a caught exception, will retry after an interval that increases exponentially with the number of retries, for a maximum number of retries. If all retries fail, the task will raise a TransportTaskException

Parameters
Returns

True if the tasks was successfully completed, False otherwise

async aiida.engine.processes.calcjobs.tasks.task_upload_job(process: CalcJob, transport_queue: aiida.engine.transports.TransportQueue, cancellable: aiida.engine.utils.InterruptableFuture)[source]

Transport task that will attempt to upload the files of a job calculation to the remote.

The task will first request a transport from the queue. Once the transport is yielded, the relevant execmanager function is called, wrapped in the exponential_backoff_retry coroutine, which, in case of a caught exception, will retry after an interval that increases exponentially with the number of retries, for a maximum number of retries. If all retries fail, the task will raise a TransportTaskException

Parameters
• process – the job calculation

• transport_queue – the TransportQueue from which to request a Transport

• cancellable – the cancelled flag that will be queried to determine whether the task was cancelled

Raises

TransportTaskException if after the maximum number of retries the transport task still excepted