Fix %autopx magic #349

Closed
wants to merge 4 commits into
from
@@ -443,12 +443,6 @@ def init_instance_attrs(self):
# ipython names that may develop later.
self.meta = Struct()
- # Object variable to store code object waiting execution. This is
- # used mainly by the multithreaded shells, but it can come in handy in
- # other situations. No need to use a Queue here, since it's a single
- # item which gets cleared once run.
- self.code_to_run = None
-
# Temporary files used for various purposes. Deleted at exit.
self.tempfiles = []
@@ -2210,16 +2204,20 @@ def run_ast_nodes(self, nodelist, cell_name, interactivity='last'):
raise ValueError("Interactivity was %r" % interactivity)
exec_count = self.execution_count
- if to_run_exec:
- mod = ast.Module(to_run_exec)
- self.code_to_run = code = self.compile(mod, cell_name, "exec")
- if self.run_code(code) == 1:
- return
-
- if to_run_interactive:
- mod = ast.Interactive(to_run_interactive)
- self.code_to_run = code = self.compile(mod, cell_name, "single")
- return self.run_code(code)
+
+ for i, node in enumerate(to_run_exec):
+ mod = ast.Module([node])
+ code = self.compile(mod, cell_name, "exec")
+ if self.run_code(code):
+ return True
+
+ for i, node in enumerate(to_run_interactive):
+ mod = ast.Interactive([node])
+ code = self.compile(mod, cell_name, "single")
+ if self.run_code(code):
+ return True
+
+ return False
# PENDING REMOVAL: this method is slated for deletion, once our new
@@ -2316,13 +2314,8 @@ def run_source(self, source, filename=None, symbol='single'):
return True
# Case 3
- # We store the code object so that threaded shells and
- # custom exception handlers can access all this info if needed.
- # The source corresponding to this can be obtained from the
- # buffer attribute as '\n'.join(self.buffer).
- self.code_to_run = code
# now actually execute the code object
- if self.run_code(code) == 0:
+ if not self.run_code(code):
return False
else:
return None
@@ -2336,11 +2329,17 @@ def run_code(self, code_obj):
When an exception occurs, self.showtraceback() is called to display a
traceback.
- Return value: a flag indicating whether the code to be run completed
- successfully:
+ Parameters
+ ----------
+ code_obj : code object
+ A compiled code object, to be executed
+ post_execute : bool [default: True]
+ whether to call post_execute hooks after this particular execution.
- - 0: successful execution.
- - 1: an error occurred.
+ Returns
+ -------
+ False : successful execution.
+ True : an error occurred.
"""
# Set our own excepthook in case the user code tries to call it
@@ -2373,8 +2372,6 @@ def run_code(self, code_obj):
if softspace(sys.stdout, 0):
print
- # Flush out code object which has been run (and source)
- self.code_to_run = None
return outflag
# For backwards compatibility
@@ -15,12 +15,10 @@
#-----------------------------------------------------------------------------
import ast
-import new
import re
from IPython.core.plugin import Plugin
from IPython.utils.traitlets import Bool, Any, Instance
-from IPython.utils.autoattr import auto_attr
from IPython.testing import decorators as testdec
#-----------------------------------------------------------------------------
@@ -36,7 +34,7 @@
class ParalleMagic(Plugin):
"""A component to manage the %result, %px and %autopx magics."""
- active_view = Any()
+ active_view = Instance('IPython.parallel.client.view.DirectView')
verbose = Bool(False, config=True)
shell = Instance('IPython.core.interactiveshell.InteractiveShellABC')
@@ -105,8 +103,10 @@ def magic_px(self, ipself, parameter_s=''):
print NO_ACTIVE_VIEW
return
print "Parallel execution on engines: %s" % self.active_view.targets
- result = self.active_view.execute(parameter_s)
- return result
+ result = self.active_view.execute(parameter_s, block=False)
+ if self.active_view.block:
+ result.get()
+ self._maybe_display_output(result)
@testdec.skip_doctest
def magic_autopx(self, ipself, parameter_s=''):
@@ -125,9 +125,13 @@ def magic_autopx(self, ipself, parameter_s=''):
%autopx to enabled
In [26]: a = 10
- <Results List>
- [0] In [8]: a = 10
- [1] In [8]: a = 10
+ Parallel execution on engines: [0,1,2,3]
+ In [27]: print a
+ Parallel execution on engines: [0,1,2,3]
+ [stdout:0] 10
+ [stdout:1] 10
+ [stdout:2] 10
+ [stdout:3] 10
In [27]: %autopx
@@ -146,10 +150,12 @@ def _enable_autopx(self):
print NO_ACTIVE_VIEW
return
+ # override run_cell and run_code
self._original_run_cell = self.shell.run_cell
- self.shell.run_cell = new.instancemethod(
- self.pxrun_cell, self.shell, self.shell.__class__
- )
+ self.shell.run_cell = self.pxrun_cell
+ self._original_run_code = self.shell.run_code
+ self.shell.run_code = self.pxrun_code
+
self.autopx = True
print "%autopx enabled"
@@ -158,17 +164,42 @@ def _disable_autopx(self):
"""
if self.autopx:
self.shell.run_cell = self._original_run_cell
+ self.shell.run_code = self._original_run_code
self.autopx = False
print "%autopx disabled"
- def pxrun_cell(self, ipself, cell, store_history=True):
+ def _maybe_display_output(self, result):
+ """Maybe display the output of a parallel result.
+
+ If self.active_view.block is True, wait for the result
+ and display the result. Otherwise, this is a noop.
+ """
+ targets = self.active_view.targets
+ if isinstance(targets, int):
+ targets = [targets]
+ if targets == 'all':
+ targets = self.active_view.client.ids
+ stdout = [s.rstrip() for s in result.stdout]
+ if any(stdout):
+ for i,eid in enumerate(targets):
+ print '[stdout:%i]'%eid, stdout[i]
+
+
+ def pxrun_cell(self, raw_cell, store_history=True):
"""drop-in replacement for InteractiveShell.run_cell.
This executes code remotely, instead of in the local namespace.
+
+ See InteractiveShell.run_cell for details.
"""
- raw_cell = cell
+
+ if (not raw_cell) or raw_cell.isspace():
+ return
+
+ ipself = self.shell
+
with ipself.builtin_trap:
- cell = ipself.prefilter_manager.prefilter_lines(cell)
+ cell = ipself.prefilter_manager.prefilter_lines(raw_cell)
# Store raw and processed history
if store_history:
@@ -187,6 +218,7 @@ def pxrun_cell(self, ipself, cell, store_history=True):
ipself.execution_count += 1
return None
except NameError:
+ # ignore name errors, because we don't know the remote keys
pass
if store_history:
@@ -195,7 +227,6 @@ def pxrun_cell(self, ipself, cell, store_history=True):
ipself.history_manager.store_output(ipself.execution_count)
# Each cell is a *single* input, regardless of how many lines it has
ipself.execution_count += 1
- print cell
if re.search(r'get_ipython\(\)\.magic\(u?"%?autopx', cell):
self._disable_autopx()
@@ -205,24 +236,49 @@ def pxrun_cell(self, ipself, cell, store_history=True):
result = self.active_view.execute(cell, block=False)
except:
ipself.showtraceback()
+ return True
+ else:
+ if self.active_view.block:
+ try:
+ result.get()
+ except:
+ self.shell.showtraceback()
+ return True
+ else:
+ self._maybe_display_output(result)
return False
-
- if self.active_view.block:
- try:
- result.get()
- except:
- ipself.showtraceback()
- else:
- targets = self.active_view.targets
- if isinstance(targets, int):
- targets = [targets]
- if targets == 'all':
- targets = self.active_view.client.ids
- stdout = [s.rstrip() for s in result.stdout]
- if any(stdout):
- for i,eid in enumerate(targets):
- print '[stdout:%i]'%eid, stdout[i]
+
+ def pxrun_code(self, code_obj):
+ """drop-in replacement for InteractiveShell.run_code.
+
+ This executes code remotely, instead of in the local namespace.
+
+ See InteractiveShell.run_code for details.
+ """
+ ipself = self.shell
+ # check code object for the autopx magic
+ if 'get_ipython' in code_obj.co_names and 'magic' in code_obj.co_names and \
+ any( [ isinstance(c, basestring) and 'autopx' in c for c in code_obj.co_consts ]):
+ self._disable_autopx()
return False
+ else:
+ try:
+ result = self.active_view.execute(code_obj, block=False)
+ except:
+ ipself.showtraceback()
+ return True
+ else:
+ if self.active_view.block:
+ try:
+ result.get()
+ except:
+ self.shell.showtraceback()
+ return True
+ else:
+ self._maybe_display_output(result)
+ return False
+
+
_loaded = False
@@ -765,11 +765,11 @@ def activate(self):
print "The IPython parallel magics (%result, %px, %autopx) only work within IPython."
else:
pmagic = ip.plugin_manager.get_plugin('parallelmagic')
- if pmagic is not None:
- pmagic.active_view = self
- else:
- print "You must first load the parallelmagic extension " \
- "by doing '%load_ext parallelmagic'"
+ if pmagic is None:
+ ip.magic_load_ext('parallelmagic')
+ pmagic = ip.plugin_manager.get_plugin('parallelmagic')
+
+ pmagic.active_view = self
@testdec.skip_doctest
Oops, something went wrong.