Ten-Go (天號作戰)
Ten-Go is my scheme to have several computers—for now azag and yzao—share one pool of long computations running over days, possibly over weeks or even months, without any of them ever recomputing what another one did, and without having to split the work by hand, also collecting the results systematically, so that all this can happen in the background.
A machine is told how much time it has ("put yzao on this for the next ten hours") and Ten-Go hands it a group of free tasks that fills that budget.
I designed with Claude Opus 5.5 on 26 October (2026), while azag was at ICMM and yzao at home, out of reach of each other.
Why
The one-atom lasing skeleton needs tens of thousands (possibly up to hundreds or thousands of thousands) of diagonalizations at high precision: closing the condensing window $0.15\le P_\sigma/g\le 12.51$ at a spacing of $0.004g$ alone is 1,952 additional pumps, about 1,760 kernel-hours (some 50 hours with both machines at full power, see Fig. (0OG5J) for the window). Until now each machine was given a fixed window and the windows were handed over by hand: that wastes time each time a machine is moved, stopped or slowed down because I'm working on it, and it requires the machines to see each other to hand over. Also it makes me having to worry about such distributions, which is prone to mistakes.
The pool
The pool lives in a GitHub repository, laussy/ten-go, public since 27 October (nothing in the computation is private, and none of it is secret, and I would nevertheless even share some secrets in the open, sometimes), the only place both machines can reach wherever they are: the institute network blocks outgoing ssh to my server, and yzao sits behind the home router, so the two machines cannot talk to each other directly. The repository holds, for each campaign:
- the list of all its pumps, each with its orders and its cost, in one global order chosen so that any set of finished pumps is evenly spread over the window (van der Corput order), never a denser stretch here and a hole there;
- the allocations: one file per allocation, saying which machine took which pumps, when, for which budget, and which of them are done;
- the results, compressed per allocation, so that every machine can draw figures with everybody's data without the machines ever meeting.
None of this is confidential: pump values, orders, machine names, times and eigenvalues. I will even publish at some point the eigenvalues.
Tasks are budgets, not pumps
A machine does not take pumps one by one. It asks for a budget, e.g. ten hours on its 8 kernels, and Ten-Go:
- turns the budget into kernel-hours of that machine;
- prices each free pump with the measured cost of its orders on that machine (on azag running 30 kernels, about 0.52 kernel-hours below $5g$, where a pump is $n_t=150$ with weights plus 200, and 1.50 kernel-hours above, where it is 150 plus 250 and 300; yzao's jobs on its 8 kernels take as long as azag's on 30; once a machine has finished five jobs of an order, the median of its own times replaces the estimate);
- takes the free pumps in the global order as long as they end within the budget when run as the driver will run them, the longest first (below), and claims them all at once as one allocation—a single pump if the budget is very small.
The faster machine simply gets larger allocations, or comes back more often. Near the end of a campaign the allocations shrink so that the machines finish together.
No clash
A claim is a commit pushed to the repository. Git accepts pushes one after the other: if two machines claim at the same moment, the second push is refused; that machine fetches the first one's allocation, recomputes its own from what is left, and pushes again. No pump can therefore be in two allocations. This is what happened in the first test (below): the two machines claimed in the same second, azag's push was refused, and its allocation was recomputed with none of yzao's pumps.
An allocation has a deadline: its start, plus one and a half times its budget, plus two hours. A pump that is not done by then goes back to the pool, so a machine that dies, sleeps or is carried away does not keep its pumps. A machine stopped on purpose (say before a move) releases its unstarted pumps at once.
On each machine
Each allocation gets its own work directory, with the plan of its pumps for the same driver as before (run_window.py, which re-reads its number of kernels every minute). The client is one Python file, tengo.py, kept in the repository itself and run as the command ten-go (a link to it on the PATH):
| Command | Does |
|---|---|
ten-go --kido MACHINE --repo CLONE --work DIR |
once per machine: its name, its clone of the pool, where the work directories go |
ten-go --honkai --jikan 10 [--ki N] [--go] [--shinshu [N]] [--shugeki] [--kikan] |
takes an allocation filling ten hours of this machine, writes its work directory, and starts the driver if asked; with --shinshu, the driver also takes the threads that stay idle; with --kikan, the allocation ends at the end of the ten hours rather than going on (below)
|
ten-go --honkai --until TAG [--ki N] [--go] |
the same, its time lasting until the moment TAG instead (below) |
ten-go --tassei |
pushes the pumps finished since the last time, their files as one archive, with the time of each job |
ten-go --uchikiri [--dry] |
ends this machine's allocations as soon as possible without loss, and says when (below) |
ten-go --chushi [MIN] [--now|--wait|--dry] |
stops them now, the only stop, but first offers to wait for a job within MIN minutes (10) of completion (below) |
ten-go launch [--shinshu [N]] |
(re)starts the driver of this machine's open allocation |
ten-go --senkyo |
what every machine holds and until when, what is done, what is free, and the expected ends |
ten-go watch [--until TAG [--kikan]] [--jikan 10] [--shinshu [N]] [--shugeki] |
runs the machine unattended, ten hours at a time: until the campaign is complete, or until TAG and then on with sorties, or with --kikan until TAG only (below)
|
ten-go --gensoku N [--jikan H] |
lowers this machine to N kernels for Ten-Go, until --zensoku or for H hours (below)
|
ten-go --zensoku [H] |
gives its kernels back, one at a time over H hours (1; 0: at once) |
ten-go --shushu DIR |
unpacks everybody's results into DIR, to draw the figures |
ten-go kh [TASK] |
the kernel-hours spent, per task and machine (below) |
ten-go kh import DIR [TASK] [--note TEXT] |
books, or brings up to date, any run of run_window.py, in Ten-Go or not |
ten-go kh add TASK KH [--note TEXT] |
books by hand a run without a log |
The time of a claim can be a moment instead of so many hours: --until TAG, TAG an AF tag (or a date 'YYYY-MM-DD HH:MM'), takes the hours from now until then in place of --jikan. Given both, the command claims nothing and prints itself again with the hours until TAG: at 07:57 on 27 October, ten-go --honkai --jikan 19 --until 0PfxC --ki 30 --go printed ten-go --honkai --jikan 25.03 --ki 30 --go # 25.03 h x 30 kernels = 750.9 kh, 0PfxC (af) being 9:00 the next day. The tag is read back by ~/bin/AF itself, rather than by one more copy of its decoder.
The watchdog (ten-go watch) pushes the finished pumps every half hour and restarts a dead driver (three times at most per allocation). As soon as an allocation has launched all its jobs, it claims the next one (until the campaign is complete, or past its --until as sorties, below), so that the kernels freed by the tail of one allocation go to the next: the new driver's number of kernels is the machine's minus the jobs of the tail.
Shinshu
With --shinshu N, an allocation takes the threads of its machine that stay idle, from its own kernels up to $N$, or up to every thread with a bare --shinshu, and gives them back as soon as other work wants them. The driver looks once a minute at what the machine spent, from /proc: all its threads' time, minus that of the run's own jobs and everything they started. What is left is the other work, and the threads free of it are the run's to take. After ten minutes in which at least one more thread than the run uses was free, it takes one more kernel; after two minutes in which other work took them back, it comes back toward the allocation's kernels, never below. A kernel above them runs the shortest job waiting rather than the next in the plan, so that the retreat is quick: lowering the count only stops new launches, and no job is killed. The shortest is judged by the run's own times once three jobs of an order are done, and before that by azag's times at home scaled to the run's speed. The budget of the allocation is unchanged; shinshu only ends it sooner. Outside Ten-Go, a file shinshu (a number, or max) in a run's directory does the same, and the current count is in shinshu.now.
On azag at 07:15 on 27 October, the 30 jobs of the refinement of the boiling window used 29.94 of the 32 threads and all other work 0.58, which left one thread for shinshu. Fed made-up loads, the rule went from 30 to 31 kernels after ten minutes at 0.6 threads of other work, back to 30 two minutes after that work rose to 4, and up again ten minutes after it fell. The extra kernels went to the 150w jobs (1,118 s at home) rather than 250e (1,580) or 300e (2,685), and below $5g$ they would go to 200e (760). The driver itself has not yet run with shinshu.
Uchikiri
Stopping at once (--chushi, below) loses what the jobs in flight had computed, as are the orders already done of the pumps not complete, since a pump counts only with all its orders. With --uchikiri, the machine instead ends as soon as possible without loss: each driver launches no new pump, only the jobs that complete the pumps started (a job done or in flight), and says when it will end, and what a STOP or a kill would waste instead. The pumps not started go back to the pool at once, for the other machines; a watch of the machine claims nothing more; and the driver pushes the last results itself when it ends. With --dry, it only says when. Outside Ten-Go, python3 run_window.py --uchikiri [--dry] does the same in a run's directory.
At 07:25 on 27 October, the dry run on the refinement of the boiling window, which a STOP was to end at 08:05, gave: 127 pumps not started given up; 30 jobs in flight completing the 15 pumps under way, none to launch; the run ends about 08:11 without loss, while a STOP now would waste nothing and a kill 14.8 kh. A pump's jobs are launched one after the other, so only the last one launched is ever cut by a STOP. It is with shinshu, whose extra kernels take the short jobs of pumps further down the plan, that many pumps are half done at any time, and a STOP wastes them all.
Chushi
--chushi is the stop for when the machine is wanted now. It aborts everything, but a job within ten minutes of completion (--chushi MIN for another span) is first offered to wait for, since a few minutes buy its whole time back:
One job is 3 min from completion (300e at P=15.79); the next are 17 min, 2 h 20, 6 h away, so those are aborted.
Abort this one too [n]ow, or [a]fter it completes? (anything else: nothing done)
After [a], the other jobs are aborted at once and the driver ends when the last job waited for does. --now or --wait gives the answer ahead (one of them is needed without a terminal), and --dry only asks. As with uchikiri, a watch of the machine claims nothing more, the pumps that will not be complete go back to the pool at once, and those the jobs waited for complete are pushed when they end. Outside Ten-Go: python3 run_window.py --chushi [--soon MIN] in a run's directory.
At 07:33 on 27 October, the dry run on the refinement of the boiling window found nine of its 30 jobs within ten minutes of completion, the last nine minutes away, and the other 21 between 10 and 48 minutes: stopping at once would have thrown away 9.1 kh, waiting the nine minutes for those nine jobs 5.0. The driver's --dir, which points it at another run's directory, now only reads (--progress and the dry runs), so that no run can be stopped by mistake from outside, and Ten-Go no longer hands a driver older than these options an option it does not know, which it would take for a plain run.
Shugeki
--shugeki, with --honkai or watch, is a sortie: a claim for a machine that may be stopped any moment. It takes the cheapest pumps first (their cost on that machine, the global order among equals), and its driver runs them whole, one after the other: no kernel above the allocation's jumps ahead to the shortest job of a pump further down, as shinshu's otherwise do. Few pumps are then ever half done, so that an uchikiri ends within a job or two and a chushi wastes little. On the copy of the pool, three hours on four kernels took, as a sortie, 20 pumps below $5g$ (10.4 kh, each pump a 200e and a 150w), where an ordinary claim of the same budget took six pumps spread between 7.3 and $11.7g$ (9.0 kh). A claim then kept back the longest job of its own pumps rather than of the whole campaign, which gave the sortie, whose longest job is 150w (19 minutes) rather than 300e (45), its full budget; since 1.33 it keeps back nothing, filled as its driver runs it (below).
The time given
The time given to a task, --jikan or --until, is a plan, not an end: when it is over, the machine goes on. What the time is for is the order of the work. Within it, an allocation runs its pumps the longest first (within one priority, each pump's jobs the longest first too), so that the time is met with the smallest pumps last, and a claim takes as many pumps as end within the time when run that way, each job launched as a kernel frees. Past its time, the allocation goes on, but as a sortie, in prevision of an imminent uchikiri: the pumps started are completed first, then those not started run the cheapest first, whole, one after the other. A watch past its --until goes on likewise, with sorties (its claims made as with --shugeki, the cheapest pumps first), from half an hour before, until the campaign is complete or an uchikiri by hand ends it. The order is by pump, not by job: running every pump's longest job first would start all the pumps at once, and an uchikiri could then give none back.
--kikan (帰還, the return to base) ends at the time instead. ten-go --honkai --jikan H --kikan makes an allocation that its driver ends without loss at the end of the $H$ hours, by an uchikiri of its own: the pumps started complete, those not started go back to the pool at once. ten-go watch --until TAG --kikan claims nothing in the last half hour, and its allocations end without loss at TAG. Outside Ten-Go, a file jikan or kikan holding a time (seconds since 1970) in a run's directory does the same.
With the pool as it was on 30 October, two hours of yzao on 6 kernels take 10 pumps (10.9 kh), whose jobs end in 1 h 52, where keeping back the longest job took 7 (7.4 kh, ending in 1 h 31), and ten hours of azag on 30 kernels 198 pumps (292.8 kh) rather than 180 (266.5 kh); in the last quarter of each, only jobs of 150w and 200e are launched. On a copy of the pool with made-up jobs (a test machine of 4 kernels, big pumps of three orders mixed with small ones of two), the plans ran the big pumps first; a watch past its time went on with sorties of small pumps, which an uchikiri by hand ended within a minute; with --kikan, the allocation running at TAG gave its pump not started back to the pool, completed its two pumps started, and the watch ended a minute after TAG.
Gensoku
--gensoku N (減速, slowing down) lowers the machine to $N$ kernels for Ten-Go, its allocations together, until --zensoku (全速, full speed), or for $H$ hours with --jikan H: a job that ends is not replaced while $N$ or more run (none is killed), the claims are made for $N$ kernels, and no idle thread is borrowed (shinshu). --zensoku [H] gives the kernels back one at a time over $H$ hours (one by default; --zensoku 0 at once), as the end of a --gensoku --jikan does over an hour. It is the machine's own (a file in its work directory, nothing pushed), kept by its watch and said by --senkyo; when it is set, it says when the open allocations will end at that pace, and which would miss their deadline. Without it, the watch keeps every kernel of the machine busy: a number of kernels lowered by hand in a driver's directory is put back within a minute, which is what a session met on yzao when it tried to leave two of its eight kernels free. yzao has run with --gensoku 6 since 08:30 on 29 October.
On a copy of the pool, a test machine of 4 kernels set to 2 while three jobs ran made its next claims for 2 kernels and ran no more than 2 jobs once those three had ended; after --zensoku 0.05 (three minutes) its claims went to 3, then 4 kernels, and shinshu came back.
Kernel-hours
What each task costs is kept in the pool, in ledger.tsv (which GitHub shows as a table): one line per allocation, which ten-go --tassei keeps up to date; one per run log imported by ten-go kh import, any run of run_window.py, inside Ten-Go or not, which a second import brings up to date; or one per run booked by hand. A kernel-hour (kh) is one kernel for one hour: the jobs' own times are summed, a failed job's included, and the first and last jobs are dated by their files, since the log has only the times of day. ten-go kh shows the totals per task and per machine, ten-go kh TASK the lines of one task, ten-go --senkyo and --tassei what each campaign and allocation has spent, and a claim prints the kh it takes, even when it only restates itself: ten-go --honkai --jikan 25.89 --go # 25.89 h x 30 kernels = 776.7 kh. On the copy of the pool, importing the runs of 26 October gave 194.3 kh for the $0.32g$ grid of the boiling window (285 jobs, 11:10 to 23:57), 229.0 for its refinement (426 jobs so far) and 12.3 for the four jobs at order 400.
Implementation
Version 1.0 was written the evening the design was made, and 1.1 and 1.2 the same night (claim --go, formerly --launch, starts the driver, and the command is ten-go, as on the emblem), and 1.3 to 1.5 the next morning (--until takes an AF tag; kh keeps the ledger of the kernel-hours spent; --shinshu takes the idle threads), 1.6 (--uchikiri ends without loss), 1.7 (--chushi stops now, after a warning), and 1.8 (claim became --honkai, and release went, --chushi being the stop), and 1.9 (--kernels became --ki, the kernels being the samurai), 1.10 (--campaign became --sakusen), and 1.11 (init became --kido and --hours --jikan, and no option can be abbreviated any more), 1.12 (status became --senkyo and done --tassei), 1.13 (fetch became --shushu), 1.14 (--shugeki, the sortie), 1.15 (mkcampaign became --sensen, which never rewrites a campaign already declared, its pump numbers being what the allocations point to), and 1.16 (campaigns declared from spec files, and adaptive campaigns, below): one file, Python's standard library and git, nothing to install. The later versions are listed at the head of tengo.py in the repository; among them, 1.30 (--gensoku and --zensoku) and 1.33 (the time a plan, not an end, and --kikan). Every command starts by resetting the machine's clone to the state of the repository, so the clone is Ten-Go's alone, and a new version of tengo.py pushed to the repository reaches every machine at its next command. The repository holds:
- campaigns/: one file per campaign, with its parameters, the prefix of its files, the reference time of each order, and its pumps in the global order;
- machines/: the kernels and the speed of each machine;
- allocs/: one file per allocation (machine, budget, deadline, pumps, those done, the time of each job, status open, finished or released), named after the machine and the time of the claim;
- results/: the result files of the pumps done, one archive per push;
- kit/: the driver and the Mathematica script, copied into each work directory.
A claim of $H$ hours on $N$ kernels takes pumps as long as their jobs, launched the longest first each as a kernel frees, end within the $H$ hours: within $N(H-t_{\max})$ kernel-hours, $t_{\max}$ the longest of their jobs on that machine, any order of launch ends in time, and only beyond that is the schedule worked out (until 1.33, that was the budget). The first campaign, closing $0.15\le P_\sigma/g\le12.51$ at $0.004g$, is declared by ten-go --sensen close-0.004: the pumps $P_\sigma=0.150+0.004k$ (the sublattice of the $0.02g$ grid of Fig. (0OG5J)) without those already done (that grid, yzao's two windows $0.5$–$1.5$ and $9.0$–$9.6g$, and $5$–$6g$, done on the other sublattice), exactly the 1,952 pumps above, in van der Corput order. azag reaches the repository through my own GitHub login; yzao will get a deploy key valid for this repository only, so that no account credential sits on it. The commits are signed tengo (machine).
Tests
On a copy of the pool on azag, with two clones playing azag and yzao:
- both claimed ten hours in the same second: yzao's push went first, azag's was refused and redone; azag got 249 pumps (277 kernel-hours for its 30 kernels), yzao 67 (73 for its 8), none in common;
- two finished pumps of azag (copies of existing files) were pushed as one archive with their job times, and counted as done;
- yzao's release gave its 67 pumps back to the pool.
Still to test: a deadline passing, and a real run through the driver.
Campaigns
A campaign (a sakusen) is one set of pumps to compute. With a single one in the pool, every command takes it; with several, --sakusen C names the one --senkyo, --honkai, watch or --shushu is about.
A campaign is declared once, by ten-go --sensen SPEC.json, from a short spec file: its name, file prefix and parameters, and either
- a grid: one or more lattices (from, to, step), the orders by range of $P_\sigma$, ranges left out, and the runs already computed (directories or archives), whose pumps are left out, each pump read from its files' header, its exact $P_\sigma$, whatever the numbering of the run; or
- an adaptive campaign: the rule of the adaptive driver (the lattice, $\delta$, the ceiling of the spacing, the window in $|\omega_p|$, the certificate's tolerances), and the runs it starts from. Their files enter the pool as the results of an allocation seed, the pumps left half done included, to be completed; then at each claim the rule, in the kit, is applied to the results in the pool and to the pumps still to come, and the midpoints (or quarter points) of the intervals still too wide are added to the campaign, worst first, in the same push as the claim, so that two machines cannot add the same pump. The campaign grows as the results come, and ends by itself when every interval is closed or at the lattice step. Short claims suit it, since it is refined only when a claim is made.
A campaign is never redeclared: its pump numbers are what the allocations point to. Tried on the copy of the pool on 27 October:
- close-0.004 rebuilt from a spec (the $0.004g$ lattice from 0.150 to $12.51g$, $5\le P_\sigma/g\le 6$ left out, and the archives of Sec. 31's $0.02g$ grid and of yzao's two windows) is the recipe exactly: the same 1,952 pumps, with the same orders and in the same global order, the 889 others read as computed;
- the boiling window at $0.08g$ (orders 300e, 250e and 150w, the colours at 80 digits from $25g$, as in the refinement run), with the runs at $0.32g$ and its refinement as computed: 118 pumps left (12.59–$42.83g$) of its 381, 263 of them done;
- an adaptive campaign seeded with azag's run on $5.2\le P_\sigma/g\le 5.6$ (2,159 files, 172 MB, three archives of 18 MB): 722 pumps, 713 complete; its first claim, at $\delta=0.001g$ on 30 kernels, added 395 pumps, the midpoints of the 395 intervals to split, and took 68 of them for four hours: first the nine left half done, with their missing orders only, then $J=84896$ ($5.531625g$) and $J=2224$, as the adaptive driver's own queue on the same files. A second claim added nothing, every interval to split waiting on a midpoint already in the campaign.
| Campaign | Size | Cost |
|---|---|---|
| close $0.15\le P_\sigma/g\le 12.51$ at $0.004g$ | 1,952 pumps | about 2,170 kernel-hours (azag on 30 kernels) |
| the boiling window $12.51\le P_\sigma/g\le 42.91$ at $0.08g$ | 118 pumps left | about 190 kernel-hours |
| adaptive refinements (such as $5.2\le P_\sigma/g\le 5.6$) | grown as the gaps are measured | open |
Listing
#!/usr/bin/env python3
"""Operation Ten-Go v1.16 (lw page [[Ten-Go]]): several machines share one pool of pump jobs through a GitHub repository (public,
github.com/laussy/ten-go), and no pump is ever computed twice. A machine asks for a BUDGET, not for pumps: `ten-go --honkai --jikan 10` takes, in one
go, the free pumps of the campaign in its global (van der Corput) order until their cost on this machine fills the ten hours, the
longest job kept back so that the last jobs launched still end in time (a single pump if the budget is very small). The claim is a
commit pushed to the repository; a push refused because another machine pushed first is redone from the new state, so that no pump
can be in two allocations. An allocation has a deadline (its start + 1.5 budget + 2 h): the pumps it has not done by then go back to
the pool. The jobs are run by run_window.py (kit/), one work directory per allocation.
Repository (the clone is Ten-Go's alone: every command starts by resetting it to origin/main):
campaigns/<c>.json params, file prefix, reference seconds per order, pumps [id, P, orders] in the global order
machines/<m>.json kernels, speed factor against the reference
allocs/<c>/<id>.json one allocation: machine, budget, deadline, pumps, done, measured job times, results, status
results/<c>/<id>-<n>.tgz the .dat files of the pumps done, one archive per push
ledger.tsv the kernel-hours (kh) spent, per task: one line per allocation (brought up to date by --tassei) or per run log
imported (any run of run_window.py, in Ten-Go or not), or booked by hand
The command is `ten-go`, a link to this file on the PATH (ln -s CLONE/tengo.py ~/bin/ten-go). Local: ~/.config/tengo.json {"machine", "repo", "work"} (written by `ten-go --kido`; env TENGO_CONF names another file).
--until TAG, an AF tag (read back by ~/bin/AF, lw page [[AF]]) or 'YYYY-MM-DD HH:MM', ends a claim or a watch at that moment: --honkai
takes the hours from now until then in place of --jikan. Given both --jikan and --until, --honkai only prints itself again, its --jikan
set to those hours and --until dropped, with the kernel-hours (kh) they make as a comment.
--shinshu [N] (--honkai, launch, watch) lets the allocation take the threads that stay idle, from its kernels up to N (no N: every
thread of the machine), and give them back as soon as other work wants them: the driver looks every minute (kit/run_window.py), one
kernel more after 10 minutes of threads free of any other work, back toward the allocation's kernels after 2 minutes of activity; a
kernel above them runs the shortest job waiting, so that the retreat is quick. The budget is unchanged: shinshu only ends it sooner.
ten-go --uchikiri ends this machine's allocations as soon as possible without loss: each driver launches
no new pump and completes those started (a job done or in flight), and says when it will end, and what a STOP or a kill would waste
instead; the pumps not started go back to the pool at once, a watch of the machine claims nothing more, and the driver pushes the last
results when it ends. With --dry, it only says when.
ten-go --chushi [MIN] stops them now (there is no other stop), but a job within MIN minutes of completion (10) is first offered to wait
for: "One job is 3 min from completion; the next are 17 min, 2 h 20, 6 h away, so those are aborted. Abort this one too [n]ow, or
[a]fter it completes?" --now or --wait answers ahead (without a terminal, one of them is needed), --dry only asks. The pumps not
complete go back to the pool at once, and those the jobs waited for complete are pushed when they end.
--shugeki (with --honkai or watch) is a sortie, for a machine that may be stopped any moment: it claims the cheapest pumps first
(their cost on this machine, the global order among equals) and its driver runs them whole, one after the other (no shortest job
taken ahead for a kernel above the allocation's), so that an uchikiri ends within a job or two and a chushi wastes little.
--sensen SPEC.json declares a campaign from a spec file (see declare): a grid, whose pumps already computed in the runs it names are
left out, or an adaptive campaign, whose pumps grow at each claim by the rule of adaptwin.py (kit/tengo_adaptive.py) applied to the results
in the pool, seeded with the pumps of the runs it names (they go to the pool as the results of an allocation "seed"). Short claims
suit an adaptive campaign: it is refined only when a claim is made.
A kh is one kernel for one hour: the jobs' own times summed, a failed job's included (an adopted job's time, unknown, is not). The
ledger keeps what each task cost: `ten-go kh` shows the totals per task and machine, `ten-go kh TASK` the lines of a task.
Usage: ten-go --kido MACHINE --repo CLONE --work DIR | --senkyo [--sakusen C] | --honkai --jikan H|--until TAG [--sakusen C] [--ki N] [--go] [--shinshu [N]] [--shugeki]
| --tassei [--alloc ID] | --uchikiri [--dry] [--alloc ID] | --chushi [MIN] [--now|--wait|--dry] [--alloc ID] | launch [--alloc ID] [--shinshu [N]] | watch --until TAG [--jikan 10] [--sakusen C] [--shinshu [N]] [--shugeki]
| --shushu [DIR] [--sakusen C] (unpack every result into DIR) | --sensen NAME|SPEC.json (declare a campaign)
| kh [TASK] | kh import DIR [TASK] [--machine M] [--note TEXT] | kh add TASK KH [--machine M] [--note TEXT]"""
import argparse, glob, json, os, random, re, shlex, shutil, subprocess, sys, tarfile, time
from fractions import Fraction
VERSION = "1.16" # 1.1: claim --go (was --launch) starts the driver; 1.2: the command is ten-go (a link to this file); 1.3: --until TAG; 1.4: kh, the ledger; 1.5: --shinshu; 1.6: --uchikiri; 1.7: --chushi; 1.8: claim is --honkai, release is gone (--chushi); 1.9: --kernels is --ki; 1.10: --campaign (-c) is --sakusen; 1.11: init is --kido, --hours is --jikan; 1.12: status is --senkyo, done --tassei; 1.13: fetch is --shushu; 1.14: --shugeki; 1.15: mkcampaign is --sensen; 1.16: spec files, adaptive campaigns
CONF = os.environ.get("TENGO_CONF", os.path.expanduser("~/.config/tengo.json")) # another file for a test identity
C = {}
def die(msg): sys.exit(f"ten-go: {msg}")
def R(*p): return os.path.join(C["repo"], *p)
def stamp(t): return time.strftime("%a %d %b %H:%M", time.localtime(t))
def moment(s):
"""the time of --until: an AF tag, read back by ~/bin/AF (not a copy of its decoder here), or 'YYYY-MM-DD HH:MM'"""
if re.fullmatch(r"[0-9A-Za-z]{5,6}", s):
try: r = subprocess.run([shutil.which("AF") or os.path.expanduser("~/bin/AF"), "-b", s], capture_output=True, text=True)
except FileNotFoundError: die(f"--until {s}: reading an AF tag needs ~/bin/AF")
if r.returncode: die(f"--until {s}: {r.stderr.strip()}")
return int(r.stdout.split("\t")[0])
try: return time.mktime(time.strptime(s, "%Y-%m-%d %H:%M"))
except ValueError: die(f"--until {s}: neither an AF tag nor 'YYYY-MM-DD HH:MM'")
def hours_until(s):
t = moment(s); h = int((t - time.time()) / 36) / 100 # floored to 0.01 h
if h <= 0: die(f"--until {s}: {stamp(t)} is past")
return h, t
def restated(h):
"""the command line as typed, its --jikan set to h and --until dropped (an option as is or with =; no abbreviations)"""
out, args = ["ten-go"], iter(sys.argv[1:])
for x in args:
k = x.split("=", 1)[0]
if k == "--jikan": out += ["--jikan", f"{h:g}"]
elif k == "--until": pass
else: out.append(x); continue
if "=" not in x: next(args, None)
return " ".join(map(shlex.quote, out))
def git(*a, check=True):
r = subprocess.run(["git", "-C", C["repo"], *a], capture_output=True, text=True)
if check and r.returncode: die(f"git {' '.join(a)} failed:\n{r.stderr.strip()}")
return r
def sync(): git("fetch", "-q", "origin"); git("reset", "-q", "--hard", "origin/main")
def transact(change, msg):
"""apply `change` to the fresh state of the repository (it returns what it did, or None when there is nothing to do) and push; a
refused push (another machine pushed first) is redone from the new state, so that a claim is recomputed from what is still free"""
for attempt in range(10):
sync(); out = change()
if out is None: return None
git("add", "-A"); git("commit", "-q", "-m", msg(out))
if git("push", "-q", "origin", "HEAD:main", check=False).returncode == 0: return out
print("ten-go: push refused (another machine pushed first), redoing from the new state", file=sys.stderr)
time.sleep(random.uniform(2, 10))
die("could not push after 10 attempts")
def save(path, obj):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f: json.dump(obj, f, indent=1); f.write("\n")
def campaigns(): return sorted(os.path.basename(f)[:-5] for f in glob.glob(R("campaigns", "*.json")))
def campaign(c): return json.load(open(R("campaigns", c + ".json")))
def allocs(c): return [json.load(open(f)) for f in sorted(glob.glob(R("allocs", c, "*.json")))]
def machine(m): return json.load(open(R("machines", m + ".json")))
def workdir(a): return os.path.join(C["work"], a["campaign"], a["id"])
def parse(o): m = re.fullmatch(r"(\d+)([we])(\d*)", o); return int(m[1]), int(m[2] == "w"), m[3] # "150w80": n_t 150, weights, 80 digits
def kh(s): return f"{s / 3600:.1f}"
def dec(x): # an exact pump as a decimal, exact on these lattices (else a/b, which hp3.wls reads as well)
x = Fraction(x)
for n in range(31):
if (x * 10 ** n).denominator == 1:
d = str(abs(x.numerator * 10 ** n // x.denominator)).rjust(n + 1, "0"); return ("-" if x < 0 else "") + (d[:-n] + "." + d[-n:] if n else d)
return f"{x.numerator}/{x.denominator}"
def upto(v): return "every thread" if v == "max" else f"{v} kernels"
def ceiling(v, kern): # --shinshu: None, max (every thread of the machine), or a number of kernels above kern
if v in (None, "max"): return v
try: n = int(v)
except ValueError: die(f"--shinshu {v}: a number of kernels, or nothing for every thread of the machine")
return n if n > kern else die(f"--shinshu {n}: not above the {kern} kernels of the allocation")
def tilde(p): h = os.path.expanduser("~"); return "~" + p[len(h):] if p.startswith(h + "/") else p
LEDGER = ["task", "machine", "source", "jobs", "failed", "kh", "first", "last", "note"] # first and last job done
def ledger(): return [dict(zip(LEDGER, l.rstrip("\n").split("\t"))) for l in open(R("ledger.tsv")).readlines()[1:]] if os.path.exists(R("ledger.tsv")) else []
def save_ledger(L):
with open(R("ledger.tsv"), "w") as f:
f.write("\t".join(LEDGER) + "\n") # a header GitHub shows as a table
for e in sorted(L, key=lambda e: (e["first"], e["source"])): f.write("\t".join(str(e[k]) for k in LEDGER) + "\n")
def logspent(W):
"""jobs, failed jobs and kernel-seconds of the jobs a run.log of run_window.py reports done, and when the first and the last of
them ended (the times of their files: the log has no dates)"""
n = bad = sec = 0; f = os.path.join(W, "run.log"); T = []
for l in (open(f) if os.path.exists(f) else []):
m = re.search(r" done (\S+) j=(\d+) P=\S+ in (\d+) s", l)
if not m: continue
n += 1; sec += int(m[3]); bad += "NO FILE" in l; d = os.path.join(W, f"{m[1]}_{m[2]}.dat")
if os.path.exists(d): T.append(os.path.getmtime(d))
T = T or [os.path.getmtime(f)] if n else [0]
return n, bad, sec, min(T), max(T)
def book(L, task, m, source, spent, note=""):
"""bring the line of `source` up to `spent` (from logspent); True when it changed"""
n, bad, sec, t0, t1 = spent; d = lambda t: time.strftime("%Y-%m-%d %H:%M", time.localtime(t)); e = next((e for e in L if e["source"] == source), None)
if e is None and n: L.append(dict(task=task, machine=m, source=source, jobs=n, failed=bad, kh=f"{sec / 3600:.2f}", first=d(t0), last=d(t1), note=note))
elif e and (str(n), f"{sec / 3600:.2f}") != (e["jobs"], e["kh"]): e.update(jobs=n, failed=bad, kh=f"{sec / 3600:.2f}", last=d(t1))
else: return False
return True
def pick(c):
cs = campaigns()
if c: return c if c in cs else die(f"no campaign {c} (there are: {', '.join(cs)})")
if len(cs) == 1: return cs[0]
die(f"say which campaign with --sakusen ({', '.join(cs)})")
def held(A, now):
"""the pumps out of the pool: those done, and all those of an open allocation before its deadline"""
out = set()
for a in A:
out |= set(a["done"])
if a["status"] == "open" and now < a["deadline"]: out |= set(a["pumps"])
return out
def secs(cm, A, m, o):
"""seconds of one job of order o on machine m: the median of its measured jobs in this campaign (5 at least), else the reference
of the campaign times the machine's factor"""
t = sorted(x for a in A if a["machine"] == m for x in a.get("times", {}).get(o, []))
return t[len(t) // 2] if len(t) >= 5 else cm["ref"][o] * machine(m).get("factor", 1.0)
def cost(cm, A, m):
S = {o: secs(cm, A, m, o) for o in cm["ref"]}
return S, {pid: sum(S[o] for o in orders) for pid, P, orders in cm["pumps"]}
# ----------------------------------------------------------------------------------------------------------------- the work directory
def setup(a):
cm = campaign(a["campaign"]); W = workdir(a); os.makedirs(W, exist_ok=True)
for f in ("hp3.wls", "run_window.py"): shutil.copy(R("kit", f), W)
P = {pid: (Pv, orders) for pid, Pv, orders in cm["pumps"]}
open(os.path.join(W, "pumps.txt"), "w").write("".join(f"{pid} {P[pid][0]}\n" for pid in a["pumps"]))
open(os.path.join(W, "params.txt"), "w").write("".join(f"{k}={v}\n" for k, v in cm["params"].items()))
open(os.path.join(W, "nproc"), "w").write(f"{a['kernels']}\n")
if a.get("shinshu"): open(os.path.join(W, "shinshu"), "w").write(f"{a['shinshu']}\n") # read by the driver every minute
if a.get("shugeki"): open(os.path.join(W, "shugeki"), "w").write("whole pumps, in the plan's order\n")
with open(os.path.join(W, "plan.txt"), "w") as f:
f.write(f"# prefix nt withWeights j [prec] (Ten-Go allocation {a['id']} of {a['campaign']}: {len(a['pumps'])} pumps in the global order)\n")
for pid in a["pumps"]:
for o in P[pid][1]:
nt, w, prec = parse(o); f.write(f"{cm['prefix']}_nt{o} {nt} {w} {pid}" + (f" {prec}" if prec else "") + "\n")
save(os.path.join(W, "alloc.json"), a)
return W
def ours(W):
"""pids of the driver and of the hp3.wls front-ends running in W"""
drv, fe = [], []
for d in glob.glob("/proc/[0-9]*"):
try:
cmd = open(d + "/cmdline").read().split("\0")
if os.readlink(d + "/cwd") != W: continue
except OSError: continue
if any(x.endswith("run_window.py") for x in cmd) and "python3" in os.path.basename(cmd[0]): drv.append(int(d[6:]))
elif "hp3.wls" in cmd and "wolframscript" in os.path.basename(cmd[0]): fe.append(int(d[6:]))
return drv, fe
def launch(a, quiet=False, shinshu=None):
W = workdir(a)
if not os.path.isdir(W): setup(a)
if shinshu: open(os.path.join(W, "shinshu"), "w").write(f"{ceiling(shinshu, a['kernels'])}\n")
if ours(W)[0]: return quiet or print(f"ten-go: the driver of {a['id']} is already running")
if os.path.exists(os.path.join(W, "STOP")): die(f"{W}/STOP exists (released or stopped by hand): remove it to run again")
for f in glob.glob(os.path.join(W, "*.dat")):
if os.path.getsize(f) < 1000: os.remove(f) # the leftover of a killed job
cmd = (["systemd-inhibit", "--what=sleep:idle", f"--why=Ten-Go {a['id']}"] if shutil.which("systemd-inhibit") else []) + ["python3", "run_window.py"]
subprocess.Popen(cmd, cwd=W, stdin=subprocess.DEVNULL, stdout=open(os.path.join(W, "run.log"), "a"), stderr=subprocess.STDOUT, start_new_session=True)
sh = open(os.path.join(W, "shinshu")).read().strip() if os.path.exists(os.path.join(W, "shinshu")) else ""
print(f"ten-go: launched {a['id']} in {W} on {a['kernels']} kernels" + (f", shinshu up to {upto(sh)}" if sh else ""))
# ----------------------------------------------------------------------------------------------------------------------- commands
def cmd_claim(c, hours, kernels=None, do_launch=False, shinshu=None, shugeki=False):
m = C["machine"]; kern = kernels or machine(m)["kernels"]; c = pick(c); sh = ceiling(shinshu, kern)
def change():
cm = campaign(c); A = allocs(c); now = time.time(); busy = held(A, now); pumps, grew = cm["pumps"], None
if cm.get("kind") == "adaptive": pumps, grew = grown(c, cm, A, busy, kern) # the rule, now: new pumps appended to cm
S, K1 = cost(cm, A, m)
take, tot, hold = [], 0.0, 0 # the budget in kernel-seconds, the longest job of the pumps taken kept back
for pid, P, orders in (sorted(pumps, key=lambda x: K1[x[0]]) if shugeki else pumps): # shugeki: the cheapest first
if pid in busy: continue
h = max(hold, *(S[o] for o in orders))
if take and tot + K1[pid] > kern * max(hours * 3600 - h, 0): break
take.append(pid); tot += K1[pid]; hold = h
if not take: return None
aid = base = f"{m}-{time.strftime('%Y%m%d-%H%M%S')}"; k = 1
while os.path.exists(R("allocs", c, aid + ".json")): k += 1; aid = f"{base}-{k}" # two claims in the same second
a = dict(id=aid, campaign=c, machine=m, kernels=kern, hours=hours, created=int(now), deadline=int(now + 1.5 * hours * 3600 + 7200),
cost_kernel_h=round(tot / 3600, 1), pumps=take, done=[], times={}, results=[], status="open", tengo=VERSION, shinshu=sh, shugeki=shugeki)
if grew: a["grown"], a["open"] = grew; save(R("campaigns", c + ".json"), cm) # the new pumps, in the claim's commit
save(R("allocs", c, aid + ".json"), a); return a
a = transact(change, lambda a: f"honkai {a['id']}: {len(a['pumps'])} pumps of {c}, {a['hours']:g} h on {a['kernels']} kernels")
if a is None: return print(f"ten-go: nothing free in {c}")
W = setup(a); P = {pid: Pv for pid, Pv, _ in campaign(c)["pumps"]}; xs = sorted(float(Fraction(P[p])) for p in a["pumps"])
print(f"ten-go: claimed {a['id']}: {len(a['pumps'])} pumps of {c} ({xs[0]:g}-{xs[-1]:g} g, {f"{a['grown']} new by the rule, {a['open']} intervals to split now" if "grown" in a else 'the cheapest first, a sortie' if shugeki else 'spread in the global order'}), "
f"{a['cost_kernel_h']} kh for {hours:g} h on {kern} kernels; deadline {stamp(a['deadline'])}; work directory {W}"
+ (f"; shinshu up to {upto(sh)}" if sh else ""))
if do_launch: launch(a)
return a
def complete(a, cm):
W = workdir(a); O = {pid: orders for pid, _, orders in cm["pumps"]}
ok = lambda f: os.path.exists(f) and os.path.getsize(f) > 1000
return [p for p in a["pumps"] if all(ok(os.path.join(W, f"{cm['prefix']}_nt{o}_{p}.dat")) for o in O[p])]
def jobtimes(W, prefix, pids):
out = {}
if os.path.exists(os.path.join(W, "run.log")):
for l in open(os.path.join(W, "run.log")):
m = re.search(rf"done {re.escape(prefix)}_nt(\w+) j=(\d+) P=\S+ in (\d+) s", l)
if m and int(m[2]) in pids and "NO FILE" not in l: out.setdefault(m[1], []).append(int(m[3]))
return out
def cmd_done(aid=None, final=None):
"""push the pumps finished since the last push (their files as one archive), their job times, and the status of the allocation"""
def change():
out = []; L = ledger(); booked = False
for c in campaigns():
cm = campaign(c); O = {pid: orders for pid, _, orders in cm["pumps"]}
for a in allocs(c):
if a["machine"] != C["machine"] or a["status"] != "open" or (aid and a["id"] != aid) or not os.path.isdir(workdir(a)): continue
W = workdir(a); have = set(a["done"]); new = [p for p in complete(a, cm) if p not in have]
if new:
rel = f"results/{c}/{a['id']}-{len(a['results']) + 1}.tgz"; os.makedirs(R("results", c), exist_ok=True)
with tarfile.open(R(rel), "w:gz") as t:
for p in new:
for o in O[p]: n = f"{cm['prefix']}_nt{o}_{p}.dat"; t.add(os.path.join(W, n), arcname=n)
a["results"].append(rel); a["done"] += new
for o, v in jobtimes(W, cm["prefix"], set(new)).items(): a["times"].setdefault(o, []).extend(v)
st = final or ("finished" if set(a["done"]) >= set(a["pumps"]) else "open")
sp = logspent(W); b = book(L, c, a["machine"], "alloc " + a["id"], sp); booked |= b
if new or st != a["status"]:
a["status"] = st; a["updated"] = int(time.time()); save(R("allocs", c, a["id"] + ".json"), a)
out.append(f"{a['id']}: +{len(new)} pumps ({len(a['done'])}/{len(a['pumps'])} done), {st}, {kh(sp[2])} kh spent")
elif b: out.append(f"{a['id']}: {kh(sp[2])} kh spent")
if booked: save_ledger(L)
return out or None
out = transact(change, lambda o: "done: " + "; ".join(o))
print("ten-go: " + ("; ".join(out) if out else "nothing new to push"))
def ending(): open(os.path.join(C["work"], "ENDING"), "w").write(f"{int(time.time())}\n") # a watch of this machine claims no more
def knows(W, opt): return opt in open(os.path.join(W, "run_window.py")).read() # an older driver would take it for a plain run
def atend(a, W): # the driver of a pushes its last results when it ends
open(os.path.join(W, "atend"), "w").write(shlex.join(["env", f"TENGO_CONF={CONF}", sys.executable, os.path.realpath(__file__), "--tassei", "--alloc", a["id"]]) + "\n")
def shrink(a, kept, why):
"""give back to the pool at once the pumps of allocation a not in kept (those its driver no longer completes); how many"""
def change():
x = json.load(open(R("allocs", a["campaign"], a["id"] + ".json")))
back = [p for p in x["pumps"] if str(p) not in kept and p not in x["done"]]
if x["status"] != "open" or not back: return None
x["pumps"] = [p for p in x["pumps"] if p not in back]; x[why] = int(time.time()); save(R("allocs", a["campaign"], a["id"] + ".json"), x)
return len(back)
return transact(change, lambda n: f"{why} {a['id']}: {n} pumps back to the pool") or 0
def mine(aid): return [a for c in campaigns() for a in allocs(c) if a["machine"] == C["machine"] and a["status"] == "open" and (not aid or a["id"] == aid)]
def cmd_uchikiri(aid=None, dry=False):
"""end this machine's open allocations as soon as possible without loss (run_window.py --uchikiri, which says when); the pumps not
started go back to the pool at once, and the driver pushes the last results when it ends (its file atend)"""
sync(); M = mine(aid)
if not M: return print("ten-go: no open allocation of this machine")
if not dry: ending()
for a in M:
W = workdir(a)
if not (os.path.isdir(W) and ours(W)[0]): print(f"ten-go: {a['id']}: no driver running (`ten-go --chushi` gives its pumps back)"); continue
if not knows(W, "--uchikiri"): print(f"ten-go: {a['id']}: its driver predates uchikiri (`ten-go --chushi` stops it)"); continue
if not dry: atend(a, W)
r = subprocess.run(["python3", "run_window.py", "--uchikiri"] + (["--dry"] if dry else []), cwd=W, capture_output=True, text=True)
say = (r.stdout + r.stderr).strip()
if dry or r.returncode: print(f"ten-go: {a['id']}: {say}"); continue
n = shrink(a, set(open(os.path.join(W, "uchikiri.kept")).read().split()), "uchikiri") # the pumps started, as the driver saw them
print(f"ten-go: {a['id']}: {n} pumps back to the pool at once; {say}; the driver pushes the last results when it ends")
def cmd_chushi(aid=None, soon=10, how=None):
"""stop this machine's open allocations now (run_window.py --chushi), a job within `soon` minutes of completion being first offered
to wait for (how: now, wait or dry answers ahead); the pumps not complete go back to the pool at once, and those the jobs waited for
complete are pushed when they end"""
sync(); M = mine(aid)
if not M: return print("ten-go: no open allocation of this machine")
if how != "dry": ending()
for a in M:
W = workdir(a); run = os.path.isdir(W) and any(ours(W))
print(f"ten-go: {a['id']}:", flush=True)
if run and not knows(W, "--chushi"): # an older driver: stopped at once, without the warning
if how == "dry": print("its driver predates chushi: it would be stopped at once"); continue
subprocess.run(["python3", "run_window.py", "--kill"], cwd=W); cmd_done(a["id"], final="released"); continue
if how == "dry":
if run: subprocess.run(["python3", "run_window.py", "--chushi", "--soon", f"{soon:g}", "--dry"], cwd=W)
else: print("no job running")
continue
if run and subprocess.run(["python3", "run_window.py", "--chushi", "--soon", f"{soon:g}"] + ([f"--{how}"] if how else []), cwd=W).returncode: continue
if os.path.isdir(W) and ours(W)[0]: # its driver waits for the jobs about to complete
atend(a, W); n = shrink(a, set(open(os.path.join(W, "chushi.kept")).read().split()), "chushi")
print(f"ten-go: {a['id']}: {n} pumps back to the pool at once; the driver pushes the last results when its jobs end")
else: cmd_done(a["id"], final="released")
def cmd_status(c=None):
sync(); now = time.time()
for c in ([pick(c)] if c else campaigns()):
cm = campaign(c); A = allocs(c); N = len(cm["pumps"])
done = set().union(*(set(a["done"]) for a in A)) if A else set(); hold = held(A, now) - done
_, K1 = cost(cm, A, "azag" if os.path.exists(R("machines", "azag.json")) else C["machine"])
left = sum(K1[p] for p, _, _ in cm["pumps"] if p not in done) / 3600; b = int(40 * len(done) / N)
print(f"{c}: [{'#' * b}{'.' * (40 - b)}] {len(done)}/{N} pumps done, {len(hold)} held, {N - len(done) - len(hold)} free; "
f"{left:.0f} kh left at the reference speed; {sum(float(e['kh']) for e in ledger() if e['task'] == c):.1f} kh spent")
for m in sorted({a["machine"] for a in A}):
if m == "seed": print(f" seed: {sum(len(a['done']) for a in A if a['machine'] == m)} pumps from the runs before Ten-Go"); continue
op = [a for a in A if a["machine"] == m and a["status"] == "open"]
S, Km = cost(cm, A, m); rest = sum(Km[p] for a in op for p in a["pumps"] if p not in set(a["done"]))
kern = max([a["kernels"] for a in op] or [machine(m)["kernels"]])
print(f" {m}: {sum(len(a['done']) for a in A if a['machine'] == m)} pumps done in all, "
f"{sum(float(e['kh']) for e in ledger() if e['task'] == c and e['machine'] == m):.1f} kh; "
+ (f"holds {sum(len(a['pumps']) - len(a['done']) for a in op)}, {rest / 3600:.0f} kh, end about {stamp(now + rest / kern)}" if op else "holds nothing"))
for a in sorted(A, key=lambda a: a["created"])[-12:]:
late = a["status"] == "open" and now > a["deadline"]
print(f" {a['id']:24s} {len(a['done']):4d}/{len(a['pumps']):<4d} {a['status']:9s} {a['hours']:g} h x {a['kernels']}"
+ (f" (shinshu up to {upto(a['shinshu'])})" if a.get("shinshu") else "") + (" (shugeki)" if a.get("shugeki") else "") + f", deadline {stamp(a['deadline'])}"
+ (f", uchikiri {stamp(a['uchikiri'])}" if a.get("uchikiri") else "")
+ (" PASSED: its pumps not done are free again" if late else ""))
def cmd_kh(args, mach=None, note=""):
"""the ledger: totals per task and machine (kh), the lines of one task (kh TASK), a run log booked (kh import DIR [TASK]: again
later, its line is brought up to date), or a run without a log booked by hand (kh add TASK KH)"""
m = mach or C["machine"]
if args and args[0] == "import":
if len(args) not in (2, 3): die("kh import DIR [TASK]")
D = os.path.abspath(os.path.expanduser(args[1])); task = args[2] if len(args) == 3 else os.path.basename(D); sp = logspent(D)
if not sp[0]: die(f"no job done in {tilde(D)}/run.log")
def change():
L = ledger()
if not book(L, task, m, f"log {m}:{tilde(D)}", sp, note): return None
save_ledger(L); return True
r = transact(change, lambda r: f"kh: {task} on {m}, {tilde(D)}: {sp[0]} jobs, {kh(sp[2])} kh")
print(f"ten-go: {task} on {m}: {sp[0]} jobs ({sp[1]} failed), {kh(sp[2])} kh" + ("" if r else ", already in the ledger"))
elif args and args[0] == "add":
if len(args) != 3: die("kh add TASK KH")
task, now = args[1], time.strftime("%Y-%m-%d %H:%M")
try: x = float(args[2])
except ValueError: die(f"kh add {task} {args[2]}: KH is a number of kernel-hours")
def change():
L = ledger(); L.append(dict(task=task, machine=m, source=f"by hand {now}", jobs="", failed="", kh=f"{x:.2f}", first=now, last=now, note=note))
save_ledger(L); return True
transact(change, lambda r: f"kh: {task} on {m}, {x:g} kh by hand"); print(f"ten-go: {task} on {m}: {x:g} kh booked by hand")
else:
sync(); L = ledger()
if not L: return print("ten-go: the ledger is empty")
if args:
E = [e for e in L if e["task"] == args[0]] or die(f"no task {args[0]} in the ledger")
for e in E: print(f" {e['machine']:6s} {float(e['kh']):9.1f} kh {e['jobs']:>6s} jobs {e['first']} to {e['last']} {e['source']} {e['note']}")
return print(f" {'all':6s} {sum(float(e['kh']) for e in E):9.1f} kh")
M = sorted({e["machine"] for e in L}); T = sorted({e["task"] for e in L}, key=lambda t: min(e["first"] for e in L if e["task"] == t))
tot = lambda t=None, mm=None: sum(float(e["kh"]) for e in L if (t is None or e["task"] == t) and (mm is None or e["machine"] == mm))
w = max(len(t) for t in T + ["task"]); print(f"kernel-hours spent (ledger.tsv, {len(L)} lines)")
print(f" {'task':{w}s}" + "".join(f"{mm:>9s}" for mm in M) + f"{'all':>9s} first job")
for t in T: print(f" {t:{w}s}" + "".join(f"{tot(t, mm):9.1f}" if tot(t, mm) else f"{'-':>9s}" for mm in M) + f"{tot(t):9.1f} {min(e['first'] for e in L if e['task'] == t)[:10]}")
print(f" {'all':{w}s}" + "".join(f"{tot(None, mm):9.1f}" for mm in M) + f"{tot():9.1f}")
def tail_state(W):
"""jobs still waiting in the driver of W (from its log) and front-ends running"""
wait = None
if os.path.exists(os.path.join(W, "run.log")):
for l in open(os.path.join(W, "run.log")):
m = re.search(r"(\d+) jobs to run|(\d+) waiting", l)
if m: wait = int(m[1] or m[2])
return wait, len(ours(W)[1])
def cmd_watch(until, hours, c, shinshu=None, shugeki=False):
"""the unattended loop: claim and launch an allocation while time remains before `until`, push the pumps done every 30 min,
restart a dead driver (3 times per allocation at most), and claim the next allocation as soon as the current one has launched all
its jobs, its tail then lending its kernels to the new one (the nproc of the new driver = kernels - jobs of the tail)"""
end = moment(until); restarts = {}; last = 0; c = pick(c); t0 = time.time(); flag = os.path.join(C["work"], "ENDING")
print(f"ten-go watch until {stamp(end)} ({C['machine']}, {c})", flush=True)
while True:
now = time.time()
if end > now and os.path.exists(flag) and os.path.getmtime(flag) > t0:
end = now; print(f"{time.strftime('%H:%M:%S')} uchikiri or chushi: no more claims, the watch ends with its drivers", flush=True)
if now - last > 1800: cmd_done(); last = now
sync(); op = [a for a in allocs(c) if a["machine"] == C["machine"] and a["status"] == "open"]
for a in op:
W = workdir(a); log = open(os.path.join(W, "run.log")).read() if os.path.exists(os.path.join(W, "run.log")) else ""
if not ours(W)[0] and "ALLDONE" not in log.split("tengo-launch")[-1] and not os.path.exists(os.path.join(W, "STOP")):
if restarts.get(a["id"], 0) >= 3: continue
restarts[a["id"]] = restarts.get(a["id"], 0) + 1; print(f"{time.strftime('%H:%M:%S')} driver of {a['id']} not running: start {restarts[a['id']]}", flush=True)
open(os.path.join(W, "run.log"), "a").write(f"{time.strftime('%H:%M:%S')} tengo-launch (watch)\n"); launch(a, quiet=True)
tails = [(a, *tail_state(workdir(a))) for a in op]
fresh = [a for a, wait, run in tails if wait] # an allocation still launching jobs
if not fresh and end - now > 1800:
busy = sum(run for _, _, run in tails); kern = machine(C["machine"])["kernels"]
a = cmd_claim(c, min(hours, (end - now) / 3600), kern, shinshu=shinshu, shugeki=shugeki)
if a:
open(os.path.join(workdir(a), "nproc"), "w").write(f"{max(kern - busy, 1)}\n"); launch(a)
if fresh and len(op) > 1: # the newest allocation takes the kernels the tails free
busy = sum(run for a, wait, run in tails if not wait); kern = machine(C["machine"])["kernels"]
for a in fresh: open(os.path.join(workdir(a), "nproc"), "w").write(f"{max(kern - busy, 1)}\n")
if now > end and not any(run for _, _, run in tails): cmd_done(); print(f"{time.strftime('%H:%M:%S')} watch ends", flush=True); return
time.sleep(60)
def cmd_fetch(c, dest):
sync(); c = pick(c); os.makedirs(dest, exist_ok=True); n = 0
for f in sorted(glob.glob(R("results", c, "*.tgz"))):
with tarfile.open(f) as t: t.extractall(dest, filter="data"); n += len(t.getnames())
print(f"ten-go: {n} files of {c} in {dest}")
# ------------------------------------------------------------------------------------------------------------------------ campaigns
REF = {"150w": 1118, "150w80": 1409, "200e": 760, "250e": 1580, "300e": 2685, "400e": 11076} # seconds per job, azag at home, 30 kernels
def computed(sources):
"""{exact pump: set of orders} of the .dat files of these runs (directories or .tgz archives), the pump read from each file's
header, its exact x, whatever the run's numbering; a file under 1000 bytes is a kill's leftover"""
out = {}
def add(name, size, first):
m = re.search(r"_nt(\d+[we]\d*)_\d+\.dat$", name)
if m and size > 1000: out.setdefault(Fraction(first.split()[1]), set()).add(m[1])
for src in sources:
if os.path.isdir(src):
for f in glob.glob(os.path.join(src, "*.dat")): add(f, os.path.getsize(f), open(f).readline())
else:
with tarfile.open(src) as t:
for x in t:
if x.isfile() and x.name.endswith(".dat"): add(x.name, x.size, t.extractfile(x).readline().decode())
return out
def declare(path):
"""a campaign from a spec file (JSON; its relative paths from the file's directory):
{"name", "about", "prefix", "params": {GA, GS, PREC, MEMMIN}, "ref": {order: seconds} (default REF), and either
"kind": "grid", "grid": [[from, to, step], ...], "orders": [[lo, hi, [orders]], ...] (for P in [lo, hi), hi may be "inf"),
"except": [[lo, hi], ...], "have": [runs]: a pump whose orders are all in these runs (directories or .tgz archives) is left out;
or "kind": "adaptive", "rule": {x0, x1, h0, lev, delta, hmax, wmax, eps, src, orders}, "seed": [run directories]: the pumps of
these runs (files <prefix>_nt<order>_<J>.dat on the rule's lattice) are its first pumps, those complete done (their files go
to the pool as the results of an allocation "seed"), the others to be completed}
returns the campaign and the seed (None, or the seed allocation with the files of its archives)"""
sp = json.load(open(path)); here = os.path.dirname(os.path.abspath(path))
at = lambda f: os.path.join(here, os.path.expanduser(f))
if not re.fullmatch(r"[\w.-]+", sp.get("name", "")): die(f"{path}: a name of letters, digits, . - _")
cm = dict(name=sp["name"], prefix=sp["prefix"], about=sp.get("about", ""), params=sp["params"], ref=sp.get("ref", REF), kind=sp.get("kind", "grid"))
if cm["kind"] == "grid":
pts = set()
for a, b, st in sp["grid"]:
a, b, st = Fraction(str(a)), Fraction(str(b)), Fraction(str(st)); pts |= {a + i * st for i in range(int((b - a) / st) + 1)}
pts = {P for P in pts if not any(Fraction(str(lo)) <= P <= Fraction(str(hi)) for lo, hi in sp.get("except", []))}
def orders(P):
for lo, hi, o in sp["orders"]:
if Fraction(str(lo)) <= P and (hi == "inf" or P < Fraction(str(hi))): return o
die(f"{path}: no orders for P = {dec(P)}")
have = computed([at(f) for f in sp.get("have", [])])
keep = sorted(P for P in pts if not set(orders(P)) <= have.get(P, set()))
cm["pumps"] = [[i, dec(keep[i]), orders(keep[i])] for i in vdc(len(keep))]
cm["left_out"] = len(pts) - len(keep); seed = None
elif cm["kind"] == "adaptive":
r = cm["rule"] = sp["rule"]; x0 = Fraction(str(r["x0"])); step = Fraction(str(r["h0"])) / 2 ** r["lev"]; have, files = {}, []
for d in sp.get("seed", []):
for f in sorted(glob.glob(os.path.join(at(d), f"{cm['prefix']}_nt*_*.dat"))):
m = re.search(rf"{re.escape(cm['prefix'])}_nt(\d+[we]\d*)_(\d+)\.dat$", f)
if not m or os.path.getsize(f) <= 1000: continue
if Fraction(open(f).readline().split()[1]) != x0 + int(m[2]) * step: die(f"{f}: not on the rule's lattice")
have.setdefault(int(m[2]), set()).add(m[1]); files.append(f)
full = sorted(J for J, o in have.items() if set(r["orders"]) <= o)
cm["pumps"] = [[J, dec(x0 + J * step), list(r["orders"]) if J in full else [o for o in r["orders"] if o not in have[J]]] for J in sorted(have)]
seed = dict(id="seed", campaign=cm["name"], machine="seed", kernels=0, hours=0, created=int(time.time()), deadline=int(time.time()),
cost_kernel_h=0, pumps=full, done=full, times={}, results=[], status="finished", tengo=VERSION,
about=f"the pumps of {', '.join(sp.get('seed', []))} before Ten-Go"), files
else: die(f"{path}: kind grid or adaptive")
for p in cm["pumps"]:
if not set(p[2]) <= set(cm["ref"]): die(f"{path}: no reference time for the orders {set(p[2]) - set(cm['ref'])}")
return cm, seed
def seeded(cm, seed): # the seed allocation and its archives (60 MB of files at most each, under GitHub's 100 MB a file)
a, files = seed; n, part = 0, []
for f in files + [None]:
if f and sum(os.path.getsize(x) for x in part) + os.path.getsize(f) < 60e6: part.append(f); continue
if part:
n += 1; rel = f"results/{cm['name']}/seed-{n}.tgz"; os.makedirs(R("results", cm["name"]), exist_ok=True)
with tarfile.open(R(rel), "w:gz") as t:
for x in part: t.add(x, arcname=os.path.basename(x))
a["results"].append(rel)
part = [f] if f else []
save(R("allocs", cm["name"], "seed.json"), a)
def cache(c): # this machine's copy of the results of campaign c, each archive unpacked once, for the rule of an adaptive campaign
D = os.path.join(C["work"], c, "cache"); os.makedirs(D, exist_ok=True); seen = os.path.join(D, ".unpacked")
done = set(open(seen).read().split()) if os.path.exists(seen) else set()
for f in sorted(glob.glob(R("results", c, "*.tgz"))):
if os.path.basename(f) in done: continue
with tarfile.open(f) as t: t.extractall(D, filter="data")
open(seen, "a").write(os.path.basename(f) + "\n")
return D
def grown(c, cm, A, busy, kern):
"""an adaptive campaign at a claim: its rule applied to the pumps whose 250e and 300e files are in the pool and to those still to
come; the new pumps are appended to cm; returns the pumps to offer (those free again first, then the new ones, worst interval
first) and (number new, intervals open)"""
sys.path[:0] = [R("kit"), os.path.dirname(os.path.realpath(__file__))]; import tengo_adaptive as adaptive # the pool's kit first
D = cache(c); have = {}
for f in os.listdir(D):
x = re.match(rf"{re.escape(cm['prefix'])}_nt(\w+)_(\d+)\.dat$", f)
if x and os.path.getsize(os.path.join(D, f)) > 1000: have.setdefault(int(x[2]), set()).add(x[1])
known = {J for J, o in have.items() if {"250e", "300e"} <= o}
done = set().union(*(set(a["done"]) for a in A)) if A else set()
waiting = {p for p, _, _ in cm["pumps"] if p not in done and p not in known}
new, nopen = adaptive.grow(cm["rule"], cm["prefix"], D, known, waiting, kern)
r = cm["rule"]; x0 = Fraction(str(r["x0"])); step = Fraction(str(r["h0"])) / 2 ** r["lev"]
cm["pumps"] += [[J, dec(x0 + J * step), list(r["orders"])] for J in new]
E = {p[0]: p for p in cm["pumps"]}
return [p for p in cm["pumps"] if p[0] not in done and p[0] not in busy and p[0] not in new] + [E[J] for J in new], (len(new), nopen)
def vdc(n):
m = max(1, (n - 1).bit_length()); return [r for r in (int(format(k, f"0{m}b")[::-1], 2) for k in range(2 ** m)) if r < n]
def mk_close004():
"""close 0.15-12.51 g at 0.004 g: P = 0.150 + 0.004 k (thousandths = 2 mod 4, the sublattice of Sec. 31), without the pumps of
Sec. 31 (k = 0 mod 5), the windows of yzao (0.502-1.498, 9.002-9.598 g) and the span 5-6 g of Sec. 32 (its 0.004 g grid is the
other sublattice); below 5 g n_t = 150 with weights and 200 (certificate B), above 150, 250 and 300 (D), as in Sec. 31"""
P = []
for k in range(3091):
th = 150 + 4 * k
if k % 5 == 0 or 502 <= th <= 1498 or 5000 <= th <= 6000 or 9002 <= th <= 9598: continue
P.append((f"{th // 1000}.{th % 1000:03d}", ["200e", "150w"] if th < 5000 else ["300e", "250e", "150w"]))
pumps = [[i, P[i][0], P[i][1]] for i in vdc(len(P))] # id = rank in pump, list = the global (van der Corput) order
return dict(name="close-0.004", prefix="hpT", about=mk_close004.__doc__.split("\n")[0].strip() + " ...",
params=dict(GA="1/10", GS="293/780364", PREC="40", MEMMIN="8"),
ref={"150w": 1118, "200e": 760, "250e": 1580, "300e": 2685}, pumps=pumps)
CAMPAIGNS = {"close-0.004": mk_close004}
def main():
ap = argparse.ArgumentParser(prog="ten-go", description="Operation Ten-Go: a shared pool of pump jobs in a git repository", allow_abbrev=False)
ap.add_argument("cmd", nargs="?"); ap.add_argument("arg", nargs="?"); ap.add_argument("more", nargs="*"); ap.add_argument("--sakusen", dest="campaign")
ap.add_argument("--jikan", dest="hours", type=float); ap.add_argument("--ki", type=int); ap.add_argument("--go", action="store_true")
ap.add_argument("--alloc"); ap.add_argument("--until"); ap.add_argument("--repo"); ap.add_argument("--work")
ap.add_argument("--machine"); ap.add_argument("--note", default=""); ap.add_argument("--shinshu", nargs="?", const="max")
ap.add_argument("--honkai", action="store_true"); ap.add_argument("--uchikiri", action="store_true"); ap.add_argument("--dry", action="store_true")
ap.add_argument("--chushi", nargs="?", const=10, type=float); ap.add_argument("--now", action="store_true"); ap.add_argument("--wait", action="store_true")
ap.add_argument("--kido", metavar="MACHINE"); ap.add_argument("--senkyo", action="store_true"); ap.add_argument("--tassei", action="store_true")
ap.add_argument("--shushu", nargs="?", const=".", metavar="DIR"); ap.add_argument("--shugeki", action="store_true")
ap.add_argument("--sensen", metavar="NAME")
OLD = {"--kernels": "--ki N, the kernels of the allocation", "-c": "--sakusen C, the campaign", "--campaign": "--sakusen C, the campaign",
"--hours": "--jikan H, the hours", "claim": "--honkai", "release": "--chushi, or --uchikiri", "init": "--kido MACHINE --repo CLONE --work DIR",
"status": "--senkyo", "done": "--tassei", "fetch": "--shushu [DIR]", "mkcampaign": "--sensen NAME"}
first = next((x for x in sys.argv[1:] if not x.startswith("-")), None) # an old command only as the command, an old option anywhere
for x in sys.argv[1:]:
if x.split("=")[0] in OLD and (x.startswith("-") or x is first): die(f"no {'option' if x.startswith('-') else 'command'} {x.split('=')[0]} any more: {OLD[x.split('=')[0]]}")
a = ap.parse_args()
if a.honkai and a.hours and a.until:
h = hours_until(a.until)[0]; k = a.ki
if not k:
try: C.update(json.load(open(CONF))); k = machine(C["machine"])["kernels"] # no sync: only this machine's kernels
except (OSError, KeyError, ValueError): pass
return print(restated(h) + (f" # {h:g} h x {k} kernels = {h * k:.1f} kh" if k else ""))
if a.cmd in ("-v", "version"): return print(f"ten-go {VERSION}")
opts = [o for o, on in (("--honkai", a.honkai), ("--uchikiri", a.uchikiri), ("--chushi", a.chushi is not None), ("--kido", a.kido is not None), ("--senkyo", a.senkyo), ("--tassei", a.tassei), ("--shushu", a.shushu is not None), ("--sensen", a.sensen is not None)) if on]
if len(opts) > 1: die(f"{' and '.join(opts)}: one at a time")
if opts and a.cmd: die(f"{opts[0]} stands alone, without a command ({a.cmd})")
if a.now and a.wait: die("--now or --wait, not both")
if not (opts or a.cmd): die(f"no command\n{__doc__}")
if a.kido is not None:
if not (a.kido and a.repo and a.work): die("ten-go --kido MACHINE --repo CLONE --work DIR")
save(CONF, dict(machine=a.kido, repo=os.path.abspath(os.path.expanduser(a.repo)), work=os.path.abspath(os.path.expanduser(a.work))))
return print(f"ten-go: {CONF} written ({a.kido})")
try: C.update(json.load(open(CONF)))
except FileNotFoundError: die(f"no {CONF}: run `ten-go --kido MACHINE --repo CLONE --work DIR` first")
if a.uchikiri: cmd_uchikiri(a.alloc, a.dry)
elif a.chushi is not None: cmd_chushi(a.alloc, a.chushi, "dry" if a.dry else "wait" if a.wait else "now" if a.now else None)
elif a.senkyo: cmd_status(a.campaign)
elif a.honkai:
if a.until: a.hours, t = hours_until(a.until); print(f"ten-go: until {a.until}, {stamp(t)}: {a.hours:g} h from now")
if not a.hours: die("--honkai --jikan H, or --until TAG")
cmd_claim(a.campaign, a.hours, a.ki, a.go, a.shinshu, a.shugeki)
elif a.tassei: cmd_done(a.alloc)
elif a.cmd == "launch":
sync(); op = [x for c in campaigns() for x in allocs(c) if x["machine"] == C["machine"] and x["status"] == "open" and (not a.alloc or x["id"] == a.alloc)]
for x in op: launch(x, shinshu=a.shinshu)
if not op: print("ten-go: no open allocation of this machine")
elif a.cmd == "watch":
if not a.until: die("watch --until TAG (an AF tag, or 'YYYY-MM-DD HH:MM')")
cmd_watch(a.until, a.hours or 10, a.campaign, a.shinshu, a.shugeki)
elif a.shushu is not None: cmd_fetch(a.campaign, a.shushu)
elif a.cmd == "kh": cmd_kh(([a.arg] if a.arg else []) + a.more, a.machine, a.note)
elif a.sensen is not None:
if a.sensen.endswith(".json"): cm, seed = declare(a.sensen)
elif a.sensen in CAMPAIGNS: cm, seed = dict(CAMPAIGNS[a.sensen](), name=a.sensen), None
else: die(f"--sensen {a.sensen}: a spec file (.json) or a known campaign ({', '.join(CAMPAIGNS)})")
name = cm["name"] # declared once: its pump ids are what the allocations point to, so it is never rewritten
def change():
if os.path.exists(R("campaigns", name + ".json")): return None
save(R("campaigns", name + ".json"), cm)
if seed: seeded(cm, seed)
return name
new = transact(change, lambda o: f"sensen {o}: {len(cm['pumps'])} pumps")
print(f"ten-go: campaign {name} ({cm.get('kind', 'grid')}), {len(cm['pumps'])} pumps"
+ (f", {cm['left_out']} left out as computed already" if cm.get("left_out") else "")
+ (f", {len(seed[0]['done'])} of them done (the seed, {len(seed[1])} files)" if seed else "")
+ ("" if new else ": already declared in the pool, left as it is"))
else: die(f"unknown command {a.cmd}\n{__doc__}")
if __name__ == "__main__":
main()