process.cpp

#include "sys.h"
#include "handle.h"
#include <openkal/process.h>
#include <openkal/memory.h>
#include <openkal/task.h>

namespace okl { extern char** g_envp; extern bool g_pipe_started_ignored; okl_ulong auxval(okl_ulong key); }
// The transfer to a function on a stack of its own (task.cpp).
namespace okl { okl_long clone_on_stack(int (*fn)(void*), void* stack, int flags, void* arg); }
// Defined with what a unit uses, below (0.21).
namespace okl {
int child_outcome(okl_long pid, bool block, int* status, int* terminated, bool& ended);
void job_forget(okl_long unit);
bool child_memory(okl_long pid, bool ended, okl_u64& now, okl_u64& peak);
bool child_cpu(okl_long pid, okl_u64& ns);
}

namespace {

bool same_prefix(const char* a, const char* b, okl_uptr n) {
    for (okl_uptr i = 0; i < n; ++i) if (a[i] != b[i]) return false;
    return true;
}


constexpr okl_uptr kMaxEntries = 512;

constexpr char     kPreopens[]  = "KAL_PREOPENS=";
constexpr okl_uptr kPreopensLen = sizeof kPreopens - 1;
constexpr okl_uptr kPidDigits   = 10;

bool names_preopens(const char* s, kal_uintptr n) {
    if (n < kPreopensLen) return false;
    for (okl_uptr i = 0; i < kPreopensLen; ++i)
        if (s[i] != kPreopens[i]) return false;
    return true;
}

okl_uptr digits(okl_uptr v) {
    okl_uptr n = 1;
    while (v >= 10) { v /= 10; ++n; }
    return n;
}

char* put_decimal(char* o, okl_uptr v) {
    char b[24]; int i = 0;
    do { b[i++] = static_cast<char>('0' + v % 10); v /= 10; } while (v);
    while (i) *o++ = b[--i];
    return o;
}

// Every descriptor from `from' upward is marked to close when the image is
// replaced. `close_range' does it in one call from Linux 5.11; before that
// kernel the descriptors the process has are listed from /proc/self/fd, and
// where that is not mounted every number below the descriptor limit is tried.
void close_on_exec_from(okl_long from) {
    const okl_long r = okl::sys(okl::nr_close_range, from, static_cast<okl_long>(~0u),
                                okl::close_range_cloexec);
    if (!okl::failed(r)) return;

    const okl_long dir = okl::sys(okl::nr_openat, okl::at_fdcwd,
                                  reinterpret_cast<okl_long>("/proc/self/fd"),
                                  okl::o_rdonly | okl::o_directory | okl::o_cloexec, 0);
    if (!okl::failed(dir)) {
        alignas(8) char buf[2048];
        for (;;) {
            const okl_long n = okl::sys(okl::nr_getdents64, dir,
                                        reinterpret_cast<okl_long>(buf), sizeof buf);
            if (n <= 0) break;
            for (okl_long at = 0; at < n; ) {
                const auto* d = reinterpret_cast<const okl::kdirent64*>(buf + at);
                okl_long fd = 0; bool number = d->name[0] != '\0';
                for (const char* c = d->name; *c; ++c) {
                    if (*c < '0' || *c > '9') { number = false; break; }
                    fd = fd * 10 + (*c - '0');
                }
                if (number && fd >= from && fd != dir)
                    okl::sys(okl::nr_fcntl, fd, okl::f_setfd, okl::fd_cloexec);
                at += d->reclen;
            }
        }
        okl::sys(okl::nr_close, dir);
        return;
    }

    okl_ulong lim[2] = { 1024, 1024 };
    okl::sys(okl::nr_prlimit64, 0, okl::rlimit_nofile, 0, reinterpret_cast<okl_long>(lim));
    for (okl_long fd = from; fd < static_cast<okl_long>(lim[0]); ++fd)
        okl::sys(okl::nr_fcntl, fd, okl::f_setfd, okl::fd_cloexec);
}

// The counted arrays the interface takes become the terminated arrays the
// kernel takes. Every allocation happens before the program is duplicated, so
// that the duplicate performs nothing but two system calls: a duplicate of a
// program that has more than one execution context may hold a lock no context
// in it will release.
struct vector {
    char** slots = nullptr;
    char*  bytes = nullptr;
    okl_uptr slots_bytes = 0;
    okl_uptr bytes_bytes = 0;
    bool   ok = true;

    bool build(const char** items, const kal_uintptr* lens, kal_uintptr n) {
        if (n > kMaxEntries) { ok = false; return false; }
        okl_uptr total = 0;
        for (kal_uintptr i = 0; i < n; ++i) total += lens[i] + 1;
        slots_bytes = (n + 1) * sizeof(char*);
        bytes_bytes = total == 0 ? 1 : total;
        slots = static_cast<char**>(kal_alloc(slots_bytes, alignof(char*)));
        bytes = static_cast<char*>(kal_alloc(bytes_bytes, 1));
        if (!slots || !bytes) { ok = false; return false; }
        okl_uptr at = 0;
        for (kal_uintptr i = 0; i < n; ++i) {
            okl::copy(bytes + at, items[i], lens[i]);
            bytes[at + lens[i]] = '\0';
            slots[i] = bytes + at;
            at += lens[i] + 1;
        }
        slots[n] = nullptr;
        return true;
    }

    // THE ENVIRONMENT, WHICH CARRIES THE NAMES OF THE GRANTED DIRECTORIES.
    //
    // A granted directory reaches the started program as a descriptor, and a
    // descriptor carries no name. The names travel in one variable,
    //
    //     KAL_PREOPENS=<pid>{;<fd>,<len>,<name>}
    //
    // which the started program reads when it first enumerates its preopens
    // (`table' in fs.cpp). It is the arrangement this environment already uses
    // to hand a started program named descriptors --- systemd's LISTEN_FDS,
    // LISTEN_FDNAMES and LISTEN_PID --- and it is bound to the process it was
    // written for in the same way: `<pid>' is ten digits written by the
    // duplicate once it knows its own number, so a value inherited through a
    // program that does not read it, a shell for example, names no one when it
    // arrives one generation further down.
    //
    // Every variable of that name the caller supplied is left out, so a value
    // is never forwarded; one is added only when the caller asked for grants,
    // and asking for none (a count of zero) is a value with no entries.
    char*    pid_digits = nullptr;

    bool build_env(const char** items, const kal_uintptr* lens, kal_uintptr n,
                   const kal_preopen* grants, kal_uintptr g) {
        if (n > kMaxEntries) { ok = false; return false; }
        kal_uintptr kept = 0;
        okl_uptr total = 0;
        for (kal_uintptr i = 0; i < n; ++i) {
            if (names_preopens(items[i], lens[i])) continue;
            ++kept; total += lens[i] + 1;
        }
        okl_uptr extra = 0;
        if (grants) {
            extra = kPreopensLen + kPidDigits + 1;
            for (kal_uintptr i = 0; i < g; ++i)
                extra += 3 + digits(3 + i) + digits(grants[i].len) + grants[i].len;
        }
        slots_bytes = (kept + (grants ? 1 : 0) + 1) * sizeof(char*);
        bytes_bytes = total + extra == 0 ? 1 : total + extra;
        slots = static_cast<char**>(kal_alloc(slots_bytes, alignof(char*)));
        bytes = static_cast<char*>(kal_alloc(bytes_bytes, 1));
        if (!slots || !bytes) { ok = false; return false; }
        okl_uptr at = 0, k = 0;
        for (kal_uintptr i = 0; i < n; ++i) {
            if (names_preopens(items[i], lens[i])) continue;
            okl::copy(bytes + at, items[i], lens[i]);
            bytes[at + lens[i]] = '\0';
            slots[k++] = bytes + at;
            at += lens[i] + 1;
        }
        if (grants) {
            char* o = bytes + at;
            slots[k++] = o;
            okl::copy(o, kPreopens, kPreopensLen); o += kPreopensLen;
            pid_digits = o;
            okl::fill(o, '0', kPidDigits); o += kPidDigits;
            for (kal_uintptr i = 0; i < g; ++i) {
                *o++ = ';'; o = put_decimal(o, 3 + i);
                *o++ = ','; o = put_decimal(o, grants[i].len);
                *o++ = ',';
                okl::copy(o, grants[i].name, grants[i].len); o += grants[i].len;
            }
            *o = '\0';
        }
        slots[k] = nullptr;
        return true;
    }

    // In the duplicate, which is the first point at which the number is known.
    void stamp(okl_long pid) const {
        if (!pid_digits) return;
        for (int i = static_cast<int>(kPidDigits) - 1; i >= 0; --i) {
            pid_digits[i] = static_cast<char>('0' + pid % 10);
            pid /= 10;
        }
    }

    ~vector() {
        if (slots) kal_free(slots, slots_bytes, alignof(char*));
        if (bytes) kal_free(bytes, bytes_bytes, 1);
    }
};

// --- reporting a replacement that failed -----------------------------------
//
// THE REPLACEMENT HAPPENS IN THE DUPLICATE, SO ITS FAILURE WAS REPORTED TO
// NOBODY.
//
// A program is started here by duplicating this image and replacing the
// duplicate. The replacement is the part that can fail --- the name is absent,
// or is a directory, or is not a program, or may not be executed --- and it
// fails inside an image the caller does not have. This implementation ended
// that image with 127 and answered `kal_ok' with a handle, so a caller learned
// something was wrong only by waiting and reading 127, which is exactly what a
// program that RAN and exited 127 reports.
//
// WHAT THAT COST, MEASURED BY A CONSUMER RATHER THAN HERE. openkal-musl
// expresses `execve' as starting a program and ending with its status, so a
// name that could not be started ended the CALLING program with 127 instead of
// returning -1. musl's `execvp' issues one `execve' per PATH entry and needs
// each to return, so the search could not survive its first miss: `bwrap',
// installed at /usr/bin/bwrap, was reported as not installed. openkal-linux#13,
// nine of nineteen test failures.
//
// openkal-musl 0.10.0 answers the part it can --- it asks `kal_fs_info' whether
// the name is there before starting. It cannot answer the rest: openkal reports
// no execute permission, so "present and not executable" is invisible above
// this line. It is not invisible HERE. The duplicate knows precisely why, and
// this is the channel that carries it.
//
// The arrangement is the ordinary one: a pipe whose ends close when the image is
// replaced. Nothing arrives ⇒ the replacement happened. A value arrives ⇒ it did
// not, and the value says why.
struct exec_report {
    int  fd[2] = { -1, -1 };
    bool armed = false;

