Add support for specifying the number of processes required for paral…

…lel execution

functional/execution.py

@@ -25,7 +25,12 @@ def evaluate(self, sequence, transformations):
 
 class ParallelExecutionEngine(ExecutionEngine):
 
+    def __init__(self, processes=None, *args, **kwargs):
+        super(ParallelExecutionEngine, self).__init__(*args, **kwargs)
+        self._processes = processes
+
     def evaluate(self, sequence, transformations):
+        processes = self._processes
         result = sequence
         staged = []
         for transform in transformations:
@@ -36,9 +41,9 @@ def evaluate(self, sequence, transformations):
                 staged.insert(0, transform.function)
             else:
                 if staged:
-                    result = parallelize(compose(*staged), result)
+                    result = parallelize(compose(*staged), result, processes)
                     staged = []
                 result = transform.function(result)
         if staged:
-            result = parallelize(compose(*staged), result)
+            result = parallelize(compose(*staged), result, processes)
         return iter(result)

functional/streams.py

@@ -13,7 +13,7 @@
 from functional.util import is_primitive, ReusableFile
 
 
-def seq(*args):
+def seq(*args, **kwargs):
     """
     Primary entrypoint for the functional package. Returns a functional.pipeline.Sequence wrapping
     the original sequence.
@@ -45,17 +45,22 @@ def seq(*args):
     :return: wrapped sequence
 
     """
-    return _seq(ExecutionEngine(), *args)
+    processes = kwargs.get("processes")
+    if processes:
+        return _seq(ParallelExecutionEngine(processes=processes), *args)
+    else:
+        return _seq(ExecutionEngine(), *args)
 
 
-def pseq(*args):
+def pseq(*args, **kwargs):
     """
     Same as functional.seq but with parallel support for maps/where and
     filter/select. Returns a functional.pipeline.Sequence wrapping
     the original sequence and passing
     functional.execution.ParallelExecutionEngine as the execution engine.
     """
-    return _seq(ParallelExecutionEngine(), *args)
+    processes = kwargs.get("processes")
+    return _seq(ParallelExecutionEngine(processes=processes), *args)
 
 
 def _seq(engine, *args):

functional/util.py

@@ -136,17 +136,22 @@ def is_serializable(func):
         return False
 
 
-def parallelize(func, result):
+def parallelize(func, result, processes=None):
     if not is_serializable(func):
         return func(result)
-    with Pool(processes=CPU_COUNT) as pool:
-        chunks = split_every(CPU_COUNT, iter(result))
+    if processes is None or processes < 1:
+        processes = CPU_COUNT
+    else:
+        processes = min(processes, CPU_COUNT)
+    with Pool(processes=processes) as pool:
+        chunks = split_every(processes, iter(result))
         packed_chunks = (pack(func, (chunk, )) for chunk in chunks)
         results = pool.map(unpack, packed_chunks)
     return chain.from_iterable(results)
 
 
 def compose(*functions):
+    # pylint: disable=undefined-variable
     return reduce(lambda f, g: lambda x: f(g(x)), functions, lambda x: x)