actors: worker memory is copy-on-write against a snapshot of the parent (memfd/shm), o3 entry tables handed over sparse

1 files changed, 100 insertions, 10 deletions

diff --git a/src/core/core.c b/src/core/core.c
index ce9e5ca..af7bba7 100644
--- a/src/core/core.c
+++ b/src/core/core.c
@@ -7363,6 +7363,85 @@ static int code_range_live(Cell start, Cell count)
 
 /* each worker owns a host thread, vm memory and translation state. */
 #if !defined(_WIN32) && !defined(__wasm__)
+#include <sys/mman.h>
+#ifdef __linux__
+#include <sys/syscall.h>
+#endif
+/* vm memory for a worker is copy-on-write against a snapshot of its parent.
+   the first spawn writes the parent's live words [0, memory_words) to an anonymous
+   shared file and maps the parent's memory privately over it at the same address,
+   so the parent reads the same pages and its later writes stay its own. a child maps
+   the same file privately, so code, genome and everything the parent has not
+   touched since are one set of physical pages for all of them, and only the pages
+   the parent changed since the snapshot are copied in. the file never changes, so a
+   child never sees a later write of its parent. a parent whose memory has drifted
+   far from its snapshot takes a fresh one. without a file (or on failure) a spawn
+   falls back to a private copy. */
+typedef struct { int fd; Cell *view; size_t words; } MemSnap;
+static VM_TLS MemSnap mem_snap = { -1, 0, 0 };
+static VM_TLS int memory_mapped;	/* m came from vm_memory_map */
+#define SNAP_PAGE 512	/* words per 4 KB page */
+static Cell *vm_memory_map(void)
+{
+	void *p = mmap(0, (size_t)MW * sizeof(Cell), PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON | MAP_NORESERVE, -1, 0);
+	return p == MAP_FAILED ? 0 : p;
+}
+static void vm_memory_unmap(Cell *p) { if (p) munmap(p, (size_t)MW * sizeof(Cell)); }
+static int snap_file(size_t bytes)
+{
+	int fd = -1;
+#ifdef __linux__
+#ifdef SYS_memfd_create
+	fd = (int)syscall(SYS_memfd_create, "zy-snapshot", 1u /* MFD_CLOEXEC */);
+#endif
+#else
+	char name[64]; snprintf(name, sizeof name, "/zy-%d-%p", (int)getpid(), (void *)&name);
+	fd = shm_open(name, O_RDWR | O_CREAT | O_EXCL, 0600); if (fd >= 0) shm_unlink(name);
+#endif
+	if (fd < 0) return -1;
+	if (ftruncate(fd, (off_t)bytes) < 0) { close(fd); return -1; }
+	return fd;
+}
+static void snap_drop(void)
+{
+	if (mem_snap.view) munmap(mem_snap.view, mem_snap.words * sizeof(Cell));
+	if (mem_snap.fd >= 0) close(mem_snap.fd);
+	mem_snap.fd = -1; mem_snap.view = 0; mem_snap.words = 0;
+}
+/* pages of [0, words) where m differs from the snapshot */
+static size_t snap_dirty(size_t words)
+{
+	size_t p, n = 0, lim = words < mem_snap.words ? words : mem_snap.words;
+	for (p = 0; p < lim; p += SNAP_PAGE) if (memcmp(m + p, mem_snap.view + p, SNAP_PAGE * sizeof(Cell))) n++;
+	return n + (words > mem_snap.words ? (words - mem_snap.words) / SNAP_PAGE : 0);
+}
+static int snap_take(void)
+{
+	size_t words = ((size_t)memory_words + SNAP_PAGE - 1) / SNAP_PAGE * SNAP_PAGE, bytes = words * sizeof(Cell), done = 0;
+	int fd; void *view, *over;
+	if (words > (size_t)MW) return -1;
+	fd = snap_file(bytes); if (fd < 0) return -1;
+	while (done < bytes) { ssize_t k = pwrite(fd, (char *)m + done, bytes - done, (off_t)done); if (k <= 0) { close(fd); return -1; } done += (size_t)k; }
+	view = mmap(0, bytes, PROT_READ, MAP_SHARED, fd, 0);
+	if (view == MAP_FAILED) { close(fd); return -1; }
+	over = mmap(m, bytes, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_FIXED | MAP_NORESERVE, fd, 0);
+	if (over == MAP_FAILED) { munmap(view, bytes); close(fd); return -1; }	/* m is untouched when the map fails */
+	snap_drop(); mem_snap.fd = fd; mem_snap.view = view; mem_snap.words = words; return 0;
+}
+/* a child's memory: the snapshot, plus the pages the parent changed since */
+static Cell *snap_clone(void)
+{
+	size_t p, words = (size_t)memory_words;
+	Cell *c;
+	if (mem_snap.fd < 0 || snap_dirty(words) * 4 > mem_snap.words / SNAP_PAGE) { if (snap_take() < 0) return 0; }
+	c = vm_memory_map(); if (!c) return 0;
+	if (mmap(c, mem_snap.words * sizeof(Cell), PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_FIXED | MAP_NORESERVE, mem_snap.fd, 0) == MAP_FAILED) { vm_memory_unmap(c); return 0; }
+	for (p = 0; p < words; p += SNAP_PAGE) {
+		size_t k = words - p < SNAP_PAGE ? words - p : SNAP_PAGE;
+		if (p >= mem_snap.words || memcmp(m + p, mem_snap.view + p, k * sizeof(Cell))) memcpy(c + p, m + p, k * sizeof(Cell));
+	}
+	return c;
+}
 typedef struct ContextChannel { Cell handle; int fd; struct ContextChannel *next; } ContextChannel;
 static VM_TLS ContextChannel *context_channels;
 static atomic_llong context_id_sequence;
