ProcessMemoryConcFuzz.cpp

Go to the documentation of this file.

 
#include "ProcessMemory.h"
#include "MemoryManager.h"
#include "ThreadManager.h"
#include "UnitTestFramework.h"
#include <map>
#include <vector>
#include <stdlib.h>
#include <string.h>
#if defined(__APPLE__)
    #include <mach-o/dyld.h>    // _NSGetExecutablePath, for the tier B re-exec
#elif !defined(WINDOWS)
    #include <unistd.h>         // readlink("/proc/self/exe")
#endif
 
namespace cmlabs {
 
// ---------------------------------------------------------------------------------
// Shared helpers
// ---------------------------------------------------------------------------------

struct ConcRand {
    uint64 s;
    ConcRand(uint64 seed) : s(seed ? seed : 0x9E3779B97F4A7C15ULL) {}
    uint64 next() { s ^= s >> 12; s ^= s << 25; s ^= s >> 27; return s * 2685821657736338717ULL; }
    uint32 below(uint32 n) { return n ? (uint32)(next() % n) : 0; }
    bool chance(uint32 pct) { return below(100) < pct; }
};
 
static uint32 ConcEnvU32(const char* name, uint32 def) {
    const char* v = getenv(name);
    if (!v || !*v) return def;
    long n = atol(v);
    return (n > 0) ? (uint32)n : def;
}
static bool ConcEnvFlag(const char* name) {
    const char* v = getenv(name);
    return v && *v == '1';
}

static uint32 ConcChecksum(const char* p, uint32 len) {
    uint32 h = 2166136261u;
    for (uint32 i = 0; i < len; i++) { h ^= (uint8)p[i]; h *= 16777619u; }
    return h;
}
static void ConcFillPayload(char* buf, uint32 len, uint32 serial) {
    for (uint32 i = 0; i < len; i++)
        buf[i] = (char)('A' + ((serial * 31 + i * 7) % 26));
    buf[len] = 0;
}

static DataMessage* ConcMakeMsg(uint32 serial, uint32 len, uint32& sumOut) {
    char* body = new char[len + 1];
    ConcFillPayload(body, len, serial);
    sumOut = ConcChecksum(body, len);
    DataMessage* m = new DataMessage();
    m->setSerial(serial);
    m->setInt("cfserial", (int64)serial);
    m->setInt("cflen", (int64)len);
    m->setInt("cfsum", (int64)sumOut);
    m->setString("cfbody", body);
    delete [] body;
    return m;
}
 
// ---------------------------------------------------------------------------------
// The set-equality oracle. Concurrency removes total order, so this records per-serial
// observation counts under a lock and reconciles at the end.
// ---------------------------------------------------------------------------------
 
struct ConcOracle {
    utils::Mutex mutex;
    std::map<uint32, uint32> readCount; // serial -> times returned by a reader
    std::map<uint32, uint32> expectSum; // serial -> expected payload checksum
    uint32 written;
    uint32 corrupt;                     // payload/len mismatches
    uint32 torn;                        // message did not parse as ours at all
    ConcOracle() : written(0), corrupt(0), torn(0) {}
 
    void noteWritten(uint32 serial, uint32 sum) {
        mutex.enter();
        expectSum[serial] = sum;
        written++;
        mutex.leave();
    }

    void noteRead(DataMessage* m, bool dupInject) {
        mutex.enter();
        if (!m || !m->isValid()) { torn++; mutex.leave(); return; }
        bool ok = false;
        int64 ser = m->getInt("cfserial", ok);
        if (!ok) { torn++; mutex.leave(); return; }
        int64 len = m->getInt("cflen");
        int64 sum = m->getInt("cfsum");
        std::string body = m->getAsString("cfbody");
        uint32 actual = ConcChecksum(body.c_str(), (uint32)body.size());
        if ((int64)body.size() != len || actual != (uint32)sum)
            corrupt++;
        readCount[(uint32)ser]++;
        // Positive control: count one read twice so the reconciliation MUST report a
        // duplicate. Proves the oracle can fail rather than always passing.
        if (dupInject) readCount[(uint32)ser]++;
        mutex.leave();
    }

