Don't wait for API request to finish to be build the next one

Author: mbaruh

bot/exts/filtering/_ui/ui.py

@@ -21,7 +21,7 @@
 from bot.exts.filtering._filter_context import Event, FilterContext
 from bot.exts.filtering._filter_lists import FilterList
 from bot.exts.filtering._utils import FakeContext, normalize_type
-from bot.utils.lock import lock_arg
+from bot.utils.async_utils import lock_arg
 from bot.utils.messages import format_channel, format_user, upload_log
 
 log = get_logger(__name__)

bot/exts/filtering/filtering.py

@@ -60,8 +60,8 @@
 from bot.exts.utils.snekbox._io import FileAttachment
 from bot.log import get_logger
 from bot.pagination import LinePaginator
+from bot.utils.async_utils import lock_arg
 from bot.utils.channel import is_mod_channel
-from bot.utils.lock import lock_arg
 from bot.utils.message_cache import MessageCache
 
 log = get_logger(__name__)

bot/exts/info/doc/_cog.py

@@ -17,7 +17,7 @@
 from bot.converters import Inventory, PackageName, ValidURL
 from bot.log import get_logger
 from bot.pagination import LinePaginator
-from bot.utils.lock import SharedEvent, lock
+from bot.utils.async_utils import SharedEvent, lock
 from bot.utils.messages import send_denial, wait_for_deletion
 
 from . import NAMESPACE, PRIORITY_PACKAGES, _batch_parser, doc_cache

bot/exts/info/doc/_redis_cache.py

@@ -5,7 +5,7 @@
 from async_rediscache.types.base import RedisObject
 
 from bot.log import get_logger
-from bot.utils.lock import lock
+from bot.utils.async_utils import lock
 
 from ._doc_item import DocItem
 

bot/exts/moderation/clean.py

@@ -4,7 +4,6 @@
 import time
 from collections import defaultdict
 from collections.abc import Callable, Iterable
-from contextlib import suppress
 from datetime import datetime
 from itertools import takewhile
 from typing import Literal, TYPE_CHECKING
@@ -19,6 +18,7 @@
 from bot.converters import Age, ISODateTime
 from bot.exts.moderation.modlog import ModLog
 from bot.log import get_logger
+from bot.utils.async_utils import AsyncExecutor
 from bot.utils.channel import is_mod_channel
 from bot.utils.messages import upload_log
 from bot.utils.modlog import send_log_message
@@ -33,6 +33,9 @@
 # Type alias for message lookup ranges.
 CleanLimit = Message | Age | ISODateTime
 
+# How many ongoing API requests a clean operation can have at the same time.
+CONCURRENT_REQUESTS_POOL = 10
+
 
 class CleanChannels(Converter):
     """A converter to turn the string into a list of channels to clean, or the literal `*` for all public channels."""
@@ -89,11 +92,11 @@ def mod_log(self) -> ModLog:
 
     @staticmethod
     def _validate_input(
-            channels: CleanChannels | None,
-            bots_only: bool,
-            users: list[User] | None,
-            first_limit: CleanLimit | None,
-            second_limit: CleanLimit | None,
+        channels: CleanChannels | None,
+        bots_only: bool,
+        users: list[User] | None,
+        first_limit: CleanLimit | None,
+        second_limit: CleanLimit | None,
     ) -> None:
         """Raise errors if an argument value or a combination of values is invalid."""
         if first_limit is None:
@@ -295,7 +298,9 @@ async def _delete_messages_individually(self, channel_messages: dict[TextChannel
                     deleted.append(message)
         return deleted
 
-    async def _delete_found(self, message_mappings: dict[TextChannel, list[Message]]) -> list[Message]:
+    async def _delete_found(
+        self, message_mappings: dict[TextChannel, list[Message]], executor:AsyncExecutor
+    ) -> list[Message]:
         """
         Delete the detected messages.
 
@@ -322,18 +327,19 @@ async def _delete_found(self, message_mappings: dict[TextChannel, list[Message]]
 
                 if len(to_delete) == 100:
                     # Only up to 100 messages can be deleted in a bulk
-                    await channel.delete_messages(to_delete)
+                    executor.submit(channel.delete_messages(to_delete))
                     deleted.extend(to_delete)
-                    to_delete.clear()
+                    to_delete = []
 
             if not self.cleaning:
                 return deleted
             if len(to_delete) > 0:
                 # Deleting any leftover messages if there are any
-                with suppress(NotFound):
-                    await channel.delete_messages(to_delete)
+                executor.submit(channel.delete_messages(to_delete))
                 deleted.extend(to_delete)
 
+        await executor.gather(return_exceptions=True)
+
         if old_messages:
             if not self.cleaning:
                 return deleted
@@ -418,9 +424,11 @@ async def _clean_messages(
         # Needs to be called after standardizing the input.
         predicate = self._build_predicate(first_limit, second_limit, bots_only, users, regex)
 
+        executor = AsyncExecutor(limit=CONCURRENT_REQUESTS_POOL)
+
         if attempt_delete_invocation:
             # Delete the invocation first
-            await self._delete_invocation(ctx)
+            executor.submit(self._delete_invocation(ctx))
 
         if self._use_cache(first_limit):
             log.trace(f"Messages for cleaning by {ctx.author.id} will be searched in the cache.")
@@ -442,8 +450,9 @@ async def _clean_messages(
 
         # Now let's delete the actual messages with purge.
         self.mod_log.ignore(Event.message_delete, *message_ids)
-        deleted_messages = await self._delete_found(message_mappings)
+        deleted_messages = await self._delete_found(message_mappings, executor)
         self.cleaning = False
+        log.trace("Cleaning completed, wrapping up")
 
         if not channels:
             channels = deletion_channels

bot/exts/moderation/silence.py

@@ -14,7 +14,7 @@
 from bot.bot import Bot
 from bot.converters import HushDurationConverter
 from bot.log import get_logger
-from bot.utils.lock import LockedResourceError, lock, lock_arg
+from bot.utils.async_utils import LockedResourceError, lock, lock_arg
 
 log = get_logger(__name__)
 

bot/exts/utils/reminders.py

@@ -29,8 +29,8 @@
 from bot.log import get_logger
 from bot.pagination import LinePaginator
 from bot.utils import time
+from bot.utils.async_utils import lock_arg
 from bot.utils.checks import has_any_role_check, has_no_roles_check
-from bot.utils.lock import lock_arg
 from bot.utils.messages import send_denial
 
 log = get_logger(__name__)

bot/exts/utils/snekbox/_cog.py

@@ -32,7 +32,7 @@
 from bot.exts.utils.snekbox._eval import EvalJob, EvalResult
 from bot.exts.utils.snekbox._io import FileAttachment
 from bot.log import get_logger
-from bot.utils.lock import LockedResourceError, lock_arg
+from bot.utils.async_utils import LockedResourceError, lock_arg
 
 if TYPE_CHECKING:
     from bot.exts.filtering.filtering import Filtering

bot/utils/lock.py bot/utils/async_utils.pybot/utils/lock.py renamed to bot/utils/async_utils.py

@@ -49,6 +49,32 @@ async def wait(self) -> None:
         await self._event.wait()
 
 
+class AsyncExecutor:
+    """Execute coroutines with a pool limit for concurrent execution."""
+
+    def __init__(self, *, limit: int):
+        self._semaphore = asyncio.Semaphore(limit)
+        self._running_tasks = set()
+
+    def submit[T](self, coro: Awaitable[T]) -> asyncio.Task[T]:
+        """Wraps the coroutine with _semaphore logic, schedules it on the event loop, and ensures cleanup."""
+        task = asyncio.create_task(self.execute(coro))
+        self._running_tasks.add(task)
+        task.add_done_callback(self._running_tasks.discard)
+        return task
+
+    async def execute[T](self, coro: Awaitable[T]) -> T:
+        """Executes the coroutine once there is an available slot in the _semaphore."""
+        async with self._semaphore:
+            return await coro
+
+    async def gather(self, return_exceptions: bool = False) -> list[Any]:
+        """Waits for all submitted coroutines to finish execution."""
+        if self._running_tasks:
+            return await asyncio.gather(*self._running_tasks, return_exceptions=return_exceptions)
+        return []
+
+
 def lock(
     namespace: Hashable,
     resource_id: ResourceId,