path: root/src/lib/actor.zc

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);