hypershell.submit


Directly embed submit capabilities either as a dedicated SubmitThread, LiveSubmitThread, or using one of the provided high-level functions, submit_from(), submit_file().

All of the parameters used below are largely the same in each instance. Configuration files have no impact when using the library; i.e., default values are not overridden by configuration.

Functions


submit_from(source: Iterable[str], queue_config: QueueConfig = None, bundlesize: int = 1, bundlewait: int = 5, cores: int = None, memory: int = None, timeout: int = None, group: int = 0, template: str = '{}', tags: Dict[str, None | bool | int | float | str | Dict[str, None | bool | int | float | str | Dict[str, JSONData] | List[JSONData]] | List[None | bool | int | float | str | Dict[str, JSONData] | List[JSONData]]] = None) → int[source]

Submit all task arguments from source, return count of submitted tasks.

If queue_config is provided, the LiveSubmitThread is used to submit task bundles directly to the shared queue hosted by the server. Otherwise, the SubmitThread submits tasks to the database.

Parameters:

  • source (Iterable[str]) – Any iterable of command-line tasks.

  • queue_config (QueueConfig) – QueueConfig instance with host, port, and auth.

  • bundlesize (int, optional) – Size of task bundles. See DEFAULT_BUNDLESIZE.

  • bundlewait (int, optional) – Waiting period before forcing task bundle push. See DEFAULT_BUNDLEWAIT.

  • template (str, optional) – Task command-line template pattern. See DEFAULT_TEMPLATE.

  • cores (int, optional) – Default number of cores to use for each task (default: none). May be overridden by inline-comment.

  • memory (int, optional) – Default memory in bytes to use for each task (default: none). May be overridden by inline-comment.

  • timeout (int, optional) – Task-level walltime limit in seconds (default: none). May be overridden by inline-comment.

  • group (int, optional) – Task group for dependency management (default: 0).

  • tags (Dict[str, JSONData], optional) – Tag dictionary for all submitted tasks.

Example

>>> from hypershell.submit import submit_from
>>> submit_from(['AAA', 'BBB', 'CCC'], template='my-script {}',
...             tags={'site': 'zzz', 'group': 37})
3

See also

Returns:

Count of submitted tasks.

Return type:

task_count (int)


submit_file(path: str, queue_config: QueueConfig = None, bundlesize: int = 1, bundlewait: int = 5, template: str = '{}', cores: int = None, memory: int = None, timeout: int = None, group: int = 0, tags: Dict[str, None | bool | int | float | str | Dict[str, None | bool | int | float | str | Dict[str, JSONData] | List[JSONData]] | List[None | bool | int | float | str | Dict[str, JSONData] | List[JSONData]]] = None, **file_options) → int[source]

Submit all task arguments by reading them from file path.

Arguments are forwarded to submit_from() with the opened file stream from path as the task source.

Parameters:

  • path (str) – Path to file containing command-line tasks.

  • queue_config (QueueConfig) – QueueConfig instance with host, port, and auth.

  • bundlesize (int, optional) – Size of task bundles. See DEFAULT_BUNDLESIZE.

  • bundlewait (int, optional) – Waiting period before forcing task bundle push. See DEFAULT_BUNDLEWAIT.

  • template (str, optional) – Task command-line template pattern. See DEFAULT_TEMPLATE.

  • cores (int, optional) – Default number of cores to use for each task (default: none). May be overridden by inline-comment.

  • memory (int, optional) – Default memory in bytes to use for each task (default: none). May be overridden by inline-comment.

  • timeout (int, optional) – Task-level walltime limit in seconds (default: none). May be overridden by inline-comment.

  • group (int, optional) – Task group for dependency management (default: 0).

  • tags (Dict[str, JSONData], optional) – Tag dictionary for all submitted tasks.

Example

>>> from hypershell.submit import submit_from
>>> from hypershell.core.queue import QueueConfig
>>> queue_config = QueueConfig(host='my.server.univ.edu', port=54321, auth='my-secret-key')
>>> submit_file('/tmp/tasks.in', queue_config=queue_config, template='my-script {}',
...             tags={'site': 'zzz', 'group': 37})
3

See also

Returns:

Count of submitted tasks.

Return type:

task_count (int)

Classes


