refactor: better names

Author: Anton

taskiq/cli/worker/args.py

@@ -41,7 +41,7 @@ class WorkerArgs:
     receiver: str = "taskiq.receiver:Receiver"
     receiver_arg: List[Tuple[str, str]] = field(default_factory=list)
     max_prefetch: int = 0
-    max_idle_tasks: Optional[int] = None
+    max_sleeping_tasks: Optional[int] = None
     no_propagate_errors: bool = False
 
     @classmethod
@@ -189,11 +189,11 @@ def from_cli(  # noqa: WPS213
             help="Maximum prefetched tasks per worker process. ",
         )
         parser.add_argument(
-            "--max-idle-tasks",
+            "--max-sleeping-tasks",
             type=int,
-            dest="max_idle_tasks",
+            dest="max_sleeping_tasks",
             default=None,
-            help="Maximum idle tasks per worker process. ",
+            help="Maximum sleeping tasks per worker process. ",
         )
 
         namespace = parser.parse_args(args)

taskiq/cli/worker/run.py

@@ -133,7 +133,7 @@ def interrupt_handler(signum: int, _frame: Any) -> None:
                 validate_params=not args.no_parse,
                 max_async_tasks=args.max_async_tasks,
                 max_prefetch=args.max_prefetch,
-                max_idle_tasks=args.max_idle_tasks,
+                max_sleeping_tasks=args.max_sleeping_tasks,
                 propagate_exceptions=not args.no_propagate_errors,
                 **receiver_args,
             )

taskiq/context.py

@@ -14,10 +14,10 @@ def __init__(
         self,
         message: TaskiqMessage,
         broker: AsyncBroker,
-        task_idler: Callable[[float], Awaitable[None]],
+        sleep: Callable[[float], Awaitable[None]],
     ) -> None:
         self.message = message
         self.broker = broker
         self.state: "TaskiqState" = None  # type: ignore
         self.state = broker.state
-        self.task_idler = task_idler
+        self.sleep = sleep

taskiq/receiver/receiver.py

@@ -15,7 +15,7 @@
 from taskiq.message import TaskiqMessage
 from taskiq.receiver.params_parser import parse_params
 from taskiq.result import TaskiqResult
-from taskiq.semaphore import DequeSemaphore
+from taskiq.semaphore import PrioritySemaphore
 from taskiq.state import TaskiqState
 from taskiq.utils import DequeQueue, maybe_awaitable
 
@@ -48,7 +48,7 @@ def __init__(  # noqa: WPS211
         validate_params: bool = True,
         max_async_tasks: "Optional[int]" = None,
         max_prefetch: int = 0,
-        max_idle_tasks: Optional[int] = None,
+        max_sleeping_tasks: Optional[int] = None,
         propagate_exceptions: bool = True,
     ) -> None:
         self.broker = broker
