db.c

#include <stdlib.h>
#include <stdint.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stdio.h>
#include <assert.h>
#include <string.h>

#include <pthread.h>
#include <signal.h>
#include <stdatomic.h>
#include <sys/syscall.h>
#include <unistd.h>

#if __has_include ("tracy/TracyC.h")
#include "tracy/TracyC.h"
#endif

#include "common.h"
#include "trie.h"
#include "epoch.h"
#include "sysmon.h"
#include "db.h"

#include "vendor/stb_ds.h"

typedef struct ListOfEdgeTo {
    size_t capacityEdges;
    size_t nEdges; // This is an upper bound.
    uint64_t edges[];
} ListOfEdgeTo;
#define SIZEOF_LIST_OF_EDGE_TO(CAPACITY_EDGES) (sizeof(ListOfEdgeTo) + (CAPACITY_EDGES)*sizeof(uint64_t))
typedef bool (*ListEdgeCheckerFn)(void* arg, uint64_t edge);


typedef struct GenRc {
    // How many acquired raw pointers to this object exist? The object
    // cannot be freed/invalidated as long as rc > 0. You must
    // increment rc before accessing any other field in the object.
    int16_t rc;

    int gen: 15;

    // The object also cannot be freed/invalidated as long as alive is
    // true; alive indicates that the object is alive in the database
    // (has supporting parent).
    bool alive: 1;
} GenRc;

bool genRcAcquire(_Atomic GenRc* genRcPtr, int32_t gen) {
    GenRc oldGenRc;
    GenRc newGenRc;
    do {
        oldGenRc = *genRcPtr;
        if (oldGenRc.gen < 0 || oldGenRc.gen != gen) {
            return false;
        }

        newGenRc = oldGenRc;
        newGenRc.rc++;

    } while (!atomic_compare_exchange_weak(genRcPtr, &oldGenRc, newGenRc));

    return true;
}
bool genRcRelease(_Atomic GenRc* genRcPtr) {
    GenRc oldGenRc;
    GenRc newGenRc;
    bool callerIsLastReleaser = false;
    do {
        oldGenRc = *genRcPtr;
        newGenRc = oldGenRc;

        --newGenRc.rc;
        callerIsLastReleaser = !oldGenRc.alive && (newGenRc.rc == 0);
        if (callerIsLastReleaser) {
            newGenRc.gen++;
        }

    } while (!atomic_compare_exchange_weak(genRcPtr, &oldGenRc, newGenRc));
    return callerIsLastReleaser;
}
void genRcMarkAsDead(_Atomic GenRc* genRcPtr) {
    // ASSUMES that rc > 0 (because the caller must have acquired the
    // rc before calling us), which means gen cannot change. We do the
    // loop here in case the rc changes.
    GenRc oldGenRc;
    GenRc newGenRc;
    do {
        oldGenRc = *genRcPtr;
        newGenRc = oldGenRc;

        newGenRc.alive = false;

        // TODO: Check that gen hasn't changed and that rc > 0.
    } while (!atomic_compare_exchange_weak(genRcPtr, &oldGenRc, newGenRc));
}

// Destructor datatype:

typedef struct Destructor {
    _Atomic int rc;

    void (*fn)(void*);
    void* arg;
} Destructor;

Destructor* destructorNew(void (*fn)(void*), void* arg) {
    Destructor* ret = malloc(sizeof(Destructor));
    ret->rc = 0;
    ret->fn = fn;
    ret->arg = arg;
    return ret;
}

void destructorRun(Destructor* d) {
    assert(d->fn != NULL);
    d->fn(d->arg);
    d->fn = NULL;
}

static void destructorRetain(Destructor* d) {
    d->rc++;
}
static void destructorRelease(Destructor* d) {
    if (--d->rc == 0) {
        destructorRun(d);
        free(d);
    }
}

typedef struct DestructorSet {
    Destructor** destructors;
    int destructorsCapacity;
    int destructorsCount;

    pthread_mutex_t destructorsMutex;
} DestructorSet;

void destructorSetInit(DestructorSet* set) {
    set->destructors = malloc(8 * sizeof(Destructor*));
    set->destructorsCapacity = 8;
    set->destructorsCount = 0;
}

static void destructorSetAddImpl(DestructorSet* set, Destructor* d) {
    assert(set->destructors != NULL);

    if (set->destructorsCount == set->destructorsCapacity) {
        set->destructorsCapacity *= 2;
        set->destructors = realloc(set->destructors, set->destructorsCapacity * sizeof(Destructor));
    }

    destructorRetain(d);
    set->destructors[set->destructorsCount++] = d;
}
void destructorSetAdd(DestructorSet* set, Destructor* d) {
    destructorSetAddImpl(set, d);
}

void destructorSetInherit(DestructorSet* to, DestructorSet* from) {
    for (int i = 0; i < from->destructorsCount; i++) {
        destructorSetAddImpl(to, from->destructors[i]);
    }
}

void destructorSetReleaseAll(DestructorSet* set) {
    for (int i = 0; i < set->destructorsCount; i++) {
        if (set->destructors[i] != NULL) {
            destructorRelease(set->destructors[i]);
        }
        set->destructors[i] = NULL;
    }
    free(set->destructors);
    set->destructors = NULL;
}

// Statement datatype:

typedef struct Statement {
    _Atomic GenRc genRc;

    // Immutable statement properties:
    // -----

    // Owned by the DB. clause cannot be mutated or invalidated while
    // rc > 0.
    Clause* _Atomic clause;

    // If the statement is removed, we wait keepMs milliseconds before
    // removing its child matches.
    _Atomic long keepMs;

    // Will be NULL if not running in an Atomically
    // convergence-tracking subgraph.
    AtomicallyVersion* atomicallyVersion;

    // Note that statement destructors are not mutable after statement
    // creation, so they can be safely looked up and inherited, unlike
    // match destructors.
    DestructorSet destructorSet;
    pthread_mutex_t destructorSetMutex;

    // Used for debugging (and stack traces for When bodies).
    char sourceFileName[100];
    int sourceLineNumber;

    // Mutable statement properties:
    // -----

    // How many living Matches (or Holds or Asserts) are supporting
    // this statement? When parentCount hits 0, we deindex the
    // statement.
    _Atomic int parentCount;

    // ListOfEdgeTo MatchRef. Used for removal.
    ListOfEdgeTo* childMatches;
    pthread_mutex_t childMatchesMutex;

    // TODO: Cache of Jim-local clause objects?

} Statement;

// Match datatype:

