ProcessMemoryFuzz.cpp

Go to the documentation of this file.

 
#include "ProcessMemory.h"
#include "MemoryManager.h"
#include "UnitTestFramework.h"
#include <deque>
#include <stdlib.h>
#include <string.h>
 
namespace cmlabs {

struct RingRand {
    uint64 s;
    RingRand(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; }
};

struct RingRef {
    uint32 serial;
    uint32 payloadLen;
    uint32 sum;
    uint32 wireSize;
};

static uint32 RingChecksum(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 RingFillPayload(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 uint32 RingEnvU32(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 const char* RingClass(const RingRef& want, DataMessage* got, uint32& gotSerial,
                             uint32& gotLen, uint32& gotSum) {
    if (!got) return "LOST";           // model had an entry, queue gave nothing
    uint32 sz = got->getSize();
    if (sz != want.wireSize) return "SHORT";
    bool ok = false;
    int64 s = got->getInt("s", ok); if (!ok) return "CORRUPT";
    gotSerial = (uint32)s;
    uint32 plen = 0;
    const char* body = got->getData("p", plen);
    gotLen = plen;
    gotSum = body ? RingChecksum(body, plen) : 0;
    if (gotSerial != want.serial) return "REORDERED";
    if (gotLen != want.payloadLen) return "SHORT";
    if (gotSum != want.sum) return "CORRUPT";
    return NULL;                        // match
}

static DataMessage* RingMakeMsg(uint32 serial, uint32 len, RingRef& ref) {
    DataMessage* m = new DataMessage(CTRL_TEST, 0);
    m->setInt("s", serial);
    char* buf = new char[len + 1];
    RingFillPayload(buf, len, serial);
    m->setData("p", buf, len);
    delete [] buf;
    ref.serial     = serial;
    ref.payloadLen = len;
    ref.sum        = RingChecksum(NULL, 0);   // placeholder, recomputed below
    // recompute sum from the same bytes we just wrote
    {
        char* tmp = new char[len + 1];
        RingFillPayload(tmp, len, serial);
        ref.sum = RingChecksum(tmp, len);
        delete [] tmp;
    }
    ref.wireSize = m->getSize();
    return m;
}

static const char* RingCheckInvariants(MessageQueueHeader* q, uint32 expectCount,
                                        char* detail, uint32 detailSize) {
    if (!q) { return expectCount == 0 ? NULL : "HEADER_NULL"; }
 
    uint64 cap  = q->size - sizeof(MessageQueueHeader);   // circular-buffer capacity
    char*  data = ((char*)q) + sizeof(MessageQueueHeader);
 
    // 1) count must match what the model expects (the primary handover symptom).
    if (q->count != expectCount) {
        snprintf(detail, detailSize,
            "count=%u expect=%u startPos=%llu endPos=%llu padding=%llu cap=%llu",
            q->count, expectCount, q->startPos, q->endPos, q->padding, cap);
        return "COUNT_MISMATCH";
    }
 
    // 2) positions must be inside the buffer.
    if (q->startPos > cap || q->endPos > cap || q->padding > cap) {
        snprintf(detail, detailSize,
            "startPos=%llu endPos=%llu padding=%llu cap=%llu",
            q->startPos, q->endPos, q->padding, cap);
        return "POS_OUT_OF_RANGE";
    }
 
    // 3) empty ambiguity: count==0 but startPos!=endPos means the ring thinks it is empty
    //    while its pointers still span live bytes (exactly the "reader stops early" shape).
    if (q->count == 0 && q->startPos != q->endPos) {
        snprintf(detail, detailSize, "count=0 but startPos=%llu endPos=%llu padding=%llu",
            q->startPos, q->endPos, q->padding);
        return "EMPTY_BUT_SPANNED";
    }
 
    // 4) walk exactly count messages from startPos, honouring the same wrap rule the reader
    //    uses (rewind when we reach cap - padding), and prove each is a well-formed DataMessage
    //    of non-zero size that fits before the wrap point.
    //    NB the reader rewinds *before* consuming the next message, not after the last one:
    //    a queue filled exactly to the end legally has endPos == cap (addToQ branch 2 does
    //    endPos += msgSize without wrapping), so rewinding after the final message would
    //    wrongly collapse cap -> 0 and flag a phantom mismatch. Rewind only when we still have
    //    another message to place.
    uint64 pos = q->startPos;
    for (uint32 i = 0; i < q->count; i++) {
        // rewind before reading, matching the reader's own timing
        if (pos >= cap - q->padding) pos = 0;
        if (GetObjID(data + pos) != DATAMESSAGEID) {
            snprintf(detail, detailSize,
                "msg %u/%u at pos=%llu not a DataMessage (startPos=%llu endPos=%llu padding=%llu)",
                i, q->count, pos, q->startPos, q->endPos, q->padding);
            return "WALK_BAD_OBJID";
        }
        uint32 msz = ((DataMessageHeader*)(data + pos))->size;
        if (msz == 0 || pos + msz > cap) {
            snprintf(detail, detailSize,
                "msg %u/%u pos=%llu size=%u overruns cap=%llu (padding=%llu)",
                i, q->count, pos, msz, cap, q->padding);
            return "WALK_SIZE_OVERRUN";
        }
        pos += msz;
    }
 
    // 5) after walking count messages we must land exactly on endPos (single-process: no
    //    concurrent writer moved it). A mismatch is the count-vs-contents inconsistency.
    //    endPos == cap is legal (full-to-the-end, not yet wrapped); so is endPos in [0,cap].
    if (pos != q->endPos) {
        snprintf(detail, detailSize,
            "walk ended at pos=%llu but endPos=%llu (count=%u startPos=%llu padding=%llu cap=%llu)",
            pos, q->endPos, q->count, q->startPos, q->padding, cap);
        return "WALK_ENDPOS_MISMATCH";
    }
    return NULL;
}

static bool RingRunSeed(MemoryManager* manager, uint64 seed, uint32 iters, bool selftest,
                        bool checkInvariants, bool invariantSelftest) {
    RingRand rng(seed);
    std::deque<RingRef> model;
    uint32 nextSerial = 1;
    uint32 injectAt = selftest ? (iters / 2) : 0xFFFFFFFFu;
    bool   injected = false;   // positive control: fire exactly once, on a read
    char   idet[256];
 
    // Invariant positive control: once, after a write, deliberately corrupt the live
    // header's count so RingCheckInvariants MUST catch it (proves the checker can fail).
    uint32 invInjectAt = invariantSelftest ? (iters / 2) : 0xFFFFFFFFu;
    bool   invInjected = false;
 
    // Resize-heavy mode (PSY_RING_RESIZEHEAVY=1): bias hard toward large writes and deep
    // queues so addToQ keeps hitting the resize path and waitForQ keeps reading across a
    // buffer that just doubled with stale padding -- the exact arithmetic the handover flags
    // as never cleanly exonerated. Off by default so the validated default mix is unchanged.
    bool resizeHeavy = getenv("PSY_RING_RESIZEHEAVY") && *getenv("PSY_RING_RESIZEHEAVY") == '1';
 
    for (uint32 op = 0; op < iters; op++) {
        // Bias toward writes when near-empty and reads when large, to sweep both wrap
        // directions and force resize, drain-to-empty and exact-fit states.
        uint32 depth = (uint32)model.size();
        bool doWrite;
        // Force a read once we reach injectAt with something queued, so the positive
        // control is guaranteed to exercise the oracle rather than depending on the RNG.
        bool forceRead = (op >= injectAt) && !injected && depth > 0;
        if (forceRead)             doWrite = false;
        else if (depth == 0)       doWrite = true;
        else if (resizeHeavy)      doWrite = (depth > 400) ? rng.chance(45) : rng.chance(80);
        else if (depth > 200)      doWrite = rng.chance(20);
        else                       doWrite = rng.chance(60);
 
        if (doWrite) {
            // occasional exact-fit-ish lengths plus a broad spread up to ~1.5KB
            uint32 len;
            uint32 pick = rng.below(100);
            if (resizeHeavy) {
                // mostly large payloads to drive repeated buffer doublings, plus some tiny
                // ones so wraps leave odd startPos/padding remainders behind
                if (pick < 20)      len = 1 + rng.below(16);
                else                len = 700 + rng.below(1400);
            }
            else if (pick < 15) len = 1 + rng.below(8);       // tiny
            else if (pick < 25) len = 900 + rng.below(700);   // large: forces resize/wrap
            else                len = 1 + rng.below(300);     // medium
            RingRef ref;
            DataMessage* m = RingMakeMsg(nextSerial++, len, ref);
            if (!manager->processMemory->addToMsgQ(0, m)) {
                unittest::fail("seed %llu op %u: addToMsgQ failed (serial %u len %u)",
                    seed, op, ref.serial, len);
                delete m;
                return false;
            }
            delete m;
            model.push_back(ref);
 
            if (checkInvariants) {
                MessageQueueHeader* q = manager->processMemory->fuzzPeekQHeader(0, MSGQ_TYPE);
                if ((op >= invInjectAt) && !invInjected && q) { q->count += 7; invInjected = true; }
                const char* iv = RingCheckInvariants(q, (uint32)model.size(), idet, sizeof(idet));
                if (iv) {
                    unittest::fail("seed %llu op %u (post-write serial %u): INVARIANT %s :: %s",
                        seed, op, ref.serial, iv, idet);
                    return false;
                }
            }
        }
        else {
            RingRef want = model.front();
            model.pop_front();
            DataMessage* got = manager->processMemory->waitForMsgQ(0, 200);
 
            // positive control: corrupt the expectation once so a correct queue "diverges",
            // proving the oracle can actually fail (not just always pass).
            if ((op >= injectAt) && !injected) { want.serial ^= 0xABCD; injected = true; }
 
            uint32 gs = 0, gl = 0, gsum = 0;
            const char* cls = RingClass(want, got, gs, gl, gsum);
            if (cls) {
                unittest::fail("seed %llu op %u: %s  want{serial=%u len=%u sum=%08x size=%u} "
                    "got{serial=%u len=%u sum=%08x size=%u}",
                    seed, op, cls, want.serial, want.payloadLen, want.sum, want.wireSize,
                    gs, gl, gsum, got ? got->getSize() : 0);
                if (got) delete got;
                return false;
            }
            delete got;
 
            if (checkInvariants) {
                MessageQueueHeader* q = manager->processMemory->fuzzPeekQHeader(0, MSGQ_TYPE);
                const char* iv = RingCheckInvariants(q, (uint32)model.size(), idet, sizeof(idet));
                if (iv) {
                    unittest::fail("seed %llu op %u (post-read): INVARIANT %s :: %s",
                        seed, op, iv, idet);
                    return false;
                }
            }
        }
    }
 
    while (!model.empty()) {
        RingRef want = model.front();
        model.pop_front();
        DataMessage* got = manager->processMemory->waitForMsgQ(0, 200);
        uint32 gs = 0, gl = 0, gsum = 0;
        const char* cls = RingClass(want, got, gs, gl, gsum);
        if (cls) {
            unittest::fail("seed %llu drain: %s serial want=%u got=%u", seed, cls, want.serial, gs);
            if (got) delete got;
            return false;
        }
        delete got;
    }
    return true;
}
 
bool ProcessMemory::FuzzTest() {
    uint32 iters    = RingEnvU32("PSY_RING_ITERS", 20000);
    uint32 seeds    = RingEnvU32("PSY_RING_SEEDS", 8);
    bool   selftest = getenv("PSY_RING_SELFTEST") && *getenv("PSY_RING_SELFTEST") == '1';
    bool   checkInvariants = getenv("PSY_RING_INVARIANTS") && *getenv("PSY_RING_INVARIANTS") == '1';
    bool   invariantSelftest = getenv("PSY_RING_INVSELFTEST") && *getenv("PSY_RING_INVSELFTEST") == '1';
    if (invariantSelftest) checkInvariants = true;   // inv-selftest implies the checker is on
    const char* sEnv = getenv("PSY_RING_SEED");
 
    // Each seed runs on its OWN freshly created+destroyed MemoryManager segment. A single
    // shared manager leaks ring geometry between seeds, which (a) makes a failure depend on
    // the seeds that ran before it, so PSY_RING_SEED replay does NOT reproduce it, and
    // (b) is exactly the kind of harness artifact the ring-defect handover warns about.
    bool ok = true;
    if (sEnv && *sEnv) {
        uint64 seed = (uint64)strtoull(sEnv, NULL, 0);
        unittest::progressf(10, "replay seed %llu (%u iters)", seed, iters);
        MemoryManager* m = new MemoryManager();
        if (!m->create(0)) { unittest::fail("MemoryManager create(0) failed (replay seed %llu)", seed); delete m; return false; }
        ok = RingRunSeed(m, seed, iters, selftest, checkInvariants, invariantSelftest);
        delete m;
    }
    else {
        for (uint32 i = 0; i < seeds && ok; i++) {
            uint64 seed = 0x1000ULL + i * 0x9E3779B1ULL;
            unittest::progressf(10 + (int)(80.0 * i / seeds),
                "seed %llu (%u/%u, %u iters)", seed, i + 1, seeds, iters);
            MemoryManager* m = new MemoryManager();
            if (!m->create(0)) { unittest::fail("MemoryManager create(0) failed (seed %llu)", seed); delete m; ok = false; break; }
            ok = RingRunSeed(m, seed, iters, selftest, checkInvariants, invariantSelftest);
            delete m;
        }
    }
 
    unittest::progress(100, "done");
    return ok;
}
 
} // namespace cmlabs