Find a burst's keys in one look

A message for a document scanned the burst's pending keys with strcmp to
see whether its key was there, and the host scanned the burst's earlier
requests the same way before each one: both quadratic in the burst (64
concurrent reads, about 31 compares each, twice). The app now indexes its
pending documents by key (open addressing over an FNV-1a hash, 1 KiB),
answers pure_app_pending in one look, and the host asks it; whether the
current event's document is pending is a flag instead of a scan by
pointer. Bursts split where they did, except that on an owner, which
applies a round's reads first, an edit of a key read in the same round
now splits the burst at the host rather than the app saving on its own.
A unit test checks the index at 200 keys.

3 files changed, 59 insertions, 7 deletions

diff --git a/core/module.c b/core/module.c
index 697e918..7adef93 100644
--- a/core/module.c
+++ b/core/module.c
@@ -1262,14 +1262,23 @@ PURE_COLD static PureList *new_list(const PureModule *m, uint32_t record) {
     return pure_list_new(m->record[record].count, m->record_field_text + m->record[record].first);
 }
 
-static bool pending(const PureApp *app, const void *document) {
-    for (unsigned i = 0; i < app->pending_count; ++i) if (app->pending[i].document == document) return true;
-    return false;
+/* Where key is in the pending index: its slot, or the empty one it would take. */
+static unsigned pending_slot(const PureApp *app, const char *key) {
+    uint32_t hash = 2166136261u; /* FNV-1a */
+    for (const char *c = key; *c; ++c) hash = (hash ^ (unsigned char)*c) * 16777619u;
+    unsigned mask = 2 * PURE_APP_PENDING - 1, slot = hash & mask;
+    while (app->pending_index[slot] && strcmp(app->pending[app->pending_index[slot] - 1].key, key)) slot = (slot + 1) & mask;
+    return slot;
+}
+
+bool pure_app_pending(const PureApp *app, const char *key) {
+    return app->pending_count && app->pending_index[pending_slot(app, key)];
 }
 
 void pure_app_finish(PureApp *app) {
     /* A document in the unsaved burst stays open until the save. */
-    if (app->document && !pending(app, app->document)) app->data.close(app->data.context, app->document);
+    if (app->document && !app->document_pending) app->data.close(app->data.context, app->document);
+    app->document_pending = false;
     free(app->copied);
     app->document = NULL;
     app->copied = NULL;
@@ -1304,7 +1313,9 @@ static void open_burst(PureApp *app) {
 static void close_burst(PureApp *app, bool keep) {
     for (unsigned i = 0; i < app->pending_count; ++i)
         if (app->pending[i].document != app->document) app->data.close(app->data.context, app->pending[i].document);
+    if (app->pending_count) memset(app->pending_index, 0, sizeof(app->pending_index));
     app->pending_count = 0;
+    app->document_pending = false;
     if (!app->open) return;
     const PureModule *m = app->module;
     for (uint32_t i = 0; i < m->states; ++i) {
@@ -1519,9 +1530,11 @@ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) {
     ++app->changes;
     if (!event) ++app->ticks;
     if (app->document) {
+        app->pending_index[pending_slot(app, ev->key)] = (uint16_t)(app->pending_count + 1); /* not there: a key joins once */
         app->pending[app->pending_count].document = app->document;
         app->pending[app->pending_count].edited = ev->modified;
         memcpy(app->pending[app->pending_count++].key, ev->key, strlen(ev->key) + 1);
+        app->document_pending = true;
         app->revision += ev->modified; /* a save moves an edited document one revision */
     }
     if (app->delivering.pending) {
@@ -1627,9 +1640,7 @@ static PureResult message(PureApp *app, uint32_t input, const char *key, int64_t
     if (shape->keyed && m->collections) {
         if (!app->data.open) return PURE_INVALID;
         /* A key joins a burst once, so its revision moves once per save. */
-        bool seen = app->pending_count == PURE_APP_PENDING;
-        for (unsigned i = 0; i < app->pending_count && !seen; ++i) seen = !strcmp(app->pending[i].key, key);
-        if (seen) {
+        if (app->pending_count == PURE_APP_PENDING || pure_app_pending(app, key)) {
             PureResult saved = pure_app_save(app);
             if (saved != PURE_OK) return saved;
         }
diff --git a/core/module.h b/core/module.h
index f62326b..b9a3450 100644
--- a/core/module.h
+++ b/core/module.h
@@ -214,6 +214,11 @@ typedef struct {
     uint64_t durable_ticks, durable_events;
     struct PurePending { void *document; char key[65]; bool edited; } *pending; /* grows to PURE_APP_PENDING */
     unsigned pending_count, pending_capacity;
+    /* The pending documents by key, so a message finds its key in one look:
+     * open addressing over the key's hash, a slot holding one more than the
+     * document's place in pending (0: none). */
+    uint16_t pending_index[2 * PURE_APP_PENDING];
+    bool document_pending; /* the current event's document is among them */
     PureFault fault;
     uint32_t fault_pc;
     uint64_t ticks, events;
@@ -258,6 +263,9 @@ PureResult pure_app_save(PureApp *);
 /* Defer saving to pure_app_save (false: save after each event). Needs a data
  * host that saves documents together (PureData.save). */
 bool pure_app_defer(PureApp *, bool deferred);
+/* Whether key's document is in the unsaved burst: a message for it would
+ * save the burst first. */
+bool pure_app_pending(const PureApp *, const char *key);
 /* Leaves unsaved frame changes unsaved: call pure_app_save first to keep them. */
 void pure_app_stop(PureApp *);
 PureResult pure_app_frame(PureApp *);
diff --git a/core/test-module.c b/core/test-module.c
index ccb3371..307d565 100644
--- a/core/test-module.c
+++ b/core/test-module.c
@@ -1716,6 +1716,38 @@ static void deferred_messages(void) {
     remove_store(path);
 }
 
+/* The burst's keys are found in one look (an index over their hashes, most
+ * slots shared at 200 keys): every pending key and no other, reads as well
+ * as edits, and a message for one of them saves the burst first. */
+static void pending_keys(void) {
+    char *path = new_store();
+    CHECK(load(editor_spec, &module_a) == PURE_OK);
+    PureApp app;
+    unsigned char buffer[64];
+    char key[8];
+    CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pure_app_defer(&app, true));
+    for (int i = 0; i < 200; ++i) {
+        snprintf(key, sizeof(key), "p%d", i);
+        CHECK(!pure_app_pending(&app, key));
+        CHECK(pure_app_message(&app, 1, key, 0, edit_payload(buffer, 0, 0, "q")) == PURE_OK);
+    }
+    CHECK(pure_app_save(&app) == PURE_OK && !pure_app_pending(&app, "p0"));
+    for (int i = 0; i < 200; ++i) {
+        snprintf(key, sizeof(key), "p%d", i);
+        CHECK((i % 2 ? pure_app_message(&app, 0, key, -1, (PureSpan){0})
+                     : pure_app_message(&app, 1, key, 1, edit_payload(buffer, 1, 0, "r"))) == PURE_OK);
+    }
+    for (int i = 0; i < 210; ++i) {
+        snprintf(key, sizeof(key), "p%d", i);
+        CHECK(pure_app_pending(&app, key) == (i < 200));
+    }
+    CHECK(!pure_app_pending(&app, "p") && !pure_app_pending(&app, "p1000") && app.pending_count == 200);
+    CHECK(pure_app_message(&app, 0, "p150", -1, (PureSpan){0}) == PURE_OK && app.pending_count == 1);
+    CHECK(pure_app_pending(&app, "p150") && !pure_app_pending(&app, "p151"));
+    pure_app_stop(&app);
+    remove_store(path);
+}
+
 static void deferred(void) {
     static const char *const list_words[] = {"a", "\xea\xb0\x80", "", "\n", "\xf0\x9f\x98\x80"};
     static const char *const text_words[] = {"a", "\xea\xb0\x80", "\n", "\xf0\x9f\x98\x80", "xy\nz", ""};
@@ -1723,6 +1755,7 @@ static void deferred(void) {
     deferred_equivalence(text_spec, text_words, LEN(text_words), 0x1b873593u);
     deferred_bursts();
     deferred_messages();
+    pending_keys();
 }
 
 /* Text inserted at a cursor goes before it; at a mark (a cursor that stays),