typedef struct Match {
    _Atomic GenRc genRc;

    // Immutable match properties:
    // -----
    int workerThreadIndex;

    // Mutable match properties:
    // -----

    // Will be NULL if not running in an Atomically
    // convergence-tracking subgraph.
    AtomicallyVersion* _Atomic atomicallyVersion;
    _Atomic bool isAtomicallyRootMatch;
    // Set to true if ANY parent statement was removed but we kept the
    // Match alive.
    _Atomic bool parentWasRemoved;

    // isCompleted is set to true once the Tcl evaluation that builds
    // the match is completed. As long as isCompleted is false,
    // workerThread is subject to termination signal (SIGUSR1) if the
    // match is removed.
    _Atomic bool isCompleted;

    DestructorSet destructorSet;
    pthread_mutex_t destructorSetMutex;

    // ListOfEdgeTo StatementRef. Used for removal.
    // NULL means the slot is fully destroyed and ready for reuse (matchNew
    // checks this). CHILD_STATEMENTS_REMOVING means matchRemoveSelf has
    // claimed removal but matchDestroy hasn't finished yet.
    ListOfEdgeTo* _Atomic childStatements;
    pthread_mutex_t childStatementsMutex;
} Match;

// Sentinel: matchRemoveSelf sets childStatements to this to prevent new
// children from being added. matchDestroy then sets it to NULL (release)
// as its final step, which is what matchNew waits for.
#define CHILD_STATEMENTS_REMOVING ((ListOfEdgeTo*)1)

// Database datatypes:

typedef struct Hold {
    // If key == NULL, then this Hold is an empty slot that can be
    // used for a new hold key.
    const char* key; // Owned by the DB.
    double version;

    StatementRef statement;
} Hold;

typedef struct AtomicallyVersion {
    int number;

    // Used to find older version to reap.
    struct Atomically* atomically;

    // When this is >0, there are still operations in flight within
    // this AtomicallyVersion arena. When this is 0, this
    // AtomicallyVersion is 'fully converged'. This should never be
    // negative.
    int _Atomic inflightCount;

    // When you do When -atomically, every time its body executes, it
    // produces a Match and a fresh AtomicallyVersion. That Match is
    // the rootMatch of the AtomicallyVersion. The rootMatch (and
    // therefore all descendant statements) is artificially kept alive
    // as long as the AtomicallyVersion hasn't been
    // invalidated. (matchRemoveSelf won't go through if
    // match->atomicallyVersion->rootMatch is match.)
    //
    // Invalidation: When a newer version converges (or when the
    // Atomically times out), we walk all older versions' rootMatches
    // and NULL them out.
    Match* _Atomic rootMatch;
} AtomicallyVersion;

typedef struct AtomicallyVersionList {
    AtomicallyVersion* version;
    struct AtomicallyVersionList* next;
} AtomicallyVersionList;

typedef struct Atomically {
    // If key == NULL, then this Atomically is an empty slot that can
    // be used for a new key.
    const char* _Atomic key; // Owned by the DB.

    int _Atomic nextNumber;

    // This is used by a newly converged version to reap older
    // versions.
    AtomicallyVersionList* _Atomic allVersions;

    AtomicallyVersion* _Atomic latestConvergedVersion;

    // Records the last time that a version of this Atomically
    // converged. The sysmon will check this every 50ms or so and
    // reap any Atomicallys that haven't converged in a while.
    int64_t latestConvergedTime;
    int64_t timeout;
} Atomically;

typedef struct Db {
    // Memory pool used to allocate statements.
    Statement statementPool[65536]; // slot 0 is reserved.
    _Atomic uint16_t statementPoolNextIdx;

    // Memory pool used to allocate matches.
    Match matchPool[65536]; // slot 0 is reserved.
    _Atomic uint16_t matchPoolNextIdx;

    // Primary trie (index) used for queries.
    const Trie* _Atomic clauseToStatementRef;

    // One for each Hold key, which always stores the highest-version
    // held statement for that key. We keep this map so that we can
    // overwrite out-of-date Holds for a key as soon as a newer one
    // comes in, without having to actually emit and react to the
    // statement.
    Hold* holds; // stb_ds string hash map (keyed by Hold.key)
    Mutex holdsMutex;

    // One for each `atomically` key.
    Atomically atomicallys[256];
    // This Mutex guards the list but not the individual Atomically
    // structs.
    Mutex atomicallysMutex;
} Db;

////////////////////////////////////////////////////////////
// EdgeTo and ListOfEdgeTo:
////////////////////////////////////////////////////////////

ListOfEdgeTo* listOfEdgeToNew(size_t capacityEdges) {
    ListOfEdgeTo* ret = calloc(SIZEOF_LIST_OF_EDGE_TO(capacityEdges), 1);
    ret->capacityEdges = capacityEdges;
    ret->nEdges = 0;
    return ret;
}
static void listOfEdgeToDefragment(ListEdgeCheckerFn checker, void* checkerArg,
                                   ListOfEdgeTo** listPtr);
// Takes a double pointer to list because it may move the list to grow
// it (requiring replacement of the original pointer). checker is used
// to discard any edges that have been invalidated if we defragment
// (to prevent unlimited growth of the edge list).
void listOfEdgeToAdd(ListEdgeCheckerFn checker, void* checkerArg,
                     ListOfEdgeTo** listPtr, uint64_t to) {
    if ((*listPtr)->nEdges == (*listPtr)->capacityEdges) {
        // We've run out of edge slots at the end of the
        // list. Try defragmenting the list.
        listOfEdgeToDefragment(checker, checkerArg, listPtr);
        if ((*listPtr)->nEdges == (*listPtr)->capacityEdges) {
            // Still no slots? Grow the statement to
            // accommodate.
            (*listPtr)->capacityEdges = (*listPtr)->capacityEdges * 2;
            *listPtr = realloc(*listPtr, SIZEOF_LIST_OF_EDGE_TO((*listPtr)->capacityEdges));
        }
    }

    assert((*listPtr)->nEdges < (*listPtr)->capacityEdges);
    // There's a free slot at the end of the edgelist in
    // the statement. Use it.
    (*listPtr)->edges[(*listPtr)->nEdges++] = to;
}
void listOfEdgeToRemove(ListOfEdgeTo* list, uint64_t to) {
    assert(list != NULL);
    for (size_t i = 0; i < list->nEdges; i++) {
        if (list->edges[i] == to) {
            list->edges[i] = to;
        }
    }
}

// Given listPtr, moves all non-EMPTY edges to the front, then updates
// nEdges accordingly. Discards any edges for which checker(edge) is
// false.
//
// Defragmentation is necessary to prevent continual growth of
// the statement edgelist if you keep adding and removing
// edges on the same statement.
static void listOfEdgeToDefragment(ListEdgeCheckerFn checker, void* checkerArg,
                                   ListOfEdgeTo** listPtr) {
    // Copy all non-EMPTY edges into a new edgelist.
    ListOfEdgeTo* list = calloc(SIZEOF_LIST_OF_EDGE_TO((*listPtr)->capacityEdges), 1);
    size_t nEdges = 0;
    for (size_t i = 0; i < (*listPtr)->nEdges; i++) {
        uint64_t edge = (*listPtr)->edges[i];
        // Also validate edge (is it a valid match / statement?)
        if (edge != 0 && checker(checkerArg, edge)) {
            list->edges[nEdges++] = edge;
        }
    }
    list->nEdges = nEdges;
    list->capacityEdges = (*listPtr)->capacityEdges;

    free(*listPtr);
    *listPtr = list;
}