    // THE PIPE MUST NOT SIT WHERE THE DUPLICATE IS ABOUT TO PLACE SOMETHING.
    // The duplicate places streams at 0, 1 and 2 and granted directories at 3
    // and upwards, so a pipe that happened to hold one of those numbers would be
    // closed by the very placement whose failure it exists to report --- and the
    // parent would then read end-of-input and call that success.
    //
    // `F_DUPFD_CLOEXEC' answers the lowest FREE descriptor at or above a bound.
    // That is the primitive for this, and `dup3' is not: `dup3' is told the
    // number and closes whatever the caller had on it.
    bool open(kal_uintptr placements) {
        const okl_long r = okl::sys(okl::nr_pipe2,
                                    reinterpret_cast<okl_long>(fd), okl::o_cloexec);
        if (okl::failed(r)) return false;
        const okl_long floor = 3 + static_cast<okl_long>(placements);
        armed = lift(fd[0], floor) && lift(fd[1], floor);
        if (!armed) close_both();
        return armed;
    }

    void close_both() {
        if (fd[0] >= 0) okl::sys(okl::nr_close, fd[0]);
        if (fd[1] >= 0) okl::sys(okl::nr_close, fd[1]);
        fd[0] = fd[1] = -1;
    }

    // In the duplicate, once the replacement has returned --- which it does only
    // when it did not happen.
    void say(okl_long failure) const {
        if (!armed) return;
        okl_long value = failure;
        okl::sys(okl::nr_write, fd[1],
                 reinterpret_cast<okl_long>(&value), sizeof value);
    }

    // In this image. Zero when the replacement happened, otherwise the kernel's
    // own negative value for why it did not.
    okl_long heard() {
        if (!armed) return 0;
        okl::sys(okl::nr_close, fd[1]);
        fd[1] = -1;
        okl_long value = 0;
        okl_long n;
        // A transfer this short is not divided, but it can be interrupted.
        do {
            n = okl::sys(okl::nr_read, fd[0],
                         reinterpret_cast<okl_long>(&value), sizeof value);
        } while (n == -okl::e_intr);
        okl::sys(okl::nr_close, fd[0]);
        fd[0] = -1;
        return (n == static_cast<okl_long>(sizeof value)) ? value : 0;
    }

private:
    static bool lift(int& f, okl_long floor) {
        const okl_long n = okl::sys(okl::nr_fcntl, f, okl::f_dupfd_cloexec, floor);
        if (okl::failed(n)) return false;
        okl::sys(okl::nr_close, f);
        f = static_cast<int>(n);
        return true;
    }
};

// A duplicate that could not be replaced is ended, and this image waits for it
// so that nothing is left for a caller to meet later. It is the one wait this
// implementation performs that a caller did not ask for, and it is bounded: the
// duplicate has already reached `exit_group'.
inline void reap(okl_long child) {
    int status = 0;
    okl_long r;
    do {
        r = okl::sys(okl::nr_wait4, child,
                     reinterpret_cast<okl_long>(&status), 0, 0);
    } while (r == -okl::e_intr);
}

}  // namespace

extern "C" {

// Starting a program. ONE FUNCTION SINCE 0.11, AND THE SAVING IS NOT ONLY IN
// THE HEADER: this file used to hold THREE bodies of sixty lines that differed
// by four. Every fix to the shared part --- and there have been several, the
// exec-report pipe among them --- had to be made three times or be made once and
// be wrong twice.
//
// The modifiers are now positions in `how': a working directory, a set of
// grants, and two flags. They compose, which the three declarations could not
// do: there was no way to grant directories AND bind a lifetime, and no way at
// all to say the two things a shell runner needs together.
// --- the handles of this interface (0.17) ------------------------------------
//
// A started program's handle is its identifier, as it always was. Two more
// kinds exist since 0.17 and are told apart by the top byte, which no
// identifier reaches: this program itself, and a program found by its
// identifier, which carries the descriptor the kernel gave for it (`pidfd')
// in the bits above the identifier. Since 0.18 a third: a program found by
// its identifier that the kernel will not give a descriptor for (another
// namespace's, or a kernel before 5.3), observed through its /proc entry, with
// the low bits of its start in the bits above the identifier --- the start the
// open recorded, so that the identifier given to another program reads as the
// first one ended (openkal `kal_process_open_id').
namespace {
constexpr unsigned kind_child = 0, kind_self = 1, kind_opened = 2, kind_observed = 3;
inline unsigned handle_kind(kal_process h) { return static_cast<unsigned>(h.h >> 56); }
inline okl_long handle_pid(kal_process h) { return static_cast<okl_long>(h.h & 0xffffffffu); }
}  // namespace

// --- where `execveat' cannot be used (0.18) ----------------------------------
//
// termux's PRoot implements `execveat' for AT_FDCWD only and answers ENOSYS to a
// descriptor; in its ptrace mode it lets the kernel run the call untranslated,
// and the started program dies at its first allocation. Every start in that
// environment failed. The answer is `execve' with an absolute name, joined HERE,
// before the duplication, from the name the kernel gives the base directory ---
// the duplicate still allocates nothing. It costs the property `execveat'
// bought, that the name is resolved through the descriptor, and is paid only
// where `execveat' is unusable: from the first start under a translator this
// implementation recognises, and, elsewhere, by each start that meets ENOSYS,
// which falls back in the started image itself. That fallback is learned by
// no other start: the image writes nothing of this program's (start_plan), so
// nothing it learns reaches this program, and the price is one refused system
// call per start.
//
// Ported from the copy mcppls carried (vendor/openkal-linux 0.15.1), which set
// the flag itself under PRoot; recognising the translator here removes that.
namespace {

int g_by_name = -1;   // -1 not yet decided, 0 execveat, 1 execve

// Whether this program is traced by PRoot: the tracer's name, and where /proc
// hides it, a PROOT_ variable. A debugger also traces and is not a translator.
bool under_translator() {
    char st[4096];
    const okl_long fd = okl::sys(okl::nr_openat, okl::at_fdcwd,
                                 reinterpret_cast<okl_long>("/proc/self/status"),
                                 okl::o_rdonly | okl::o_cloexec, 0);
    okl_long n = 0;
    if (!okl::failed(fd)) {
        n = okl::sys(okl::nr_read, fd, reinterpret_cast<okl_long>(st), sizeof st - 1);
        okl::sys(okl::nr_close, fd);
    }
    if (n > 0) {
        st[n] = '\0';
        const char* t = nullptr;
        for (okl_long i = 0; i + 10 < n; ++i)
            if (st[i] == 'T' && same_prefix(st + i, "TracerPid:", 10)) { t = st + i + 10; break; }
        okl_long pid = 0;
        while (t && (*t == ' ' || *t == '\t')) ++t;
        while (t && *t >= '0' && *t <= '9') pid = pid * 10 + (*t++ - '0');
        if (pid == 0) return false;
        char path[48] = "/proc/";
        char* o = put_decimal(path + 6, static_cast<okl_uptr>(pid));
        okl::copy(o, "/comm", 6);
        const okl_long cfd = okl::sys(okl::nr_openat, okl::at_fdcwd, reinterpret_cast<okl_long>(path),
                                      okl::o_rdonly | okl::o_cloexec, 0);
        if (!okl::failed(cfd)) {
            char comm[32] = {};
            const okl_long m = okl::sys(okl::nr_read, cfd, reinterpret_cast<okl_long>(comm), sizeof comm - 1);
            okl::sys(okl::nr_close, cfd);
            return m >= 5 && same_prefix(comm, "proot", 5);
        }
    }
    for (char** e = okl::g_envp; e && *e; ++e)
        if (same_prefix(*e, "PROOT_", 6)) return true;
    return false;
}

bool by_name() {
    int v = __atomic_load_n(&g_by_name, __ATOMIC_ACQUIRE);
    if (v < 0) {
        v = under_translator() ? 1 : 0;
        __atomic_store_n(&g_by_name, v, __ATOMIC_RELEASE);
    }
    return v == 1;
}

// `/proc/self/fd/<fd>' read, and `name' joined beneath it.
bool absolute_under(int fd, const char* name, okl_uptr len, char* out, okl_uptr cap) {
    char link[32] = "/proc/self/fd/";
    char* e = put_decimal(link + 14, static_cast<okl_uptr>(fd));
    *e = '\0';
    const okl_long r = okl::sys(okl::nr_readlinkat, okl::at_fdcwd, reinterpret_cast<okl_long>(link),
                                reinterpret_cast<okl_long>(out), static_cast<okl_long>(cap));
    if (okl::failed(r) || r <= 0) return false;
    okl_uptr at = static_cast<okl_uptr>(r);
    if (at + 1 + len + 1 > cap) return false;
    if (out[at - 1] != '/') out[at++] = '/';
    okl::copy(out + at, name, len);
    out[at + len] = '\0';
    return true;
}

// Whether `name' reaches the file `fd' holds: the same device and inode.
bool same_file(okl_long fd, const char* name) {
    okl::kstat held{}, named{};
    if (okl::failed(okl::sys(okl::nr_fstat, fd, reinterpret_cast<okl_long>(&held)))) return false;
    if (okl::failed(okl::sys(okl::nr_newfstatat, okl::at_fdcwd, reinterpret_cast<okl_long>(name),
                             reinterpret_cast<okl_long>(&named), 0)))
        return false;
    return held.dev == named.dev && held.ino == named.ino;
}

}  // namespace

// EVERY ENVIRONMENT ENTRY IS `NAME=value' (0.17, process.h). Refused before
// anything is built, so that one implementation does not pass on an entry
// another refuses.
static bool entries_well_formed(const char** envp, const kal_uintptr* lens, kal_uintptr n) {
    for (kal_uintptr i = 0; i < n; ++i) {
        if (envp[i] == nullptr) return false;
        kal_uintptr eq = 0;
        while (eq < lens[i] && envp[i][eq] != '=') ++eq;
        if (eq == 0 || eq == lens[i]) return false;
    }
    return true;
}

namespace {

// WHAT THE STARTED IMAGE IS GIVEN, prepared before it exists. It runs in this
// program's memory until it is replaced, so it reads this and writes nothing of
// this program's but the digits of its own number in its own environment,
// which this program releases unread; the descriptors it moves are its own
// copies.
struct start_plan {
    const kal_spawn*       how;
    const okl::terminated* name;
    const okl_long*        granted;
    int                    b, w;
    okl_long               in, ou, er;
    const vector*          args;
    const vector*          envs;
    const exec_report*     report;
    const char*            resolved;
    bool                   named, use_name, unit, bind;
    okl_long               join, mine;
    const unsigned long*   mask;
};

constexpr okl_uptr kStartStack = 64 * 1024;

// THE STARTED IMAGE, until it is replaced: on its own stack, in this program's
// memory, with every request held (kal_process_spawn).
int start_image(void* arg) {
    const start_plan& plan = *static_cast<const start_plan*>(arg);
    const kal_spawn* const how = plan.how;
    const okl::terminated& p = *plan.name;
    const exec_report& report = *plan.report;
    const vector& args = *plan.args;
    const vector& envs = *plan.envs;
    const int b = plan.b, w = plan.w;
    const okl_long in = plan.in, ou = plan.ou, er = plan.er, join = plan.join, mine = plan.mine;
    const bool named = plan.named, use_name = plan.use_name, unit = plan.unit, bind = plan.bind;
    const char* const resolved = plan.resolved;
    okl_long granted[16];
    for (kal_uintptr i = 0; i < how->grant_count; ++i) granted[i] = plan.granted[i];

    // A HANDLER OF THE CALLER'S IS NOT ENTERED HERE: each request it handles
    // is given its default until the replacement, which gives every handled
    // request its default in any case. One it ignores stays ignored, as the
    // replacement keeps it.
    for (int signo = 1; signo <= 64; ++signo) {
        if (signo == 9 || signo == 19) continue;   // SIGKILL, SIGSTOP
        struct { void* handler; unsigned long flags; void* restorer; unsigned long mask; } act{};
        if (okl::failed(okl::sys(okl::nr_rt_sigaction, signo, 0, reinterpret_cast<okl_long>(&act),
                                 sizeof act.mask)))
            continue;
        if (act.handler == nullptr || act.handler == reinterpret_cast<void*>(1)) continue;
        decltype(act) standard{};
        okl::sys(okl::nr_rt_sigaction, signo, reinterpret_cast<okl_long>(&standard), 0, sizeof standard.mask);
    }

    // WHAT THE STARTED PROGRAM RECEIVES IS THE THREE STREAMS AND THE GRANTED
    // DIRECTORIES, AND NOTHING ELSE (SPEC.md clause 7.13).
    //
    // Every source is first moved above the positions being filled, because
    // placing one source on its position must not overwrite another that
    // still has to be read: a granted directory, a stream, the base or the
    // working directory may each occupy a number between 0 and 3+n. Placing
    // them one at a time where they stood is how a grant named `work'
    // arrived as a second copy of `/', and how a placement overwrote the
    // base the program's name is resolved under. A source that is moved has
    // its own descriptor flag; `dup3' then clears it on the copy it places,
    // including where the source happened to be the position itself ---
    // which `dup3' refuses, and which used to leave the copy marked to
    // close, so that the grant never arrived.
    const okl_long top = static_cast<okl_long>(3 + how->grant_count);
    auto lift = [&](okl_long fd) -> okl_long {
        const okl_long r = okl::sys(okl::nr_fcntl, fd, okl::f_dupfd_cloexec, top);
        if (okl::failed(r)) {
            report.say(r);
            okl::sys(okl::nr_exit_group, 127);
            for (;;) { }
        }
        return r;
    };
    // THE NULL DEVICE (openkal 0.22), opened here, in the started image, so
    // that the caller holds no descriptor for it; it is placed as any other
    // source is. Its descriptor is marked to close, so the replacement keeps
    // only the positions it was placed at.
    const kal_uintptr nulls = how->flags & (KAL_SPAWN_NULL_IN | KAL_SPAWN_NULL_OUT | KAL_SPAWN_NULL_ERR);
    okl_long device = 0;
    if (nulls != 0) {
        const okl_long fd = okl::sys(okl::nr_openat, okl::at_fdcwd, reinterpret_cast<okl_long>("/dev/null"),
                                     okl::o_rdwr | okl::o_cloexec, 0);
        if (okl::failed(fd)) {
            report.say(fd);
            okl::sys(okl::nr_exit_group, 127);
            for (;;) { }
        }
        device = lift(fd);
    }
    const okl_long sin = (how->flags & KAL_SPAWN_NULL_IN)  ? device : in != 0 ? lift(in) : 0;
    const okl_long sou = (how->flags & KAL_SPAWN_NULL_OUT) ? device : ou != 0 ? lift(ou) : 0;
    const okl_long ser = (how->flags & KAL_SPAWN_NULL_ERR) ? device : er != 0 ? lift(er) : 0;
    for (kal_uintptr i = 0; i < how->grant_count; ++i) granted[i] = lift(granted[i]);
    const okl_long base = lift(b);
    const okl_long work = lift(w);

    if (sin != 0) okl::sys(okl::nr_dup3, sin, 0, 0);
    if (sou != 0) okl::sys(okl::nr_dup3, sou, 1, 0);
    if (ser != 0) okl::sys(okl::nr_dup3, ser, 2, 0);
    for (kal_uintptr i = 0; i < how->grant_count; ++i)
        okl::sys(okl::nr_dup3, granted[i], static_cast<okl_long>(3 + i), 0);

    // What the CALLER itself inherited without the flag is not a handle it
    // granted. A program that starts others inside a sandbox passes every
    // descriptor it holds to them, and one of those reaching beyond the
    // sandbox is exactly what the sandbox was for.
    close_on_exec_from(top);

    envs.stamp(okl::sys(okl::nr_getpid));

    // THE DIRECTORY THE PROGRAM RUNS IN, AND THIS LINE IS THE WHOLE OF IT.
    //
    // `openat' below takes the base as a dirfd, but that only RESOLVES the
    // name --- resolving a name is not entering a directory, which is what
    // the comment here used to get wrong. Until 0.11 there was no second
    // directory to enter, and a started program ran wherever this
    // implementation happened to be.
    //
    // A FAILURE HERE MUST NOT REACH `execveat'. Running the right program
    // in the wrong directory is precisely the silent wrongness this exists to
    // remove, so it is reported through the same pipe an exec failure uses.
    if (const okl_long e = okl::sys(okl::nr_fchdir, work); okl::failed(e)) {
        report.say(e);
        okl::sys(okl::nr_exit_group, 127);
        for (;;) { }
    }

    // THE UNIT, ENTERED HERE AND NOT FROM THE PARENT: the parent's own
    // `setpgid' on this child races the replacement below and loses once the
    // program has been replaced. Zero means "your own", which is how a group
    // comes into existence at all --- there is nothing to create beforehand,
    // which is why the interface reports the identity rather than taking it.
    if (unit) okl::sys(okl::nr_setpgid, 0, join);

    // SIGPIPE AS THIS PROGRAM WAS STARTED WITH, not as this implementation
    // set it for its own writes (env.cpp): an ignore would outlive `execve'.
    if (!okl::g_pipe_started_ignored) {
        struct { void* handler; unsigned long flags; void* restorer; unsigned long mask; }
            standard{ nullptr /* SIG_DFL */, 0, nullptr, 0 };
        okl::sys(okl::nr_rt_sigaction, 13 /* SIGPIPE */,
                 reinterpret_cast<okl_long>(&standard), 0, sizeof standard.mask);
    }

    if (bind) {
        // 9 is SIGKILL: the binding must not be something the started program
        // can decline, because the caller asked for a program that does not
        // outlive it and not for one that is invited not to.
        okl::sys(okl::nr_prctl, okl::pr_set_pdeathsig, 9, 0, 0, 0);
        // The window: if the caller ended between the clone and the line
        // above, the signal is already spent and this image would survive it.
        if (okl::sys(okl::nr_getppid) != mine)
            okl::sys(okl::nr_exit_group, 127);
    }

    // The caller's mask, which the program is started with, once nothing of
    // the caller's can be entered here any more (above).
    okl::sys(okl::nr_rt_sigprocmask, 2 /* SIG_SETMASK */, reinterpret_cast<okl_long>(plan.mask), 0,
             sizeof *plan.mask);

    // WHAT THE STARTED PROGRAM KEEPS IS A DESCRIPTOR FOR THE PROGRAM, NOT
    // FOR THE DIRECTORY IT WAS FOUND IN, AND WITHOUT A DESCRIPTOR AT ALL A
    // WHOLE CLASS OF PROGRAMS COULD NOT BE STARTED.
    //
    // `execveat' with a dirfd and a relative name gives the program's name to
    // the kernel as `/dev/fd/<dirfd>/<name>'. That spelling is invisible to a
    // caller and harmless for an ordinary executable --- the kernel already
    // holds the file open. It stops being harmless the moment the program
    // needs an INTERPRETER: a `#!' script, or a binary of another
    // architecture registered through `binfmt_misc'. The kernel then starts
    // the interpreter and hands it that name to open --- AFTER the
    // replacement, by which time a close-on-exec descriptor is gone. The
    // interpreter is told the script does not exist.
    //
    // Measured in twenty lines of plain C, with everything else identical:
    //
    //     descriptor WITH O_CLOEXEC       execveat -> ENOENT
    //     descriptor WITHOUT O_CLOEXEC    STARTED ok
    //
    // It is not a property of one architecture. It was FOUND on aarch64,
    // where every foreign binary needs the binfmt interpreter and so every
    // start failed --- and it was mistaken there for a limit of the emulator.
    // It reproduces natively on x86_64 with a `#!' script, which is what a
    // consumer meets on any machine.
    //
    // The descriptor that survives must not be the base. A copy of the base
    // is a handle on a whole directory, often `/', and a program that runs
    // other programs in a sandbox hands it on to them, where
    // `/proc/self/fd/<n>/' reaches whatever the sandbox hid. So the program
    // itself is opened here, with `O_PATH' because it is only to be named
    // and not read, and `execveat' is given that descriptor and an empty
    // name. The interpreter's `/dev/fd/<n>' then names the file, and nothing
    // can be walked out of a descriptor for a file.
    //
    // AND ONLY A PROGRAM THAT NEEDS AN INTERPRETER KEEPS EVEN THAT. The
    // descriptor is opened with `O_CLOEXEC', and an ordinary executable,
    // which the kernel holds open itself, starts with nothing left behind.
    // A program that needs an interpreter is refused with ENOENT BEFORE the
    // point of no return --- `binfmt_script' and `binfmt_misc' test whether
    // the name will still be reachable and answer that it will not --- so
    // the image is still here to clear the flag and ask again. The other
    // sources of ENOENT on an open descriptor, an interpreter that does not
    // exist, answer the same the second time.
    //
    // The descriptor is moved above the placed positions as well: where
    // the caller had no standard input and supplied none, the lowest free
    // number is 0, and a script would read its own descriptor as input.
    //
    // A SCRIPT IS GIVEN ITS NAME WHERE THE NAME REACHES THE SAME FILE
    // (0.18.1). Through the descriptor a script sees `$0' as `/dev/fd/<n>',
    // and one that locates its neighbours with `dirname "$0"' --- a wrapper
    // installed beside the files it runs, a test's stub provider --- finds
    // none of them: measured in mcpp, whose provider stub wrote beside
    // itself into `/dev/fd/ticks'. So a program that needs an interpreter is
    // started by its absolute name, as every shell and `posix_spawn' start
    // it, when that name is known and names the file the descriptor holds
    // (the same device and inode, so the name grants nothing the descriptor
    // did not). Only where it does not --- a base the kernel cannot name, a
    // name rebound since it was opened --- is the descriptor kept, with the
    // price stated above: the descriptor still names the program and can be
    // reopened through `/proc/self/fd/<n>', as `/proc/self/exe' can.
    //
    // Opened HERE, in the started image, and not where the preopens are made:
    // the caller's own descriptors stay close-on-exec, which is what every
    // other operation of this implementation relies upon.
    //
    // A name that cannot be opened is reported exactly as one that cannot be
    // started is, through the same pipe.
    if (use_name) {
        const okl_long why = okl::sys(okl::nr_execve, reinterpret_cast<okl_long>(resolved),
                                      reinterpret_cast<okl_long>(args.slots),
                                      reinterpret_cast<okl_long>(envs.slots));
        report.say(why);
        okl::sys(okl::nr_exit_group, 127);
        for (;;) { }
    }
    okl_long exe = okl::sys(okl::nr_openat, base, reinterpret_cast<okl_long>(p.buf),
                            okl::o_path | okl::o_cloexec, 0);
    if (okl::failed(exe)) {
        report.say(exe);
        okl::sys(okl::nr_exit_group, 127);
        for (;;) { }
    }
    if (exe < top) exe = lift(exe);

    auto start = [&] {
        return okl::sys(okl::nr_execveat, exe, reinterpret_cast<okl_long>(""),
                        reinterpret_cast<okl_long>(args.slots),
                        reinterpret_cast<okl_long>(envs.slots), okl::at_empty_path);
    };
    okl_long why = start();
    if (why == -okl::e_noent && named && same_file(exe, resolved))
        why = okl::sys(okl::nr_execve, reinterpret_cast<okl_long>(resolved),
                       reinterpret_cast<okl_long>(args.slots),
                       reinterpret_cast<okl_long>(envs.slots));
    if (why == -okl::e_noent) {
        okl::sys(okl::nr_fcntl, exe, okl::f_setfd, 0);
        why = start();
    }
    if (why == -okl::e_nosys && named)
        why = okl::sys(okl::nr_execve, reinterpret_cast<okl_long>(resolved),
                       reinterpret_cast<okl_long>(args.slots),
                       reinterpret_cast<okl_long>(envs.slots));
    // Reached only when the replacement did not happen, because when it does
    // there is nothing here to reach.
    report.say(why);
    okl::sys(okl::nr_exit_group, 127);
    for (;;) { }
    
}

}  // namespace

int kal_process_spawn(const kal_spawn* how,
                      const char* path, kal_uintptr path_len,
                      const char** argv, const kal_uintptr* argv_lens, kal_uintptr argc,
                      const char** envp, const kal_uintptr* envp_lens, kal_uintptr envc,
                      const kal_spawn_streams* streams,
                      kal_process* out) {
    if (how == nullptr || out == nullptr) return kal_err_invalid;

    const int b = okl::unpack(how->base.h);
    const int w = okl::unpack(how->work.h);
    if (b < 0 || w < 0) return kal_err_invalid;
    if (!okl::acceptable(path, path_len)) return kal_err_invalid;
    if (how->grant_count > 0 && how->grants == nullptr) return kal_err_invalid;

    // REFUSED BEFORE ANYTHING IS STARTED, not after. A caller that asked for a
    // bound lifetime and received a program without one has been given a program
    // that outlives it --- which is the failure the flag exists to remove --- so an
    // unclaimed position is an error and not a thing to proceed without.
    constexpr kal_uintptr nulls = KAL_SPAWN_NULL_IN | KAL_SPAWN_NULL_OUT | KAL_SPAWN_NULL_ERR;
    if (how->flags & ~(KAL_SPAWN_BOUND_LIFETIME | nulls)) return kal_err_not_supported;
    // THE NULL DEVICE BESIDE A STREAM AT ONE POSITION is two answers to one
    // question (openkal 0.22, clause 7.15).
    if (streams != nullptr
        && (((how->flags & KAL_SPAWN_NULL_IN) && streams->in.h != 0)
            || ((how->flags & KAL_SPAWN_NULL_OUT) && streams->out.h != 0)
            || ((how->flags & KAL_SPAWN_NULL_ERR) && streams->err.h != 0)))
        return kal_err_invalid;

    okl::terminated p(path, path_len);
    if (!p.ok) return kal_err_invalid;

    // Resolved before the duplication, because a failure after it would leave a
    // child to be reaped and a caller with an error it cannot act upon. A name
    // travels in the environment, which cannot carry a zero byte.
    constexpr kal_uintptr max_grants = 16;
    if (how->grant_count > max_grants) return kal_err_invalid;
    okl_long granted[max_grants];
    for (kal_uintptr i = 0; i < how->grant_count; ++i) {
        const kal_preopen& g = how->grants[i];
        const int fd = okl::unpack(g.dir.h);
        if (fd < 0) return kal_err_invalid;
        if (g.len > okl::max_name || (g.len > 0 && g.name == nullptr)) return kal_err_invalid;
        for (kal_uintptr c = 0; c < g.len; ++c)
            if (g.name[c] == '\0') return kal_err_invalid;
        granted[i] = fd;
    }

    if (envc > 0 && (envp == nullptr || envp_lens == nullptr)) return kal_err_invalid;
    if (!entries_well_formed(envp, envp_lens, envc)) return kal_err_invalid;

    // The absolute name, for `execve' where `execveat' cannot be used. Computed
    // always, because the first start that meets ENOSYS learns it in the child.
    char resolved[2 * okl::max_name + 2];
    const bool named = absolute_under(b, p.buf, path_len, resolved, sizeof resolved);
    const bool use_name = by_name() && named;

    vector args, envs;
    if (!args.build(argv, argv_lens, argc)) return kal_err_no_memory;
    if (!envs.build_env(envp, envp_lens, envc, how->grants, how->grant_count))
        return kal_err_no_memory;

    const okl_long in = streams ? static_cast<okl_long>(streams->in.h)  : 0;
    const okl_long ou = streams ? static_cast<okl_long>(streams->out.h) : 0;
    const okl_long er = streams ? static_cast<okl_long>(streams->err.h) : 0;

    const bool bind = (how->flags & KAL_SPAWN_BOUND_LIFETIME) != 0;

    // THE UNIT, WHOSE IDENTITY HERE IS A PROCESS GROUP'S --- which is to say,
    // the identifier of whichever program formed it first. `join' is zero for the
    // first member, and the child then makes the group its own; a later member is
    // given the number to join.
    const okl_long join = how->job ? static_cast<okl_long>(how->job->h) : 0;
    const bool     unit = how->job != nullptr;

    const okl_long mine = bind ? okl::sys(okl::nr_getpid) : 0;

    // The bound is 3 + grant_count, because the placements below reach that far.
    exec_report report;
    report.open(how->grant_count);

    // THE STARTED IMAGE SHARES THIS PROGRAM'S MEMORY UNTIL IT IS REPLACED
    // (CLONE_VM | CLONE_VFORK), on a stack of its own, and this context waits
    // for the replacement, as a C library's `posix_spawn' starts a program. A
    // copy of the address space (clone with SIGCHLD alone, until 0.22) copied the
    // page tables of the whole program and marked every page it had written to
    // be copied, so that each page it wrote afterwards faulted again: measured in
    // a language server that starts a worker for each module it builds, 1.4 ms in
    // the kernel for each start and a third of a million faults over 176 starts,
    // a fifth of its processor time.
    //
    // Every request is held until the image is replaced. A handler of this
    // program run in the image would run in this program's memory; so the image
    // sets every handler to the default first, and only then takes the mask the
    // caller had (start_image).
    unsigned long every = ~0ul, caller_mask = 0;
    okl::sys(okl::nr_rt_sigprocmask, 2 /* SIG_SETMASK */, reinterpret_cast<okl_long>(&every),
             reinterpret_cast<okl_long>(&caller_mask), sizeof every);
    const start_plan plan{ how, &p, granted, b, w, in, ou, er, &args, &envs, &report, resolved,
                           named, use_name, unit, bind, join, mine, &caller_mask };
    const okl_long stack = okl::sys(okl::nr_mmap, 0, static_cast<okl_long>(kStartStack),
                                    okl::prot_read | okl::prot_write,
                                    okl::map_private | okl::map_anonymous, -1, 0);
    const okl_long child = okl::failed(stack)
        ? stack
        : okl::clone_on_stack(&start_image, reinterpret_cast<char*>(stack) + kStartStack,
                              0x100 /* CLONE_VM */ | 0x4000 /* CLONE_VFORK */ | 17 /* SIGCHLD */,
                              const_cast<start_plan*>(&plan));
    okl::sys(okl::nr_rt_sigprocmask, 2 /* SIG_SETMASK */, reinterpret_cast<okl_long>(&caller_mask), 0,
             sizeof caller_mask);
    if (!okl::failed(stack)) okl::sys(okl::nr_munmap, stack, static_cast<okl_long>(kStartStack));
    if (okl::failed(child)) { report.close_both(); return okl::translate(child); }

    if (const okl_long why = report.heard()) {
        reap(child);
        return okl::translate(why);
    }

    // WRITTEN ONLY AFTER THE START HAS SUCCEEDED, and only when the unit was
    // new. The first member's identifier IS the group's, so this is where the
    // caller learns it; a later member joins one the caller already holds and
    // there is nothing to report.
    if (unit && join == 0) how->job->h = static_cast<kal_uintptr>(child);

    *out = kal_process{ static_cast<kal_uintptr>(child) };
    return kal_ok;
}

// A channel: a pair of streams of which one end is meant to cross a spawn.
//
// WHY THIS IS A KERNEL FACILITY AND kal::kit's CHANNEL IS NOT. A started program
// is another address space, so a pointer into this one is not something it can
// be handed. The pair must therefore be made of whatever the environment carries
// across a spawn, which here is a descriptor.
//
// BOTH ENDS ARE OWNED AND BOTH ARE RELEASED THROUGH kal_process_channel_close.
// A parent that does not release the far end after the spawn never observes the
// end of input on its own --- the classic deadlock of this arrangement, and the
// reason the release is declared beside the operation rather than left to
// openkal.stream, which has no release at all.
int kal_process_channel(kal_stream* mine, kal_stream* theirs) {
    if (mine == nullptr || theirs == nullptr) return kal_err_invalid;

    int fds[2] = { -1, -1 };
    // O_CLOEXEC on both. The far end is placed deliberately, by the spawn that
    // receives it; an end that leaked into every other started program would
    // keep the channel open after the intended reader had closed it, and the
    // writer would then never see the end of input.
    const okl_long r = okl::sys(okl::nr_pipe2, reinterpret_cast<okl_long>(fds),
                                okl::o_cloexec);
    if (okl::failed(r)) return okl::translate(r);

    // THE STREAMS ARE BARE DESCRIPTORS AND NOT PACKED HANDLES, because
    // openkal.stream's transfer operations take what the environment takes.
    // kal_fs_stream reports a file's stream the same way and for the same
    // reason.
    *mine   = kal_stream{ static_cast<kal_uintptr>(fds[0]) };   // the reading end
    *theirs = kal_stream{ static_cast<kal_uintptr>(fds[1]) };   // the writing end
    return kal_ok;
}

void kal_process_channel_close(kal_stream s) {
    // A bare descriptor, so there is no generation to retire. The standard
    // streams are borrowed and are numbered 0, 1 and 2; closing one of those
    // through this operation would take a stream away from the whole program,
    // so they are refused rather than closed.
    const okl_long fd = static_cast<okl_long>(s.h);
    if (fd < 3) return;
    okl::sys(okl::nr_close, fd);
}


int kal_process_wait(kal_process h, int* status, int* terminated_by_environment) {
    if (h.h == 0) return kal_err_invalid;
    if (handle_kind(h) == kind_self) return kal_err_invalid;
    if (handle_kind(h) == kind_opened || handle_kind(h) == kind_observed) return kal_err_not_supported;
    // NOT COLLECTED HERE (0.21): the program is waited for and left in the
    // kernel's record until `kal_process_close', so that what it used --- its
    // peak memory, which the record keeps --- can still be asked about. A
    // program ended by its unit's limit is told apart by the unit (openkal 0.20).
    bool ended = false;
    return okl::child_outcome(static_cast<okl_long>(h.h & 0xffffffffu), true, status,
                              terminated_by_environment, ended);
}

// ONE PROGRAM, WHATEVER UNIT IT IS IN.
//
// An earlier draft made this reach the whole group when the started program had
// formed one, recovering that fact with `getpgid(pid) == pid'. It worked, and it
// was the wrong shape: the meaning of this operation then turned on a property of
// the handle that no caller could see. The unit has its own operation below, and
// the caller says which of the two it means.
int kal_process_terminate(kal_process h) {
    if (h.h == 0) return kal_err_invalid;
    if (handle_kind(h) == kind_self) return kal_err_invalid;
    if (handle_kind(h) == kind_opened || handle_kind(h) == kind_observed) return kal_err_not_supported;
    const okl_long r = okl::sys(okl::nr_kill, static_cast<okl_long>(h.h), 15 /* SIGTERM */);
    return okl::failed(r) ? okl::translate(r) : kal_ok;
}

// This program itself joins or forms a unit --- the operation `kal_spawn.job'
// cannot express, because that one places a program the caller STARTS and a copy
// wishing to lead a unit must say so about ITSELF before it replaces itself.
int kal_process_job_enter(kal_job* j) {
    if (j == nullptr) return kal_err_invalid;
    const okl_long join = static_cast<okl_long>(j->h);
    const okl_long r = okl::sys(okl::nr_setpgid, 0, join);
    if (okl::failed(r)) return okl::translate(r);
    // The identity of a group is its leader's, so a program that has just formed
    // one reports its own. Read back rather than assumed: `setpgid(0, 0)' makes
    // this program the leader, and `getpid' is that leader's identifier.
    if (join == 0) j->h = static_cast<kal_uintptr>(okl::sys(okl::nr_getpid));
    return kal_ok;
}

// Every program in the unit, including ones this implementation never held a
// handle to --- which is the whole reason a unit exists.
//
// AND IT IS THE SIGNAL THAT CANNOT BE DECLINED, WHICH IS A DECISION AND NOT
// A DETAIL.
//
// `kal_process_terminate' upon ONE program uses the polite one: a caller holds
// that program's handle, can wait for it, and can terminate it again. None of
// that is true of a unit. A unit exists because its members include programs the
// caller never held a handle to and cannot enumerate --- and a request that any
// one of them may ignore does not terminate the unit, it terminates the part of
// it that agreed.
//
// Measured with a consumer's own test: a shell that traps the polite signal
// and loops. Asked politely, the unit outlived every deadline; the caller's
// escalation could not help, because openkal has no vocabulary for "and this
// time I mean it" --- it has no signals at all.
//
// ⇒ So the operation does what its name says. WHAT THIS COSTS IS REAL: a
// member gets no chance to clean up, where on a system programmed directly a
// caller would send the polite signal first and wait. A caller that wants that
// still has it --- `kal_process_terminate' upon the member it holds --- and what it
// cannot do is ask a unit politely.
//
// A GROUP IS NAMED BY A PROCESS IDENTIFIER, AND THOSE ARE REUSED. Once the
// program that formed the group has ended and the numbers have wrapped, this can
// reach a different group. That is what this system does --- every program that
// calls `killpg' lives with it --- and the interface records it rather than
// reading as though it were not so.
int kal_process_job_terminate(kal_job j) {
    if (j.h == 0) return kal_err_invalid;
    const okl_long r = okl::sys(okl::nr_kill, -static_cast<okl_long>(j.h), 9 /* SIGKILL */);
    return okl::failed(r) ? okl::translate(r) : kal_ok;
}

// RELEASES NOTHING AND ENDS NOTHING. A group here is a number, not a resource,
// so there is no handle to close --- and the operation exists so that a caller
// need not know that. Where the unit IS a resource, releasing it must still not
// end its members; the interface says so at the declaration.
void kal_process_job_close(kal_job j) { if (j.h != 0) okl::job_forget(static_cast<okl_long>(j.h)); }

// Releasing the handle does not affect the program. A program that has not been
// waited for continues, and this environment collects it when the caller exits.
void kal_process_close(kal_process h) {
    // A started program that has ended is collected now, which `kal_process_wait'
    // did not do (0.21); one still running continues, as before.
    if (handle_kind(h) == kind_child && h.h != 0) {
        alignas(8) unsigned char info[128] = {};
        okl_long r;
        do r = okl::sys(okl::nr_waitid, 1 /* P_PID */, handle_pid(h), reinterpret_cast<okl_long>(info),
                        4 | 1 /* WEXITED | WNOHANG */, 0);
        while (okl::interrupted(r));
        return;
    }
    if (handle_kind(h) == kind_opened) {
        const okl_long fd = static_cast<okl_long>((h.h >> 32) & 0xffffu) - 1;
        if (fd >= 0) okl::sys(okl::nr_close, fd);
    }
}

// A WORD THE ENVIRONMENT SETS WHEN SOMEBODY HAS ASKED THIS PROGRAM TO END.
//
// A HANDLER AND NOT A WAITING CONTEXT, AND THE REASON IS WHICH ONE CAN BE
// ARMED WITHOUT DISTURBING A PROGRAM THAT NEVER ASKS. Consuming these signals
// from a context of its own would require them BLOCKED IN EVERY context, and
// blocking is per-context and inherited: a program that already had contexts
// running when it first asked would have some that still take the default
// action, and a program that never asks would have been made unkillable at
// startup. A disposition is per PROGRAM and can be installed at any moment.
//
// The handler does two things, and both are safe to do from one: store a word,
// and wake whoever waits on it. `kal_task_wait' is what a caller waits with, so
// the wake is the same operation `kal_task_wake' performs --- issued here as the
// raw call, because a handler may not enter code that takes a lock.
//
// THE RESTORER IS SUPPLIED HERE ON ONE ARCHITECTURE AND BY THE KERNEL ON THE
// OTHER. On x86_64 a disposition installed without SA_RESTORER faults on return
// from the handler --- the C library normally supplies the three instructions,
// and this implementation has no C library beneath it. On aarch64 the kernel
// supplies it and the flag must NOT be set.
namespace {

kal_u32 g_stop_word = 0;
int     g_stop_armed = 0;

#if defined(__x86_64__)
extern "C" void okl_sigreturn_trampoline(void);
// IN `.text', SAID IN SO MANY WORDS (0.18). A top-level `asm' is emitted into
// whatever section the compiler is in at that point of the file, and GCC at
// -O0 was in `.bss': the three instructions were zero bytes in memory that may
// not be executed, and the first request to end a program faulted on its way
// back from the handler (measured: SIGSEGV after SIGTERM, GCC -O0 only).
asm(".pushsection .text,\"ax\",@progbits\n"
    ".globl okl_sigreturn_trampoline\n"
    "okl_sigreturn_trampoline:\n"
    "  movq $15, %rax\n"      // rt_sigreturn
    "  syscall\n"
    ".popsection\n");
constexpr unsigned long sa_restorer_flag = 0x04000000u;   // SA_RESTORER
#endif

// Which request arrived first, for `kal_process_end_as_requested'.
int g_stop_signal = 0;

void stop_handler(int signo) {
    int none = 0;
    __atomic_compare_exchange_n(&g_stop_signal, &none, signo, false,
                                __ATOMIC_RELAXED, __ATOMIC_RELAXED);
    __atomic_store_n(&g_stop_word, 1u, __ATOMIC_RELEASE);
    // PRIVATE, AS `kal_task_wait' WAITS (0.18.2). The kernel keys a private
    // futex apart from a shared one at the same address, so a shared wake
    // reached no context waiting through `kal_task_wait' with no bound, and a
    // program that waits for its end to be requested never ended.
    okl::sys(okl::nr_futex, reinterpret_cast<okl_long>(&g_stop_word),
             okl::futex_wake | okl::futex_private, 0x7fffffff, 0, 0, 0);
}

// THE RESULT IS EXAMINED, AND IT WAS NOT WHEN THIS SHIPPED IN 0.11. An
// installation that failed would leave a word that can never change, and
// answering the caller with one is `reporting success having done nothing' in
// its exact form: the program asks whether its end has been requested, is told
// no, and goes on being told no after it has been. Found reviewing the same
// code written for the other kernel, where the trampoline made the failure
// mode obvious.
bool arm_one(int signo) {
    struct { void* handler; unsigned long flags; void* restorer; unsigned long mask; } act {};
    // A REQUEST THE STARTER ARRANGED NOT TO DELIVER STAYS UNDELIVERED (0.17,
    // the take-over rule). A program started in the background is started with
    // these requests ignored, and taking them over would end it when its
    // starter meant it to survive them --- which this did until 0.18.
    {
        decltype(act) old{};
        const okl_long r = okl::sys(okl::nr_rt_sigaction, signo, 0,
                                    reinterpret_cast<okl_long>(&old), sizeof old.mask);
        if (!okl::failed(r) && old.handler == reinterpret_cast<void*>(1) /* SIG_IGN */) return true;
    }
    act.handler = reinterpret_cast<void*>(&stop_handler);
#if defined(__x86_64__)
    act.flags   = sa_restorer_flag;
    act.restorer = reinterpret_cast<void*>(&okl_sigreturn_trampoline);
#endif
    return !okl::failed(okl::sys(okl::nr_rt_sigaction, signo,
                                 reinterpret_cast<okl_long>(&act), 0, sizeof act.mask));
}

}  // namespace

// ARMED ON THE FIRST ENQUIRY AND NOT AT STARTUP. A program that never asks
// keeps the default action, which is what every program that has never heard of
// this operation expects --- and it is the only arrangement under which adding
// this operation changes nothing for anyone who does not use it.
const kal_u32* kal_process_stop_requested(void) {
    // THREE STATES AND NOT TWO: not yet tried, armed, refused. A second
    // caller is told what the first found rather than arming again.
    int state = __atomic_load_n(&g_stop_armed, __ATOMIC_ACQUIRE);
    if (state == 0) {
        // SIGTERM, SIGINT and SIGHUP (0.18.1): the terminal that started a
        // program closing is a request to end it as much as the other two, and
        // a program that took the requests over and not this one was ended by
        // it before it could finish in order.
        const bool ok = arm_one(15) && arm_one(2) && arm_one(1);
        state = ok ? 1 : -1;
        __atomic_store_n(&g_stop_armed, state, __ATOMIC_RELEASE);
    }
    return state == 1 ? &g_stop_word : nullptr;
}

// --- what is known of a program (0.17) --------------------------------------
namespace {

// `pidfd_open' is the kernel's handle on a program, and it is what lets a
// handle found by an identifier keep naming that program after the number is
// given again (5.3 and later). Asked once.
int g_pidfd = -1;
bool open_id_available() {
    int v = __atomic_load_n(&g_pidfd, __ATOMIC_ACQUIRE);
    if (v < 0) {
        const okl_long fd = okl::sys(okl::nr_pidfd_open, okl::sys(okl::nr_getpid), 0);
        v = okl::failed(fd) ? 0 : 1;
        if (!okl::failed(fd)) okl::sys(okl::nr_close, fd);
        __atomic_store_n(&g_pidfd, v, __ATOMIC_RELEASE);
    }
    return v == 1;
}

// /proc/<pid>/stat, or /proc/self/stat for pid 0, read whole into `b'; the
// length, or 0 when there is no such entry.
okl_long read_stat(okl_long pid, char (&b)[1024]) {
    char path[40] = "/proc/";
    char* o = path + 6;
    if (pid == 0) { okl::copy(o, "self", 4); o += 4; }
    else o = put_decimal(o, static_cast<okl_uptr>(pid));
    okl::copy(o, "/stat", 6);
    const okl_long fd = okl::sys(okl::nr_openat, okl::at_fdcwd, reinterpret_cast<okl_long>(path),
                                 okl::o_rdonly | okl::o_cloexec, 0);
    if (okl::failed(fd)) return 0;
    const okl_long n = okl::sys(okl::nr_read, fd, reinterpret_cast<okl_long>(b), sizeof b - 1);
    okl::sys(okl::nr_close, fd);
    return n > 0 ? n : 0;
}

// The fields after the name, which is in parentheses and may itself contain
// them, so the last `)' is where the fields begin: the position of field
// `wanted' (the state is 3, the processor times 14 and 15, the start 22), or -1.
okl_long stat_field(const char* b, okl_long n, int wanted) {
    okl_long close = -1;
    for (okl_long i = 0; i < n; ++i) if (b[i] == ')') close = i;
    if (close < 0) return -1;
    int field = 2;                                   // the name was field 2
    okl_long i = close + 1;
    while (i < n && field < wanted) { if (b[i] == ' ') ++field; ++i; }
    return field == wanted && i < n ? i : -1;
}

okl_u64 stat_number(const char* b, okl_long n, int wanted, bool& ok) {
    const okl_long at = stat_field(b, n, wanted);
    ok = at >= 0;
    okl_u64 v = 0;
    for (okl_long i = at; ok && i < n && b[i] >= '0' && b[i] <= '9'; ++i) v = v * 10 + static_cast<okl_u64>(b[i] - '0');
    return v;
}

// Field 22: when the program started, in the kernel's clock ticks since boot.
// Of this program (pid 0) through /proc/self: under a sandbox that numbers its
// programs apart from the /proc it shows, `getpid' names another program's entry
// --- measured beneath a language server's own sandbox, every instance read pid
// 1's start, and a killed instance and its restart were one program.
bool started_ticks(okl_long pid, okl_u64& ticks) {
    char b[1024];
    const okl_long n = read_stat(pid, b);
    bool ok = false;
    ticks = stat_number(b, n, 22, ok);
    return ok;
}

// The state letter (field 3), or 0 when there is no entry: a program whose
// entry is gone, or is Z or X, has ended.
char stat_state(okl_long pid) {
    char b[1024];
    const okl_long n = read_stat(pid, b);
    const okl_long at = stat_field(b, n, 3);
    return at >= 0 ? b[at] : 0;
}

// This program's number as the /proc it shows names it, which is the number
// another program finds it by there; `getpid' where there is no /proc.
okl_long self_number() {
    char b[24];
    const okl_long n = okl::sys(okl::nr_readlinkat, okl::at_fdcwd, reinterpret_cast<okl_long>("/proc/self"),
                                reinterpret_cast<okl_long>(b), sizeof b);
    okl_long v = 0;
    if (n > 0) for (okl_long i = 0; i < n && b[i] >= '0' && b[i] <= '9'; ++i) v = v * 10 + (b[i] - '0');
    return v > 0 ? v : okl::sys(okl::nr_getpid);
}

okl_u64 ticks_to_ns(okl_u64 t) {
    okl_ulong hz = okl::auxval(17 /* AT_CLKTCK */);
    if (hz == 0) hz = 100;
    return t / hz * 1000000000ull + t % hz * (1000000000ull / hz);
}

// The processor time of a program, through the clock the kernel keeps for
// every program (`MAKE_PROCESS_CPUCLOCK(pid, CPUCLOCK_SCHED)'): nanoseconds,
// for any program the caller may ask about.
//
// A program this one cannot name to the kernel --- one numbered by another
// namespace, observed through /proc --- is answered from its entry's processor
// times (fields 14 and 15, in clock ticks).
bool cpu_ns(okl_long pid, bool self, okl_u64& out) {
    const okl_long clock = self ? 2 /* CLOCK_PROCESS_CPUTIME_ID */
                                : static_cast<okl_long>(static_cast<int>((~static_cast<unsigned>(pid)) << 3) | 2);
    struct { okl_long sec, nsec; } ts{};
    const okl_long r = okl::sys(okl::nr_clock_gettime, clock, reinterpret_cast<okl_long>(&ts));
    if (!okl::failed(r)) {
        out = static_cast<okl_u64>(ts.sec) * 1000000000ull + static_cast<okl_u64>(ts.nsec);
        return true;
    }
    char b[1024];
    const okl_long n = read_stat(self ? 0 : pid, b);
    bool user_ok = false, system_ok = false;
    const okl_u64 user = stat_number(b, n, 14, user_ok), system = stat_number(b, n, 15, system_ok);
    if (!user_ok || !system_ok) return false;
    out = ticks_to_ns(user + system);
    return true;
}

// Whether a program has ended. A started one: `waitid' without collecting it
// (WNOWAIT), and one already collected has ended too. One found by its
// identifier: its descriptor is readable once it has ended.
bool has_ended(kal_process h) {
    if (handle_kind(h) == kind_self) return false;
    if (handle_kind(h) == kind_observed) {
        const char state = stat_state(handle_pid(h));
        okl_u64 t = 0;
        return state == 0 || state == 'Z' || state == 'X' || !started_ticks(handle_pid(h), t)
            || (t & 0xffffffu) != ((h.h >> 32) & 0xffffffu);
    }
    if (handle_kind(h) == kind_opened) {
        const okl_long fd = static_cast<okl_long>((h.h >> 32) & 0xffffu) - 1;
        if (fd < 0) return false;
        struct { int fd; short events, revents; } pfd{ static_cast<int>(fd), 1 /* POLLIN */, 0 };
        struct { okl_long sec, nsec; } zero{0, 0};
        const okl_long r = okl::sys(okl::nr_ppoll, reinterpret_cast<okl_long>(&pfd), 1,
                                    reinterpret_cast<okl_long>(&zero), 0, 0);
        return r > 0 && (pfd.revents & 1) != 0;
    }
    alignas(8) unsigned char info[128] = {};
    const okl_long r = okl::sys(okl::nr_waitid, 1 /* P_PID */, handle_pid(h),
                                reinterpret_cast<okl_long>(info),
                                4 | 1 | 0x01000000 /* WEXITED | WNOHANG | WNOWAIT */, 0);
    if (r == -okl::e_child) return true;
    if (okl::failed(r)) return false;
    int si_pid = 0;
    okl::copy(reinterpret_cast<char*>(&si_pid), reinterpret_cast<const char*>(info + 16), sizeof si_pid);
    return si_pid != 0;
}

}  // namespace

int kal_process_self(kal_process* out) {
    if (out == nullptr) return kal_err_invalid;
    *out = kal_process{ (static_cast<kal_uintptr>(kind_self) << 56)
                        | static_cast<kal_uintptr>(okl::sys(okl::nr_getpid)) };
    return kal_ok;
}

int kal_process_open_id(kal_u64 id, kal_process* out) {
    if (out == nullptr || id == 0 || id > 0x7fffffffu) return id == 0 || out == nullptr
                                                             ? kal_err_invalid : kal_err_not_found;
    // NUMBERED AS /proc NUMBERS IT, WHERE THE KERNEL DOES NOT. A program in
    // another namespace than this one's --- a sandbox that shows the system's
    // /proc --- is not found by `pidfd_open', and a kernel before 5.3 has no
    // such call; its entry is observed instead, with no descriptor: ended when
    // the entry is gone or is a zombie's.
    const auto observed = [&]() -> int {
        const char state = stat_state(static_cast<okl_long>(id));
        okl_u64 t = 0;
        if (state == 0 || state == 'Z' || state == 'X' || !started_ticks(static_cast<okl_long>(id), t))
            return kal_err_not_found;
        *out = kal_process{ (static_cast<kal_uintptr>(kind_observed) << 56)
                            | (static_cast<kal_uintptr>(t & 0xffffffu) << 32) | static_cast<kal_uintptr>(id) };
        return kal_ok;
    };
    if (!open_id_available()) {
        if (stat_state(0) == 0) return kal_err_not_supported;
        return observed();
    }
    const okl_long fd = okl::sys(okl::nr_pidfd_open, static_cast<okl_long>(id), 0);
    if (fd == -3 /* ESRCH */) return observed();
    if (okl::failed(fd)) return okl::translate(fd);
    if (fd >= 0xffff) { okl::sys(okl::nr_close, fd); return kal_err_no_memory; }
    okl::sys(okl::nr_fcntl, fd, okl::f_setfd, okl::fd_cloexec);
    *out = kal_process{ (static_cast<kal_uintptr>(kind_opened) << 56)
                        | (static_cast<kal_uintptr>(fd + 1) << 32)
                        | static_cast<kal_uintptr>(id) };
    return kal_ok;
}

int kal_process_info(kal_process p, kal_u32 wanted, kal_program_info* out) {
    if (out == nullptr || p.h == 0 || out->self_size < 8) return kal_err_invalid;
    const kal_u32 self = out->self_size;
    kal_program_info v{};
    v.self_size = self;
    const bool me = handle_kind(p) == kind_self;
    const okl_long pid = handle_pid(p);
    v.id = static_cast<kal_u64>(me ? self_number() : pid);
    v.present |= KAL_PINFO_ID;
    const bool ended = has_ended(p);
    v.ended = ended ? 1u : 0u;
    v.present |= KAL_PINFO_ENDED;
    if (!ended && (wanted & KAL_PINFO_STARTED)) {
        okl_u64 t = 0;
        // Read, then checked again: a program found by its identifier that
        // ended meanwhile may have had its number given to another, whose start
        // this would have read.
        if (started_ticks(me ? 0 : pid, t) && !has_ended(p)) {
            v.started_ns = ticks_to_ns(t);
            v.present |= KAL_PINFO_STARTED;
        }
    }
    // CPU (0.21): of a started program that has ended, from the record the
    // kernel keeps until it is collected, as its peak memory below.
    if (wanted & KAL_PINFO_CPU) {
        okl_u64 c = 0;
        const bool ok = !ended ? cpu_ns(pid, me, c)
                     : handle_kind(p) == kind_child && okl::child_cpu(pid, c);
        if (ok) { v.cpu_ns = c; v.present |= KAL_PINFO_CPU; }
    }
    // MEMORY (openkal 0.20): resident now and at most, from the kernel's record;
    // of a started program that has ended, its peak from the record it keeps
    // until it is collected. Only where the structure has room for the fields.
    if ((wanted & KAL_PINFO_MEMORY) && self >= sizeof v) {
        okl_u64 now = 0, peak = 0;
        const bool ok = me ? okl::child_memory(0, false, now, peak)
                     : handle_kind(p) == kind_child ? okl::child_memory(pid, ended, now, peak)
                     : (!ended && okl::child_memory(pid, false, now, peak));
        if (ok) { v.memory_bytes = now; v.memory_peak_bytes = peak; v.present |= KAL_PINFO_MEMORY; }
    }
    const kal_u32 n = self < sizeof v ? self : static_cast<kal_u32>(sizeof v);
    okl::copy(reinterpret_cast<char*>(out), reinterpret_cast<const char*>(&v), n);
    return kal_ok;
}

// ENDS AS THE REQUEST WOULD HAVE ENDED THE PROGRAM (0.17): the request's own
// default treatment restored, the request unblocked and sent again. Its
// default is to end the program, so the starter's wait reports the signal; a
// program that is somehow still here a moment later ends with the status a
// shell gives such a program.
int kal_process_end_as_requested(void) {
    if (__atomic_load_n(&g_stop_word, __ATOMIC_ACQUIRE) == 0) return kal_err_invalid;
    int signo = __atomic_load_n(&g_stop_signal, __ATOMIC_RELAXED);
    if (signo == 0) signo = 15;
    struct { void* handler; unsigned long flags; void* restorer; unsigned long mask; } act {};
#if defined(__x86_64__)
    act.flags    = sa_restorer_flag;
    act.restorer = reinterpret_cast<void*>(&okl_sigreturn_trampoline);
#endif
    okl::sys(okl::nr_rt_sigaction, signo, reinterpret_cast<okl_long>(&act), 0, sizeof act.mask);
    unsigned long set = 1ul << (signo - 1);
    okl::sys(okl::nr_rt_sigprocmask, 1 /* SIG_UNBLOCK */, reinterpret_cast<okl_long>(&set), 0, sizeof set);
    okl::sys(okl::nr_kill, okl::sys(okl::nr_getpid), signo);
    for (int i = 0; i < 100; ++i) {
        struct { okl_long sec, nsec; } ts{0, 1000000};
        okl::sys(okl::nr_nanosleep, reinterpret_cast<okl_long>(&ts), 0);
    }
    okl::sys(okl::nr_exit_group, 128 + signo);
    for (;;) { }
}

kal_uintptr kal_process_props(void) {
    return KAL_PROCESS_PROP_TERMINATE | KAL_PROCESS_PROP_STREAM_PASSING
         | KAL_PROCESS_PROP_EXIT_STATUS
         | KAL_PROCESS_PROP_CHANNEL | KAL_PROCESS_PROP_GRANT_DIR
         | KAL_PROCESS_PROP_BOUND_LIFETIME
         | KAL_PROCESS_PROP_JOB | KAL_PROCESS_PROP_JOB_USAGE
         | KAL_PROCESS_PROP_JOB_LIMIT_SUPERVISED
         | KAL_PROCESS_PROP_NULL_STREAMS
         | (open_id_available() || stat_state(0) != 0 ? KAL_PROCESS_PROP_OPEN_ID : 0u)
         // AGREES WITH `kal_process_stop_requested', because the header
         // defines null there as the absence this position reports. Read and
         // never armed: asking what an implementation can do must not install a
         // disposition, so the position is claimed until an installation has
         // actually been refused.
         | (__atomic_load_n(&g_stop_armed, __ATOMIC_ACQUIRE) == -1
                ? 0u : KAL_PROCESS_PROP_STOP_REQUESTED);
}


// --- what a program and a unit use, and a limit on a unit (0.20) ------------
//
// A PROGRAM'S MEMORY is what this kernel says is resident for it, and the most
// that ever was (/proc/<pid>/status: VmRSS and VmHWM). An ended program that has
// not been collected keeps its peak in the record the kernel holds until it is
// collected, and `waitid' with WNOWAIT reads it from there --- which is why
// `kal_process_wait' leaves the program uncollected and `kal_process_close'
// collects it (0.21): what it used remains readable while its handle is held.
//
// A UNIT IS A PROCESS GROUP, and the kernel records every program's group: the
// programs in a unit are the entries of /proc whose group is the unit's, which
// counts the programs its members started too. The kernel lists no group's
// members, so a unit's use is a walk of /proc --- one read per program on the
// machine, which is why nothing walks it unless a caller asked for a use or
// placed a limit.
//
// A LIMIT IS SUPERVISED. A context of this implementation, started with the
// first limit, watches the limited units and ends one it sees over its limit,
// at an interval of at most 500 ms (500 ms below half the limit, 200 ms below
// four fifths, 100 ms above).
//
// NOT HARD, AND WHY. The kernel holds a set of programs to a ceiling through a
// control group, and a program may make one only in a part of the hierarchy
// that was delegated to it; whether a part was delegated is the manager's
// record, not the kernel's, so an implementation that made groups wherever it
// could write would make them where it was not entitled to --- on an ordinary
// desktop the program's own slice is writable and is not delegated. And where
// a group is delegated, a program inside it may give it children with a
// ceiling only after moving itself out of it, which is the program's decision
// to make and not an interface's. A program that holds a delegation and wants
// the kernel's ceiling uses it directly (clause 1.1).
namespace {

okl_long read_small(const char* path, char* b, okl_long cap) {
    const okl_long fd = okl::sys(okl::nr_openat, okl::at_fdcwd, reinterpret_cast<okl_long>(path),
                                 okl::o_rdonly | okl::o_cloexec, 0);
    if (okl::failed(fd)) return -1;
    okl_long n = 0;
    while (n < cap - 1) {
        const okl_long r = okl::sys(okl::nr_read, fd, reinterpret_cast<okl_long>(b + n), cap - 1 - n);
        if (okl::interrupted(r)) continue;
        if (r <= 0) break;
        n += r;
    }
    okl::sys(okl::nr_close, fd);
    b[n] = 0;
    return n;
}

okl_uptr length_of(const char* s) { okl_uptr n = 0; while (s[n]) ++n; return n; }

// The number after `key' in a text of lines `key: number unit', or false.
bool field_value(const char* b, okl_long n, const char* key, okl_u64& out) {
    const okl_uptr k = length_of(key);
    for (okl_long i = 0; i + static_cast<okl_long>(k) <= n; ++i) {
        if ((i == 0 || b[i - 1] == '\n') && same_prefix(b + i, key, k)) {
            okl_long j = i + static_cast<okl_long>(k);
            while (j < n && (b[j] == ' ' || b[j] == '\t' || b[j] == ':')) ++j;
            okl_u64 v = 0; bool any = false;
            while (j < n && b[j] >= '0' && b[j] <= '9') { v = v * 10 + static_cast<okl_u64>(b[j] - '0'); ++j; any = true; }
            out = v;
            return any;
        }
    }
    return false;
}

okl_u64 page_bytes() {
    const okl_ulong p = okl::auxval(6 /* AT_PAGESZ */);
    return p != 0 ? p : 4096;
}

// A running program's memory, from /proc/<pid>/status (pid 0: this one).
bool resident(okl_long pid, okl_u64& now, okl_u64& peak) {
    char path[40] = "/proc/";
    char* o = path + 6;
    if (pid == 0) { okl::copy(o, "self", 4); o += 4; }
    else o = put_decimal(o, static_cast<okl_uptr>(pid));
    okl::copy(o, "/status", 8);
    char b[2048];
    const okl_long n = read_small(path, b, sizeof b);
    if (n <= 0) return false;
    okl_u64 rss = 0, hwm = 0;
    if (!field_value(b, n, "VmRSS", rss)) return false;
    if (!field_value(b, n, "VmHWM", hwm)) hwm = rss;
    now = rss * 1024; peak = (hwm < rss ? rss : hwm) * 1024;
    return true;
}

// An ended program not yet collected: its record, read without collecting it.
struct k_rusage { okl_long utime_s, utime_us, stime_s, stime_us, maxrss; okl_long rest[13]; };
bool ended_usage(okl_long pid, k_rusage& ru) {
    alignas(8) unsigned char info[128] = {};
    const okl_long r = okl::sys(okl::nr_waitid, 1 /* P_PID */, pid, reinterpret_cast<okl_long>(info),
                                4 | 1 | 0x01000000 /* WEXITED | WNOHANG | WNOWAIT */,
                                reinterpret_cast<okl_long>(&ru));
    if (okl::failed(r)) return false;
    int si_pid = 0;
    okl::copy(&si_pid, info + 16, sizeof si_pid);
    return si_pid != 0;
}

// --- the units this program limited -----------------------------------------

struct limited {
    okl_long unit;          // the group's identifier; 0 for a free row
    okl_u64  limit;         // bytes; 0 for none
    int      ended;         // the supervisor ended it for its limit
};
constexpr int kLimited = 64;
limited g_limited[kLimited];
volatile int g_limited_lock = 0;
void lock_limited()   { while (__atomic_exchange_n(&g_limited_lock, 1, __ATOMIC_ACQUIRE)) okl::sys(okl::nr_sched_yield); }
void unlock_limited() { __atomic_store_n(&g_limited_lock, 0, __ATOMIC_RELEASE); }

limited* row_of(okl_long unit) {
    for (auto& r : g_limited) if (r.unit == unit) return &r;
    return nullptr;
}

// --- the programs of a unit -----------------------------------------------------

// Every program whose group is `unit', with its resident memory summed.
// Returns how many there are.
kal_u32 members_of(okl_long unit, okl_u64& memory) {
    memory = 0;
    kal_u32 count = 0;
    const okl_long dir = okl::sys(okl::nr_openat, okl::at_fdcwd, reinterpret_cast<okl_long>("/proc"),
                                  okl::o_rdonly | okl::o_directory | okl::o_cloexec, 0);
    if (okl::failed(dir)) return 0;
    alignas(8) char buf[8192];
    const okl_u64 page = page_bytes();
    for (;;) {
        const okl_long n = okl::sys(okl::nr_getdents64, dir, reinterpret_cast<okl_long>(buf), sizeof buf);
        if (n <= 0) break;
        for (okl_long at = 0; at < n;) {
            const auto* d = reinterpret_cast<const okl::kdirent64*>(buf + at);
            at += d->reclen;
            const char* name = d->name;
            if (name[0] < '1' || name[0] > '9') continue;
            okl_long pid = 0;
            for (const char* c = name; *c; ++c) pid = pid * 10 + (*c - '0');
            char b[1024];
            const okl_long len = read_stat(pid, b);
            if (len <= 0) continue;
            bool ok = false;
            if (static_cast<okl_long>(stat_number(b, len, 5, ok)) != unit || !ok) continue;
            bool rok = false;
            const okl_u64 rss = stat_number(b, len, 24, rok);
            if (rok) memory += rss * page;
            ++count;
        }
    }
    okl::sys(okl::nr_close, dir);
    return count;
}

// --- the supervisor -------------------------------------------------------------

int g_supervising = 0;

void supervise(void*) {
    for (;;) {
        okl_u64 worst = 0;   // the largest use, in thousandths of its limit
        bool any = false;
        for (int i = 0; i < kLimited; ++i) {
            lock_limited();
            const limited r = g_limited[i];
            unlock_limited();
            if (r.unit == 0 || r.limit == 0 || r.ended) continue;
            any = true;
            okl_u64 memory = 0;
            const kal_u32 n = members_of(r.unit, memory);
            if (n == 0) continue;
            if (memory > r.limit) {
                // Marked before it is ended, so that a wait that returns
                // because of the end already finds why.
                lock_limited();
                if (g_limited[i].unit == r.unit) g_limited[i].ended = 1;
                unlock_limited();
                okl::sys(okl::nr_kill, -r.unit, 9 /* SIGKILL */);
                continue;
            }
            const okl_u64 share = memory * 1000 / r.limit;
            if (share > worst) worst = share;
        }
        if (!any) {
            __atomic_store_n(&g_supervising, 0, __ATOMIC_RELEASE);
            // A limit placed between the last look and the store above found
            // a supervisor still running and started none.
            lock_limited();
            bool again = false;
            for (const auto& r : g_limited) if (r.unit && r.limit && !r.ended) again = true;
            unlock_limited();
            if (!again || __atomic_exchange_n(&g_supervising, 1, __ATOMIC_ACQ_REL)) return;
            continue;
        }
        const okl_i64 ms = worst < 500 ? 500 : worst < 800 ? 200 : 100;
        okl::ktimespec ts{ 0, ms * 1000000 };
        okl::sys(okl::nr_nanosleep, reinterpret_cast<okl_long>(&ts), 0);
    }
}

// Why a program that ended on SIGKILL ended: its unit's limit, where the
// supervisor ended the unit for it.
bool ended_by_limit(okl_long pid) {
    char b[1024];
    const okl_long len = read_stat(pid, b);
    if (len <= 0) return false;
    bool ok = false;
    const okl_long unit = static_cast<okl_long>(stat_number(b, len, 5, ok));
    if (!ok || unit <= 0) return false;
    lock_limited();
    const limited* r = row_of(unit);
    const limited copy = r ? *r : limited{};
    unlock_limited();
    return copy.unit != 0 && copy.ended != 0;
}

}  // namespace

extern "C++" {
namespace okl {
void job_forget(okl_long unit) {
    lock_limited();
    if (limited* r = row_of(unit)) *r = limited{};
    unlock_limited();
}

bool child_memory(okl_long pid, bool ended, okl_u64& now, okl_u64& peak) {
    if (!ended) return resident(pid, now, peak);
    k_rusage ru{};
    if (!ended_usage(pid, ru)) return false;
    now = 0;
    peak = static_cast<okl_u64>(ru.maxrss) * 1024;
    return true;
}

bool child_cpu(okl_long pid, okl_u64& ns) {
    k_rusage ru{};
    if (!ended_usage(pid, ru)) return false;
    ns = static_cast<okl_u64>(ru.utime_s + ru.stime_s) * 1000000000ull
       + static_cast<okl_u64>(ru.utime_us + ru.stime_us) * 1000ull;
    return true;
}

// How a started program ended, read without collecting it; `block' waits for
// it. Shared with openkal.timeout.
int child_outcome(okl_long pid, bool block, int* status, int* terminated, bool& ended) {
    alignas(8) unsigned char info[128] = {};
    okl_long r;
    do r = okl::sys(okl::nr_waitid, 1 /* P_PID */, pid, reinterpret_cast<okl_long>(info),
                    4 | 0x01000000 | (block ? 0 : 1) /* WEXITED | WNOWAIT [| WNOHANG] */, 0);
    while (okl::interrupted(r));
    ended = false;
    if (okl::failed(r)) return okl::translate(r);
    int si_pid = 0, si_code = 0, si_status = 0;
    okl::copy(&si_code, info + 8, sizeof si_code);
    okl::copy(&si_pid, info + 16, sizeof si_pid);
    okl::copy(&si_status, info + 24, sizeof si_status);
    if (si_pid == 0) return kal_ok;          // not yet
    ended = true;
    if (si_code == 1 /* CLD_EXITED */) {
        if (status) *status = si_status & 0xff;
        if (terminated) *terminated = 0;
    } else {
        if (status) *status = si_status;
        if (terminated)
            *terminated = (si_status == 9 && ended_by_limit(pid)) ? KAL_TERMINATED_LIMIT : 1;
    }
    return kal_ok;
}
}  // namespace okl
}  // extern "C++"

int kal_process_job_usage(kal_job j, kal_job_usage* out) {
    if (out == nullptr || j.h == 0 || out->self_size < 8) return kal_err_invalid;
    const okl_long unit = static_cast<okl_long>(j.h);
    kal_job_usage v{};
    v.self_size = out->self_size;
    okl_u64 memory = 0;
    v.members = members_of(unit, memory);
    if (v.members == 0) return kal_err_invalid;
    v.memory_bytes = memory;
    const kal_u32 n = out->self_size < sizeof v ? out->self_size : static_cast<kal_u32>(sizeof v);
    okl::copy(reinterpret_cast<char*>(out), reinterpret_cast<const char*>(&v), n);
    return kal_ok;
}

int kal_process_job_limit(kal_job j, kal_uintptr kind, kal_u64 value) {
    if (kind != KAL_JOB_LIMIT_MEMORY) return kal_err_invalid;
    if (j.h == 0) return kal_err_invalid;
    const okl_long unit = static_cast<okl_long>(j.h);

    lock_limited();
    limited* r = row_of(unit);
    if (r == nullptr && value != 0) r = row_of(0);
    if (r == nullptr) { unlock_limited(); return value == 0 ? kal_ok : kal_err_no_memory; }
    if (r->unit == 0) *r = limited{ unit, 0, 0 };
    r->limit = value;
    unlock_limited();

    if (value != 0 && !__atomic_exchange_n(&g_supervising, 1, __ATOMIC_ACQ_REL)) {
        kal_task t{};
        if (kal_task_start(&supervise, nullptr, &t) != kal_ok) {
            __atomic_store_n(&g_supervising, 0, __ATOMIC_RELEASE);
            return kal_err_no_memory;
        }
    }
    return kal_ok;
}

}