multiprocess.multiproc_distributor
==================================

.. py:module:: multiprocess.multiproc_distributor

.. autoapi-nested-parse::

   Multiprocess executes tasks in parallel if multiple threads are available.



Attributes
----------

.. autoapisummary::

   multiprocess.multiproc_distributor.logger


Classes
-------

.. autoapisummary::

   multiprocess.multiproc_distributor.PickableTask
   multiprocess.multiproc_distributor.Consumer
   multiprocess.multiproc_distributor.MultiprocessDistributor
   multiprocess.multiproc_distributor.MultiprocessWorkerNode


Module Contents
---------------

.. py:data:: logger

.. py:class:: PickableTask(distributor_id: str, command: str, task_type: str = 'array')

   Object passed between processes.

   Initialization of PickableTask.

   array_step_id: is only meaningful and has int value when for array


   .. py:attribute:: distributor_id


   .. py:attribute:: command


   .. py:attribute:: task_type
      :value: 'array'



.. py:class:: Consumer(task_queue: multiprocessing.JoinableQueue, response_queue: multiprocessing.Queue)

   Bases: :py:obj:`multiprocessing.Process`


   Consumes the tasks to be run.

   Consume work sent from LocalExecutor through multiprocessing queue.

   this class is structured based on
   https://pymotw.com/2/multiprocessing/communication.html

       task_queue:
           a (multiprocessing.JoinableQueue[Optional[PickableTask]]) object
           created by LocalExecutor used to retrieve work from the
           distributor.
       response_queue:
           A (Queue[Tuple[int, Optional[int], Optional[int]]]) object,
           that will hold information with Queue:
           Tuple[distributor_id, array_step_id if applicable, pid]


   .. py:attribute:: task_queue
      :type:  multiprocessing.JoinableQueue[Optional[PickableTask]]


   .. py:attribute:: response_queue
      :type:  multiprocessing.Queue[Tuple[str, Optional[int]]]


   .. py:method:: run() -> None

      Wait for work, the execute it.



.. py:class:: MultiprocessDistributor(cluster_name: str, parallelism: int = 3, *args: tuple, **kwargs: dict)

   Bases: :py:obj:`jobmon.core.cluster_protocol.ClusterDistributor`


   Executes tasks locally in parallel.

   It uses the multiprocessing Python library and queues to parallelize the execution of
   tasks. The subprocessing pattern looks like this::

       LocalExec
       --> consumer1
       ----> subconsumer1
       --> consumer2
       ----> subconsumer2
       ...
       --> consumerN
       ----> subconsumerN

   Initialization of the multiprocess distributor.

   :param cluster_name: the name of the cluster.
   :param parallelism: how many parallel jobs to distribute at a
                       time
   :type parallelism: int, optional


   .. py:attribute:: temp_dir
      :type:  Optional[str]
      :value: None



   .. py:attribute:: started
      :value: False



   .. py:attribute:: task_queue
      :type:  multiprocessing.JoinableQueue[Optional[PickableTask]]


   .. py:attribute:: response_queue
      :type:  multiprocessing.Queue[Tuple[str, Optional[int]]]


   .. py:attribute:: consumers
      :type:  List[Consumer]
      :value: []



   .. py:property:: worker_node_entry_point
      :type: str


      Path to jobmon worker_node_entry_point.


   .. py:property:: cluster_name
      :type: str


      Return the name of the cluster type.


   .. py:method:: start() -> None

      Fire up N task consuming processes using Multiprocessing.

      Number of consumers is controlled by parallelism.



   .. py:method:: stop() -> None

      Terminate consumers and call sync 1 final time.



   .. py:method:: terminate_task_instances(distributor_ids: List[str]) -> None

      Terminate task instances.

      Only terminate the task instances that are running, not going to kill the jobs that
      are actually still in a waiting or a transitioning state.

      :param distributor_ids: A list of distributor IDs.



   .. py:method:: get_submitted_or_running(distributor_ids: Optional[List[str]] = None) -> Set[str]

      Get tasks that are active.



   .. py:method:: submit_to_batch_distributor(command: str, name: str, requested_resources: Dict[str, Any]) -> str

      Submit the command on the cluster technology and return a distributor_id.

      The distributor_id can be used to identify the associated TaskInstance, terminate
      it, monitor for missingness, or collect usage statistics. If an exception is raised by
      this method the task instance will move to "W" state and the exception will be logged
      in the database under the task_instance_error_log table.

      :param command: command to be run
      :param name: name of task
      :param requested_resources: resource requests sent to distributor API

      :returns: A tuple indicating the distributor id, the full output file location,
                and full error location.



   .. py:method:: submit_array_to_batch_distributor(command: str, name: str, requested_resources: Dict[str, Any], array_length: int) -> Dict[int, str]

      Submit an array task to the multiprocess cluster.

      Return: a mapping of array_step_id to distributor_id, output path, and error path



   .. py:method:: get_queueing_errors(distributor_ids: List[str]) -> Dict[str, str]

      Get the task instances that have errored out.



   .. py:method:: get_remote_exit_info(distributor_id: str) -> Tuple[str, str]

      Get the exit info about the task instance once it is done running.



.. py:class:: MultiprocessWorkerNode

   Bases: :py:obj:`jobmon.core.cluster_protocol.ClusterWorkerNode`


   Task instance info for an instance run with the Multiprocessing distributor.

   Initialization of the multiprocess distributor worker node.


   .. py:property:: distributor_id
      :type: Optional[str]


      The id from the distributor.


   .. py:method:: get_exit_info(exit_code: int, error_msg: str) -> Tuple[str, str]

      Exit code and message.



   .. py:method:: get_usage_stats() -> Dict

      Usage information specific to the distributor.



   .. py:method:: initialize_logfile(log_type: str, log_dir: str, name: str) -> str

      Error and exit code info from the executor.



   .. py:property:: array_step_id
      :type: Optional[int]


      Return array_step_id .