-
- from multiprocessing.pool import ThreadPool
- import numpy as np
- def clean_index_list(obj):
- n = len(obj)
- converted = np.array(obj, dtype=object)
- floats = np.empty(n, dtype='f8')
- complexes = np.empty(n, dtype='c16')
- ints = np.empty(n, dtype='i8')
- uints = np.empty(n, dtype='u8')
-
- for i in xrange(n):
- val = converted[i]
- uints[i] = val
-
- return uints, 0
-
- n = 20000
- def run_test(*args):
- index = range(n)
- converted, all_arrays = clean_index_list(index)
- print("%3i %i" % args[0])
- return converted
-
- def test_multithread(i):
- pool = ThreadPool(10)
- results = pool.map(run_test, zip(range(200), [i]*200))
- for r in results:
- assert np.max(r) == n - 1,'max(r) is %d not %d' % (np.max(r), n - 1)
-
-
- if __name__ == '__main__':
- for i in range(10):
- test_multithread(i)
- print('ok')
-