@@ -7374,6 +7453,7 @@ struct VmContext {
  Cell token, entry, sp, rp, stack, words, limit, growth, code_size;
  Cell frame_floor, frame_top, frame_guard;
  Cell *memory;
+ int memory_mapped;
  int exit_code, requested_exit;
  void *input_snapshot;
  int argc, envc;
@@ -7387,6 +7467,7 @@ struct VmContext {
  int region_count;
  unsigned short *entries;
  unsigned char *entry_indices;
+ int *entry_at, entry_count;
 };
 static int context_descriptor(Cell handle) {
  ContextChannel *channel;
@@ -7405,7 +7486,7 @@ static void context_snapshot_free(VmContext *context) {
  int i;
  for (i = 0; i < NENH; i++) { free(context->enhancements[i].u); free(context->enhancements[i].reg); }
  for (i = 0; i < context->region_count; i++) free(context->regions[i]);
- free(context->input_snapshot); free(context->entries); free(context->entry_indices); if (context->memory) { free(context->memory); atomic_fetch_sub(&context_memory_live, 1); } context->memory = 0;
+ free(context->input_snapshot); free(context->entries); free(context->entry_indices); free(context->entry_at); if (context->memory) { if (context->memory_mapped) vm_memory_unmap(context->memory); else free(context->memory); atomic_fetch_sub(&context_memory_live, 1); } context->memory = 0;
 }
 static int context_cancelled(void) { return context_current && atomic_load_explicit(&context_current->stop, memory_order_acquire); }
 static void context_exit(int value) { context_current->exit_code = value; context_current->requested_exit = 1; atomic_store_explicit(&context_current->stop, 1, memory_order_release); }
@@ -7425,7 +7506,7 @@ static void context_engine_free(void) {
  free(lcov); free(lt); free(hot);
 #endif
 #endif
- free(m); m = 0; atomic_fetch_sub(&context_memory_live, 1);
+ if (memory_mapped) vm_memory_unmap(m); else free(m); m = 0; snap_drop(); atomic_fetch_sub(&context_memory_live, 1);
 }
 static void *context_worker(void *opaque) {
  VmContext *context = opaque;
@@ -7434,7 +7515,7 @@ static void *context_worker(void *opaque) {
  optimizer_storage();
  context_current = context; io_context_restore(context->input_snapshot); free(context->input_snapshot); context->input_snapshot = 0; trace = getenv("ZYTRACE") != 0;
  context->worker_channel.next = context_channels; context_channels = &context->worker_channel;
- m = context->memory; context->memory = 0;
+ m = context->memory; context->memory = 0; memory_mapped = context->memory_mapped;
  memory_limit = (int)context->limit; memory_words = (int)context->words; memory_growth_limit = (int)context->growth;
  stack_control = context->stack; frame_floor_word = context->frame_floor; frame_top_word = context->frame_top; frame_guard_word = context->frame_guard;
  zy_argc = context->argc; zy_argv = context->argv; zy_envc = context->envc; zy_envp = context->envp;
@@ -7463,8 +7544,8 @@ static void *context_worker(void *opaque) {
  memcpy(enh, context->enhancements, sizeof enh); memset(context->enhancements, 0, sizeof context->enhancements);
  for (i = 0; i < context->region_count; i++) { rgs[i] = context->regions[i]; context->regions[i] = 0; }
  nrg = context->region_count; context->region_count = 0;
- memcpy(o3ent, context->entries, (size_t)code_words * sizeof *o3ent); memcpy(o3ei, context->entry_indices, (size_t)code_words);
- free(context->entries); context->entries = 0; free(context->entry_indices); context->entry_indices = 0;
+ for (i = 0; i < context->entry_count; i++) { o3ent[context->entry_at[i]] = context->entries[i]; o3ei[context->entry_at[i]] = context->entry_indices[i]; }
+ free(context->entries); context->entries = 0; free(context->entry_indices); context->entry_indices = 0; free(context->entry_at); context->entry_at = 0;
 #ifdef ZYJIT_X64
  if (x64_boot(sp, rp, &context->stop)) { context->exit_code = -1; context_channel_close(context->worker_channel.handle); context_engine_free(); return 0; }
  atomic_fetch_add(&context_code_live, 1);
@@ -7506,12 +7587,17 @@ static Cell context_start(Cell entry, Cell channel) {
  context->parent_channel.fd = fd[0]; context->worker_channel.fd = fd[1]; atomic_init(&context->stop, 0);
  context->input_snapshot = malloc(io_context_size());
  if (!context->input_snapshot) goto failed; io_context_snapshot(context->input_snapshot);
- context->memory = calloc((size_t)memory_limit, sizeof(Cell));
+ context->memory = snap_clone(); context->memory_mapped = context->memory != 0;
+ if (!context->memory) context->memory = calloc((size_t)memory_limit, sizeof(Cell));
  if (context->memory) atomic_fetch_add(&context_memory_live, 1);
- context->entries = malloc((size_t)code_words * sizeof *o3ent); context->entry_indices = malloc((size_t)code_words);
- if (!context->memory || !context->entries || !context->entry_indices) goto failed;
- saved = m[channel]; m[channel] = context->worker_channel.handle; memcpy(context->memory, m, (size_t)memory_words * sizeof(Cell)); m[channel] = saved;
- memcpy(context->entries, o3ent, (size_t)code_words * sizeof *o3ent); memcpy(context->entry_indices, o3ei, (size_t)code_words);
+ /* o3 entry points are few and the tables mostly zero: hand over only the set ones */
+ { Cell k; context->entry_count = 0; for (k = 0; k < code_words; k++) if (o3ent[k]) context->entry_count++; }
+ context->entry_at = malloc((size_t)(context->entry_count + 1) * sizeof *context->entry_at);
+ context->entries = malloc((size_t)(context->entry_count + 1) * sizeof *o3ent); context->entry_indices = malloc((size_t)(context->entry_count + 1));
+ if (!context->memory || !context->entries || !context->entry_indices || !context->entry_at) goto failed;
+ if (context->memory_mapped) context->memory[channel] = context->worker_channel.handle;
+ else { saved = m[channel]; m[channel] = context->worker_channel.handle; memcpy(context->memory, m, (size_t)memory_words * sizeof(Cell)); m[channel] = saved; }
+ { Cell k; int j = 0; for (k = 0; k < code_words; k++) if (o3ent[k]) { context->entry_at[j] = (int)k; context->entries[j] = o3ent[k]; context->entry_indices[j] = o3ei[k]; j++; } }
  for (i = 0; i < NENH; i++) {
   context->enhancements[i] = enh[i]; context->enhancements[i].u = 0; context->enhancements[i].reg = 0;
   if (enh[i].u) { context->enhancements[i].u = malloc((size_t)(enh[i].n + 1) * sizeof(Uop)); if (!context->enhancements[i].u) goto failed; memcpy(context->enhancements[i].u, enh[i].u, (size_t)(enh[i].n + 1) * sizeof(Uop)); }
@@ -7603,6 +7689,10 @@ int main(int argc, char **argv)
 	memory_words = memory_option("TINYVM_MEMORY_WORDS", HEAP_HI > PW ? HEAP_HI : PW, HEAP_HI > PW ? HEAP_HI : PW);
 	if (memory_words < 0) return 1;
 	if (memory_words > memory_limit) { fprintf(stderr, "zy: initial memory exceeds its limit\n"); return 1; }
+#if !defined(_WIN32) && !defined(__wasm__)
+	m = vm_memory_map(); memory_mapped = m != 0;
+	if (!m)
+#endif
 	m = calloc((size_t)MW, sizeof *m);
 #ifdef ZYNATIVE
 	cov = calloc((size_t)MW, 1);