    const char* reconcile(char* det, size_t detSize, bool lostInject) {
        mutex.enter();
        uint32 dup = 0, lost = 0, extra = 0;
        uint32 firstDup = 0, firstLost = 0;
        std::map<uint32, uint32>::const_iterator it;
        for (it = expectSum.begin(); it != expectSum.end(); ++it) {
            std::map<uint32, uint32>::const_iterator r = readCount.find(it->first);
            uint32 n = (r == readCount.end()) ? 0 : r->second;
            // Positive control: pretend one written serial was never read.
            if (lostInject && it->first == 1) n = 0;
            if (n == 0)      { lost++; if (!firstLost) firstLost = it->first; }
            else if (n > 1)  { dup++;  if (!firstDup)  firstDup  = it->first; }
        }
        for (it = readCount.begin(); it != readCount.end(); ++it)
            if (expectSum.find(it->first) == expectSum.end()) extra++;
        uint32 w = written, c = corrupt, t = torn;
        mutex.leave();
 
        snprintf(det, detSize,
            "written=%u distinct-read=%u LOST=%u DUPLICATED=%u never-written=%u "
            "corrupt=%u torn=%u (first lost=%u first dup=%u)",
            w, (uint32)readCount.size(), lost, dup, extra, c, t, firstLost, firstDup);
        if (t)     return "TORN";
        if (c)     return "CORRUPT";
        if (extra) return "READ_BUT_NEVER_WRITTEN";
        if (dup)   return "DUPLICATED";
        if (lost)  return "LOST";
        return NULL;
    }
};
 
// ---------------------------------------------------------------------------------
// Tier A / A' / C / D: threads inside one process, one shared queue.
// ---------------------------------------------------------------------------------
 
struct ConcCtx {
    MemoryManager* manager;
    ConcOracle*    oracle;
    uint16         procID;
    uint32         totalMsgs;
    uint32         payloadFixed;    // 0 = mixed sizes
    uint32         burst;           // 0 = steady; else write burst then drain burst
    uint64         seed;
    bool           resizeStress;    // tier C: force repeated growth under readers
    bool           blockingReads;   // tier D: readers wait on an empty queue
    volatile bool  writerDone;
    volatile bool  stop;
    uint32         dupInjectAt;     // positive control
    volatile uint32 readsSeen;
};
 
static THREAD_RET THREAD_FUNCTION_CALL ConcWriter(THREAD_ARG arg) {
    ConcCtx* c = (ConcCtx*)arg;
    ConcRand rng(c->seed ^ 0x1517ULL);
    uint32 serial = 1;
    while (serial <= c->totalMsgs && !c->stop) {
        // Burst mode reproduces the real shape: fill to ~burst entries, then let the
        // readers drain the lot in one go. The real defect always loses the TAIL of such
        // a burst (serials 61-65 of ~64 published), so a steady trickle may never
        // construct the state at all.
        uint32 n = c->burst ? c->burst : 1;
        for (uint32 i = 0; i < n && serial <= c->totalMsgs && !c->stop; i++) {
            uint32 len;
            if (c->payloadFixed)      len = c->payloadFixed;
            else if (c->resizeStress) len = 700 + rng.below(1600);
            else                      len = 1 + rng.below(300);
            uint32 sum = 0;
            DataMessage* m = ConcMakeMsg(serial, len, sum);
            // Note the serial BEFORE the write: if addToMsgQ succeeds and the message is
            // then lost, the oracle must already know it was owed.
            c->oracle->noteWritten(serial, sum);
            if (!c->manager->processMemory->addToMsgQ(c->procID, m)) {
                // A refused write is not a loss - withdraw the expectation so the
                // reconciliation does not blame the ring for backpressure.
                c->oracle->mutex.enter();
                c->oracle->expectSum.erase(serial);
                c->oracle->written--;
                c->oracle->mutex.leave();
            }
            delete m;
            serial++;
        }
        if (c->burst) {
            // let readers catch up so the next burst starts from a drained-ish ring
            for (uint32 g = 0; g < 50 && !c->stop; g++) {
                if (c->manager->processMemory->getMsgQCount(c->procID) == 0) break;
                utils::Sleep(1);
            }
        }
    }
    c->writerDone = true;
    return 0;
}
 
static THREAD_RET THREAD_FUNCTION_CALL ConcReader(THREAD_ARG arg) {
    ConcCtx* c = (ConcCtx*)arg;
    uint32 idleRounds = 0;
    while (!c->stop) {
        // Tier D uses a real timeout so waitForQ takes its blocking path (semaphore wait,
        // mutex released mid-call, re-enter). Tier A polls with a short timeout so the
        // readers contend hard on the non-empty fast path instead.
        DataMessage* m = c->manager->processMemory->waitForMsgQ(c->procID,
            c->blockingReads ? 50 : 2);
        if (m) {
            idleRounds = 0;
            uint32 seen = ++c->readsSeen;
            c->oracle->noteRead(m, c->dupInjectAt && seen == c->dupInjectAt);
            delete m;
            continue;
        }
        if (c->writerDone) {
            if (c->manager->processMemory->getMsgQCount(c->procID) == 0 && ++idleRounds > 20)
                break;
        }
    }
    return 0;
}

static THREAD_RET THREAD_FUNCTION_CALL ConcResizer(THREAD_ARG arg) {
    ConcCtx* c = (ConcCtx*)arg;
    ConcRand rng(c->seed ^ 0x9E5121ULL);
    while (!c->stop && !c->writerDone) {
        // Writing a very large message is the only public way to force the resize path.
        // It is registered with the oracle like any other message and reconciled the same
        // way: a resize driver that gets LOST is exactly as interesting as a normal
        // message that gets lost, so excluding it would discard evidence. Serials are
        // allocated from a high, non-overlapping range so the writer thread and this
        // thread can never collide on one.
        uint32 sum = 0;
        uint32 serial = 0xF0000000u + (c->readsSeen & 0xFFFFF);
        DataMessage* big = ConcMakeMsg(serial, 4000 + rng.below(4000), sum);
        c->oracle->noteWritten(serial, sum);
        if (!c->manager->processMemory->addToMsgQ(c->procID, big)) {
            c->oracle->mutex.enter();
            c->oracle->expectSum.erase(serial);
            c->oracle->written--;
            c->oracle->mutex.leave();
        }
        delete big;
        utils::Sleep(2);
    }
    return 0;
}
 
// ---------------------------------------------------------------------------------
// Tier drivers
// ---------------------------------------------------------------------------------
 
static bool ConcCrossProcessChild();        // tier B child half, defined below

static bool ConcRunThreaded(const char* label, uint32 readers, uint32 msgs,
        uint32 payload, uint32 burst, uint64 seed, bool resizeStress,
        bool blockingReads, bool dupSelftest, bool lostSelftest) {
    MemoryManager* manager = new MemoryManager();
    if (!manager->create(0)) {
        unittest::fail("%s: MemoryManager create(0) failed", label);
        delete manager;
        return false;
    }
 
    ConcOracle oracle;
    ConcCtx ctx;
    memset(&ctx, 0, sizeof(ctx));
    ctx.manager = manager;
    ctx.oracle = &oracle;
    ctx.procID = 0;                 // process 0's MSGQ, same queue Phase 1 uses
    ctx.totalMsgs = msgs;
    ctx.payloadFixed = payload;
    ctx.burst = burst;
    ctx.seed = seed;
    ctx.resizeStress = resizeStress;
    ctx.blockingReads = blockingReads;
    ctx.writerDone = false;
    ctx.stop = false;
    ctx.readsSeen = 0;
    // Positive control: duplicate one specific read once the run is under way.
    ctx.dupInjectAt = dupSelftest ? (msgs / 2) : 0;
 
    std::vector<uint32> tids;
    uint32 id = 0;
    for (uint32 i = 0; i < readers; i++) {
        if (!ThreadManager::CreateThread(ConcReader, &ctx, id)) {
            unittest::fail("%s: could not create reader thread %u", label, i);
            ctx.stop = true;
            delete manager;
            return false;
        }
        tids.push_back(id);
    }
    if (resizeStress) {
        if (ThreadManager::CreateThread(ConcResizer, &ctx, id))
            tids.push_back(id);
    }
    uint32 wid = 0;
    if (!ThreadManager::CreateThread(ConcWriter, &ctx, wid)) {
        unittest::fail("%s: could not create writer thread", label);
        ctx.stop = true;
        delete manager;
        return false;
    }
    tids.push_back(wid);
 
    // Bound the run: readers exit on their own once the writer is done and the queue is
    // drained, but a hang must not wedge the suite.
    uint64 start = GetTimeNow();
    uint32 budgetMs = 60000 * UT_TIME_SCALE;
    while (GetTimeAgeMS(start) < (int64)budgetMs) {
        bool anyAlive = false;
        for (size_t i = 0; i < tids.size(); i++)
            if (ThreadManager::IsThreadRunning(tids[i])) { anyAlive = true; break; }
        if (!anyAlive) break;
        utils::Sleep(20);
    }
    ctx.stop = true;
    for (uint32 g = 0; g < 200; g++) {
        bool anyAlive = false;
        for (size_t i = 0; i < tids.size(); i++)
            if (ThreadManager::IsThreadRunning(tids[i])) { anyAlive = true; break; }
        if (!anyAlive) break;
        utils::Sleep(25);
    }
 
    char det[512];
    const char* cls = oracle.reconcile(det, sizeof(det), lostSelftest);
    // NOTE: an earlier revision also asserted a read-pointer CONTINUITY check recorded
    // inside waitForQ (BakeLiveBug.md §7). That instrumentation sat on the queue hot path
    // under the process mutex and was removed when the investigation was filed - it found
    // no violation on any run, including genuine bakelive failures. The set-equality
    // oracle below is the durable check: a re-read or a skipped entry shows up as
    // DUPLICATED or LOST regardless of how the pointer got there.
    if (cls) {
        unittest::fail("%s: %s :: %s", label, cls, det);
        delete manager;
        return false;
    }
    unittest::detail("%s: OK :: %s", label, det);
    delete manager;
    return true;
}

static bool ConcRunCrossProcess(uint32 msgs, uint32 payload, uint64 seed) {
    // The parent CREATES the segment and writes; a re-exec of this same binary OPENs it by
    // name and reads. That is the only way to get the REAL contended cross-process mutex
    // and two independent `header` mappings - the dimension no other tier reaches.
    //
    // Reconciliation crosses the boundary through a file: the child appends every serial
    // it read, and the parent reconciles that list against what it wrote. Coarser than the
    // in-process oracle (no per-read timing), but it answers the one question that matters:
    // did anything go missing or double across the process boundary?
    // ⚠️ NOT a hardcoded "/tmp/...": there is no /tmp on Windows, so the child cannot
    // create its read-record file and the tier fails with "child produced NO read
    // records" - i.e. it looks like a cross-process DEFECT when only the harness path
    // was wrong. (The tier is right to refuse: unmeasured is not clean.) %TEMP% on
    // Windows, /tmp elsewhere - same rule as PsyTestTempPath / sslTestTempPath.
#if defined(WINDOWS)
    const char* tmpEnv = getenv("TEMP");
    if (!tmpEnv || !*tmpEnv) tmpEnv = getenv("TMP");
    std::string childList = std::string((tmpEnv && *tmpEnv) ? tmpEnv : ".")
        + "\\cmsdk_concfuzz_child_reads.txt";
#else
    std::string childList = "/tmp/cmsdk_concfuzz_child_reads.txt";
#endif
    utils::DeleteAFile(childList.c_str(), true);
 
    MemoryManager* manager = new MemoryManager();
    if (!manager->create(0)) {
        unittest::fail("TIER B: MemoryManager create(0) failed");
        delete manager;
        return false;
    }
    // Both sides address process 0's MSGQ - the same queue the in-process tiers use - and
    // the child attaches with connect(0, false), so no port needs passing: system id 0 is
    // what names the shared segments.
    // ⚠️ utils::GetCommandLine() and friends return EMPTY here: they are populated by
    // SetCommandLine(argc, argv), which CMSDKTest's main() never calls. Using them made
    // this tier fail with "cannot determine own executable path" on every run - correctly,
    // because the guard refuses to proceed rather than silently skip. Resolve the running
    // executable from the OS instead, with an env override for awkward setups.
    std::string exe;
    const char* exeEnv = getenv("PSY_CONC_EXE");
    if (exeEnv && *exeEnv)
        exe = exeEnv;
    else {
#if defined(WINDOWS)
        char buf[4096];
        DWORD n = GetModuleFileNameA(NULL, buf, sizeof(buf));
        if (n > 0 && n < sizeof(buf)) { buf[n] = 0; exe = buf; }
#elif defined(__APPLE__)
        char buf[4096];
        uint32_t sz = sizeof(buf);
        if (_NSGetExecutablePath(buf, &sz) == 0)
            exe = buf;
#else
        char buf[4096];
        ssize_t n = readlink("/proc/self/exe", buf, sizeof(buf) - 1);
        if (n > 0) { buf[n] = 0; exe = buf; }
#endif
    }
    if (exe.empty()) {
        unittest::fail("TIER B: cannot determine own executable path for re-exec "
            "(set PSY_CONC_EXE=<path to CMSDKTest> to override)");
        delete manager;
        return false;
    }
    std::string cmd = utils::StringFormat("%s test=processmemory_concfuzz", exe.c_str());
 
    // Child mode is selected by env so no new command-line parsing is needed.
    // PSY_CONC_CHILD=read tells the child to open() the segment and drain the queue.
#if defined(WINDOWS)
    _putenv_s("PSY_CONC_CHILD", "read");
    _putenv_s("PSY_CONC_CHILD_OUT", childList.c_str());
    _putenv_s("PSY_CONC_CHILD_MSGS", utils::StringFormat("%u", msgs).c_str());
#else
    setenv("PSY_CONC_CHILD", "read", 1);
    setenv("PSY_CONC_CHILD_OUT", childList.c_str(), 1);
    setenv("PSY_CONC_CHILD_MSGS", utils::StringFormat("%u", msgs).c_str(), 1);
#endif
    uint32 childPid = utils::NewProcess(cmd.c_str());
    // ⚠️ Do NOT unset PSY_CONC_CHILD here. NewProcess forks and the child inherits the
    // environment at EXEC time, not at fork time, so clearing it immediately is a race:
    // a child that exec's after the unset runs the PARENT path and recurses. Leave it set
    // for the lifetime of this function; the parent already took its own branch above.
    if (!childPid) {
        unittest::fail("TIER B: could not spawn reader child (%s)", cmd.c_str());
        delete manager;
        return false;
    }
 
    // Give the child time to attach before writing, otherwise the whole burst is written
    // and drained by nobody and the test measures nothing.
    utils::Sleep(1500);
 
    ConcRand rng(seed ^ 0xB10CULL);
    std::map<uint32, uint32> written;   // serial -> checksum
    for (uint32 serial = 1; serial <= msgs; serial++) {
        uint32 len = payload ? payload : (1 + rng.below(300));
        uint32 sum = 0;
        DataMessage* m = ConcMakeMsg(serial, len, sum);
        if (manager->processMemory->addToMsgQ(0, m))
            written[serial] = sum;
        delete m;
        if ((serial % 64) == 0)
            utils::Sleep(1);            // let the child drain, keeps the ring cycling
    }
 
    // Wait for the child to finish or time out.
    uint64 start = GetTimeNow();
    int rc = 0;
    bool exited = false;
    while (GetTimeAgeMS(start) < (int64)(30000 * UT_TIME_SCALE)) {
        if (utils::GetProcessStatus(childPid, rc) == PROC_TERMINATED) { exited = true; break; }
        utils::Sleep(50);
    }
    if (!exited) {
        utils::EndProcess(childPid);
        unittest::detail("TIER B: child did not exit in time; terminated (partial result)");
    }
 
    // Reconcile: every serial written exactly once in the child's list.
    std::string data = utils::ReadAFileString(childList.c_str());
    std::vector<std::string> lines = utils::TextListSplitLines(data.c_str(), false, true);
    std::map<uint32, uint32> readCount;
    uint32 badSum = 0;
    for (size_t i = 0; i < lines.size(); i++) {
        if (lines[i].find("R|") != 0) continue;
        std::vector<std::string> f = utils::TextListSplit(lines[i].c_str(), "|", true, true);
        if (f.size() < 3) continue;
        uint32 s = (uint32)strtoul(f[1].c_str(), NULL, 10);
        uint32 c = (uint32)strtoul(f[2].c_str(), NULL, 10);
        readCount[s]++;
        std::map<uint32, uint32>::const_iterator w = written.find(s);
        if (w != written.end() && w->second != c) badSum++;
    }
    uint32 lost = 0, dup = 0, extra = 0, firstLost = 0;
    for (std::map<uint32, uint32>::const_iterator it = written.begin(); it != written.end(); ++it) {
        std::map<uint32, uint32>::const_iterator r = readCount.find(it->first);
        uint32 n = (r == readCount.end()) ? 0 : r->second;
        if (n == 0)     { lost++; if (!firstLost) firstLost = it->first; }
        else if (n > 1) dup++;
    }
    for (std::map<uint32, uint32>::const_iterator it = readCount.begin(); it != readCount.end(); ++it)
        if (written.find(it->first) == written.end()) extra++;
 
    delete manager;
 
    if (!lines.size()) {
        // An empty list means the child never reported - that is UNMEASURED, not clean.
        unittest::fail("TIER B: child produced NO read records (%s) - the cross-process "
            "path is UNMEASURED, not clean. child exited=%d rc=%d",
            childList.c_str(), (int)exited, (int)rc);
        return false;
    }
    if (lost || dup || extra || badSum) {
        unittest::fail("TIER B (cross-process): written=%u child-read-distinct=%u LOST=%u "
            "DUPLICATED=%u never-written=%u badChecksum=%u (first lost=%u)",
            (uint32)written.size(), (uint32)readCount.size(), lost, dup, extra, badSum,
            firstLost);
        return false;
    }
    unittest::detail("TIER B (cross-process): OK :: written=%u child-read-distinct=%u "
        "LOST=0 DUPLICATED=0 badChecksum=0 (real cross-process mutex + two header mappings)",
        (uint32)written.size(), (uint32)readCount.size());
    return true;
}

static bool ConcCrossProcessChild() {
    const char* outPath = getenv("PSY_CONC_CHILD_OUT");
    uint32 want = ConcEnvU32("PSY_CONC_CHILD_MSGS", 20000);
    if (!outPath || !*outPath)
        return false;
 
    MemoryManager* manager = new MemoryManager();
    // open() attaches to the segment the PARENT created - this is the whole point of the
    // tier: a second independent mapping and the real named cross-process mutex.
    if (!manager->connect(0, false)) {
        delete manager;
        return false;
    }
    uint32 got = 0, idle = 0;
    std::string batch;
    while (got < want && idle < 600) {
        DataMessage* m = manager->processMemory->waitForMsgQ(0, 50);
        if (!m) { idle++; continue; }
        idle = 0;
        bool ok = false;
        int64 ser = m->getInt("cfserial", ok);
        std::string body = m->getAsString("cfbody");
        uint32 sum = ConcChecksum(body.c_str(), (uint32)body.size());
        if (ok)
            batch += utils::StringFormat("R|%llu|%u\n", (unsigned long long)ser, sum);
        delete m;
        got++;
        // Batch the appends: one write per message would make the child the bottleneck
        // and could itself change the timing under test.
        if (batch.size() > 8192) {
            utils::AppendToAFile(outPath, batch.c_str(), (uint32)batch.size());
            batch.clear();
        }
    }
    if (!batch.empty())
        utils::AppendToAFile(outPath, batch.c_str(), (uint32)batch.size());
    delete manager;
    return true;
}
 
bool ProcessMemory::ConcFuzzTest() {
    // Tier B child half: this same binary re-exec'd with PSY_CONC_CHILD=read. Must be the
    // FIRST thing checked - the child must not run the tier drivers and recurse.
    const char* childMode = getenv("PSY_CONC_CHILD");
    if (childMode && *childMode) {
        bool ok = ConcCrossProcessChild();
        // The child's verdict reaches the parent through the read-record file, not through
        // this return value, so a false here only marks the child's own failure to attach.
        return ok;
    }
 
    const char* tierEnv = getenv("PSY_CONC_TIER");
    std::string tier = (tierEnv && *tierEnv) ? tierEnv : "ALL";
    uint32 readers = ConcEnvU32("PSY_CONC_READERS", 4);
    uint32 msgs    = ConcEnvU32("PSY_CONC_MSGS", 20000);
    uint32 burst   = ConcEnvU32("PSY_CONC_BURST", 64);
    uint64 seed    = (uint64)ConcEnvU32("PSY_CONC_SEED", 0x2000);
    bool   dupSt   = ConcEnvFlag("PSY_CONC_DUPSELFTEST");
    bool   lostSt  = ConcEnvFlag("PSY_CONC_LOSTSELFTEST");
    // Default payload ~2280 bytes: the real bake wire size is 2314 and message size is
    // what determines how many entries fit before a wrap/resize, so matching it matters
    // more than randomising it.
    uint32 payload = ConcEnvU32("PSY_CONC_PAYLOAD", 2280);
 
    bool all = (tier == "ALL");
    bool ok = true;
 
    if (ok && (all || tier == "A")) {
        unittest::progress(10, "tier A: multi-reader contention on one queue");
        ok = ConcRunThreaded("TIER A (multi-reader)", readers, msgs, payload,
            0 /*steady*/, seed, false, false, dupSt, lostSt);
    }
    if (ok && (all || tier == "A2")) {
        unittest::progress(30, "tier A': burst-drain replay of the real shape");
        ok = ConcRunThreaded("TIER A' (burst-drain)", readers, msgs, payload,
            burst, seed, false, false, dupSt, lostSt);
    }
    if (ok && (all || tier == "B")) {
        unittest::progress(50, "tier B: cross-process");
        ok = ConcRunCrossProcess(msgs, payload, seed);
    }
    if (ok && (all || tier == "C")) {
        unittest::progress(65, "tier C: resize under concurrent readers");
        ok = ConcRunThreaded("TIER C (resize-under-read)", readers, msgs, payload,
            0, seed, true /*resizeStress*/, false, dupSt, lostSt);
    }
    if (ok && (all || tier == "D")) {
        unittest::progress(85, "tier D: blocking waitForQ path");
        ok = ConcRunThreaded("TIER D (blocking reads)", readers, msgs, payload,
            burst, seed, false, true /*blockingReads*/, dupSt, lostSt);
    }
 
    unittest::progress(100, "done");
    return ok;
}
 
} // namespace cmlabs