make flatten resilient to 'generator is already generating' while rem…

…aining resilient to 'max recur depth exceeded'

kioss/_exec.py

@@ -119,17 +119,21 @@ def __next__(self):
             except KeyError:  # KeyError: 'pop from an empty set'
                 raise StopIteration()
 
-            def f():
-                try:
-                    elem = next(next_iterator_elem)
-                except StopIteration:
+            def f(backoff: int = 0.005):
+                while True:
                     try:
-                        self.iterators_pool.remove(next_iterator_elem)
-                    except KeyError:
-                        pass
-                    return ThreadedFlatteningIteratorWrapper._SKIP
-
-                return elem
+                        elem = next(next_iterator_elem)
+                    except StopIteration:
+                        try:
+                            self.iterators_pool.remove(next_iterator_elem)
+                        except KeyError:
+                            pass
+                        return ThreadedFlatteningIteratorWrapper._SKIP
+                    except ValueError:  # generator anlready executing
+                        time.sleep(backoff)
+                        backoff *= 2
+                        continue
+                    return elem
 
             return f
 

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='kioss',
-    version='0.4.0',
+    version='0.4.1',
     packages=['kioss'],
     url='http://github.com/bonnal-enzo/kioss',
     license='Apache 2.',

tests/test_pipe.py

@@ -483,3 +483,13 @@ def test_planning_and_execution_decoupling(self):
         self.assertListEqual(a.collect(), list(range(N)))
         # test b not affected by a execution
         self.assertListEqual(b.collect(), [list(range(N))])
+
+    def test_generator_already_generating(self):
+        self.assertEqual(
+            Counter(
+                Pipe(
+                    lambda: [(ten_ms_identity(x) for x in range(N)) for _ in range(3)]
+                ).flatten(n_threads=2)
+            ),
+            Counter(list(range(N)) + list(range(N)) + list(range(N))),
+        )