////////////////////////////////////////////////////////////
// Statement:
////////////////////////////////////////////////////////////

Statement* statementAcquire(Db* db, StatementRef ref) {
    if (ref.idx == 0) { return NULL; }

    Statement* s = &db->statementPool[ref.idx];
    if (genRcAcquire(&s->genRc, ref.gen)) {
        return s;
    }
    return NULL;
}
Statement* statementUnsafeGet(Db* db, StatementRef ref) {
    if (ref.idx == 0) { return NULL; }
    return &db->statementPool[ref.idx];
}

static void statementDestroy(Statement* stmt);
void statementRelease(Db* db, Statement* stmt) {
    if (genRcRelease(&stmt->genRc)) {
        statementDestroy(stmt);
    }
}

bool statementCheck(Db* db, StatementRef ref) {
    Statement* s = &db->statementPool[ref.idx];
    GenRc genRc = s->genRc;
    return ref.gen >= 0 && ref.gen == genRc.gen;
}

StatementRef statementRef(Db* db, Statement* stmt) {
    GenRc genRc = stmt->genRc;
    return (StatementRef) {
        .gen = genRc.gen,
        .idx = stmt - &db->statementPool[0]
    };
}

// Creates a new statement. Internal helper for the DB, not callable
// from the outside (they need to insert into the DB as a complete
// operation). Note: clause ownership transfers to the DB, which then
// becomes responsible for freeing it. 
static StatementRef statementNew(Db* db, Clause* clause,
                                 long keepMs, AtomicallyVersion* atomicallyVersion,
                                 const char* sourceFileName,
                                 int sourceLineNumber) {
    StatementRef ret;
    Statement* stmt = NULL;

    // Look for a free statement slot to use:
    while (1) {
        int32_t idx = db->statementPoolNextIdx++;
        if (idx == 0) { continue; } // skip null statement
        stmt = &db->statementPool[idx];

        GenRc oldGenRc = stmt->genRc;
        if (oldGenRc.rc == 0 && !oldGenRc.alive && stmt->clause == NULL) {
            GenRc newGenRc = oldGenRc;
            newGenRc.alive = true;

            if (atomic_compare_exchange_weak(&stmt->genRc, &oldGenRc, newGenRc)) {
                ret = (StatementRef) { .gen = newGenRc.gen, .idx = idx };
                break;
            }
        }
    }

    // We should now have exclusive access to stmt, as its rc
    // is 0 and we were the ones who made it alive.

    atomic_store(&stmt->clause, clause);
    stmt->keepMs = keepMs;

    // inflightCount must start incremented so that this
    // atomicallyVersion never reports convergence (inflightCount = 0)
    // before the first reaction is dispatched.
    if (atomicallyVersion != NULL) {
        atomicallyVersion->inflightCount++;
        stmt->atomicallyVersion = atomicallyVersion;
    } else {
        stmt->atomicallyVersion = NULL;
    }
    stmt->parentCount = 1;

    destructorSetInit(&stmt->destructorSet);
    pthread_mutex_init(&stmt->destructorSetMutex, NULL);

    stmt->childMatches = listOfEdgeToNew(8);

    pthread_mutexattr_t mta;
    pthread_mutexattr_init(&mta);
    pthread_mutexattr_settype(&mta, PTHREAD_MUTEX_RECURSIVE);
    pthread_mutex_init(&stmt->childMatchesMutex, &mta);
    pthread_mutexattr_destroy(&mta);

    snprintf(stmt->sourceFileName, sizeof(stmt->sourceFileName),
             "%s", sourceFileName);
    stmt->sourceLineNumber = sourceLineNumber;

    return ret;
}

static void statementDestroy(Statement* stmt) {
    stmt->parentCount = 0;
    // They should have removed the children first.
    assert(stmt->childMatches == NULL);

    pthread_mutex_lock(&stmt->destructorSetMutex);
    destructorSetReleaseAll(&stmt->destructorSet);
    pthread_mutex_unlock(&stmt->destructorSetMutex);

    Clause* stmtClause = statementClause(stmt);
    // Marks this statement slot as being fully free and ready for
    // reuse.
    stmt->clause = NULL;

    /* TracyCFreeS(stmt, 4); */
    clauseFree(stmtClause);
}

Clause* statementClause(Statement* stmt) { return stmt->clause; }

AtomicallyVersion* statementAtomicallyVersion(Statement* stmt) {
    return stmt->atomicallyVersion;
}
int statementParentCount(Statement* stmt) {
    return stmt->parentCount;
}

char* statementSourceFileName(Statement* stmt) {
    return stmt->sourceFileName;
}
int statementSourceLineNumber(Statement* stmt) {
    return stmt->sourceLineNumber;
}

int statementIncompleteChildMatchesCount(Db* db, Statement* stmt) {
    int count = 0;

    pthread_mutex_lock(&stmt->childMatchesMutex);
    if (stmt->childMatches == NULL) { goto done; }
    for (size_t i = 0; i < stmt->childMatches->nEdges; i++) {
        MatchRef childRef = { .val = stmt->childMatches->edges[i] };
        Match* child = matchAcquire(db, childRef);
        if (child != NULL) {
            if (!child->isCompleted) { count++; }
            matchRelease(db, child);
        }
    }
 done:
    pthread_mutex_unlock(&stmt->childMatchesMutex);
    return count;
}

static bool matchChecker(void* db, uint64_t ref) {
    return matchCheck((Db*) db, (MatchRef) { .val = ref });
}
// You must call this with the childMatchesMutex held.
static void statementAddChildMatch(Db* db, Statement* stmt, MatchRef child) {
    listOfEdgeToAdd(&matchChecker, db,
                    &stmt->childMatches, child.val);
}

void statementAddDestructor(Statement* stmt, Destructor* d) {
    pthread_mutex_lock(&stmt->destructorSetMutex);
    destructorSetAdd(&stmt->destructorSet, d);
    pthread_mutex_unlock(&stmt->destructorSetMutex);
}
void statementInheritDestructors(Statement* stmt, Statement* fromStmt) {
    pthread_mutex_lock(&fromStmt->destructorSetMutex);
    pthread_mutex_lock(&stmt->destructorSetMutex);
    destructorSetInherit(&stmt->destructorSet,
                         &fromStmt->destructorSet);
    pthread_mutex_unlock(&stmt->destructorSetMutex);
    pthread_mutex_unlock(&fromStmt->destructorSetMutex);
}

// Fails to increment parentCount & returns false if parentCount is 0,
// meaning that the statement is in the process of being destroyed by
// someone else and you should back off.
bool statementTryIncrParentCount(Statement* stmt) {
    int oldParentCount;
    int newParentCount;
    do {
        oldParentCount = stmt->parentCount;
        if (oldParentCount == 0) {
            return false;
        }
        newParentCount = oldParentCount + 1;
    } while (!atomic_compare_exchange_weak(&stmt->parentCount, &oldParentCount,
                                           newParentCount));
    return true;
}

void statementDecrParentCountAndMaybeRemoveSelf(Db* db, Statement* stmt) {
    if (stmt->keepMs > 0) {
        if (--stmt->parentCount == 0) {
            // Note that we should have exclusive access to stmt at
            // this point.

            // Prevent future removers now that we've already
            // scheduled removal.
            long keepMs = stmt->keepMs;
            stmt->keepMs = -keepMs;

            // Tentatively trigger a removal in `keepMs` ms, but the
            // statement is still able to be revived in the
            // intervening time.
            sysmonScheduleRemoveAfter(statementRef(db, stmt), keepMs);

            stmt->parentCount++;
        }

    } else if (stmt->keepMs < 0) {
        // We're carrying out a previously-scheduled removal.
        if (--stmt->parentCount == 0) {
            // Note that we should have exclusive access to stmt at
            // this point.
            statementRemoveSelf(db, stmt, true);

        } else {
            // The statement's been revived; restore keepMs.

            // TODO: there's a race here if parentCount gets zeroed
            // without ever getting scheduled for removal.
            stmt->keepMs = -stmt->keepMs;
        }

    } else if (stmt->keepMs == 0) {
        if (--stmt->parentCount == 0) {
            // Note that we should have exclusive access to stmt at
            // this point.
            statementRemoveSelf(db, stmt, true);
        }
    }
}

// Call statementRemoveSelf when ALL of the statement's parents
// (matches or other) are removed (parentCount has hit 0).
void statementRemoveSelf(Db* db, Statement* stmt, bool doDeindex) {
    assert(stmt->parentCount == 0);

    if (doDeindex) {
        uint64_t results[100]; int resultsCount;
        epochBegin();

        const Trie* oldClauseToStatementRef;
        const Trie* newClauseToStatementRef;
        do {
            epochReset();
            oldClauseToStatementRef = db->clauseToStatementRef;
            newClauseToStatementRef =
                trieRemove(db->clauseToStatementRef,
                           epochAlloc, epochFree,
                           stmt->clause,
                           (uint64_t*) results, sizeof(results)/sizeof(results[0]),
                           &resultsCount);
            if (newClauseToStatementRef == oldClauseToStatementRef) {
                break;
            }
        } while (!atomic_compare_exchange_weak(&db->clauseToStatementRef,
                                               &oldClauseToStatementRef,
                                               newClauseToStatementRef));
        epochEnd();
    }

    /* printf("reactToRemovedStatement: s%d:%d (%s)\n", stmt - &db->statementPool[0], stmt->gen, */
    /*        clauseToString(stmt->clause)); */
    pthread_mutex_lock(&stmt->childMatchesMutex);
    ListOfEdgeTo* childMatches = stmt->childMatches;
    assert(childMatches != NULL);
    // Guarantees that no further matches can be added (we would be
    // unable to remove those).
    stmt->childMatches = NULL;
    genRcMarkAsDead(&stmt->genRc);
    pthread_mutex_unlock(&stmt->childMatchesMutex);

    for (size_t i = 0; i < childMatches->nEdges; i++) {
        MatchRef childRef = { .val = childMatches->edges[i] };
        Match* child = matchAcquire(db, childRef);
        if (child != NULL) {
            // The removal of _any_ of a Match's statement parents
            // means the removal of that Match.
            matchRemoveSelf(db, child);
            matchRelease(db, child);
        }
    }
    free(childMatches);
}

////////////////////////////////////////////////////////////
// Match:
////////////////////////////////////////////////////////////

Match* matchAcquire(Db* db, MatchRef ref) {
    if (ref.idx == 0) { return NULL; }

    Match* m = &db->matchPool[ref.idx];
    if (genRcAcquire(&m->genRc, ref.gen)) {
        return m;
    } else {
        return NULL;
    }
}
static void matchDestroy(Match* match);
void matchRelease(Db* db, Match* match) {
    if (genRcRelease(&match->genRc)) {
        matchDestroy(match);
    }
}

bool matchCheck(Db* db, MatchRef ref) {
    Match* m = &db->matchPool[ref.idx];
    GenRc genRc = m->genRc;
    return ref.gen >= 0 && ref.gen == genRc.gen;
}

MatchRef matchRef(Db* db, Match* match) {
    GenRc genRc = match->genRc;
    return (MatchRef) {
        .gen = genRc.gen,
        .idx = match - &db->matchPool[0]
    };
}

static MatchRef matchNew(Db* db,
                         AtomicallyVersion* atomicallyVersion,
                         int workerThreadIndex) {
    MatchRef ret;
    Match* match = NULL;

    // Look for a free match slot to use:
    while (1) {
        int32_t idx = db->matchPoolNextIdx++;
        if (idx == 0) { continue; } // skip null match
        match = &db->matchPool[idx];

        GenRc oldGenRc = match->genRc;
        if (oldGenRc.rc == 0 && !oldGenRc.alive &&
            atomic_load_explicit(&match->childStatements, memory_order_acquire) == NULL) {
            GenRc newGenRc = oldGenRc;
            newGenRc.alive = true;

            if (atomic_compare_exchange_weak(&match->genRc, &oldGenRc, newGenRc)) {
                ret = (MatchRef) { .gen = newGenRc.gen, .idx = idx };
                break;
            }
        }
    }

    // We should have exclusive access to match right now.

    atomic_store_explicit(&match->childStatements, listOfEdgeToNew(8), memory_order_relaxed);
    match->parentWasRemoved = false;

    pthread_mutexattr_t mta;
    pthread_mutexattr_init(&mta);
    pthread_mutexattr_settype(&mta, PTHREAD_MUTEX_RECURSIVE);
    pthread_mutex_init(&match->childStatementsMutex, &mta);
    pthread_mutexattr_destroy(&mta);

    match->atomicallyVersion = atomicallyVersion;
    match->isAtomicallyRootMatch = false;
    match->workerThreadIndex = workerThreadIndex;
    match->isCompleted = false;

    destructorSetInit(&match->destructorSet);
    pthread_mutex_init(&match->destructorSetMutex, NULL);

    return ret;
}

static void matchDestroy(Match* match) {
    assert(atomic_load_explicit(&match->childStatements, memory_order_relaxed)
           == CHILD_STATEMENTS_REMOVING);

    // Fire any destructors.
    pthread_mutex_lock(&match->destructorSetMutex);
    destructorSetReleaseAll(&match->destructorSet);
    pthread_mutex_unlock(&match->destructorSetMutex);

    // Release store: synchronizes with matchNew's acquire load so that
    // all writes above are visible before the slot is reused.
    atomic_store_explicit(&match->childStatements, NULL, memory_order_release);
}

AtomicallyVersion* matchAtomicallyVersion(Match* m) {
    return m->atomicallyVersion;
}
void matchSetAtomicallyVersion(Match* m, AtomicallyVersion* a) {
    m->atomicallyVersion = a;
}

static bool statementChecker(void* db, uint64_t ref) {
    return statementCheck((Db*) db, (StatementRef) { .val = ref });
}
// You must call this with the childStatementsMutex held.
static void matchAddChildStatement(Db* db, Match* match, StatementRef child) {
    ListOfEdgeTo* list = atomic_load_explicit(&match->childStatements, memory_order_relaxed);
    listOfEdgeToAdd(statementChecker, db, &list, child.val);
    atomic_store_explicit(&match->childStatements, list, memory_order_relaxed);
}
void matchAddDestructor(Match* m, Destructor* d) {
    pthread_mutex_lock(&m->destructorSetMutex);
    destructorSetAdd(&m->destructorSet, d);
    pthread_mutex_unlock(&m->destructorSetMutex);
}

void matchCompleted(Match* match) {
    match->isCompleted = true;
}
// Call matchRemoveSelf when ANY of the match's parent statements is
// removed.
//
// FIXME: Make this thread-safe (if called by multiple removers at the
// same time, it shouldn't double-free).
extern ThreadControlBlock threads[];
extern void traceItem(char* buf, size_t bufsz, WorkQueueItem item);
void matchRemoveSelf(Db* db, Match* match) {
    /* assert(match > &db->matchPool[0] && match < &db->matchPool[65536]); */
    if (match->isAtomicallyRootMatch &&
        atomic_load(&match->atomicallyVersion) != NULL) {
        // The AtomicallyVersion owns this root until the reaper clears
        // atomicallyVersion. Don't dereference it here: the reaper may be
        // freeing it concurrently.
        match->parentWasRemoved = true;
        if (atomic_load(&match->atomicallyVersion) != NULL) {
            return;
        }
        // The reaper cleared the pin between our two loads. Exactly one of
        // us must claim removal.
        if (!atomic_exchange(&match->parentWasRemoved, false)) {
            return;
        }
    }

    // Walk through each child statement and remove this match as a
    // parent of that statement.
    pthread_mutex_lock(&match->childStatementsMutex);
    ListOfEdgeTo* childStatements = atomic_load_explicit(&match->childStatements, memory_order_relaxed);
    if (childStatements == NULL || childStatements == CHILD_STATEMENTS_REMOVING) {
        // Someone else has done / is doing removal. Abort.
        pthread_mutex_unlock(&match->childStatementsMutex);
        return;
    }
    // This blocks further child statements from being added to this
    // match (if they were added, then we wouldn't be able to remove
    // them).
    atomic_store_explicit(&match->childStatements, CHILD_STATEMENTS_REMOVING, memory_order_relaxed);
    genRcMarkAsDead(&match->genRc);
    pthread_mutex_unlock(&match->childStatementsMutex);

    for (size_t i = 0; i < childStatements->nEdges; i++) {
        StatementRef childRef = { .val = childStatements->edges[i] };
        Statement* child = statementAcquire(db, childRef);
        if (child != NULL) {
            statementDecrParentCountAndMaybeRemoveSelf(db, child);
            statementRelease(db, child);
        }
    }
    free(childStatements);

    if (!match->isCompleted) {
        // Signal the match worker thread to terminate the match
        // execution.
        ThreadControlBlock *workerThread = &threads[match->workerThreadIndex];
        if (timestamp_get(workerThread->clockid) - workerThread->currentItemStartTimestamp > 100000000) {
            mutexLock(&workerThread->currentItemMutex);
            WorkQueueItem item = workerThread->currentItem;
            mutexUnlock(&workerThread->currentItemMutex);

            char buf[10000]; traceItem(buf, sizeof(buf), item);
            fprintf(stderr, "KILL (%.150s)\n", buf);
#ifdef __LINUX__
            syscall(SYS_tgkill, getpid(), workerThread->tid, SIGUSR1);
#else
            kill(workerThread->tid, SIGUSR1);
#endif
        }
    }
}

////////////////////////////////////////////////////////////
// Database:
////////////////////////////////////////////////////////////

Db* dbNew() {
    Db* ret = calloc(sizeof(Db), 1);

    ret->statementPool[0].genRc = (GenRc) { .gen = -1, .rc = 0 };
    ret->statementPoolNextIdx = 1;

    ret->matchPool[0].genRc = (GenRc) { .gen = -1, .rc = 0 };
    ret->matchPoolNextIdx = 1;

    ret->clauseToStatementRef = trieNew();

    sh_new_arena(ret->holds);
    mutexInit(&ret->holdsMutex);

    mutexInit(&ret->atomicallysMutex);

    return ret;
}

// Used by trie-graph.folk. Avoid if you can.
void dbLockClauseToStatementRef(Db* db) {
    epochBegin();
}
const Trie* dbGetClauseToStatementRef(Db* db) {
    return db->clauseToStatementRef;
}
void dbUnlockClauseToStatementRef(Db* db) {
    epochEnd();
}

// Query
ResultSet* dbQuery(Db* db, Clause* pattern) {
    ResultSet *resultSet;
    size_t maxResults = 500;
    do {
        resultSet = malloc(SIZEOF_RESULTSET(maxResults));

        epochBegin();
        resultSet->nResults =
            trieLookup(db->clauseToStatementRef, pattern,
                       (uint64_t*) resultSet->results, maxResults);
        epochEnd();

        if (resultSet->nResults < maxResults) {
            break;
        }

        maxResults *= 2;
        if (maxResults > 10 * 1000) {
            fprintf(stderr, "dbQuery: Too many results for query (%s)\n",
                    clauseToString(pattern));
            exit(1);
        }
        free(resultSet);
    } while (true);

    return resultSet;
}

AtomicallyVersion* dbFreshAtomicallyVersionOnKey(Db* db, const char* key,
                                                 MatchRef rootMatchRef) {
    mutexLock(&db->atomicallysMutex);

    Atomically* atomically = NULL;
    for (unsigned long i = 0; i < sizeof(db->atomicallys)/sizeof(db->atomicallys[0]); i++) {
        if (db->atomicallys[i].key != NULL &&
            strcmp(db->atomicallys[i].key, key) == 0) {

            atomically = &db->atomicallys[i];
            break;
        }
    }
    if (atomically == NULL) {
        for (unsigned long i = 0; i < sizeof(db->atomicallys)/sizeof(db->atomicallys[0]); i++) {
            if (db->atomicallys[i].key == NULL) {
                atomically = &db->atomicallys[i];
                atomically->key = strdup(key);
                atomically->nextNumber = 0;
                atomically->allVersions = NULL;
                atomically->timeout = 100000000; // 100ms
                atomically->latestConvergedTime = 0;
                break;
            }
        }
    }
    if (atomically == NULL) {
        fprintf(stderr, "dbGetOrCreateAtomicallyByKey: Ran out of Atomically slots\n");
        exit(1);
    }
    mutexUnlock(&db->atomicallysMutex);

    AtomicallyVersion* atomicallyVersion = malloc(sizeof(AtomicallyVersion));
    atomicallyVersion->atomically = atomically;

    // FIXME: assert old values are bad, do something with refcount

    atomicallyVersion->number = atomically->nextNumber++;
    // An AtomicallyVersion should start unconverged, assuming that it
    // always gets set into a currently running (incomplete) match
    // that can mark it as converged when done.
    atomicallyVersion->inflightCount = 1;
    atomicallyVersion->rootMatch = matchAcquire(db, rootMatchRef);
    assert(atomicallyVersion->rootMatch != NULL);
    atomicallyVersion->rootMatch->isAtomicallyRootMatch = true;

    // Add this version to atomically->allVersions list
    AtomicallyVersionList* newNode = malloc(sizeof(AtomicallyVersionList));
    newNode->version = atomicallyVersion;

    AtomicallyVersionList* oldList;
    do {
        oldList = atomically->allVersions;
        newNode->next = oldList;
    } while (!atomic_compare_exchange_weak(&atomically->allVersions,
                                           &oldList,
                                           newNode));

    return atomicallyVersion;
}
static void dbAtomicallyReapAllVersions(Db* db, Atomically* atomically,
                                        AtomicallyVersion* newlyConvergedVersion,
                                        bool onlyReapConvergedVersions) {
    // Swap out the entire list atomically, then process it after the
    // CAS loop.
    AtomicallyVersionList* allVersions;
    do {
        allVersions = atomically->allVersions;
    } while (!atomic_compare_exchange_weak(&atomically->allVersions,
                                           &allVersions,
                                           NULL));

    // Now we have exclusive ownership of allVersions. Process it to
    // separate old versions.
    AtomicallyVersionList* newList = NULL;
    AtomicallyVersionList* x = allVersions;
    while (x != NULL) {
        AtomicallyVersionList* next = x->next;
        if (x->version != NULL &&
            (newlyConvergedVersion == NULL ||
             x->version->number < newlyConvergedVersion->number) &&
            (!onlyReapConvergedVersions || x->version->inflightCount == 0)) {
            // Old version - clear and free it
            Match* rootMatch = x->version->rootMatch;
            x->version->rootMatch = NULL;
            if (rootMatch != NULL) {
                atomic_store(&rootMatch->atomicallyVersion, NULL);
                if (atomic_exchange(&rootMatch->parentWasRemoved, false)) {
                    matchRemoveSelf(db, rootMatch);
                }
                matchRelease(db, rootMatch);
            }
            free(x);
        } else {
            // Keep this version - add to new list
            x->next = newList;
            newList = x;
        }
        x = next;
    }

    // Write back the filtered list. We need to handle
    // concurrently-added versions. Find the tail of newList once,
    // then use CAS loop to atomically append.
    if (newList != NULL) {
        // Find the end of newList (which was built in reverse order)
        AtomicallyVersionList* tail = newList;
        while (tail->next != NULL) {
            tail = tail->next;
        }

        // CAS loop to atomically set tail->next and update allVersions
        AtomicallyVersionList* expected;
        do {
            expected = atomically->allVersions;
            tail->next = expected;  // Point tail to current head
        } while (!atomic_compare_exchange_weak(&atomically->allVersions,
                                               &expected,
                                               newList));
    }
    // Note: If newList is NULL (all versions were reaped), we
    // don't write anything back.  Concurrently-added versions
    // remain in allVersions and will be handled on next
    // convergence.
}
void dbGarbageCollectAtomicallys(Db* db, int64_t now) {
    mutexLock(&db->atomicallysMutex);
    for (int i = 0; i < sizeof(db->atomicallys)/sizeof(db->atomicallys[0]); i++) {
        if (db->atomicallys[i].key != NULL &&
            now - db->atomicallys[i].latestConvergedTime > db->atomicallys[i].timeout) {

            dbAtomicallyReapAllVersions(db, &db->atomicallys[i], NULL, true);
        }
    }
    mutexUnlock(&db->atomicallysMutex);
}

bool dbAtomicallyVersionHasConverged(AtomicallyVersion* atomicallyVersion) {
    return atomicallyVersion->inflightCount == 0;
}
int dbAtomicallyVersionInflightCount(AtomicallyVersion* atomicallyVersion) {
    /* printf("%p -- inflight count %d\n", atomicallyVersion, atomicallyVersion->inflightCount); */
    return atomicallyVersion->inflightCount;
}
int dbAtomicallyVersionNumber(AtomicallyVersion* atomicallyVersion) {
    return atomicallyVersion->number;
}
const char* dbAtomicallyVersionKey(AtomicallyVersion* atomicallyVersion) {
    return atomicallyVersion->atomically->key;
}
void dbInflightIncr(Statement* stmt) {
    if (stmt != NULL && stmt->atomicallyVersion != NULL) {
        stmt->atomicallyVersion->inflightCount++;
        /* printf("dbInflightIncr (%s) %p -> %d\n", clauseToString(stmt->clause), */
        /*        stmt->atomicallyVersion, */
        /*        stmt->atomicallyVersion->inflightCount); */
    }
}
void dbInflightDecr(Db* db, Statement* stmt) {
    if (stmt != NULL && stmt->atomicallyVersion != NULL) {
        /* printf("dbInflightDecr (%s) %p -> %d\n", clauseToString(stmt->clause), */
        /*        stmt->atomicallyVersion, */
        /*        stmt->atomicallyVersion->inflightCount - 1); */
        dbAtomicallyVersionInflightDecr(db, stmt->atomicallyVersion);
    }
}
void dbAtomicallyVersionInflightIncr(AtomicallyVersion* atomicallyVersion) {
    atomicallyVersion->inflightCount++;
    /* printf("dbAtomicallyVersionInflightIncr %p -> %d\n", */
    /*        atomicallyVersion, */
    /*        atomicallyVersion->inflightCount); */
}
void dbAtomicallyVersionInflightDecr(Db* db, AtomicallyVersion* atomicallyVersion) {
    if (--atomicallyVersion->inflightCount == 0) {
        Atomically* atomically = atomicallyVersion->atomically;
        atomically->latestConvergedVersion = atomicallyVersion;
        atomically->latestConvergedTime = timestamp_get(CLOCK_MONOTONIC);
        dbAtomicallyReapAllVersions(db, atomically, atomicallyVersion, false);
    }
    /* printf("dbAtomicallyVersionInflightDecr %p -> %d\n", */
    /*        atomicallyVersion, */
    /*        atomicallyVersion->inflightCount); */
}

// parentMatch is allowed to be NULL (meaning that no parent match is
// specified; the statement impulse comes directly from Assert! or
// Hold!). If parentMatch is not NULL, then you need to have
// (obviously) acquired the match _and_ to be holding its
// childStatementsMutex when you call this function.
static bool tryReuseStatement(Db* db, Statement* stmt, Match* parentMatch) {
    if (parentMatch != NULL) {
        // TODO: Update the sourceFileName and sourceLineNumber of the
        // stmt to reflect this new parent? (this isn't quite correct
        // either but better than nothing)

        if (!statementTryIncrParentCount(stmt)) {
            // We still want to insert/reuse the statement, but this
            // one is on the way out, so tell the caller to retry.
            return false;
        }

        matchAddChildStatement(db, parentMatch, statementRef(db, stmt));
        return true;
    }

    if (!statementTryIncrParentCount(stmt)) {
        return false;
    } else {
        return true;
    }
}

