client.py
1875 lines
| 67.0 KiB
| text/x-python
|
PythonLexer
|
r16567 | """A semi-synchronous Client for IPython parallel""" | ||
|
r4018 | |||
|
r16567 | # Copyright (c) IPython Development Team. | ||
# Distributed under the terms of the Modified BSD License. | ||||
|
r4018 | |||
|
r13348 | from __future__ import print_function | ||
|
r3539 | |||
|
r3575 | import os | ||
|
r3631 | import json | ||
|
r4155 | import sys | ||
|
r6484 | from threading import Thread, Event | ||
|
r3551 | import time | ||
|
r3631 | import warnings | ||
from datetime import datetime | ||||
|
r3581 | from getpass import getpass | ||
|
r3540 | from pprint import pprint | ||
|
r3631 | |||
|
r3614 | pjoin = os.path.join | ||
|
r3540 | |||
|
r3551 | import zmq | ||
|
r4071 | from IPython.config.configurable import MultipleInstanceError | ||
from IPython.core.application import BaseIPythonApplication | ||||
|
r7476 | from IPython.core.profiledir import ProfileDir, ProfileDirError | ||
|
r4071 | |||
|
r12229 | from IPython.utils.capture import RichOutput | ||
|
r7026 | from IPython.utils.coloransi import TermColors | ||
|
r13520 | from IPython.utils.jsonutil import rekey, extract_dates, parse_date | ||
|
r12591 | from IPython.utils.localinterfaces import localhost, is_local_ip | ||
|
r17139 | from IPython.utils.path import get_ipython_dir, compress_user | ||
|
r13361 | from IPython.utils.py3compat import cast_bytes, string_types, xrange, iteritems | ||
|
r5344 | from IPython.utils.traitlets import (HasTraits, Integer, Instance, Unicode, | ||
|
r6484 | Dict, List, Bool, Set, Any) | ||
|
r3539 | from IPython.external.decorator import decorator | ||
|
r5821 | from IPython.parallel import Reference | ||
|
r3673 | from IPython.parallel import error | ||
from IPython.parallel import util | ||||
|
r9372 | from IPython.kernel.zmq.session import Session, Message | ||
from IPython.kernel.zmq import serialize | ||||
|
r4006 | |||
|
r3666 | from .asyncresult import AsyncResult, AsyncHubResult | ||
|
r3642 | from .view import DirectView, LoadBalancedView | ||
|
r3539 | |||
|
r3584 | #-------------------------------------------------------------------------- | ||
|
r3555 | # Decorators for Client methods | ||
#-------------------------------------------------------------------------- | ||||
|
r3539 | @decorator | ||
|
r3664 | def spin_first(f, self, *args, **kwargs): | ||
|
r3555 | """Call spin() to sync state prior to calling the method.""" | ||
|
r3539 | self.spin() | ||
return f(self, *args, **kwargs) | ||||
|
r3598 | |||
#-------------------------------------------------------------------------- | ||||
# Classes | ||||
#-------------------------------------------------------------------------- | ||||
|
r6819 | |||
|
r12229 | class ExecuteReply(RichOutput): | ||
|
r6819 | """wrapper for finished Execute results""" | ||
def __init__(self, msg_id, content, metadata): | ||||
self.msg_id = msg_id | ||||
self._content = content | ||||
self.execution_count = content['execution_count'] | ||||
self.metadata = metadata | ||||
|
r12229 | # RichOutput overrides | ||
@property | ||||
def source(self): | ||||
|
r16568 | execute_result = self.metadata['execute_result'] | ||
if execute_result: | ||||
return execute_result.get('source', '') | ||||
|
r12229 | |||
@property | ||||
def data(self): | ||||
|
r16568 | execute_result = self.metadata['execute_result'] | ||
if execute_result: | ||||
return execute_result.get('data', {}) | ||||
|
r12229 | |||
@property | ||||
def _metadata(self): | ||||
|
r16568 | execute_result = self.metadata['execute_result'] | ||
if execute_result: | ||||
return execute_result.get('metadata', {}) | ||||
|
r12229 | |||
def display(self): | ||||
from IPython.display import publish_display_data | ||||
|
r16585 | publish_display_data(self.data, self.metadata) | ||
|
r12229 | |||
def _repr_mime_(self, mime): | ||||
if mime not in self.data: | ||||
return | ||||
data = self.data[mime] | ||||
if mime in self._metadata: | ||||
return data, self._metadata[mime] | ||||
else: | ||||
return data | ||||
|
r6819 | def __getitem__(self, key): | ||
return self.metadata[key] | ||||
def __getattr__(self, key): | ||||
if key not in self.metadata: | ||||
raise AttributeError(key) | ||||
return self.metadata[key] | ||||
def __repr__(self): | ||||
|
r16568 | execute_result = self.metadata['execute_result'] or {'data':{}} | ||
text_out = execute_result['data'].get('text/plain', '') | ||||
|
r6819 | if len(text_out) > 32: | ||
text_out = text_out[:29] + '...' | ||||
return "<ExecuteReply[%i]: %s>" % (self.execution_count, text_out) | ||||
|
r7026 | def _repr_pretty_(self, p, cycle): | ||
|
r16568 | execute_result = self.metadata['execute_result'] or {'data':{}} | ||
text_out = execute_result['data'].get('text/plain', '') | ||||
|
r7026 | |||
if not text_out: | ||||
return | ||||
try: | ||||
ip = get_ipython() | ||||
except NameError: | ||||
colors = "NoColor" | ||||
else: | ||||
colors = ip.colors | ||||
if colors == "NoColor": | ||||
out = normal = "" | ||||
else: | ||||
out = TermColors.Red | ||||
normal = TermColors.Normal | ||||
|
r7239 | if '\n' in text_out and not text_out.startswith('\n'): | ||
# add newline for multiline reprs | ||||
text_out = '\n' + text_out | ||||
|
r7026 | p.text( | ||
|
r7239 | out + u'Out[%i:%i]: ' % ( | ||
self.metadata['engine_id'], self.execution_count | ||||
) + normal + text_out | ||||
|
r7026 | ) | ||
|
r6819 | |||
|
r3602 | class Metadata(dict): | ||
|
r3607 | """Subclass of dict for initializing metadata values. | ||
|
r4872 | |||
|
r3607 | Attribute access works on keys. | ||
|
r4872 | |||
|
r3607 | These objects have a strict set of keys - errors will raise if you try | ||
to add new keys. | ||||
""" | ||||
|
r3602 | def __init__(self, *args, **kwargs): | ||
dict.__init__(self) | ||||
md = {'msg_id' : None, | ||||
'submitted' : None, | ||||
'started' : None, | ||||
'completed' : None, | ||||
'received' : None, | ||||
'engine_uuid' : None, | ||||
'engine_id' : None, | ||||
'follow' : None, | ||||
'after' : None, | ||||
'status' : None, | ||||
|
r16567 | 'execute_input' : None, | ||
|
r16568 | 'execute_result' : None, | ||
|
r16569 | 'error' : None, | ||
|
r3602 | 'stdout' : '', | ||
'stderr' : '', | ||||
|
r6812 | 'outputs' : [], | ||
|
r8103 | 'data': {}, | ||
|
r7494 | 'outputs_ready' : False, | ||
|
r3602 | } | ||
self.update(md) | ||||
self.update(dict(*args, **kwargs)) | ||||
|
r4872 | |||
|
r3607 | def __getattr__(self, key): | ||
"""getattr aliased to getitem""" | ||||
|
r13360 | if key in self: | ||
|
r3607 | return self[key] | ||
else: | ||||
raise AttributeError(key) | ||||
|
r3602 | |||
|
r3607 | def __setattr__(self, key, value): | ||
"""setattr aliased to setitem, with strict""" | ||||
|
r13360 | if key in self: | ||
|
r3607 | self[key] = value | ||
else: | ||||
raise AttributeError(key) | ||||
|
r4872 | |||
|
r3607 | def __setitem__(self, key, value): | ||
"""strict static key enforcement""" | ||||
|
r13360 | if key in self: | ||
|
r3607 | dict.__setitem__(self, key, value) | ||
else: | ||||
raise KeyError(key) | ||||
|
r4872 | |||
|
r3602 | |||
|
r3636 | class Client(HasTraits): | ||
|
r3664 | """A semi-synchronous client to the IPython ZMQ cluster | ||
|
r4872 | |||
|
r3555 | Parameters | ||
---------- | ||||
|
r4872 | |||
|
r7889 | url_file : str/unicode; path to ipcontroller-client.json | ||
This JSON file should contain all the information needed to connect to a cluster, | ||||
and is likely the only argument needed. | ||||
|
r3619 | Connection information for the Hub's registration. If a json connector | ||
file is given, then likely no further configuration is necessary. | ||||
|
r3620 | [Default: use profile] | ||
profile : bytes | ||||
The name of the Cluster profile to be used to find connector information. | ||||
|
r4071 | If run from an IPython application, the default profile will be the same | ||
as the running application, otherwise it will be 'default'. | ||||
|
r8233 | cluster_id : str | ||
String id to added to runtime files, to prevent name collisions when using | ||||
multiple clusters with a single profile simultaneously. | ||||
When set, will look for files named like: 'ipcontroller-<cluster_id>-client.json' | ||||
Since this is text inserted into filenames, typical recommendations apply: | ||||
Simple character strings are ideal, and spaces are not recommended (but | ||||
should generally work) | ||||
|
r3572 | context : zmq.Context | ||
|
r3620 | Pass an existing zmq.Context instance, otherwise the client will create its own. | ||
|
r3572 | debug : bool | ||
flag for lots of message printing for debug purposes | ||||
|
r4872 | timeout : int/float | ||
|
r4013 | time (in seconds) to wait for connection replies from the Hub | ||
[Default: 10] | ||||
|
r4872 | |||
|
r4013 | #-------------- session related args ---------------- | ||
|
r4872 | |||
|
r4013 | config : Config object | ||
If specified, this will be relayed to the Session for configuration | ||||
username : str | ||||
set username for the session object | ||||
|
r4872 | |||
|
r3572 | #-------------- ssh related args ---------------- | ||
# These are args for configuring the ssh tunnel to be used | ||||
# credentials are used to forward connections over ssh to the Controller | ||||
# Note that the ip given in `addr` needs to be relative to sshserver | ||||
# The most basic case is to leave addr as pointing to localhost (127.0.0.1), | ||||
|
r4872 | # and set sshserver as the same machine the Controller is on. However, | ||
|
r3572 | # the only requirement is that sshserver is able to see the Controller | ||
# (i.e. is within the same trusted network). | ||||
|
r4872 | |||
|
r3572 | sshserver : str | ||
A string of the form passed to ssh, i.e. 'server.tld' or 'user@server.tld:port' | ||||
If keyfile or password is specified, and this is not, it will default to | ||||
the ip given in addr. | ||||
|
r4589 | sshkey : str; path to ssh private key file | ||
|
r3572 | This specifies a key to be used in ssh login, default None. | ||
Regular default ssh keys will be used without specifying this argument. | ||||
|
r4872 | password : str | ||
|
r3572 | Your ssh password to sshserver. Note that if this is left None, | ||
you will be prompted for it if passwordless key based login is unavailable. | ||||
|
r3619 | paramiko : bool | ||
flag for whether to use paramiko instead of shell ssh for tunneling. | ||||
[default: True on win32, False else] | ||||
|
r4872 | |||
|
r3539 | Attributes | ||
---------- | ||||
|
r4872 | |||
|
r3664 | ids : list of int engine IDs | ||
|
r3539 | requesting the ids attribute always synchronizes | ||
the registration state. To request ids without synchronization, | ||||
|
r3575 | use semi-private _ids attributes. | ||
|
r4872 | |||
|
r3539 | history : list of msg_ids | ||
a list of msg_ids, keeping track of all the execution | ||||
|
r3555 | messages you have submitted in order. | ||
|
r4872 | |||
|
r3539 | outstanding : set of msg_ids | ||
a set of msg_ids that have been submitted, but whose | ||||
|
r3555 | results have not yet been received. | ||
|
r4872 | |||
|
r3539 | results : dict | ||
a dict of all our results, keyed by msg_id | ||||
|
r4872 | |||
|
r3539 | block : bool | ||
determines default behavior when block not specified | ||||
in execution methods | ||||
|
r4872 | |||
|
r3539 | Methods | ||
------- | ||||
|
r4872 | |||
|
r3635 | spin | ||
flushes incoming results and registration state changes | ||||
control methods spin, and requesting `ids` also ensures up to date | ||||
|
r4872 | |||
|
r3664 | wait | ||
|
r3635 | wait on one or more msg_ids | ||
|
r4872 | |||
|
r3635 | execution methods | ||
apply | ||||
|
r3539 | legacy: execute, run | ||
|
r4872 | |||
|
r3664 | data movement | ||
push, pull, scatter, gather | ||||
|
r4872 | |||
|
r3635 | query methods | ||
|
r3664 | queue_status, get_result, purge, result_status | ||
|
r4872 | |||
|
r3635 | control methods | ||
abort, shutdown | ||||
|
r4872 | |||
|
r3539 | """ | ||
|
r4872 | |||
|
r3636 | block = Bool(False) | ||
|
r3651 | outstanding = Set() | ||
results = Instance('collections.defaultdict', (dict,)) | ||||
metadata = Instance('collections.defaultdict', (Metadata,)) | ||||
|
r3636 | history = List() | ||
debug = Bool(False) | ||||
|
r6484 | _spin_thread = Any() | ||
_stop_spinning = Any() | ||||
|
r4872 | |||
|
r4071 | profile=Unicode() | ||
def _profile_default(self): | ||||
if BaseIPythonApplication.initialized(): | ||||
# an IPython app *might* be running, try to get its profile | ||||
try: | ||||
return BaseIPythonApplication.instance().profile | ||||
except (AttributeError, MultipleInstanceError): | ||||
# could be a *different* subclass of config.Application, | ||||
# which would raise one of these two errors. | ||||
return u'default' | ||||
else: | ||||
return u'default' | ||||
|
r4872 | |||
|
r3651 | _outstanding_dict = Instance('collections.defaultdict', (set,)) | ||
|
r3636 | _ids = List() | ||
_connected=Bool(False) | ||||
_ssh=Bool(False) | ||||
_context = Instance('zmq.Context') | ||||
_config = Dict() | ||||
|
r3664 | _engines=Instance(util.ReverseDict, (), {}) | ||
|
r3657 | # _hub_socket=Instance('zmq.Socket') | ||
|
r3636 | _query_socket=Instance('zmq.Socket') | ||
_control_socket=Instance('zmq.Socket') | ||||
_iopub_socket=Instance('zmq.Socket') | ||||
_notification_socket=Instance('zmq.Socket') | ||||
|
r3664 | _mux_socket=Instance('zmq.Socket') | ||
_task_socket=Instance('zmq.Socket') | ||||
|
r3988 | _task_scheme=Unicode() | ||
|
r3636 | _closed = False | ||
|
r5344 | _ignored_control_replies=Integer(0) | ||
_ignored_hub_replies=Integer(0) | ||||
|
r4872 | |||
|
r4071 | def __new__(self, *args, **kw): | ||
# don't raise on positional args | ||||
return HasTraits.__new__(self, **kw) | ||||
|
r4872 | |||
|
r7889 | def __init__(self, url_file=None, profile=None, profile_dir=None, ipython_dir=None, | ||
context=None, debug=False, | ||||
|
r3575 | sshserver=None, sshkey=None, password=None, paramiko=None, | ||
|
r8233 | timeout=10, cluster_id=None, **extra_args | ||
|
r3614 | ): | ||
|
r4071 | if profile: | ||
super(Client, self).__init__(debug=debug, profile=profile) | ||||
else: | ||||
super(Client, self).__init__(debug=debug) | ||||
|
r3539 | if context is None: | ||
|
r3661 | context = zmq.Context.instance() | ||
|
r3636 | self._context = context | ||
|
r6484 | self._stop_spinning = Event() | ||
|
r7889 | |||
if 'url_or_file' in extra_args: | ||||
url_file = extra_args['url_or_file'] | ||||
warnings.warn("url_or_file arg no longer supported, use url_file", DeprecationWarning) | ||||
if url_file and util.is_url(url_file): | ||||
raise ValueError("single urls cannot be specified, url-files must be used.") | ||||
|
r4872 | |||
|
r4071 | self._setup_profile_dir(self.profile, profile_dir, ipython_dir) | ||
|
r7889 | |||
|
r3614 | if self._cd is not None: | ||
|
r7889 | if url_file is None: | ||
|
r8234 | if not cluster_id: | ||
|
r8233 | client_json = 'ipcontroller-client.json' | ||
else: | ||||
client_json = 'ipcontroller-%s-client.json' % cluster_id | ||||
|
r8273 | url_file = pjoin(self._cd.security_dir, client_json) | ||
|
r7889 | if url_file is None: | ||
|
r6869 | raise ValueError( | ||
"I can't find enough information to connect to a hub!" | ||||
|
r7889 | " Please specify at least one of url_file or profile." | ||
|
r6869 | ) | ||
|
r7889 | |||
|
r17139 | if not os.path.exists(url_file): | ||
raise IOError("Connection file %r not found. Is a controller running?" % \ | ||||
compress_user(url_file) | ||||
) | ||||
|
r7889 | with open(url_file) as f: | ||
cfg = json.load(f) | ||||
self._task_scheme = cfg['task_scheme'] | ||||
|
r4872 | |||
|
r3614 | # sync defaults from args, json: | ||
if sshserver: | ||||
cfg['ssh'] = sshserver | ||||
|
r7889 | |||
|
r3614 | location = cfg.setdefault('location', None) | ||
|
r7890 | |||
proto,addr = cfg['interface'].split('://') | ||||
|
r8354 | addr = util.disambiguate_ip_address(addr, location) | ||
|
r7890 | cfg['interface'] = "%s://%s" % (proto, addr) | ||
# turn interface,port into full urls: | ||||
for key in ('control', 'task', 'mux', 'iopub', 'notification', 'registration'): | ||||
cfg[key] = cfg['interface'] + ':%i' % cfg[key] | ||||
|
r7889 | url = cfg['registration'] | ||
|
r7890 | |||
|
r12591 | if location is not None and addr == localhost(): | ||
|
r4119 | # location specified, and connection is expected to be local | ||
|
r12591 | if not is_local_ip(location) and not sshserver: | ||
|
r4119 | # load ssh from JSON *only* if the controller is not on | ||
# this machine | ||||
sshserver=cfg['ssh'] | ||||
|
r12591 | if not is_local_ip(location) and not sshserver: | ||
|
r4119 | # warn if no ssh specified, but SSH is probably needed | ||
# This is only a warning, because the most likely cause | ||||
# is a local Controller on a laptop whose IP is dynamic | ||||
|
r4117 | warnings.warn(""" | ||
Controller appears to be listening on localhost, but not on this machine. | ||||
If this is true, you should specify Client(...,sshserver='you@%s') | ||||
or instruct your controller to listen on an external IP."""%location, | ||||
RuntimeWarning) | ||||
|
r4207 | elif not sshserver: | ||
# otherwise sync with cfg | ||||
sshserver = cfg['ssh'] | ||||
|
r4872 | |||
|
r3614 | self._config = cfg | ||
|
r4872 | |||
|
r3575 | self._ssh = bool(sshserver or sshkey or password) | ||
|
r3572 | if self._ssh and sshserver is None: | ||
|
r3614 | # default to ssh via localhost | ||
|
r7890 | sshserver = addr | ||
|
r3572 | if self._ssh and password is None: | ||
|
r17054 | from zmq.ssh import tunnel | ||
|
r3575 | if tunnel.try_passwordless_ssh(sshserver, sshkey, paramiko): | ||
|
r3572 | password=False | ||
else: | ||||
password = getpass("SSH Password for %s: "%sshserver) | ||||
|
r3575 | ssh_kwargs = dict(keyfile=sshkey, password=password, paramiko=paramiko) | ||
|
r4872 | |||
|
r4013 | # configure and construct the session | ||
|
r11987 | try: | ||
extra_args['packer'] = cfg['pack'] | ||||
extra_args['unpacker'] = cfg['unpack'] | ||||
extra_args['key'] = cast_bytes(cfg['key']) | ||||
extra_args['signature_scheme'] = cfg['signature_scheme'] | ||||
except KeyError as exc: | ||||
msg = '\n'.join([ | ||||
"Connection file is invalid (missing '{}'), possibly from an old version of IPython.", | ||||
"If you are reusing connection files, remove them and start ipcontroller again." | ||||
]) | ||||
raise ValueError(msg.format(exc.message)) | ||||
|
r7889 | |||
|
r4013 | self.session = Session(**extra_args) | ||
|
r4872 | |||
|
r4725 | self._query_socket = self._context.socket(zmq.DEALER) | ||
|
r7888 | |||
|
r3572 | if self._ssh: | ||
|
r17054 | from zmq.ssh import tunnel | ||
|
r7890 | tunnel.tunnel_connection(self._query_socket, cfg['registration'], sshserver, **ssh_kwargs) | ||
|
r3572 | else: | ||
|
r7890 | self._query_socket.connect(cfg['registration']) | ||
|
r4872 | |||
|
r3636 | self.session.debug = self.debug | ||
|
r4872 | |||
|
r3539 | self._notification_handlers = {'registration_notification' : self._register_engine, | ||
'unregistration_notification' : self._unregister_engine, | ||||
|
r3664 | 'shutdown_notification' : lambda msg: self.close(), | ||
|
r3539 | } | ||
self._queue_handlers = {'execute_reply' : self._handle_execute_reply, | ||||
'apply_reply' : self._handle_apply_reply} | ||||
|
r12252 | |||
try: | ||||
self._connect(sshserver, ssh_kwargs, timeout) | ||||
except: | ||||
self.close(linger=0) | ||||
raise | ||||
|
r7476 | |||
# last step: setup magics, if we are in IPython: | ||||
try: | ||||
ip = get_ipython() | ||||
except NameError: | ||||
return | ||||
else: | ||||
if 'px' not in ip.magics_manager.magics: | ||||
# in IPython but we are the first Client. | ||||
# activate a default view for parallel magics. | ||||
self.activate() | ||||
|
r4872 | |||
|
r3654 | def __del__(self): | ||
"""cleanup sockets, but _not_ context.""" | ||||
self.close() | ||||
|
r4872 | |||
|
r3992 | def _setup_profile_dir(self, profile, profile_dir, ipython_dir): | ||
|
r3614 | if ipython_dir is None: | ||
ipython_dir = get_ipython_dir() | ||||
|
r3992 | if profile_dir is not None: | ||
|
r3614 | try: | ||
|
r3992 | self._cd = ProfileDir.find_profile_dir(profile_dir) | ||
|
r3641 | return | ||
|
r3992 | except ProfileDirError: | ||
|
r3614 | pass | ||
elif profile is not None: | ||||
try: | ||||
|
r3992 | self._cd = ProfileDir.find_profile_dir_by_name( | ||
|
r3614 | ipython_dir, profile) | ||
|
r3641 | return | ||
|
r3992 | except ProfileDirError: | ||
|
r3614 | pass | ||
|
r3641 | self._cd = None | ||
|
r4872 | |||
|
r3539 | def _update_engines(self, engines): | ||
|
r3555 | """Update our engines dict and _ids from a dict of the form: {id:uuid}.""" | ||
|
r13361 | for k,v in iteritems(engines): | ||
|
r3539 | eid = int(k) | ||
|
r7891 | if eid not in self._engines: | ||
self._ids.append(eid) | ||||
|
r4155 | self._engines[eid] = v | ||
|
r3625 | self._ids = sorted(self._ids) | ||
|
r13383 | if sorted(self._engines.keys()) != list(range(len(self._engines))) and \ | ||
|
r3664 | self._task_scheme == 'pure' and self._task_socket: | ||
|
r3622 | self._stop_scheduling_tasks() | ||
|
r4872 | |||
|
r3622 | def _stop_scheduling_tasks(self): | ||
"""Stop scheduling tasks because an engine has been unregistered | ||||
from a pure ZMQ scheduler. | ||||
""" | ||||
|
r3664 | self._task_socket.close() | ||
self._task_socket = None | ||||
|
r3622 | msg = "An engine has been unregistered, and we are using pure " +\ | ||
"ZMQ task scheduling. Task farming will be disabled." | ||||
if self.outstanding: | ||||
msg += " If you were running tasks when this happened, " +\ | ||||
"some `outstanding` msg_ids may never resolve." | ||||
warnings.warn(msg, RuntimeWarning) | ||||
|
r4872 | |||
|
r3539 | def _build_targets(self, targets): | ||
|
r3555 | """Turn valid target IDs or 'all' into two lists: | ||
(int_ids, uuids). | ||||
""" | ||||
|
r3777 | if not self._ids: | ||
# flush notification socket if no engines yet, just in case | ||||
if not self.ids: | ||||
raise error.NoEnginesRegistered("Can't build targets without any engines") | ||||
|
r4872 | |||
|
r3539 | if targets is None: | ||
targets = self._ids | ||||
|
r13353 | elif isinstance(targets, string_types): | ||
|
r3539 | if targets.lower() == 'all': | ||
targets = self._ids | ||||
else: | ||||
raise TypeError("%r not valid str target, must be 'all'"%(targets)) | ||||
elif isinstance(targets, int): | ||||
|
r3665 | if targets < 0: | ||
targets = self.ids[targets] | ||||
|
r3777 | if targets not in self._ids: | ||
|
r3665 | raise IndexError("No such engine: %i"%targets) | ||
|
r3539 | targets = [targets] | ||
|
r4872 | |||
|
r3665 | if isinstance(targets, slice): | ||
|
r13383 | indices = list(range(len(self._ids))[targets]) | ||
|
r3665 | ids = self.ids | ||
targets = [ ids[i] for i in indices ] | ||||
|
r4872 | |||
|
r3665 | if not isinstance(targets, (tuple, list, xrange)): | ||
raise TypeError("targets by int/slice/collection of ints only, not %s"%(type(targets))) | ||||
|
r4872 | |||
|
r6813 | return [cast_bytes(self._engines[t]) for t in targets], list(targets) | ||
|
r4872 | |||
|
r3664 | def _connect(self, sshserver, ssh_kwargs, timeout): | ||
"""setup all our socket connections to the cluster. This is called from | ||||
|
r3555 | __init__.""" | ||
|
r4872 | |||
|
r3622 | # Maybe allow reconnecting? | ||
|
r3539 | if self._connected: | ||
return | ||||
self._connected=True | ||||
|
r3572 | |||
|
r3614 | def connect_socket(s, url): | ||
|
r3572 | if self._ssh: | ||
|
r17054 | from zmq.ssh import tunnel | ||
|
r3614 | return tunnel.tunnel_connection(s, url, sshserver, **ssh_kwargs) | ||
|
r3572 | else: | ||
|
r3614 | return s.connect(url) | ||
|
r4872 | |||
|
r3657 | self.session.send(self._query_socket, 'connection_request') | ||
|
r4072 | # use Poller because zmq.select has wrong units in pyzmq 2.1.7 | ||
poller = zmq.Poller() | ||||
poller.register(self._query_socket, zmq.POLLIN) | ||||
# poll expects milliseconds, timeout is seconds | ||||
evts = poller.poll(timeout*1000) | ||||
if not evts: | ||||
|
r3664 | raise error.TimeoutError("Hub connection request timed out") | ||
|
r3657 | idents,msg = self.session.recv(self._query_socket,mode=0) | ||
|
r3540 | if self.debug: | ||
pprint(msg) | ||||
|
r7889 | content = msg['content'] | ||
# self._config['registration'] = dict(content) | ||||
cfg = self._config | ||||
if content['status'] == 'ok': | ||||
self._mux_socket = self._context.socket(zmq.DEALER) | ||||
connect_socket(self._mux_socket, cfg['mux']) | ||||
self._task_socket = self._context.socket(zmq.DEALER) | ||||
connect_socket(self._task_socket, cfg['task']) | ||||
self._notification_socket = self._context.socket(zmq.SUB) | ||||
self._notification_socket.setsockopt(zmq.SUBSCRIBE, b'') | ||||
connect_socket(self._notification_socket, cfg['notification']) | ||||
self._control_socket = self._context.socket(zmq.DEALER) | ||||
connect_socket(self._control_socket, cfg['control']) | ||||
self._iopub_socket = self._context.socket(zmq.SUB) | ||||
self._iopub_socket.setsockopt(zmq.SUBSCRIBE, b'') | ||||
connect_socket(self._iopub_socket, cfg['iopub']) | ||||
self._update_engines(dict(content['engines'])) | ||||
|
r3539 | else: | ||
self._connected = False | ||||
raise Exception("Failed to connect!") | ||||
|
r4872 | |||
|
r3555 | #-------------------------------------------------------------------------- | ||
# handlers and callbacks for incoming messages | ||||
#-------------------------------------------------------------------------- | ||||
|
r4872 | |||
|
r3639 | def _unwrap_exception(self, content): | ||
|
r3664 | """unwrap exception, and remap engine_id to int.""" | ||
|
r3644 | e = error.unwrap_exception(content) | ||
|
r3658 | # print e.traceback | ||
|
r3639 | if e.engine_info: | ||
|
r3641 | e_uuid = e.engine_info['engine_uuid'] | ||
|
r3639 | eid = self._engines[e_uuid] | ||
|
r3641 | e.engine_info['engine_id'] = eid | ||
|
r3639 | return e | ||
|
r4872 | |||
|
r7957 | def _extract_metadata(self, msg): | ||
header = msg['header'] | ||||
parent = msg['parent_header'] | ||||
msg_meta = msg['metadata'] | ||||
content = msg['content'] | ||||
|
r3598 | md = {'msg_id' : parent['msg_id'], | ||
'received' : datetime.now(), | ||||
|
r7957 | 'engine_uuid' : msg_meta.get('engine', None), | ||
'follow' : msg_meta.get('follow', []), | ||||
'after' : msg_meta.get('after', []), | ||||
|
r3602 | 'status' : content['status'], | ||
} | ||||
|
r4872 | |||
|
r3607 | if md['engine_uuid'] is not None: | ||
md['engine_id'] = self._engines.get(md['engine_uuid'], None) | ||||
|
r4872 | |||
|
r3607 | if 'date' in parent: | ||
|
r4006 | md['submitted'] = parent['date'] | ||
|
r7957 | if 'started' in msg_meta: | ||
|
r13520 | md['started'] = parse_date(msg_meta['started']) | ||
|
r3607 | if 'date' in header: | ||
|
r4006 | md['completed'] = header['date'] | ||
|
r3598 | return md | ||
|
r4872 | |||
|
r3651 | def _register_engine(self, msg): | ||
"""Register a new engine, and update our connection info.""" | ||||
content = msg['content'] | ||||
eid = content['id'] | ||||
|
r7891 | d = {eid : content['uuid']} | ||
|
r3651 | self._update_engines(d) | ||
def _unregister_engine(self, msg): | ||||
"""Unregister an engine that has died.""" | ||||
content = msg['content'] | ||||
eid = int(content['id']) | ||||
if eid in self._ids: | ||||
self._ids.remove(eid) | ||||
uuid = self._engines.pop(eid) | ||||
|
r4872 | |||
|
r3651 | self._handle_stranded_msgs(eid, uuid) | ||
|
r4872 | |||
|
r3664 | if self._task_socket and self._task_scheme == 'pure': | ||
|
r3651 | self._stop_scheduling_tasks() | ||
|
r4872 | |||
|
r3651 | def _handle_stranded_msgs(self, eid, uuid): | ||
"""Handle messages known to be on an engine when the engine unregisters. | ||||
|
r4872 | |||
|
r3651 | It is possible that this will fire prematurely - that is, an engine will | ||
go down after completing a result, and the client will be notified | ||||
of the unregistration and later receive the successful result. | ||||
""" | ||||
|
r4872 | |||
|
r3651 | outstanding = self._outstanding_dict[uuid] | ||
|
r4872 | |||
|
r3651 | for msg_id in list(outstanding): | ||
if msg_id in self.results: | ||||
|
r4872 | # we already | ||
|
r3651 | continue | ||
try: | ||||
raise error.EngineError("Engine %r died while running task %r"%(eid, msg_id)) | ||||
except: | ||||
content = error.wrap_exception() | ||||
# build a fake message: | ||||
|
r8199 | msg = self.session.msg('apply_reply', content=content) | ||
msg['parent_header']['msg_id'] = msg_id | ||||
msg['metadata']['engine'] = uuid | ||||
|
r3651 | self._handle_apply_reply(msg) | ||
|
r4872 | |||
|
r3539 | def _handle_execute_reply(self, msg): | ||
|
r3598 | """Save the reply to an execute_request into our results. | ||
|
r4872 | |||
|
r3598 | execute messages are never actually used. apply is used instead. | ||
""" | ||||
|
r4872 | |||
|
r3539 | parent = msg['parent_header'] | ||
msg_id = parent['msg_id'] | ||||
if msg_id not in self.outstanding: | ||||
|
r3607 | if msg_id in self.history: | ||
|
r13386 | print("got stale result: %s"%msg_id) | ||
|
r3607 | else: | ||
|
r13386 | print("got unknown result: %s"%msg_id) | ||
|
r3539 | else: | ||
self.outstanding.remove(msg_id) | ||||
|
r6793 | |||
content = msg['content'] | ||||
header = msg['header'] | ||||
# construct metadata: | ||||
md = self.metadata[msg_id] | ||||
|
r7957 | md.update(self._extract_metadata(msg)) | ||
|
r6793 | # is this redundant? | ||
self.metadata[msg_id] = md | ||||
e_outstanding = self._outstanding_dict[md['engine_uuid']] | ||||
if msg_id in e_outstanding: | ||||
e_outstanding.remove(msg_id) | ||||
# construct result: | ||||
if content['status'] == 'ok': | ||||
|
r6819 | self.results[msg_id] = ExecuteReply(msg_id, content, md) | ||
|
r6793 | elif content['status'] == 'aborted': | ||
self.results[msg_id] = error.TaskAborted(msg_id) | ||||
elif content['status'] == 'resubmitted': | ||||
# TODO: handle resubmission | ||||
pass | ||||
else: | ||||
self.results[msg_id] = self._unwrap_exception(content) | ||||
|
r4872 | |||
|
r3539 | def _handle_apply_reply(self, msg): | ||
|
r3555 | """Save the reply to an apply_request into our results.""" | ||
|
r4008 | parent = msg['parent_header'] | ||
|
r3539 | msg_id = parent['msg_id'] | ||
if msg_id not in self.outstanding: | ||||
|
r3607 | if msg_id in self.history: | ||
|
r13386 | print("got stale result: %s"%msg_id) | ||
|
r13348 | print(self.results[msg_id]) | ||
print(msg) | ||||
|
r3607 | else: | ||
|
r13386 | print("got unknown result: %s"%msg_id) | ||
|
r3539 | else: | ||
self.outstanding.remove(msg_id) | ||||
content = msg['content'] | ||||
|
r4008 | header = msg['header'] | ||
|
r4872 | |||
|
r3602 | # construct metadata: | ||
|
r3651 | md = self.metadata[msg_id] | ||
|
r7957 | md.update(self._extract_metadata(msg)) | ||
|
r3651 | # is this redundant? | ||
|
r3602 | self.metadata[msg_id] = md | ||
|
r4872 | |||
|
r3651 | e_outstanding = self._outstanding_dict[md['engine_uuid']] | ||
if msg_id in e_outstanding: | ||||
e_outstanding.remove(msg_id) | ||||
|
r4872 | |||
|
r3602 | # construct result: | ||
|
r3539 | if content['status'] == 'ok': | ||
|
r8103 | self.results[msg_id] = serialize.unserialize_object(msg['buffers'])[0] | ||
|
r3540 | elif content['status'] == 'aborted': | ||
|
r3664 | self.results[msg_id] = error.TaskAborted(msg_id) | ||
|
r3540 | elif content['status'] == 'resubmitted': | ||
|
r3559 | # TODO: handle resubmission | ||
pass | ||||
|
r3539 | else: | ||
|
r3639 | self.results[msg_id] = self._unwrap_exception(content) | ||
|
r4872 | |||
|
r3539 | def _flush_notifications(self): | ||
|
r3555 | """Flush notifications of engine registrations waiting | ||
in ZMQ queue.""" | ||||
|
r4006 | idents,msg = self.session.recv(self._notification_socket, mode=zmq.NOBLOCK) | ||
|
r3539 | while msg is not None: | ||
|
r3540 | if self.debug: | ||
pprint(msg) | ||||
|
r4230 | msg_type = msg['header']['msg_type'] | ||
|
r3539 | handler = self._notification_handlers.get(msg_type, None) | ||
if handler is None: | ||||
|
r9029 | raise Exception("Unhandled message type: %s" % msg_type) | ||
|
r3539 | else: | ||
handler(msg) | ||||
|
r4006 | idents,msg = self.session.recv(self._notification_socket, mode=zmq.NOBLOCK) | ||
|
r4872 | |||
|
r3539 | def _flush_results(self, sock): | ||
|
r3555 | """Flush task or queue results waiting in ZMQ queue.""" | ||
|
r4006 | idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK) | ||
|
r3539 | while msg is not None: | ||
|
r3540 | if self.debug: | ||
pprint(msg) | ||||
|
r4230 | msg_type = msg['header']['msg_type'] | ||
|
r3539 | handler = self._queue_handlers.get(msg_type, None) | ||
if handler is None: | ||||
|
r9029 | raise Exception("Unhandled message type: %s" % msg_type) | ||
|
r3539 | else: | ||
handler(msg) | ||||
|
r4006 | idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK) | ||
|
r4872 | |||
|
r3540 | def _flush_control(self, sock): | ||
|
r3555 | """Flush replies from the control channel waiting | ||
|
r3559 | in the ZMQ queue. | ||
|
r4872 | |||
|
r3559 | Currently: ignore them.""" | ||
|
r3664 | if self._ignored_control_replies <= 0: | ||
return | ||||
|
r4006 | idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK) | ||
|
r3540 | while msg is not None: | ||
|
r3664 | self._ignored_control_replies -= 1 | ||
|
r3540 | if self.debug: | ||
pprint(msg) | ||||
|
r4006 | idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK) | ||
|
r4872 | |||
|
r3664 | def _flush_ignored_control(self): | ||
"""flush ignored control replies""" | ||||
while self._ignored_control_replies > 0: | ||||
self.session.recv(self._control_socket) | ||||
self._ignored_control_replies -= 1 | ||||
|
r4872 | |||
|
r3664 | def _flush_ignored_hub_replies(self): | ||
|
r4006 | ident,msg = self.session.recv(self._query_socket, mode=zmq.NOBLOCK) | ||
|
r3664 | while msg is not None: | ||
|
r4006 | ident,msg = self.session.recv(self._query_socket, mode=zmq.NOBLOCK) | ||
|
r4872 | |||
|
r3602 | def _flush_iopub(self, sock): | ||
"""Flush replies from the iopub channel waiting | ||||
in the ZMQ queue. | ||||
""" | ||||
|
r4006 | idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK) | ||
|
r3602 | while msg is not None: | ||
if self.debug: | ||||
pprint(msg) | ||||
parent = msg['parent_header'] | ||||
|
r6091 | # ignore IOPub messages with no parent. | ||
# Caused by print statements or warnings from before the first execution. | ||||
if not parent: | ||||
|
r10565 | idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK) | ||
|
r6091 | continue | ||
|
r3602 | msg_id = parent['msg_id'] | ||
content = msg['content'] | ||||
header = msg['header'] | ||||
|
r4230 | msg_type = msg['header']['msg_type'] | ||
|
r4872 | |||
|
r3602 | # init metadata: | ||
|
r3651 | md = self.metadata[msg_id] | ||
|
r4872 | |||
|
r3602 | if msg_type == 'stream': | ||
name = content['name'] | ||||
s = md[name] or '' | ||||
md[name] = s + content['data'] | ||||
|
r16569 | elif msg_type == 'error': | ||
md.update({'error' : self._unwrap_exception(content)}) | ||||
|
r16567 | elif msg_type == 'execute_input': | ||
md.update({'execute_input' : content['code']}) | ||||
|
r6812 | elif msg_type == 'display_data': | ||
|
r6891 | md['outputs'].append(content) | ||
|
r16568 | elif msg_type == 'execute_result': | ||
md['execute_result'] = content | ||||
|
r8103 | elif msg_type == 'data_message': | ||
data, remainder = serialize.unserialize_object(msg['buffers']) | ||||
md['data'].update(data) | ||||
|
r7494 | elif msg_type == 'status': | ||
# idle message comes after all outputs | ||||
if content['execution_state'] == 'idle': | ||||
md['outputs_ready'] = True | ||||
|
r3602 | else: | ||
|
r6812 | # unhandled msg_type (status, etc.) | ||
pass | ||||
|
r4872 | |||
|
r3651 | # reduntant? | ||
|
r3602 | self.metadata[msg_id] = md | ||
|
r4872 | |||
|
r4006 | idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK) | ||
|
r4872 | |||
|
r3555 | #-------------------------------------------------------------------------- | ||
|
r3636 | # len, getitem | ||
|
r3555 | #-------------------------------------------------------------------------- | ||
|
r4872 | |||
|
r3636 | def __len__(self): | ||
"""len(client) returns # of engines.""" | ||||
return len(self.ids) | ||||
|
r4872 | |||
|
r3539 | def __getitem__(self, key): | ||
|
r3635 | """index access returns DirectView multiplexer objects | ||
|
r4872 | |||
|
r3635 | Must be int, slice, or list/tuple/xrange of ints""" | ||
if not isinstance(key, (int, slice, tuple, list, xrange)): | ||||
raise TypeError("key by int/slice/iterable of ints only, not %s"%(type(key))) | ||||
|
r3539 | else: | ||
|
r3665 | return self.direct_view(key) | ||
|
r16200 | |||
def __iter__(self): | ||||
"""Since we define getitem, Client is iterable | ||||
but unless we also define __iter__, it won't work correctly unless engine IDs | ||||
start at zero and are continuous. | ||||
""" | ||||
for eid in self.ids: | ||||
yield self.direct_view(eid) | ||||
|
r3555 | #-------------------------------------------------------------------------- | ||
# Begin public methods | ||||
#-------------------------------------------------------------------------- | ||||
|
r4872 | |||
|
r3664 | @property | ||
def ids(self): | ||||
"""Always up-to-date ids property.""" | ||||
self._flush_notifications() | ||||
# always copy: | ||||
return list(self._ids) | ||||
|
r4872 | |||
|
r7476 | def activate(self, targets='all', suffix=''): | ||
"""Create a DirectView and register it with IPython magics | ||||
Defines the magics `%px, %autopx, %pxresult, %%px` | ||||
Parameters | ||||
---------- | ||||
targets: int, list of ints, or 'all' | ||||
The engines on which the view's magics will run | ||||
suffix: str [default: ''] | ||||
The suffix, if any, for the magics. This allows you to have | ||||
multiple views associated with parallel magics at the same time. | ||||
e.g. ``rc.activate(targets=0, suffix='0')`` will give you | ||||
the magics ``%px0``, ``%pxresult0``, etc. for running magics just | ||||
on engine 0. | ||||
""" | ||||
view = self.direct_view(targets) | ||||
view.block = True | ||||
view.activate(suffix) | ||||
return view | ||||
|
r12251 | def close(self, linger=None): | ||
"""Close my zmq Sockets | ||||
If `linger`, set the zmq LINGER socket option, | ||||
which allows discarding of messages. | ||||
""" | ||||
|
r3664 | if self._closed: | ||
return | ||||
|
r6484 | self.stop_spin_thread() | ||
|
r12251 | snames = [ trait for trait in self.trait_names() if trait.endswith("socket") ] | ||
for name in snames: | ||||
socket = getattr(self, name) | ||||
if socket is not None and not socket.closed: | ||||
if linger is not None: | ||||
socket.close(linger=linger) | ||||
else: | ||||
socket.close() | ||||
|
r3664 | self._closed = True | ||
|
r4872 | |||
|
r6484 | def _spin_every(self, interval=1): | ||
"""target func for use in spin_thread""" | ||||
while True: | ||||
if self._stop_spinning.is_set(): | ||||
return | ||||
time.sleep(interval) | ||||
self.spin() | ||||
def spin_thread(self, interval=1): | ||||
"""call Client.spin() in a background thread on some regular interval | ||||
This helps ensure that messages don't pile up too much in the zmq queue | ||||
while you are working on other things, or just leaving an idle terminal. | ||||
It also helps limit potential padding of the `received` timestamp | ||||
on AsyncResult objects, used for timings. | ||||
Parameters | ||||
---------- | ||||
interval : float, optional | ||||
The interval on which to spin the client in the background thread | ||||
(simply passed to time.sleep). | ||||
Notes | ||||
----- | ||||
For precision timing, you may want to use this method to put a bound | ||||
on the jitter (in seconds) in `received` timestamps used | ||||
in AsyncResult.wall_time. | ||||
""" | ||||
if self._spin_thread is not None: | ||||
self.stop_spin_thread() | ||||
self._stop_spinning.clear() | ||||
self._spin_thread = Thread(target=self._spin_every, args=(interval,)) | ||||
self._spin_thread.daemon = True | ||||
self._spin_thread.start() | ||||
def stop_spin_thread(self): | ||||
"""stop background spin_thread, if any""" | ||||
if self._spin_thread is not None: | ||||
self._stop_spinning.set() | ||||
self._spin_thread.join() | ||||
self._spin_thread = None | ||||
|
r3539 | def spin(self): | ||
|
r3555 | """Flush any registration notifications and execution results | ||
waiting in the ZMQ queue. | ||||
""" | ||||
if self._notification_socket: | ||||
|
r3539 | self._flush_notifications() | ||
|
r6819 | if self._iopub_socket: | ||
self._flush_iopub(self._iopub_socket) | ||||
|
r3664 | if self._mux_socket: | ||
self._flush_results(self._mux_socket) | ||||
if self._task_socket: | ||||
self._flush_results(self._task_socket) | ||||
|
r3555 | if self._control_socket: | ||
self._flush_control(self._control_socket) | ||||
|
r3664 | if self._query_socket: | ||
self._flush_ignored_hub_replies() | ||||
|
r4872 | |||
|
r3664 | def wait(self, jobs=None, timeout=-1): | ||
|
r3639 | """waits on one or more `jobs`, for up to `timeout` seconds. | ||
|
r4872 | |||
|
r3539 | Parameters | ||
---------- | ||||
|
r4872 | |||
|
r3639 | jobs : int, str, or list of ints and/or strs, or one or more AsyncResult objects | ||
|
r3555 | ints are indices to self.history | ||
strs are msg_ids | ||||
default: wait on all outstanding messages | ||||
timeout : float | ||||
a time in seconds, after which to give up. | ||||
default is -1, which means no timeout | ||||
|
r4872 | |||
|
r3555 | Returns | ||
------- | ||||
|
r4872 | |||
|
r3555 | True : when all msg_ids are done | ||
False : timeout reached, some msg_ids still outstanding | ||||
""" | ||||
tic = time.time() | ||||
|
r3639 | if jobs is None: | ||
|
r3555 | theids = self.outstanding | ||
else: | ||||
|
r13353 | if isinstance(jobs, string_types + (int, AsyncResult)): | ||
|
r3639 | jobs = [jobs] | ||
|
r3555 | theids = set() | ||
|
r3639 | for job in jobs: | ||
if isinstance(job, int): | ||||
# index access | ||||
job = self.history[job] | ||||
elif isinstance(job, AsyncResult): | ||||
|
r13383 | theids.update(job.msg_ids) | ||
|
r3590 | continue | ||
|
r3639 | theids.add(job) | ||
|
r3590 | if not theids.intersection(self.outstanding): | ||
return True | ||||
|
r3555 | self.spin() | ||
while theids.intersection(self.outstanding): | ||||
if timeout >= 0 and ( time.time()-tic ) > timeout: | ||||
break | ||||
time.sleep(1e-3) | ||||
self.spin() | ||||
return len(theids.intersection(self.outstanding)) == 0 | ||||
|
r4872 | |||
|
r3555 | #-------------------------------------------------------------------------- | ||
# Control methods | ||||
#-------------------------------------------------------------------------- | ||||
|
r4872 | |||
|
r3664 | @spin_first | ||
|
r3540 | def clear(self, targets=None, block=None): | ||
|
r3555 | """Clear the namespace in target(s).""" | ||
|
r3774 | block = self.block if block is None else block | ||
|
r3540 | targets = self._build_targets(targets)[0] | ||
for t in targets: | ||||
|
r3559 | self.session.send(self._control_socket, 'clear_request', content={}, ident=t) | ||
|
r3540 | error = False | ||
|
r3774 | if block: | ||
|
r3664 | self._flush_ignored_control() | ||
|
r3540 | for i in range(len(targets)): | ||
|
r3555 | idents,msg = self.session.recv(self._control_socket,0) | ||
|
r3540 | if self.debug: | ||
pprint(msg) | ||||
if msg['content']['status'] != 'ok': | ||||
|
r3639 | error = self._unwrap_exception(msg['content']) | ||
|
r3664 | else: | ||
self._ignored_control_replies += len(targets) | ||||
|
r3540 | if error: | ||
|
r3654 | raise error | ||
|
r4872 | |||
|
r3664 | @spin_first | ||
|
r3639 | def abort(self, jobs=None, targets=None, block=None): | ||
"""Abort specific jobs from the execution queues of target(s). | ||||
|
r4872 | |||
|
r3639 | This is a mechanism to prevent jobs that have already been submitted | ||
from executing. | ||||
|
r4872 | |||
|
r3639 | Parameters | ||
---------- | ||||
|
r4872 | |||
|
r3639 | jobs : msg_id, list of msg_ids, or AsyncResult | ||
The jobs to be aborted | ||||
|
r5645 | |||
If unspecified/None: abort all outstanding jobs. | ||||
|
r4872 | |||
|
r3639 | """ | ||
|
r3774 | block = self.block if block is None else block | ||
|
r5645 | jobs = jobs if jobs is not None else list(self.outstanding) | ||
|
r3540 | targets = self._build_targets(targets)[0] | ||
|
r5645 | |||
|
r3639 | msg_ids = [] | ||
|
r13353 | if isinstance(jobs, string_types + (AsyncResult,)): | ||
|
r3639 | jobs = [jobs] | ||
|
r13383 | bad_ids = [obj for obj in jobs if not isinstance(obj, string_types + (AsyncResult,))] | ||
|
r3639 | if bad_ids: | ||
raise TypeError("Invalid msg_id type %r, expected str or AsyncResult"%bad_ids[0]) | ||||
for j in jobs: | ||||
if isinstance(j, AsyncResult): | ||||
msg_ids.extend(j.msg_ids) | ||||
else: | ||||
msg_ids.append(j) | ||||
|
r3540 | content = dict(msg_ids=msg_ids) | ||
for t in targets: | ||||
|
r4872 | self.session.send(self._control_socket, 'abort_request', | ||
|
r3540 | content=content, ident=t) | ||
error = False | ||||
|
r3774 | if block: | ||
|
r3664 | self._flush_ignored_control() | ||
|
r3540 | for i in range(len(targets)): | ||
|
r3555 | idents,msg = self.session.recv(self._control_socket,0) | ||
|
r3540 | if self.debug: | ||
pprint(msg) | ||||
if msg['content']['status'] != 'ok': | ||||
|
r3639 | error = self._unwrap_exception(msg['content']) | ||
|
r3664 | else: | ||
self._ignored_control_replies += len(targets) | ||||
|
r3540 | if error: | ||
|
r3654 | raise error | ||
|
r4872 | |||
|
r3664 | @spin_first | ||
|
r7481 | def shutdown(self, targets='all', restart=False, hub=False, block=None): | ||
"""Terminates one or more engine processes, optionally including the hub. | ||||
Parameters | ||||
---------- | ||||
targets: list of ints or 'all' [default: all] | ||||
Which engines to shutdown. | ||||
hub: bool [default: False] | ||||
Whether to include the Hub. hub=True implies targets='all'. | ||||
block: bool [default: self.block] | ||||
Whether to wait for clean shutdown replies or not. | ||||
restart: bool [default: False] | ||||
NOT IMPLEMENTED | ||||
whether to restart engines after shutting them down. | ||||
""" | ||||
|
r9026 | from IPython.parallel.error import NoEnginesRegistered | ||
|
r7481 | if restart: | ||
raise NotImplementedError("Engine restart is not yet implemented") | ||||
|
r3774 | block = self.block if block is None else block | ||
|
r3664 | if hub: | ||
|
r3580 | targets = 'all' | ||
|
r9026 | try: | ||
targets = self._build_targets(targets)[0] | ||||
except NoEnginesRegistered: | ||||
targets = [] | ||||
|
r3540 | for t in targets: | ||
|
r4872 | self.session.send(self._control_socket, 'shutdown_request', | ||
|
r3575 | content={'restart':restart},ident=t) | ||
|
r3540 | error = False | ||
|
r3664 | if block or hub: | ||
self._flush_ignored_control() | ||||
|
r3540 | for i in range(len(targets)): | ||
|
r3664 | idents,msg = self.session.recv(self._control_socket, 0) | ||
|
r3540 | if self.debug: | ||
pprint(msg) | ||||
if msg['content']['status'] != 'ok': | ||||
|
r3639 | error = self._unwrap_exception(msg['content']) | ||
|
r3664 | else: | ||
self._ignored_control_replies += len(targets) | ||||
|
r4872 | |||
|
r3664 | if hub: | ||
|
r3580 | time.sleep(0.25) | ||
self.session.send(self._query_socket, 'shutdown_request') | ||||
idents,msg = self.session.recv(self._query_socket, 0) | ||||
if self.debug: | ||||
pprint(msg) | ||||
if msg['content']['status'] != 'ok': | ||||
|
r3639 | error = self._unwrap_exception(msg['content']) | ||
|
r4872 | |||
|
r3540 | if error: | ||
|
r3584 | raise error | ||
|
r4872 | |||
|
r3555 | #-------------------------------------------------------------------------- | ||
|
r3774 | # Execution related methods | ||
|
r3555 | #-------------------------------------------------------------------------- | ||
|
r4872 | |||
|
r3583 | def _maybe_raise(self, result): | ||
"""wrapper for maybe raising an exception if apply failed.""" | ||||
if isinstance(result, error.RemoteError): | ||||
raise result | ||||
|
r4872 | |||
|
r3583 | return result | ||
|
r4872 | |||
|
r7957 | def send_apply_request(self, socket, f, args=None, kwargs=None, metadata=None, track=False, | ||
|
r3664 | ident=None): | ||
"""construct and send an apply message via a socket. | ||||
|
r4872 | |||
|
r3664 | This is the principal method with which all engine execution is performed by views. | ||
|
r3555 | """ | ||
|
r4872 | |||
|
r6868 | if self._closed: | ||
raise RuntimeError("Client cannot be used after its sockets have been closed") | ||||
|
r3555 | # defaults: | ||
args = args if args is not None else [] | ||||
kwargs = kwargs if kwargs is not None else {} | ||||
|
r7957 | metadata = metadata if metadata is not None else {} | ||
|
r4872 | |||
|
r3664 | # validate arguments | ||
|
r5821 | if not callable(f) and not isinstance(f, Reference): | ||
|
r3555 | raise TypeError("f must be callable, not %s"%type(f)) | ||
if not isinstance(args, (tuple, list)): | ||||
raise TypeError("args must be tuple or list, not %s"%type(args)) | ||||
if not isinstance(kwargs, dict): | ||||
raise TypeError("kwargs must be dict, not %s"%type(kwargs)) | ||||
|
r7957 | if not isinstance(metadata, dict): | ||
raise TypeError("metadata must be dict, not %s"%type(metadata)) | ||||
|
r4872 | |||
|
r8103 | bufs = serialize.pack_apply_message(f, args, kwargs, | ||
|
r8033 | buffer_threshold=self.session.buffer_threshold, | ||
item_threshold=self.session.item_threshold, | ||||
) | ||||
|
r4872 | |||
|
r3664 | msg = self.session.send(socket, "apply_request", buffers=bufs, ident=ident, | ||
|
r7957 | metadata=metadata, track=track) | ||
|
r4872 | |||
|
r4230 | msg_id = msg['header']['msg_id'] | ||
|
r3539 | self.outstanding.add(msg_id) | ||
|
r3664 | if ident: | ||
# possibly routed to a specific engine | ||||
if isinstance(ident, list): | ||||
|
r6793 | ident = ident[-1] | ||
if ident in self._engines.values(): | ||||
# save for later, in case of engine death | ||||
self._outstanding_dict[ident].add(msg_id) | ||||
self.history.append(msg_id) | ||||
self.metadata[msg_id]['submitted'] = datetime.now() | ||||
return msg | ||||
|
r7957 | def send_execute_request(self, socket, code, silent=True, metadata=None, ident=None): | ||
|
r6793 | """construct and send an execute request via a socket. | ||
""" | ||||
|
r6868 | if self._closed: | ||
raise RuntimeError("Client cannot be used after its sockets have been closed") | ||||
|
r6793 | # defaults: | ||
|
r7957 | metadata = metadata if metadata is not None else {} | ||
|
r6793 | |||
# validate arguments | ||||
|
r13353 | if not isinstance(code, string_types): | ||
|
r6793 | raise TypeError("code must be text, not %s" % type(code)) | ||
|
r7957 | if not isinstance(metadata, dict): | ||
raise TypeError("metadata must be dict, not %s" % type(metadata)) | ||||
|
r6793 | |||
|
r16570 | content = dict(code=code, silent=bool(silent), user_expressions={}) | ||
|
r6793 | |||
msg = self.session.send(socket, "execute_request", content=content, ident=ident, | ||||
|
r7957 | metadata=metadata) | ||
|
r6793 | |||
msg_id = msg['header']['msg_id'] | ||||
self.outstanding.add(msg_id) | ||||
if ident: | ||||
# possibly routed to a specific engine | ||||
if isinstance(ident, list): | ||||
|
r3664 | ident = ident[-1] | ||
if ident in self._engines.values(): | ||||
# save for later, in case of engine death | ||||
self._outstanding_dict[ident].add(msg_id) | ||||
|
r3539 | self.history.append(msg_id) | ||
|
r3651 | self.metadata[msg_id]['submitted'] = datetime.now() | ||
|
r4872 | |||
|
r3664 | return msg | ||
|
r3589 | #-------------------------------------------------------------------------- | ||
|
r3636 | # construct a View object | ||
|
r3589 | #-------------------------------------------------------------------------- | ||
|
r4872 | |||
|
r3664 | def load_balanced_view(self, targets=None): | ||
"""construct a DirectView object. | ||||
|
r4872 | |||
|
r3664 | If no arguments are specified, create a LoadBalancedView | ||
using all engines. | ||||
|
r4872 | |||
|
r3664 | Parameters | ||
---------- | ||||
|
r4872 | |||
|
r3664 | targets: list,slice,int,etc. [default: use all engines] | ||
The subset of engines across which to load-balance | ||||
""" | ||||
|
r4492 | if targets == 'all': | ||
targets = None | ||||
|
r3666 | if targets is not None: | ||
|
r3665 | targets = self._build_targets(targets)[1] | ||
return LoadBalancedView(client=self, socket=self._task_socket, targets=targets) | ||||
|
r4872 | |||
|
r3664 | def direct_view(self, targets='all'): | ||
"""construct a DirectView object. | ||||
|
r4872 | |||
|
r5290 | If no targets are specified, create a DirectView using all engines. | ||
rc.direct_view('all') is distinguished from rc[:] in that 'all' will | ||||
evaluate the target engines at each execution, whereas rc[:] will connect to | ||||
all *current* engines, and that list will not change. | ||||
That is, 'all' will always use all engines, whereas rc[:] will not use | ||||
engines added after the DirectView is constructed. | ||||
|
r4872 | |||
|
r3664 | Parameters | ||
---------- | ||||
|
r4872 | |||
|
r3664 | targets: list,slice,int,etc. [default: use all engines] | ||
The engines to use for the View | ||||
""" | ||||
|
r3665 | single = isinstance(targets, int) | ||
|
r4492 | # allow 'all' to be lazily evaluated at each execution | ||
if targets != 'all': | ||||
targets = self._build_targets(targets)[1] | ||||
|
r3665 | if single: | ||
targets = targets[0] | ||||
return DirectView(client=self, socket=self._mux_socket, targets=targets) | ||||
|
r4872 | |||
|
r3555 | #-------------------------------------------------------------------------- | ||
# Query methods | ||||
#-------------------------------------------------------------------------- | ||||
|
r4872 | |||
|
r3664 | @spin_first | ||
|
r16739 | def get_result(self, indices_or_msg_ids=None, block=None, owner=True): | ||
|
r3639 | """Retrieve a result by msg_id or history index, wrapped in an AsyncResult object. | ||
|
r4872 | |||
|
r3639 | If the client already has the results, no request to the Hub will be made. | ||
|
r4872 | |||
|
r3639 | This is a convenient way to construct AsyncResult objects, which are wrappers | ||
that include metadata about execution, and allow for awaiting results that | ||||
were not submitted by this Client. | ||||
|
r4872 | |||
|
r3639 | It can also be a convenient way to retrieve the metadata associated with | ||
blocking execution, since it always retrieves | ||||
|
r4872 | |||
|
r3639 | Examples | ||
-------- | ||||
:: | ||||
|
r4872 | |||
|
r3639 | In [10]: r = client.apply() | ||
|
r4872 | |||
|
r3539 | Parameters | ||
---------- | ||||
|
r4872 | |||
|
r3639 | indices_or_msg_ids : integer history index, str msg_id, or list of either | ||
The indices or msg_ids of indices to be retrieved | ||||
|
r4872 | |||
|
r3639 | block : bool | ||
Whether to wait for the result to be done | ||||
|
r16739 | owner : bool [default: True] | ||
Whether this AsyncResult should own the result. | ||||
If so, calling `ar.get()` will remove data from the | ||||
client's result and metadata cache. | ||||
There should only be one owner of any given msg_id. | ||||
|
r4872 | |||
|
r3639 | Returns | ||
------- | ||||
|
r4872 | |||
|
r3639 | AsyncResult | ||
A single AsyncResult object will always be returned. | ||||
|
r4872 | |||
|
r3639 | AsyncHubResult | ||
A subclass of AsyncResult that retrieves results from the Hub | ||||
|
r4872 | |||
|
r3639 | """ | ||
|
r3774 | block = self.block if block is None else block | ||
|
r3639 | if indices_or_msg_ids is None: | ||
indices_or_msg_ids = -1 | ||||
|
r11416 | |||
single_result = False | ||||
|
r3639 | if not isinstance(indices_or_msg_ids, (list,tuple)): | ||
indices_or_msg_ids = [indices_or_msg_ids] | ||||
|
r11416 | single_result = True | ||
|
r4872 | |||
|
r3639 | theids = [] | ||
for id in indices_or_msg_ids: | ||||
if isinstance(id, int): | ||||
id = self.history[id] | ||||
|
r13353 | if not isinstance(id, string_types): | ||
|
r3639 | raise TypeError("indices must be str or int, not %r"%id) | ||
theids.append(id) | ||||
|
r4872 | |||
|
r13383 | local_ids = [msg_id for msg_id in theids if (msg_id in self.outstanding or msg_id in self.results)] | ||
remote_ids = [msg_id for msg_id in theids if msg_id not in local_ids] | ||||
|
r11416 | |||
# given single msg_id initially, get_result shot get the result itself, | ||||
# not a length-one list | ||||
if single_result: | ||||
theids = theids[0] | ||||
|
r4872 | |||
|
r3639 | if remote_ids: | ||
|
r16739 | ar = AsyncHubResult(self, msg_ids=theids, owner=owner) | ||
|
r3639 | else: | ||
|
r16739 | ar = AsyncResult(self, msg_ids=theids, owner=owner) | ||
|
r4872 | |||
|
r3639 | if block: | ||
ar.wait() | ||||
|
r4872 | |||
|
r3639 | return ar | ||
|
r3874 | |||
@spin_first | ||||
|
r7957 | def resubmit(self, indices_or_msg_ids=None, metadata=None, block=None): | ||
|
r3874 | """Resubmit one or more tasks. | ||
in-flight tasks may not be resubmitted. | ||||
Parameters | ||||
---------- | ||||
indices_or_msg_ids : integer history index, str msg_id, or list of either | ||||
The indices or msg_ids of indices to be retrieved | ||||
block : bool | ||||
Whether to wait for the result to be done | ||||
Returns | ||||
------- | ||||
AsyncHubResult | ||||
A subclass of AsyncResult that retrieves results from the Hub | ||||
""" | ||||
block = self.block if block is None else block | ||||
if indices_or_msg_ids is None: | ||||
indices_or_msg_ids = -1 | ||||
if not isinstance(indices_or_msg_ids, (list,tuple)): | ||||
indices_or_msg_ids = [indices_or_msg_ids] | ||||
theids = [] | ||||
for id in indices_or_msg_ids: | ||||
if isinstance(id, int): | ||||
id = self.history[id] | ||||
|
r13353 | if not isinstance(id, string_types): | ||
|
r3874 | raise TypeError("indices must be str or int, not %r"%id) | ||
theids.append(id) | ||||
content = dict(msg_ids = theids) | ||||
self.session.send(self._query_socket, 'resubmit_request', content) | ||||
zmq.select([self._query_socket], [], []) | ||||
idents,msg = self.session.recv(self._query_socket, zmq.NOBLOCK) | ||||
if self.debug: | ||||
pprint(msg) | ||||
content = msg['content'] | ||||
if content['status'] != 'ok': | ||||
raise self._unwrap_exception(content) | ||||
|
r6817 | mapping = content['resubmitted'] | ||
new_ids = [ mapping[msg_id] for msg_id in theids ] | ||||
|
r3874 | |||
|
r6817 | ar = AsyncHubResult(self, msg_ids=new_ids) | ||
|
r3874 | |||
if block: | ||||
ar.wait() | ||||
return ar | ||||
|
r4872 | |||
|
r3664 | @spin_first | ||
|
r3639 | def result_status(self, msg_ids, status_only=True): | ||
"""Check on the status of the result(s) of the apply request with `msg_ids`. | ||||
|
r4872 | |||
|
r3639 | If status_only is False, then the actual results will be retrieved, else | ||
only the status of the results will be checked. | ||||
|
r4872 | |||
|
r3639 | Parameters | ||
---------- | ||||
|
r4872 | |||
|
r3639 | msg_ids : list of msg_ids | ||
|
r3555 | if int: | ||
Passed as index to self.history for convenience. | ||||
|
r3639 | status_only : bool (default: True) | ||
|
r3555 | if False: | ||
|
r3639 | Retrieve the actual results of completed tasks. | ||
|
r4872 | |||
|
r3598 | Returns | ||
------- | ||||
|
r4872 | |||
|
r3598 | results : dict | ||
There will always be the keys 'pending' and 'completed', which will | ||||
|
r3639 | be lists of msg_ids that are incomplete or complete. If `status_only` | ||
is False, then completed results will be keyed by their `msg_id`. | ||||
|
r3539 | """ | ||
|
r3641 | if not isinstance(msg_ids, (list,tuple)): | ||
|
r3651 | msg_ids = [msg_ids] | ||
|
r4872 | |||
|
r3539 | theids = [] | ||
|
r3641 | for msg_id in msg_ids: | ||
|
r3539 | if isinstance(msg_id, int): | ||
msg_id = self.history[msg_id] | ||||
|
r13353 | if not isinstance(msg_id, string_types): | ||
|
r3555 | raise TypeError("msg_ids must be str, not %r"%msg_id) | ||
|
r3539 | theids.append(msg_id) | ||
|
r4872 | |||
|
r3555 | completed = [] | ||
local_results = {} | ||||
|
r4872 | |||
|
r3622 | # comment this block out to temporarily disable local shortcut: | ||
|
r3639 | for msg_id in theids: | ||
|
r3622 | if msg_id in self.results: | ||
completed.append(msg_id) | ||||
local_results[msg_id] = self.results[msg_id] | ||||
theids.remove(msg_id) | ||||
|
r4872 | |||
|
r3559 | if theids: # some not locally cached | ||
|
r3555 | content = dict(msg_ids=theids, status_only=status_only) | ||
msg = self.session.send(self._query_socket, "result_request", content=content) | ||||
zmq.select([self._query_socket], [], []) | ||||
idents,msg = self.session.recv(self._query_socket, zmq.NOBLOCK) | ||||
if self.debug: | ||||
pprint(msg) | ||||
content = msg['content'] | ||||
if content['status'] != 'ok': | ||||
|
r3639 | raise self._unwrap_exception(content) | ||
|
r3598 | buffers = msg['buffers'] | ||
|
r3555 | else: | ||
content = dict(completed=[],pending=[]) | ||||
|
r4872 | |||
|
r3598 | content['completed'].extend(completed) | ||
|
r4872 | |||
|
r3598 | if status_only: | ||
return content | ||||
|
r4872 | |||
|
r3598 | failures = [] | ||
# load cached results into result: | ||||
content.update(local_results) | ||||
|
r4872 | |||
|
r3598 | # update cache with results: | ||
for msg_id in sorted(theids): | ||||
if msg_id in content['completed']: | ||||
rec = content[msg_id] | ||||
|
r13512 | parent = extract_dates(rec['header']) | ||
header = extract_dates(rec['result_header']) | ||||
|
r3598 | rcontent = rec['result_content'] | ||
|
r3602 | iodict = rec['io'] | ||
|
r3598 | if isinstance(rcontent, str): | ||
rcontent = self.session.unpack(rcontent) | ||||
|
r4872 | |||
|
r3651 | md = self.metadata[msg_id] | ||
|
r7957 | md_msg = dict( | ||
content=rcontent, | ||||
parent_header=parent, | ||||
header=header, | ||||
metadata=rec['result_metadata'], | ||||
) | ||||
md.update(self._extract_metadata(md_msg)) | ||||
|
r6469 | if rec.get('received'): | ||
|
r13520 | md['received'] = parse_date(rec['received']) | ||
|
r3602 | md.update(iodict) | ||
|
r7508 | |||
|
r3598 | if rcontent['status'] == 'ok': | ||
|
r7508 | if header['msg_type'] == 'apply_reply': | ||
|
r8103 | res,buffers = serialize.unserialize_object(buffers) | ||
|
r7508 | elif header['msg_type'] == 'execute_reply': | ||
res = ExecuteReply(msg_id, rcontent, md) | ||||
else: | ||||
|
r9027 | raise KeyError("unhandled msg type: %r" % header['msg_type']) | ||
|
r3598 | else: | ||
|
r3639 | res = self._unwrap_exception(rcontent) | ||
|
r3598 | failures.append(res) | ||
|
r4872 | |||
|
r3598 | self.results[msg_id] = res | ||
content[msg_id] = res | ||||
|
r4872 | |||
|
r3639 | if len(theids) == 1 and failures: | ||
|
r7508 | raise failures[0] | ||
|
r4872 | |||
|
r3639 | error.collect_exceptions(failures, "result_status") | ||
|
r3555 | return content | ||
|
r3664 | @spin_first | ||
|
r3635 | def queue_status(self, targets='all', verbose=False): | ||
|
r3555 | """Fetch the status of engine queues. | ||
|
r4872 | |||
|
r3555 | Parameters | ||
---------- | ||||
|
r4872 | |||
|
r3555 | targets : int/str/list of ints/strs | ||
|
r3635 | the engines whose states are to be queried. | ||
|
r3555 | default : all | ||
verbose : bool | ||||
|
r3584 | Whether to return lengths only, or lists of ids for each element | ||
|
r3555 | """ | ||
|
r6093 | if targets == 'all': | ||
# allow 'all' to be evaluated on the engine | ||||
engine_ids = None | ||||
else: | ||||
engine_ids = self._build_targets(targets)[1] | ||||
|
r3664 | content = dict(targets=engine_ids, verbose=verbose) | ||
|
r3555 | self.session.send(self._query_socket, "queue_request", content=content) | ||
idents,msg = self.session.recv(self._query_socket, 0) | ||||
|
r3540 | if self.debug: | ||
pprint(msg) | ||||
|
r3555 | content = msg['content'] | ||
status = content.pop('status') | ||||
if status != 'ok': | ||||
|
r3639 | raise self._unwrap_exception(content) | ||
|
r4036 | content = rekey(content) | ||
|
r3664 | if isinstance(targets, int): | ||
return content[targets] | ||||
else: | ||||
return content | ||||
|
r4872 | |||
|
r8409 | def _build_msgids_from_target(self, targets=None): | ||
"""Build a list of msg_ids from the list of engine targets""" | ||||
|
r8415 | if not targets: # needed as _build_targets otherwise uses all engines | ||
|
r8409 | return [] | ||
target_ids = self._build_targets(targets)[0] | ||||
|
r13383 | return [md_id for md_id in self.metadata if self.metadata[md_id]["engine_uuid"] in target_ids] | ||
|
r8409 | |||
def _build_msgids_from_jobs(self, jobs=None): | ||||
"""Build a list of msg_ids from "jobs" """ | ||||
|
r8415 | if not jobs: | ||
return [] | ||||
|
r8409 | msg_ids = [] | ||
|
r13353 | if isinstance(jobs, string_types + (AsyncResult,)): | ||
|
r8409 | jobs = [jobs] | ||
|
r13383 | bad_ids = [obj for obj in jobs if not isinstance(obj, string_types + (AsyncResult,))] | ||
|
r8409 | if bad_ids: | ||
raise TypeError("Invalid msg_id type %r, expected str or AsyncResult"%bad_ids[0]) | ||||
for j in jobs: | ||||
if isinstance(j, AsyncResult): | ||||
msg_ids.extend(j.msg_ids) | ||||
else: | ||||
msg_ids.append(j) | ||||
return msg_ids | ||||
def purge_local_results(self, jobs=[], targets=[]): | ||||
|
r13917 | """Clears the client caches of results and their metadata. | ||
|
r8409 | Individual results can be purged by msg_id, or the entire | ||
history of specific targets can be purged. | ||||
|
r13917 | Use `purge_local_results('all')` to scrub everything from the Clients's | ||
results and metadata caches. | ||||
|
r8409 | |||
After this call all `AsyncResults` are invalid and should be discarded. | ||||
If you must "reget" the results, you can still do so by using | ||||
`client.get_result(msg_id)` or `client.get_result(asyncresult)`. This will | ||||
redownload the results from the hub if they are still available | ||||
(i.e `client.purge_hub_results(...)` has not been called. | ||||
Parameters | ||||
---------- | ||||
jobs : str or list of str or AsyncResult objects | ||||
the msg_ids whose results should be purged. | ||||
|
r13917 | targets : int/list of ints | ||
The engines, by integer ID, whose entire result histories are to be purged. | ||||
Raises | ||||
------ | ||||
RuntimeError : if any of the tasks to be purged are still outstanding. | ||||
|
r8409 | |||
""" | ||||
if not targets and not jobs: | ||||
raise ValueError("Must specify at least one of `targets` and `jobs`") | ||||
|
r13917 | |||
|
r8409 | if jobs == 'all': | ||
|
r13917 | if self.outstanding: | ||
raise RuntimeError("Can't purge outstanding tasks: %s" % self.outstanding) | ||||
|
r8409 | self.results.clear() | ||
self.metadata.clear() | ||||
else: | ||||
|
r13917 | msg_ids = set() | ||
msg_ids.update(self._build_msgids_from_target(targets)) | ||||
msg_ids.update(self._build_msgids_from_jobs(jobs)) | ||||
still_outstanding = self.outstanding.intersection(msg_ids) | ||||
if still_outstanding: | ||||
raise RuntimeError("Can't purge outstanding tasks: %s" % still_outstanding) | ||||
|
r13383 | for mid in msg_ids: | ||
|
r16739 | self.results.pop(mid, None) | ||
self.metadata.pop(mid, None) | ||||
|
r8409 | |||
|
r3664 | @spin_first | ||
|
r8409 | def purge_hub_results(self, jobs=[], targets=[]): | ||
|
r3664 | """Tell the Hub to forget results. | ||
|
r4872 | |||
|
r3555 | Individual results can be purged by msg_id, or the entire | ||
|
r3584 | history of specific targets can be purged. | ||
|
r4872 | |||
|
r4146 | Use `purge_results('all')` to scrub everything from the Hub's db. | ||
|
r4872 | |||
|
r3555 | Parameters | ||
---------- | ||||
|
r4872 | |||
|
r3664 | jobs : str or list of str or AsyncResult objects | ||
|
r3584 | the msg_ids whose results should be forgotten. | ||
|
r3555 | targets : int/str/list of ints/strs | ||
|
r4146 | The targets, by int_id, whose entire history is to be purged. | ||
|
r4872 | |||
|
r3555 | default : None | ||
""" | ||||
|
r3639 | if not targets and not jobs: | ||
raise ValueError("Must specify at least one of `targets` and `jobs`") | ||||
|
r3555 | if targets: | ||
targets = self._build_targets(targets)[1] | ||||
|
r4872 | |||
|
r3639 | # construct msg_ids from jobs | ||
|
r4146 | if jobs == 'all': | ||
msg_ids = jobs | ||||
else: | ||||
|
r8409 | msg_ids = self._build_msgids_from_jobs(jobs) | ||
|
r4146 | |||
content = dict(engine_ids=targets, msg_ids=msg_ids) | ||||
|
r3555 | self.session.send(self._query_socket, "purge_request", content=content) | ||
idents, msg = self.session.recv(self._query_socket, 0) | ||||
if self.debug: | ||||
pprint(msg) | ||||
content = msg['content'] | ||||
if content['status'] != 'ok': | ||||
|
r3639 | raise self._unwrap_exception(content) | ||
|
r3548 | |||
|
r8409 | def purge_results(self, jobs=[], targets=[]): | ||
"""Clears the cached results from both the hub and the local client | ||||
Individual results can be purged by msg_id, or the entire | ||||
history of specific targets can be purged. | ||||
Use `purge_results('all')` to scrub every cached result from both the Hub's and | ||||
the Client's db. | ||||
Equivalent to calling both `purge_hub_results()` and `purge_client_results()` with | ||||
the same arguments. | ||||
Parameters | ||||
---------- | ||||
jobs : str or list of str or AsyncResult objects | ||||
the msg_ids whose results should be forgotten. | ||||
targets : int/str/list of ints/strs | ||||
The targets, by int_id, whose entire history is to be purged. | ||||
default : None | ||||
""" | ||||
self.purge_local_results(jobs=jobs, targets=targets) | ||||
self.purge_hub_results(jobs=jobs, targets=targets) | ||||
def purge_everything(self): | ||||
"""Clears all content from previous Tasks from both the hub and the local client | ||||
In addition to calling `purge_results("all")` it also deletes the history and | ||||
other bookkeeping lists. | ||||
""" | ||||
self.purge_results("all") | ||||
self.history = [] | ||||
self.session.digest_history.clear() | ||||
|
r3780 | @spin_first | ||
def hub_history(self): | ||||
"""Get the Hub's history | ||||
|
r4872 | |||
|
r3780 | Just like the Client, the Hub has a history, which is a list of msg_ids. | ||
This will contain the history of all clients, and, depending on configuration, | ||||
may contain history across multiple cluster sessions. | ||||
|
r4872 | |||
|
r3780 | Any msg_id returned here is a valid argument to `get_result`. | ||
|
r4872 | |||
|
r3780 | Returns | ||
------- | ||||
|
r4872 | |||
|
r3780 | msg_ids : list of strs | ||
list of all msg_ids, ordered by task submission time. | ||||
""" | ||||
|
r4872 | |||
|
r3780 | self.session.send(self._query_socket, "history_request", content={}) | ||
idents, msg = self.session.recv(self._query_socket, 0) | ||||
|
r4872 | |||
|
r3780 | if self.debug: | ||
pprint(msg) | ||||
content = msg['content'] | ||||
if content['status'] != 'ok': | ||||
raise self._unwrap_exception(content) | ||||
else: | ||||
return content['history'] | ||||
@spin_first | ||||
def db_query(self, query, keys=None): | ||||
"""Query the Hub's TaskRecord database | ||||
|
r4872 | |||
|
r3780 | This will return a list of task record dicts that match `query` | ||
|
r4872 | |||
|
r3780 | Parameters | ||
---------- | ||||
|
r4872 | |||
|
r3780 | query : mongodb query dict | ||
The search dict. See mongodb query docs for details. | ||||
keys : list of strs [optional] | ||||
|
r3876 | The subset of keys to be returned. The default is to fetch everything but buffers. | ||
|
r3780 | 'msg_id' will *always* be included. | ||
""" | ||||
|
r13353 | if isinstance(keys, string_types): | ||
|
r3876 | keys = [keys] | ||
|
r3780 | content = dict(query=query, keys=keys) | ||
self.session.send(self._query_socket, "db_request", content=content) | ||||
idents, msg = self.session.recv(self._query_socket, 0) | ||||
if self.debug: | ||||
pprint(msg) | ||||
content = msg['content'] | ||||
if content['status'] != 'ok': | ||||
raise self._unwrap_exception(content) | ||||
|
r4872 | |||
|
r3780 | records = content['records'] | ||
|
r4872 | |||
|
r3780 | buffer_lens = content['buffer_lens'] | ||
result_buffer_lens = content['result_buffer_lens'] | ||||
buffers = msg['buffers'] | ||||
has_bufs = buffer_lens is not None | ||||
has_rbufs = result_buffer_lens is not None | ||||
for i,rec in enumerate(records): | ||||
|
r13512 | # unpack datetime objects | ||
|
r13520 | for hkey in ('header', 'result_header'): | ||
if hkey in rec: | ||||
rec[hkey] = extract_dates(rec[hkey]) | ||||
for dtkey in ('submitted', 'started', 'completed', 'received'): | ||||
|
r13512 | if dtkey in rec: | ||
|
r13520 | rec[dtkey] = parse_date(rec[dtkey]) | ||
|
r3780 | # relink buffers | ||
if has_bufs: | ||||
blen = buffer_lens[i] | ||||
rec['buffers'], buffers = buffers[:blen],buffers[blen:] | ||||
if has_rbufs: | ||||
blen = result_buffer_lens[i] | ||||
rec['result_buffers'], buffers = buffers[:blen],buffers[blen:] | ||||
|
r4872 | |||
|
r3780 | return records | ||
|
r3587 | |||
|
r3666 | __all__ = [ 'Client' ] | ||