merge work/actors: wasm seed bootstrap (seed/MANIFEST), shared copy-on-write actor memory, top-level hold blocks on all backends and lsp

3 files changed, 132 insertions, 15 deletions

diff --git a/src/micro/micro.c b/src/micro/micro.c
index 23994dd..e095451 100644
--- a/src/micro/micro.c
+++ b/src/micro/micro.c
@@ -34,6 +34,10 @@
 #include "micro_port.h"
 #include "fops.h"
 #include "bops.h"
+#ifdef ZY_THREADS
+#define HOLD_THREADS
+#endif
+#include "hold.h"
 #define NENH 512
 #define XOP0 22
 #define XPOOL 4096
@@ -537,6 +541,11 @@ static int run_from(int p, int s, int r)
 					/* poll: a streaming port suspends once, so its host can deliver input that is waiting */
 					result = micro_syscall(m, b, 0, 0, 0);
 					if (result == MICRO_WAIT) { sus_p = p - 1; sus_s = s; sus_r = r; return MICRO_SUSPENDED; }
+				} else if (b == 57) {
+					result = hold_enter(m, mw, m[a + 1], m[a + 2]);
+					if (result == -2) { micro_msg("zy: a hold block was entered while a later one was held; hold blocks are entered in the order they are declared\n"); context_call(7, 2, 0, 0, s + 2, r); return 1; }	/* a trap: exit status 2 */
+				} else if (b == 58) {
+					result = hold_leave(m, mw, m[a + 1], m[a + 2]);
 				} else result = context_call(b, m[a + 1], m[a + 2], m[a + 3], s + 2, r);
 				m[a] = result;
 				s += 2;
@@ -647,6 +656,11 @@ extern size_t micro_context_port_size(void);
 extern void micro_context_port_save(void *);
 extern void micro_context_port_restore(const void *);
 extern void micro_context_port_free(void);
