aboutsummaryrefslogtreecommitdiffstats
path: root/prototypes/connections
diff options
context:
space:
mode:
Diffstat (limited to 'prototypes/connections')
-rw-r--r--prototypes/connections/serve.c241
-rw-r--r--prototypes/connections/state.h4
-rw-r--r--prototypes/connections/store-tree.c22
-rw-r--r--prototypes/connections/test-tree-store.c60
4 files changed, 238 insertions, 89 deletions
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index 39f4f33..f90772a 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -10,7 +10,13 @@
* borrowed from the request). After each round of events every ready
* request is applied to the module with deferred saving, one save stores
* the burst, and only then are its requests answered: an answer is durable,
- * and a failed save fails the whole burst, reads included. */
+ * and a failed save fails the whole burst, reads included.
+ *
+ * With --threads N each thread is a shard with its own listener on the same
+ * port (SO_REUSEPORT), epoll set, runtime domain, app and store handle; the
+ * kernel spreads connections over them. Shards share nothing but the store,
+ * whose commits take turns and whose revisions settle concurrent edits of a
+ * key. A module with states (not only keyed documents) runs on one thread. */
#include "module-store.h"
#include "runtime.h"
#include "store.h"
@@ -20,7 +26,9 @@
#include <inttypes.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
+#include <pthread.h>
#include <signal.h>
+#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@@ -68,7 +76,8 @@ typedef struct {
} Applied;
struct Host {
- PureModule module;
+ const PureModule *module;
+ PureStore *store;
PureApp app;
uint32_t show, edit;
PureRuntime runtime;
@@ -83,8 +92,8 @@ struct Host {
uint64_t accepted, requests, bursts, burst_events, expired;
};
-static volatile sig_atomic_t stopping;
-static void stop(int signal_number) { (void)signal_number; stopping = 1; }
+static atomic_bool stopping; /* lock-free, so a signal handler may set it */
+static void stop(int signal_number) { (void)signal_number; atomic_store(&stopping, true); }
static uint64_t now_ms(void) {
struct timespec t;
@@ -428,7 +437,11 @@ static void finish_burst(Host *host, size_t count) {
static void serve_bursts(Host *host) {
size_t count = 0, keys = 0;
PureEffect effect;
+ /* One look at the store per burst: a request made after another shard's
+ * answer was sent is read after that answer's commit. */
+ bool pinned = false;
while (pure_runtime_take(&host->runtime, &effect) == PURE_TAKE_EFFECT) {
+ if (!pinned) pinned = pure_store_pin(host->store, true) == PURE_OK;
bool repeated = false;
for (size_t i = 0; i < count && !repeated; ++i) repeated = !strcmp(host->applied[i].key, effect.key);
if (repeated || keys == PURE_APP_PENDING) {
@@ -439,6 +452,7 @@ static void serve_bursts(Host *host) {
++keys;
}
finish_burst(host, count);
+ if (pinned) pure_store_pin(host->store, false);
}
/* PUT replies are borrowed from the request, so an edit handler may reply
@@ -456,18 +470,17 @@ static bool edit_replies_borrow(const PureModule *m, uint32_t edit) {
/* ---- Main ------------------------------------------------------------- */
-static bool load_module(Host *host, const char *path) {
+static bool load_module(PureModule *m, const char *path, uint32_t *show_out, uint32_t *edit_out) {
static unsigned char bytes[1 << 17];
char error[128];
FILE *file = fopen(path, "rb");
size_t size = file ? fread(bytes, 1, sizeof(bytes), file) : 0;
bool read_ok = file && !ferror(file) && size < sizeof(bytes);
if (file) fclose(file);
- if (!read_ok || pure_module_load(bytes, size, &host->module, error, sizeof(error)) != PURE_OK) {
+ if (!read_ok || pure_module_load(bytes, size, m, error, sizeof(error)) != PURE_OK) {
fprintf(stderr, "Module: %s\n", read_ok ? error : "cannot read");
return false;
}
- const PureModule *m = &host->module;
int show = pure_module_input(m, "show"), edit = pure_module_input(m, "edit");
if (show < 0 || edit < 0 || m->input[show].kind != PURE_KIND_MESSAGE || !m->input[show].keyed ||
m->input[edit].kind != PURE_KIND_MESSAGE || !m->input[edit].keyed) {
@@ -478,8 +491,8 @@ static bool load_module(Host *host, const char *path) {
fputs("Module: edit may reply only a field of its message\n", stderr);
return false;
}
- host->show = (uint32_t)show;
- host->edit = (uint32_t)edit;
+ *show_out = (uint32_t)show;
+ *edit_out = (uint32_t)edit;
return true;
}
@@ -519,113 +532,169 @@ static void accept_all(Host *host) {
}
}
-int main(int argc, char **argv) {
- static Host host;
- const char *module_path = NULL, *store = NULL;
- unsigned port = 8767;
- size_t budget = 64u << 20;
- host.capacity = 1024;
- host.timeout_ms = 10000;
- for (int i = 1; i < argc; i += 2) {
- uint64_t value;
- if (i + 1 == argc) goto usage;
- if (!strcmp(argv[i], "--module")) module_path = argv[i + 1];
- else if (!strcmp(argv[i], "--store")) store = argv[i + 1];
- else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value;
- else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value;
- else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) host.capacity = (size_t)value;
- else if (!strcmp(argv[i], "--timeout-ms") && number(argv[i + 1], 3600000, &value) && value) host.timeout_ms = value;
- else goto usage;
- }
- if (!module_path || !store || !load_module(&host, module_path)) goto usage;
- if (pure_store_init(store) != PURE_OK) { fputs("Cannot initialize the store directory\n", stderr); return 1; }
- PureStore *handle;
- if (pure_store_open(store, &handle) != PURE_OK) { fputs("Cannot open the store\n", stderr); return 1; }
- if (pure_app_start(&host.app, &host.module, pure_store_data(handle)) != PURE_OK || !pure_app_defer(&host.app, true)) {
+/* One shard: its own listener on the shared port, epoll set, runtime domain,
+ * app and store handle. */
+static bool start_shard(Host *host, const PureModule *module, uint32_t show, uint32_t edit, const char *store,
+ size_t budget, unsigned *port) {
+ host->module = module;
+ host->show = show;
+ host->edit = edit;
+ if (pure_store_open(store, &host->store) != PURE_OK) { fputs("Cannot open the store\n", stderr); return false; }
+ if (pure_app_start(&host->app, module, pure_store_data(host->store)) != PURE_OK || !pure_app_defer(&host->app, true)) {
fputs("Cannot start the module on the store\n", stderr);
- return 1;
+ return false;
}
- host.work = calloc(host.capacity, sizeof(*host.work));
- host.table = calloc(host.capacity, sizeof(*host.table));
- host.by_slot = calloc(host.capacity, sizeof(*host.by_slot));
- host.applied = calloc(host.capacity, sizeof(*host.applied));
- if (!host.work || !host.table || !host.by_slot || !host.applied) { fputs("Cannot allocate connections\n", stderr); return 1; }
- if (!pure_runtime_init(&host.runtime, host.work, host.capacity, &host.app,
+ host->work = calloc(host->capacity, sizeof(*host->work));
+ host->table = calloc(host->capacity, sizeof(*host->table));
+ host->by_slot = calloc(host->capacity, sizeof(*host->by_slot));
+ host->applied = calloc(host->capacity, sizeof(*host->applied));
+ if (!host->work || !host->table || !host->by_slot || !host->applied) { fputs("Cannot allocate connections\n", stderr); return false; }
+ if (!pure_runtime_init(&host->runtime, host->work, host->capacity, &host->app,
(PureHost){.allocate = allocate_ingress}, budget) ||
- !pure_runtime_enable_async(&host.runtime)) abort();
-
- signal(SIGPIPE, SIG_IGN);
- struct sigaction stop_action = {.sa_handler = stop};
- sigemptyset(&stop_action.sa_mask);
- sigaction(SIGINT, &stop_action, NULL);
- sigaction(SIGTERM, &stop_action, NULL);
- host.listener = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
+ !pure_runtime_enable_async(&host->runtime)) return false;
+ host->listener = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
int one = 1;
- setsockopt(host.listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
- struct sockaddr_in address = {.sin_family = AF_INET, .sin_port = htons((uint16_t)port),
+ setsockopt(host->listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
+ setsockopt(host->listener, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(one));
+ struct sockaddr_in address = {.sin_family = AF_INET, .sin_port = htons((uint16_t)*port),
.sin_addr.s_addr = htonl(INADDR_LOOPBACK)};
socklen_t address_size = sizeof(address);
- if (host.listener < 0 || bind(host.listener, (struct sockaddr *)&address, sizeof(address)) ||
- listen(host.listener, 4096) || getsockname(host.listener, (struct sockaddr *)&address, &address_size)) {
+ if (host->listener < 0 || bind(host->listener, (struct sockaddr *)&address, sizeof(address)) ||
+ listen(host->listener, 4096) || getsockname(host->listener, (struct sockaddr *)&address, &address_size)) {
perror("listen");
- return 1;
+ return false;
}
- host.epoll = epoll_create1(EPOLL_CLOEXEC);
+ *port = ntohs(address.sin_port); /* the first shard picks it, the others join it */
+ host->epoll = epoll_create1(EPOLL_CLOEXEC);
struct epoll_event listen_event = {.events = EPOLLIN, .data.ptr = NULL};
- if (host.epoll < 0 || epoll_ctl(host.epoll, EPOLL_CTL_ADD, host.listener, &listen_event)) { perror("epoll"); return 1; }
- printf("{\"port\":%u}\n", ntohs(address.sin_port));
- fflush(stdout);
+ if (host->epoll < 0 || epoll_ctl(host->epoll, EPOLL_CTL_ADD, host->listener, &listen_event)) {
+ perror("epoll");
+ return false;
+ }
+ return true;
+}
+static void *run_shard(void *context) {
+ Host *host = context;
struct epoll_event events[EVENTS];
uint64_t next_sweep = now_ms() + 1000;
- while (!stopping) {
- int ready = epoll_wait(host.epoll, events, EVENTS, 1000);
+ while (!atomic_load(&stopping)) {
+ int ready = epoll_wait(host->epoll, events, EVENTS, 1000);
for (int e = 0; e < ready; ++e) {
Connection *c = events[e].data.ptr;
- if (!c) { accept_all(&host); continue; }
+ if (!c) { accept_all(host); continue; }
if (c->phase == CLOSED) continue;
if (events[e].events & EPOLLERR) { close_connection(c); continue; }
if (events[e].events & (EPOLLIN | EPOLLRDHUP | EPOLLHUP)) c->readable = true;
if (events[e].events & EPOLLOUT) c->writable = true;
drive(c);
}
- serve_bursts(&host);
+ serve_bursts(host);
uint64_t now = now_ms();
if (now >= next_sweep) {
next_sweep = now + 1000;
- for (size_t i = 0; i < host.capacity; ++i) {
- Connection *c = host.table[i];
- if (c && c->phase != WAITING && now >= c->deadline) { ++host.expired; close_connection(c); }
+ for (size_t i = 0; i < host->capacity; ++i) {
+ Connection *c = host->table[i];
+ if (c && c->phase != WAITING && now >= c->deadline) { ++host->expired; close_connection(c); }
}
}
/* The round is over: closed connections can be reused. */
- while (host.closed) {
- Connection *c = host.closed;
- host.closed = c->next;
- c->next = host.spare;
- host.spare = c;
+ while (host->closed) {
+ Connection *c = host->closed;
+ host->closed = c->next;
+ c->next = host->spare;
+ host->spare = c;
}
}
- pure_runtime_stop(&host.runtime);
- serve_bursts(&host);
- for (size_t i = 0; i < host.capacity; ++i) if (host.table[i]) close_connection(host.table[i]);
+ pure_runtime_stop(&host->runtime);
+ serve_bursts(host);
+ for (size_t i = 0; i < host->capacity; ++i) if (host->table[i]) close_connection(host->table[i]);
+ return NULL;
+}
+
+static void stop_shard(Host *host) {
+ pure_app_stop(&host->app);
+ pure_store_close(host->store);
+ if (host->listener >= 0) close(host->listener);
+ if (host->epoll >= 0) close(host->epoll);
+ while (host->closed) {
+ Connection *c = host->closed;
+ host->closed = c->next;
+ c->next = host->spare;
+ host->spare = c;
+ }
+ for (Connection *c = host->spare, *next; c; c = next) { next = c->next; free(c); }
+ free(host->work);
+ free(host->table);
+ free(host->by_slot);
+ free(host->applied);
+}
+
+int main(int argc, char **argv) {
+ static PureModule module;
+ const char *module_path = NULL, *store = NULL;
+ unsigned port = 8767, threads = 1;
+ size_t budget = 64u << 20, capacity = 1024;
+ uint64_t timeout_ms = 10000;
+ for (int i = 1; i < argc; i += 2) {
+ uint64_t value;
+ if (i + 1 == argc) goto usage;
+ if (!strcmp(argv[i], "--module")) module_path = argv[i + 1];
+ else if (!strcmp(argv[i], "--store")) store = argv[i + 1];
+ else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value;
+ else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value;
+ else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value;
+ else if (!strcmp(argv[i], "--threads") && number(argv[i + 1], 256, &value) && value) threads = (unsigned)value;
+ else if (!strcmp(argv[i], "--timeout-ms") && number(argv[i + 1], 3600000, &value) && value) timeout_ms = value;
+ else goto usage;
+ }
+ uint32_t show, edit;
+ if (!module_path || !store || !load_module(&module, module_path, &show, &edit)) goto usage;
+ if (threads > 1 && module.states) {
+ fputs("Module: its states have one owner, so it runs with --threads 1\n", stderr);
+ return 2;
+ }
+ if (pure_store_init(store) != PURE_OK) { fputs("Cannot initialize the store directory\n", stderr); return 1; }
+ signal(SIGPIPE, SIG_IGN);
+ struct sigaction stop_action = {.sa_handler = stop};
+ sigemptyset(&stop_action.sa_mask);
+ sigaction(SIGINT, &stop_action, NULL);
+ sigaction(SIGTERM, &stop_action, NULL);
+ Host *hosts = calloc(threads, sizeof(*hosts));
+ pthread_t *thread = calloc(threads, sizeof(*thread));
+ if (!hosts || !thread) return 1;
+ for (unsigned i = 0; i < threads; ++i) {
+ hosts[i].capacity = capacity;
+ hosts[i].timeout_ms = timeout_ms;
+ hosts[i].listener = hosts[i].epoll = -1;
+ if (!start_shard(&hosts[i], &module, show, edit, store, budget, &port)) return 1;
+ }
+ printf("{\"port\":%u,\"threads\":%u}\n", port, threads);
+ fflush(stdout);
+ for (unsigned i = 1; i < threads; ++i)
+ if (pthread_create(&thread[i], NULL, run_shard, &hosts[i])) { fputs("Cannot start a thread\n", stderr); return 1; }
+ run_shard(&hosts[0]);
+ for (unsigned i = 1; i < threads; ++i) pthread_join(thread[i], NULL);
+ size_t active = 0, retained = 0;
+ uint64_t accepted = 0, requests = 0, bursts = 0, events = 0, expired = 0;
+ for (unsigned i = 0; i < threads; ++i) {
+ active += hosts[i].active;
+ retained += hosts[i].runtime.retained;
+ accepted += hosts[i].accepted;
+ requests += hosts[i].requests;
+ bursts += hosts[i].bursts;
+ events += hosts[i].burst_events;
+ expired += hosts[i].expired;
+ stop_shard(&hosts[i]);
+ }
fprintf(stderr, "{\"active\":%zu,\"retained_payload_bytes\":%zu,\"accepted\":%" PRIu64 ",\"requests\":%" PRIu64
- ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"expired\":%" PRIu64 "}\n",
- host.active, host.runtime.retained, host.accepted, host.requests, host.bursts, host.burst_events,
- host.expired);
- pure_app_stop(&host.app);
- pure_store_close(handle);
- close(host.listener);
- close(host.epoll);
- for (Connection **list = (Connection *[]){host.spare, host.closed}, **end = list + 2; list < end; ++list)
- for (Connection *c = *list, *next; c; c = next) { next = c->next; free(c); }
- free(host.work);
- free(host.table);
- free(host.by_slot);
- free(host.applied);
+ ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"expired\":%" PRIu64 ",\"threads\":%u}\n",
+ active, retained, accepted, requests, bursts, events, expired, threads);
+ free(hosts);
+ free(thread);
return 0;
usage:
- fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--budget BYTES] [--connections 1024] "
- "[--timeout-ms 10000]\n", stderr);
+ fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1] [--budget BYTES] "
+ "[--connections 1024] [--timeout-ms 10000]\n", stderr);
return 2;
}
diff --git a/prototypes/connections/state.h b/prototypes/connections/state.h
index 25c0f61..60d333c 100644
--- a/prototypes/connections/state.h
+++ b/prototypes/connections/state.h
@@ -30,6 +30,10 @@ PureResult pure_store_open(const char *path, PureStore **out);
void pure_store_close(PureStore *);
PureResult pure_store_draft(PureStore *, const char *key, PureDraft **out);
PureResult pure_store_state(PureStore *, const char *name, PureDraft **out);
+/* Pinned, drafts open at the version found when pinning, without checking
+ * the file each time (a host pins for one burst of requests); commits made
+ * through the handle still move it on. */
+PureResult pure_store_pin(PureStore *, bool pinned);
/* Memory only. Fails without change on a bad range or when memory runs out. */
PureResult pure_draft_splice(PureDraft *, size_t offset, size_t remove, PureSpan insert);
size_t pure_draft_size(const PureDraft *);
diff --git a/prototypes/connections/store-tree.c b/prototypes/connections/store-tree.c
index c41393b..97add45 100644
--- a/prototypes/connections/store-tree.c
+++ b/prototypes/connections/store-tree.c
@@ -17,6 +17,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
+#include <pthread.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
@@ -36,6 +37,10 @@ enum { COMPACT_SLACK = 1048576 };
static const unsigned char head_magic[8] = {'P','U','R','E','T','R','E','1'};
static const unsigned char branch_magic[4] = {'P','T','B','1'};
static const char tree_name[] = "pure.tree";
+/* Threads of one process take turns as the writer: the file lock arbitrates
+ * only between processes (it is the process's, and closing any descriptor of
+ * the lock file drops it). */
+static pthread_mutex_t writers = PTHREAD_MUTEX_INITIALIZER;
static uint64_t temporary_sequence;
/* A reference is a file offset, or an index into this edit's node arena.
@@ -704,6 +709,7 @@ PureResult pure_store_init(const char *path) {
if (!created && errno != EEXIST) return PURE_STORE_ERROR;
int directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
if (directory < 0) return PURE_STORE_ERROR; /* Never reinterpret an old DB file. */
+ pthread_mutex_lock(&writers);
int lock = openat(directory, ".writer.lock", O_RDWR | O_CREAT | O_NOFOLLOW | O_CLOEXEC, 0600);
struct stat info;
bool ok = lock >= 0 && fstat(lock, &info) == 0 && S_ISREG(info.st_mode) && lock_writer(lock, true);
@@ -724,6 +730,7 @@ PureResult pure_store_init(const char *path) {
}
ok = ok && sync_fd(lock) && sync_fd(directory);
if (lock >= 0) close(lock);
+ pthread_mutex_unlock(&writers);
close(directory);
if (ok && created) {
char *parent = strdup(path);
@@ -827,6 +834,7 @@ typedef struct {
Head head;
Tree tree;
struct stat file;
+ bool turn; /* holds this process's writer turn */
} Session;
static void close_session(Session *session) {
@@ -834,10 +842,14 @@ static void close_session(Session *session) {
if (session->fd >= 0) close(session->fd);
if (session->lock >= 0) close(session->lock);
if (session->directory >= 0) close(session->directory);
+ if (session->turn) pthread_mutex_unlock(&writers);
+ session->turn = false;
}
static PureResult open_session(const char *path, Session *session) {
*session = (Session){.directory = -1, .lock = -1, .fd = -1};
+ pthread_mutex_lock(&writers); /* another thread of this process waits, not fails */
+ session->turn = true;
session->directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
if (session->directory < 0) return PURE_STORE_ERROR;
session->lock = openat(session->directory, ".writer.lock", O_RDWR | O_NOFOLLOW | O_CLOEXEC);
@@ -972,6 +984,7 @@ struct PureStore {
ino_t inode;
Head head;
Mapping *mapping;
+ bool pinned; /* drafts open at the version of the last check */
};
struct PureDraft {
@@ -1120,9 +1133,16 @@ void pure_store_close(PureStore *store) {
free(store);
}
+PureResult pure_store_pin(PureStore *store, bool pinned) {
+ store->pinned = false;
+ PureResult result = pinned ? refresh(store) : PURE_OK;
+ store->pinned = pinned && result == PURE_OK;
+ return result;
+}
+
static PureResult open_through(PureStore *store, const char *key, PureDraft **out) {
*out = NULL;
- PureResult result = refresh(store);
+ PureResult result = store->pinned ? PURE_OK : refresh(store);
if (result != PURE_OK) return result;
PureDraft *draft = calloc(1, sizeof(*draft));
if (!draft) return PURE_ALLOCATION;
diff --git a/prototypes/connections/test-tree-store.c b/prototypes/connections/test-tree-store.c
index 3f6d2f0..bc8ba4b 100644
--- a/prototypes/connections/test-tree-store.c
+++ b/prototypes/connections/test-tree-store.c
@@ -8,6 +8,7 @@
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
+#include <pthread.h>
#include <sys/wait.h>
#include <unistd.h>
@@ -824,6 +825,61 @@ static void store_handle(const char *path) {
pure_store_close(store);
}
+/* Threads of one process commit through their own handles at once: none is
+ * refused as busy, every commit lands, and edits of one shared key conflict
+ * instead of overwriting each other. */
+typedef struct { const char *path; unsigned number, conflicts; } Committer;
+
+static void *commit_many(void *context) {
+ Committer *c = context;
+ PureStore *store;
+ CHECK(pure_store_open(c->path, &store) == PURE_OK);
+ char key[16];
+ snprintf(key, sizeof(key), "t%u", c->number);
+ for (unsigned i = 0; i < 40; ++i) {
+ PureDraft *own, *shared;
+ int64_t revisions[2];
+ CHECK(pure_store_draft(store, key, &own) == PURE_OK && pure_store_draft(store, "shared", &shared) == PURE_OK);
+ CHECK(pure_draft_splice(own, pure_draft_size(own), 0, (PureSpan){(const unsigned char *)"x", 1}) == PURE_OK);
+ CHECK(pure_draft_splice(shared, pure_draft_size(shared), 0, (PureSpan){(const unsigned char *)"y", 1}) == PURE_OK);
+ PureDraft *batch[2] = {own, shared};
+ PureResult result = pure_draft_commit_all(batch, 2, revisions);
+ CHECK(result == PURE_OK || result == PURE_CONFLICT);
+ if (result == PURE_CONFLICT) {
+ ++c->conflicts;
+ CHECK(pure_draft_commit(own, &revisions[0]) == PURE_OK);
+ }
+ pure_draft_close(own);
+ pure_draft_close(shared);
+ }
+ pure_store_close(store);
+ return NULL;
+}
+
+static void threads(const char *path) {
+ pthread_t thread[4];
+ Committer committer[4];
+ for (unsigned i = 0; i < 4; ++i) {
+ committer[i] = (Committer){path, i, 0};
+ CHECK(pthread_create(&thread[i], NULL, commit_many, &committer[i]) == 0);
+ }
+ unsigned conflicts = 0;
+ for (unsigned i = 0; i < 4; ++i) {
+ CHECK(pthread_join(thread[i], NULL) == 0);
+ conflicts += committer[i].conflicts;
+ }
+ PureStoredView view;
+ char key[16];
+ for (unsigned i = 0; i < 4; ++i) {
+ snprintf(key, sizeof(key), "t%u", i);
+ CHECK(pure_store_get(path, key, PURE_OBJECT_LIMIT, &view) == PURE_OK && view.body.size == 40 && view.revision == 40);
+ pure_store_release(&view);
+ }
+ CHECK(pure_store_get(path, "shared", PURE_OBJECT_LIMIT, &view) == PURE_OK);
+ CHECK(view.body.size == 160 - conflicts && view.revision == (int64_t)(160 - conflicts));
+ pure_store_release(&view);
+}
+
static void clean(const char *path) {
DIR *entries = opendir(path);
CHECK(entries);
@@ -840,7 +896,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};
+ draft_versions, anonymous_draft, batch, store_handle, threads};
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);
@@ -851,6 +907,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\"}");
+ "atomic-batch-runtime-keys-store-handle-threads\"}");
return 0;
}
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 16:34:31 +0000