Source code for distributed

"""
Distributed evaluation of genomes.

About compute nodes:
The primary node (=the node which creates and mutates genomes) and the secondary
nodes (=the nodes which evaluate genomes) can execute the same script. The
role of a compute node is determined using the ``mode`` argument of the
DistributedEvaluator. If the mode is MODE_AUTO, the `host_is_local()` function
is used to check if the ``addr`` argument points to the localhost. If it does,
the compute node starts as a primary node, otherwise as a secondary node. If
``mode`` is MODE_PRIMARY, the compute node always starts as a primary node. If
``mode`` is MODE_SECONDARY, the compute node will always start as a secondary node.

There can only be one primary node per NEAT, but any number of secondary nodes.
The primary node will not evaluate any genomes, which means you will always need
at least two compute nodes.

You can run any number of compute nodes on the same physical machine (or VM).
However, if a machine has both a primary node and one or more secondary nodes,
MODE_AUTO cannot be used for those secondary nodes - MODE_SECONDARY will need to be
specified.

NOTE: This module is in a **beta** state, and still *unstable* even in single-machine testing. Reliability is likely to
vary, including depending on the Python version and implementation (e.g., cpython vs pypy) in use and the likelihoods of
timeouts (due to machine and/or network slowness). In particular, while the code can try to reconnect between between
primary and secondary nodes, as noted in the `multiprocessing` documentation this may not work due to data
loss/corruption. Note also that this module is not responsible for starting the script copies on the different compute
nodes, since this is very site/configuration-dependent.

Usage:
1. Import modules and define the evaluation logic (the eval_genome function).
  (After this, check for ``if __name__ == '__main__'``, and put the rest of
  the code inside the body of the statement.)
2. Load config and create a population - here, the variable ``p``.
3. If required, create and add reporters.
4. Create a ``DistributedEvaluator(addr_of_primary_node, b'some_password',
  eval_function, mode=MODE_AUTO)`` - here, the variable ``de``.
5. Call ``de.start(exit_on_stop=True)``. The `start()` call will block on the
  secondary nodes and call `sys.exit(0)` when the NEAT evolution finishes. This
  means that the following code will only be executed on the primary node.
6. Start the evaluation using ``p.run(de.evaluate, number_of_generations)``.
7. Stop the secondary nodes using ``de.stop()``.
8. You are done. You may want to save the winning genome or show some statistics.

See ``examples/xor/evolve-feedforward-distributed.py`` for a complete example.

Utility functions:

``host_is_local(hostname, port=22)`` returns True if ``hostname`` points to
the local node/host. This can be used to check if a compute node will run as
a primary node or as a secondary node with MODE_AUTO.

``chunked(data, chunksize)``: splits data into a list of chunks with at most
``chunksize`` elements.
"""
from __future__ import print_function

import socket
import sys
import time
import warnings

# below still needed for queue.Empty
try:
    # pylint: disable=import-error
    import Queue as queue
except ImportError:
    # pylint: disable=import-error
    import queue

import multiprocessing
from multiprocessing import managers
from argparse import Namespace

# Some of this code is based on
# http://eli.thegreenplace.net/2012/01/24/distributed-computing-in-python-with-multiprocessing
# According to the website, the code is in the public domain
# ('public domain' links to unlicense.org).
# This means that we can use the code from this website.
# Thanks to Eli Bendersky for making his code open for use.


# modes to determine the role of a compute node
# the primary handles the evolution of the genomes
# the secondary handles the evaluation of the genomes
MODE_AUTO = 0  # auto-determine mode
MODE_PRIMARY = MODE_MASTER = 1  # enforce primary mode
MODE_SECONDARY = MODE_SLAVE = 2  # enforce secondary mode

# states to determine whether the secondaries should shut down
_STATE_RUNNING = 0
_STATE_SHUTDOWN = 1
_STATE_FORCED_SHUTDOWN = 2