+#ifndef _WIN32
+extern Cell *micro_mem_clone(void);
+extern void micro_mem_adopt(Cell *mem, UCell words);
+extern void micro_mem_release(Cell *mem);
+#endif
 #define CONTEXT_HANDLE_LO 0x40000000
 typedef struct ContextChannel { Cell handle; int fd; struct ContextChannel *next; } ContextChannel;
 typedef struct VmContext {
@@ -658,6 +672,7 @@ typedef struct VmContext {
  UCell words, limit, initial;
  int sp, rp, exit_code, requested_exit, enhancement_count, input_flag, input_known;
  Cell *memory;
+ int memory_mapped;
  void *port;
  Xop operations[XPOOL];
  Enh enhancements[NENH];
@@ -683,15 +698,26 @@ static int context_close(Cell handle) {
 }
 static void context_snapshot_free(VmContext *context) {
  free(context->port);
- if (context->memory) { free(context->memory); atomic_fetch_sub(&context_memory_live, 1); }
+ if (context->memory) {
+#ifndef _WIN32
+  if (context->memory_mapped) micro_mem_release(context->memory); else
+#endif
+  free(context->memory);
+  atomic_fetch_sub(&context_memory_live, 1);
+ }
 }
 static void *context_worker(void *opaque) {
  VmContext *context = opaque;
  context_current = context; context->worker_channel.next = context_channels; context_channels = &context->worker_channel;
  micro_context_port_restore(context->port); free(context->port); context->port = 0;
  memory_limit = context->limit; memory_initial = context->initial;
+ if (context->memory_mapped) {
+#ifndef _WIN32
+  micro_mem_adopt(context->memory, context->words); context->memory = 0;
+#endif
+ }
  if (!mem_grow_words(context->words)) { context->exit_code = -1; context_close(context->worker_channel.handle); return 0; }
- memcpy(m, context->memory, (size_t)context->words * sizeof(Cell)); free(context->memory); context->memory = 0;
+ if (context->memory) { memcpy(m, context->memory, (size_t)context->words * sizeof(Cell)); free(context->memory); context->memory = 0; }
  stack_control = context->stack; jflag = context->input_flag; jknown = context->input_known;
  xuse = context->enhancement_count; memcpy(xops, context->operations, sizeof xops); memcpy(enh, context->enhancements, sizeof enh);
  int sp = context->sp, rp = context->rp; Cell entry = context->entry;
@@ -701,7 +727,7 @@ static void *context_worker(void *opaque) {
  if (!context->requested_exit) context->exit_code = context_cancelled() ? 0 : result ? -1 : inmem(final_s) ? (int)m[final_s] : 0;
  context_close(context->worker_channel.handle);
  while (context_children) context_join(context_children->token, 1);
- micro_context_port_free(); m = 0; atomic_fetch_sub(&context_memory_live, 1); context_current = 0; return 0;
+ hold_release_all(); micro_context_port_free(); m = 0; atomic_fetch_sub(&context_memory_live, 1); context_current = 0; return 0;
 }
 static Cell context_start(Cell entry, Cell channel, int sp, int rp) {
  if (!inmem(entry) || !inmem(channel) || (entry >= HEAP_LO && (!inmem(entry - 3) || m[entry - 3] != (entry ^ 0x2B7E1516)))) return -1;
@@ -714,12 +740,17 @@ static Cell context_start(Cell entry, Cell channel, int sp, int rp) {
  context->parent_channel.handle = CONTEXT_HANDLE_LO + atomic_fetch_add(&context_sequence, 1) + 1;
  context->worker_channel.handle = CONTEXT_HANDLE_LO + atomic_fetch_add(&context_sequence, 1) + 1;
  context->parent_channel.fd = fd[0]; context->worker_channel.fd = fd[1]; atomic_init(&context->stop, 0);
- context->port = malloc(micro_context_port_size()); context->memory = malloc((size_t)mw * sizeof(Cell));
+ context->port = malloc(micro_context_port_size());
+#ifndef _WIN32
+ context->memory = micro_mem_clone(); context->memory_mapped = context->memory != 0;
+ if (!context->memory)
+#endif
+ context->memory = malloc((size_t)mw * sizeof(Cell));
  if (context->memory) atomic_fetch_add(&context_memory_live, 1);
  if (!context->port || !context->memory) goto failed;
- micro_context_port_save(context->port); memcpy(context->memory, m, (size_t)mw * sizeof(Cell)); context->memory[channel] = context->worker_channel.handle;
+ micro_context_port_save(context->port); if (!context->memory_mapped) memcpy(context->memory, m, (size_t)mw * sizeof(Cell)); context->memory[channel] = context->worker_channel.handle;
  memcpy(context->operations, xops, sizeof xops); memcpy(context->enhancements, enh, sizeof enh);
- pthread_attr_t attributes; pthread_attr_init(&attributes); pthread_attr_setstacksize(&attributes, 16u << 20);
+ pthread_attr_t attributes; pthread_attr_init(&attributes); pthread_attr_setstacksize(&attributes, 4u << 20);	/* the vm keeps its own stacks in vm memory; 4 MB of address space covers the deepest native recursion (jit compile, o3) */
  int result = pthread_create(&context->thread, &attributes, context_worker, context); pthread_attr_destroy(&attributes);
  if (result) goto failed;
  context->parent_channel.next = context_channels; context_channels = &context->parent_channel;
@@ -776,6 +807,13 @@ int micro_run(void)
 		micro_msg("zy: genome does not fit\n");
 		return 1;
 	}
+#ifdef MICRO_PORT_IMAGE
+	/* the port may hand over an image of its own in place of the embedded one
+	   (the wasm bootstrap seed does, to run a freshly built compiler) */
+	i = micro_port_image(m, (UCell)GORG, (UCell)FBOT);
+	if (i < 0) { micro_msg("zy: bad image\n"); return 1; }
+	if (i == 0)
+#endif
 	for (i = 0; i < GLEN; i++) m[GORG + i] = genome_word(i);
 	jknown = 0;
 	return run(GORG);
diff --git a/src/micro/micro_port.h b/src/micro/micro_port.h
index 9f77987..1653250 100644
--- a/src/micro/micro_port.h
+++ b/src/micro/micro_port.h
@@ -56,6 +56,9 @@ int micro_eof(void);
 #define MICRO_WAIT (-2)
 #define MICRO_SUSPENDED 3
 int micro_run(void);
+/* with MICRO_PORT_IMAGE defined, the port's micro_port_image(mem, org, end) may load an
+   image of its own at mem[org..end): 1 when it did, 0 to use the embedded genome, -1 on error. */
+int micro_port_image(Cell *mem, UCell org, UCell end);
 int micro_resume(void);
 
 /* write the low 8 bits. */
diff --git a/src/micro/port_linux.c b/src/micro/port_linux.c
index 6245f4f..a39e3e8 100644
--- a/src/micro/port_linux.c
+++ b/src/micro/port_linux.c
@@ -38,16 +38,38 @@
 #define VM_TLS
 #endif
 
-/* the machine's memory: a block that grows by realloc, zero in the new part. the
-   base can move on a grow, which micro.c takes into account (it re-reads m). */
+/* the machine's memory. with threads (actors) on a posix host it is one large
+   reservation that grows in place, so that a worker's memory can be a
+   copy-on-write map of a snapshot of its parent (micro_mem_clone); a machine that
+   outgrows the reservation moves to a block that grows by realloc. without threads it
+   is a block that grows by realloc, zero in the new part. the base can move on a
+   grow, which micro.c takes into account (it re-reads m). */
 static VM_TLS Cell *lmem;
 static VM_TLS UCell lcap;
+#if defined(ZY_THREADS) && !defined(_WIN32)
+#include <sys/mman.h>
+#ifdef __linux__
+#include <sys/syscall.h>
+#endif
+#define RESW ((UCell)1 << 28)	/* words reserved per machine: 2 GB of address space */
+static VM_TLS int lres;	/* lmem is a reservation of RESW words */
+static Cell *res_map(void) { void *p = mmap(0, RESW * sizeof(Cell), PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON | MAP_NORESERVE, -1, 0); return p == MAP_FAILED ? 0 : p; }
+#endif
 
 Cell *micro_mem_alloc(UCell words)
 {
 	Cell *n;
 	if (words <= lcap && lmem) return lmem;
 	if (words > ((UCell)-1) / sizeof(Cell)) return NULL;
+#if defined(ZY_THREADS) && !defined(_WIN32)
+	if (!lmem && words <= RESW && (lmem = res_map())) { lres = 1; lcap = words; return lmem; }
+	if (lres && words <= RESW) { lcap = words; return lmem; }	/* the reservation reads zero past lcap */
+	if (lres) {
+		n = malloc(words * sizeof(Cell)); if (!n) return NULL;
+		memcpy(n, lmem, lcap * sizeof(Cell)); memset(n + lcap, 0, (words - lcap) * sizeof(Cell));
+		munmap(lmem, RESW * sizeof(Cell)); lres = 0; lmem = n; lcap = words; return n;
+	}
+#endif
 	n = realloc(lmem, words * sizeof(Cell));
 	if (!n) return NULL;
 	memset(n + lcap, 0, (words - lcap) * sizeof(Cell));
@@ -56,15 +78,62 @@ Cell *micro_mem_alloc(UCell words)
 	return n;
 }
 
+#if defined(ZY_THREADS) && !defined(_WIN32)
+/* a worker's memory is a private map of an unchanging snapshot of its parent's,
+   plus the pages the parent changed since; see the same scheme in src/core/core.c */
+#define SNAP_PAGE 512
+static VM_TLS int snap_fd = -1; static VM_TLS Cell *snap_view; static VM_TLS UCell snap_words;
+static void snap_drop(void) { if (snap_view) munmap(snap_view, snap_words * sizeof(Cell)); if (snap_fd >= 0) close(snap_fd); snap_fd = -1; snap_view = 0; snap_words = 0; }
+static int snap_take(void)
+{
+	UCell words = (lcap + SNAP_PAGE - 1) / SNAP_PAGE * SNAP_PAGE; size_t bytes = words * sizeof(Cell), done = 0; int fd = -1; void *view;
+#ifdef __linux__
+#ifdef SYS_memfd_create
+	fd = (int)syscall(SYS_memfd_create, "zym-snapshot", 1u);
+#endif
+#else
+	{ char name[64]; snprintf(name, sizeof name, "/zym-%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; }
+	while (done < bytes) { ssize_t k = pwrite(fd, (char *)lmem + 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; }
+	if (mmap(lmem, bytes, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_FIXED | MAP_NORESERVE, fd, 0) == MAP_FAILED) { munmap(view, bytes); close(fd); return -1; }
+	snap_drop(); snap_fd = fd; snap_view = view; snap_words = words; return 0;
+}
+Cell *micro_mem_clone(void)
+{
+	UCell p, dirty = 0; Cell *c;
+	if (!lres || !lmem) return 0;
+	if (snap_fd >= 0) {
+		for (p = 0; p < lcap; p += SNAP_PAGE) if (p >= snap_words || memcmp(lmem + p, snap_view + p, SNAP_PAGE * sizeof(Cell))) dirty++;
+		if (dirty * 4 > snap_words / SNAP_PAGE) snap_drop();
+	}
+	if (snap_fd < 0 && snap_take() < 0) return 0;
+	c = res_map(); if (!c) return 0;
+	if (mmap(c, snap_words * sizeof(Cell), PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_FIXED | MAP_NORESERVE, snap_fd, 0) == MAP_FAILED) { munmap(c, RESW * sizeof(Cell)); return 0; }
+	for (p = 0; p < lcap; p += SNAP_PAGE) {
+		UCell k = lcap - p < SNAP_PAGE ? lcap - p : SNAP_PAGE;
+		if (p >= snap_words || memcmp(lmem + p, snap_view + p, k * sizeof(Cell))) memcpy(c + p, lmem + p, k * sizeof(Cell));
+	}
+	return c;
+}
+void micro_mem_adopt(Cell *mem, UCell words) { lmem = mem; lcap = words; lres = 1; }
+void micro_mem_release(Cell *mem) { if (mem) munmap(mem, RESW * sizeof(Cell)); }
+#endif
+
 #if defined(ZY_THREADS)
-static VM_TLS unsigned char input_bytes[65536];
+/* allocated when the machine first reads, so a worker that never reads has none */
+static VM_TLS unsigned char *input_buffer;
+#define input_bytes (input_buffer ? input_buffer : (input_buffer = malloc(65536)) ? input_buffer : (fputs("zym: out of memory\n", stderr), exit(2), input_buffer))
 static VM_TLS int input_position, input_count, input_ended;
 static int input_fill(int wait) {
  if (input_position < input_count) return 1;
  if (input_ended) return 0;
  if (!wait) { struct pollfd descriptor = {0, POLLIN, 0}; if (poll(&descriptor, 1, 0) <= 0) return 0; }
  else fflush(stdout);
- int n = (int)read(0, input_bytes, sizeof input_bytes);
+ int n = (int)read(0, input_bytes, 65536);
  if (n <= 0) { input_ended = 1; return 0; }
  input_position = 0; input_count = n; return 1;
 }
@@ -392,22 +461,29 @@ int main(int argc, char **argv) {
 
 #ifdef ZY_THREADS
 struct PortSnapshot {
- unsigned char bytes[65536];
  int position, count, ended, argc, envc;
  char **argv, **envp;
+ unsigned char bytes[];	/* the input read ahead and not consumed yet */
 };
-size_t micro_context_port_size(void) { return sizeof(struct PortSnapshot); }
+size_t micro_context_port_size(void) { return sizeof(struct PortSnapshot) + (size_t)(input_position < input_count ? input_count - input_position : 0); }
 void micro_context_port_save(void *destination) {
  struct PortSnapshot *snapshot = destination;
- memcpy(snapshot->bytes, input_bytes, sizeof input_bytes);
+ if (input_position < input_count) memcpy(snapshot->bytes, input_bytes + input_position, (size_t)(input_count - input_position));
  snapshot->position = input_position; snapshot->count = input_count; snapshot->ended = input_ended;
  snapshot->argc = zy_argc; snapshot->argv = zy_argv; snapshot->envc = zy_envc; snapshot->envp = zy_envp;
 }
 void micro_context_port_restore(const void *source) {
  const struct PortSnapshot *snapshot = source;
- memcpy(input_bytes, snapshot->bytes, sizeof input_bytes);
  input_position = snapshot->position; input_count = snapshot->count; input_ended = snapshot->ended;
+ if (input_position < input_count) memcpy(input_bytes + input_position, snapshot->bytes, (size_t)(input_count - input_position));
  zy_argc = snapshot->argc; zy_argv = snapshot->argv; zy_envc = snapshot->envc; zy_envp = snapshot->envp;
 }
-void micro_context_port_free(void) { free(lmem); lmem = 0; lcap = 0; }
+void micro_context_port_free(void) {
+ free(input_buffer); input_buffer = 0;
+#ifndef _WIN32
+ if (lres) { munmap(lmem, RESW * sizeof(Cell)); lres = 0; snap_drop(); } else
+#endif
+ free(lmem);
+ lmem = 0; lcap = 0;
+}
 #endif