change concurrency param to workers for multiprocessing

Author: RichardZhu2

src/pyper/_core/async_helper/stage.py

@@ -18,14 +18,14 @@
 
 class AsyncProducer:
     def __init__(self, task: Task, next_task: Task):
-        if task.concurrency > 1:
-            raise RuntimeError(f"The first task in a pipeline ({task.func.__qualname__}) cannot have concurrency greater than 1")
+        if task.workers > 1:
+            raise RuntimeError(f"The first task in a pipeline ({task.func.__qualname__}) cannot have more than 1 worker")
         if task.join:
             raise RuntimeError(f"The first task in a pipeline ({task.func.__qualname__}) cannot join previous results")
         self.task = task
         self.q_out = asyncio.Queue(maxsize=task.throttle)
 
-        self._n_consumers = 1 if next_task is None else next_task.concurrency
+        self._n_consumers = 1 if next_task is None else next_task.workers
         self._enqueue = AsyncEnqueueFactory(self.q_out, self.task)
 
     async def _worker(self, *args, **kwargs):
@@ -42,8 +42,8 @@ class AsyncProducerConsumer:
     def __init__(self, q_in: asyncio.Queue, task: Task, next_task: Task):
         self.q_out = asyncio.Queue(maxsize=task.throttle)
 
-        self._n_workers = task.concurrency
-        self._n_consumers = 1 if next_task is None else next_task.concurrency
+        self._n_workers = task.workers
+        self._n_consumers = 1 if next_task is None else next_task.workers
         self._dequeue = AsyncDequeueFactory(q_in, task)
         self._enqueue = AsyncEnqueueFactory(self.q_out, task)
         self._workers_done = 0

src/pyper/_core/decorators.py

@@ -19,13 +19,26 @@
 
 
 class task:
-    """Decorator class to initialize a `Pipeline` consisting of one task.
+    """Decorator class to initialize a `Pipeline` consisting of one task, from a function or callable.
 
-    The behaviour of each task within a Pipeline is determined by the parameters:
-    * `join`: allows the function to take all previous results as input, instead of single results
-    * `concurrency`: runs the functions with multiple (async or threaded) workers
-    * `throttle`: limits the number of results the function is able to produce when all consumers are busy
-    * `bind`: additional args and kwargs to bind to the function when defining a pipeline
+    Args:
+        func (callable): A positional-only param defining the task function
+        join (bool): Allows the task to take all previous results as input, instead of single results
+        workers (int): Defines the number of workers to run the task
+        throttle (int): Limits the number of results the task is able to produce when all consumers are busy
+        multiprocess (bool): Allows the task to be multiprocessed (cannot be `True` for async tasks)
+        bind (tuple[args, kwargs]): Additional args and kwargs to bind to the task when defining a pipeline
+
+    Returns:
+        Pipeline: A `Pipeline` instance consisting of one task.
+
+     Example:
+    ```python
+    def f(x: int):
+        return x + 1
+
+    p = task(f, workers=10, multiprocess=True)
+    ```
     """
     @t.overload
     def __new__(
@@ -34,7 +47,7 @@ def __new__(
             /,
             *,
             join: bool = False,
-            concurrency: int = 1,
+            workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
             bind: _ArgsKwargs = None) -> t.Type[task]: ...
@@ -46,7 +59,7 @@ def __new__(
             /,
             *,
             join: bool = False,
-            concurrency: int = 1,
+            workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
             bind: _ArgsKwargs = None) -> AsyncPipeline[_P, _R]: ...
@@ -58,7 +71,7 @@ def __new__(
             /,
             *,
             join: bool = False,
-            concurrency: int = 1,
+            workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
             bind: _ArgsKwargs = None) -> AsyncPipeline[_P, _R]: ...
@@ -70,7 +83,7 @@ def __new__(
             /,
             *,
             join: bool = False,
-            concurrency: int = 1,
+            workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
             bind: _ArgsKwargs = None) -> Pipeline[_P, _R]: ...
@@ -82,7 +95,7 @@ def __new__(
             /,
             *,
             join: bool = False,
-            concurrency: int = 1,
+            workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
             bind: _ArgsKwargs = None) -> Pipeline[_P, _R]: ...
@@ -93,14 +106,14 @@ def __new__(
             /,
             *,
             join: bool = False,
-            concurrency: int = 1,
+            workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
             bind: _ArgsKwargs = None):
         # Classic decorator trick: @task() means func is None, @task without parentheses means func is passed. 
         if func is None:
-            return functools.partial(cls, join=join, concurrency=concurrency, throttle=throttle, multiprocess=multiprocess, bind=bind)
-        return Pipeline([Task(func=func, join=join, concurrency=concurrency, throttle=throttle, multiprocess=multiprocess, bind=bind)])
+            return functools.partial(cls, join=join, workers=workers, throttle=throttle, multiprocess=multiprocess, bind=bind)
+        return Pipeline([Task(func=func, join=join, workers=workers, throttle=throttle, multiprocess=multiprocess, bind=bind)])
 
     @staticmethod
     def bind(*args, **kwargs) -> _ArgsKwargs:
@@ -113,6 +126,7 @@ def f(x: int, y: int):
 
         p = task(f, bind=task.bind(y=1))
         p(x=1)
+        ```
         """
         if not args and not kwargs:
             return None

