feature/fix-manager

benchmarks/100_million_downloads/client-pypeln-idiomatic.py

@@ -0,0 +1,26 @@
+# client-pypeln-pl.task.py
+
+from aiohttp import ClientSession, TCPConnector
+import asyncio
+import sys
+import pypeln as pl
+
+
+limit = 1000
+urls = ("http://localhost:8080/{}".format(i) for i in range(int(sys.argv[1])))
+
+
+async def main():
+
+    async with ClientSession(connector=TCPConnector(limit=0)) as session:
+
+        async def fetch(url):
+            async with session.get(url) as response:
+                return await response.read()
+
+        await pl.task.each(
+            fetch, urls, workers=limit,
+        )
+
+
+asyncio.run(main())

pypeln/process/utils.py

@@ -7,9 +7,13 @@
 
 from pypeln import utils as pypeln_utils
 
-# multiprocessing = get_context("fork")
-MANAGER = multiprocessing.Manager()
+MANAGER = None
 
 
 def Namespace(**kwargs) -> tp.Any:
+    global MANAGER
+
+    if MANAGER is None:
+        MANAGER = multiprocessing.Manager()
+
     return MANAGER.Namespace(**kwargs)

pypeln/sync/__init__.py

@@ -41,4 +41,4 @@ def slow_gt3(x):
 from .api.to_iterable import to_iterable
 from .api.ordered import ordered
 from .stage import Stage
-from .utils import get_namespace
+from .utils import Namespace

pypeln/sync/api/map_sync_test.py

@@ -47,7 +47,7 @@ def test_map_square_event_start(nums):
     nums_py = map(lambda x: x ** 2, nums)
     nums_py = list(nums_py)
 
-    namespace = pl.sync.get_namespace()
+    namespace = pl.sync.Namespace()
     namespace.x = 0
 
     def on_start():
@@ -99,7 +99,7 @@ def test_kwargs():
     nums = range(100)
     n_workers = 4
     letters = "abc"
-    namespace = pl.sync.get_namespace()
+    namespace = pl.sync.Namespace()
     namespace.on_done = None
 
     def on_start():
@@ -121,7 +121,7 @@ def on_done(y):
 @hp.settings(max_examples=MAX_EXAMPLES)
 def test_map_square_event_end(nums):
 
-    namespace = pl.sync.get_namespace()
+    namespace = pl.sync.Namespace()
     namespace.x = 0
     namespace.done = False
     namespace.active_workers = -1
@@ -156,6 +156,4 @@ def test_map_square_workers(nums):
 
     assert sorted(nums_pl) == sorted(nums_py)
 
-
-
     assert nums_pl == nums_py

pypeln/sync/utils.py

@@ -1,12 +1,13 @@
 from collections import namedtuple
 from queue import Empty, Full, Queue
 from threading import Lock
+import typing as tp
 
 from pypeln import utils as pypeln_utils
 
 
-def get_namespace():
-    return pypeln_utils.Namespace()
+def Namespace(**kwargs) -> tp.Any:
+    return pypeln_utils.Namespace(**kwargs)
 
 
 class StageStatus(object):

pypeln/task/api/each.py

@@ -39,8 +39,7 @@ def each(
     timeout: float = 0,
     on_start: tp.Callable = None,
     on_done: tp.Callable = None,
-    run: bool = False,
-) -> tp.Optional[Stage[B]]:
+) -> Stage[B]:
     ...
 
 
@@ -52,8 +51,21 @@ def each(
     timeout: float = 0,
     on_start: tp.Callable = None,
     on_done: tp.Callable = None,
+) -> pypeln_utils.Partial[Stage[B]]:
+    ...
+
+
+@tp.overload
+def each(
+    f: EachFn,
+    stage: tp.Union[Stage[A], tp.Iterable[A], tp.AsyncIterable[A]],
+    workers: int = 1,
+    maxsize: int = 0,
+    timeout: float = 0,
+    on_start: tp.Callable = None,
+    on_done: tp.Callable = None,
     run: bool = False,
-) -> pypeln_utils.Partial[tp.Optional[Stage[B]]]:
+) -> tp.Optional[Stage[B]]:
     ...
 
 

pypeln/task/utils.py

@@ -7,15 +7,8 @@
 from pypeln import utils as pypeln_utils
 
 
-class _Namespace:
-    def __init__(self, **kwargs):
-
-        for key, value in kwargs.items():
-            setattr(self, key, value)
-
-
 def Namespace(**kwargs) -> tp.Any:
-    return _Namespace(**kwargs)
+    return pypeln_utils.Namespace(**kwargs)
 
 
 def get_running_loop() -> asyncio.AbstractEventLoop:

pypeln/thread/api/map_thread_test.py

@@ -75,7 +75,7 @@ def f(x):
 
         return x
 
-    nums_pl = pl.thread.map(f, nums, timeout=0.5)
+    nums_pl = pl.thread.map(f, nums, timeout=0.2)
     nums_pl = list(nums_pl)
 
     assert len(nums_pl) == 9

pypeln/thread/queue_thread_test.py

@@ -66,6 +66,7 @@ def worker():
             for i in nums:
                 queue.put(i)
 
+            time.sleep(0.01)
             queue.stop()
 
         processes = pl.thread.start_workers(worker)

pypeln/thread/utils.py

@@ -7,9 +7,6 @@
 
 from pypeln import utils as pypeln_utils
 
-# multiprocessing = get_context("fork")
-MANAGER = multiprocessing.Manager()
-
 
 def Namespace(**kwargs) -> tp.Any:
-    return MANAGER.Namespace(**kwargs)
+    return pypeln_utils.Namespace(**kwargs)