diff options
Diffstat (limited to 'src/micro/micro.c')
| -rw-r--r-- | src/micro/micro.c | 172 |
1 files changed, 157 insertions, 15 deletions
diff --git a/src/micro/micro.c b/src/micro/micro.c index 4542bd8..528382a 100644 --- a/src/micro/micro.c +++ b/src/micro/micro.c @@ -19,6 +19,17 @@ words). a smaller board compiles this file against the headers src/tools/genmap writes, and against the genome built for that same map. */ #include <stdint.h> +#ifdef ZY_THREADS +#include <pthread.h> +#include <stdatomic.h> +#include <stdlib.h> +#include <string.h> +#include <unistd.h> +#include <sys/socket.h> +#define VM_TLS _Thread_local +#else +#define VM_TLS +#endif #ifdef USE_LSP_GENOME #include "lsp_genome.h" #elif defined(GENOME_H) @@ -68,22 +79,22 @@ typedef struct { /* the machine: m is mw words, and grows on demand through the port (micro_mem_alloc). every access goes through m and mw, which the port's memory may move under. */ -static Cell *m; -static UCell mw; -static UCell memory_initial = MICRO_MW_INIT, memory_limit = MICRO_MW_LIMIT; +static VM_TLS Cell *m; +static VM_TLS UCell mw; +static VM_TLS UCell memory_initial = MICRO_MW_INIT, memory_limit = MICRO_MW_LIMIT; int micro_memory_config(UCell initial, UCell limit) { if (m || initial < MICRO_MW_INIT || limit < initial || limit > 0x7fffffff) return 0; memory_initial = initial; memory_limit = limit; return 1; } -static Xop xops[XPOOL]; -static int xuse; -static Enh enh[NENH]; -static Xop build[MAXOPS]; -static int bn, bnv, berr; -static Cell xv[MAXV]; -static int jflag; -static int jknown; /* 0 until something asks for the end-of-input flag: the probe reads the port, so it waits for the first ask */ +static VM_TLS Xop xops[XPOOL]; +static VM_TLS int xuse; +static VM_TLS Enh enh[NENH]; +static VM_TLS Xop build[MAXOPS]; +static VM_TLS int bn, bnv, berr; +static VM_TLS Cell xv[MAXV]; +static VM_TLS int jflag; +static VM_TLS int jknown; /* 0 until something asks for the end-of-input flag: the probe reads the port, so it waits for the first ask */ UCell micro_mem_words(void) { return mw; } static int inmem(Cell a) { return (UCell)a < mw; } @@ -109,7 +120,7 @@ static void b_store(Cell addr, Cell val, int n) #define PC_PAST_END (-5) #define PC_BAD_OP (-6) #define PC_HALT (-7) /* the zero word: the program's halt, not an error */ -static int bad_op; +static VM_TLS int bad_op; /* grows the machine to at least words words, through the port. the contents stay, and the new words are zero. 1 on success, 0 when the port has no more memory. */ @@ -482,13 +493,16 @@ static int xrun(Enh *e, int s) /* a run suspends when the port has no input yet but more may come (streaming, see micro_port.h): micro_getc returns MICRO_WAIT, the getc is run again on micro_resume, and the stacks wait where they are */ -static int sus_p, sus_s, sus_r; +static VM_TLS int sus_p, sus_s, sus_r; +static VM_TLS int final_s; +static Cell context_call(Cell op, Cell a, Cell b, Cell c, int sp, int rp); +static int context_cancelled(void); static int run_from(int p, int s, int r) { int op; Cell w, a, b; - while ((UCell)p < mw) { + while ((UCell)p < mw && !context_cancelled()) { if (s < stack_data_limit() || r > stack_return_limit()) { Cell sp = s, rp = r; int ret = r > stack_return_limit(); if (!stack_control || stack_expand(ret, stack_get(ret ? SC_RETURN_TOP : SC_DATA_TOP) - stack_get(ret ? SC_RETURN_LOW : SC_DATA_LOW) + 1, &sp, &rp) < 0) { @@ -517,7 +531,16 @@ static int run_from(int p, int s, int r) Cell limit = m[a + 2]; if (limit < 0 || (UCell)limit < mw || limit > 0x7fffffff) result = -1; else { memory_limit = (UCell)limit; result = limit; } - } else result = micro_syscall(m, b, m[a + 1], m[a + 2], m[a + 3]); + } else if (b == 41) { + char text[400]; int i, n = fop_format_mode((uint64_t)m[a + 1], 0, 17, text); + Cell dst = m[a + 2], cap = m[a + 3]; + if (dst < 0 || cap <= n || (UCell)dst > mw || (UCell)cap > mw - (UCell)dst) result = -1; + else { for (i = 0; i < n; i++) m[dst + i] = (unsigned char)text[i]; m[dst + n] = 0; result = n; } + } else if (b == 42) { + /* 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 result = context_call(b, m[a + 1], m[a + 2], m[a + 3], s + 2, r); m[a] = result; s += 2; break; @@ -604,6 +627,7 @@ static int run_from(int p, int s, int r) else { bad_op = op; p = PC_BAD_OP; } } } + final_s = s; if (p == PC_DATA_OVERFLOW) { micro_msg("zy: data stack overflow\n"); return 2; } if (p == PC_RETURN_OVERFLOW) { micro_msg("zy: return stack overflow\n"); return 2; } if (p == PC_PAST_END) { micro_msg("zy: the program counter is past the end of memory\n"); return 2; } @@ -620,6 +644,124 @@ static int run_from(int p, int s, int r) return 0; } +/* hosted contexts own their memory, input buffer and channel endpoints. */ +#ifdef ZY_THREADS +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); +#define CONTEXT_HANDLE_LO 0x40000000 +typedef struct ContextChannel { Cell handle; int fd; struct ContextChannel *next; } ContextChannel; +typedef struct VmContext { + ContextChannel parent_channel, worker_channel; + pthread_t thread; + struct VmContext *next; + atomic_int stop; + Cell token, entry, stack; + UCell words, limit, initial; + int sp, rp, exit_code, requested_exit, enhancement_count, input_flag, input_known; + Cell *memory; + void *port; + Xop operations[XPOOL]; + Enh enhancements[NENH]; +} VmContext; +static VM_TLS VmContext *context_current, *context_children; +static VM_TLS ContextChannel *context_channels; +static atomic_llong context_sequence; +static atomic_int context_records_live, context_memory_live; +static Cell context_join(Cell token, int cancel); +static int context_cancelled(void) { return context_current && atomic_load_explicit(&context_current->stop, memory_order_acquire); } +static int context_descriptor(Cell handle) { + if (handle < CONTEXT_HANDLE_LO) return handle >= 0 && handle <= 0x7fffffff ? (int)handle : -1; + for (ContextChannel *channel = context_channels; channel; channel = channel->next) if (channel->handle == handle) return channel->fd; + return -1; +} +static int context_close(Cell handle) { + ContextChannel **link = &context_channels; + while (*link && (*link)->handle != handle) link = &(*link)->next; + if (!*link) return -1; + ContextChannel *channel = *link; *link = channel->next; + if (channel->fd < 0) return -1; + int result = close(channel->fd); channel->fd = -1; return result; +} +static void context_snapshot_free(VmContext *context) { + free(context->port); + if (context->memory) { 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 (!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; + 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; + if (entry >= HEAP_LO && m[entry - 3] == (entry ^ 0x2B7E1516)) { m[--sp] = entry + 3; entry = m[entry]; } + m[rp++] = PC_HALT; + int result = run_from((int)entry, sp, rp); + 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; +} +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; + int fd[2]; if (socketpair(AF_UNIX, SOCK_STREAM, 0, fd) < 0) return -1; + VmContext *context = calloc(1, sizeof *context); if (!context) { close(fd[0]); close(fd[1]); return -1; } + context->entry = entry; context->sp = sp; context->rp = rp; context->stack = stack_control; + context->words = mw; context->limit = memory_limit; context->initial = memory_initial; + context->input_flag = jflag; context->input_known = jknown; context->enhancement_count = xuse; + context->token = atomic_fetch_add(&context_sequence, 1) + 1; + 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)); + 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; + 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); + 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; + atomic_fetch_add(&context_records_live, 1); m[channel] = context->parent_channel.handle; + context->next = context_children; context_children = context; return context->token; +failed: + context_snapshot_free(context); free(context); close(fd[0]); close(fd[1]); return -1; +} +static Cell context_join(Cell token, int cancel) { + VmContext **link = &context_children; + while (*link && (*link)->token != token) link = &(*link)->next; + if (!*link) return -1; + VmContext *context = *link; + if (cancel) { atomic_store_explicit(&context->stop, 1, memory_order_release); if (context->parent_channel.fd >= 0) shutdown(context->parent_channel.fd, SHUT_RDWR); } + if (pthread_join(context->thread, 0)) return -1; + if (context->parent_channel.fd >= 0) context_close(context->parent_channel.handle); + int result = context->exit_code; *link = context->next; context_snapshot_free(context); free(context); + atomic_fetch_sub(&context_records_live, 1); return result; +} +static Cell context_call(Cell op, Cell a, Cell b, Cell c, int sp, int rp) { + switch (op) { + case 51: return context_start(a, b, sp, rp); + case 52: return context_join(a, b != 0); + case 53: return getpid(); + case 54: return (Cell)(uintptr_t)pthread_self(); + case 55: return a == 0 ? atomic_load(&context_records_live) : a == 1 ? 0 : a == 2 ? atomic_load(&context_memory_live) : -1; + case 7: + if (context_current) { context_current->exit_code = (int)a; context_current->requested_exit = 1; atomic_store_explicit(&context_current->stop, 1, memory_order_release); return 0; } + break; + case 2: if (a >= CONTEXT_HANDLE_LO) return context_close(a); break; + case 44: case 45: case 46: a = context_descriptor(a); if (a < 0) return -1; break; + } + return micro_syscall(m, op, a, b, c); +} +#else +static int context_cancelled(void) { return 0; } +static Cell context_call(Cell op, Cell a, Cell b, Cell c, int sp, int rp) { return micro_syscall(m, op, a, b, c); } +#endif + static int run(int p) { return run_from(p, DS0, RS0); } /* goes on from a run that suspended for input (MICRO_SUSPENDED), once the port |