Feat: Introduce random delays to task handlers to stagger requests to sites.
This commit is contained in:
parent
ac4ce99b01
commit
0c0fe9f5e6
5 changed files with 99 additions and 51 deletions
3
FAQ.md
3
FAQ.md
|
|
@ -4,7 +4,7 @@ Certain configuration values can be set via environment variables, using the `-e
|
||||||
or the `environment:` section in `compose.yaml` file.
|
or the `environment:` section in `compose.yaml` file.
|
||||||
|
|
||||||
| Environment Variable | Description | Default |
|
| Environment Variable | Description | Default |
|
||||||
| ------------------------------ | ------------------------------------------------------------------ | --------------------- |
|
| ------------------------------ | ------------------------------------------------------------------- | --------------------- |
|
||||||
| TZ | The timezone to use for the application | `(not_set)` |
|
| TZ | The timezone to use for the application | `(not_set)` |
|
||||||
| YTP_OUTPUT_TEMPLATE | The template for the filenames of the downloaded videos | `%(title)s.%(ext)s` |
|
| YTP_OUTPUT_TEMPLATE | The template for the filenames of the downloaded videos | `%(title)s.%(ext)s` |
|
||||||
| YTP_DEFAULT_PRESET | The default preset to use for the download | `default` |
|
| YTP_DEFAULT_PRESET | The default preset to use for the download | `default` |
|
||||||
|
|
@ -50,6 +50,7 @@ or the `environment:` section in `compose.yaml` file.
|
||||||
| YTP_STATIC_UI_PATH | Path to custom static UI files. | `(not_set)` |
|
| YTP_STATIC_UI_PATH | Path to custom static UI files. | `(not_set)` |
|
||||||
| YTP_AUTO_CLEAR_HISTORY_DAYS | Number of days after which completed download history is cleared. | `0` |
|
| YTP_AUTO_CLEAR_HISTORY_DAYS | Number of days after which completed download history is cleared. | `0` |
|
||||||
| YTP_DEFAULT_PAGINATION | The default number of items per page for history. | `50` |
|
| YTP_DEFAULT_PAGINATION | The default number of items per page for history. | `50` |
|
||||||
|
| YTP_TASK_HANDLER_RANDOM_DELAY | The maximum random delay in seconds before starting a task handler. | `60` |
|
||||||
|
|
||||||
> [!NOTE]
|
> [!NOTE]
|
||||||
> To raise the maximum workers for specific extractor, you need to add a ENV variable that follows the pattern `YTP_MAX_WORKERS_FOR_<EXTRACTOR_NAME>`.
|
> To raise the maximum workers for specific extractor, you need to add a ENV variable that follows the pattern `YTP_MAX_WORKERS_FOR_<EXTRACTOR_NAME>`.
|
||||||
|
|
|
||||||
|
|
@ -3,6 +3,7 @@ import inspect
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
import pkgutil
|
import pkgutil
|
||||||
|
import random
|
||||||
import time
|
import time
|
||||||
import uuid
|
import uuid
|
||||||
from dataclasses import asdict, dataclass, field
|
from dataclasses import asdict, dataclass, field
|
||||||
|
|
@ -696,6 +697,8 @@ class HandleTask:
|
||||||
def _dispatcher(self):
|
def _dispatcher(self):
|
||||||
s: dict[list[str]] = {"h": [], "d": [], "u": [], "f": []}
|
s: dict[list[str]] = {"h": [], "d": [], "u": [], "f": []}
|
||||||
|
|
||||||
|
handler_groups: dict[str, list[tuple[Task, type]]] = {}
|
||||||
|
|
||||||
for task in self._tasks.get_all():
|
for task in self._tasks.get_all():
|
||||||
if not task.enabled or not task.handler_enabled:
|
if not task.enabled or not task.handler_enabled:
|
||||||
s["d"].append(task.name)
|
s["d"].append(task.name)
|
||||||
|
|
@ -712,19 +715,53 @@ class HandleTask:
|
||||||
s["u"].append(task.name)
|
s["u"].append(task.name)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
coro = self.dispatch(task, handler=handler)
|
handler_name = handler.__name__
|
||||||
t = asyncio.create_task(coro, name=f"task-{task.id}")
|
if handler_name not in handler_groups:
|
||||||
t.add_done_callback(lambda fut, t=task: self._handle_exception(fut, t))
|
handler_groups[handler_name] = []
|
||||||
|
handler_groups[handler_name].append((task, handler))
|
||||||
s["h"].append(task.name)
|
s["h"].append(task.name)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
LOG.error(f"Failed to handle task '{task.name}'. '{e!s}'.")
|
LOG.error(f"Failed to handle task '{task.name}'. '{e!s}'.")
|
||||||
s["f"].append(task.name)
|
s["f"].append(task.name)
|
||||||
|
|
||||||
|
for tasks_with_handlers in handler_groups.values():
|
||||||
|
for idx, (task, handler) in enumerate(tasks_with_handlers):
|
||||||
|
try:
|
||||||
|
t = asyncio.create_task(
|
||||||
|
coro=self._dispatch(
|
||||||
|
task,
|
||||||
|
handler,
|
||||||
|
delay=0.0 if 0 == idx else random.uniform(1.0, self._config.task_handler_random_delay),
|
||||||
|
),
|
||||||
|
name=f"task-{task.id}",
|
||||||
|
)
|
||||||
|
t.add_done_callback(lambda fut, t=task: self._handle_exception(fut, t))
|
||||||
|
except Exception as e:
|
||||||
|
LOG.error(f"Failed to dispatch task '{task.name}'. '{e!s}'.")
|
||||||
|
|
||||||
if len(self._tasks.get_all()) > 0:
|
if len(self._tasks.get_all()) > 0:
|
||||||
LOG.info(
|
LOG.info(
|
||||||
f"Tasks handler summary: Handled: {len(s['h'])}, Unhandled: {len(s['u'])}, Disabled: {len(s['d'])}, Failed: {len(s['f'])}."
|
f"Tasks handler summary: Handled: {len(s['h'])}, Unhandled: {len(s['u'])}, Disabled: {len(s['d'])}, Failed: {len(s['f'])}."
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def _dispatch(self, task: Task, handler: type, delay: float) -> TaskResult | TaskFailure | None:
|
||||||
|
"""
|
||||||
|
Dispatch a task after a random delay to avoid rate limiting.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
task (Task): The task to dispatch.
|
||||||
|
handler (type): The handler to use.
|
||||||
|
delay (float): The delay in seconds before dispatching.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
TaskResult|TaskFailure|None: The dispatch result.
|
||||||
|
|
||||||
|
"""
|
||||||
|
if delay > 0:
|
||||||
|
LOG.debug(f"Delaying dispatch of task '{task.name}' by {delay:.1f} seconds.")
|
||||||
|
await asyncio.sleep(delay)
|
||||||
|
return await self.dispatch(task, handler=handler)
|
||||||
|
|
||||||
def _handle_exception(self, fut: asyncio.Task, task: Task) -> None:
|
def _handle_exception(self, fut: asyncio.Task, task: Task) -> None:
|
||||||
if fut.cancelled():
|
if fut.cancelled():
|
||||||
return
|
return
|
||||||
|
|
|
||||||
|
|
@ -201,6 +201,9 @@ class Config(metaclass=Singleton):
|
||||||
default_pagination: int = 50
|
default_pagination: int = 50
|
||||||
"""The default number of items per page for pagination."""
|
"""The default number of items per page for pagination."""
|
||||||
|
|
||||||
|
task_handler_random_delay: float = 60.0
|
||||||
|
"""The maximum random delay in seconds before starting a task handler."""
|
||||||
|
|
||||||
pictures_backends: list[str] = [
|
pictures_backends: list[str] = [
|
||||||
"https://unsplash.it/1920/1080?random",
|
"https://unsplash.it/1920/1080?random",
|
||||||
"https://picsum.photos/1920/1080",
|
"https://picsum.photos/1920/1080",
|
||||||
|
|
@ -265,6 +268,9 @@ class Config(metaclass=Singleton):
|
||||||
)
|
)
|
||||||
"The variables that are booleans."
|
"The variables that are booleans."
|
||||||
|
|
||||||
|
_float_vars: tuple = ("task_handler_random_delay",)
|
||||||
|
"The variables that are floats."
|
||||||
|
|
||||||
_frontend_vars: tuple = (
|
_frontend_vars: tuple = (
|
||||||
"download_path",
|
"download_path",
|
||||||
"keep_archive",
|
"keep_archive",
|
||||||
|
|
@ -365,6 +371,9 @@ class Config(metaclass=Singleton):
|
||||||
if k in self._int_vars:
|
if k in self._int_vars:
|
||||||
setattr(self, k, int(v))
|
setattr(self, k, int(v))
|
||||||
|
|
||||||
|
if k in self._float_vars:
|
||||||
|
setattr(self, k, float(v))
|
||||||
|
|
||||||
if isinstance(self.pictures_backends, str) and self.pictures_backends:
|
if isinstance(self.pictures_backends, str) and self.pictures_backends:
|
||||||
self.pictures_backends = self.pictures_backends.split(",")
|
self.pictures_backends = self.pictures_backends.split(",")
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -120,7 +120,7 @@ async def get_background(request: Request, config: Config, cache: Cache) -> Resp
|
||||||
|
|
||||||
try:
|
try:
|
||||||
IS_REQUESTING_BACKGROUND = True
|
IS_REQUESTING_BACKGROUND = True
|
||||||
backend = random.choice(config.pictures_backends) # noqa: S311
|
backend = random.choice(config.pictures_backends)
|
||||||
CACHE_KEY_BING = "random_background_bing"
|
CACHE_KEY_BING = "random_background_bing"
|
||||||
CACHE_KEY = "random_background"
|
CACHE_KEY = "random_background"
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -168,6 +168,7 @@ ignore = [
|
||||||
"TRY004", # Like it's our choice to use ValuesError :|
|
"TRY004", # Like it's our choice to use ValuesError :|
|
||||||
"PT011",
|
"PT011",
|
||||||
"RUF001", # We like unicode chars.
|
"RUF001", # We like unicode chars.
|
||||||
|
"S311", # Not used for cryptography
|
||||||
]
|
]
|
||||||
|
|
||||||
# Allow fix for all enabled rules (when `--fix`) is provided.
|
# Allow fix for all enabled rules (when `--fix`) is provided.
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue