Group the commits of concurrent threads

Threads that commit at the same time now share one commit: they queue,
the first becomes the leader and publishes every queued request of the
same store with one head, and the others wait for its result. Each
request stays all or nothing on its own: one with a stale draft is left
out, and so is one whose key an earlier request of the group changes.
When several threads have been committing together, a leader waits up to
2 ms for as many to join as groups held lately; a single thread never
waits.

A test pauses a publication and checks that five commits make exactly two
publications. Under eight writers on three pure-serve shards, 1,234
bursts went out in 421 publications. pure_store_group_counts reports the
groups, and pure-serve prints them.

1 files changed, 49 insertions, 2 deletions

diff --git a/prototypes/connections/test-tree-store.c b/prototypes/connections/test-tree-store.c
index bc8ba4b..9cb0701 100644
--- a/prototypes/connections/test-tree-store.c
+++ b/prototypes/connections/test-tree-store.c
@@ -10,13 +10,20 @@
 #include <sys/stat.h>
 #include <pthread.h>
 #include <sys/wait.h>
+#include <time.h>
 #include <unistd.h>
 
 #define CHECK(x) do { if (!(x)) { fprintf(stderr, "%s:%d: %s\n", __FILE__, __LINE__, #x); abort(); } } while (0)
 
 static int fail_stage = -1;
 static int crash;
+static int slow_once; /* the next publication pauses before its data sync */
+static void pause_ms(long ms) {
+    struct timespec wait = {ms / 1000, (ms % 1000) * 1000000};
+    nanosleep(&wait, NULL);
+}
 int pure_store_test_fault(PureStoreStage stage) {
+    if (stage == PURE_BEFORE_FILE_SYNC && __atomic_exchange_n(&slow_once, 0, __ATOMIC_SEQ_CST)) pause_ms(100);
     if ((int)stage != fail_stage) return 0;
     if (crash) _exit(77);
     return 1;
@@ -25,6 +32,7 @@ bool pure_tree_test_shape(const char *path, const char *key, unsigned *height,
                           uint64_t *leaves, uint64_t *small, uint64_t *file_end);
 
 void pure_draft_test_arena(const PureDraft *draft, size_t *nodes, size_t *bytes);
+unsigned long pure_tree_test_publications(void);
 
 enum { KEYS = 4, CAPACITY = 1 << 20 };
 typedef struct { unsigned char *bytes; size_t size; int64_t revision; } Oracle;
@@ -880,6 +888,45 @@ static void threads(const char *path) {
     pure_store_release(&view);
 }
 
+/* Commits that arrive while another is being published share the next
+ * publication: one head for all of them. */
+typedef struct { const char *path; unsigned number; PureResult result; } Single;
+
+static void *commit_one(void *context) {
+    Single *s = context;
+    PureStore *store;
+    PureDraft *draft;
+    char key[16];
+    int64_t revision;
+    snprintf(key, sizeof(key), "g%u", s->number);
+    CHECK(pure_store_open(s->path, &store) == PURE_OK && pure_store_draft(store, key, &draft) == PURE_OK);
+    CHECK(pure_draft_splice(draft, 0, 0, (PureSpan){(const unsigned char *)key, strlen(key)}) == PURE_OK);
+    s->result = pure_draft_commit(draft, &revision);
+    CHECK(s->result != PURE_OK || revision == 1);
+    pure_draft_close(draft);
+    pure_store_close(store);
+    return NULL;
+}
+
+static void grouped(const char *path) {
+    pthread_t thread[5];
+    Single single[5];
+    unsigned long before = pure_tree_test_publications();
+    __atomic_store_n(&slow_once, 1, __ATOMIC_SEQ_CST);
+    for (unsigned i = 0; i < 5; ++i) {
+        single[i] = (Single){path, i, PURE_INVALID};
+        CHECK(pthread_create(&thread[i], NULL, commit_one, &single[i]) == 0);
+        if (!i) pause_ms(20); /* the first is now publishing, slowly */
+    }
+    for (unsigned i = 0; i < 5; ++i) CHECK(pthread_join(thread[i], NULL) == 0 && single[i].result == PURE_OK);
+    CHECK(pure_tree_test_publications() - before == 2);
+    char key[16];
+    for (unsigned i = 0; i < 5; ++i) {
+        snprintf(key, sizeof(key), "g%u", i);
+        expect(path, key, &(Oracle){(unsigned char *)key, strlen(key), 1});
+    }
+}
+
 static void clean(const char *path) {
     DIR *entries = opendir(path);
     CHECK(entries);
@@ -896,7 +943,7 @@ int main(void) {
     void (*checks[])(const char *) = {oracle, amplification, faults, crashes, torn_head,
                                       isolation, corruption, many_keys, revision_limit, keys,
                                       drafts, typing, draft_conflict, draft_relocation, draft_faults,
-                                      draft_versions, anonymous_draft, batch, store_handle, threads};
+                                      draft_versions, anonymous_draft, batch, store_handle, threads, grouped};
     for (size_t i = 0; i < sizeof(checks) / sizeof(*checks); ++i) {
         char path[] = "/tmp/pure-tree-check-XXXXXX";
         CHECK(mkdtemp(path) && pure_store_init(path) == PURE_OK);
@@ -907,6 +954,6 @@ int main(void) {
     puts("{\"tree_store\":\"random-oracle-balance-held-views-one-byte-amplification-faults-"
          "process-crash-torn-head-lockless-isolation-compaction-corruption-1024-keys-revision-limit-key-listing-"
          "draft-oracle-typing-arena-conflict-relocation-commit-faults-held-versions-anonymous-"
-         "atomic-batch-runtime-keys-store-handle-threads\"}");
+         "atomic-batch-runtime-keys-store-handle-threads-group-commit\"}");
     return 0;
 }