1 files changed, 280 insertions, 0 deletions
diff --git a/src/lib/actor.zc b/src/lib/actor.zc
new file mode 100644
index 0000000..29fd600
--- /dev/null
+++ b/src/lib/actor.zc
@@ -0,0 +1,280 @@
+namespace actor;
+include rt, run;
+
+// stream entries are raw words: integer payloads never acquire heap ownership.
+unchecked word vector() { return rt.dnew(1, 0, 0); }
+unchecked int append(word vector, word value) {
+ rt.dgrow(vector, vector[0] + 1); vector[2][vector[0]] = value; vector[0]++; return 0;
+}
+unchecked int find(word refs, word value) {
+ int i = 0; while (i < refs[0]) { if (refs[2][i] == value) return i; i++; } return -1;
+}
+unchecked int size(word type) { return type[1]; }
+unchecked int packed(word type) { if (type[0] == 1 || type[0] == 12) return type[5]; return 0; }
+
+// encode pointer graphs with identity records; decoding creates fresh objects.
+// typed fields determine references, so an integer equal to a heap address
+// remains an integer. repeated references and cycles preserve identity.
+unchecked int encode(word data, word refs, word address, word type) {
+ int kind = type[0], i = 0, offset, bytes, n, id; word pointer, element;
+ if (kind == 8) {
+  while (i < type[3]) {
+   element = type[4][i * 3 + 1]; offset = type[4][i * 3 + 2];
+   if (offset < 0) { bytes = packed(element); n = 0; int k = 0; while (k < bytes) { n |= rt.ld8(rt.addr_b(address) - offset - 1 + k) << (k * 8); k++; } append(data, n); }
+   else encode(data, refs, address + offset, element);
+   i++;
+  }
+  return 0;
+ }
+ if (kind == 6) {
+  element = type[4]; bytes = packed(element);
+  while (i < type[3]) {
+   if (bytes) { n = 0; int k = 0; while (k < bytes) { n |= rt.ld8(rt.addr_b(address) + i * bytes + k) << (k * 8); k++; } append(data, n); }
+   else encode(data, refs, address + i * size(element), element);
+   i++;
+  }
+  return 0;
+ }
+ if (kind != 5 && kind != 11 && kind != 13 && kind != 7) { append(data, *address); return 0; }
+ pointer = *address;
+ if (!pointer) { append(data, 0); if (kind == 7) append(data, 0); return 0; }
+ if (kind == 7 && rt.callable_code(pointer)) { append(data, 0); append(data, pointer); return 0; }
+ id = find(refs, pointer);
+ if (id >= 0) { append(data, id + 1); return 0; }
+ id = refs[0]; append(refs, pointer); append(data, 0 - id - 1);
+ if (kind == 11) {
+  n = pointer[0]; append(data, n); element = type[4]; bytes = packed(element);
+  while (i < n) {
+   if (bytes) { int k = 0; int value = 0; while (k < bytes) { value |= rt.ld8(rt.addr_b(pointer[2]) + i * bytes + k) << (k * 8); k++; } append(data, value); }
+   else encode(data, refs, pointer[2] + i * size(element), element);
+   i++;
+  }
+  return 0;
+ }
+ if (kind == 13) { n = rt.len_s(pointer); append(data, n); while (i < n) { append(data, rt.at_s(pointer, i)); i++; } return 0; }
+ if (kind == 7) {
+  element = pointer[-1]; n = pointer[-5] - rt.HDR;
+  append(data, n); append(data, element); encode(data, refs, pointer, element); return 0;
+ }
+ element = type[4]; bytes = packed(element);
+ if (rt.isobj(pointer)) n = pointer[-5] - rt.HDR;
+ else if (element[0] == 2) n = rt.len_w(pointer) + 1;
+ else n = size(element);
+ append(data, n);
+ if (bytes || size(element) <= 0) { while (i < n) { append(data, pointer[i]); i++; } }
+ else { while (i + size(element) <= n) { encode(data, refs, pointer + i, element); i += size(element); } }
+ return 0;
+}
+unchecked word take(word frame) { word result = frame[0][2][frame[1]]; frame[1]++; return result; }
+unchecked int decode(word frame, word refs, word address, word type) {
+ int kind = type[0], i = 0, offset, bytes, n, id; word pointer, element;
+ if (kind == 8) {
+  while (i < type[3]) {
+   element = type[4][i * 3 + 1]; offset = type[4][i * 3 + 2];
+   if (offset < 0) { bytes = packed(element); word value = take(frame); int k = 0; while (k < bytes) { rt.st8(rt.addr_b(address) - offset - 1 + k, value >> (k * 8)); k++; } }
+   else decode(frame, refs, address + offset, element);
+   i++;
+  }
+  return 0;
+ }
+ if (kind == 6) {
+  element = type[4]; bytes = packed(element);
+  while (i < type[3]) {
+   if (bytes) { word value = take(frame); int k = 0; while (k < bytes) { rt.st8(rt.addr_b(address) + i * bytes + k, value >> (k * 8)); k++; } }
+   else decode(frame, refs, address + i * size(element), element);
+   i++;
+  }
+  return 0;
+ }
+ if (kind != 5 && kind != 11 && kind != 13 && kind != 7) { *address = take(frame); return 0; }
+ id = take(frame);
+ if (!id) { if (kind == 7) *address = take(frame); else *address = 0; return 0; }
+ if (id > 0) { *address = refs[2][id - 1]; return 0; }
+ n = take(frame);
+ if (kind == 11) {
+  element = type[4]; bytes = packed(element);
+  if (bytes) pointer = rt.dnew(0 - bytes, n, type); else pointer = rt.dnew(size(element), n, type);
+  append(refs, pointer); *address = pointer;
+  while (i < n) {
+   if (bytes) { word value = take(frame); int k = 0; while (k < bytes) { rt.st8(rt.addr_b(pointer[2]) + i * bytes + k, value >> (k * 8)); k++; } }
+   else decode(frame, refs, pointer[2] + i * size(element), element);
+   i++;
+  }
+  return 0;
+ }
+ if (kind == 13) {
+  pointer = rt.new((n + 7) / 8 + 1); pointer[0] = n; append(refs, pointer); *address = pointer;
+  while (i < n) { rt.st8(rt.addr_b(pointer + 1) + i, take(frame)); i++; } return 0;
+ }
+ if (kind == 7) { element = take(frame); pointer = rt.new_t(n, element); append(refs, pointer); *address = pointer; decode(frame, refs, pointer, element); return 0; }
+ pointer = rt.new_t(n, type[4]); append(refs, pointer); *address = pointer;
+ element = type[4]; bytes = packed(element);
+ if (bytes || size(element) <= 0) { while (i < n) { pointer[i] = take(frame); i++; } }
+ else { while (i + size(element) <= n) { decode(frame, refs, pointer + i, element); i += size(element); } }
+ return 0;
+}
+unchecked int write_message(int fd, word address, word type) {
+ word data = vector(), refs = vector(); int count;
+ encode(data, refs, address, type); count = data[0];
+ int result = rt.syscall(44, fd, &count, 1);
+ if (result >= 0) result = rt.syscall(44, fd, data[2], count);
+ rt.free(refs); rt.free(data); return result;
+}
+unchecked int read_message(int fd, word address, word type) {
+ int count = 0; word data, refs, frame[2];
+ if (rt.syscall(45, fd, &count, 1) < 0 || count < 0 || count > 1048576) return -1;
+ data = rt.dnew(1, count, 0); refs = vector(); frame[0] = data; frame[1] = 0;
+ if (rt.syscall(45, fd, data[2], count) < 0) { rt.free(refs); rt.free(data); return -1; }
+ decode(frame, refs, address, type); rt.free(refs); rt.free(data); return 0;
+}
+
+int busy, worker;
+type Dispatcher = fn();
+Dispatcher[?] dispatchers;
+unchecked int dispatch() {
+ if (busy || worker || !dispatchers) return 0;
+ busy = 1; int i = 0;
+ while (i < len(dispatchers)) { if (dispatchers[i]) dispatchers[i](); i++; }
+ busy = 0; return 0;
+}
+
+// declared actors expose a typed handle method and receive their own endpoint.
+// lowering lives in this imported source; the compiler only dispatches callbacks.
+unchecked int declaration_parameters(word type) {
+ if (type[0] != run.TY_GEN) return 0;
+ rt.put_c('<'); int i = 0;
+ while (i < type[3]) { if (i) rt.put_c(','); rt.put_s(type[6][i]); i++; }
+ rt.put_c('>'); return 0;
+}
+unchecked int declaration_type(word type) { rt.put_s(type[2]); declaration_parameters(type); return 0; }
+unchecked int declaration_method(word type, char* method) {
+ rt.put_s("unchecked fn "); rt.put_s(type[2]); rt.put_c('.'); rt.put_s(method); declaration_parameters(type); rt.put_c('('); declaration_type(type); rt.put_s("* this"); return 0;
+}
+unchecked int parse_actor() {
+ int symbol = run.kw_struct(); word type = run.sval[symbol], signature, fields, oldfields;
+ int handler, count, mode, length, capacity; word output, previous_grow; char* buffer;
+ if (type[0] == run.TY_GEN && rt.eq_w(type[2], "actor.Actor")) return 0;
+ handler = run.methof(type, run.intern("handle"));
+ if (handler < 0) { run.serr("an actor declaration requires handle(input) with a typed result"); return 0; }
+ signature = run.stype[handler];
+ if (!signature || signature[3] != 2) { run.serr("an actor handle method requires one input parameter"); return 0; }
+ int generic = type[0] == run.TY_GEN, stride = 3 - generic;
+ count = type[3]; if (generic) count = type[5]; int field_index = 0;
+ while (field_index < count) { if (type[4][field_index * stride] == run.intern("__mailbox")) { run.serr("__mailbox is reserved for actor state"); return 0; } field_index++; }
+ oldfields = type[4]; fields = run.talloc((count + 1) * stride + 1);
+ rt.copy_w(fields, oldfields, count * stride); fields[count * stride] = run.intern("__mailbox");
+ fields[count * stride + 1] = run.tword;
+ if (generic) type[5]++; else { fields[count * stride + 2] = type[1]; type[3]++; type[1]++; }
+ type[4] = fields;
+ buffer = rt.new(16384); mode = rt.outmode; output = rt.outbuf; length = rt.outn; capacity = rt.outcap; previous_grow = rt.outgrow;
+ rt.outgrow = 0; rt.outmode = 2; rt.outbuf = buffer; rt.outn = 0; rt.outcap = 16383;
+ declaration_method(type, "spawn"); rt.put_s(") { if (this.__mailbox) return -1; actor.Actor<");
+ run.tprint(signature[8]); rt.put_s(","); run.tprint(signature[4]); rt.put_s(">* endpoint = rt.new(actor.Actor<");
+ run.tprint(signature[8]); rt.put_s(","); run.tprint(signature[4]); rt.put_s(">); this.__mailbox = endpoint; return endpoint.spawn(fn(");
+ run.tprint(signature[8]); rt.put_s(" message) { return this.handle(message); }); } ");
+ int index = 0;
+ while (index < 5) {
+  if (index == 0) declaration_method(type, "send"); else if (index == 1) declaration_method(type, "recv"); else if (index == 2) declaration_method(type, "await"); else if (index == 3) declaration_method(type, "close"); else declaration_method(type, "poll");
+  if (index == 0) { rt.put_s(","); run.tprint(signature[8]); rt.put_s(" message"); }
+  if (index == 1) { rt.put_s(",int("); run.tprint(signature[4]); rt.put_s(").d callback"); }
+  rt.put_s(") { actor.Actor<"); run.tprint(signature[8]); rt.put_s(","); run.tprint(signature[4]); rt.put_s(">* endpoint = this.__mailbox as actor.Actor<");
+  run.tprint(signature[8]); rt.put_s(","); run.tprint(signature[4]); rt.put_s(">*; if (!endpoint) return -1; return endpoint.");
+  if (index == 0) rt.put_s("send(message);"); else if (index == 1) rt.put_s("recv(callback);"); else if (index == 2) rt.put_s("await();"); else if (index == 3) rt.put_s("close();"); else rt.put_s("poll();");
+  rt.put_s(" } "); index++;
+ }
+ buffer[rt.outn] = 0; rt.outgrow = previous_grow; rt.outmode = mode; rt.outbuf = output; rt.outn = length; rt.outcap = capacity;
+ word tokens = run.tokens(buffer);
+ if (generic) {
+  int position = 0, variable;
+  while (position < tokens[0]) {
+   variable = 0;
+   while (variable < run.ntl) {
+    word descriptor = run.typelist[variable];
+    if (descriptor[0] == run.TY_VAR && descriptor[3] < type[3] && tokens[1 + position * 3] == run.T_ID && tokens[2 + position * 3] == descriptor[2]) tokens[2 + position * 3] = type[6][descriptor[3]];
+    variable++;
+   }
+   position++;
+  }
+ }
+ word lowered = rt.new(tokens[0] * 3 + 1); int read = 0, written = 0;
+ while (read < tokens[0]) {
+  if (read > 0 && tokens[2 + (read - 1) * 3] == run.s_fn && read + 2 < tokens[0] && tokens[2 + (read + 1) * 3] == run.s_dot) {
+   lowered[1 + written * 3] = run.T_ID; lowered[2 + written * 3] = run.methname(tokens[2 + read * 3], tokens[2 + (read + 2) * 3]); lowered[3 + written * 3] = tokens[3 + read * 3]; read += 3;
+  } else { rt.copy_w(lowered + 1 + written * 3, tokens + 1 + read * 3, 3); read++; }
+  written++;
+ }
+ lowered[0] = written; run.compile_list_context(lowered, 1, run.module_name, run.module_imports); rt.free(lowered); rt.free(tokens); rt.free(buffer); return 0;
+}
+run.syntax("actor", actor.parse_actor);
+
+actor Actor<I, O> {
+ O(I) handler;
+ int(O) receiver;
+ int fd, process, pending, closed, dispatcher_index;
+
+ // threads receive a private vm snapshot and an owned channel endpoint.
+ unchecked fn spawn(O(I).d callback) {
+  if (this.process || this.closed || !callback) return -1;
+  this.handler = callback;
+  worker = 1;
+  this.process = rt.syscall(51, fn() {
+   worker = 1;
+   while (1) {
+    I message; O result;
+    if (read_message(this.fd, &message, typeof(I)) < 0) break;
+    result = this.handler(message);
+    if (write_message(this.fd, &result, typeof(O)) < 0) break;
+   }
+   return 0;
+  } as word, &this.fd, 0);
+  if (this.process < 0) { worker = 0; return -1; }
+  worker = 0;
+  if (!dispatchers) dispatchers = rt.new(Dispatcher[?]);
+  this.dispatcher_index = len(dispatchers);
+  push(dispatchers, fn:word() { this.poll(); return 0; });
+  return 0;
+ }
+
+ // sends a snapshot; the worker executes it asynchronously on its own cpu.
+ unchecked fn send(I message) {
+  if (this.process <= 0 || this.closed) return -1;
+  if (write_message(this.fd, &message, typeof(I)) < 0) { this.close(); return -1; }
+  this.pending++; return 0;
+ }
+
+ // registration does not wait; statement callbacks deliver ready replies.
+ unchecked fn recv(int(O).d callback) {
+  if (this.closed || !callback) return -1;
+  this.receiver = callback; this.poll(); return 0;
+ }
+
+ unchecked fn poll() {
+  O result;
+  if (!this.receiver || this.closed || this.process <= 0) return 0;
+  int previous_busy = busy; busy = 1;
+  while (!this.closed && this.pending > 0 && rt.syscall(46, this.fd, 0, 0) > 0) {
+   if (read_message(this.fd, &result, typeof(O)) < 0) { busy = previous_busy; this.close(); return -1; }
+   this.pending--; this.receiver(result);
+  }
+  busy = previous_busy; return 0;
+ }
+
+ // waits until every submitted message has reached the registered receiver.
+ unchecked fn await() {
+  if (!this.receiver && this.pending) return -1;
+  while (this.pending > 0 && !this.closed) {
+   if (rt.syscall(46, this.fd, -1, 0) < 0 || this.poll() < 0) return -1;
+  }
+  return 0;
+ }
+
+ unchecked fn close() {
+  if (this.closed) return 0;
+  this.closed = 1;
+  if (this.process > 0 && dispatchers) dispatchers[this.dispatcher_index] = 0;
+  if (this.process > 0) rt.syscall(52, this.process, 1, 0);
+  this.handler = 0; this.receiver = 0; return 0;
+ }
+}
+
+run.statement_hook(actor.dispatch);