From 3b4966243d372f4be255fa61766563832a9d8660 Mon Sep 17 00:00:00 2001 From: Shinya Okano Date: Fri, 2 Oct 2026 14:11:28 +0900 Subject: [PATCH] Add --threads to run several tasks at once inside a worker The worker stays one process with one polling thread, which still fetches, receives from the broker and handles the signals; the task it found is handed to one of M executor threads instead of being run inline. The claim is made by the polling thread with the executor's worker id, so the next fetch never hands out the same row, and a broker receive asks for no more messages than there are idle threads. Each thread records its own id on the tasks it runs, has a database connection of its own and closes old connections after every task. --max-tasks counts the tasks handed out, failures are counted across the threads under a lock, and on SIGTERM the tasks in the threads finish before the worker exits. Several threads are refused on SQLite, as several workers are: claims from the broker path and finishing writes from several threads collide on the database lock. The default of 1 leaves the loop as it was. Refs #25 --- CHANGELOG.md | 17 ++ README.md | 68 ++++- .../locale/es/LC_MESSAGES/django.mo | Bin 9653 -> 10101 bytes .../locale/es/LC_MESSAGES/django.po | 41 ++- .../locale/ja/LC_MESSAGES/django.mo | Bin 10088 -> 10580 bytes .../locale/ja/LC_MESSAGES/django.po | 47 +-- .../locale/pt_BR/LC_MESSAGES/django.mo | Bin 9575 -> 10021 bytes .../locale/pt_BR/LC_MESSAGES/django.po | 47 +-- .../locale/zh_Hans/LC_MESSAGES/django.mo | Bin 8908 -> 9304 bytes .../locale/zh_Hans/LC_MESSAGES/django.po | 43 +-- .../management/commands/run_database_tasks.py | 278 +++++++++++++++--- django_database_task/threads.py | 156 ++++++++++ tests/postgres/test_threads.py | 179 +++++++++++ tests/tasks.py | 14 + tests/test_commands.py | 36 ++- tests/test_threads.py | 102 +++++++ 16 files changed, 899 insertions(+), 129 deletions(-) create mode 100644 django_database_task/threads.py create mode 100644 tests/postgres/test_threads.py create mode 100644 tests/test_threads.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 33371e7..44846b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,23 @@ ### Added +- **`--threads M` on `run_database_tasks`.** The worker runs M tasks at + once in executor threads while staying one process with one polling + thread, which still does the fetching, the broker receives and the signal + handling. The polling load stays at one fetch per `--interval`, a broker + receive asks for no more messages than there are idle threads, and the + claim is made by the polling thread with the executor's worker id, so the + next fetch never hands out the same row. Each thread records its own id + (`-t1` to `-tM`) on the tasks it runs, has a database + connection of its own, and calls `close_old_connections()` after every + task. `--max-tasks` counts the tasks handed out, failures are counted + across the threads, and on `SIGTERM` the tasks in the threads finish + before the worker exits. The task code must be thread-safe, and the option + only helps I/O-bound tasks; the README's *`--threads M`* and *Trade-offs* + sections say what it saves (memory, not connections) and what it costs. + Several threads are refused on SQLite, as several workers are. The default + is 1, with the loop unchanged, so existing projects need no changes. + ([#25](https://github.com/tokibito/django-database-task/issues/25)) - **`claim_task()` and `run_claimed_task()` on the backend.** The two steps of `DatabaseTaskBackend.run_task()`, which now calls one and then the other, so that a worker can claim a task in the thread that fetched it and diff --git a/README.md b/README.md index 41654b7..7ba4c96 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ A database-backed task queue backend for Django's built-in task framework. - **Django Admin integration** - View and manage tasks from the admin interface - **Async support** - Supports async task functions - **Graceful shutdown** - Workers finish the running task before exiting on `SIGTERM` -- **Parallel workers** - `--workers N` keeps N worker processes running from one command, or run one per systemd unit / Kubernetes replica as before +- **Parallel workers** - `--workers N` keeps N worker processes running from one command, `--threads M` runs M tasks at once inside a worker, or run one process per systemd unit / Kubernetes replica as before - **Crash recovery** - Tasks stranded in `RUNNING` by a killed worker are found and requeued - **Monitoring** - Counts per status and the age of the oldest pending task, from Python or `/tasks/status/` - **Job scheduler friendly** - Opt-in exit codes that tell an idle run from a failed one, plus structured log fields for JP1 / Hinemos / cron / systemd timers @@ -260,6 +260,7 @@ python manage.py run_database_tasks [options] | `--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) | | `--workers` | Number of worker processes to run from this command (default: 1). See [Running several workers](#running-several-workers) | +| `--threads` | Number of threads running tasks inside the worker (default: 1). The task code must be thread-safe. See [`--threads M`](#--threads-m) | | `--shutdown-timeout` | Maximum seconds to wait for the running task after `SIGTERM`/`SIGINT` before forcing exit (0=wait indefinitely, default: 0) | | `--no-graceful-shutdown` | Do not install signal handlers (terminate immediately, even while a task is running) | | `--empty-exit-code` | Exit with this code when no task was processed (0=exit normally, default: 0) | @@ -699,12 +700,13 @@ with GracefulShutdown(timeout=50) as shutdown: `run_database_tasks` is one process running one task at a time. To run more tasks at once, run more of them: every worker claims its own tasks with `SELECT FOR UPDATE SKIP LOCKED`, so workers on one host or on many never run -the same task. There are two ways to keep N of them running. +the same task. There are three tiers, and they combine. -| | The processes are kept running by | Reach for it when | +| | Unit of parallelism | Reach for it when | |---|---|---| -| A process manager | systemd template units, Kubernetes replicas, supervisord `numprocs` | You have one. It restarts, logs and health-checks the workers, and the command's default (one process, with the signals and exit codes described above) is what it expects | -| `--workers N` | The command itself | One host without a process manager, or one unit or container that should hold N workers: a small supervisor inside the command keeps N worker processes running | +| A process manager | Processes, kept running by systemd template units, Kubernetes replicas, supervisord `numprocs` | You have one. It restarts, logs and health-checks the workers, and the command's default (one process, with the signals and exit codes described above) is what it expects | +| `--workers N` | Processes, kept running by the command itself | One host without a process manager, or one unit or container that should hold N workers: a small supervisor inside the command keeps N worker processes running | +| `--threads M` | Threads inside one worker process | Memory is tight and the tasks are I/O-bound and written thread-safe: M tasks in progress for the footprint of one process, with one poller | ### `--workers N` @@ -767,11 +769,52 @@ Shutdown complete: every worker exited. [requeue_stale_database_tasks](#recovering-tasks-left-in-running-status) is for. +### `--threads M` + +```bash +python manage.py run_database_tasks --continuous --threads 8 +``` + +The worker stays one process with one polling thread, which is still the +one that fetches, receives from the broker and handles the signals. What +changes is that the task it found is handed to one of M executor threads +instead of being run inline, so M tasks are in progress at once. The polling +load stays at one fetch per `--interval`; a broker receive asks for no more +messages than there are idle threads, up to `--max-messages`. + +- **Thread safety is on you.** The M tasks share one interpreter, one set of + module globals and whatever your task code caches. A task that is not + thread-safe, that calls `os._exit()`, or that crashes the interpreter takes + the other M-1 with it. `--workers` asks nothing of the task code; this + option does, and that is the trade it makes explicit. +- **I/O-bound only.** The GIL serialises Python bytecode across the threads, + so CPU-bound tasks gain nothing from M; HTTP calls, database-heavy tasks + and anything that waits do. For CPU-bound work use `--workers`. +- **Memory, not connections.** Threads share the process, so M tasks cost + roughly one worker's footprint rather than M. They do not save database + connections: Django keeps one per thread, so a worker with M threads holds + M connections, plus the polling thread's, plus the `LISTEN` connection with + the PostgreSQL broker. Each executor calls `close_old_connections()` after + every task, the way a request does, so `CONN_MAX_AGE` applies. +- **Worker ids.** Each thread records its own id on the tasks it runs, + `-t1` to `-tM`, so `worker_ids` and the log fields + say which thread ran what. The claim itself is made by the polling thread, + with the executor's id, so the next fetch never hands out the same row. +- **Shutdown and `--max-tasks`** work as with one thread: on `SIGTERM` no + more tasks are handed out and the ones in the threads finish, within + `--shutdown-timeout` if set; `--max-tasks` counts the tasks handed out, so + no more than that many are ever started. +- **Combining.** `--workers N --threads M` is N processes with M threads + each: N pollers, N×M tasks in progress, process isolation between the N. +- **SQLite** is refused above one thread, as with `--workers`: without row + locks, claims and finishing writes from several threads collide on the + database lock. + ### Trade-offs Processes are the simple, robust unit of parallelism, and they are not free. What `--workers N` buys and what it costs, so you can pick N with your eyes -open: +open, and where `--threads M` stands against it: - **Isolation.** Every worker is a process of its own. A task that crashes the interpreter, leaks until the OOM killer steps in, or blocks forever takes @@ -818,13 +861,12 @@ open: running `--workers N`, is a reasonable middle: the manager handles the machine-level concerns and the supervisor the fan-out. - **SQLite** has no row locks, so the command refuses `--workers` above 1 - there. Use PostgreSQL, MySQL or MariaDB. - -Threads inside a worker, the memory-lean option for I/O-bound tasks that are -written thread-safe, are not part of this release; the design is tracked in -[#25](https://github.com/tokibito/django-database-task/issues/25). On this -backend they would save memory but not database connections, since each -thread needs a connection of its own. + there, and `--threads` above 1 too. Use PostgreSQL, MySQL or MariaDB. +- **Threads instead.** [`--threads M`](#--threads-m) trades the isolation + and the CPU scaling above for memory: M tasks in progress for one + process's footprint, as long as the tasks are I/O-bound and thread-safe. + It does not reduce database connections, and one bad task can take the + whole worker down. When memory is not the constraint, prefer processes. ## Running from a job scheduler diff --git a/django_database_task/locale/es/LC_MESSAGES/django.mo b/django_database_task/locale/es/LC_MESSAGES/django.mo index 4198addf3f333f33dc3a35b162f3f2584463f2db..c19ee004266032c269248e6ff4b8ae1688d766b2 100644 GIT binary patch delta 1897 zcmYk+Uu+ab9Ki7(+Ex%sOR*l6e^abN1^SOl;o4d#Ez)WPW0e<;nci*NO_$yCcCW3* zI7cnY7! z@!LiA;W_NWs^ucn_%Iqgj~eJ@Y{Po8N%MXgqH&FZS5YS(s}#8tKSkD(?{NqIlxSAQ z0S_S=h>IG)^SA-OO~(JmyXdd1il1x5v-G?03;YXT=Kb)CbV{Wy(1_yHcmtLWnaZj#Y)ycaK^?%0i_-?ApTD|mwONj!t=c&px}uTU@F*T_$D zy^j6Ak4B1vx`@3viS_L3K|GHAxQNHGnNMUZeu59+mBjY-B9GD!v4%_+u#f(Q4e?!C zzcJp~cGP+Mu?LS9Y4D6J;NAEmPT^7`vIVD6FVQLFCs{z&mP`06-da!g_y@kFZi;v2 z4HnbHK0-a3MI69iupHA3@evgF(0GS|$53Bz6<3f!>Yn(_8`x*=l{6>zBBvpbBfBmh z>YaHJ`K0Cb_&ajAZj4P zNPN;W^LE+h{$^SX2x*tDY7_py=wY!nrSC_tuHLRLD(6=EucM)*XDZ7(>e^{(;iD)8 zuJ6_z?3j6~d~LCpmfjfsj=HH@wo^N)dIR+q@1$xby3hNmT6C)pP&IY^g!HU6{e4s| zdMmZ)rD&x-s2$2)DkmU&sQsGWE*d+iyQw!z8x8%Swe-!^RjgT7wBdMYT;D`vZfK5J z|5;jj$7Xdfbh8d!$4mzKaVOtohG<93Fqwc~jugVk4C@m*VS8vtocNLWaT7+a=b2pK zd7%mXjAL?6o>e0!KViL2UF_WNl=JZRoJR?>m{u4SW-f7&l^pN#AVA zIwQ8=MQPL8IJdQG^{Sb4s!7>1)W5b tjpv)lb$pgCm|Q-{xx^Xx&XhYp=aa5yO(|PqA@d-dsjsh{+i8mB{{Z~#ANv3R delta 1485 zcmZA1OGs2v9LMp0rk0KG@|B~`RK6`=<>VtDIX*JTiXb8(vdBec5rkR9l+Z?k7UIGP ziINuSCRFGF;=)}nvPEso7PacJG+fl!_c!;p=`d$L=N#|c`#=BlzpfscJses|jUF{x zC*4c$i84EjkD~da`C`p7(2wz0i+R|N0la{fIF9A`4x7+6nO%ft)bFQIzgxgv_!U#l zLKe5#Y#R$+ypNT559g3rEEwnfume+=_n{95aULJy89cqktQFs21Fqv;Z16bO&!GbS zj8&MzZt8fx1sVKdA%yzj?F6%Qe2v`0marO^U6T`?fP=^m>=Y`1OPGs`?)rD!&OGeS zvy#l-GcUtoTt^?zx8Y>73YC$bfrTmunA_ed|ZvXaIHNA=HB&xa(u68%?pXUCFnfzxT$i5aWUK z4~KCApW`d+&U9)hj;E{W;*lk*#1ibvBL6E4uCp+JCpfTfT*Nl?lIJn(#Zvr@jabC1 zBKz2JfDz{t?`wE znLY{!Yw)r3I~n+h+EEyFB*i<-4q*kx;7wGq-od9hg1T=gz|-T2{3PnE&yy!zc*S)H z^)`$mS+yzDjuwzt*gm*_UqMdIR*__yC*P@+eAM;jNIhE%Quub!(SFjA%H{b``9sbuTu<#ZK;Dpp(Rq_@%4l&TK8UK6ELhpg=%q^m_a xMBZ;zaJxHiMm8ELj-a*|U{XcjPyg2{8R$h;Yn?8R&5WB4Cncmzhy4RFe*t?7h+hBz diff --git a/django_database_task/locale/es/LC_MESSAGES/django.po b/django_database_task/locale/es/LC_MESSAGES/django.po index f78b41c..c61c524 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-09-30 17:27+0000\n" +"POT-Creation-Date: 2026-10-02 05:07+0000\n" "PO-Revision-Date: 2026-09-30 00:00+0000\n" "Last-Translator: Shinya Okano \n" "Language-Team: Spanish\n" @@ -135,31 +135,31 @@ 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:55 +#: management/commands/run_database_tasks.py:63 msgid "Execute tasks queued in the database" msgstr "Ejecuta las tareas encoladas en la base de datos" -#: management/commands/run_database_tasks.py:62 +#: management/commands/run_database_tasks.py:70 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:68 +#: management/commands/run_database_tasks.py:76 msgid "Backend name (default: default)" msgstr "Nombre del backend (predeterminado: default)" -#: management/commands/run_database_tasks.py:73 +#: management/commands/run_database_tasks.py:81 msgid "Continuous mode (keep polling even when no tasks)" msgstr "Modo continuo (sigue sondeando aunque no haya tareas)" -#: management/commands/run_database_tasks.py:79 +#: management/commands/run_database_tasks.py:87 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:85 +#: management/commands/run_database_tasks.py:93 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:92 +#: management/commands/run_database_tasks.py:100 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 +171,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:104 +#: management/commands/run_database_tasks.py:112 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 +180,12 @@ msgstr "" "Sustituye a --interval como espera inactiva al recibir de un broker " "(predeterminado: 20)" -#: management/commands/run_database_tasks.py:114 +#: management/commands/run_database_tasks.py:122 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:123 +#: management/commands/run_database_tasks.py:131 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,7 +196,18 @@ msgstr "" "se reemplaza; sin él, el comando termina cuando todos los workers lo hayan " "hecho (predeterminado: 1)" -#: management/commands/run_database_tasks.py:134 +#: management/commands/run_database_tasks.py:143 +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 " +"(predeterminado: 1)" + +#: management/commands/run_database_tasks.py:154 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -205,7 +216,7 @@ msgstr "" "SIGINT antes de forzar la salida (0=esperar indefinidamente, predeterminado: " "0)" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:163 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" @@ -213,7 +224,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:153 +#: management/commands/run_database_tasks.py:173 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 +233,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:164 +#: management/commands/run_database_tasks.py:184 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 130d80b705eee79ecc78f97c88edd5c21084bbcb..532ff556aa722f44008c53d85441c3131814c059 100644 GIT binary patch delta 1957 zcmY+^eN0tl9Ki9P4nZIVEd`|Fu@{9DW8~Fo4%!Jr>}jSdRNKfMd8Bf55GnohH(b z52KEsLLGM+Z^7SjrAS=zmWo_Q!$$1I8tletWG>Q_?hV|DE2wv&!Nd45j^LBno*{BC zeu>qXl_}!j188s>b)pxr6xXtvO5QI`6fV>70_woNWg@Hb3nZ4D!!rEY*DUuY+=Q$^ zf~XUC25-dg{PsU^4fW+&UcVxoqF#+(<0X8a_sdt=-VArLyb|gMkX@8_aSNVAJ&p6o z}wNV#v0tg+Xip|wSUnKB6nefZx233{c|j1r3GAsTI!7j-Vq%`-MRO0 zJAUGu5jWn78ht|;r9(IB3ohYWEaQjGkVlZ%5>gk5^k6BDu&f!{&))3yALARllllyD z&m`?u7J)gaNB0!!-HFF2{6gVHEJkOYNQ8jnsApPG=$+L%)D={t4y^aJ{pSv9;FnPQ z4g2k5s9XJ+|NI;O`8ls%Dd+tT7ySpBMc#~aeG5?oR-oSAI%IybwSPEeecfhCdLVkL zOUawaSEmhB*rJ8+N2A<9t|sg4;Wi2vLn`~1Eh*2tosuTKpn5s8$+wZq`=3~nAJ<#P zWm@=-wxA~6t}3$L;EiP6Z{6#9@_l4YdS=_n>16$c^z8H#)BjYO7JJCu7$D#8*K3dy zk!|EUU0wx+GO}KZE2)%%Zi%MaiM-U@^tcsmkD6fE#M**Uv&RZQO{vqiTC_15Y_X}@ zrYq9fZg*}nO_XDdFwGI3?AaHMncdn$2NXxGR@>|1bvMyiFcdO-BcV{#M8eIs*=u(a zHD-72w?b9Mwwl{ag*Vu=Sy5w|7At1$wxTxi!(qEQ7L0^Vq}2pt1n3Hz!WO&L+82sd znn2OShOB`V@#MIhoN^N<_?>;W+jZV{oqpFDn;SgIubY^3lcR1j=_U{Bqr@p~NsPOR zqqAeL&ky&y&T%)<;~q-<_ee)gXiObK&#NLNK3r%YTzCsZdS*|>yXW6>T!%nf*FeeF z7CMc%$q6^XB*tf_7|{3Q`!DnR@8XT?47koau5-c!%88Q_C`MrD}#%%T=XXtW5DS|k)$R8Y|-?f;xJZ5o{Oo0;#M`QEwty5-X$|GR|nAtN=B z)5vXMW~cE^I6ov$q}gW7#AqzV3~ay>Y{xKEg6w!E0EyKJ92E52~;OSpylva6@W zn!Td##eQ5t58t=`II{vAM^?`kupB?2b|I5#6SU>v2o_-~&f_wEK^DP26Ag2)U#P%W zZ!{xji$P+rG}JhGsPB}ZUkw!$7{d-a+mO|_3#fsvpa#B<`p|%DA4Ux{=GwDU z>sFb~B7YW_XvUvq^P>s-5?TLa6gp_A$7jxH*8d3g7QBRGn2brBBpuRxtifi~77RG= z;!EmN_!hhAq%(9X#ViZYpx*DrTpUhe{l8LpOhXq=uwh5Bfc34$9=wImQ61X&dMS20 z1K3LaHx@I~mUL!Jy_bVP7n?<`+(&G}AI|cO(1d#Zt}u$1dEo)-16~jNj_s(!*oXAB zQ7qy^0i2`$otO094Aa|(i^#dM)NKTVd8jSxLIu=^-*60fp?`?Wc##30p!TvjC$vXC zROE+I9Zxtr-SZ*Tz~iX*rd<0BYKh;t=S%MSPgf6P8TC9G>9^A|C@5mDvn=$$no*~= z6^YL(1}DOE>vvP4A-J15qMF11N$a6|qpQHjo zh=x@hJRgzjuco9Er~6YzR`Qa4\n" "Language-Team: Japanese\n" @@ -129,31 +129,31 @@ msgstr "一度に処理するタスク数 (デフォルト: 1000)" msgid "Show what would happen without changing anything" msgstr "何も変更せずに、何が行われるかを表示" -#: management/commands/run_database_tasks.py:55 +#: management/commands/run_database_tasks.py:63 msgid "Execute tasks queued in the database" msgstr "データベースのキューに登録されたタスクを実行" -#: management/commands/run_database_tasks.py:62 +#: management/commands/run_database_tasks.py:70 msgid "Queue name to process (all queues if not specified)" msgstr "処理するキュー名 (指定しない場合はすべてのキュー)" -#: management/commands/run_database_tasks.py:68 +#: management/commands/run_database_tasks.py:76 msgid "Backend name (default: default)" msgstr "バックエンド名 (デフォルト: default)" -#: management/commands/run_database_tasks.py:73 +#: management/commands/run_database_tasks.py:81 msgid "Continuous mode (keep polling even when no tasks)" msgstr "継続モード (タスクがなくてもポーリングを続ける)" -#: management/commands/run_database_tasks.py:79 +#: management/commands/run_database_tasks.py:87 msgid "Polling interval in seconds for continuous mode (default: 5)" msgstr "継続モードのポーリング間隔（秒）(デフォルト: 5)" -#: management/commands/run_database_tasks.py:85 +#: management/commands/run_database_tasks.py:93 msgid "Maximum number of tasks to process (0=unlimited, default: 0)" msgstr "処理する最大タスク数 (0=無制限、デフォルト: 0)" -#: management/commands/run_database_tasks.py:92 +#: management/commands/run_database_tasks.py:100 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 +165,7 @@ msgstr "" "バックする。'auto' (デフォルト) は、受信できるブローカーがバックエンドにあれ" "ば 'both'、なければ 'db' になる" -#: management/commands/run_database_tasks.py:104 +#: management/commands/run_database_tasks.py:112 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,21 +173,32 @@ msgstr "" "再確認するまでブローカーのメッセージを待つ秒数。ブローカーから受信する場合、" "アイドル時の待機として --interval の代わりに使われる (デフォルト: 20)" -#: management/commands/run_database_tasks.py:114 +#: management/commands/run_database_tasks.py:122 msgid "Maximum number of broker messages to receive at a time (default: 1)" msgstr "一度に受信するブローカーメッセージの最大数 (デフォルト: 1)" -#: management/commands/run_database_tasks.py:123 +#: management/commands/run_database_tasks.py:131 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; " "without it the command exits when every worker has (default: 1)" msgstr "" -"このコマンドから起動するワーカープロセスの数。各ワーカーは --workers を除いたこのコマンドの" -"コピー。--continuous 指定時は終了したワーカーを再起動し、未指定時は全ワーカーが終了する" -"とコマンドも終了する (デフォルト: 1)" +"このコマンドから起動するワーカープロセスの数。各ワーカーは --workers を除いた" +"このコマンドのコピー。--continuous 指定時は終了したワーカーを再起動し、未指定" +"時は全ワーカーが終了するとコマンドも終了する (デフォルト: 1)" -#: management/commands/run_database_tasks.py:134 +#: management/commands/run_database_tasks.py:143 +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)" + +#: management/commands/run_database_tasks.py:154 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -195,7 +206,7 @@ msgstr "" "SIGTERM/SIGINTを受信してから強制終了するまで、実行中のタスクを待つ最大秒数 " "(0=無期限に待つ、デフォルト: 0)" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:163 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" @@ -203,7 +214,7 @@ msgstr "" "SIGTERM/SIGINTハンドラーを設定しない。タスクの実行中でもプロセスは即座に終了" "する" -#: management/commands/run_database_tasks.py:153 +#: management/commands/run_database_tasks.py:173 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)" @@ -211,7 +222,7 @@ msgstr "" "タスクを1件も処理しなかった場合にこの終了コードで終了する。ジョブスケジュー" "ラーが空振りの実行を区別できるようにする (0=通常どおり終了、デフォルト: 0)" -#: management/commands/run_database_tasks.py:164 +#: management/commands/run_database_tasks.py:184 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 7dd4daa3cbfa7d6f3e876171fb7ea0ff535e13f0..275e84e367352efaabc34d9aa5bf5111315b1067 100644 GIT binary patch delta 1912 zcmY+@YiJx*6u|Lg$-|noO=H_+ZM4^F)5IoiwpCL%iAiizjfn9>un3lVvokScc4xLb zlZT=#E%bwr;)kGA(SoRmR-_ORA_)k=FDgZ)ps2MfNKr}@AD}I@{?E?%8o2r0dw1@c zbMHMld*`cnm%gg6ct&^zX-za|nTW+VEBM3HQ6h?7g@tZ8$N~ocoTN6`$;&Tz3@_n&JVq9O;ycJ;$lR6jRV|`!;3Fh=`2xvFzDJGoJC^iA6`a3L)F3gX zA+Z&?1?fT!v>Q1j8AAQg{mJ|&YMcYf{B$yZDDep1cbUjhO!3~&Yq=h*UB~%vq|?NC z4`459Q5{CTiUr(*=Wz=AsU(AQs73TSrtuPLoOC1iio^IKPU0)LhOcTZ&7rO-hnS*AAW~=v#08g*xD+*wGy5vnmu{n{+ty5aX!^9%x@h`8+eW*YrWp>>_R#d9aLa|HN2lAT9?HAgh59Uu zky}wK`SxVok9v=Nv;p1TPC7eiT5W$n+v(`p>$7_{Rn@q>WW({$c)p3oywK#W|1iCx zYaN{oJ;!C}nyH{T?iRP2Vfqm(Og7+`{6rX;5q(1!w1;-ojbDjhH(})Ea;6aEa-j+Q ztZNExk*JYdoV2+vc3V>UF_I5x5)Hgt*a`))S!0^bCq#)}9r<(n4A9kmm= zC}YyCvs+fxte(j<)Rywb4I^s~Ojtki97|&5tm92Mt{)Wx;{^ZgVaQ|`XC3cBCi!M$ zi^L2~q3CA4(3+4G|F@r^WTB=MdH#gePDm2gC+I}p?lY6_Bkc0utp!~sYaN@!F_D{R uYauHO#l_h|(JO~=rx#xih?O^i8L@trbV%GfL1ZPURspK}lQp8G%l^S?eg@TDvC(H(co zXdU!idS9&BQM?n!56u&Awi&&+0qZdz+t80Eum(r53g2Nf+FG;I5I{Zu2=&|y7T{-e zn}sZ4o!KTPa`6V%;5D2=VzFRC^ucya<@zLg@DfhrZ9Ia#iDoVM0vmA^uVCZ)=>1cu zKtG`mQ`t=e@3$a>KTL#B4_;0(%fRQz8a9iyIOmwMF&c0X*@5+-0yu+P@U=7l9kaP! zajvtH&E9iej>EW$9^P-mDQ49;j%3f~a1YL-PQlBz3EGNq6n&V3v$%qbND^#;XxM}O zLIu7i&5W3BBNBt&A1i2)5(7;gY!(Nn4Y1IUf^!(ei|D}xRPiKHQ8nm8 zE!2-*9KyT!5XZ1RGg?CnsQZ3k9=eI97^}0$|2GDQnYfNE99Rdw#QYK$2ubsA9T@E}TFW*;C}0Z3gu@e?YCbh|0h+DgYPJ>i!I*_DZdQfg-F% z6`|j`p%WFzAtXL)3O|l3ZmndblOy~{iB{nMjTA(_8@+Z1eK(zRiG-ys)Q9t2{;XO? zY6_Ziiv(ASyTWH(IibCbv~7jjNLMSTH_)r-W%QkNMW}*lrK{;5M>}1q)Al*u$gcLG zT5Oca_g&3cRjwAGv(ZSA1hqX?nU7va|JPLV%Dr04WJ!Ex!sJS_D|Paxx5o7slX!;U 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 69700ab..8ca5a27 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-09-30 17:27+0000\n" +"POT-Creation-Date: 2026-10-02 05:07+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,31 @@ 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:55 +#: management/commands/run_database_tasks.py:63 msgid "Execute tasks queued in the database" msgstr "Executa as tarefas enfileiradas no banco de dados" -#: management/commands/run_database_tasks.py:62 +#: management/commands/run_database_tasks.py:70 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:68 +#: management/commands/run_database_tasks.py:76 msgid "Backend name (default: default)" msgstr "Nome do backend (padrão: default)" -#: management/commands/run_database_tasks.py:73 +#: management/commands/run_database_tasks.py:81 msgid "Continuous mode (keep polling even when no tasks)" msgstr "Modo contínuo (continua consultando mesmo sem tarefas)" -#: management/commands/run_database_tasks.py:79 +#: management/commands/run_database_tasks.py:87 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:85 +#: management/commands/run_database_tasks.py:93 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:92 +#: management/commands/run_database_tasks.py:100 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 +168,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:104 +#: management/commands/run_database_tasks.py:112 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,22 +176,33 @@ 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:114 +#: management/commands/run_database_tasks.py:122 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:123 +#: management/commands/run_database_tasks.py:131 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; " "without it the command exits when every worker has (default: 1)" msgstr "" -"Executa esta quantidade de processos worker a partir de um único comando, cada " -"um uma cópia deste comando sem --workers. Com --continuous, um worker que sai " -"é substituído; sem ele, o comando termina quando todos os workers terminarem " +"Executa esta quantidade de processos worker a partir de um único comando, " +"cada um uma cópia deste comando sem --workers. Com --continuous, um worker " +"que sai é substituído; sem ele, o comando termina quando todos os workers " +"terminarem (padrão: 1)" + +#: management/commands/run_database_tasks.py:143 +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 "" +"Executa as tarefas em esta quantidade de threads dentro do worker. O código " +"das tarefas precisa ser thread-safe. O worker continua consultando uma vez " +"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:134 +#: management/commands/run_database_tasks.py:154 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -199,7 +210,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:143 +#: management/commands/run_database_tasks.py:163 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" @@ -207,7 +218,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:153 +#: management/commands/run_database_tasks.py:173 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)" @@ -216,7 +227,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:164 +#: management/commands/run_database_tasks.py:184 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 f77b2be7891fa4ae900a61acaf253f95a64f3948..1920335620dbf67f9475b39947231f26210e0ffb 100644 GIT binary patch delta 1878 zcmYk+e@xV69Ki8McVcP+YN;sVLqA3&f~ci|hNcv3ZMDhbpMHJizQm35PPs#HW1X@f zA;~7V5E+GOBW7Tu`GY$;H`m(CT(kaY>&J54_qf|wXFvX^HLdsOzTY;V?YYKbGPcmf+7=f!TAF zYQ*PJ+DA~@eTxs_pSW15kjlA7srx7>!XvmHJMcO(7gaOQZdi+p$hV-zF1(DR_zE`8 zS84}-iCZvhfl`O?S=4wPWuhrukNK>ojPX?siQ5z$MQPZxP^qPO73oXe#0@y(s59*e z*B~oU9+U~RVIKbAl>dV($Y*BR^$IaUehc2fzplH7xQcum%4HgG%A<}q@O8>9>_+!;`!1zWlm8uM=a%HK z{{7|%#tFt3uWj0 zC<|?KJRc&#Zm3(18F^et^0_Dz+lTV~1TwZ7N7;d^_!<6)NgQ0wi@;88515^AU)my+ zqj?pd#MhkshbR*YeN1AE#3af<9?s*Yd_ZYvJZM*JaBM~{u4;EY=9Hg8sXy-I&pY3* zpiKB1lsgl5zTdL*W$Lz5VJ-(LGg^VtQGt^$b}UC3xC&=sMeneypt3}25}9rKsuttj zK@N$PrQeVA{uohi6z7xv*WtWXS?|I*n{qailE9-#Zv>B0Jxbiv`_i11Avs(wP5K>e zMG104%ZYMhi->YjWQNtmr->4{xT=aMOP43aaiyQw4wNw4CVe)gq@Qu}+wo!IHe#hL zZ!?JvM7cfxhxH`nK}*;c}QAswWJuJiL{PjeuscS{Xvur^@Sn|2kux()5;dr&W`E6kwo-F?9yTD^g!(5==5Zi@d*b6q=`i{3r^$i7#MD?~ z{8X&s;=%aY;Z)?L{mt$oKG>CbbJV)<3BAORpNsd5S(Ckq=urIaGpWeoWVk;ub~+Uq OC{0DWCbIITT>k;kJ6q%c delta 1485 zcmZA1Nk~;u9LMp$rj`v(r8z%`Oj6U-^7YKAd=9iJ$W0|h5*ai=tB|NiQ9%;5kcdj8 z#GR3Zw5UW}6fH7HDG0KpVI$R|1Q&wZ_xJ8;(}BOAbI!Z>+%x=-A6mzoLZ6eOZW*PO z$Rye$&5mGi6n`i=F=p#A7gu8deb|hCJci}ihh_K{>(Q2*oqz_^`wvm?y}%9l9h1yL z7Q4c19UYl?9n0|wjv{lhV669I3ntR;#2oCxF}#b1@!(3cMtqJnxP%w6W|jB+C~Bh9 zSb>SGrkd|tkjerbA=HcK*O;Z?Gh_^#z)GBQCd7FY4k9bC{iq3?#B3aQ{Xa0B_M&U2 z#hXpiF2QbG!W_PD-3ew}a1hx&o5EWBhB}2@md&JXBlckhX5a)a;xA+q>?gBf4K|0G z__DQT%-Q0QIanrYoMO~>{1{S49Tmo~-OhGo_w6)lptGofFQGnk)Aip$4K(2T2VMUY z=NRsbFnf*40bjB=jvw=A2b0C>g7NIYUi4<5Ia)Y|Cdy{={Syi)5utSid8s|U0BXR>cv4U##hc+ z+(p~RkH4`4%dw5Cr01`qcCH@-IOLo`)@b=5mZM4l+i(wRW&^H&9QlTQMeV=>&R{~e z*)^O;edw~!OVazOBYB5)_|dh~bG(V<;v@P?QQr-HqVk2xEb7Jgx!!|Q&KV?8ZO$3P z!D&S)sOR%ty9o81A2rbyRIVIwzaMq&Gw%0uUVo9@prV29c^%g896}8^f)QBP^Ej%o zsg#=bht6<{DqH?9%rty&G|jEV4ubs-|CixBEzpw{?N6(urb2FpcY@1iMMPE4spyQ* zPHN;t`2N(ON(oU-lo3jTZG>jnN;DBFB%-wtTDql=ys3)P2;`~Uy| 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 adff526..0035b39 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-09-30 17:27+0000\n" +"POT-Creation-Date: 2026-10-02 05:07+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,31 @@ msgstr "每次处理的任务数（默认：1000）" msgid "Show what would happen without changing anything" msgstr "显示将会执行的操作，但不做任何更改" -#: management/commands/run_database_tasks.py:55 +#: management/commands/run_database_tasks.py:63 msgid "Execute tasks queued in the database" msgstr "执行在数据库中排队的任务" -#: management/commands/run_database_tasks.py:62 +#: management/commands/run_database_tasks.py:70 msgid "Queue name to process (all queues if not specified)" msgstr "要处理的队列名称（未指定时为所有队列）" -#: management/commands/run_database_tasks.py:68 +#: management/commands/run_database_tasks.py:76 msgid "Backend name (default: default)" msgstr "后端名称（默认：default）" -#: management/commands/run_database_tasks.py:73 +#: management/commands/run_database_tasks.py:81 msgid "Continuous mode (keep polling even when no tasks)" msgstr "持续模式（没有任务时也继续轮询）" -#: management/commands/run_database_tasks.py:79 +#: management/commands/run_database_tasks.py:87 msgid "Polling interval in seconds for continuous mode (default: 5)" msgstr "持续模式的轮询间隔秒数（默认：5）" -#: management/commands/run_database_tasks.py:85 +#: management/commands/run_database_tasks.py:93 msgid "Maximum number of tasks to process (0=unlimited, default: 0)" msgstr "要处理的最大任务数（0=不限，默认：0）" -#: management/commands/run_database_tasks.py:92 +#: management/commands/run_database_tasks.py:100 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 +158,7 @@ msgstr "" "消息代理再回退到数据库，'auto'（默认）在后端有可接收的消息代理时为 'both'，否" "则为 'db'" -#: management/commands/run_database_tasks.py:104 +#: management/commands/run_database_tasks.py:112 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,20 +166,29 @@ msgstr "" "再次查看前等待消息代理消息的秒数。从消息代理接收时，取代 --interval 作为空闲" "时的等待（默认：20）" -#: management/commands/run_database_tasks.py:114 +#: management/commands/run_database_tasks.py:122 msgid "Maximum number of broker messages to receive at a time (default: 1)" msgstr "每次接收的消息代理消息的最大数量（默认：1）" -#: management/commands/run_database_tasks.py:123 +#: management/commands/run_database_tasks.py:131 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; " "without it the command exits when every worker has (default: 1)" msgstr "" -"从一条命令启动这么多个工作进程，每个都是不带 --workers 的本命令副本。指定 --continuous 时，" -"退出的工作进程会被替换；未指定时，所有工作进程退出后命令即退出（默认：1）" +"从一条命令启动这么多个工作进程，每个都是不带 --workers 的本命令副本。指定 --" +"continuous 时，退出的工作进程会被替换；未指定时，所有工作进程退出后命令即退出" +"（默认：1）" -#: management/commands/run_database_tasks.py:134 +#: management/commands/run_database_tasks.py:143 +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）" + +#: management/commands/run_database_tasks.py:154 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -187,13 +196,13 @@ msgstr "" "收到 SIGTERM/SIGINT 后、强制退出前等待正在运行的任务的最长秒数（0=无限期等" "待，默认：0）" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:163 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:153 +#: management/commands/run_database_tasks.py:173 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)" @@ -201,7 +210,7 @@ msgstr "" "未处理任何任务时以此代码退出，以便作业调度器区分空转的运行与实际的运行（0=正" "常退出，默认：0）" -#: management/commands/run_database_tasks.py:164 +#: management/commands/run_database_tasks.py:184 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 19f8f3d..524e2c6 100644 --- a/django_database_task/management/commands/run_database_tasks.py +++ b/django_database_task/management/commands/run_database_tasks.py @@ -1,7 +1,9 @@ import argparse +import functools import logging import socket import sys +import threading import uuid from contextlib import ExitStack @@ -17,6 +19,7 @@ from django_database_task.models import DatabaseTask from django_database_task.shutdown import GracefulShutdown, signal_name from django_database_task.supervisor import WorkerSupervisor, worker_arguments +from django_database_task.threads import ExecutorThreads #: Where the worker looks for tasks to run. SOURCE_AUTO = "auto" @@ -27,6 +30,11 @@ logger = logging.getLogger("django_database_task") +#: How long the main thread waits for an executor thread to report back +#: before looking at the shutdown flag again. A result wakes it at once; +#: this only bounds the wait. +COLLECT_WAIT = 1.0 + def _exit_code_argument(value): """Parse an exit code option, rejecting what a shell cannot report.""" @@ -126,6 +134,18 @@ def add_arguments(self, parser): "exits when every worker has (default: 1)" ), ) + parser.add_argument( + "--threads", + type=int, + default=1, + metavar="M", + help=_( + "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)" + ), + ) parser.add_argument( "--shutdown-timeout", type=float, @@ -185,6 +205,9 @@ def handle(self, *args, **options): raise CommandError("--max-messages must be at least 1") if options["workers"] < 1: raise CommandError("--workers must be at least 1") + threads = options["threads"] + if threads < 1: + raise CommandError("--threads must be at least 1") backend = task_backends[backend_name] source = self._resolve_source(backend, options["source"]) @@ -200,6 +223,8 @@ def handle(self, *args, **options): self.stdout.write(f"Worker ID: {worker_id}") self.stdout.write(f"Backend: {backend_name}") self.stdout.write(f"Source: {source}") + if threads > 1: + self.stdout.write(f"Threads: {threads}") if queue_name: self.stdout.write(f"Queue: {queue_name}") if continuous: @@ -224,8 +249,10 @@ def handle(self, *args, **options): self.verbosity = verbosity # Counted here rather than returned from the loop because a task can # fail at several depths (the run itself, the broker message that - # named it) and every one of them feeds the same exit code. + # named it) and every one of them feeds the same exit code. With + # --threads it is counted from several threads, hence the lock. self.tasks_failed = 0 + self._failures_lock = threading.Lock() logger.info( "Worker started: id=%s backend=%s source=%s", @@ -238,6 +265,7 @@ def handle(self, *args, **options): "source": source, "queue_name": queue_name, "continuous": continuous, + "threads": threads, }, ) @@ -255,6 +283,11 @@ def handle(self, *args, **options): # A broker a worker receives from may hold a connection # open, so release it however the loop ends. stack.callback(self._close_broker, backend.broker) + executors = None + if threads > 1: + self._check_database_supports_concurrency("--threads", threads) + executors = ExecutorThreads(threads, worker_id).start() + stack.callback(executors.stop) tasks_processed = self._process_tasks( shutdown=shutdown, backend=backend, @@ -268,6 +301,7 @@ def handle(self, *args, **options): wait_time=wait_time, max_messages=max_messages, verbosity=verbosity, + executors=executors, ) if shutdown.is_set() and verbosity >= 1: @@ -315,7 +349,7 @@ def _supervise(self, options): workers = options["workers"] verbosity = options["verbosity"] - self._check_database_supports_workers(workers) + self._check_database_supports_concurrency("--workers", workers) supervisor = WorkerSupervisor( worker_arguments(), @@ -333,23 +367,25 @@ def _supervise(self, options): if exit_code: sys.exit(exit_code) - def _check_database_supports_workers(self, workers): + def _check_database_supports_concurrency(self, option, value): """ - Refuse several workers on SQLite. + Refuse several workers, or several threads, on SQLite. Nothing runs twice there, since a task is claimed with a conditional - UPDATE, but SQLite has no row locks: two workers claiming at once - both hold the read lock and one is refused the write with "database - is locked", which the worker counts as a failed task. The README - says SQLite is for development and single-worker deployments; this - is where that stops being a note. + UPDATE, but SQLite has no row locks: two claims at once both hold the + read lock and one is refused the write with "database is locked", + which the worker counts as a failed task. Threads claim from one + thread when polling the database, but not for broker messages, and + their finishing writes collide the same way. The README says SQLite + is for development and single-worker deployments; this is where + that stops being a note. """ alias = router.db_for_write(DatabaseTask) if connections[alias].vendor == "sqlite": raise CommandError( - f"--workers {workers} needs a database with row locks, and " - f"the {alias!r} database is SQLite. Run one worker, or use " - "PostgreSQL, MySQL or MariaDB." + f"{option} {value} needs a database with row locks, and the " + f"{alias!r} database is SQLite. Run one worker with one " + "thread, or use PostgreSQL, MySQL or MariaDB." ) def _exit_code(self, tasks_processed, empty_exit_code, failed_exit_code): @@ -407,19 +443,62 @@ def _process_tasks( wait_time=20.0, max_messages=1, verbosity=1, + executors=None, ): + """ + The worker loop: find tasks and run them until there is reason to stop. + + Without ``executors`` every task is run right here, in the thread + that found it. With them (``--threads``) this thread only fetches, + receives and claims, and hands each task to an idle executor; the + executors report back what they ran, and that is added up here. + """ use_broker = source in (SOURCE_BROKER, SOURCE_BOTH) use_database = source in (SOURCE_DATABASE, SOURCE_BOTH) broker = backend.broker if use_broker else None tasks_processed = 0 + exhausted = False + + def in_flight(): + return executors.busy if executors is not None else 0 + + def free_slots(): + return executors.idle_count() if executors is not None else 1 + + def collect(wait=False): + """Add up what the executors have finished since last time.""" + nonlocal tasks_processed + if executors is None: + return + outcomes = executors.collect(timeout=COLLECT_WAIT if wait else None) + tasks_processed += sum(1 for processed in outcomes if processed) def remaining(): - """How many more tasks may be run, capped by --max-messages.""" + """How many more tasks may be started right now.""" + limit = max_messages + if executors is not None: + limit = min(limit, free_slots()) if not max_tasks: - return max_messages - return min(max_messages, max_tasks - tasks_processed) + return limit + return min(limit, max_tasks - tasks_processed - in_flight()) + + def reached_max(): + # Tasks handed to an executor count towards the cap, so no more + # than --max-tasks are ever started. One that turns out not to + # run (a broker message for a task that is gone) can leave the + # run short of the cap by that many. + return self._reached_max_tasks( + tasks_processed + in_flight(), max_tasks, verbosity + ) while not shutdown.is_set(): + collect() + if executors is not None and free_slots() == 0: + # Every executor is busy: wait for one to come back, not + # for the poll interval. + collect(wait=True) + continue + worked = False if broker is not None: @@ -435,10 +514,12 @@ def remaining(): wait, remaining(), verbosity, + executors, ) - tasks_processed += count + if executors is None: + tasks_processed += count worked = count > 0 - if self._reached_max_tasks(tasks_processed, max_tasks, verbosity): + if reached_max(): break if shutdown.is_set(): break @@ -447,19 +528,24 @@ def remaining(): task = fetch_task(queue_name=queue_name, backend_name=backend_name) if task is not None: worked = True - if self._run_database_task(backend, task, worker_id, verbosity): + if executors is None: + counted = self._run_database_task( + backend, task, worker_id, verbosity + ) + else: + counted = self._dispatch_database_task( + executors, backend, task, verbosity + ) + if counted: tasks_processed += 1 - if self._reached_max_tasks( - tasks_processed, max_tasks, verbosity - ): - break + if reached_max(): + break if worked: continue if not continuous: - if verbosity >= 1: - self.stdout.write("No more tasks to process.") + exhausted = True break if verbosity >= 2: @@ -471,7 +557,7 @@ def remaining(): if source == SOURCE_BOTH and wait_time > 0: # The broker's own wait doubles as the idle interval, so a # message wakes the worker up straight away. - tasks_processed += self._receive_and_run( + count = self._receive_and_run( broker, backend_name, queue_name, @@ -479,8 +565,11 @@ def remaining(): wait_time, remaining(), verbosity, + executors, ) - if self._reached_max_tasks(tasks_processed, max_tasks, verbosity): + if executors is None: + tasks_processed += count + if reached_max(): break elif source == SOURCE_BROKER and wait_time > 0: # receive() has already waited. @@ -490,6 +579,16 @@ def remaining(): # requested instead of waiting out the interval. break + while in_flight(): + # The tasks already handed out finish first. A shutdown that + # runs past --shutdown-timeout ends the process from the + # GracefulShutdown timer, as it does for a task run inline. + collect(wait=True) + + if exhausted and verbosity >= 1: + # Reported after the executors are done, so it reads in order. + self.stdout.write("No more tasks to process.") + return tasks_processed def _reached_max_tasks(self, tasks_processed, max_tasks, verbosity): @@ -501,7 +600,7 @@ def _reached_max_tasks(self, tasks_processed, max_tasks, verbosity): def _run_database_task(self, backend, task, worker_id, verbosity): """ - Run a task fetched straight from the database. + Run a task fetched straight from the database, here and now. Returns: True if the task was run here, False if another worker claimed @@ -513,14 +612,7 @@ def _run_database_task(self, backend, task, worker_id, verbosity): try: result = backend.run_task(task, worker_id=worker_id) except Exception as e: - logger.exception( - "Worker could not run task: id=%s path=%s", - task.id, - task.task_path, - extra=task_log_fields(task, worker_id), - ) - self.stdout.write(self.style.ERROR(f" Error running task: {e}")) - self.tasks_failed += 1 + self._report_run_error(task, worker_id, e) return True if result is None: @@ -528,10 +620,78 @@ def _run_database_task(self, backend, task, worker_id, verbosity): self.stdout.write(" Task is not ready to run; nothing to do") return False + self._record_result(result, verbosity) + return True + + def _dispatch_database_task(self, executors, backend, task, verbosity): + """ + Claim a task fetched from the database and hand it to an executor. + + The claim happens here, in the fetching thread, so that the next + fetch does not return the same READY row; it is made with the + executor's worker id, which is what the task records. The executor + reports the run back through the loop's ``collect()``. + + Returns: + True if the task is to be counted now (the claim itself failed, + which is a failed task that reached no executor), False if it + was handed out or was no longer READY. + """ + index = executors.acquire() + thread_worker_id = executors.worker_ids[index] + if verbosity >= 1: + self.stdout.write(f"\nProcessing task: {task.id} ({task.task_path})") + + try: + claimed = backend.claim_task(task, worker_id=thread_worker_id) + except Exception as e: + executors.release(index) + self._report_run_error(task, thread_worker_id, e) + return True + + if not claimed: + executors.release(index) + if verbosity >= 1: + self.stdout.write(" Task is not ready to run; nothing to do") + return False + + executors.submit( + index, + functools.partial( + self._run_claimed_task, backend, task, thread_worker_id, verbosity + ), + ) + return False + + def _run_claimed_task(self, backend, task, worker_id, verbosity): + """Run a task :meth:`_dispatch_database_task` claimed, in an executor.""" + try: + result = backend.run_claimed_task(task, worker_id=worker_id) + except Exception as e: + self._report_run_error(task, worker_id, e) + return True + self._record_result(result, verbosity) + return True + + def _report_run_error(self, task, worker_id, error): + """Record that this worker could not run ``task`` at all.""" + logger.exception( + "Worker could not run task: id=%s path=%s", + task.id, + task.task_path, + extra=task_log_fields(task, worker_id), + ) + self.stdout.write(self.style.ERROR(f" Error running task: {error}")) + self._count_failure() + + def _record_result(self, result, verbosity): if result.status != TaskResultStatus.SUCCESSFUL: - self.tasks_failed += 1 + self._count_failure() self._report_result(result.status, verbosity) - return True + + def _count_failure(self): + with self._failures_lock: + self.tasks_failed += 1 def _receive_and_run( self, @@ -542,8 +702,16 @@ def _receive_and_run( wait_seconds, max_messages, verbosity, + executors=None, ): - """Receive messages from the broker and run the tasks they name.""" + """ + Receive messages from the broker and run the tasks they name. + + Returns: + Without executors, how many tasks were run. With them, how many + messages were handed out; the executors report the runs back + through the loop's ``collect()``. + """ if max_messages < 1: return 0 @@ -567,11 +735,29 @@ def _receive_and_run( self.stdout.write(self.style.ERROR(f"\nError receiving from broker: {e}")) return 0 - return sum( - 1 - for message in messages or [] - if self._run_broker_message(broker, message, worker_id, verbosity) - ) + messages = list(messages or []) + if executors is None: + return sum( + 1 + for message in messages + if self._run_broker_message(broker, message, worker_id, verbosity) + ) + + for message in messages: + # remaining() capped the receive by the idle executors, so one + # is free for each message. + index = executors.acquire() + executors.submit( + index, + functools.partial( + self._run_broker_message, + broker, + message, + executors.worker_ids[index], + verbosity, + ), + ) + return len(messages) def _run_broker_message(self, broker, message, worker_id, verbosity): """ @@ -602,7 +788,7 @@ def _run_broker_message(self, broker, message, worker_id, verbosity): extra=self._broker_task_log_fields(message, worker_id), ) self.stdout.write(self.style.ERROR(f" Error running task: {e}")) - self.tasks_failed += 1 + self._count_failure() self._nack(broker, message) return False @@ -613,9 +799,7 @@ def _run_broker_message(self, broker, message, worker_id, verbosity): self.stdout.write(" Task is not ready to run; nothing to do") return False - if result.status != TaskResultStatus.SUCCESSFUL: - self.tasks_failed += 1 - self._report_result(result.status, verbosity) + self._record_result(result, verbosity) return True def _broker_task_log_fields(self, message, worker_id): diff --git a/django_database_task/threads.py b/django_database_task/threads.py new file mode 100644 index 0000000..ccb9d4c --- /dev/null +++ b/django_database_task/threads.py @@ -0,0 +1,156 @@ +""" +Executor threads for ``run_database_tasks --threads M``. + +The main thread of the worker stays the one that fetches and receives: it +is the only thread polling the database, the only one on the broker's +connection, and the one the signal handlers run in. What ``--threads`` +changes is that a task it found is run by one of M executor threads instead +of inline, so M tasks are in progress at once while there is still one +poller per process. + +Each executor has a worker id of its own, derived from the process's, so +the ``worker_ids`` recorded on a task say which thread ran it. Django keeps +database connections per thread, so each executor has a connection of its +own too; after every task it closes the ones that outlived ``CONN_MAX_AGE`` +or broke, the way a request does, since a worker has no ``request_finished`` +to do it for it. + +Tasks run under ``--threads`` share one process. A task that is not +thread-safe, calls ``os._exit()`` or crashes the interpreter takes the other +M-1 with it; that is the trade the option makes, and the README says so. +""" + +import logging +import queue +import threading + +from django.db import close_old_connections, connections + +logger = logging.getLogger("django_database_task") + + +class ExecutorThreads: + """ + M threads that run the jobs the main thread hands them. + + Only the main thread calls :meth:`acquire`, :meth:`submit`, + :meth:`release` and :meth:`collect`; the executors put themselves back + on the idle list and their outcomes on the results queue. A job is a + callable taking no arguments; what it returns is handed back by + :meth:`collect`. + + Args: + count: How many executors. + worker_id: The process's worker id; executor ``n`` is + ``f"{worker_id}-t{n}"``. + """ + + def __init__(self, count, worker_id): + if count < 1: + raise ValueError("count must be at least 1") + self.count = count + self.worker_ids = [f"{worker_id}-t{n}" for n in range(1, count + 1)] + self._inboxes = [queue.Queue(maxsize=1) for _ in range(count)] + self._idle = queue.Queue() + for index in range(count): + self._idle.put(index) + self._results = queue.Queue() + #: Jobs handed out and not yet collected. Main thread only. + self._busy = 0 + self._threads = [ + threading.Thread( + target=self._run, + args=(index,), + name=f"executor-{index + 1}", + daemon=True, + ) + for index in range(count) + ] + + def start(self): + for thread in self._threads: + thread.start() + return self + + def stop(self): + """Let every executor finish its job and exit, and wait for them.""" + for inbox in self._inboxes: + inbox.put(None) + for thread in self._threads: + thread.join() + + @property + def busy(self): + """How many jobs are out: submitted and not yet collected.""" + return self._busy + + def idle_count(self): + """How many executors are free to take a job right now.""" + return self._idle.qsize() + + def acquire(self): + """ + Take an idle executor, returning its index, or None if none is idle. + + The index is held until :meth:`submit` hands it a job or + :meth:`release` gives it back, so the caller can do work in between + (claim a task with the executor's worker id) knowing the slot is its. + """ + try: + return self._idle.get_nowait() + except queue.Empty: + return None + + def release(self, index): + """Give back an executor taken with :meth:`acquire` and not used.""" + self._idle.put(index) + + def submit(self, index, job): + """Hand ``job`` to the executor taken with :meth:`acquire`.""" + self._busy += 1 + self._inboxes[index].put(job) + + def collect(self, timeout=None): + """ + Return the outcomes of the jobs that have finished since last time. + + With ``timeout``, wait up to that many seconds for the first one when + none has finished yet; without it, return at once. + """ + outcomes = [] + try: + if timeout: + outcomes.append(self._results.get(timeout=timeout)) + while True: + outcomes.append(self._results.get_nowait()) + except queue.Empty: + pass + self._busy -= len(outcomes) + return outcomes + + def _run(self, index): + inbox = self._inboxes[index] + try: + while True: + job = inbox.get() + if job is None: + break + try: + outcome = job() + except BaseException: + # The jobs report their own failures; this is a last + # resort so that a slot is never lost to a thread that + # died, and so that SystemExit raised by task code ends + # the task rather than the thread. + logger.exception( + "Executor thread %s failed", + self.worker_ids[index], + extra={"worker_id": self.worker_ids[index]}, + ) + outcome = None + finally: + close_old_connections() + self._results.put(outcome) + self._idle.put(index) + finally: + connections.close_all() diff --git a/tests/postgres/test_threads.py b/tests/postgres/test_threads.py new file mode 100644 index 0000000..00e07b7 --- /dev/null +++ b/tests/postgres/test_threads.py @@ -0,0 +1,179 @@ +""" +``--threads M`` at work, against PostgreSQL. + +The executor threads claim and write on connections of their own, so they +need a database with row locks, which is why the command refuses several +threads on SQLite; tests/test_commands.py covers that refusal, and +tests/test_threads.py the pool on its own. Here the command runs with +threads for real: tasks that can only finish together, one worker id per +thread, the cap, the failure count, broker messages spread over the +threads, and the shutdown that lets the running tasks finish. + +Run the suite with DJANGO_DATABASE_ENGINE=postgresql to include them. +""" + +import sys +import threading +from io import StringIO +from unittest.mock import patch + +import pytest +from django.core.management import call_command +from django.db import connections +from django.tasks.base import TaskResultStatus + +from django_database_task.brokers import BrokerMessage +from django_database_task.models import DatabaseTask +from tests import tasks as test_tasks +from tests.tasks import ( + barrier_task, + failing_task, + shutdown_signal_task, + simple_task, + slow_task, +) +from tests.test_commands import FakePullBroker, make_backend, run_worker + +if connections["default"].vendor != "postgresql": + pytest.skip( + "needs a PostgreSQL database; run the suite with " + "DJANGO_DATABASE_ENGINE=postgresql", + allow_module_level=True, + ) + +posix_signals = pytest.mark.skipif( + sys.platform == "win32", + reason="os.kill() cannot deliver a signal to the test process on Windows", +) + +# Real commits: the executor threads work on connections of their own, and +# must see the rows the test enqueued and be free to write theirs while the +# test's connection is not holding a transaction. +pytestmark = pytest.mark.django_db(transaction=True) + + +class TestThreads: + def test_tasks_run_at_the_same_time(self): + """Three tasks that each wait for the other two can only pass together.""" + test_tasks.barrier = threading.Barrier(3, timeout=10) + results = [barrier_task.enqueue() for _ in range(3)] + out = StringIO() + + call_command("run_database_tasks", threads=3, stdout=out) + + statuses = {DatabaseTask.objects.get(id=r.id).status for r in results} + assert statuses == {TaskResultStatus.SUCCESSFUL} + output = out.getvalue() + assert "Threads: 3" in output + assert "Total tasks processed: 3" in output + assert "Tasks failed" not in output + + def test_each_thread_records_its_own_worker_id(self): + test_tasks.barrier = threading.Barrier(2, timeout=10) + results = [barrier_task.enqueue() for _ in range(2)] + + call_command("run_database_tasks", threads=2, stdout=StringIO()) + + worker_ids = sorted( + DatabaseTask.objects.get(id=r.id).worker_ids_json[0] for r in results + ) + assert [w.rsplit("-", 1)[1] for w in worker_ids] == ["t1", "t2"] + assert worker_ids[0].rsplit("-", 1)[0] == worker_ids[1].rsplit("-", 1)[0] + + def test_max_tasks_counts_the_tasks_handed_out(self): + for _ in range(5): + slow_task.enqueue(seconds=0.1) + out = StringIO() + + call_command("run_database_tasks", threads=2, max_tasks=3, stdout=out) + + assert "Reached max tasks limit: 3" in out.getvalue() + assert "Total tasks processed: 3" in out.getvalue() + assert DatabaseTask.objects.filter(status=TaskResultStatus.READY).count() == 2 + + def test_failures_are_counted_across_threads(self): + for _ in range(4): + failing_task.enqueue() + out = StringIO() + + with pytest.raises(SystemExit) as exc_info: + call_command( + "run_database_tasks", threads=2, failed_exit_code=5, stdout=out + ) + + assert exc_info.value.code == 5 + assert "Tasks failed: 4" in out.getvalue() + assert "Total tasks processed: 4" in out.getvalue() + + def test_connections_are_closed_after_every_task(self): + for _ in range(3): + simple_task.enqueue(1, 2) + + with patch("django_database_task.threads.close_old_connections") as close: + call_command("run_database_tasks", threads=2, stdout=StringIO()) + + assert close.call_count == 3 + + def test_output_ends_after_the_last_task(self): + slow_task.enqueue(seconds=0.3) + out = StringIO() + + call_command("run_database_tasks", threads=2, stdout=out) + + output = out.getvalue() + assert output.index("Task completed successfully") < output.index( + "No more tasks to process." + ) + + def test_broker_messages_are_spread_over_the_threads(self): + test_tasks.barrier = threading.Barrier(3, timeout=10) + results = [barrier_task.enqueue() for _ in range(3)] + broker = FakePullBroker( + batches=[[BrokerMessage(task_id=r.id) for r in results]] + ) + + output = run_worker( + make_backend(broker), + source="broker", + threads=3, + max_messages=10, + wait_time=0, + ) + + statuses = {DatabaseTask.objects.get(id=r.id).status for r in results} + assert statuses == {TaskResultStatus.SUCCESSFUL} + assert sorted(broker.acked) == sorted(r.id for r in results) + assert "Total tasks processed: 3" in output + + def test_receive_asks_for_no_more_than_the_free_threads(self): + broker = FakePullBroker(batches=[]) + + run_worker( + make_backend(broker), + source="broker", + threads=2, + max_messages=10, + wait_time=0, + ) + + assert broker.received[0]["max_messages"] == 2 + + @posix_signals + def test_running_tasks_finish_before_shutdown(self): + """A signal lets the tasks in the threads finish and starts no more.""" + first = slow_task.enqueue(seconds=0.5) + second = shutdown_signal_task.enqueue() + third = simple_task.enqueue(1, 2) + out = StringIO() + + call_command("run_database_tasks", threads=2, stdout=out) + + assert DatabaseTask.objects.get(id=first.id).status == ( + TaskResultStatus.SUCCESSFUL + ) + assert DatabaseTask.objects.get(id=second.id).status == ( + TaskResultStatus.SUCCESSFUL + ) + assert DatabaseTask.objects.get(id=third.id).status == TaskResultStatus.READY + assert "Shutdown complete" in out.getvalue() + assert "Total tasks processed: 2" in out.getvalue() diff --git a/tests/tasks.py b/tests/tasks.py index ef0e01c..2ed7cad 100644 --- a/tests/tasks.py +++ b/tests/tasks.py @@ -122,3 +122,17 @@ def shutdown_aware_task(iterations=10): break completed += 1 return completed + + +#: Set by a test to a threading.Barrier; barrier_task() waits on it, so the +#: tasks only finish if as many of them are running at the same time. +barrier = None + + +@task +def barrier_task(): + """Task that waits for the other parties of ``barrier`` before returning.""" + from . import tasks + + tasks.barrier.wait() + return "met" diff --git a/tests/test_commands.py b/tests/test_commands.py index de934c9..5a4bd93 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -1306,7 +1306,7 @@ def run_workers(self, code=0, **options): out = StringIO() with ( patch(f"{COMMAND_MODULE}.worker_arguments", return_value=stub_worker(code)), - patch.object(RunDatabaseTasks, "_check_database_supports_workers"), + patch.object(RunDatabaseTasks, "_check_database_supports_concurrency"), ): call_command("run_database_tasks", stdout=out, stderr=out, **options) return out.getvalue() @@ -1379,3 +1379,37 @@ def stop(): assert "; restarting." in output assert "Received SIGTERM: stopping" in output assert "Shutdown complete: every worker exited." in output + + +@pytest.mark.django_db +class TestRunDatabaseTasksThreads: + """ + What the command checks for --threads before starting any thread. + + The threads at work are tested against PostgreSQL in + tests/postgres/test_threads.py, and the pool on its own in + tests/test_threads.py. + """ + + def test_threads_must_be_positive(self): + with pytest.raises(CommandError, match="--threads must be at least 1"): + call_command("run_database_tasks", threads=0, stdout=StringIO()) + + def test_one_thread_is_the_worker_itself(self): + result = simple_task.enqueue(1, 2) + out = StringIO() + + call_command("run_database_tasks", threads=1, stdout=out) + + assert "Threads:" not in out.getvalue() + worker_ids = DatabaseTask.objects.get(id=result.id).worker_ids_json + assert len(worker_ids) == 1 + assert "-t" not in worker_ids[0] + + @pytest.mark.skipif( + connections["default"].vendor != "sqlite", + reason="the refusal is for SQLite", + ) + def test_several_threads_are_refused_on_sqlite(self): + with pytest.raises(CommandError, match="--threads 2 needs a database"): + call_command("run_database_tasks", threads=2, stdout=StringIO()) diff --git a/tests/test_threads.py b/tests/test_threads.py new file mode 100644 index 0000000..1b9e97c --- /dev/null +++ b/tests/test_threads.py @@ -0,0 +1,102 @@ +"""Tests for the executor threads behind ``--threads``, without a database.""" + +import threading +import time +from unittest.mock import patch + +import pytest + +from django_database_task.threads import ExecutorThreads + + +@pytest.fixture +def executors(): + pool = ExecutorThreads(2, "host-abcd1234").start() + yield pool + pool.stop() + + +def wait_for(condition, timeout=10): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if condition(): + return + time.sleep(0.01) + pytest.fail("condition not met") + + +class TestExecutorThreads: + def test_rejects_less_than_one(self): + with pytest.raises(ValueError): + ExecutorThreads(0, "w") + + def test_worker_ids_are_derived_from_the_process_id(self, executors): + assert executors.worker_ids == ["host-abcd1234-t1", "host-abcd1234-t2"] + + def test_jobs_run_and_report_back(self, executors): + ran = [] + assert executors.idle_count() == 2 + + with patch("django_database_task.threads.close_old_connections") as close: + for value in ("a", "b"): + index = executors.acquire() + executors.submit(index, lambda value=value: ran.append(value) or value) + assert executors.busy == 2 + assert executors.acquire() is None + + outcomes = [] + while executors.busy: + outcomes += executors.collect(timeout=5) + + assert sorted(outcomes) == ["a", "b"] + assert sorted(ran) == ["a", "b"] + assert executors.idle_count() == 2 + assert close.call_count == 2 + + def test_jobs_run_at_the_same_time(self, executors): + barrier = threading.Barrier(2, timeout=5) + + def meet(): + barrier.wait() + return "met" + + for _ in range(2): + executors.submit(executors.acquire(), meet) + + outcomes = [] + while executors.busy: + outcomes += executors.collect(timeout=5) + + assert outcomes == ["met", "met"] + + def test_release_gives_the_slot_back(self, executors): + index = executors.acquire() + assert executors.idle_count() == 1 + executors.release(index) + assert executors.idle_count() == 2 + assert executors.busy == 0 + + def test_a_job_that_raises_does_not_lose_the_slot(self, executors, caplog): + def boom(): + raise SystemExit(3) + + executors.submit(executors.acquire(), boom) + outcomes = executors.collect(timeout=5) + + assert outcomes == [None] + assert executors.idle_count() == 2 + assert "Executor thread host-abcd1234-t" in caplog.text + + def test_collect_without_timeout_does_not_wait(self, executors): + started = time.monotonic() + assert executors.collect() == [] + assert time.monotonic() - started < 1 + + def test_stop_waits_for_the_running_job(self): + pool = ExecutorThreads(1, "w").start() + done = [] + pool.submit(pool.acquire(), lambda: time.sleep(0.2) or done.append(True)) + + pool.stop() + + assert done == [True]