diff --git a/IPython/parallel/apps/launcher.py b/IPython/parallel/apps/launcher.py
index f5b7d7f..171a248 100644
--- a/IPython/parallel/apps/launcher.py
+++ b/IPython/parallel/apps/launcher.py
@@ -21,13 +21,21 @@ import os
import re
import stat
+# signal imports, handling various platforms, versions
+
from signal import SIGINT, SIGTERM
try:
from signal import SIGKILL
except ImportError:
- # windows
+ # Windows
SIGKILL=SIGTERM
+try:
+ # Windows >= 2.7, 3.2
+ from signal import CTRL_C_EVENT as SIGINT
+except ImportError:
+ pass
+
from subprocess import Popen, PIPE, STDOUT
try:
from subprocess import check_output
@@ -51,24 +59,10 @@ from IPython.parallel.factory import LoggingFactory
from .win32support import forward_read_events
-# load winhpcjob only on Windows
-try:
- from .winhpcjob import (
- IPControllerTask, IPEngineTask,
- IPControllerJob, IPEngineSetJob
- )
-except ImportError:
- pass
+from .winhpcjob import IPControllerTask, IPEngineTask, IPControllerJob, IPEngineSetJob
WINDOWS = os.name == 'nt'
-if WINDOWS:
- try:
- # >= 2.7, 3.2
- from signal import CTRL_C_EVENT as SIGINT
- except ImportError:
- pass
-
#-----------------------------------------------------------------------------
# Paths to the kernel apps
#-----------------------------------------------------------------------------
@@ -282,11 +276,19 @@ class LocalProcessLauncher(BaseLauncher):
def signal(self, sig):
if self.state == 'running':
- self.process.send_signal(sig)
+ if WINDOWS and sig != SIGINT:
+ # use Windows tree-kill for better child cleanup
+ check_output(['taskkill', '-pid', str(self.process.pid), '-t', '-f'])
+ else:
+ self.process.send_signal(sig)
def interrupt_then_kill(self, delay=2.0):
"""Send INT, wait a delay and then send KILL."""
- self.signal(SIGINT)
+ try:
+ self.signal(SIGINT)
+ except Exception:
+ self.log.debug("interrupt failed")
+ pass
self.killer = ioloop.DelayedCallback(lambda : self.signal(SIGKILL), delay*1000, self.loop)
self.killer.start()
diff --git a/IPython/parallel/apps/winhpcjob.py b/IPython/parallel/apps/winhpcjob.py
index c9219e2..492df58 100644
--- a/IPython/parallel/apps/winhpcjob.py
+++ b/IPython/parallel/apps/winhpcjob.py
@@ -16,8 +16,6 @@ Job and task components for writing .xml files that the Windows HPC Server
# Imports
#-----------------------------------------------------------------------------
-from __future__ import with_statement
-
import os
import re
import uuid
diff --git a/IPython/parallel/tests/__init__.py b/IPython/parallel/tests/__init__.py
index 23d9ef9..0b1e013 100644
--- a/IPython/parallel/tests/__init__.py
+++ b/IPython/parallel/tests/__init__.py
@@ -14,21 +14,46 @@
import os
import tempfile
import time
-from subprocess import Popen, PIPE, STDOUT
+from subprocess import Popen
from IPython.utils.path import get_ipython_dir
from IPython.parallel import Client
+from IPython.parallel.apps.launcher import (LocalProcessLauncher,
+ ipengine_cmd_argv,
+ ipcontroller_cmd_argv,
+ SIGKILL)
-processes = []
-blackhole = tempfile.TemporaryFile()
+# globals
+launchers = []
+blackhole = open(os.devnull, 'w')
+
+# Launcher class
+class TestProcessLauncher(LocalProcessLauncher):
+ """subclass LocalProcessLauncher, to prevent extra sockets and threads being created on Windows"""
+ def start(self):
+ if self.state == 'before':
+ self.process = Popen(self.args,
+ stdout=blackhole, stderr=blackhole,
+ env=os.environ,
+ cwd=self.work_dir
+ )
+ self.notify_start(self.process.pid)
+ self.poll = self.process.poll
+ else:
+ s = 'The process was already started and has state: %r' % self.state
+ raise ProcessStateError(s)
# nose setup/teardown
def setup():
- cp = Popen('ipcontroller --profile iptest -r --log-level 10 --log-to-file --usethreads'.split(), stdout=blackhole, stderr=STDOUT)
- processes.append(cp)
- engine_json = os.path.join(get_ipython_dir(), 'cluster_iptest', 'security', 'ipcontroller-engine.json')
- client_json = os.path.join(get_ipython_dir(), 'cluster_iptest', 'security', 'ipcontroller-client.json')
+ cp = TestProcessLauncher()
+ cp.cmd_and_args = ipcontroller_cmd_argv + \
+ ['--profile', 'iptest', '--log-level', '99', '-r', '--usethreads']
+ cp.start()
+ launchers.append(cp)
+ cluster_dir = os.path.join(get_ipython_dir(), 'cluster_iptest')
+ engine_json = os.path.join(cluster_dir, 'security', 'ipcontroller-engine.json')
+ client_json = os.path.join(cluster_dir, 'security', 'ipcontroller-client.json')
tic = time.time()
while not os.path.exists(engine_json) or not os.path.exists(client_json):
if cp.poll() is not None:
@@ -44,9 +69,10 @@ def add_engines(n=1, profile='iptest'):
base = len(rc)
eps = []
for i in range(n):
- ep = Popen(['ipengine']+ ['--profile', profile, '--log-level', '10', '--log-to-file'], stdout=blackhole, stderr=STDOUT)
- # ep.start()
- processes.append(ep)
+ ep = TestProcessLauncher()
+ ep.cmd_and_args = ipengine_cmd_argv + ['--profile', profile, '--log-level', '99']
+ ep.start()
+ launchers.append(ep)
eps.append(ep)
tic = time.time()
while len(rc) < base+n:
@@ -61,11 +87,11 @@ def add_engines(n=1, profile='iptest'):
def teardown():
time.sleep(1)
- while processes:
- p = processes.pop()
+ while launchers:
+ p = launchers.pop()
if p.poll() is None:
try:
- p.terminate()
+ p.stop()
except Exception, e:
print e
pass
@@ -73,8 +99,9 @@ def teardown():
time.sleep(.25)
if p.poll() is None:
try:
- print 'killing'
- p.kill()
+ print 'cleaning up test process...'
+ p.signal(SIGKILL)
except:
print "couldn't shutdown process: ", p
+ blackhole.close()
diff --git a/IPython/parallel/tests/clienttest.py b/IPython/parallel/tests/clienttest.py
index 8cee15f..b6c2ed0 100644
--- a/IPython/parallel/tests/clienttest.py
+++ b/IPython/parallel/tests/clienttest.py
@@ -20,7 +20,8 @@ from IPython.external.decorator import decorator
from IPython.parallel import error
from IPython.parallel import Client
-from IPython.parallel.tests import processes,add_engines
+
+from IPython.parallel.tests import launchers, add_engines
# simple tasks for use in apply tests
@@ -112,8 +113,8 @@ class ClusterTestCase(BaseZMQTestCase):
def tearDown(self):
# self.client.clear(block=True)
# close fds:
- for e in filter(lambda e: e.poll() is not None, processes):
- processes.remove(e)
+ for e in filter(lambda e: e.poll() is not None, launchers):
+ launchers.remove(e)
# allow flushing of incoming messages to prevent crash on socket close
self.client.wait(timeout=2)
diff --git a/IPython/parallel/tests/test_asyncresult.py b/IPython/parallel/tests/test_asyncresult.py
index e3ef072..9457033 100644
--- a/IPython/parallel/tests/test_asyncresult.py
+++ b/IPython/parallel/tests/test_asyncresult.py
@@ -38,7 +38,6 @@ class AsyncResultTest(ClusterTestCase):
def test_get_after_done(self):
ar = self.client[-1].apply_async(lambda : 42)
- self.assertFalse(ar.ready())
ar.wait()
self.assertTrue(ar.ready())
self.assertEquals(ar.get(), 42)