Hojun-Cho/ww
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wwtest: bounded package concurrency via exec start/poll

Browse Source
This commit is contained in:
Hojun-Cho
2026-08-08 13:43:55 +09:00
parent 0f0fd36563
commit 15137054e2
2 changed files with 162 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

211
internal/wwpackage/package.ww
View File

@@ -37,8 +37,25 @@ type pkggroup = struct {
builderr: str,
runout: str,
runerr: str,
state: i32,
fail: i32,
buildres: exec.result,
runres: exec.result,
};
// pkggroup.state values for the bounded scheduler.
def PKGQUEUED: i32 = 0;
def PKGBUILDING: i32 = 1;
def PKGRUNNING: i32 = 2;
def PKGDONE: i32 = 3;
// pkggroup.fail values recorded at launch, reported at ordered emission.
def PKGFAILNONE: i32 = 0;
def PKGFAILSETUP: i32 = 1;
def PKGFAILCOMPOSE: i32 = 2;
def pkgpoll: time.duration = 1000000i64: time.duration;
type pkgdiscover = struct {
paths: []str,
errors: i32,
@@ -90,7 +107,7 @@ fn pkgusage() void = {
let s: str = strings.concat(
"usage: wwtest package [-c] [-list] [-j N] [-I DIR] [-run|-filter GLOB] [-timeout-ms=N] [DIR | DIR/...] [-- GLOB ...]\n",
" *_test.ww is canonical; noncanonical files require an actual @test declaration\n",
" -c retains the compiled package binaries; -j is reserved by sequential v1\n");
" -c retains the compiled package binaries; -j N runs up to N package groups at once\n");
pkgput(os.STDERR_FILENO, s);
};
@@ -381,18 +398,19 @@ fn pkgdedup(ss: []str) []str = {
return out;
};
fn pkgparsedec(s: str, max: i64) bool = {
if (s.len == 0) { return false; };
// Decimal value, or -1 on empty, non-digit, or overflow past max.
fn pkgparsedec(s: str, max: i64) i64 = {
if (s.len == 0) { return -1i64; };
let i: i32 = 0;
let n: i64 = 0i64;
for (i < s.len) {
if (s[i] < '0' || s[i] > '9') { return false; };
if (s[i] < '0' || s[i] > '9') { return -1i64; };
let digit: i64 = (s[i] - '0'): i64;
if (n > (max - digit) / 10i64) { return false; };
if (n > (max - digit) / 10i64) { return -1i64; };
n = n * 10i64 + digit;
i += 1;
};
return n > 0i64;
return n;
};
fn pkgdefaultbuilder() str = {
@@ -537,13 +555,8 @@ fn pkgreportcommand(kind: str, g: *pkggroup, r: *exec.result) void = {
pkgputln(os.STDERR_FILENO, ")");
};
fn pkgrungroup(g: *pkggroup, srcs: []pkgsource, builder: str,
filters: []str, includes: []str, timeoutarg: str,
list: bool, compileonly: bool) bool = {
if (!pkgcombined(g, srcs)) {
pkgfailpath(g.combined, "cannot compose package test source");
return false;
};
fn pkgstartbuild(g: *pkggroup, builder: str, includes: []str,
h: *exec.process) void = {
let ba: []str = alloc([], (9 + includes.len * 2): u64)!;
append(ba, builder);
append(ba, "test");
@@ -559,37 +572,21 @@ fn pkgrungroup(g: *pkggroup, srcs: []pkgsource, builder: str,
ii += 1;
};
append(ba, g.combined);
let env: []str = toolenv(g.root);
let bcmd: exec.command;
bcmd.path = builder;
bcmd.argv = ba;
bcmd.env = env;
bcmd.env = toolenv(g.root);
bcmd.dir = "";
bcmd.stdoutpath = g.buildout;
bcmd.stderrpath = g.builderr;
bcmd.deadline.sec = 0i64;
bcmd.deadline.nsec = 0i64;
bcmd.grace = 0i64: time.duration;
let br: exec.result;
exec.run(&bcmd, &br);
let buildcaptures: bool = pkgemitfile(g.buildout, os.STDOUT_FILENO);
buildcaptures = pkgemitfile(g.builderr, os.STDERR_FILENO) && buildcaptures;
if (!buildcaptures) {
pkgfailpath(g.root, "cannot read build capture");
return false;
};
if (br.errno != 0 || br.cleanuperrno != 0
|| br.termination != exec.termination.EXIT || br.code != 0) {
pkgreportcommand("build", g, &br);
return false;
};
if (compileonly) {
pkgput(os.STDOUT_FILENO, "built ");
pkglabel(g);
pkgput(os.STDOUT_FILENO, " -> ");
pkgputln(os.STDOUT_FILENO, g.bin);
return true;
};
exec.start(h, &bcmd);
};
fn pkgstartrun(g: *pkggroup, filters: []str, timeoutarg: str,
list: bool, h: *exec.process) void = {
let ra: []str = alloc([], (filters.len + 4): u64)!;
append(ra, g.bin);
append(ra, strings.concat("-package=", g.pkg));
@@ -600,24 +597,61 @@ fn pkgrungroup(g: *pkggroup, srcs: []pkgsource, builder: str,
let rcmd: exec.command;
rcmd.path = g.bin;
rcmd.argv = ra;
rcmd.env = env;
rcmd.env = toolenv(g.root);
rcmd.dir = "";
rcmd.stdoutpath = g.runout;
rcmd.stderrpath = g.runerr;
rcmd.deadline.sec = 0i64;
rcmd.deadline.nsec = 0i64;
rcmd.grace = 0i64: time.duration;
let rr: exec.result;
exec.run(&rcmd, &rr);
exec.start(h, &rcmd);
};
fn pkgbuildok(g: *pkggroup) bool = {
return g.buildres.errno == 0 && g.buildres.cleanuperrno == 0
&& g.buildres.termination == exec.termination.EXIT
&& g.buildres.code == 0;
};
// Ordered emission: a group's captures, diagnostics, and verdict line are
// written only here, strictly in group order, so concurrent scheduling
// produces the byte stream sequential scheduling produced.
fn pkgemitgroup(g: *pkggroup, tmproot: str, compileonly: bool) bool = {
if (g.fail == PKGFAILSETUP) {
pkgfailpath(tmproot, "cannot create group temporary directory");
return false;
};
if (g.fail == PKGFAILCOMPOSE) {
pkgfailpath(g.combined, "cannot compose package test source");
return false;
};
let buildcaptures: bool = pkgemitfile(g.buildout, os.STDOUT_FILENO);
buildcaptures = pkgemitfile(g.builderr, os.STDERR_FILENO) && buildcaptures;
if (!buildcaptures) {
pkgfailpath(g.root, "cannot read build capture");
return false;
};
if (!pkgbuildok(g)) {
pkgreportcommand("build", g, &g.buildres);
return false;
};
if (compileonly) {
pkgput(os.STDOUT_FILENO, "built ");
pkglabel(g);
pkgput(os.STDOUT_FILENO, " -> ");
pkgputln(os.STDOUT_FILENO, g.bin);
return true;
};
let runcaptures: bool = pkgemitfile(g.runout, os.STDOUT_FILENO);
runcaptures = pkgemitfile(g.runerr, os.STDERR_FILENO) && runcaptures;
if (!runcaptures) {
pkgfailpath(g.root, "cannot read test capture");
return false;
};
if (rr.errno != 0 || rr.cleanuperrno != 0
|| rr.termination != exec.termination.EXIT || rr.code != 0) {
pkgreportcommand("test", g, &rr);
if (g.runres.errno != 0 || g.runres.cleanuperrno != 0
|| g.runres.termination != exec.termination.EXIT
|| g.runres.code != 0) {
pkgreportcommand("test", g, &g.runres);
return false;
};
pkgput(os.STDOUT_FILENO, "ok ");
@@ -629,7 +663,7 @@ fn pkgrungroup(g: *pkggroup, srcs: []pkgsource, builder: str,
export fn packagecommand(args: []str) int = {
let compileonly: bool = false;
let list: bool = false;
let sawj: bool = false;
let jobs: i32 = 1;
let afterdash: bool = false;
let roots: []str = alloc([], 2u64)!;
let filters: []str = alloc([], (args.len + 1): u64)!;
@@ -656,7 +690,7 @@ export fn packagecommand(args: []str) int = {
};
if (strings.hasprefix(a, "-timeout-ms=")) {
if (timeoutarg.len != 0 || a.len == 12
|| !pkgparsedec(a[12:a.len], 3600000i64)) {
|| pkgparsedec(a[12:a.len], 3600000i64) <= 0i64) {
pkgusage();
return 2;
};
@@ -665,12 +699,10 @@ export fn packagecommand(args: []str) int = {
continue;
};
if (strings.compare(a, "-j") == 0) {
if (i + 1 >= args.len
|| !pkgparsedec(args[i + 1], 2147483647i64)) {
pkgusage();
return 2;
};
sawj = true;
if (i + 1 >= args.len) { pkgusage(); return 2; };
let v: i64 = pkgparsedec(args[i + 1], 2147483647i64);
if (v <= 0i64) { pkgusage(); return 2; };
jobs = v: i32;
i += 2;
continue;
};
@@ -703,10 +735,6 @@ export fn packagecommand(args: []str) int = {
pkgusage();
return 2;
};
if (sawj) {
pkgputln(os.STDERR_FILENO,
"wwtest package: -j is reserved; sequential package scheduling is active");
};
// Go's ./... form: a trailing "..." path element walks the tree
// rooted at the prefix instead of one explicit directory.
@@ -846,19 +874,78 @@ export fn packagecommand(args: []str) int = {
let borrowed: str = temp.dir();
let tmproot: str = strings.dup(borrowed);
let failed: i32 = 0;
// Bounded scheduler: up to `jobs` groups in flight, each a
// build-then-run process chain supervised with exec.start/poll.
// Launch order, run-slot accounting, and ordered emission keep
// -j 1 byte-identical to the former sequential loop; a setup
// failure stops new launches exactly where that loop broke.
let handles: []exec.process = alloc([], groups.len: u64)!;
i = 0;
for (i < groups.len) {
if (!pkgsetpaths(&groups[i], tmproot, i, compileonly)) {
pkgfailpath(tmproot, "cannot create group temporary directory");
failed += 1;
break;
};
if (!pkgrungroup(&groups[i], srcs, builder, filters, includes,
timeoutarg, list, compileonly)) {
failed += 1;
};
let h: exec.process;
append(handles, h);
i += 1;
};
let launched: i32 = 0;
let emitted: i32 = 0;
let active: i32 = 0;
let stopped: bool = false;
for (emitted < groups.len) {
for (!stopped && launched < groups.len && active < jobs) {
let g: *pkggroup = &groups[launched];
if (!pkgsetpaths(g, tmproot, launched, compileonly)) {
g.fail = PKGFAILSETUP;
g.state = PKGDONE;
stopped = true;
} else if (!pkgcombined(g, srcs)) {
g.fail = PKGFAILCOMPOSE;
g.state = PKGDONE;
} else {
pkgstartbuild(g, builder, includes,
&handles[launched]);
g.state = PKGBUILDING;
active += 1;
};
launched += 1;
};
let k: i32 = emitted;
for (k < launched) {
let g: *pkggroup = &groups[k];
// Pointer alias: an aggregate copy from
// `handles[k].result` (N_DOT over N_INDEX source) is a
// loud cgen gap in both stages; the deref-base spine is
// wired. Same idiom as wwfixture's collect().
let hp: *exec.process = &handles[k];
if (g.state == PKGBUILDING && exec.poll(hp)) {
g.buildres = hp.result;
if (pkgbuildok(g) && !compileonly) {
pkgstartrun(g, filters, timeoutarg,
list, hp);
g.state = PKGRUNNING;
} else {
g.state = PKGDONE;
active -= 1;
};
} else if (g.state == PKGRUNNING && exec.poll(hp)) {
g.runres = hp.result;
g.state = PKGDONE;
active -= 1;
};
k += 1;
};
for (emitted < launched && groups[emitted].state == PKGDONE) {
if (!pkgemitgroup(&groups[emitted], tmproot,
compileonly)) {
failed += 1;
};
emitted += 1;
};
if (stopped && active == 0 && emitted == launched) { break; };
if (active > 0) {
time.sleep(pkgpoll, time.clock.monotonic);
};
};
if (!pkgremoveall(tmproot)) {
pkgput(os.STDERR_FILENO, "wwtest package: cleanup failed; retained ");
pkgputln(os.STDERR_FILENO, tmproot);