class SubmitThread(source: Iterable[str], cores: int = None, memory: int = None, timeout: int = None, bundlesize: int = 1, bundlewait: int = 5, template: str = '{}', group: int = 0, tags: Dict[str, None | bool | int | float | str | Dict[str, None | bool | int | float | str | Dict[str, JSONData] | List[JSONData]] | List[None | bool | int | float | str | Dict[str, JSONData] | List[JSONData]]] = None)[source]

Bases: Thread

Submit tasks to database within dedicated thread.

Parameters:

  • source (Iterable[str]) – Any iterable of command-line tasks.

  • bundlesize (int, optional) – Size of task bundles. See DEFAULT_BUNDLESIZE.

  • bundlewait (int, optional) – Waiting period before forcing task bundle push. See DEFAULT_BUNDLEWAIT.

  • template (str, optional) – Task command-line template pattern. See DEFAULT_TEMPLATE.

  • cores (int, optional) – Default number of cores to use for each task (default: none). May be overridden by inline-comment.

  • memory (int, optional) – Default memory in bytes to use for each task (default: none). May be overridden by inline-comment.

  • timeout (int, optional) – Task-level walltime limit in seconds (default: none).

  • group (int, optional) – Task group for dependency management (default: 0).

  • tags (Dict[str, JSONData], optional) – Tag dictionary for all submitted tasks.

Example

>>> from hypershell.submit import SubmitThread
>>> submitter = SubmitThread.new(['AAA', 'BBB', 'CCC'],
...                              template='my-script {}'
...                              tags={'site': 'zzz', 'group': 37})
>>> submitter.join()

See also

classmethod new(*args, **kwargs) → Thread

Initialize and start the thread.

start()

Start the thread’s activity.

It must be called at most once per thread object. It arranges for the object’s run() method to be invoked in a separate thread of control.

This method will raise a RuntimeError if called more than once on the same thread object.

join(timeout: float | None = None) → None

Calls Thread.join but re-raises exceptions.

stop(wait: bool = False, timeout: int = None) → None[source]

Stop child threads before main thread.


class LiveSubmitThread(source: Iterable[str], queue_config: QueueConfig, template: str = '{}', cores: int = None, memory: int = None, timeout: int = None, group: int = 0, bundlesize: int = 1, bundlewait: int = 5, tags: Dict[str, str] = None)[source]

Bases: Thread

Submit tasks directly to queue within dedicated thread.

Parameters:

  • source (Iterable[str]) – Any iterable of command-line tasks.

  • queue_config (QueueConfig) – QueueConfig instance with host, port, and auth.

  • bundlesize (int, optional) – Size of task bundles. See DEFAULT_BUNDLESIZE.

  • bundlewait (int, optional) – Waiting period before forcing task bundle push. See DEFAULT_BUNDLEWAIT.

  • template (str, optional) – Task command-line template pattern. See DEFAULT_TEMPLATE.

  • cores (int, optional) – Default number of cores to use for each task (default: none). May be overridden by inline-comment.

  • memory (int, optional) – Default memory in bytes to use for each task (default: none). May be overridden by inline-comment.

  • timeout (int, optional) – Task-level walltime limit in seconds (default: none). May be overridden by inline-comment.

  • group (int, optional) – Task group for dependency management (default: 0).

  • tags (Dict[str, JSONData], optional) – Tag dictionary for all submitted tasks.

Example

>>> from hypershell.submit import LiveSubmitThread
>>> from hypershell.core.queue import QueueConfig
>>> queue_config = QueueConfig(host='localhost', port=54321, auth='my-secret-key')
>>> submitter = LiveSubmitThread.new(['AAA', 'BBB', 'CCC'],
...                                  queue_config=queue_config,
...                                  template='my-script {}'
...                                  tags={'site': 'zzz', 'group': 37})
>>> submitter.join()

See also

classmethod new(*args, **kwargs) → Thread

Initialize and start the thread.

start()

Start the thread’s activity.

It must be called at most once per thread object. It arranges for the object’s run() method to be invoked in a separate thread of control.

This method will raise a RuntimeError if called more than once on the same thread object.

join(timeout: float | None = None) → None

Calls Thread.join but re-raises exceptions.

stop(wait: bool = False, timeout: int = None) → None[source]

Stop child threads before main thread.

Constants


DEFAULT_BUNDLESIZE: Final[int] = 1

Default size of task bundles.

DEFAULT_BUNDLEWAIT: Final[int] = 5

Default waiting period before forcing task bundle push.

DEFAULT_TASK_GROUP: Final[int] = 0

Default task group for backwards compatibility.