@@ -62,22 +62,22 @@ def __init__(  # noqa: WPS211
             self.task_signatures[task.task_name] = inspect.signature(task.original_func)
             self.task_hints[task.task_name] = get_type_hints(task.original_func)
             self.dependency_graphs[task.task_name] = DependencyGraph(task.original_func)
-        self.sem: "Optional[DequeSemaphore]" = None
+        self.sem: "Optional[PrioritySemaphore]" = None
         if max_async_tasks is not None and max_async_tasks > 0:
-            self.sem = DequeSemaphore(max_async_tasks)
+            self.sem = PrioritySemaphore(max_async_tasks)
         else:
             logger.warning(
                 "Setting unlimited number of async tasks "
                 + "can result in undefined behavior",
             )
-        self.sem_prefetch = DequeSemaphore(max_prefetch)
+        self.sem_prefetch = PrioritySemaphore(max_prefetch)
         self.queue: DequeQueue[bytes] = DequeQueue()
 
-        self.sem_idle: Optional[asyncio.Semaphore] = None
-        if max_idle_tasks is not None and max_idle_tasks <= 0:
+        self.sem_sleeping: Optional[asyncio.Semaphore] = None
+        if max_sleeping_tasks is not None and max_sleeping_tasks <= 0:
             raise ValueError("`max_idle_tasks` should be greater then zero or None.")
-        if max_idle_tasks is not None and max_idle_tasks > 0:
-            self.sem_idle = asyncio.Semaphore(max_idle_tasks)
+        if max_sleeping_tasks is not None and max_sleeping_tasks > 0:
+            self.sem_sleeping = asyncio.Semaphore(max_sleeping_tasks)
 
     async def callback(  # noqa: C901, WPS213
         self,
@@ -190,7 +190,7 @@ async def run_task(  # noqa: C901, WPS210
             broker_ctx = self.broker.custom_dependency_context
             broker_ctx.update(
                 {
-                    Context: Context(message, self.broker, self.task_idler),
+                    Context: Context(message, self.broker, self.task_sleeper),
                     TaskiqState: self.broker.state,
                 },
             )
@@ -344,24 +344,26 @@ def task_cb(task: "asyncio.Task[Any]") -> None:
             # https://textual.textualize.io/blog/2023/02/11/the-heisenbug-lurking-in-your-async-code/
             task.add_done_callback(task_cb)
 
-    async def task_idler(self, wait: float) -> None:  # noqa: WPS213, WPS217
+    async def task_sleeper(self, wait: float) -> None:  # noqa: WPS213, WPS217
         """
-        Temporary increasing `max_async_tasks` for at least `wait` amount of time.
+        Non-blocking sleep for running tasks.
+
+        Temporary increasing `max_async_tasks` for at least `wait` amount of time
 
         :param wait: time
         """
         if not self.sem:
             await asyncio.sleep(wait)
             return
 
-        if not self.sem_idle:
-            logger.warning("`max_idle_tasks` is undefined. Idle is unavailable.")
+        if not self.sem_sleeping:
+            logger.warning("`max_sleeping_tasks` is undefined. Sleep is unavailable.")
             await asyncio.sleep(wait)
             return
 
         start_time = time()
         with anyio.move_on_after(wait) as scope:
-            await self.sem_idle.acquire()
+            await self.sem_sleeping.acquire()
 
         if scope.cancel_called:  # noqa: WPS441
             return
@@ -381,4 +383,4 @@ async def task_idler(self, wait: float) -> None:  # noqa: WPS213, WPS217
             await task
 
         finally:
-            self.sem_idle.release()
+            self.sem_sleeping.release()

taskiq/semaphore.py

@@ -5,9 +5,9 @@
 from typing_extensions import Literal
 
 
-class DequeSemaphore:
+class PrioritySemaphore:
     """
-    Custom deque based semaphore.
+    Custom semaphore with prioerities.
 
     https://neopythonic.blogspot.com/2022/10/reasoning-about-asynciosemaphore.html
     """

tests/cli/worker/test_receiver.py

@@ -34,7 +34,7 @@ def get_receiver(
     broker: Optional[AsyncBroker] = None,
     no_parse: bool = False,
     max_async_tasks: Optional[int] = None,
-    max_idle_tasks: Optional[int] = None,
+    max_sleeping_tasks: Optional[int] = None,
 ) -> Receiver:
     """
     Returns receiver with custom broker and args.
@@ -51,7 +51,7 @@ def get_receiver(
         executor=ThreadPoolExecutor(max_workers=10),
         validate_params=not no_parse,
         max_async_tasks=max_async_tasks,
-        max_idle_tasks=max_idle_tasks,
+        max_sleeping_tasks=max_sleeping_tasks,
     )
 
 
@@ -333,13 +333,13 @@ async def task_add_one(val: int) -> int:
     @broker.task
     async def task_map(vals: List[int], ctx: Context = Depends()) -> List[int]:
         tasks = [await task_add_one.kiq(val) for val in vals]
-        await ctx.task_idler(0.5)
+        await ctx.sleep(0.5)
         resps_tasks = [asyncio.create_task(t.wait_result(timeout=1)) for t in tasks]
         resps = await asyncio.gather(*resps_tasks)
         res = [r.return_value for r in resps]
         return res
 
-    receiver = get_receiver(broker, max_async_tasks=1, max_idle_tasks=1)
+    receiver = get_receiver(broker, max_async_tasks=1, max_sleeping_tasks=1)
     listen_task = asyncio.create_task(receiver.listen())
 
     task = await task_map.kiq(list(range(0, 10)))
@@ -349,7 +349,7 @@ async def task_map(vals: List[int], ctx: Context = Depends()) -> List[int]:
     await broker.shutdown()
     await listen_task
 
-    assert receiver.sem_idle._value == 1  # type: ignore
+    assert receiver.sem_sleeping._value == 1  # type: ignore
     assert receiver.sem._value == 1  # type: ignore
 
 
@@ -376,14 +376,14 @@ async def wait_for_task(
             resp_task = asyncio.create_task(
                 task.wait_result(interval * 0.4, timeout=interval),
             )
-            await ctx.task_idler(interval)
+            await ctx.sleep(interval)
 
             try:
                 return await resp_task
             except TaskiqResultTimeoutError:
                 continue
 
-    receiver = get_receiver(broker, max_async_tasks=1, max_idle_tasks=10)
+    receiver = get_receiver(broker, max_async_tasks=1, max_sleeping_tasks=10)
     listen_task = asyncio.create_task(receiver.listen())
 
     task = await task_run.kiq(10, "hello world!")
@@ -393,7 +393,7 @@ async def wait_for_task(
     await broker.shutdown()
     await listen_task
 
-    assert receiver.sem_idle._value == 10  # type: ignore
+    assert receiver.sem_sleeping._value == 10  # type: ignore
     assert receiver.sem._value == 1  # type: ignore
 
 
@@ -404,10 +404,10 @@ async def test_tasks_sleep() -> None:
 
     @broker.task
     async def task_run(ind: int, ctx: Context = Depends()) -> int:
-        await ctx.task_idler(0.1)
+        await ctx.sleep(0.1)
         return ind
 
-    receiver = get_receiver(broker, max_async_tasks=1, max_idle_tasks=20)
+    receiver = get_receiver(broker, max_async_tasks=1, max_sleeping_tasks=20)
     listen_task = asyncio.create_task(receiver.listen())
 
     with anyio.fail_after(1):
@@ -423,10 +423,11 @@ async def task_run(ind: int, ctx: Context = Depends()) -> int:
     await broker.shutdown()
     await listen_task
 
-    assert receiver.sem_idle._value == 20  # type: ignore
+    assert receiver.sem_sleeping._value == 20  # type: ignore
     assert receiver.sem._value == 1  # type: ignore
 
 
+@pytest.mark.anyio
 async def test_no_result_error() -> None:
     broker = InMemoryBroker()
     executed = asyncio.Event()

tests/test_semaphore.py

@@ -3,18 +3,18 @@
 import anyio
 import pytest
 
-from taskiq.semaphore import DequeSemaphore
+from taskiq.semaphore import PrioritySemaphore
 
 
 @pytest.mark.anyio
 async def test_semaphore_exception() -> None:
     with pytest.raises(ValueError):
-        DequeSemaphore(-1)
+        PrioritySemaphore(-1)
 
 
 @pytest.mark.anyio
 async def test_semaphore() -> None:
-    sem = DequeSemaphore(1)
+    sem = PrioritySemaphore(1)
 
     async def c1() -> None:
         await sem.acquire()