Merge pull request #3 from halprin/parallel

Parallel Version

Author: halprin

README.md

@@ -32,15 +32,17 @@ import iterator_chain
 ```
 
 ### Start the chain
-To start the chain, use the `from_iterable` function.  It takes an iterable.
+To start the chain, use the `from_iterable` or `from_iterable_parallel` function.  They take an iterable.
 ```python
 an_iterable = [5, 78, 12, 26]
-iterator_chain.from_iterable(an_iterable)
+chain = iterator_chain.from_iterable(an_iterable)
+parallel_chain = iterator_chain.from_iterable_parallel(an_iterable)
 ```
 
-| Function | Arguments | Description | 
+| Function | Arguments | Description |
 | --- | --- | --- |
 | `from_iterable` | • `iterable` - An iterable to be used in the iterator chain | Starts the iterator chain with the supplied iterable.  Chaining and terminating methods can now be called on the result. |
+| `from_iterable_parallel` | • `iterable` - An iterable to be used in the iterator chain<br/>• `chunksize` - Keyword.  How big of chunks to split the iterator up across the parallel execution units.  If unspecified or None, the chunk size will start at 1 and send that many elements to each execution unit.  The chunk size will then increment in powers of two and send that many items to each execution unit.  This is repeated until the iterator is exhausted.  This value is used as the default chunksize for all the following parallel based methods.  A specific parallel based method's chunksize can be overrided by supplying the `chunksize` keyword to that method. | Starts the iterator chain with the supplied iterable.  Chaining and terminating methods can now be called on the result.  Certain chaining and terminating methods will occur in parallel.  Parallel means separate processes to get around Python's GIL. |
 
 
 ### Continuing the chain
@@ -55,33 +57,44 @@ called.
 an actual value.  This value will depend on all the previous chaining methods being executed first.
 
 #### Chaining methods
-| Method | Arguments | Description | 
+| Method | Arguments | Description |
 | --- | --- | --- |
 | `map` | • `function` - A function that takes a single argument | Will run the `function` across all the elements in the iterator. |
 | `filter` | • `function` - A function that takes a single argument | Will run the `function` on every element.  `function` should return a truthy or falsy value.  On true, the element will stay; on false, the element will be removed. |
 | `skip` | • `number` - An integer | The `number` number of elements will be skipped over and effectively removed. |
 | `distinct` |  | Any duplicates will be removed. |
 | `limit` | • `max_size` - An integer | The iterator will stop after `max_size` elements.  Any elements afterward are effectively removed. |
 | `flatten` |  | Any element that is an iterable itself will have its elements iterated over first before continuing with the remaining elements.  Strings (`str`) do not count as an iterable for this method.  Dictionaries flatten to its item tuples. |
-| `sort` | • `key` - Keyword.  A function of one argument that is used to extract a comparison key from each element<br/>• `cmp` - Keyword.  A Python 2.x "cmp" function that takes two arguments<br/>• `reverse` - Keyword.  If set to `True`, the elements will be sorted in the reverse order. | Sorts the iterator based on the elements' values.  Use `key` or `cmp` to make a custom comparison.  If `key` is specified, `cmp` cannot be used.  This method is expensive because it must serialize all the values into a sequence. |
-| `reverse` |  | Reverses the iterator.  The last time will be first, and the first item will be last.  This method is expensive because it must serialize all the values into a list. |
+| `sort` | • `key` - Keyword.  A function of one argument that is used to extract a comparison key from each element<br/>• `cmp` - Keyword.  A Python 2.x "cmp" function that takes two arguments<br/>• `reverse` - Keyword.  If set to `True`, the elements will be sorted in the reverse order | Sorts the iterator based on the elements' values.  Use `key` or `cmp` to make a custom comparison.  If `key` is specified, `cmp` cannot be used.  This method is expensive because it must serialize all the values into a sequence. |
+| `reverse` |  | Reverses the iterator.  The last item will be first, and the first item will be last.  This method is expensive because it must serialize all the values into a list. |
+
+##### Parallel Versions
+| Method | Arguments | Description |
+| --- | --- | --- |
+| `map` | • `function` - A function that takes a single argument<br/>• `chunksize` - Keyword.  Overrides the chunksize supplied to the original `from_iterable_parallel`  | Will run the `function` across all the elements in the iterator in parallel. |
+| `filter` | • `function` - A function that takes a single argument<br/>• `chunksize` - Keyword.  Overrides the chunksize supplied to the original `from_iterable_parallel` | Will run the `function` on every element in parallel.  `function` should return a truthy or falsy value.  On true, the element will stay; on false, the element will be removed. |
 
 #### Terminating methods
-| Method | Arguments | Description | 
+| Method | Arguments | Description |
 | --- | --- | --- |
 | `list` |  | Serializes the iterator chain into a `list` and returns it. |
-| `count` |  | Returns the number of elements in the iterator |
+| `count` |  | Returns the number of elements in the iterator. |
 | `first` | • `default` - Keyword.  Any value. | Returns just the first item in the iterator.  If the iterator is empty, the `default` is returned. |
 | `last` | • `default` - Keyword.  Any value. | Returns just the last item in the iterator.  If the iterator is empty, the `default` is returned. |
 | `max` | • `default` - Keyword.  Any value. | Returns the largest valued element in the iterator.  If the iterator is empty, the `default` is returned. |
 | `min` | • `default` - Keyword.  Any value. | Returns the smallest valued element in the iterator.  If the iterator is empty, the `default` is returned. |
 | `sum` | • `default` - Keyword.  Any value. | Sums all the elements in the iterator together.  If any of the elements are un-summable, the `default` is returned. |
-| `reduce` | • `function` - A function that takes two arguments | Applies the function to two elements in the iterator cumulatively.  Subsequent calls to `function` uses the previous return value from `function` as the first argument and the next element in the iterator as the second argument.  The final value is returned. |
+| `reduce` | • `function` - A function that takes two arguments<br/>• `initial` - Keyword.  Any value. | Applies the function to two elements in the iterator cumulatively.  Subsequent calls to `function` uses the previous return value from `function` as the first argument and the next element in the iterator as the second argument.  The final value is returned.  If `initial` is present, it is placed before the items of the sequence in the calculation, and serves as a default when the sequence is empty. |
 | `for_each` | • `function` - A function that takes one argument and returns nothing | Executes `function` on every element in the iterator.  There is no return value.  If you are wanting to return a list of values based on the function, use `.map(_function_).list()`. |
 | `all_match` | • `function` - A function that takes one argument and returns a boolean | Returns `True` only if _all_ the elements return `True` after applying the `function` to them.  Else returns `False`. |
 | `any_match` | • `function` - A function that takes one argument and returns a boolean | Returns `True` if just one element return `True` after applying the `function` to it.  If all elements result in `False`, `False` is returned. |
 | `none_match` | • `function` - A function that takes one argument and returns a boolean | Returns `True` only if _all_ the elements return `False` after applying the `function` to them.  Else returns `True`. |
 
+##### Parallel Versions
+| Method | Arguments | Description |
+| --- | --- | --- |
+| `for_each` | • `function` - A function that takes one argument and returns nothing<br/>• `chunksize` - Keyword.  Overrides the chunksize supplied to the original `from_iterable_parallel` | Executes `function` on every element in the iterator in parallel.  There is no return value.  If you are wanting to return a list of values based on the function, use `.map(function).list()`. |
+
 ## Examples
 ```python
 import iterator_chain

