actor: a declared actor without its own drop gets one that awaits owed replies and closes the endpoint; the handler reaches the struct through endpoint.state so the struct isn't shared; spawn hint skips mailbox structs

1 files changed, 13 insertions, 2 deletions

diff --git a/src/lib/actor.zc b/src/lib/actor.zc
index a129f7f..c013441 100644
--- a/src/lib/actor.zc
+++ b/src/lib/actor.zc
@@ -170,8 +170,10 @@ unchecked int parse_actor() {
  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); }); } ");
+ run.tprint(signature[8]); rt.put_s(","); run.tprint(signature[4]); rt.put_s(">); this.__mailbox = endpoint; endpoint.state = this as word; return endpoint.spawn(fn(");
+ // the handler reaches the struct through the endpoint: capturing this would share it, and a
+ // shared struct is the collector's, so its drop would not run when its owner's block ends
+ run.tprint(signature[8]); rt.put_s(" message) { return (endpoint.state as "); declaration_type(type); rt.put_s("*).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");
@@ -182,6 +184,14 @@ unchecked int parse_actor() {
   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++;
  }
+ // the actor's process is a resource: unless the struct declares its own drop, its drop
+ // waits for the replies a registered receiver is owed, then closes the endpoint (cancelling
+ // the worker) when its owner's block ends, when it is freed, or when the collector reclaims
+ // it. a worker's snapshot of the struct closes nothing
+ if (run.methof(type, run.intern("drop")) < 0) {
+  declaration_method(type, "drop"); rt.put_s(") { if (actor.worker) return 0; 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) { endpoint.await(); endpoint.close(); } return 0; } ");
+ }
  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) {
@@ -211,6 +221,7 @@ actor Actor<I, O> {
  O(I) handler;
  int(O) receiver;
  int fd, process, pending, closed, dispatcher_index;
+ word state;	// a declared actor struct, for its handler (see parse_actor)
 
  // threads receive a private vm snapshot and an owned channel endpoint.
  unchecked fn spawn(O(I).d callback) {