// Inserts a new statement with clause `clause` & returns a ref to
// that newly created statement & sets outReusedStatementRef to a null
// ref, UNLESS:
// 
//   - a statement is already present with that clause, in which case
//     we increment that statement's parent count & return a null ref
//     & set outReusedStatementRef to the already-present statement
//
//   - parentMatchRef has been invalidated, or its parentMatch has
//     childStatements invalidated, in which case we do nothing &
//     return a null ref & set outReusedStatementRef to a null ref
//     (because the whole situation has been invalidated)
// 
// (both of these mean that the caller shouldn't trigger a reaction,
// since no new statement is being created).
//
// Takes ownership of `clause` (i.e., you can't touch clause at the
// caller after calling this!).
Statement* dbInsertOrReuseStatement(Db* db, Clause* clause,
                                    long keepMs, AtomicallyVersion* atomicallyVersion,
                                    const char* sourceFileName, int sourceLineNumber,
                                    MatchRef parentMatchRef,
                                    StatementRef* outReusedStatementRef) {
#define setReusedStatementRef(_ref) \
    if (outReusedStatementRef != NULL) { \
        *outReusedStatementRef = (_ref); \
    }

    Match* parentMatch = NULL;
    if (!matchRefIsNull(parentMatchRef)) {
        // Need to set up parent match.
        parentMatch = matchAcquire(db, parentMatchRef);
        if (parentMatch == NULL) {
            setReusedStatementRef(STATEMENT_REF_NULL);

            /* fprintf(stderr, "parentMatch == NULL; aborted Say (%s)\n", clauseToString(clause)); */
            clauseFree(clause);
            return NULL; // Abort!
        }

        pthread_mutex_lock(&parentMatch->childStatementsMutex);
        ListOfEdgeTo* cs = atomic_load_explicit(&parentMatch->childStatements, memory_order_relaxed);
        if (cs == NULL || cs == CHILD_STATEMENTS_REMOVING) {
            pthread_mutex_unlock(&parentMatch->childStatementsMutex);
            matchRelease(db, parentMatch);

            setReusedStatementRef(STATEMENT_REF_NULL);

            /* fprintf(stderr, "parentMatch->childStatements == NULL (%p); aborted Say (%s)\n", */
            /*         parentMatch, */
            /*         clauseToString(clause)); */
            clauseFree(clause);
            return NULL; // Abort!
        }

        // Given that we have a parent match, if we've reached this
        // point, we now have a guarantee that the parentMatch is
        // acquired and we hold its childStatementsMutex and can add
        // to its childStatements list.
    }

    // We'll provisionally create a new statement to add.
    // 
    // Also transfers ownership of `clause` to the DB.
    StatementRef ref = statementNew(db, clause,
                                    keepMs, atomicallyVersion,
                                    sourceFileName, sourceLineNumber);

    // Now try to add to the trie: the trieAdd operation will
    // atomically detect if the clause is already present.
    epochBegin();
    const Trie* oldClauseToStatementRef;
    const Trie* newClauseToStatementRef;
    do {
        epochReset();
        oldClauseToStatementRef = db->clauseToStatementRef;
        newClauseToStatementRef = trieAdd(oldClauseToStatementRef,
                                          epochAlloc, epochFree,
                                          clause, ref.val);

        if (newClauseToStatementRef == oldClauseToStatementRef) {
            // The statement is possibly already present in the db --
            // trieAdd reported that it did not add anything -- so we
            // should try to reuse the existing statement.
            StatementRef existingRefs[10];
            int existingRefsCount = 
                trieLookupLiteral(oldClauseToStatementRef, clause,
                                  (uint64_t*)existingRefs,
                                  sizeof(existingRefs)/sizeof(existingRefs[0]));
            Statement* stmt;
            if (existingRefsCount == 1 && (stmt = statementAcquire(db, existingRefs[0]))) {
                // This is sort of the expected case. The statement is
                // indeed already present in the db, and we've been
                // able to acquire it.
                //
                // TODO: Warn if keepMs differs between existing and
                // newly-proposed statement?
                if (tryReuseStatement(db, stmt, parentMatch)) {
                    // TODO: Add the new destructor passed in?
                    if (parentMatch != NULL) {
                        pthread_mutex_lock(&parentMatch->destructorSetMutex);
                        pthread_mutex_lock(&stmt->destructorSetMutex);
                        destructorSetInherit(&stmt->destructorSet,
                                             &parentMatch->destructorSet);
                        pthread_mutex_unlock(&stmt->destructorSetMutex);
                        pthread_mutex_unlock(&parentMatch->destructorSetMutex);
                    }

                    statementRelease(db, stmt);

                    epochReset();
                    epochEnd();

                    // Free the new statement `ref` that we created,
                    // since we won't be using it.
                    Statement* newStmt = statementAcquire(db, ref);
                    newStmt->parentCount = 0;
                    statementRemoveSelf(db, newStmt, false);
                    statementRelease(db, newStmt);

                    if (parentMatch != NULL) { 
                        pthread_mutex_unlock(&parentMatch->childStatementsMutex);

                        matchRelease(db, parentMatch);
                    }

                    setReusedStatementRef(existingRefs[0]);
                    return NULL;
                } else {
                    // Reuse failed, but not for operation-aborting
                    // reasons -- we just need to actually make the
                    // new statement, probably.
                    statementRelease(db, stmt);
                    continue; // Retry.
                }

            } else if (existingRefsCount > 1) {
                // The database invariant has been violated.
                fprintf(stderr, "dbInsertOrReuseStatement: FATAL: "
                        "More than 1 statement with same clause.\n");
                exit(1);
            } else {
                // Something changed and now that duplicate statement
                // is gone (existingRefsCount == 0 or acquire
                // failed). Let's start over and try to add again.
                continue;
            }
        }

        // Note: continue statements from reuse logic above jump here
    } while (newClauseToStatementRef == oldClauseToStatementRef ||
             !atomic_compare_exchange_weak(&db->clauseToStatementRef,
                                           &oldClauseToStatementRef,
                                           newClauseToStatementRef));
    epochEnd();

    Statement* newStmt = statementAcquire(db, ref);
    assert(newStmt != NULL);

    // OK, we've made a new statement. trieAdd added the statement to
    // the db and we committed the new db.
    if (parentMatch != NULL) {
        matchAddChildStatement(db, parentMatch, ref);

        pthread_mutex_lock(&parentMatch->destructorSetMutex);
        pthread_mutex_lock(&newStmt->destructorSetMutex);
        destructorSetInherit(&newStmt->destructorSet,
                             &parentMatch->destructorSet);
        pthread_mutex_unlock(&newStmt->destructorSetMutex);
        pthread_mutex_unlock(&parentMatch->destructorSetMutex);

        pthread_mutex_unlock(&parentMatch->childStatementsMutex);
        matchRelease(db, parentMatch);
    }

    setReusedStatementRef(STATEMENT_REF_NULL);
    return newStmt;

#undef setReusedStatementRef
}

