diff --git a/CHANGELOG.md b/CHANGELOG.md index 91ea5a6..b352337 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ ### Added +- **`--max-tasks-jitter J` on `run_database_tasks`.** Each worker adds a + random number from 0 to J, drawn once when it starts, to `--max-tasks`, + the way gunicorn's `max_requests_jitter` does, so the workers of + `--workers N --continuous --max-tasks K` no longer all reach K and restart + at the same moment. The draw is made by the worker, so every worker and + every restart gets its own, and from the operating system's random source, + so a project that seeds `random` does not make them alike; the limit drawn is printed at startup and + carried as a new `max_tasks` field on `Worker started`. Without + `--max-tasks` the option is ignored. The default is 0, which leaves + `--max-tasks` exactly as it was, and the only change to existing output is + the added log field, so existing projects need no changes. + ([#64](https://github.com/tokibito/django-database-task/issues/64)) - **A summary of the `--workers` supervisor's state in its log records.** When every worker is exiting with an error within ten seconds of starting at once, the supervisor logs one `Every worker is failing to start` diff --git a/README.md b/README.md index 4746b50..e0a38fe 100644 --- a/README.md +++ b/README.md @@ -256,6 +256,7 @@ python manage.py run_database_tasks [options] | `--continuous` | Keep polling even when no tasks | | `--interval` | Polling interval in seconds (default: 5) | | `--max-tasks` | Maximum number of tasks to process (0=unlimited) | +| `--max-tasks-jitter` | Add a random number from 0 to this to `--max-tasks`, drawn once per worker at start (default: 0). Ignored without `--max-tasks`. See [Running several workers](#running-several-workers) | | `--source` | Where to look for tasks: `auto` (default), `db`, `broker` or `both`. See [Task sources](#task-sources) | | `--wait-time` | Seconds to wait for a broker message before looking again (default: 20) | | `--max-messages` | Maximum number of broker messages to receive at a time (default: 1) | @@ -766,7 +767,24 @@ Shutdown complete: every worker exited. time. - **`--max-tasks`** goes to each worker. With `--continuous` that is gunicorn's `max_requests`: a worker exits after that many tasks and is - replaced, which bounds a leak in task code. + replaced, which bounds a leak in task code. The N workers start together + and process at similar rates, so on their own they reach the limit at + about the same time and restart together, leaving the host without a + worker for that moment. Add `--max-tasks-jitter J`, gunicorn's + `max_requests_jitter`, to spread them out: each worker draws a number + from 0 to J when it starts, a restarted one included, and stops after + `--max-tasks` plus that many tasks. A jitter of a tenth of `--max-tasks` + or more is a reasonable start: + + ```bash + python manage.py run_database_tasks --continuous --workers 4 \ + --max-tasks 1000 --max-tasks-jitter 100 + ``` + + The limit a worker drew is printed at startup + (`Max tasks: 1037 (1000 + jitter 37 of up to 100)`) and carried as + `max_tasks` on its `Worker started` log record. A worker run by a process + manager rather than `--workers` takes the option the same way. - **Signals.** `SIGTERM` and `SIGINT` (Ctrl-Break too on Windows) are handled by the supervisor, which forwards them to every worker and waits for the workers to finish their running task, within `--shutdown-timeout` if one is diff --git a/django_database_task/locale/es/LC_MESSAGES/django.mo b/django_database_task/locale/es/LC_MESSAGES/django.mo index c19ee00..8b2f083 100644 Binary files a/django_database_task/locale/es/LC_MESSAGES/django.mo and b/django_database_task/locale/es/LC_MESSAGES/django.mo differ diff --git a/django_database_task/locale/es/LC_MESSAGES/django.po b/django_database_task/locale/es/LC_MESSAGES/django.po index c61c524..835fcce 100644 --- a/django_database_task/locale/es/LC_MESSAGES/django.po +++ b/django_database_task/locale/es/LC_MESSAGES/django.po @@ -6,7 +6,7 @@ msgid "" msgstr "" "Project-Id-Version: django-database-task 0.1.0\n" "Report-Msgid-Bugs-To: \n" -"POT-Creation-Date: 2026-10-02 05:07+0000\n" +"POT-Creation-Date: 2026-10-12 16:27+0000\n" "PO-Revision-Date: 2026-09-30 00:00+0000\n" "Last-Translator: Shinya Okano \n" "Language-Team: Spanish\n" @@ -135,31 +135,42 @@ msgstr "Número de tareas a procesar de una vez (predeterminado: 1000)" msgid "Show what would happen without changing anything" msgstr "Muestra lo que ocurriría sin cambiar nada" -#: management/commands/run_database_tasks.py:63 +#: management/commands/run_database_tasks.py:70 msgid "Execute tasks queued in the database" msgstr "Ejecuta las tareas encoladas en la base de datos" -#: management/commands/run_database_tasks.py:70 +#: management/commands/run_database_tasks.py:77 msgid "Queue name to process (all queues if not specified)" msgstr "Nombre de la cola a procesar (todas las colas si no se indica)" -#: management/commands/run_database_tasks.py:76 +#: management/commands/run_database_tasks.py:83 msgid "Backend name (default: default)" msgstr "Nombre del backend (predeterminado: default)" -#: management/commands/run_database_tasks.py:81 +#: management/commands/run_database_tasks.py:88 msgid "Continuous mode (keep polling even when no tasks)" msgstr "Modo continuo (sigue sondeando aunque no haya tareas)" -#: management/commands/run_database_tasks.py:87 +#: management/commands/run_database_tasks.py:94 msgid "Polling interval in seconds for continuous mode (default: 5)" msgstr "Intervalo de sondeo en segundos en modo continuo (predeterminado: 5)" -#: management/commands/run_database_tasks.py:93 +#: management/commands/run_database_tasks.py:100 msgid "Maximum number of tasks to process (0=unlimited, default: 0)" msgstr "Número máximo de tareas a procesar (0=ilimitado, predeterminado: 0)" -#: management/commands/run_database_tasks.py:100 +#: management/commands/run_database_tasks.py:108 +msgid "" +"Add a random number from 0 to this to --max-tasks, drawn once when the " +"worker starts, so workers started together do not all reach the limit at " +"once. Ignored without --max-tasks (default: 0)" +msgstr "" +"Suma a --max-tasks un número aleatorio de 0 hasta este valor, sorteado una " +"vez cuando el worker se inicia, para que los workers iniciados a la vez no " +"alcancen el límite al mismo tiempo. Se ignora sin --max-tasks " +"(predeterminado: 0)" + +#: management/commands/run_database_tasks.py:119 msgid "" "Where to look for tasks: 'db' polls the database, 'broker' receives from the " "backend's broker, 'both' does the broker first and falls back to the " @@ -171,7 +182,7 @@ msgstr "" "datos, y 'auto' (predeterminado) significa 'both' cuando el backend tiene un " "broker del que recibir y 'db' en caso contrario" -#: management/commands/run_database_tasks.py:112 +#: management/commands/run_database_tasks.py:131 msgid "" "Seconds to wait for a broker message before looking again. Replaces --" "interval as the idle wait when receiving from a broker (default: 20)" @@ -180,12 +191,12 @@ msgstr "" "Sustituye a --interval como espera inactiva al recibir de un broker " "(predeterminado: 20)" -#: management/commands/run_database_tasks.py:122 +#: management/commands/run_database_tasks.py:141 msgid "Maximum number of broker messages to receive at a time (default: 1)" msgstr "" "Número máximo de mensajes del broker a recibir de una vez (predeterminado: 1)" -#: management/commands/run_database_tasks.py:131 +#: management/commands/run_database_tasks.py:150 msgid "" "Run this many worker processes from one command, each a copy of this command " "without --workers. With --continuous a worker that exits is replaced; " @@ -196,18 +207,18 @@ msgstr "" "se reemplaza; sin él, el comando termina cuando todos los workers lo hayan " "hecho (predeterminado: 1)" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:162 msgid "" "Run tasks in this many threads inside the worker. The task code must be " "thread-safe. The worker still polls once per interval; each thread has a " "database connection of its own (default: 1)" msgstr "" -"Ejecuta las tareas en esta cantidad de hilos dentro del worker. El código " -"de las tareas debe ser seguro para hilos. El worker sigue consultando una " -"vez por intervalo; cada hilo tiene su propia conexión a la base de datos " +"Ejecuta las tareas en esta cantidad de hilos dentro del worker. El código de " +"las tareas debe ser seguro para hilos. El worker sigue consultando una vez " +"por intervalo; cada hilo tiene su propia conexión a la base de datos " "(predeterminado: 1)" -#: management/commands/run_database_tasks.py:154 +#: management/commands/run_database_tasks.py:173 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -216,7 +227,7 @@ msgstr "" "SIGINT antes de forzar la salida (0=esperar indefinidamente, predeterminado: " "0)" -#: management/commands/run_database_tasks.py:163 +#: management/commands/run_database_tasks.py:182 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" @@ -224,7 +235,7 @@ msgstr "" "No instala manejadores de SIGTERM/SIGINT; el proceso termina de inmediato, " "incluso con una tarea en ejecución" -#: management/commands/run_database_tasks.py:173 +#: management/commands/run_database_tasks.py:192 msgid "" "Exit with this code when no task was processed, so a job scheduler can tell " "an idle run from a real one (0=exit normally, default: 0)" @@ -233,7 +244,7 @@ msgstr "" "planificador de trabajos distinga una ejecución vacía de una real " "(0=terminar normalmente, predeterminado: 0)" -#: management/commands/run_database_tasks.py:184 +#: management/commands/run_database_tasks.py:203 msgid "" "Exit with this code when at least one task failed or could not be run. Takes " "precedence over --empty-exit-code (0=exit normally, default: 0)" diff --git a/django_database_task/locale/ja/LC_MESSAGES/django.mo b/django_database_task/locale/ja/LC_MESSAGES/django.mo index 532ff55..b3fc286 100644 Binary files a/django_database_task/locale/ja/LC_MESSAGES/django.mo and b/django_database_task/locale/ja/LC_MESSAGES/django.mo differ diff --git a/django_database_task/locale/ja/LC_MESSAGES/django.po b/django_database_task/locale/ja/LC_MESSAGES/django.po index 007e2a2..bcb0c63 100644 --- a/django_database_task/locale/ja/LC_MESSAGES/django.po +++ b/django_database_task/locale/ja/LC_MESSAGES/django.po @@ -7,7 +7,7 @@ msgid "" msgstr "" "Project-Id-Version: django-database-task 0.1.0\n" "Report-Msgid-Bugs-To: \n" -"POT-Creation-Date: 2026-10-02 05:07+0000\n" +"POT-Creation-Date: 2026-10-12 16:27+0000\n" "PO-Revision-Date: 2025-12-27 00:00+0000\n" "Last-Translator: Shinya Okano \n" "Language-Team: Japanese\n" @@ -129,31 +129,41 @@ msgstr "一度に処理するタスク数 (デフォルト: 1000)" msgid "Show what would happen without changing anything" msgstr "何も変更せずに、何が行われるかを表示" -#: management/commands/run_database_tasks.py:63 +#: management/commands/run_database_tasks.py:70 msgid "Execute tasks queued in the database" msgstr "データベースのキューに登録されたタスクを実行" -#: management/commands/run_database_tasks.py:70 +#: management/commands/run_database_tasks.py:77 msgid "Queue name to process (all queues if not specified)" msgstr "処理するキュー名 (指定しない場合はすべてのキュー)" -#: management/commands/run_database_tasks.py:76 +#: management/commands/run_database_tasks.py:83 msgid "Backend name (default: default)" msgstr "バックエンド名 (デフォルト: default)" -#: management/commands/run_database_tasks.py:81 +#: management/commands/run_database_tasks.py:88 msgid "Continuous mode (keep polling even when no tasks)" msgstr "継続モード (タスクがなくてもポーリングを続ける)" -#: management/commands/run_database_tasks.py:87 +#: management/commands/run_database_tasks.py:94 msgid "Polling interval in seconds for continuous mode (default: 5)" msgstr "継続モードのポーリング間隔(秒)(デフォルト: 5)" -#: management/commands/run_database_tasks.py:93 +#: management/commands/run_database_tasks.py:100 msgid "Maximum number of tasks to process (0=unlimited, default: 0)" msgstr "処理する最大タスク数 (0=無制限、デフォルト: 0)" -#: management/commands/run_database_tasks.py:100 +#: management/commands/run_database_tasks.py:108 +msgid "" +"Add a random number from 0 to this to --max-tasks, drawn once when the " +"worker starts, so workers started together do not all reach the limit at " +"once. Ignored without --max-tasks (default: 0)" +msgstr "" +"--max-tasks に 0 からこの値までの乱数を加える。乱数はワーカーの起動時に 1 回" +"だけ引くので、同時に起動したワーカーが一斉に上限に達することがなくなる。--" +"max-tasks がないときは無視される (デフォルト: 0)" + +#: management/commands/run_database_tasks.py:119 msgid "" "Where to look for tasks: 'db' polls the database, 'broker' receives from the " "backend's broker, 'both' does the broker first and falls back to the " @@ -165,7 +175,7 @@ msgstr "" "バックする。'auto' (デフォルト) は、受信できるブローカーがバックエンドにあれ" "ば 'both'、なければ 'db' になる" -#: management/commands/run_database_tasks.py:112 +#: management/commands/run_database_tasks.py:131 msgid "" "Seconds to wait for a broker message before looking again. Replaces --" "interval as the idle wait when receiving from a broker (default: 20)" @@ -173,11 +183,11 @@ msgstr "" "再確認するまでブローカーのメッセージを待つ秒数。ブローカーから受信する場合、" "アイドル時の待機として --interval の代わりに使われる (デフォルト: 20)" -#: management/commands/run_database_tasks.py:122 +#: management/commands/run_database_tasks.py:141 msgid "Maximum number of broker messages to receive at a time (default: 1)" msgstr "一度に受信するブローカーメッセージの最大数 (デフォルト: 1)" -#: management/commands/run_database_tasks.py:131 +#: management/commands/run_database_tasks.py:150 msgid "" "Run this many worker processes from one command, each a copy of this command " "without --workers. With --continuous a worker that exits is replaced; " @@ -187,18 +197,17 @@ msgstr "" "このコマンドのコピー。--continuous 指定時は終了したワーカーを再起動し、未指定" "時は全ワーカーが終了するとコマンドも終了する (デフォルト: 1)" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:162 msgid "" "Run tasks in this many threads inside the worker. The task code must be " "thread-safe. The worker still polls once per interval; each thread has a " "database connection of its own (default: 1)" msgstr "" -"ワーカー内でこの数のスレッドでタスクを実行する。タスクのコードはスレッドセーフである必要がある。ポーリングは引き続き " -"interval ごとに 1 " -"回で、スレッドごとにデータベース接続を 1 本持つ " -"(デフォルト: 1)" +"ワーカー内でこの数のスレッドでタスクを実行する。タスクのコードはスレッドセー" +"フである必要がある。ポーリングは引き続き interval ごとに 1 回で、スレッドごと" +"にデータベース接続を 1 本持つ (デフォルト: 1)" -#: management/commands/run_database_tasks.py:154 +#: management/commands/run_database_tasks.py:173 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -206,7 +215,7 @@ msgstr "" "SIGTERM/SIGINTを受信してから強制終了するまで、実行中のタスクを待つ最大秒数 " "(0=無期限に待つ、デフォルト: 0)" -#: management/commands/run_database_tasks.py:163 +#: management/commands/run_database_tasks.py:182 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" @@ -214,7 +223,7 @@ msgstr "" "SIGTERM/SIGINTハンドラーを設定しない。タスクの実行中でもプロセスは即座に終了" "する" -#: management/commands/run_database_tasks.py:173 +#: management/commands/run_database_tasks.py:192 msgid "" "Exit with this code when no task was processed, so a job scheduler can tell " "an idle run from a real one (0=exit normally, default: 0)" @@ -222,7 +231,7 @@ msgstr "" "タスクを1件も処理しなかった場合にこの終了コードで終了する。ジョブスケジュー" "ラーが空振りの実行を区別できるようにする (0=通常どおり終了、デフォルト: 0)" -#: management/commands/run_database_tasks.py:184 +#: management/commands/run_database_tasks.py:203 msgid "" "Exit with this code when at least one task failed or could not be run. Takes " "precedence over --empty-exit-code (0=exit normally, default: 0)" diff --git a/django_database_task/locale/pt_BR/LC_MESSAGES/django.mo b/django_database_task/locale/pt_BR/LC_MESSAGES/django.mo index 275e84e..2628a31 100644 Binary files a/django_database_task/locale/pt_BR/LC_MESSAGES/django.mo and b/django_database_task/locale/pt_BR/LC_MESSAGES/django.mo differ diff --git a/django_database_task/locale/pt_BR/LC_MESSAGES/django.po b/django_database_task/locale/pt_BR/LC_MESSAGES/django.po index 8ca5a27..4b42e28 100644 --- a/django_database_task/locale/pt_BR/LC_MESSAGES/django.po +++ b/django_database_task/locale/pt_BR/LC_MESSAGES/django.po @@ -6,7 +6,7 @@ msgid "" msgstr "" "Project-Id-Version: django-database-task 0.1.0\n" "Report-Msgid-Bugs-To: \n" -"POT-Creation-Date: 2026-10-02 05:07+0000\n" +"POT-Creation-Date: 2026-10-12 16:27+0000\n" "PO-Revision-Date: 2026-09-30 00:00+0000\n" "Last-Translator: Shinya Okano \n" "Language-Team: Brazilian Portuguese\n" @@ -132,31 +132,41 @@ msgstr "Número de tarefas a processar de uma vez (padrão: 1000)" msgid "Show what would happen without changing anything" msgstr "Mostra o que aconteceria sem alterar nada" -#: management/commands/run_database_tasks.py:63 +#: management/commands/run_database_tasks.py:70 msgid "Execute tasks queued in the database" msgstr "Executa as tarefas enfileiradas no banco de dados" -#: management/commands/run_database_tasks.py:70 +#: management/commands/run_database_tasks.py:77 msgid "Queue name to process (all queues if not specified)" msgstr "Nome da fila a processar (todas as filas se não for informado)" -#: management/commands/run_database_tasks.py:76 +#: management/commands/run_database_tasks.py:83 msgid "Backend name (default: default)" msgstr "Nome do backend (padrão: default)" -#: management/commands/run_database_tasks.py:81 +#: management/commands/run_database_tasks.py:88 msgid "Continuous mode (keep polling even when no tasks)" msgstr "Modo contínuo (continua consultando mesmo sem tarefas)" -#: management/commands/run_database_tasks.py:87 +#: management/commands/run_database_tasks.py:94 msgid "Polling interval in seconds for continuous mode (default: 5)" msgstr "Intervalo de consulta em segundos no modo contínuo (padrão: 5)" -#: management/commands/run_database_tasks.py:93 +#: management/commands/run_database_tasks.py:100 msgid "Maximum number of tasks to process (0=unlimited, default: 0)" msgstr "Número máximo de tarefas a processar (0=ilimitado, padrão: 0)" -#: management/commands/run_database_tasks.py:100 +#: management/commands/run_database_tasks.py:108 +msgid "" +"Add a random number from 0 to this to --max-tasks, drawn once when the " +"worker starts, so workers started together do not all reach the limit at " +"once. Ignored without --max-tasks (default: 0)" +msgstr "" +"Soma a --max-tasks um número aleatório de 0 até este valor, sorteado uma vez " +"quando o worker inicia, para que workers iniciados juntos não atinjam o " +"limite ao mesmo tempo. Ignorado sem --max-tasks (padrão: 0)" + +#: management/commands/run_database_tasks.py:119 msgid "" "Where to look for tasks: 'db' polls the database, 'broker' receives from the " "backend's broker, 'both' does the broker first and falls back to the " @@ -168,7 +178,7 @@ msgstr "" "e 'auto' (padrão) significa 'both' quando o backend tem um broker do qual " "receber e 'db' caso contrário" -#: management/commands/run_database_tasks.py:112 +#: management/commands/run_database_tasks.py:131 msgid "" "Seconds to wait for a broker message before looking again. Replaces --" "interval as the idle wait when receiving from a broker (default: 20)" @@ -176,11 +186,11 @@ msgstr "" "Segundos de espera por uma mensagem do broker antes de verificar novamente. " "Substitui --interval como espera ociosa ao receber de um broker (padrão: 20)" -#: management/commands/run_database_tasks.py:122 +#: management/commands/run_database_tasks.py:141 msgid "Maximum number of broker messages to receive at a time (default: 1)" msgstr "Número máximo de mensagens do broker a receber de uma vez (padrão: 1)" -#: management/commands/run_database_tasks.py:131 +#: management/commands/run_database_tasks.py:150 msgid "" "Run this many worker processes from one command, each a copy of this command " "without --workers. With --continuous a worker that exits is replaced; " @@ -191,7 +201,7 @@ msgstr "" "que sai é substituído; sem ele, o comando termina quando todos os workers " "terminarem (padrão: 1)" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:162 msgid "" "Run tasks in this many threads inside the worker. The task code must be " "thread-safe. The worker still polls once per interval; each thread has a " @@ -202,7 +212,7 @@ msgstr "" "por intervalo; cada thread tem a sua própria conexão com o banco de dados " "(padrão: 1)" -#: management/commands/run_database_tasks.py:154 +#: management/commands/run_database_tasks.py:173 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -210,7 +220,7 @@ msgstr "" "Tempo máximo, em segundos, de espera pela tarefa em execução após receber " "SIGTERM/SIGINT antes de forçar a saída (0=esperar indefinidamente, padrão: 0)" -#: management/commands/run_database_tasks.py:163 +#: management/commands/run_database_tasks.py:182 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" @@ -218,7 +228,7 @@ msgstr "" "Não instala manipuladores de SIGTERM/SIGINT; o processo é encerrado " "imediatamente, mesmo com uma tarefa em execução" -#: management/commands/run_database_tasks.py:173 +#: management/commands/run_database_tasks.py:192 msgid "" "Exit with this code when no task was processed, so a job scheduler can tell " "an idle run from a real one (0=exit normally, default: 0)" @@ -227,7 +237,7 @@ msgstr "" "agendador de jobs distinga uma execução ociosa de uma real (0=sair " "normalmente, padrão: 0)" -#: management/commands/run_database_tasks.py:184 +#: management/commands/run_database_tasks.py:203 msgid "" "Exit with this code when at least one task failed or could not be run. Takes " "precedence over --empty-exit-code (0=exit normally, default: 0)" diff --git a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.mo b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.mo index 1920335..034f0c4 100644 Binary files a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.mo and b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.mo differ diff --git a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po index 0035b39..379333c 100644 --- a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po +++ b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po @@ -6,7 +6,7 @@ msgid "" msgstr "" "Project-Id-Version: django-database-task 0.1.0\n" "Report-Msgid-Bugs-To: \n" -"POT-Creation-Date: 2026-10-02 05:07+0000\n" +"POT-Creation-Date: 2026-10-12 16:27+0000\n" "PO-Revision-Date: 2026-09-30 00:00+0000\n" "Last-Translator: Shinya Okano \n" "Language-Team: Simplified Chinese\n" @@ -123,31 +123,40 @@ msgstr "每次处理的任务数(默认:1000)" msgid "Show what would happen without changing anything" msgstr "显示将会执行的操作,但不做任何更改" -#: management/commands/run_database_tasks.py:63 +#: management/commands/run_database_tasks.py:70 msgid "Execute tasks queued in the database" msgstr "执行在数据库中排队的任务" -#: management/commands/run_database_tasks.py:70 +#: management/commands/run_database_tasks.py:77 msgid "Queue name to process (all queues if not specified)" msgstr "要处理的队列名称(未指定时为所有队列)" -#: management/commands/run_database_tasks.py:76 +#: management/commands/run_database_tasks.py:83 msgid "Backend name (default: default)" msgstr "后端名称(默认:default)" -#: management/commands/run_database_tasks.py:81 +#: management/commands/run_database_tasks.py:88 msgid "Continuous mode (keep polling even when no tasks)" msgstr "持续模式(没有任务时也继续轮询)" -#: management/commands/run_database_tasks.py:87 +#: management/commands/run_database_tasks.py:94 msgid "Polling interval in seconds for continuous mode (default: 5)" msgstr "持续模式的轮询间隔秒数(默认:5)" -#: management/commands/run_database_tasks.py:93 +#: management/commands/run_database_tasks.py:100 msgid "Maximum number of tasks to process (0=unlimited, default: 0)" msgstr "要处理的最大任务数(0=不限,默认:0)" -#: management/commands/run_database_tasks.py:100 +#: management/commands/run_database_tasks.py:108 +msgid "" +"Add a random number from 0 to this to --max-tasks, drawn once when the " +"worker starts, so workers started together do not all reach the limit at " +"once. Ignored without --max-tasks (default: 0)" +msgstr "" +"在 --max-tasks 上加一个 0 到此值之间的随机数,在工作进程启动时抽取一次,使同" +"时启动的工作进程不会同时达到上限。未指定 --max-tasks 时忽略(默认:0)" + +#: management/commands/run_database_tasks.py:119 msgid "" "Where to look for tasks: 'db' polls the database, 'broker' receives from the " "backend's broker, 'both' does the broker first and falls back to the " @@ -158,7 +167,7 @@ msgstr "" "消息代理再回退到数据库,'auto'(默认)在后端有可接收的消息代理时为 'both',否" "则为 'db'" -#: management/commands/run_database_tasks.py:112 +#: management/commands/run_database_tasks.py:131 msgid "" "Seconds to wait for a broker message before looking again. Replaces --" "interval as the idle wait when receiving from a broker (default: 20)" @@ -166,11 +175,11 @@ msgstr "" "再次查看前等待消息代理消息的秒数。从消息代理接收时,取代 --interval 作为空闲" "时的等待(默认:20)" -#: management/commands/run_database_tasks.py:122 +#: management/commands/run_database_tasks.py:141 msgid "Maximum number of broker messages to receive at a time (default: 1)" msgstr "每次接收的消息代理消息的最大数量(默认:1)" -#: management/commands/run_database_tasks.py:131 +#: management/commands/run_database_tasks.py:150 msgid "" "Run this many worker processes from one command, each a copy of this command " "without --workers. With --continuous a worker that exits is replaced; " @@ -180,15 +189,16 @@ msgstr "" "continuous 时,退出的工作进程会被替换;未指定时,所有工作进程退出后命令即退出" "(默认:1)" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:162 msgid "" "Run tasks in this many threads inside the worker. The task code must be " "thread-safe. The worker still polls once per interval; each thread has a " "database connection of its own (default: 1)" msgstr "" -"在工作进程内用这么多个线程运行任务。任务代码必须是线程安全的。工作进程仍每个间隔轮询一次;每个线程有自己的数据库连接(默认:1)" +"在工作进程内用这么多个线程运行任务。任务代码必须是线程安全的。工作进程仍每个" +"间隔轮询一次;每个线程有自己的数据库连接(默认:1)" -#: management/commands/run_database_tasks.py:154 +#: management/commands/run_database_tasks.py:173 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -196,13 +206,13 @@ msgstr "" "收到 SIGTERM/SIGINT 后、强制退出前等待正在运行的任务的最长秒数(0=无限期等" "待,默认:0)" -#: management/commands/run_database_tasks.py:163 +#: management/commands/run_database_tasks.py:182 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" msgstr "不安装 SIGTERM/SIGINT 处理程序;即使有任务正在运行,进程也会立即终止" -#: management/commands/run_database_tasks.py:173 +#: management/commands/run_database_tasks.py:192 msgid "" "Exit with this code when no task was processed, so a job scheduler can tell " "an idle run from a real one (0=exit normally, default: 0)" @@ -210,7 +220,7 @@ msgstr "" "未处理任何任务时以此代码退出,以便作业调度器区分空转的运行与实际的运行(0=正" "常退出,默认:0)" -#: management/commands/run_database_tasks.py:184 +#: management/commands/run_database_tasks.py:203 msgid "" "Exit with this code when at least one task failed or could not be run. Takes " "precedence over --empty-exit-code (0=exit normally, default: 0)" diff --git a/django_database_task/management/commands/run_database_tasks.py b/django_database_task/management/commands/run_database_tasks.py index 88484a0..b45bcdb 100644 --- a/django_database_task/management/commands/run_database_tasks.py +++ b/django_database_task/management/commands/run_database_tasks.py @@ -1,6 +1,7 @@ import argparse import functools import logging +import random import socket import sys import threading @@ -98,6 +99,18 @@ def add_arguments(self, parser): default=0, help=_("Maximum number of tasks to process (0=unlimited, default: 0)"), ) + parser.add_argument( + "--max-tasks-jitter", + type=int, + default=0, + metavar="J", + help=_( + "Add a random number from 0 to this to --max-tasks, drawn once " + "when the worker starts, so workers started together do not " + "all reach the limit at once. Ignored without --max-tasks " + "(default: 0)" + ), + ) parser.add_argument( "--source", choices=SOURCES, @@ -214,6 +227,9 @@ def handle(self, *args, **options): threads = options["threads"] if threads < 1: raise CommandError("--threads must be at least 1") + max_tasks_jitter = options["max_tasks_jitter"] + if max_tasks_jitter < 0: + raise CommandError("--max-tasks-jitter must not be negative") backend = task_backends[backend_name] source = self._resolve_source(backend, options["source"]) @@ -226,6 +242,12 @@ def handle(self, *args, **options): worker_id = f"{socket.gethostname()}-{uuid.uuid4().hex[:8]}" worker_index = self._read_worker_index() + # Drawn here, in the worker, rather than by the --workers supervisor, + # so that every worker and every restart gets a value of its own; and + # from the OS rather than the random module's shared generator, which + # a project that seeds it at import would make draw alike everywhere. + jitter = random.SystemRandom().randint(0, max_tasks_jitter) if max_tasks else 0 + max_tasks += jitter if verbosity >= 1: self.stdout.write(f"Worker ID: {worker_id}") if worker_index is not None: @@ -243,7 +265,12 @@ def handle(self, *args, **options): f"Broker: {type(backend.broker).__name__} " f"(wait={wait_time}s, max_messages={max_messages})" ) - if max_tasks: + if max_tasks and max_tasks_jitter: + self.stdout.write( + f"Max tasks: {max_tasks} ({options['max_tasks']} + jitter " + f"{jitter} of up to {max_tasks_jitter})" + ) + elif max_tasks: self.stdout.write(f"Max tasks: {max_tasks}") if graceful: timeout_label = ( @@ -275,6 +302,7 @@ def handle(self, *args, **options): "queue_name": queue_name, "continuous": continuous, "threads": threads, + "max_tasks": max_tasks, **worker_log_fields(), }, ) diff --git a/tests/test_commands.py b/tests/test_commands.py index bfc4510..e36a606 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -1,6 +1,8 @@ """Tests for management commands.""" import os +import random +import re import signal import sqlite3 import subprocess @@ -156,6 +158,94 @@ def test_run_database_tasks_no_tasks(self): assert "No more tasks to process" in out.getvalue() +@pytest.mark.django_db +class TestRunDatabaseTasksMaxTasksJitter: + """--max-tasks-jitter: a random addition to --max-tasks, drawn per worker.""" + + def run(self, drawn=0, **options): + out = StringIO() + with patch("random.SystemRandom.randint", return_value=drawn) as randint: + call_command("run_database_tasks", stdout=out, **options) + return out.getvalue(), randint + + def test_the_drawn_value_is_added_to_max_tasks(self): + for i in range(5): + simple_task.enqueue(i, i) + + output, randint = self.run(drawn=2, max_tasks=1, max_tasks_jitter=3) + + randint.assert_called_once_with(0, 3) + assert "Max tasks: 3 (1 + jitter 2 of up to 3)" in output + assert "Reached max tasks limit: 3" in output + assert ( + DatabaseTask.objects.filter(status=TaskResultStatus.SUCCESSFUL).count() == 3 + ) + assert DatabaseTask.objects.filter(status=TaskResultStatus.READY).count() == 2 + + def test_a_draw_of_zero_leaves_max_tasks_as_it_is(self): + for i in range(3): + simple_task.enqueue(i, i) + + output, _ = self.run(drawn=0, max_tasks=2, max_tasks_jitter=3) + + assert "Max tasks: 2 (2 + jitter 0 of up to 3)" in output + assert ( + DatabaseTask.objects.filter(status=TaskResultStatus.SUCCESSFUL).count() == 2 + ) + + def test_the_default_is_no_jitter(self): + """Without the option --max-tasks is what it has always been.""" + output, randint = self.run(max_tasks=2) + + randint.assert_called_once_with(0, 0) + assert "Max tasks: 2\n" in output + assert "jitter" not in output + + def test_jitter_without_max_tasks_is_ignored(self): + """Unlimited stays unlimited, as with gunicorn's max_requests_jitter.""" + for i in range(3): + simple_task.enqueue(i, i) + + output, randint = self.run(drawn=1, max_tasks_jitter=5) + + randint.assert_not_called() + assert "Max tasks" not in output + assert ( + DatabaseTask.objects.filter(status=TaskResultStatus.SUCCESSFUL).count() == 3 + ) + + def test_a_seeded_random_module_does_not_make_the_draws_alike(self): + """A project that seeds random at import must not undo the jitter.""" + drawn = [] + state = random.getstate() + try: + for _ in range(2): + random.seed(0) + out = StringIO() + call_command( + "run_database_tasks", + max_tasks=1, + max_tasks_jitter=10**9, + stdout=out, + ) + drawn.append(re.search(r"Max tasks: (\d+)", out.getvalue())[1]) + finally: + random.setstate(state) + + assert drawn[0] != drawn[1] + + def test_jitter_must_not_be_negative(self): + with pytest.raises( + CommandError, match="--max-tasks-jitter must not be negative" + ): + call_command( + "run_database_tasks", + max_tasks=1, + max_tasks_jitter=-1, + stdout=StringIO(), + ) + + @posix_signals @pytest.mark.django_db class TestRunDatabaseTasksGracefulShutdown: @@ -1375,6 +1465,12 @@ def test_the_workers_exit_codes_become_the_commands(self): self.run_workers(code=1, workers=2, failed_exit_code=1) assert exc_info.value.code == 1 + def test_the_jitter_is_drawn_by_each_worker_not_the_supervisor(self): + with patch("random.SystemRandom.randint") as randint: + self.run_workers(workers=2, max_tasks=10, max_tasks_jitter=5) + + randint.assert_not_called() + def test_a_normal_exit_of_every_worker_exits_normally(self): self.run_workers(code=0, workers=2, empty_exit_code=4, failed_exit_code=1) diff --git a/tests/test_logging.py b/tests/test_logging.py index 199d313..c8d6e3c 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -275,6 +275,20 @@ def test_start_is_logged_with_fields(self, task_logs): assert record.source == "db" assert record.queue_name == "emails" assert record.continuous is False + assert record.max_tasks == 0 + + def test_start_carries_the_max_tasks_with_the_jitter_drawn(self, task_logs): + """The drawn jitter is otherwise invisible to a fleet's logs.""" + with patch("random.SystemRandom.randint", return_value=7): + call_command( + "run_database_tasks", + max_tasks=100, + max_tasks_jitter=10, + stdout=StringIO(), + ) + + (record,) = records_matching(task_logs, "Worker started") + assert record.max_tasks == 107 def test_a_failed_receive_is_logged_with_the_worker_fields(self, task_logs): class BrokenBroker(FakePullBroker):