iterator_chain/__init__.py

@@ -1,2 +1,3 @@
 __version__ = '1.0.0'
 from iterator_chain.begin import from_iterable
+from iterator_chain.begin import from_iterable_parallel

iterator_chain/begin.py

@@ -1,6 +1,27 @@
+from concurrent.futures import ProcessPoolExecutor
 from iterator_chain.intermediate import _IntermediateIteratorChain
+from iterator_chain.parallel_intermediate import _IntermediateParallelIteratorChain
 
 
 def from_iterable(iterable):
+    """
+    Starts the iterator chain with the supplied iterable.  Chaining and terminating methods can now be called on the result.
+
+    :param iterable: An iterable to be used in the iterator chain.
+    :return: An intermediate object that subsequent chaining and terminating methods can be called on.
+    """
     iterator = iter(iterable)
     return _IntermediateIteratorChain(iterator)
+
+
+def from_iterable_parallel(iterable, chunksize=None):
+    """
+    Starts the iterator chain with the supplied iterable.  Chaining and terminating methods can now be called on the result.  Certain chaining and terminating methods will occur in parallel.  Parallel means separate processes to get around Python's GIL.
+
+    :param iterable: An iterable to be used in the iterator chain.
+    :param chunksize: How big of chunks to split the iterator up across the parallel execution units.  If unspecified or None, the chunk size will start at 1 and send that many elements to each execution unit.  The chunk size will then increment in powers of two and send that many items to each execution unit.  This is repeated until the iterator is exhausted.  This value is used as the default chunksize for all the following parallel based methods.  A specific parallel based method's chunksize can be overrided by supplying the `chunksize` keyword to that method.
+    :return: An intermediate object that subsequent chaining and terminating methods can be called on.
+    """
+    iterator = iter(iterable)
+    executor = ProcessPoolExecutor()
+    return _IntermediateParallelIteratorChain(iterator, executor, chunksize=chunksize)

