# pathos module documentation

## abstract_launcher module

This module contains the base classes for pathos pool and pipe objects, and describes the map and pipe interfaces. A pipe is defined as a connection between two ‘nodes’, where a node is something that does work. A pipe may be a one-way or two-way connection. A map is defined as a one-to-many connection between nodes. In both map and pipe connections, results from the connected nodes can be returned to the calling node. There are several variants of pipe and map, such as whether the connection is blocking, or ordered, or asynchronous. For pipes, derived methods must overwrite the ‘pipe’ method, while maps must overwrite the ‘map’ method. Pipes and maps are available from worker pool objects, where the work is done by any of the workers in the pool. For more specific point-to-point connections (such as a pipe between two specific compute nodes), use the pipe object directly.

### Usage

A typical call to a pathos map will roughly follow this example:

    >>> # instantiate and configure the worker pool
    >>> from pathos.pools import ProcessPool
    >>> pool = ProcessPool(nodes=4)
    >>>
    >>> # do a blocking map on the chosen function
    >>> results = pool.map(pow, [1,2,3,4], [5,6,7,8])
    >>>
    >>> # do a non-blocking map, then extract the results from the iterator
    >>> results = pool.imap(pow, [1,2,3,4], [5,6,7,8])
    >>> print("...")
    >>> results = list(results)
    >>>
    >>> # do an asynchronous map, then get the results
    >>> results = pool.amap(pow, [1,2,3,4], [5,6,7,8])
    >>> while not results.ready():
    ...     time.sleep(5); print(".", end=' ')
    ...
    >>> results = results.get()

Notes

Each of the pathos worker pools rely on a different transport protocol (e.g. threads, multiprocessing, etc), where the use of each pool comes with a few caveats. See the usage documentation and examples for each worker pool for more information.

class AbstractPipeConnection(*args, **kwds)

  Bases: `object`

  AbstractPipeConnection base class for pathos pipes.

  Required input:

    ???

  Additional inputs:

    ???

  Important class members:

    ???

  Other class members:

    ???

  __annotate_func__ = None
  __annotations_cache__ = {}
  __firstlineno__ = 61
  __repr__()

    Return repr(self).

  __static_attributes__ = ()

class AbstractWorkerPool(*args, **kwds)

  Bases: `object`

  AbstractWorkerPool base class for pathos pools.

  Important class members:

    nodes - number (and potentially description) of workers ncpus - number of worker processors servers - list of worker servers scheduler - the associated scheduler workdir - associated $WORKDIR for scratch calculations/files

  Other class members:

    scatter - True, if uses ‘scatter-gather’ (instead of ‘worker-pool’) source - False, if minimal use of TemporaryFiles is desired timeout - number of seconds to wait for return value from scheduler

  __annotations_cache__ = {}
  __enter__()
  __exit__(*args)
  __firstlineno__ = 87
  __get_nodes()

    get the number of nodes in the pool

  __imap(f, *args, **kwds)

    default filter for imap inputs

  __init(*args, **kwds)

    default filter for __init__ inputs

  __map(f, *args, **kwds)

    default filter for map inputs

  __nodes = 1
  __pipe(f, *args, **kwds)

    default filter for pipe inputs

  __repr__()

    Return repr(self).

  __set_nodes(nodes)

    set the number of nodes in the pool

  __static_attributes__ = ('_id', 'nodes')
  _serve(*args, **kwds)

    Create a new server if one isn’t already initialized

  amap(f, *args, **kwds)

    run a batch of jobs with an asynchronous map

    Returns a results object which containts the results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until all results are retrieved. Use the ready() method on the result object to check if all results are ready. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  apipe(f, *args, **kwds)

    submit a job asynchronously to a queue

    Returns a results object which containts the result of calling the function f on a selected worker. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until the result is available. Use the ready() method on the results object to check if the result is ready.

  clear()

    Remove server with matching state

  imap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and ordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  map(f, *args, **kwds)

    run a batch of jobs with a blocking and ordered map

    Returns a list of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  pipe(f, *args, **kwds)

    submit a job and block until results are available

    Returns result of calling the function f on a selected worker. This function will block until results are available.

  uimap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and unordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. The order of the resulting sequence is not guaranteed. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

## connection module

This module contains the base class for popen pipes, and describes the popen pipe interface. The ‘config’ method can be overwritten for pipe customization. The pipe’s ‘launch’ method can be overwritten with a derived pipe’s new execution algorithm. See the following for an example of standard use.

### Usage

A typical call to a popen ‘pipe’ will roughly follow this example:

    >>> # instantiate the pipe
    >>> pipe = Pipe()
    >>>
    >>> # configure the pipe to stage the command
    >>> pipe(command='hostname')
    >>>
    >>> # execute the launch and retrieve the response
    >>> pipe.launch()
    >>> print(pipe.response())

class Pipe(name=None, **kwds)

  Bases: `object`

  a popen-based pipe for parallel and distributed computing

  create a popen-pipe

  Inputs:

    name: a unique identifier (string) for the pipe command: a command to send [default = ‘echo <name>’] background: run in background [default = False] decode: ensure response is ‘ascii’ [default = True] stdin: file-like object to serve as standard input for the remote process

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(**kwds)

    configure the pipe using given keywords

    (Re)configure the pipe for the following inputs:

      command: a command to send [default = ‘echo <name>’] background: run in background [default = False] decode: ensure response is ‘ascii’ [default = True] stdin: file-like object to serve as standard input for the remote process

  __firstlineno__ = 49
  __repr__()

    Return repr(self).

  __static_attributes__ = ('_pid', '_response', '_stdout', 'background', 'codec', 'message', 'name', 'stdin')
  _debug = <Logger pathos (WARNING)>
  _execute()
  config(**kwds)

    configure the pipe using given keywords

    (Re)configure the pipe for the following inputs:

      command: a command to send [default = ‘echo <name>’] background: run in background [default = False] decode: ensure response is ‘ascii’ [default = True] stdin: file-like object to serve as standard input for the remote process

  kill()

    terminate the pipe

  launch()

    launch a configured command

  pid()

    get pipe pid

  response()

    Return the response from the launched process. Return None if no response was received yet from a background process.

  verbose = True

exception PipeException

  Bases: `Exception`

  Exception for failure to launch a command

  __firstlineno__ = 44
  __static_attributes__ = ()

## core module

high-level programming interface to core pathos utilities

connect(host, port=None, through=None)

  establish a secure tunnel connection to a remote host at the given port

  Parameters:

    - established (host -- hostname to which a tunnel should be)

    - number (port -- port)

    - None] (through -- 'tunnel-through' hostname [default =)

copy(source, destination=None, **kwds)

  copy source to (possibly) remote destination

  Execute a copy, and return the copier. Use ‘kill’ to kill the copier, and ‘pid’ to get the process id for the copier.

  Parameters:

    - 'file' (source -- path string of source)

    - target (destination -- path string for destination)

execute(command, host=None, bg=True, **kwds)

  execute a command (possibly) on a remote host

  Execute a process, and return the launcher. Use ‘response’ to retrieve the response from the executed command. Use ‘kill’ to kill the launcher, and ‘pid’ to get the process id for the launcher.

  Parameters:

    - executed (command -- command string to be)

    - None (host -- hostname of execution target [default =)

    - True] (bg -- run as background process? [default =)

getchild(pid=None, host=None, group=False)

  get all child process ids for the given parent process id (ppid)

  If pid is None, the pid of the __main__ python instance will be used.

  Parameters:

    - id (group -- get process ids for the parent group)

    - running (host -- hostname where process is)

    - id

getpid(target=None, host=None, all=False, **kwds)

  get the process id for a target process (possibly) running on remote host

  This method should only be used as a last-ditch effort to find a process id. This method __may__ work when a child has been spawned and the pid was not registered… but there’s no guarantee.

  If target is None, then get the process id of the __main__ python instance.

  Parameters:

    - process (target -- string name of target)

    - running (host -- hostname where process is)

    - False] (all -- get all resulting lines from query? [default =)

getppid(pid=None, host=None, group=False)

  get parent process id (ppid) for the given process

  If pid is None, the pid of the __main__ python instance will be used.

  Parameters:

    - id (group -- get parent group)

    - running (host -- hostname where process is)

    - id

kill(pid, host=None, **kwds)

  kill a process (possibly) on a remote host

  Parameters:

    - id (pid -- process)

    - None (host -- hostname where process is running [default =)

randomport(host=None)

  select a open port on a (possibly) remote host

  Parameters:

    port (host -- hostname on which to select a open)

serve(server, host=None, port=None, profile='.bash_profile')

  begin serving RPC requests

  Parameters:

    - server – name of RPC server (i.e. ‘ppserver’)

    - host – hostname on which a server should be launched

    - port – port number (on host) that server will accept request at

    - profile – file to configure the user’s environment [default=’.bash_profile’]

## helpers module

- pathos.helpers module documentation

ProcessPool

  alias of `Pool`

ThreadPool(processes=None, initializer=None, initargs=())

cpu_count()

  Returns the number of CPUs in the system

freeze_support()

  Check whether this is a fake forked process in a frozen executable. If so then run code specified by commandline and exit.

shutdown(type=None)

  destroy all cached pools (of the given type)

## hosts module

high-level programming interface to pathos host registry

_profiles = {}

  For example, to register two ‘known’ host profiles:

  _profiles = { ‘foobar.danse.us’:’.profile’, ‘computer.cacr.caltech.edu’:’.cshrc’, }

get_profile(rhost, assume=True)

  get the default $PROFILE for a remote host

get_profiles()

  get $PROFILE for each registered host

register(rhost, profile=None)

  register a host and $PROFILE

register_profiles(profiles)

  add dict of {‘host’:$PROFILE} to registered host profiles

## maps module

maps: stand-alone map-like objects using lazy pool instantiation

class Amap(pool=None, *args, **kwds)

  Bases: `Map`

  async map instance with internal lazy pool instantiation

  Parameters:

    - pool – pool object (i.e. pathos.pools.ProcessPool)

    - *args – positional arguments for pool initialization

    - **kwds – keyword arguments for pool initialization

    - close – if True, close the pool to any new jobs [Default: False]

    - join – if True, reclaim the pool’s closed workers [Default: False]

    - clear – if True, delete the pool singleton [Default: False]

  NOTE: if a pool object is not provided, NotImplemented is returned

    upon use.

  NOTE: pools from both multiprocess and pathos.pools can be used,

    however the behavior is slightly different. Pools from both pathos and multiprocess have close and join methods, to close the pool to new jobs, and to shut down the pool’s workers. Pools from pathos, however, are launched as singletons, so they also include a clear method that deletes the singleton. In either case, a pool that has been “closed” will throw a ValueError if map is then called, and similarly, a ValueError will be thrown if join is called before a pool is “closed”. The major difference is that if a pathos.pool is closed, the map instance cannot run new jobs until “clear” is called, while a new multiprocess pool will be created each time the map is executed. This leads to pathos.pools generally being called with either `clear=True` or `clear=False`, and pools from multprocess either using `close=True` or `join=True` or both. Some hierarchical parallel workflows are not allowed, and will result in an error being thrown; however, changing close, join, or clear can often remove the error.

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(func, *args, **kwds)

    instantiate a pool and execute the pool’s async map

    Parameters:

      - func – function object to map

      - *args – positional arguments for async map

      - **kwds – keyword arguments for async map

    Returns:

      results from execution of async `map(func, *args, **kwds)`

    NOTE: initializes a new worker pool with each call

  __firstlineno__ = 245
  __static_attributes__ = ('_pool',)

class Asmap(pool=None, *args, **kwds)

  Bases: `Map`

  async starmap instance with internal lazy pool instantiation

  Parameters:

    - pool – pool object (i.e. pathos.pools.ProcessPool)

    - *args – positional arguments for pool initialization

    - **kwds – keyword arguments for pool initialization

    - close – if True, close the pool to any new jobs [Default: False]

    - join – if True, reclaim the pool’s closed workers [Default: False]

    - clear – if True, delete the pool singleton [Default: False]

  NOTE: if a pool object is not provided, NotImplemented is returned

    upon use.

  NOTE: pools from both multiprocess and pathos.pools can be used,

    however the behavior is slightly different. Pools from both pathos and multiprocess have close and join methods, to close the pool to new jobs, and to shut down the pool’s workers. Pools from pathos, however, are launched as singletons, so they also include a clear method that deletes the singleton. In either case, a pool that has been “closed” will throw a ValueError if map is then called, and similarly, a ValueError will be thrown if join is called before a pool is “closed”. The major difference is that if a pathos.pool is closed, the map instance cannot run new jobs until “clear” is called, while a new multiprocess pool will be created each time the map is executed. This leads to pathos.pools generally being called with either `clear=True` or `clear=False`, and pools from multprocess either using `close=True` or `join=True` or both. Some hierarchical parallel workflows are not allowed, and will result in an error being thrown; however, changing close, join, or clear can often remove the error.

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(func, *args, **kwds)

    instantiate a pool and execute the pool’s async starmap

    Parameters:

      - func – function object to map

      - *args – positional arguments for async starmap

      - **kwds – keyword arguments for async starmap

    Returns:

      results from execution of async `starmap(func, *args, **kwds)`

    NOTE: initializes a new worker pool with each call

  __firstlineno__ = 307
  __static_attributes__ = ('_pool',)

class Imap(pool=None, *args, **kwds)

  Bases: `Map`

  map iterator with internal lazy pool instantiation

  Parameters:

    - pool – pool object (i.e. pathos.pools.ProcessPool)

    - *args – positional arguments for pool initialization

    - **kwds – keyword arguments for pool initialization

    - close – if True, close the pool to any new jobs [Default: False]

    - join – if True, reclaim the pool’s closed workers [Default: False]

    - clear – if True, delete the pool singleton [Default: False]

  NOTE: if a pool object is not provided, a builtins.map will be

    used.

  NOTE: pools from both multiprocess and pathos.pools can be used,

    however the behavior is slightly different. Pools from both pathos and multiprocess have close and join methods, to close the pool to new jobs, and to shut down the pool’s workers. Pools from pathos, however, are launched as singletons, so they also include a clear method that deletes the singleton. In either case, a pool that has been “closed” will throw a ValueError if map is then called, and similarly, a ValueError will be thrown if join is called before a pool is “closed”. The major difference is that if a pathos.pool is closed, the map instance cannot run new jobs until “clear” is called, while a new multiprocess pool will be created each time the map is executed. This leads to pathos.pools generally being called with either `clear=True` or `clear=False`, and pools from multprocess either using `close=True` or `join=True` or both. Some hierarchical parallel workflows are not allowed, and will result in an error being thrown; however, changing close, join, or clear can often remove the error.

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(func, *args, **kwds)

    instantiate a pool and execute the pool’s map iterator

    Parameters:

      - func – function object to map

      - *args – positional arguments for map iterator

      - **kwds – keyword arguments for map iterator

    Returns:

      results from execution of `map(func, *args, **kwds)` iterator

    NOTE: initializes a new worker pool with each call

  __firstlineno__ = 183
  __static_attributes__ = ('_pool',)

class Ismap(pool=None, *args, **kwds)

  Bases: `Map`

  starmap iterator with internal lazy pool instantiation

  Parameters:

    - pool – pool object (i.e. pathos.pools.ProcessPool)

    - *args – positional arguments for pool initialization

    - **kwds – keyword arguments for pool initialization

    - close – if True, close the pool to any new jobs [Default: False]

    - join – if True, reclaim the pool’s closed workers [Default: False]

    - clear – if True, delete the pool singleton [Default: False]

  NOTE: if a pool object is not provided, an itertools.starmap will

    be used.

  NOTE: pools from both multiprocess and pathos.pools can be used,

    however the behavior is slightly different. Pools from both pathos and multiprocess have close and join methods, to close the pool to new jobs, and to shut down the pool’s workers. Pools from pathos, however, are launched as singletons, so they also include a clear method that deletes the singleton. In either case, a pool that has been “closed” will throw a ValueError if map is then called, and similarly, a ValueError will be thrown if join is called before a pool is “closed”. The major difference is that if a pathos.pool is closed, the map instance cannot run new jobs until “clear” is called, while a new multiprocess pool will be created each time the map is executed. This leads to pathos.pools generally being called with either `clear=True` or `clear=False`, and pools from multprocess either using `close=True` or `join=True` or both. Some hierarchical parallel workflows are not allowed, and will result in an error being thrown; however, changing close, join, or clear can often remove the error.

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(func, *args, **kwds)

    instantiate a pool and execute the pool’s starmap iterator

    Parameters:

      - func – function object to map

      - *args – positional arguments for starmap iterator

      - **kwds – keyword arguments for starmap iterator

    Returns:

      results from execution of `starmap(func, *args, **kwds)` iterator

    NOTE: initializes a new worker pool with each call

  __firstlineno__ = 432
  __static_attributes__ = ('_pool',)

class Map(pool=None, *args, **kwds)

  Bases: `object`

  map instance with internal lazy pool instantiation

  Parameters:

    - pool – pool object (i.e. pathos.pools.ProcessPool)

    - *args – positional arguments for pool initialization

    - **kwds – keyword arguments for pool initialization

    - close – if True, close the pool to any new jobs [Default: False]

    - join – if True, reclaim the pool’s closed workers [Default: False]

    - clear – if True, delete the pool singleton [Default: False]

  NOTE: if a pool object is not provided, a builtins.map will be

    used with the returned iterator cast to a list.

  NOTE: pools from both multiprocess and pathos.pools can be used,

    however the behavior is slightly different. Pools from both pathos and multiprocess have close and join methods, to close the pool to new jobs, and to shut down the pool’s workers. Pools from pathos, however, are launched as singletons, so they also include a clear method that deletes the singleton. In either case, a pool that has been “closed” will throw a ValueError if map is then called, and similarly, a ValueError will be thrown if join is called before a pool is “closed”. The major difference is that if a pathos.pool is closed, the map instance cannot run new jobs until “clear” is called, while a new multiprocess pool will be created each time the map is executed. This leads to pathos.pools generally being called with either `clear=True` or `clear=False`, and pools from multprocess either using `close=True` or `join=True` or both. Some hierarchical parallel workflows are not allowed, and will result in an error being thrown; however, changing close, join, or clear can often remove the error.

  __annotate_func__ = None
  __annotations_cache__ = {}
  __attr__()
  __call__(func, *args, **kwds)

    instantiate a pool and execute the pool’s map

    Parameters:

      - func – function object to map

      - *args – positional arguments for map

      - **kwds – keyword arguments for map

    Returns:

      results from execution of `map(func, *args, **kwds)`

    NOTE: initializes a new worker pool with each call

  __cls__()
  __del__()

    shutdown the worker pool and tidy up

  __firstlineno__ = 11
  property __func__
  property __get__
  __meth__()
  property __self__
  __static_attributes__ = ('_clear', '_close', '_join', '_pool', 'args', 'kwds', 'pool')
  clear()

    remove pool singleton, if exists

  close()

    close the map to any new jobs

  join()

    reclaim the closed workers

class Smap(pool=None, *args, **kwds)

  Bases: `Map`

  starmap instance with internal lazy pool instantiation

  Parameters:

    - pool – pool object (i.e. pathos.pools.ProcessPool)

    - *args – positional arguments for pool initialization

    - **kwds – keyword arguments for pool initialization

    - close – if True, close the pool to any new jobs [Default: False]

    - join – if True, reclaim the pool’s closed workers [Default: False]

    - clear – if True, delete the pool singleton [Default: False]

  NOTE: if a pool object is not provided, an itertools.starmap will

    be used with the returned iterator cast to a list.

  NOTE: pools from both multiprocess and pathos.pools can be used,

    however the behavior is slightly different. Pools from both pathos and multiprocess have close and join methods, to close the pool to new jobs, and to shut down the pool’s workers. Pools from pathos, however, are launched as singletons, so they also include a clear method that deletes the singleton. In either case, a pool that has been “closed” will throw a ValueError if map is then called, and similarly, a ValueError will be thrown if join is called before a pool is “closed”. The major difference is that if a pathos.pool is closed, the map instance cannot run new jobs until “clear” is called, while a new multiprocess pool will be created each time the map is executed. This leads to pathos.pools generally being called with either `clear=True` or `clear=False`, and pools from multprocess either using `close=True` or `join=True` or both. Some hierarchical parallel workflows are not allowed, and will result in an error being thrown; however, changing close, join, or clear can often remove the error.

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(func, *args, **kwds)

    instantiate a pool and execute the pool’s starmap

    Parameters:

      - func – function object to map

      - *args – positional arguments for starmap

      - **kwds – keyword arguments for starmap

    Returns:

      results from execution of `starmap(func, *args, **kwds)`

    NOTE: initializes a new worker pool with each call

  __firstlineno__ = 119
  __static_attributes__ = ('_pool',)

class Uimap(pool=None, *args, **kwds)

  Bases: `Map`

  unordered map iterator with internal lazy pool instantiation

  Parameters:

    - pool – pool object (i.e. pathos.pools.ProcessPool)

    - *args – positional arguments for pool initialization

    - **kwds – keyword arguments for pool initialization

    - close – if True, close the pool to any new jobs [Default: False]

    - join – if True, reclaim the pool’s closed workers [Default: False]

    - clear – if True, delete the pool singleton [Default: False]

  NOTE: if a pool object is not provided, NotImplemented is returned

    upon use.

  NOTE: pools from both multiprocess and pathos.pools can be used,

    however the behavior is slightly different. Pools from both pathos and multiprocess have close and join methods, to close the pool to new jobs, and to shut down the pool’s workers. Pools from pathos, however, are launched as singletons, so they also include a clear method that deletes the singleton. In either case, a pool that has been “closed” will throw a ValueError if map is then called, and similarly, a ValueError will be thrown if join is called before a pool is “closed”. The major difference is that if a pathos.pool is closed, the map instance cannot run new jobs until “clear” is called, while a new multiprocess pool will be created each time the map is executed. This leads to pathos.pools generally being called with either `clear=True` or `clear=False`, and pools from multprocess either using `close=True` or `join=True` or both. Some hierarchical parallel workflows are not allowed, and will result in an error being thrown; however, changing close, join, or clear can often remove the error.

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(func, *args, **kwds)

    instantiate a pool and execute the pool’s unordered map iterator

    Parameters:

      - func – function object to map

      - *args – positional arguments for unordered map iterator

      - **kwds – keyword arguments for unordered map iterator

    Returns:

      results from execution of unordered `map(func, *args, **kwds)` iterator

    NOTE: initializes a new worker pool with each call

  __firstlineno__ = 370
  __static_attributes__ = ('_pool',)

## mp_map module

minimal interface to python’s multiprocessing module

Notes

This module has been deprecated in favor of `pathos.pools`.

mp_map(function, sequence, *args, **kwds)

  extend python’s parallel map function to multiprocessing

  Parameters:

    - function (function - target)

    - parallel (sequence - sequence to process in)

    - 'autodetect'] (nproc - number of 'local' cpus to use [defaut =)

    - ['blocking' (type - processing type)

    - 'non-blocking'

    - 'unordered']

    - True (threads - if)

    - multiprocessing (use threading instead of)

## multiprocessing module

This module contains map and pipe interfaces to python’s multiprocessing module.

Pipe methods provided:

  pipe - blocking communication pipe [returns: value] apipe - asynchronous communication pipe [returns: object]

Map methods provided:

  map - blocking and ordered worker pool [returns: list] imap - non-blocking and ordered worker pool [returns: iterator] uimap - non-blocking and unordered worker pool [returns: iterator] amap - asynchronous worker pool [returns: object]

### Usage

A typical call to a pathos multiprocessing map will roughly follow this example:

    >>> # instantiate and configure the worker pool
    >>> from pathos.multiprocessing import ProcessPool
    >>> pool = ProcessPool(nodes=4)
    >>>
    >>> # do a blocking map on the chosen function
    >>> print(pool.map(pow, [1,2,3,4], [5,6,7,8]))
    >>>
    >>> # do a non-blocking map, then extract the results from the iterator
    >>> results = pool.imap(pow, [1,2,3,4], [5,6,7,8])
    >>> print("...")
    >>> print(list(results))
    >>>
    >>> # do an asynchronous map, then get the results
    >>> results = pool.amap(pow, [1,2,3,4], [5,6,7,8])
    >>> while not results.ready():
    ...     time.sleep(5); print(".", end=' ')
    ...
    >>> print(results.get())
    >>>
    >>> # do one item at a time, using a pipe
    >>> print(pool.pipe(pow, 1, 5))
    >>> print(pool.pipe(pow, 2, 6))
    >>>
    >>> # do one item at a time, using an asynchronous pipe
    >>> result1 = pool.apipe(pow, 1, 5)
    >>> result2 = pool.apipe(pow, 2, 6)
    >>> print(result1.get())
    >>> print(result2.get())

Notes

This worker pool leverages the python’s multiprocessing module, and thus has many of the limitations associated with that module. The function f and the sequences in args must be serializable. The maps in this worker pool have full functionality whether run from a script or in the python interpreter, and work reliably for both imported and interactively-defined functions. Unlike python’s multiprocessing module, pathos.multiprocessing maps can directly utilize functions that require multiple arguments.

class ProcessPool(*args, **kwds)

  Bases: `AbstractWorkerPool`

  Mapper that leverages python’s multiprocessing.

  Important class members:

    nodes - number (and potentially description) of workers ncpus - number of worker processors servers - list of worker servers scheduler - the associated scheduler workdir - associated $WORKDIR for scratch calculations/files

  Other class members:

    scatter - True, if uses ‘scatter-gather’ (instead of ‘worker-pool’) source - False, if minimal use of TemporaryFiles is desired timeout - number of seconds to wait for return value from scheduler

  NOTE: if number of nodes is not given, will autodetect processors.

  NOTE: additional keyword input is optional, with:

    id - identifier for the pool initializer - function that takes no input, called when node is spawned initargs - tuple of args for initializers that have args maxtasksperchild - int that limits the max number of tasks per node

  __annotate_func__ = None
  __annotations_cache__ = {}
  __firstlineno__ = 84
  __get_nodes()

    get the number of nodes used in the map

  __repr__()

    Return repr(self).

  __set_nodes(nodes)

    set the number of nodes used in the map

  __state__ = {4: <multiprocess.pool.Pool state=RUN pool_size=4>}
  __static_attributes__ = ('__nodes', '_id', '_kwds')
  _clear()

    Remove server with matching state

  _serve(nodes=None)

    Create a new server if one isn’t already initialized

  amap(f, *args, **kwds)

    run a batch of jobs with an asynchronous map

    Returns a results object which containts the results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until all results are retrieved. Use the ready() method on the result object to check if all results are ready. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  apipe(f, *args, **kwds)

    submit a job asynchronously to a queue

    Returns a results object which containts the result of calling the function f on a selected worker. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until the result is available. Use the ready() method on the results object to check if the result is ready.

  clear()

    Remove server with matching state

  close()

    close the pool to any new jobs

  imap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and ordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  join()

    cleanup the closed worker processes

  map(f, *args, **kwds)

    run a batch of jobs with a blocking and ordered map

    Returns a list of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  property ncpus

    get the number of nodes used in the map

  property nodes

    get the number of nodes used in the map

  pipe(f, *args, **kwds)

    submit a job and block until results are available

    Returns result of calling the function f on a selected worker. This function will block until results are available.

  restart(force=False)

    restart a closed pool

  terminate()

    a more abrupt close

  uimap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and unordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. The order of the resulting sequence is not guaranteed. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

_ProcessPool

  alias of `Pool`

## parallel module

This module contains map and pipe interfaces to the parallelpython (pp) module.

Pipe methods provided:

  pipe - blocking communication pipe [returns: value] apipe - asynchronous communication pipe [returns: object]

Map methods provided:

  map - blocking and ordered worker pool [returns: list] imap - non-blocking and ordered worker pool [returns: iterator] uimap - non-blocking and unordered worker pool [returns: iterator] amap - asynchronous worker pool [returns: object]

### Usage

A typical call to a pathos pp map will roughly follow this example:

    >>> # instantiate and configure the worker pool
    >>> from pathos.pp import ParallelPool
    >>> pool = ParallelPool(nodes=4)
    >>>
    >>> # do a blocking map on the chosen function
    >>> print(pool.map(pow, [1,2,3,4], [5,6,7,8]))
    >>>
    >>> # do a non-blocking map, then extract the results from the iterator
    >>> results = pool.imap(pow, [1,2,3,4], [5,6,7,8])
    >>> print("...")
    >>> print(list(results))
    >>>
    >>> # do an asynchronous map, then get the results
    >>> results = pool.amap(pow, [1,2,3,4], [5,6,7,8])
    >>> while not results.ready():
    ...     time.sleep(5); print(".", end=' ')
    ...
    >>> print(results.get())
    >>>
    >>> # do one item at a time, using a pipe
    >>> print(pool.pipe(pow, 1, 5))
    >>> print(pool.pipe(pow, 2, 6))
    >>>
    >>> # do one item at a time, using an asynchronous pipe
    >>> result1 = pool.apipe(pow, 1, 5)
    >>> result2 = pool.apipe(pow, 2, 6)
    >>> print(result1.get())
    >>> print(result2.get())

Notes

This worker pool leverages the parallelpython (pp) module, and thus has many of the limitations associated with that module. The function f and the sequences in args must be serializable. The maps in this worker pool have full functionality when run from a script, but may be somewhat limited when used in the python interpreter. Both imported and interactively-defined functions in the interpreter session may fail due to the pool failing to find the source code for the target function. For a work-around, try:

    >>> # instantiate and configure the worker pool
    >>> from pathos.pp import ParallelPool
    >>> pool = ParallelPool(nodes=4)
    >>>
    >>> # wrap the function, so it can be used interactively by the pool
    >>> def wrapsin(*args, **kwds):
    >>>      from math import sin
    >>>      return sin(*args, **kwds)
    >>>
    >>> # do a blocking map using the wrapped function
    >>> results = pool.map(wrapsin, [1,2,3,4,5])

class ParallelPool(*args, **kwds)

  Bases: `AbstractWorkerPool`

  Mapper that leverages parallelpython (i.e. pp) maps.

  Important class members:

    nodes - number (and potentially description) of workers ncpus - number of worker processors servers - list of worker servers scheduler - the associated scheduler workdir - associated $WORKDIR for scratch calculations/files

  Other class members:

    scatter - True, if uses ‘scatter-gather’ (instead of ‘worker-pool’) source - False, if minimal use of TemporaryFiles is desired timeout - number of seconds to wait for return value from scheduler

  NOTE: if number of nodes is not given, will autodetect processors.

  NOTE: if a tuple of servers is not provided, defaults to localhost only.

  NOTE: additional keyword input is optional, with:

    id - identifier for the pool servers - tuple of pp.Servers

  __annotate_func__ = None
  __annotations_cache__ = {}
  __firstlineno__ = 138
  __get_nodes()

    get the number of nodes used in the map

  __get_servers()

    get the servers used in the map

  __repr__()

    Return repr(self).

  __set_nodes(nodes)

    set the number of nodes used in the map

  __set_servers(servers)

    set the servers used in the map

  __state__ = {}
  __static_attributes__ = ('__nodes', '__servers', '_exiting', '_id')
  _clear()

    Remove server with matching state

  _equals(server)

    check if the server is compatible

  _is_alive(server=None, negate=False, run=True)
  _serve(nodes=None, servers=None)

    Create a new server if one isn’t already initialized

  amap(f, *args, **kwds)

    run a batch of jobs with an asynchronous map

    Returns a results object which containts the results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until all results are retrieved. Use the ready() method on the result object to check if all results are ready. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  apipe(f, *args, **kwds)

    submit a job asynchronously to a queue

    Returns a results object which containts the result of calling the function f on a selected worker. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until the result is available. Use the ready() method on the results object to check if the result is ready.

  clear()

    Remove server with matching state

  close()

    close the pool to any new jobs

  imap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and ordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  join()

    cleanup the closed worker processes

  map(f, *args, **kwds)

    run a batch of jobs with a blocking and ordered map

    Returns a list of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  property ncpus

    get the number of nodes used in the map

  property nodes

    get the number of nodes used in the map

  pipe(f, *args, **kwds)

    submit a job and block until results are available

    Returns result of calling the function f on a selected worker. This function will block until results are available.

  restart(force=False)

    restart a closed pool

  property servers

    get the servers used in the map

  terminate()

    a more abrupt close

  uimap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and unordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. The order of the resulting sequence is not guaranteed. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

stats(pool=None)

  return a string containing stats response from the pp.Server

## pools module

pools: pools of pathos workers, providing map and pipe constructs

class ParallelPool(*args, **kwds)

  Bases: `AbstractWorkerPool`

  Mapper that leverages parallelpython (i.e. pp) maps.

  Important class members:

    nodes - number (and potentially description) of workers ncpus - number of worker processors servers - list of worker servers scheduler - the associated scheduler workdir - associated $WORKDIR for scratch calculations/files

  Other class members:

    scatter - True, if uses ‘scatter-gather’ (instead of ‘worker-pool’) source - False, if minimal use of TemporaryFiles is desired timeout - number of seconds to wait for return value from scheduler

  NOTE: if number of nodes is not given, will autodetect processors.

  NOTE: if a tuple of servers is not provided, defaults to localhost only.

  NOTE: additional keyword input is optional, with:

    id - identifier for the pool servers - tuple of pp.Servers

  __annotations_cache__ = {}
  __firstlineno__ = 138
  __get_nodes()

    get the number of nodes used in the map

  __get_servers()

    get the servers used in the map

  __repr__()

    Return repr(self).

  __set_nodes(nodes)

    set the number of nodes used in the map

  __set_servers(servers)

    set the servers used in the map

  __state__ = {}
  __static_attributes__ = ('__nodes', '__servers', '_exiting', '_id')
  _clear()

    Remove server with matching state

  _equals(server)

    check if the server is compatible

  _is_alive(server=None, negate=False, run=True)
  _serve(nodes=None, servers=None)

    Create a new server if one isn’t already initialized

  amap(f, *args, **kwds)

    run a batch of jobs with an asynchronous map

    Returns a results object which containts the results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until all results are retrieved. Use the ready() method on the result object to check if all results are ready. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  apipe(f, *args, **kwds)

    submit a job asynchronously to a queue

    Returns a results object which containts the result of calling the function f on a selected worker. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until the result is available. Use the ready() method on the results object to check if the result is ready.

  clear()

    Remove server with matching state

  close()

    close the pool to any new jobs

  imap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and ordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  join()

    cleanup the closed worker processes

  map(f, *args, **kwds)

    run a batch of jobs with a blocking and ordered map

    Returns a list of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  property ncpus

    get the number of nodes used in the map

  property nodes

    get the number of nodes used in the map

  pipe(f, *args, **kwds)

    submit a job and block until results are available

    Returns result of calling the function f on a selected worker. This function will block until results are available.

  restart(force=False)

    restart a closed pool

  property servers

    get the servers used in the map

  terminate()

    a more abrupt close

  uimap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and unordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. The order of the resulting sequence is not guaranteed. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

class ProcessPool(*args, **kwds)

  Bases: `AbstractWorkerPool`

  Mapper that leverages python’s multiprocessing.

  Important class members:

    nodes - number (and potentially description) of workers ncpus - number of worker processors servers - list of worker servers scheduler - the associated scheduler workdir - associated $WORKDIR for scratch calculations/files

  Other class members:

    scatter - True, if uses ‘scatter-gather’ (instead of ‘worker-pool’) source - False, if minimal use of TemporaryFiles is desired timeout - number of seconds to wait for return value from scheduler

  NOTE: if number of nodes is not given, will autodetect processors.

  NOTE: additional keyword input is optional, with:

    id - identifier for the pool initializer - function that takes no input, called when node is spawned initargs - tuple of args for initializers that have args maxtasksperchild - int that limits the max number of tasks per node

  __annotations_cache__ = {}
  __firstlineno__ = 84
  __get_nodes()

    get the number of nodes used in the map

  __repr__()

    Return repr(self).

  __set_nodes(nodes)

    set the number of nodes used in the map

  __state__ = {4: <multiprocess.pool.Pool state=RUN pool_size=4>}
  __static_attributes__ = ('__nodes', '_id', '_kwds')
  _clear()

    Remove server with matching state

  _serve(nodes=None)

    Create a new server if one isn’t already initialized

  amap(f, *args, **kwds)

    run a batch of jobs with an asynchronous map

    Returns a results object which containts the results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until all results are retrieved. Use the ready() method on the result object to check if all results are ready. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  apipe(f, *args, **kwds)

    submit a job asynchronously to a queue

    Returns a results object which containts the result of calling the function f on a selected worker. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until the result is available. Use the ready() method on the results object to check if the result is ready.

  clear()

    Remove server with matching state

  close()

    close the pool to any new jobs

  imap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and ordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  join()

    cleanup the closed worker processes

  map(f, *args, **kwds)

    run a batch of jobs with a blocking and ordered map

    Returns a list of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  property ncpus

    get the number of nodes used in the map

  property nodes

    get the number of nodes used in the map

  pipe(f, *args, **kwds)

    submit a job and block until results are available

    Returns result of calling the function f on a selected worker. This function will block until results are available.

  restart(force=False)

    restart a closed pool

  terminate()

    a more abrupt close

  uimap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and unordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. The order of the resulting sequence is not guaranteed. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

class SerialPool(*args, **kwds)

  Bases: `AbstractWorkerPool`

  Mapper that leverages standard (i.e. serial) python maps.

  Important class members:

    nodes - number (and potentially description) of workers ncpus - number of worker processors servers - list of worker servers scheduler - the associated scheduler workdir - associated $WORKDIR for scratch calculations/files

  Other class members:

    scatter - True, if uses ‘scatter-gather’ (instead of ‘worker-pool’) source - False, if minimal use of TemporaryFiles is desired timeout - number of seconds to wait for return value from scheduler

  __annotate_func__ = None
  __annotations_cache__ = {}
  __firstlineno__ = 65
  __get_nodes()

    get the number of nodes in the pool

  __set_nodes(nodes)

    set the number of nodes in the pool

  __state__ = {}
  __static_attributes__ = ('_exiting',)
  _exiting = False
  _is_alive(negate=False, run=True)
  clear()

    hard restart

  close()

    close the pool to any new jobs

  imap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and ordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  join()

    cleanup the closed worker processes

  map(f, *args, **kwds)

    run a batch of jobs with a blocking and ordered map

    Returns a list of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  property nodes

    get the number of nodes in the pool

  pipe(f, *args, **kwds)

    submit a job and block until results are available

    Returns result of calling the function f on a selected worker. This function will block until results are available.

  restart(force=False)

    restart a closed pool

  terminate()

    a more abrupt close

class ThreadPool(*args, **kwds)

  Bases: `AbstractWorkerPool`

  Mapper that leverages python’s threading.

  Important class members:

    nodes - number (and potentially description) of workers ncpus - number of worker processors servers - list of worker servers scheduler - the associated scheduler workdir - associated $WORKDIR for scratch calculations/files

  Other class members:

    scatter - True, if uses ‘scatter-gather’ (instead of ‘worker-pool’) source - False, if minimal use of TemporaryFiles is desired timeout - number of seconds to wait for return value from scheduler

  NOTE: if number of nodes is not given, will autodetect processors.

  NOTE: additional keyword input is optional, with:

    id - identifier for the pool initializer - function that takes no input, called when node is spawned initargs - tuple of args for initializers that have args

  __annotate_func__ = None
  __annotations_cache__ = {}
  __firstlineno__ = 78
  __get_nodes()

    get the number of nodes used in the map

  __repr__()

    Return repr(self).

  __set_nodes(nodes)

    set the number of nodes used in the map

  __state__ = {4: <multiprocess.pool.ThreadPool state=RUN pool_size=4>}
  __static_attributes__ = ('__nodes', '_id', '_kwds')
  _clear()

    Remove server with matching state

  _serve(nodes=None)

    Create a new server if one isn’t already initialized

  amap(f, *args, **kwds)

    run a batch of jobs with an asynchronous map

    Returns a results object which containts the results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until all results are retrieved. Use the ready() method on the result object to check if all results are ready. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  apipe(f, *args, **kwds)

    submit a job asynchronously to a queue

    Returns a results object which containts the result of calling the function f on a selected worker. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until the result is available. Use the ready() method on the results object to check if the result is ready.

  clear()

    Remove server with matching state

  close()

    close the pool to any new jobs

  imap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and ordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  join()

    cleanup the closed worker processes

  map(f, *args, **kwds)

    run a batch of jobs with a blocking and ordered map

    Returns a list of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  property nodes

    get the number of nodes used in the map

  property nthreads

    get the number of nodes used in the map

  pipe(f, *args, **kwds)

    submit a job and block until results are available

    Returns result of calling the function f on a selected worker. This function will block until results are available.

  restart(force=False)

    restart a closed pool

  terminate()

    a more abrupt close

  uimap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and unordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. The order of the resulting sequence is not guaranteed. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

_ProcessPool

  alias of `Pool`

_ThreadPool(processes=None, initializer=None, initargs=())

_clear(type=None)

  destroy all cached pools (of the given type)

## portpicker module

This script prints out an available port number.

class portnumber(min=0, max=64 * 1024)

  Bases: `object`

  port selector

  Usage:

        >>> pick = portnumber(min=1024,max=65535)
        >>> print( pick() )

  select a port number from a given range.

  The first call will return a random number from the available range, and each subsequent call will return the next number in the range.

  Parameters:

    - 0] (min -- minimum port number [default =)

    - 65536] (max -- maximum port number [default =)

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__()

    Call self as a function.

  __firstlineno__ = 15
  __static_attributes__ = ('current', 'first', 'max', 'min')

randomport(min=1024, max=65536)

  select a random port number

  Parameters:

    - 1024] (min -- minimum port number [default =)

    - 65536] (max -- maximum port number [default =)

## pp module

This module contains map and pipe interfaces to the parallelpython (pp) module.

Pipe methods provided:

  pipe - blocking communication pipe [returns: value] apipe - asynchronous communication pipe [returns: object]

Map methods provided:

  map - blocking and ordered worker pool [returns: list] imap - non-blocking and ordered worker pool [returns: iterator] uimap - non-blocking and unordered worker pool [returns: iterator] amap - asynchronous worker pool [returns: object]

### Usage

A typical call to a pathos pp map will roughly follow this example:

    >>> # instantiate and configure the worker pool
    >>> from pathos.pp import ParallelPool
    >>> pool = ParallelPool(nodes=4)
    >>>
    >>> # do a blocking map on the chosen function
    >>> print(pool.map(pow, [1,2,3,4], [5,6,7,8]))
    >>>
    >>> # do a non-blocking map, then extract the results from the iterator
    >>> results = pool.imap(pow, [1,2,3,4], [5,6,7,8])
    >>> print("...")
    >>> print(list(results))
    >>>
    >>> # do an asynchronous map, then get the results
    >>> results = pool.amap(pow, [1,2,3,4], [5,6,7,8])
    >>> while not results.ready():
    ...     time.sleep(5); print(".", end=' ')
    ...
    >>> print(results.get())
    >>>
    >>> # do one item at a time, using a pipe
    >>> print(pool.pipe(pow, 1, 5))
    >>> print(pool.pipe(pow, 2, 6))
    >>>
    >>> # do one item at a time, using an asynchronous pipe
    >>> result1 = pool.apipe(pow, 1, 5)
    >>> result2 = pool.apipe(pow, 2, 6)
    >>> print(result1.get())
    >>> print(result2.get())

Notes

This worker pool leverages the parallelpython (pp) module, and thus has many of the limitations associated with that module. The function f and the sequences in args must be serializable. The maps in this worker pool have full functionality when run from a script, but may be somewhat limited when used in the python interpreter. Both imported and interactively-defined functions in the interpreter session may fail due to the pool failing to find the source code for the target function. For a work-around, try:

    >>> # instantiate and configure the worker pool
    >>> from pathos.pp import ParallelPool
    >>> pool = ParallelPool(nodes=4)
    >>>
    >>> # wrap the function, so it can be used interactively by the pool
    >>> def wrapsin(*args, **kwds):
    >>>      from math import sin
    >>>      return sin(*args, **kwds)
    >>>
    >>> # do a blocking map using the wrapped function
    >>> results = pool.map(wrapsin, [1,2,3,4,5])

ParallelPythonPool

  alias of `ParallelPool`

stats(pool=None)

  return a string containing stats response from the pp.Server

## pp_map module

minimal interface to python’s `pp` (parallel python) module

Implements a work-alike of the builtin `map` function that distributes work across many processes. As it uses `ppft` to do the actual parallel processing, code using this must conform to the usual `ppft` restrictions (arguments must be serializable, etc).

Notes

This module has been deprecated in favor of `pathos.pools`.

ppServer

  alias of `Server`

pp_map(function, sequence, *args, **kwds)

  extend python’s parallel map function to parallel python

  Parameters:

    - function (function - target)

    - parallel (sequence - sequence to process in)

    - 'autodetect'] (ncpus - number of 'local' processors to use [defaut =)

    - ()] (servers - available distributed parallel python servers [default =)

ppmap(processes, function, sequence, *sequences)

  Split the work of ‘function’ across the given number of processes. Set ‘processes’ to None to let Parallel Python autodetect the number of children to use.

  Although the calling semantics should be identical to __builtin__.map (even using __builtin__.map to process arguments), it differs in that it returns a generator instead of a list. This enables lazy evaluation of the results so that other work can be done while the subprocesses are still running.

      >>> def rangetotal(n): return n, sum(range(n))
      >>> list(map(rangetotal, range(1, 6)))
      [(1, 0), (2, 1), (3, 3), (4, 6), (5, 10)]
      >>> list(ppmap(1, rangetotal, range(1, 6)))
      [(1, 0), (2, 1), (3, 3), (4, 6), (5, 10)]

print_stats(servers=None)

  print stats from the pp.Server

stats(pool=None)

  return a string containing stats response from the pp.Server

## profile module

This module contains functions for profiling in other threads and processes.

Functions for identifying a thread/process:

  process_id - get the identifier (process id) for the current process thread_id - get the identifier for the current thread

Functions for controlling profiling:

  enable_profiling - initialize a profiler in the current thread/process start_profiling - begin profiling everything in the current thread/process stop_profiling - stop profiling everything in the current thread/process disable_profiling - remove the profiler from the current thread/process

Functions that control profile statstics (pstats) output

  clear_stats - clear stored pstats from the current thread/process get_stats - get stored pstats for the current thread/process print_stats - print stored pstats for the current thread/process dump_stats - dump stored pstats for the current thread/process

Functions that add/remove profiling:

  profiled - decorator to add profiling to a function not_profiled - decorator to remove profiling from a function profile - decorator for profiling a function (will enable_profiling)

### Usage

Typical calls to pathos profiling will roughly follow this example:

     >>> import time
     >>> import random
     >>> import pathos.profile as pr
     >>>
     >>> # build a worker function
     >>> def _work(i):
     ...     x = random.random()
     ...     time.sleep(x)
     ...     return (i,x)
     >>>
     >>> # generate a 'profiled' work function
     >>> config = dict(gen=pr.process_id)
     >>> work = pr.profiled(**config)(_work)
     >>>
     >>> # enable profiling
     >>> pr.enable_profiling()
     >>>
     >>> # profile the work (not the map internals) in the main process
     >>> for i in map(work, range(-10,0)):
     ...     print(i)
     ...
     >>> # profile the map in the main process, and work in the other process
     >>> from pathos.helpers import mp
     >>> pool = mp.Pool(10)
     >>> _uimap = pr.profiled(**config)(pool.imap_unordered)
     >>> for i in _uimap(work, range(-10,0)):
     ...     print(i)
     ...
     >>> # deactivate all profiling
     >>> pr.disable_profiling() # in the main process
     >>> tuple(_uimap(pr.disable_profiling, range(10))) # in the workers
     >>> for i in _uimap(work, range(-20,-10)):
     ...     print(i)
     ...
     >>> # re-activate profiling
     >>> pr.enable_profiling()
     >>>
     >>> # print stats for profile of 'import math' in another process
     >>> def test_import(module):
     ...    __import__(module)
     ...
     >>> import pathos.pools as pp
     >>> pool = pp.ProcessPool(1)
     >>> pr.profile('cumulative', pipe=pool.pipe)(test_import, 'pox')
          10 function calls in 0.003 seconds

    Ordered by: cumulative time

    ncalls  tottime  percall  cumtime  percall filename:lineno(function)
         1    0.000    0.000    0.003    0.003 <stdin>:1(test_import)
         1    0.002    0.002    0.003    0.003 {__import__}
         1    0.001    0.001    0.001    0.001 __init__.py:8(<module>)
         1    0.000    0.000    0.000    0.000 shutils.py:11(<module>)
         1    0.000    0.000    0.000    0.000 _disk.py:15(<module>)
         1    0.000    0.000    0.000    0.000 {eval}
         1    0.000    0.000    0.000    0.000 utils.py:11(<module>)
         1    0.000    0.000    0.000    0.000 <string>:1(<module>)
         1    0.000    0.000    0.000    0.000 info.py:2(<module>)
         1    0.000    0.000    0.000    0.000 {method 'disable' of '_lsprof.Profiler' objects}

     >>> pool.close()
     >>> pool.join()
     >>> pool.clear()

Notes

This module leverages the python’s cProfile module, and is primarily a high-level interface to that module that strives to make profiling in a different thread or process easier. The use of pathos.pools are suggested, however are not required (as seen in the example above).

In many cases, profiling in another thread is not necessary, and either of the following can be sufficient/better for timing and profiling:

    $ python -c "import time; s=time.time(); import pathos; print (time.time()-s)"
    $ python -c "import cProfile; p=cProfile.Profile(); p.enable(); import pathos; p.print_stats('cumulative')"

This module was inspired by: http://stackoverflow.com/a/32522579/4646678.

clear_stats(*args)

  clear all stored profiling results from the current thread/process

disable_profiling(*args)

  remove the profiler instance from the current thread/process

dump_stats(*args, **kwds)

  dump all stored profiling results for the current thread/process

  Notes

  see `pathos.profile.profiled` for settings for `*args` and `**kwds`

enable_profiling(*args)

  initialize a profiler instance in the current thread/process

get_stats(*args)

  get all stored profiling results for the current thread/process

not_profiled(f)

  decorator to remove profiling (due to ‘profiled’) from a function

print_stats(*args, **kwds)

  print all stored profiling results for the current thread/process

process_id()

  get the identifier (process id) for the current process

class profile(sort=None, **config)

  Bases: `object`

  decorator for profiling a function (will enable profiling)

  sort is integer index of column in pstats output for sorting

  Important class members:

    pipe - pipe instance in which profiling is active

  Example

      >>> import time
      >>> import random
      >>> import pathos.profile as pr
      >>>
      ... def work():
      ...     x = random.random()
      ...     time.sleep(x)
      ...     return x
      ...
      >>> # profile the work; print pstats info
      >>> pr.profile()(work)
               4 function calls in 0.136 seconds

  > Ordered by: standard name

  > ncalls tottime percall cumtime percall filename:lineno(function)

  >   1 0.000 0.000 0.136 0.136 <stdin>:1(work) 1 0.000 0.000 0.000 0.000 {method ‘disable’ of ‘_lsprof.Profiler’ objects} 1 0.000 0.000 0.000 0.000 {method ‘random’ of ‘_random.Random’ objects} 1 0.136 0.136 0.136 0.136 {time.sleep}

  0.1350568110491419 >>>

  NOTE: pipe provided should come from pool built with nodes=1. Other

    configuration keywords (config) are passed to ‘pr.profiled’. Output can be ordered by setting sort to one of the following: ‘calls’ - call count ‘cumulative’ - cumulative time ‘cumtime’ - cumulative time ‘file’ - file name ‘filename’ - file name ‘module’ - file name ‘ncalls’ - call count ‘pcalls’ - primitive call count ‘line’ - line number ‘name’ - function name ‘nfl’ - name/file/line ‘stdname’ - standard name ‘time’ - internal time ‘tottime’ - internal time

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(function, *args, **kwds)

    Call self as a function.

  __firstlineno__ = 284
  __static_attributes__ = ('_pool', 'config', 'pipe')

class profiled(gen=None, prefix='id-', suffix='.prof')

  Bases: `object`

  decorator for profiling a function (does not call enable profiling)

  y=gen(), with y an indentifier (e.g. current_process().pid)

  Important class members:

    prefix - string prefix for pstats filename [default: ‘id-‘] suffix - string suffix for pstats filename [default: ‘.prof’] pid - function for obtaining id of current process/thread sort - integer index of column in pstats output for sorting

  Example

      >>> import time
      >>> import random
      >>> import pathos.profile as pr
      >>>
      >>> config = dict(gen=pr.process_id)
      >>> @pr.profiled(**config)
      ... def work(i):
      ...     x = random.random()
      ...     time.sleep(x)
      ...     return (i,x)
      ...
      >>> pr.enable_profiling()
      >>> # profile the work (not the map internals); write to file for pstats
      >>> for i in map(work, range(-10,0)):
      ...     print(i)
      ...

  NOTE: If gen is a bool or string, then sort=gen and pid is not used.

    Otherwise, pid=gen and sort is not used. Output can be ordered by setting gen to one of the following: ‘calls’ - call count ‘cumulative’ - cumulative time ‘cumtime’ - cumulative time ‘file’ - file name ‘filename’ - file name ‘module’ - file name ‘ncalls’ - call count ‘pcalls’ - primitive call count ‘line’ - line number ‘name’ - function name ‘nfl’ - name/file/line ‘stdname’ - standard name ‘time’ - internal time ‘tottime’ - internal time

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(f)

    Call self as a function.

  __firstlineno__ = 132
  __static_attributes__ = ('pid', 'prefix', 'sort', 'suffix')

start_profiling(*args)

  begin profiling everything in the current thread/process

stop_profiling(*args)

  stop profiling everything in the current thread/process

thread_id()

  get the identifier for the current thread

## python module

This module contains map and pipe interfaces to standard (i.e. serial) python.

Pipe methods provided:

  pipe - blocking communication pipe [returns: value]

Map methods provided:

  map - blocking and ordered worker pool [returns: list] imap - non-blocking and ordered worker pool [returns: iterator]

### Usage

A typical call to a pathos python map will roughly follow this example:

    >>> # instantiate and configure the worker pool
    >>> from pathos.serial import SerialPool
    >>> pool = SerialPool()
    >>>
    >>> # do a blocking map on the chosen function
    >>> print(pool.map(pow, [1,2,3,4], [5,6,7,8]))
    >>>
    >>> # do a non-blocking map, then extract the results from the iterator
    >>> results = pool.imap(pow, [1,2,3,4], [5,6,7,8])
    >>> print("...")
    >>> print(list(results))
    >>>
    >>> # do one item at a time, using a pipe
    >>> print(pool.pipe(pow, 1, 5))
    >>> print(pool.pipe(pow, 2, 6))

Notes

This worker pool leverages the built-in python maps, and thus does not have limitations due to serialization of the function f or the sequences in args. The maps in this worker pool have full functionality whether run from a script or in the python interpreter, and work reliably for both imported and interactively-defined functions.

PythonSerial

  alias of `SerialPool`

## secure module

- pathos.secure module documentation

class Copier(name=None, **kwds)

  Bases: `Pipe`

  a popen-based copier for parallel and distributed computing.

  create a copier

  Inputs:

    name: a unique identifier (string) for the launcher source: hostname:path of original [user@host:path is also valid] destination: hostname:path for copy [user@host:path is also valid] launcher: remote service mechanism (i.e. scp, cp) [default = ‘scp’] options: remote service options (i.e. -v, -P) [default = ‘’] background: run in background [default = False] decode: ensure response is ‘ascii’ [default = True] stdin: file-like object to serve as standard input for the remote process

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(**kwds)

    configure the copier using given keywords:

    (Re)configure the copier for the following inputs:

      source: hostname:path of original [user@host:path is also valid] destination: hostname:path for copy [user@host:path is also valid] launcher: remote service mechanism (i.e. scp, cp) [default = ‘scp’] options: remote service options (i.e. -v, -P) [default = ‘’] background: run in background [default = False] decode: ensure response is ‘ascii’ [default = True] stdin: file-like object to serve as standard input for the remote process

  __firstlineno__ = 42
  __static_attributes__ = ('_stdout', 'background', 'codec', 'destination', 'launcher', 'message', 'options', 'source', 'stdin')
  config(**kwds)

    configure the copier using given keywords:

    (Re)configure the copier for the following inputs:

      source: hostname:path of original [user@host:path is also valid] destination: hostname:path for copy [user@host:path is also valid] launcher: remote service mechanism (i.e. scp, cp) [default = ‘scp’] options: remote service options (i.e. -v, -P) [default = ‘’] background: run in background [default = False] decode: ensure response is ‘ascii’ [default = True] stdin: file-like object to serve as standard input for the remote process

class Pipe(name=None, **kwds)

  Bases: `Pipe`

  a popen-based ssh-pipe for parallel and distributed computing.

  create a ssh pipe

  Inputs:

    name: a unique identifier (string) for the pipe host: hostname to recieve command [user@host is also valid] command: a command to send [default = ‘echo <name>’] launcher: remote service mechanism (i.e. ssh, rsh) [default = ‘ssh’] options: remote service options (i.e. -v, -N, -L) [default = ‘’] background: run in background [default = False] decode: ensure response is ‘ascii’ [default = True] stdin: file-like object to serve as standard input for the remote process

  __annotate_func__ = None
  __annotations_cache__ = {}
  __call__(**kwds)

    configure a remote command using given keywords:

    (Re)configure the copier for the following inputs:

      host: hostname to recieve command [user@host is also valid] command: a command to send [default = ‘echo <name>’] launcher: remote service mechanism (i.e. ssh, rsh) [default = ‘ssh’] options: remote service options (i.e. -v, -N, -L) [default = ‘’] background: run in background [default = False] decode: ensure response is ‘ascii’ [default = True] stdin: file-like object to serve as standard input for the remote process

  __firstlineno__ = 36
  __static_attributes__ = ('_stdout', 'background', 'codec', 'host', 'launcher', 'message', 'options', 'stdin')
  config(**kwds)

    configure a remote command using given keywords:

    (Re)configure the copier for the following inputs:

      host: hostname to recieve command [user@host is also valid] command: a command to send [default = ‘echo <name>’] launcher: remote service mechanism (i.e. ssh, rsh) [default = ‘ssh’] options: remote service options (i.e. -v, -N, -L) [default = ‘’] background: run in background [default = False] decode: ensure response is ‘ascii’ [default = True] stdin: file-like object to serve as standard input for the remote process

class Tunnel(name=None, **kwds)

  Bases: `object`

  a ssh-tunnel launcher for parallel and distributed computing.

  create a ssh tunnel launcher

  Inputs:

    name – a unique identifier (string) for the launcher

  MAXPORT = 65535
  MINPORT = 1024
  __annotations_cache__ = {}
  __disconnect()

    disconnect tunnel internals

  __firstlineno__ = 48
  __repr__()

    Return repr(self).

  __static_attributes__ = ('_host', '_launcher', '_lport', '_pid', '_rport', 'connected', 'name')
  _connect(localport, remotehost, remoteport, through=None)
  connect(host, port=None, through=None)

    establish a secure shell tunnel between local and remote host

    Input:

      host – remote hostname [user@host:path is also valid] port – remote port number

    Additional Input:

      through – ‘tunnel-through’ hostname [default = None]

  disconnect()

    destroy the ssh tunnel

  verbose = True

exception TunnelException

  Bases: `Exception`

  Exception for failure to establish ssh tunnel

  __firstlineno__ = 44
  __static_attributes__ = ()

## selector module

This module implements a selector class, which can be used to dispatch events and for event handler wrangling.

class Selector

  Bases: `object`

  Selector object for watching and event notification.

  Takes no initial input.

  _TIMEOUT = 0.5
  __annotations_cache__ = {}
  __firstlineno__ = 25
  __static_attributes__ = ('_exception', '_idle', '_input', '_interrupt', '_output', '_timeout', 'state')
  _cleanup()
  _debug = <Logger pathos.selector (WARNING)>
  _dispatch(handlers, entities)
  _watch()
  notifyOnException(fd, handler)

    add <handler> to the list of routines to call when <fd> raises an exception

  notifyOnInterrupt(handler)

    add <handler> to the list of routines to call when a signal arrives

  notifyOnReadReady(fd, handler)

    add <handler> to the list of routines to call when <fd> is read ready

  notifyOnWriteReady(fd, handler)

    add <handler> to the list of routines to call when <fd> is write ready

  notifyWhenIdle(handler)

    add <handler> to the list of routines to call when a timeout occurs

  watch(timeout=None)

    dispatch events to the registered hanlders

## serial module

This module contains map and pipe interfaces to standard (i.e. serial) python.

Pipe methods provided:

  pipe - blocking communication pipe [returns: value]

Map methods provided:

  map - blocking and ordered worker pool [returns: list] imap - non-blocking and ordered worker pool [returns: iterator]

### Usage

A typical call to a pathos python map will roughly follow this example:

    >>> # instantiate and configure the worker pool
    >>> from pathos.serial import SerialPool
    >>> pool = SerialPool()
    >>>
    >>> # do a blocking map on the chosen function
    >>> print(pool.map(pow, [1,2,3,4], [5,6,7,8]))
    >>>
    >>> # do a non-blocking map, then extract the results from the iterator
    >>> results = pool.imap(pow, [1,2,3,4], [5,6,7,8])
    >>> print("...")
    >>> print(list(results))
    >>>
    >>> # do one item at a time, using a pipe
    >>> print(pool.pipe(pow, 1, 5))
    >>> print(pool.pipe(pow, 2, 6))

Notes

This worker pool leverages the built-in python maps, and thus does not have limitations due to serialization of the function f or the sequences in args. The maps in this worker pool have full functionality whether run from a script or in the python interpreter, and work reliably for both imported and interactively-defined functions.

class SerialPool(*args, **kwds)

  Bases: `AbstractWorkerPool`

  Mapper that leverages standard (i.e. serial) python maps.

  Important class members:

    nodes - number (and potentially description) of workers ncpus - number of worker processors servers - list of worker servers scheduler - the associated scheduler workdir - associated $WORKDIR for scratch calculations/files

  Other class members:

    scatter - True, if uses ‘scatter-gather’ (instead of ‘worker-pool’) source - False, if minimal use of TemporaryFiles is desired timeout - number of seconds to wait for return value from scheduler

  __annotations_cache__ = {}
  __firstlineno__ = 65
  __get_nodes()

    get the number of nodes in the pool

  __set_nodes(nodes)

    set the number of nodes in the pool

  __state__ = {}
  __static_attributes__ = ('_exiting',)
  _exiting = False
  _is_alive(negate=False, run=True)
  clear()

    hard restart

  close()

    close the pool to any new jobs

  imap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and ordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  join()

    cleanup the closed worker processes

  map(f, *args, **kwds)

    run a batch of jobs with a blocking and ordered map

    Returns a list of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  property nodes

    get the number of nodes in the pool

  pipe(f, *args, **kwds)

    submit a job and block until results are available

    Returns result of calling the function f on a selected worker. This function will block until results are available.

  restart(force=False)

    restart a closed pool

  terminate()

    a more abrupt close

__get_nodes__(self)

  get the number of nodes in the pool

__set_nodes__(self, nodes)

  set the number of nodes in the pool

## server module

This module contains the base class for pathos servers, and describes the pathos server interface. If a third-party RPC server is selected, such as ‘parallel python’ (i.e. ‘pp’) or ‘RPyC’, direct calls to the third-party interface are currently used.

class Server

  Bases: `object`

  Server base class for pathos servers for parallel and distributed computing.

  Takes no initial input.

  __annotate_func__ = None
  __annotations_cache__ = {}
  __firstlineno__ = 28
  __static_attributes__ = ('_selector',)
  activate(onTimeout=None, selector=None)

    configure the selector and install the timeout callback

  deactivate()

    turn off the selector

  selector()

    get the selector

  serve(timeout)

    begin serving, and set the timeout

## threading module

This module contains map and pipe interfaces to python’s threading module.

Pipe methods provided:

  pipe - blocking communication pipe [returns: value] apipe - asynchronous communication pipe [returns: object]

Map methods provided:

  map - blocking and ordered worker pool [returns: list] imap - non-blocking and ordered worker pool [returns: iterator] uimap - non-blocking and unordered worker pool [returns: iterator] amap - asynchronous worker pool [returns: object]

### Usage

A typical call to a pathos threading map will roughly follow this example:

    >>> # instantiate and configure the worker pool
    >>> from pathos.threading import ThreadPool
    >>> pool = ThreadPool(nodes=4)
    >>>
    >>> # do a blocking map on the chosen function
    >>> print(pool.map(pow, [1,2,3,4], [5,6,7,8]))
    >>>
    >>> # do a non-blocking map, then extract the results from the iterator
    >>> results = pool.imap(pow, [1,2,3,4], [5,6,7,8])
    >>> print("...")
    >>> print(list(results))
    >>>
    >>> # do an asynchronous map, then get the results
    >>> results = pool.amap(pow, [1,2,3,4], [5,6,7,8])
    >>> while not results.ready():
    ...     time.sleep(5); print(".", end=' ')
    ...
    >>> print(results.get())
    >>>
    >>> # do one item at a time, using a pipe
    >>> print(pool.pipe(pow, 1, 5))
    >>> print(pool.pipe(pow, 2, 6))
    >>>
    >>> # do one item at a time, using an asynchronous pipe
    >>> result1 = pool.apipe(pow, 1, 5)
    >>> result2 = pool.apipe(pow, 2, 6)
    >>> print(result1.get())
    >>> print(result2.get())

Notes

This worker pool leverages the python’s multiprocessing.dummy module, and thus has many of the limitations associated with that module. The function f and the sequences in args must be serializable. The maps in this worker pool have full functionality whether run from a script or in the python interpreter, and work reliably for both imported and interactively-defined functions. Unlike python’s multiprocessing.dummy module, pathos.threading maps can directly utilize functions that require multiple arguments.

class ThreadPool(*args, **kwds)

  Bases: `AbstractWorkerPool`

  Mapper that leverages python’s threading.

  Important class members:

    nodes - number (and potentially description) of workers ncpus - number of worker processors servers - list of worker servers scheduler - the associated scheduler workdir - associated $WORKDIR for scratch calculations/files

  Other class members:

    scatter - True, if uses ‘scatter-gather’ (instead of ‘worker-pool’) source - False, if minimal use of TemporaryFiles is desired timeout - number of seconds to wait for return value from scheduler

  NOTE: if number of nodes is not given, will autodetect processors.

  NOTE: additional keyword input is optional, with:

    id - identifier for the pool initializer - function that takes no input, called when node is spawned initargs - tuple of args for initializers that have args

  __annotations_cache__ = {}
  __firstlineno__ = 78
  __get_nodes()

    get the number of nodes used in the map

  __repr__()

    Return repr(self).

  __set_nodes(nodes)

    set the number of nodes used in the map

  __state__ = {4: <multiprocess.pool.ThreadPool state=RUN pool_size=4>}
  __static_attributes__ = ('__nodes', '_id', '_kwds')
  _clear()

    Remove server with matching state

  _serve(nodes=None)

    Create a new server if one isn’t already initialized

  amap(f, *args, **kwds)

    run a batch of jobs with an asynchronous map

    Returns a results object which containts the results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until all results are retrieved. Use the ready() method on the result object to check if all results are ready. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  apipe(f, *args, **kwds)

    submit a job asynchronously to a queue

    Returns a results object which containts the result of calling the function f on a selected worker. To retrieve the results, call the get() method on the returned results object. The call to get() is blocking, until the result is available. Use the ready() method on the results object to check if the result is ready.

  clear()

    Remove server with matching state

  close()

    close the pool to any new jobs

  imap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and ordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  join()

    cleanup the closed worker processes

  map(f, *args, **kwds)

    run a batch of jobs with a blocking and ordered map

    Returns a list of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

  property nodes

    get the number of nodes used in the map

  property nthreads

    get the number of nodes used in the map

  pipe(f, *args, **kwds)

    submit a job and block until results are available

    Returns result of calling the function f on a selected worker. This function will block until results are available.

  restart(force=False)

    restart a closed pool

  terminate()

    a more abrupt close

  uimap(f, *args, **kwds)

    run a batch of jobs with a non-blocking and unordered map

    Returns a list iterator of results of applying the function f to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence. The order of the resulting sequence is not guaranteed. Some maps accept the chunksize keyword, which causes the sequence to be split into tasks of approximately the given size.

_ThreadPool(processes=None, initializer=None, initargs=())

## util module

utilities for distributed computing

_b(string, codec=None)

  convert string to bytes using the given codec (default is ‘ascii’)

_str(byte, codec=None)

  convert bytes to string using the given codec (default is ‘ascii’)

print_exc_info()

  thread-safe return of string from print_exception call

spawn(onParent, onChild)

  a unidirectional fork wrapper

  Calls onParent(pid, fromchild) in parent process,

    onChild(pid, toparent) in child process.

spawn2(onParent, onChild)

  a bidirectional fork wrapper

  Calls onParent(pid, fromchild, tochild) in parent process,

    onChild(pid, fromparent, toparent) in child process.

## xmlrpc module

- pathos.xmlrpc module documentation

class XMLRPCRequestHandler(server, socket)

  Bases: `BaseHTTPRequestHandler`

  create a XML-RPC request handler

  Override BaseHTTPRequestHandler.__init__(): we need to be able to have (potentially) multiple handler objects at a given time.

  Inputs:

    server – server object to handle requests for socket – socket connection

  __annotate_func__ = None
  __annotations_cache__ = {}
  __firstlineno__ = 188
  __static_attributes__ = ('_server', 'client_address', 'connection', 'rfile', 'wfile')
  _debug = <Logger pathos.xmlrpc (WARNING)>
  _sendResponse(response)

    Write the XML-RPC response

  do_POST()

    Access point from HTTP handler

  log_message(format, *args)

    Overriding BaseHTTPRequestHandler.log_message()

class XMLRPCServer(host, port)

  Bases: `Server`, `SimpleXMLRPCDispatcher`

  extends base pathos server to an XML-RPC dispatcher

  create a XML-RPC server

  Takes two initial inputs:

    host – hostname of XML-RPC server host port – port number for server requests

  __annotate_func__ = None
  __annotations_cache__ = {}
  __firstlineno__ = 60
  __static_attributes__ = ('_activeProcesses', '_socket', 'host', 'port')
  _handleMessageFromChild(selector, fd)

    handler for message from a child process

  _installSocket(host, port)

    prepare a listening socket

  _marshaled_dispatch(data, dispatch_method=None)

    override SimpleXMLRPCDispatcher._marshaled_dispatch() fault string

  _onConnection(selector, fd)

    upon socket connection, establish a request handler

  _onSelectorIdle(selector)

    something to do when there’s no requests

  _onSocketConnection(socket)

    upon socket connections, establish a request handler

  _registerChild(pid, fromchild)

    register a child process so it can be retrieved on select events

  _unRegisterChild(fd)

    remove a child process from active process register

  activate()

    install callbacks

  serve()

    enter the select loop… and wait for service requests