src/pyper/_core/sync_helper/stage.py

@@ -21,8 +21,8 @@ def __init__(
             next_task: Task,
             q_err: Union[mp.Queue, queue.Queue],
             shutdown_event: Union[MpEvent, threading.Event]):
-        if task.concurrency > 1:
-            raise RuntimeError(f"The first task in a pipeline ({task.func.__qualname__}) cannot have concurrency greater than 1")
+        if task.workers > 1:
+            raise RuntimeError(f"The first task in a pipeline ({task.func.__qualname__}) cannot have more than 1 worker")
         if task.join:
             raise RuntimeError(f"The first task in a pipeline ({task.func.__qualname__}) cannot join previous results")
         self.q_out = mp.Queue(maxsize=task.throttle) \
@@ -31,8 +31,8 @@ def __init__(
 
         self._q_err = q_err
         self._shutdown_event = shutdown_event
-        self._n_workers = task.concurrency
-        self._n_consumers = 1 if next_task is None else next_task.concurrency
+        self._n_workers = task.workers
+        self._n_consumers = 1 if next_task is None else next_task.workers
         self._enqueue = EnqueueFactory(self.q_out, task)
 
     def _worker(self, *args, **kwargs):
@@ -65,8 +65,8 @@ def __init__(
 
         self._q_err = q_err
         self._shutdown_event = shutdown_event
-        self._n_workers = task.concurrency
-        self._n_consumers = 1 if next_task is None else next_task.concurrency
+        self._n_workers = task.workers
+        self._n_consumers = 1 if next_task is None else next_task.workers
         self._dequeue = DequeueFactory(q_in, task)
         self._enqueue = EnqueueFactory(self.q_out, task)
 

src/pyper/_core/task.py

@@ -13,7 +13,7 @@ class Task:
         "is_async",
         "func",
         "join",
-        "concurrency",
+        "workers",
         "throttle",
         "multiprocess"
     )
@@ -22,14 +22,14 @@ def __init__(
             self,
             func: Callable,
             join: bool = False,
-            concurrency: int = 1,
+            workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
             bind: Optional[Tuple[Tuple, Dict]] = None):
-        if not isinstance(concurrency, int):
-            raise TypeError("concurrency must be an integer")
-        if concurrency < 1:
-            raise ValueError("concurrency cannot be less than 1")
+        if not isinstance(workers, int):
+            raise TypeError("workers must be an integer")
+        if workers < 1:
+            raise ValueError("workers cannot be less than 1")
         if not isinstance(throttle, int):
             raise TypeError("throttle must be an integer")
         if throttle < 0:
@@ -51,6 +51,6 @@ def __init__(
 
         self.func = func if bind is None else functools.partial(func, *bind[0], **bind[1])
         self.join = join
-        self.concurrency = concurrency
+        self.workers = workers
         self.throttle = throttle
         self.multiprocess = multiprocess

src/pyper/_core/util/asynchronize.py

@@ -32,6 +32,6 @@ async def wrapper(*args, **kwargs):
     return Task(
         func=wrapper,
         join=task.join,
-        concurrency=task.concurrency,
+        workers=task.workers,
         throttle=task.throttle
     )