Match* dbInsertMatch(Db* db, int nParents, StatementRef parents[],
                     AtomicallyVersion* atomicallyVersion,
                     int workerThreadIndex) {
    MatchRef ref = matchNew(db, atomicallyVersion, workerThreadIndex);
    Match* match = matchAcquire(db, ref);
    assert(match != NULL);

    // All parents need to be valid and need to have locked
    // childMatches before we insert the match. Otherwise, abort.
    bool failed = false;

    Statement* parentStatements[nParents];
    memset(parentStatements, 0, sizeof(parentStatements));
    for (int i = 0; i < nParents; i++) {
        parentStatements[i] = statementAcquire(db, parents[i]);
        if (parentStatements[i] == NULL) {
            failed = true; goto done;
        }

        pthread_mutex_lock(&parentStatements[i]->childMatchesMutex);
        if (parentStatements[i]->childMatches == NULL) {
            failed = true; goto done;
        }
    }

    // We have now acquired all parent statements and are holding
    // their childMatchesMutexes, and none have childMatches == NULL.

    // Now we can do the actual insertion.
    for (int i = 0; i < nParents; i++) {
        statementAddChildMatch(db, parentStatements[i], ref);

        // We should also inherit all destructors from each parent
        // statement.
        pthread_mutex_lock(&parentStatements[i]->destructorSetMutex);
        pthread_mutex_lock(&match->destructorSetMutex);
        destructorSetInherit(&match->destructorSet,
                             &parentStatements[i]->destructorSet);
        pthread_mutex_unlock(&match->destructorSetMutex);
        pthread_mutex_unlock(&parentStatements[i]->destructorSetMutex);
    }

done:
    for (int i = nParents - 1; i >= 0; i--) {
        if (parentStatements[i] == NULL) {
            continue;
        }
        pthread_mutex_unlock(&parentStatements[i]->childMatchesMutex);
        statementRelease(db, parentStatements[i]);
    }

    if (failed) {
        matchRemoveSelf(db, match);
        matchRelease(db, match);
        return NULL;
    } else {
        return match;
    }
}

void dbRetractStatements(Db* db, Clause* pattern) {
    StatementRef results[500];
    size_t maxResults = sizeof(results)/sizeof(results[0]);

    // TODO: Should we accept a StatementRef and enforce that is what
    // gets removed?
    epochBegin();
    size_t nResults = trieLookup(db->clauseToStatementRef, pattern,
                                 (uint64_t*) results, maxResults);
    epochEnd();

    if (nResults == maxResults) {
        // TODO: Try again with a larger maxResults?
        fprintf(stderr, "dbQuery: Hit max results\n"); exit(1);
    }

    for (size_t i = 0; i < nResults; i++) {
        Statement* stmt = statementAcquire(db, results[i]);

        statementDecrParentCountAndMaybeRemoveSelf(db, stmt);
        statementRelease(db, stmt);
    }
}

// Takes ownership of `clause` and `destructorSet`.
Statement* dbHoldStatement(Db* db,
                           const char* key, double version,
                           Clause* clause, long keepMs,
                           const char* sourceFileName, int sourceLineNumber,
                           StatementRef* outOldStatement) {
    if (outOldStatement) { *outOldStatement = STATEMENT_REF_NULL; }

    mutexLock(&db->holdsMutex);

    Hold* hold = shgetp_null(db->holds, key);
    if (hold == NULL) {
        Hold newHold = { .key = (char*)key, .version = -1, .statement = STATEMENT_REF_NULL };
        shputs(db->holds, newHold);
        hold = shgetp_null(db->holds, key);
    }

    if (version < 0) {
        version = hold->version + 1;
    }
    if (version > hold->version) {
        StatementRef oldStmt = hold->statement;
        // TODO: Should we accept a StatementRef and enforce that
        // is what gets removed?
        Statement* oldStmtPtr = statementAcquire(db, oldStmt);
        if (oldStmtPtr && clauseIsEqual(clause, statementClause(oldStmtPtr))) {
            statementRelease(db, oldStmtPtr);
            mutexUnlock(&db->holdsMutex);
            clauseFree(clause);
            return NULL;
        }

        Statement* newStmt = NULL;
        if (clause->nTerms > 0) {
            hold->version = version;

            StatementRef reusedStatementRef;
            newStmt = dbInsertOrReuseStatement(db, clause, keepMs, NULL,
                                               sourceFileName, sourceLineNumber,
                                               MATCH_REF_NULL,
                                               &reusedStatementRef);
            if (newStmt != NULL) {
                hold->statement = statementRef(db, newStmt);
            } else if (!statementRefIsNull(reusedStatementRef)) {
                hold->statement = reusedStatementRef;
            } else {
                fprintf(stderr, "dbHoldStatement: ERROR: Ref neither reused nor created\n");
                exit(1);
            }
        } else {
            clauseFree(clause);
            shdel(db->holds, key);
            hold = NULL;
        }


        if (oldStmtPtr) {
            // We deindex (trieRemove) the old statement immediately,
            // but we leave it to the caller to actually destroy the
            // statement itself (and therefore remove all its
            // children).
            epochBegin();

            uint64_t results[10]; int resultsCount;

            const Trie* oldClauseToStatementRef;
            const Trie* newClauseToStatementRef;
            do {
                epochReset();
                oldClauseToStatementRef = db->clauseToStatementRef;
                newClauseToStatementRef =
                    trieRemove(db->clauseToStatementRef,
                               epochAlloc, epochFree,
                               statementClause(oldStmtPtr),
                               (uint64_t*) results, sizeof(results)/sizeof(results[0]),
                               &resultsCount);
                if (newClauseToStatementRef == oldClauseToStatementRef) {
                    break;
                }
            } while (!atomic_compare_exchange_weak(&db->clauseToStatementRef,
                                                   &oldClauseToStatementRef,
                                                   newClauseToStatementRef));
            epochEnd();

            statementRelease(db, oldStmtPtr);
        } else if (oldStmt.idx != 0) {
            fprintf(stderr, "Somehow old statement from Hold (%d:%d) was already removed?\n",
                    oldStmt.idx, oldStmt.gen);
        }

        /* assert(nRemoved == 1); */
        /* assert(results[0] == oldStmt.val); */

        if (outOldStatement) { *outOldStatement = oldStmt; }

        mutexUnlock(&db->holdsMutex);
        return newStmt;
    } else {
        // The new version is older than the version already in the
        // hold, so we just shouldn't do anything / we shouldn't
        // install the new statement.
        mutexUnlock(&db->holdsMutex);
        clauseFree(clause);
        return NULL;
    }
}