Initialize a new instance. Args: system_status: Provides data about system utilization (load). concurrency_settings: Settings of concurrency levels. run_task_function: A function that performs an asynchronous resource-intensive task. is_task_r
(
self,
*,
system_status: SystemStatus,
concurrency_settings: ConcurrencySettings | None = None,
run_task_function: Callable[[], Awaitable],
is_task_ready_function: Callable[[], Awaitable[bool]],
is_finished_function: Callable[[], Awaitable[bool]],
)
| 62 | """Timeout within which the `run_task_function` must complete.""" |
| 63 | |
| 64 | def __init__( |
| 65 | self, |
| 66 | *, |
| 67 | system_status: SystemStatus, |
| 68 | concurrency_settings: ConcurrencySettings | None = None, |
| 69 | run_task_function: Callable[[], Awaitable], |
| 70 | is_task_ready_function: Callable[[], Awaitable[bool]], |
| 71 | is_finished_function: Callable[[], Awaitable[bool]], |
| 72 | ) -> None: |
| 73 | """Initialize a new instance. |
| 74 | |
| 75 | Args: |
| 76 | system_status: Provides data about system utilization (load). |
| 77 | concurrency_settings: Settings of concurrency levels. |
| 78 | run_task_function: A function that performs an asynchronous resource-intensive task. |
| 79 | is_task_ready_function: A function that indicates whether `run_task_function` should be called. This |
| 80 | function is called every time there is free capacity for a new task and it should indicate whether |
| 81 | it should start a new task or not by resolving to either `True` or `False`. Besides its obvious use, |
| 82 | it is also useful for task throttling to save resources. |
| 83 | is_finished_function: A function that is called only when there are no tasks to be processed. If it |
| 84 | resolves to `True` then the pool's run finishes. Being called only when there are no tasks being |
| 85 | processed means that as long as `is_task_ready_function` keeps resolving to `True`, |
| 86 | `is_finished_function` will never be called. To abort a run, use the `abort` method. |
| 87 | """ |
| 88 | concurrency_settings = concurrency_settings or ConcurrencySettings() |
| 89 | |
| 90 | self._system_status = system_status |
| 91 | self._run_task_function = run_task_function |
| 92 | self._is_task_ready_function = is_task_ready_function |
| 93 | self._is_finished_function = is_finished_function |
| 94 | self._desired_concurrency = concurrency_settings.desired_concurrency |
| 95 | self._max_concurrency = concurrency_settings.max_concurrency |
| 96 | self._min_concurrency = concurrency_settings.min_concurrency |
| 97 | self._max_tasks_per_minute = concurrency_settings.max_tasks_per_minute |
| 98 | |
| 99 | self._log_system_status_task = RecurringTask(self._log_system_status, self._LOGGING_INTERVAL) |
| 100 | self._autoscale_task = RecurringTask(self._autoscale, self._AUTOSCALE_INTERVAL) |
| 101 | |
| 102 | self._is_paused = False |
| 103 | self._current_run: _AutoscaledPoolRun | None = None |
| 104 | |
| 105 | async def run(self) -> None: |
| 106 | """Start the autoscaled pool and return when all tasks are completed and `is_finished_function` returns True. |
nothing calls this directly
no test coverage detected