[docs]class ModeError(RuntimeError): """ An exception raised when a mode-specific method is being called without being in the mode - either a primary-specific method called by a secondary node or a secondary-specific method called by a primary node. """ pass
[docs]def host_is_local(hostname, port=22): # no port specified, just use the ssh port """ Returns True if the hostname points to the localhost, otherwise False. """ hostname = socket.getfqdn(hostname) if hostname in ("localhost", "0.0.0.0", "127.0.0.1", "1.0.0.127.in-addr.arpa", "1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.ip6.arpa"): return True localhost = socket.gethostname() if hostname == localhost: return True localaddrs = socket.getaddrinfo(localhost, port) targetaddrs = socket.getaddrinfo(hostname, port) for (ignored_family, ignored_socktype, ignored_proto, ignored_canonname, sockaddr) in localaddrs: for (ignored_rfamily, ignored_rsocktype, ignored_rproto, ignored_rcanonname, rsockaddr) in targetaddrs: if rsockaddr[0] == sockaddr[0]: return True return False
[docs]def _determine_mode(addr, mode): """ Returns the mode which should be used. If mode is MODE_AUTO, this is determined by checking if 'addr' points to the local host. If it does, return MODE_PRIMARY, else return MODE_SECONDARY. If mode is either MODE_PRIMARY or MODE_SECONDARY, return the 'mode' argument. Otherwise, a ValueError is raised. """ if isinstance(addr, tuple): host = addr[0] elif type(addr) == type(b"binary_string"): host = addr else: raise TypeError("'addr' needs to be a tuple or an bytestring!") if mode == MODE_AUTO: if host_is_local(host): return MODE_PRIMARY return MODE_SECONDARY elif mode in (MODE_SECONDARY, MODE_PRIMARY): return mode else: raise ValueError("Invalid mode {!r}!".format(mode))
[docs]def chunked(data, chunksize): """ Returns a list of chunks containing at most ``chunksize`` elements of data. """ if chunksize < 1: raise ValueError("Chunksize must be at least 1!") if int(chunksize) != chunksize: raise ValueError("Chunksize needs to be an integer") res = [] cur = [] for e in data: cur.append(e) if len(cur) >= chunksize: res.append(cur) cur = [] if cur: res.append(cur) return res
[docs]class _ExtendedManager(object): """A class for managing the multiprocessing.managers.SyncManager""" __safe_for_unpickling__ = True # this may not be safe for unpickling, # but this is required by pickle. def __init__(self, addr, authkey, mode, start=False): self.addr = addr self.authkey = authkey self.mode = _determine_mode(addr, mode) self.manager = None self._secondary_state= multiprocessing.managers.Value(int, _STATE_RUNNING) if start: self.start()
[docs] def __reduce__(self): """ This method is used by pickle to serialize instances of this class. """ return ( self.__class__, (self.addr, self.authkey, self.mode, True), )
[docs] def start(self): """Starts or connects to the manager.""" if self.mode == MODE_PRIMARY: i = self._start() else: i = self._connect() self.manager = i
[docs] def stop(self): """Stops the manager.""" self.manager.shutdown()
[docs] def set_secondary_state(self, value): """Sets the value for 'secondary_state'.""" if value not in (_STATE_RUNNING, _STATE_SHUTDOWN, _STATE_FORCED_SHUTDOWN): raise ValueError( "State {!r} is invalid - needs to be one of _STATE_RUNNING, _STATE_SHUTDOWN, or _STATE_FORCED_SHUTDOWN".format( value) ) if self.manager is None: raise RuntimeError("Manager not started") self.manager.set_state(value)
def _get_secondary_state(self): """ Returns the value for 'secondary_state'. This is required for the manager. """ return self._secondary_state def _get_manager_class(self, register_callables=False): """ Returns a new 'Manager' subclass with registered methods. If 'register_callable' is True, defines the 'callable' arguments. """ class _EvaluatorSyncManager(managers.BaseManager): """ A custom BaseManager. Please see the documentation of `multiprocessing` for more information. """ pass inqueue = queue.Queue() outqueue = queue.Queue() namespace = Namespace() if register_callables: _EvaluatorSyncManager.register( "get_inqueue", callable=lambda: inqueue, ) _EvaluatorSyncManager.register( "get_outqueue", callable=lambda: outqueue, ) _EvaluatorSyncManager.register( "get_state", callable=self._get_secondary_state, ) _EvaluatorSyncManager.register( "set_state", callable=lambda v: self._secondary_state.set(v), ) _EvaluatorSyncManager.register( "get_namespace", callable=lambda: namespace, ) else: _EvaluatorSyncManager.register( "get_inqueue", ) _EvaluatorSyncManager.register( "get_outqueue", ) _EvaluatorSyncManager.register( "get_state", ) _EvaluatorSyncManager.register( "set_state", ) _EvaluatorSyncManager.register( "get_namespace", ) return _EvaluatorSyncManager def _connect(self): """Connects to the manager.""" cls = self._get_manager_class(register_callables=False) ins = cls(address=self.addr, authkey=self.authkey) ins.connect() return ins def _start(self): """Starts the manager.""" cls = self._get_manager_class(register_callables=True) ins = cls(address=self.addr, authkey=self.authkey) ins.start() return ins @property def secondary_state(self): """Whether the secondary nodes should still process elements.""" v = self.manager.get_state() return v.get()
[docs] def get_inqueue(self): """Returns the inqueue.""" if self.manager is None: raise RuntimeError("Manager not started") return self.manager.get_inqueue()
[docs] def get_outqueue(self): """Returns the outqueue.""" if self.manager is None: raise RuntimeError("Manager not started") return self.manager.get_outqueue()
[docs] def get_namespace(self): """Returns the namespace.""" if self.manager is None: raise RuntimeError("Manager not started") return self.manager.get_namespace()
[docs]class DistributedEvaluator(object): """An evaluator working across multiple machines""" def __init__( self, addr, authkey, eval_function, secondary_chunksize=1, num_workers=None, worker_timeout=60, mode=MODE_AUTO, ): """ ``addr`` should be a tuple of (hostname, port) pointing to the machine running the DistributedEvaluator in primary mode. If mode is MODE_AUTO, the mode is determined by checking whether the hostname points to this host or not. ``authkey`` is the password used to restrict access to the manager; see ``Authentication Keys`` in the `multiprocessing` manual for more information. All DistributedEvaluators need to use the same authkey. Note that this needs to be a `bytes` object for Python 3.X, and should be in 2.7 for compatibility (identical in 2.7 to a `str` object). ``eval_function`` should take two arguments (a genome object and the configuration) and return a single float (the genome's fitness). 'secondary_chunksize' specifies the number of genomes that will be sent to a secondary at any one time. ``num_workers`` is the number of child processes to use if in secondary mode. It defaults to None, which means `multiprocessing.cpu_count()` is used to determine this value. If 1 in a secondary node, the process creating the DistributedEvaluator instance will also do the evaulations. ``worker_timeout`` specifies the timeout (in seconds) for a secondary node getting the results from a worker subprocess; if None, there is no timeout. ``mode`` specifies the mode to run in; it defaults to MODE_AUTO. """ self.addr = addr self.authkey = authkey self.eval_function = eval_function self.secondary_chunksize = secondary_chunksize self.slave_chunksize = secondary_chunksize # backward compatibility if num_workers: self.num_workers = num_workers else: try: self.num_workers = max(1,multiprocessing.cpu_count()) except (RuntimeError, AttributeError): # pragma: no cover print("multiprocessing.cpu_count() gave an error; assuming 1", file=sys.stderr) self.num_workers = 1 self.worker_timeout = worker_timeout self.mode = _determine_mode(self.addr, mode) self.em = _ExtendedManager(self.addr, self.authkey, mode=self.mode, start=False) self.inqueue = None self.outqueue = None self.namespace = None self.started = False def __getstate__(self): """Required by the pickle protocol.""" # we do not actually save any state, but we need __getstate__ to be # called. return True # return some nonzero value def __setstate__(self, state): """Called when instances of this class are unpickled.""" self._set_shared_instances()
[docs] def is_primary(self): """Returns True if the caller is the primary node""" return (self.mode == MODE_PRIMARY)
[docs] def is_master(self): # pragma: no cover """Returns True if the caller is the primary (master) node""" warnings.warn("Use is_primary, not is_master", DeprecationWarning) return self.is_primary()
[docs] def start(self, exit_on_stop=True, secondary_wait=0, reconnect=False): """ If the DistributedEvaluator is in primary mode, starts the manager process and returns. In this case, the ``exit_on_stop`` argument will be ignored. If the DistributedEvaluator is in secondary mode, it connects to the manager and waits for tasks. If in secondary mode and ``exit_on_stop`` is True, sys.exit() will be called when the connection is lost. ``secondary_wait`` specifies the time (in seconds) to sleep before actually starting when in secondary mode. If 'reconnect' is True, the secondary nodes will try to reconnect when the connection is lost. In this case, sys.exit() will only be called when 'exit_on_stop' is True and the primary node send a forced shutdown command. """ if self.started: raise RuntimeError("DistributedEvaluator already started!") self.started = True if self.mode == MODE_PRIMARY: self._start_primary() elif self.mode == MODE_SECONDARY: time.sleep(secondary_wait) self._start_secondary() self._secondary_loop(reconnect=reconnect) if exit_on_stop: sys.exit(0) else: raise ValueError("Invalid mode {!r}!".format(self.mode))
[docs] def stop(self, wait=1, shutdown=True, force_secondary_shutdown=False): """ Stops all secondaries. 'wait' specifies the time (in seconds) to wait before shutting down the manager or returning. If 'shutdown', shutdown the manager. If 'force_secondary_shutdown', shutdown the secondary nodes even if they are started with 'reconnect=True'. """ if self.mode != MODE_PRIMARY: raise ModeError("Not in primary mode!") if not self.started: raise RuntimeError("Not yet started!") if force_secondary_shutdown: state = _STATE_FORCED_SHUTDOWN else: state = _STATE_SHUTDOWN self.em.set_secondary_state(state) time.sleep(wait) if shutdown: self.em.stop() self.started = False self.inqueue = self.outqueue = self.namespace = None
def _start_primary(self): """Start as the primary""" self.em.start() self.em.set_secondary_state(_STATE_RUNNING) self._set_shared_instances() def _start_secondary(self): """Start as a secondary.""" self.em.start() self._set_shared_instances() def _set_shared_instances(self): """Sets attributes from the shared instances.""" self.inqueue = self.em.get_inqueue() self.outqueue = self.em.get_outqueue() self.namespace = self.em.get_namespace() def _reset_em(self): """Resets self.em and the shared instances.""" self.em = _ExtendedManager(self.addr, self.authkey, mode=self.mode, start=False) self.em.start() self._set_shared_instances() def _secondary_loop(self, reconnect=False): """The worker loop for the secondary nodes.""" if self.num_workers > 1: pool = multiprocessing.Pool(self.num_workers) else: pool = None should_reconnect = True while should_reconnect: i = 0 running = True try: self._reset_em() except (socket.error, EOFError, IOError, OSError, socket.gaierror, TypeError): continue while running: i += 1 if i % 5 == 0: # for better performance, only check every 5 cycles try: state = self.em.secondary_state except (socket.error, EOFError, IOError, OSError, socket.gaierror, TypeError): if not reconnect: raise else: break if state == _STATE_FORCED_SHUTDOWN: running = False should_reconnect = False elif state == _STATE_SHUTDOWN: running = False if not running: continue try: tasks = self.inqueue.get(block=True, timeout=0.2) except queue.Empty: continue except (socket.error, EOFError, IOError, OSError, socket.gaierror, TypeError): break except (managers.RemoteError, multiprocessing.ProcessError) as e: if ('Empty' in repr(e)) or ('TimeoutError' in repr(e)): continue if (('EOFError' in repr(e)) or ('PipeError' in repr(e)) or ('AuthenticationError' in repr(e))): # Second for Python 3.X, Third for 3.6+ break raise if pool is None: res = [] for genome_id, genome, config in tasks: fitness = self.eval_function(genome, config) res.append((genome_id, fitness)) else: genome_ids = [] jobs = [] for genome_id, genome, config in tasks: genome_ids.append(genome_id) jobs.append( pool.apply_async( self.eval_function, (genome, config) ) ) results = [ job.get(timeout=self.worker_timeout) for job in jobs ] res = zip(genome_ids, results) try: self.outqueue.put(res) except (socket.error, EOFError, IOError, OSError, socket.gaierror, TypeError): break except (managers.RemoteError, multiprocessing.ProcessError) as e: if ('Empty' in repr(e)) or ('TimeoutError' in repr(e)): continue if (('EOFError' in repr(e)) or ('PipeError' in repr(e)) or ('AuthenticationError' in repr(e))): # Second for Python 3.X, Third for 3.6+ break raise if not reconnect: should_reconnect = False break if pool is not None: pool.terminate()
[docs] def evaluate(self, genomes, config): """ Evaluates the genomes. This method raises a ModeError if the DistributedEvaluator is not in primary mode. """ if self.mode != MODE_PRIMARY: raise ModeError("Not in primary mode!") tasks = [(genome_id, genome, config) for genome_id, genome in genomes] id2genome = {genome_id: genome for genome_id, genome in genomes} tasks = chunked(tasks, self.secondary_chunksize) n_tasks = len(tasks) for task in tasks: self.inqueue.put(task) tresults = [] while len(tresults) < n_tasks: try: sr = self.outqueue.get(block=True, timeout=0.2) except (queue.Empty, managers.RemoteError): continue tresults.append(sr) results = [] for sr in tresults: results += sr for genome_id, fitness in results: genome = id2genome[genome_id] genome.fitness = fitness