iterator_chain/intermediate.py

@@ -9,18 +9,44 @@ def __init__(self, iterator):
 
     # Chain methods
     def map(self, function):
-        iterator = map(function, self._iterator)
-        return _IntermediateIteratorChain(iterator)
+        """
+        Will run the `function` across all the elements in the iterator.
 
-    def skip(self, number):
-        iterator = itertools.islice(self._iterator, number, None)
+        :param function: A function that takes a single argument.
+        :return: An intermediate object that subsequent chaining and terminating methods can be called on.
+        """
+        iterator = map(function, self._iterator)
         return _IntermediateIteratorChain(iterator)
 
     def filter(self, function):
+        """
+        Will run the `function` on every element.  `function` should return a truthy or falsy value.  On true, the element will stay; on false, the element will be removed.
+
+        :param function: A function that takes a single argument.
+        :return: An intermediate object that subsequent chaining and terminating methods can be called on.
+        """
         iterator = filter(function, self._iterator)
         return _IntermediateIteratorChain(iterator)
 
+    def skip(self, number):
+        """
+        The `number` number of elements will be skipped over and effectively removed.
+
+        :param number: An integer.
+        :return: An intermediate object that subsequent chaining and terminating methods can be called on.
+        """
+        iterator = self._skip(number)
+        return _IntermediateIteratorChain(iterator)
+
+    def _skip(self, number):
+        return itertools.islice(self._iterator, number, None)
+
     def distinct(self):
+        """
+        Any duplicates will be removed.
+
+        :return: An intermediate object that subsequent chaining and terminating methods can be called on.
+        """
         iterator = self._distinct()
         return _IntermediateIteratorChain(iterator)
 
@@ -31,9 +57,18 @@ def _distinct(self):
             yield item
 
     def limit(self, max_size):
-        iterator = itertools.islice(self._iterator, max_size)
+        """
+        The iterator will stop after `max_size` elements.  Any elements afterward are effectively removed.
+
+        :param max_size: An integer.
+        :return: An intermediate object that subsequent chaining and terminating methods can be called on.
+        """
+        iterator = self._limit(max_size)
         return _IntermediateIteratorChain(iterator)
 
+    def _limit(self, max_size):
+        return itertools.islice(self._iterator, max_size)
+
     @staticmethod
     def _is_dict(something):
         return isinstance(something, dict)
@@ -54,62 +89,159 @@ def _flatten(self, iterable, force_stop=False):
                 yield item
 
     def flatten(self):
+        """
+        Any element that is an iterable itself will have its elements iterated over first before continuing with the remaining elements.  Strings (`str`) do not count as an iterable for this method.  Dictionaries flatten to its item tuples.
+
+        :return: An intermediate object that subsequent chaining and terminating methods can be called on.
+        """
         iterator = self._flatten(self._iterator)
         return _IntermediateIteratorChain(iterator)
 
     def sort(self, key=None, cmp=None, reverse=False):
+        """
+        Sorts the iterator based on the elements' values.  Use `key` or `cmp` to make a custom comparison.  If `key` is specified, `cmp` cannot be used.  This method is expensive because it must serialize all the values into a sequence.
+
+        :param key: Keyword.  A function of one argument that is used to extract a comparison key from each element.
+        :param cmp: Keyword.  A Python 2.x "cmp" function that takes two arguments.
+        :param reverse: Keyword.  If set to `True`, the elements will be sorted in the reverse order.
+        :return: An intermediate object that subsequent chaining and terminating methods can be called on.
+        """
+        iterator = self._sort(key=key, cmp=cmp, reverse=reverse)
+        return _IntermediateIteratorChain(iterator)
+
+    def _sort(self, key=None, cmp=None, reverse=False):
         if key is None and cmp is not None:
             key = functools.cmp_to_key(cmp)
-        iterator = iter(sorted(self._iterator, key=key, reverse=reverse))
-        return _IntermediateIteratorChain(iterator)
+        return iter(sorted(self._iterator, key=key, reverse=reverse))
 
     def reverse(self):
-        forward = list(self._iterator)
-        iterator = reversed(forward)
+        """
+        Reverses the iterator.  The last item will be first, and the first item will be last.  This method is expensive because it must serialize all the values into a list.
+
+        :return: An intermediate object that subsequent chaining and terminating methods can be called on.
+        """
+        iterator = self._reverse()
         return _IntermediateIteratorChain(iterator)
 
+    def _reverse(self):
+        forward = list(self._iterator)
+        return reversed(forward)
+
     # Termination methods
     def list(self):
+        """
+        Serializes the iterator chain into a `list` and returns it.
+
+        :return: A list whose elements come from the iterator.
+        """
         return list(self._iterator)
 
     def count(self):
+        """
+        Returns the number of elements in the iterator
+
+        :return: An integer.
+        """
         return sum(1 for _ in self._iterator)
 
     def first(self, default=None):
+        """
+        Returns just the first item in the iterator.  If the iterator is empty, the `default` is returned.
+
+        :param default: Keyword.  Any value.
+        :return: The first element.
+        """
         return next(itertools.islice(self._iterator, 1), default)
 
     def last(self, default=None):
+        """
+         Returns just the last item in the iterator.  If the iterator is empty, the `default` is returned.
+
+        :param default: Keyword.  Any value.
+        :return: The last element.
+        """
         try:
             end = collections.deque(self._iterator, maxlen=1).pop()
         except IndexError:
             end = default
         return end
 
     def max(self, default=None):
+        """
+        Returns the largest valued element in the iterator.  If the iterator is empty, the `default` is returned.
+
+        :param default: Keyword.  Any value.
+        :return: The largest element.
+        """
         return max(self._iterator, default=default)
 
     def min(self, default=None):
+        """
+        Returns the smallest valued element in the iterator.  If the iterator is empty, the `default` is returned.
+
+        :param default: Keyword.  Any value.
+        :return: The smallest element.
+        """
         return min(self._iterator, default=default)
 
     def sum(self, default=None):
+        """
+        Sums all the elements in the iterator together.  If any of the elements are un-summable, the `default` is returned.
+
+        :param default: Keyword.  Any value.
+        :return: The sum of all the elements.
+        """
         try:
             total = sum(self._iterator)
         except TypeError:
             total = default
         return total
 
-    def reduce(self, function):
-        return functools.reduce(function, self._iterator)
+    def reduce(self, function, initial=None):
+        """
+        Applies the function to two elements in the iterator cumulatively.  Subsequent calls to `function` uses the previous return value from `function` as the first argument and the next element in the iterator as the second argument.  The final value is returned.  If `initial` is present, it is placed before the items of the sequence in the calculation, and serves as a default when the sequence is empty.
+
+        :param function: A function that takes two arguments.
+        :param initial: Keyword.  Any value.
+        :return: The final reduced value.
+        """
+        if initial is None:
+            return functools.reduce(function, self._iterator)
+        else:
+            return functools.reduce(function, self._iterator, initial)
 
     def for_each(self, function):
+        """
+        Executes `function` on every element in the iterator.  There is no return value.  If you are wanting to return a list of values based on the function, use `.map(function).list()`.
+
+        :param function: A function that takes one argument and returns nothing.
+        """
         for item in self._iterator:
             function(item)
 
     def all_match(self, function):
+        """
+        Returns `True` only if all the elements return `True` after applying the `function` to them.  Else returns `False`.
+
+        :param function: A function that takes one argument and returns a boolean.
+        :return: True or False
+        """
         return all(map(function, self._iterator))
 
     def any_match(self, function):
+        """
+        Returns `True` if just one element return `True` after applying the `function` to it.  If all elements result in `False`, `False` is returned.
+
+        :param function: A function that takes one argument and returns a boolean.
+        :return: True or False
+        """
         return any(map(function, self._iterator))
 
     def none_match(self, function):
+        """
+        Returns `True` only if all the elements return `False` after applying the `function` to them.  Else returns `True`.
+
+        :param function: A function that takes one argument and returns a boolean.
+        :return: True or False
+        """
         return not self.any_match(function)