I have a class that stores a function and a method that tries to run the function in parallel using the multiprocessing
module. I'm testing the class with the nose
module and the nose.main()
function hangs (nothing happens, no output is displayed, the kernel must be manually told to stop and restart) when run. My directory looks like:
Parent/
test.py
Code/
__init__.py
code.py
tests/
test_code.py
code.py
contains:
import multiprocessing as mp
import numpy as np
import itertools
def target(): pass
def target_star(args): return target(*args)
class FuncWrapper():
def __init__(self, fun):
self.fun = fun
def run(self, X):
try:
arg_list_objects = []
arg_list_inputs = []
for i in range(len(X)):
arg_list_objects.append(self.fun.im_self)
arg_list_inputs.append(X[i])
target.__code__ = self.fun.im_func.__code__
pool = mp.Pool(processes=mp.cpu_count()-1)
F = np.array(pool.map(target_star, itertools.izip(arg_list_objects, arg_list_inputs))).reshape(X.shape)
pool.close()
pool.join()
except:
F = self.fun(X)
return F
Most of the code in the try:
block is used to make the pool.map
function work with class methods.
test_code.py
contains:
import numpy as np
from unittest import TestCase
import unittest
from Code.code import FuncWrapper
class TestCode(TestCase):
def f(self, x):
return x
def test_FuncWrapper(self):
x = np.random.uniform(0, 1, (50, 1))
return FuncWrapper(self.f).run(x)
if __name__ == '__main__':
unittest.main()
test.py
contains:
import nose
nose.main()
When I run test_code.py
from the Parent folder, it successfully performs the test but when test.py
is run, nothing happens. How should I use the nose.main
function so that it doesn't hang, or is there another function in the nose
module that would work here?
I'm using Ubuntu 14.04 LTS 64-bit with the Enthought Python distribution (Python 2.7.11 64-bit) and their Canopy IDE.