proper beam evaluator

4 files changed, 44 insertions, 428 deletions

diff --git a/subsystem/karuta/Beam.ml b/subsystem/karuta/Beam.ml
index 0f9e93c..f46a131 100644
--- a/subsystem/karuta/Beam.ml
+++ b/subsystem/karuta/Beam.ml
@@ -1,48 +1,14 @@
-(* The beam evaluator. Its programs are functions on a BEAM node, each
-   taking one relation per input and giving back a relation of the
-   heading the program declares. rnt-node.lfe serves them. Tuples cross
-   as Translator writes them, and a message is a byte for its kind
-   followed by big-endian integers and binaries:
-
-     RNT   i Cursor:32 M:16 Module F:16 Function Reader:32...  invoke
-     RNT   f Cursor:32 Limit:32                                 fetch
-     RNT   c Cursor:32                                          close
-     node  n Reader:32 Limit:32                                 next
-     node  l (Size:32 Segment)...                               lookup
-     both  b Exhausted:8 (Size:32 Tuple)...                     a batch
-     RNT   r Reader:32                                          a reader
-     both  e Message                                            a failure
-
-   fetch and next are answered with a batch or a failure, lookup with a
-   reader or a failure, invoke and close not at all. The node applies
-   the function at the first fetch, so a function that cannot be applied
-   fails it. Calls nest: until its answer comes, whoever waits serves the
-   calls that come instead.
-
-   lookup asks for the relation at a path, within what the program runs
-   within, and FOL answers it. The node reads it with next like an
-   input, and the cursor being fetched, whose program asked, keeps it
-   until it closes. *)
-
 open Rnt
 
 module Error = struct
   open Concepts.Condition
 
-  let not_a_program () =
-    condition "not-a-beam-program" "The evaluator was given a program of another kind" empty
-
-  let failed message =
-    condition "beam-failed" "The BEAM node failed to answer a call"
-      ("message" |=| Concepts.Value.String message)
-
   let closed () = condition "beam-closed" "The BEAM node is gone" empty
 
-  let unexpected () =
-    condition "unexpected-beam-message" "The BEAM node said something out of turn" empty
+  let unimplemented () =
+    condition "beam-unimplemented" "The beam evaluator cannot do this yet" empty
 
-  let unheld_input () =
-    condition "unheld-input" "An input could not be held for as long as the program needs it" empty
+  let taken () = condition "beam-taken" "An evaluator is already registered as karuta:beam" empty
 end
 
 module type TRANSPORT = sig
@@ -50,11 +16,9 @@ module type TRANSPORT = sig
 
   val send : t -> string -> (unit, Concepts.Condition.condition) result
   val receive : t -> (string, Concepts.Condition.condition) result
+  val stop : t -> unit
 end
 
-(* A node RNT starts as its child and speaks to over the child's stdin
-   and stdout, each message behind its length in four bytes, as a port
-   opened with {packet, 4} frames it. *)
 module Port = struct
   type t = {input: in_channel; output: out_channel}
 
@@ -70,292 +34,69 @@ module Port = struct
   let receive {input; _} =
     try
       let header = really_input_string input 4 in
-      Ok (really_input_string input (Int32.to_int (String.get_int32_be header 0)))
+      Ok (really_input_string input (Int32.to_int (String.get_int32_be header 0) land 0xffff_ffff))
     with End_of_file | Sys_error _ -> Error (Error.closed ())
 
-  (* The node compiles every LFE module in [directories], which must
-     hold rnt.lfe and rnt-node.lfe, and serves until [stop] closes its
-     stdin. If it cannot, it says why on stderr and halts. It is started
-     with erl and not lfe, whose shell would read stdin, so LFE has to be
-     where erl finds it, as ERL_LIBS says. *)
   let start ?(erl = "erl") directories =
     let boot =
-      "try [begin {ok, Ms, _} = lfe_comp:file(F, [binary, return]), [{module, M} = \
-       code:load_binary(M, F, B) || {ok, M, B, _} <- Ms] end || D <- init:get_plain_arguments(), F \
-       <- filelib:wildcard(D ++ \"/*.lfe\")], 'rnt-node':start() catch C:R:S -> \
+      "try [{module, _} = code:load_abs(filename:rootname(F)) || D <- init:get_plain_arguments(), \
+       F <- filelib:wildcard(D ++ \"/*.beam\")], 'rnt-node':start() catch C:R:S -> \
        io:format(standard_error, \"~p:~p~n~p~n\", [C, R, S]), halt(1) end."
     in
-    let input, output =
+    match
       Unix.open_process_args erl
         (Array.of_list ([erl; "-noshell"; "-noinput"; "-eval"; boot; "-extra"] @ directories))
-    in
-    {input; output}
+    with
+    | input, output -> Ok {input; output}
+    | exception Unix.Unix_error _ -> Error (Error.closed ())
 
   let stop {input; output} = ignore (Unix.close_process (input, output))
 end
 
-let subsystem = "karuta"
-let language = "beam"
-let name = subsystem ^ ":" ^ language
+let name = "karuta"
 
 type program =
   {module_: string; function_: string; heading: Protocols.Schematics.relation_description}
 
 type Kernel.Program.kind += Beam of program
 
-module Bencode = Concepts.Encoding.Bencode
-
-let encode {module_; function_; heading} =
-  Bencode.Dict
-    [ "module", Bencode.String module_;
-      "function", Bencode.String function_;
-      ( "heading",
-        Bencode.Dict
-          ( BatMap.String.bindings heading
-          |> List.map (fun (attribute, {Protocols.Schematics.domain; _}) ->
-              attribute, Bencode.String domain ) ) ) ]
-  |> Bencode.to_bytes
-  |> Bytes.to_string
-
-let decode source =
-  let open Utilities.Result in
-  let* program = Bencode.of_bytes (Bytes.of_string source) in
-  let* module_ = Concepts.Encoding.Field.require_string "module" program in
-  let* function_ = Concepts.Encoding.Field.require_string "function" program in
-  let* heading = Bencode.field "heading" program |>> Bencode.as_dict in
-  let* heading =
-    heading
-    |> List.map (fun (attribute, domain) ->
-        Bencode.as_string domain
-        |> Result.map (fun domain -> attribute, {Protocols.Schematics.domain; provenance= []}) )
-    |> Utilities.List.sequence
-  in
-  Ok {module_; function_; heading= BatMap.String.of_list heading}
-
-let int32 buffer n = Buffer.add_int32_be buffer (Int32.of_int n)
-
-let invoke_message cursor {module_; function_; _} readers =
-  let buffer = Buffer.create 64 in
-  Buffer.add_char buffer 'i';
-  int32 buffer cursor;
-  List.iter
-    (fun name ->
-      Buffer.add_uint16_be buffer (String.length name);
-      Buffer.add_string buffer name )
-    [module_; function_];
-  List.iter (int32 buffer) readers;
-  Buffer.contents buffer
-
-let fetch_message cursor limit =
-  let buffer = Buffer.create 9 in
-  Buffer.add_char buffer 'f'; int32 buffer cursor; int32 buffer limit; Buffer.contents buffer
-
-let close_message cursor =
-  let buffer = Buffer.create 5 in
-  Buffer.add_char buffer 'c'; int32 buffer cursor; Buffer.contents buffer
-
-let batch_message {Protocols.Cursor.tuples; exhausted} =
-  let buffer = Buffer.create 256 in
-  Buffer.add_char buffer 'b';
-  Buffer.add_uint8 buffer (Bool.to_int exhausted);
-  BatFingerTree.iter
-    (fun tuple ->
-      let term = Translator.encode tuple in
-      int32 buffer (String.length term);
-      Buffer.add_string buffer term )
-    tuples;
-  Buffer.contents buffer
-
-let reader_message reader =
-  let buffer = Buffer.create 5 in
-  Buffer.add_char buffer 'r'; int32 buffer reader; Buffer.contents buffer
-
-let failure_message condition = "e" ^ Concepts.Condition.to_string_hum condition
-
-(* The pieces of [message] from [at] on, each behind its size in four bytes. *)
-let chunks message at =
-  let length = String.length message in
-  let rec read at pieces =
-    if at = length then Ok (List.rev pieces)
-    else if at + 4 > length then Error (Error.unexpected ())
-    else
-      let size = Int32.to_int (String.get_int32_be message at) land 0xffff_ffff in
-      if at + 4 + size > length then Error (Error.unexpected ())
-      else read (at + 4 + size) (String.sub message (at + 4) size :: pieces)
-  in
-  read at []
-
-let batch_of message =
-  let open Utilities.Result in
-  let* terms = chunks message 2 in
-  let* tuples = List.map Translator.decode terms |> Utilities.List.sequence in
-  Ok {Protocols.Cursor.tuples= BatFingerTree.of_list tuples; exhausted= message.[1] <> '\000'}
-
 module Make (T : TRANSPORT) = struct
-  (* The root programs find their evaluators in, the cursors the node
-     reads through us by the reader id it knows each by, and the last id
-     given to a reader or a cursor. *)
-  type node =
-    { root: Protocols.Handle.t;
-      transport: T.t;
-      readers: (int, Protocols.Handle.t) Hashtbl.t;
-      mutable last: int }
-
-  let fresh node =
-    node.last <- node.last + 1;
-    node.last
-
-  let next node message =
-    let open Utilities.Result in
-    let reader = Int32.to_int (String.get_int32_be message 1) in
-    let limit = Int32.to_int (String.get_int32_be message 5) in
-    let* cursor =
-      Hashtbl.find_opt node.readers reader |> Option.to_result ~none:(Error.unexpected ())
-    in
-    let* scan = Protocols.Cursor.require cursor in
-    Protocols.Cursor.fetch scan limit
-
-  let lookup node within message =
-    let open Utilities.Result in
-    let* path = chunks message 1 in
-    let program =
-      Evaluators.FOL.program ~root:node.root (Evaluators.FOL.Base (Kernel.Path.of_list path))
-    in
-    let* relation =
-      Protocols.Handle.releasing program (fun () ->
-          let* executable = Protocols.Executable.require program in
-          Protocols.Executable.invoke executable [] ~within )
-    in
-    let* cursor = Kernel.EphemeralRelation.enumerate_owned relation in
-    let reader = fresh node in
-    Hashtbl.replace node.readers reader cursor;
-    Ok reader
-
-  (* [own] keeps a reader the node looked up for the cursor being fetched. *)
-  let rec await node within own =
-    let open Utilities.Result in
-    let* message = T.receive node.transport in
-    let reply answer =
-      let* () = T.send node.transport answer in
-      await node within own
-    in
-    match if message = "" then ' ' else message.[0] with
-    | 'b' when String.length message >= 2 -> batch_of message
-    | 'e' -> Error (Error.failed (String.sub message 1 (String.length message - 1)))
-    | 'n' when String.length message = 9 -> (
-        reply
-        @@
-        match next node message with
-        | Ok batch -> batch_message batch
-        | Error c -> failure_message c )
-    | 'l' -> (
-        reply
-        @@
-        match lookup node within message with
-        | Ok reader -> own reader; reader_message reader
-        | Error c -> failure_message c )
-    | _ -> Error (Error.unexpected ())
-
-  let forget node readers =
-    List.iter
-      (fun reader ->
-        Hashtbl.find_opt node.readers reader |> Option.iter Protocols.Handle.release;
-        Hashtbl.remove node.readers reader )
-      readers
-
-  (* It keeps the readers of its inputs, and of what its program looked up. *)
-  class cursor node id inputs within =
+  class evaluator (transport : T.t) =
     object (self)
       inherit Kernel.Lifecycle.counted
       inherit Kernel.Identity.of_id
-      val mutable readers = inputs
-      method to_string = "beam-cursor"
-
-      method fetch limit =
-        Result.bind
-          (T.send node.transport (fetch_message id limit))
-          (fun () -> await node within (fun reader -> readers <- reader :: readers))
-
-      method destroy =
-        ignore (T.send node.transport (close_message id));
-        forget node readers
-
-      method protocols : Protocols.Handle.protocol list = Protocols.[Cursor.make self]
-    end
+      method to_string = name
+      method destroy = T.stop transport
 
-  let open_readers node inputs =
-    Utilities.List.foldm_left
-      (fun readers input ->
-        match Protocols.Enumerable.cursor input with
-        | Ok cursor ->
-            let reader = fresh node in
-            Hashtbl.replace node.readers reader cursor;
-            Ok (reader :: readers)
-        | Error c -> forget node readers; Error c )
-      [] inputs
-    |> Result.map List.rev
+      method parse : string -> (Kernel.Program.kind, Concepts.Condition.condition) result =
+        fun _ -> Error (Error.unimplemented ())
 
-  let invoke node program inputs ~within =
-    let open Utilities.Result in
-    let* readers = open_readers node inputs in
-    let id = fresh node in
-    match T.send node.transport (invoke_message id program readers) with
-    | Ok () -> Ok (Protocols.Handle.make (new cursor node id readers within))
-    | Error c -> forget node readers; Error c
-
-  (* The relation reads its inputs each time it is enumerated, so it holds them. *)
-  let hold inputs =
-    let held = List.filter_map Protocols.Handle.copy inputs in
-    if List.length held = List.length inputs then Ok held
-    else begin
-      List.iter Protocols.Handle.release held;
-      Error (Error.unheld_input ())
-    end
-
-  (* It looks relations up within [within] too, so it holds that as well.
-     TODO: A program should declare its modes as it declares its heading.
-     Until then every program is taken to end. *)
-  let generate node program inputs ~within =
-    let open Utilities.Result in
-    let* held = hold (within :: inputs) in
-    match held with
-    | [] -> Error (Error.unheld_input ())
-    | within :: inputs ->
-        Ok
-          (Kernel.EphemeralRelation.instantiate ~inputs:held program.heading
-             ~modes:(fun () -> Ok [Concepts.Mode.mode [] Concepts.Cardinality.Finite])
-             (Kernel.EphemeralRelation.temporary program.heading)
-             (fun () -> invoke node program inputs ~within) )
-
-  class evaluator node =
-    object (self)
-      inherit Kernel.Lifecycle.null
-      inherit Kernel.Identity.of_id
-      method to_string = "beam-evaluator"
-      method parse source = decode source |> Result.map (fun program -> Beam program)
-
-      method eval kind inputs ~within =
-        match kind with
-        | Beam program -> generate node program inputs ~within
-        | _ -> Error (Error.not_a_program ())
+      method eval :
+          Kernel.Program.kind ->
+          Protocols.Handle.t list ->
+          within:Protocols.Handle.t ->
+          (Protocols.Handle.t, Concepts.Condition.condition) result =
+        fun _ _ ~within:_ -> Error (Error.unimplemented ())
 
       method protocols : Protocols.Handle.protocol list = [Protocols.Evaluator.make self]
     end
-
-  (* \karuta is the subsystem's root, and its programs name the evaluator karuta:beam. *)
-  let register root transport =
-    let open Utilities.Result in
-    let node = {root; transport; readers= Hashtbl.create 8; last= 0} in
-    let* _ = Kernel.Initialization.mount root subsystem in
-    let* _ =
-      Kernel.Path.update root
-        Kernel.Path.(subsystem @/ "evaluator" @/ this)
-        language None
-        (Some (Protocols.Handle.make (new evaluator node)))
-    in
-    Ok ()
 end
 
-let program ?root ~module_ ~function_ heading =
-  let program = {module_; function_; heading} in
-  Kernel.Program.make ?root ~evaluator:name ~source:(encode program) (Beam program)
+module Node = Make (Port)
+
+let register root directories =
+  let open Utilities.Result in
+  let* port = Port.start directories in
+  let evaluator = Protocols.Handle.make (new Node.evaluator port) in
+  let bound =
+    let* () = Kernel.Initialization.mount root name in
+    Kernel.Path.update root Kernel.Path.(name @/ "evaluator" @/ this) "beam" None (Some evaluator)
+  in
+  match bound with
+  | Ok true -> Ok ()
+  | Ok false ->
+      Protocols.Handle.release evaluator;
+      Error (Error.taken ())
+  | Error c ->
+      Protocols.Handle.release evaluator;
+      Error c
diff --git a/subsystem/karuta/Translator.ml b/subsystem/karuta/Translator.ml
index 1fbbd9d..0d7e7b7 100644
--- a/subsystem/karuta/Translator.ml
+++ b/subsystem/karuta/Translator.ml
@@ -1,16 +1,11 @@
-(* A tuple of an ephemeral relation as an Erlang term in the external
-   term format, for binary_to_term on the BEAM: {Type, #{Name =>
-   Value}}, with the type, names and strings as binaries. And back,
-   from what term_to_binary writes for such a term. *)
-
 open Rnt
 
 module Error = struct
   open Concepts.Condition
+
   let malformed () =
-  Concepts.Condition.condition "malformed-beam-tuple"
-    "A term is not {Type, #{Name => Value}} of binaries, and integers that fit an int"
-    Concepts.Condition.empty
+    condition "malformed-beam-tuple"
+      "A term is not {Type, #{Name => Value}} of binaries, and integers that fit an int" empty
 end
 
 let binary buffer s =
@@ -18,9 +13,6 @@ let binary buffer s =
   Buffer.add_int32_be buffer (Int32.of_int (String.length s));
   Buffer.add_string buffer s
 
-(* Every integer as a small big: a sign, then the bytes of the magnitude,
-   least significant first. binary_to_term gives back an ordinary
-   integer when it fits one, and lsr reads abs min_int as 2^62. *)
 let integer buffer n =
   let rec digits m = if m = 0 then [] else (m land 0xff) :: digits (m lsr 8) in
   let digits = digits (abs n) in
@@ -46,7 +38,6 @@ let encode {Concepts.Tuple.type_; attributes} =
     attributes;
   Buffer.contents buffer
 
-(* Decoding raises within and answers a result at the end. *)
 exception Malformed
 
 let decode bytes =
@@ -65,9 +56,6 @@ let decode bytes =
     let n = u32 () in
     String.sub bytes (take n) n
   in
-  (* a sign, then at most 8 bytes of the magnitude, least significant
-     first; an int holds magnitudes up to 2^62 - 1, or 2^62 when
-     negative *)
   let big n =
     let negative = u8 () = 1 in
     if n > 8 then raise Malformed;
@@ -83,7 +71,6 @@ let decode bytes =
     if Int64.compare m 0L < 0 || Int64.compare m limit > 0 then raise Malformed;
     Int64.to_int (if negative then Int64.neg m else m)
   in
-  (* term_to_binary writes an integer in the least of these that holds it *)
   let value () =
     match u8 () with
     | 109 -> Concepts.Value.String (binary ())
diff --git a/subsystem/karuta/rnt-node.lfe b/subsystem/karuta/rnt-node.lfe
deleted file mode 100644
index 94a4b76..0000000
--- a/subsystem/karuta/rnt-node.lfe
+++ /dev/null
@@ -1,86 +0,0 @@
-;; The node of the beam evaluator, which Beam.ml starts and speaks to
-;; over stdin and stdout, with the messages Beam.ml lists. A program is
-;; a function taking one relation per input and giving back a relation:
-;; a function from a limit to at most that many tuples and the rest of
-;; the relation, or done when there are no more. Its tuples are as
-;; rnt:recv/1 gives and rnt:send/1 takes them, and it can read any
-;; relation RNT holds with lookup/1.
-(defmodule rnt-node
-  (export (start 0) (lookup 1)))
-
-(defun start ()
-  ;; The default handler prints to stdout, where it would break a frame.
-  (logger:remove_handler 'default)
-  (logger:add_handler 'default 'logger_std_h (map 'config (map 'type 'standard_error)))
-  (put 'port (open_port #(fd 0 1) '(binary #(packet 4) eof)))
-  (loop))
-
-(defun loop () (serve (read-frame)) (loop))
-
-(defun read-frame ()
-  (let ((port (get 'port)))
-    (receive
-      ((tuple p (tuple 'data frame)) (when (=:= p port)) frame)
-      ((tuple p 'eof) (when (=:= p port)) (halt)))))
-
-(defun write-frame (message) (port_command (get 'port) message))
-
-;; The function is applied at the first fetch, while RNT waits on it and
-;; can answer what the function asks.
-(defun serve
-  (((binary #\i (cursor (size 32)) (m (size 16)) (module binary (size m))
-            (f (size 16)) (func binary (size f)) (readers binary)))
-   (let ((inputs (lc ((<= (binary (reader (size 32))) readers)) (relation reader))))
-     (put (tuple 'cursor cursor)
-          (lambda (limit)
-            (funcall (apply (binary_to_existing_atom module) (binary_to_existing_atom func) inputs)
-                     limit)))))
-  (((binary #\f (cursor (size 32)) (limit (size 32))))
-   (write-frame
-     (try (fetch cursor limit)
-       (catch
-         ((tuple class reason _) (list #\e (io_lib:format "~p:~p" (list class reason))))))))
-  (((binary #\c (cursor (size 32))))
-   (erase (tuple 'cursor cursor))))
-
-(defun fetch (cursor limit)
-  (case (get (tuple 'cursor cursor))
-    ('done (batch '() 'true))
-    (relation
-     (let (((tuple tuples rest) (funcall relation limit)))
-       (put (tuple 'cursor cursor) rest)
-       (batch tuples (=:= rest 'done))))))
-
-(defun batch (tuples exhausted)
-  (list* #\b (if exhausted 1 0)
-         (lc ((<- tuple tuples))
-           (let ((term (rnt:send tuple)))
-             (list (binary ((byte_size term) (size 32))) term)))))
-
-;; The relation at path, a list of segments, within what the program
-;; runs within.
-(defun lookup (path)
-  (write-frame
-    (cons #\l (lc ((<- segment path))
-                (let ((bytes (iolist_to_binary segment)))
-                  (list (binary ((byte_size bytes) (size 32))) bytes)))))
-  (let (((tuple 'reader reader) (answer)))
-    (relation reader)))
-
-;; A relation read through RNT. Reading it moves the reader on, so it
-;; can be read once.
-(defun relation (reader)
-  (lambda (limit)
-    (write-frame (binary #\n (reader (size 32)) (limit (size 32))))
-    (let (((tuple tuples exhausted) (answer)))
-      (tuple tuples (if exhausted 'done (relation reader))))))
-
-;; The answer to a next or a lookup, serving the calls that come before it.
-(defun answer ()
-  (case (read-frame)
-    ((binary #\b exhausted (tuples binary))
-     (tuple (lc ((<= (binary (n (size 32)) (term binary (size n))) tuples)) (rnt:recv term))
-            (=:= exhausted 1)))
-    ((binary #\r (reader (size 32))) (tuple 'reader reader))
-    ((binary #\e (message binary)) (error (tuple 'rnt message)))
-    (call (serve call) (answer))))
diff --git a/subsystem/karuta/rnt.lfe b/subsystem/karuta/rnt.lfe
deleted file mode 100644
index dc0fbb7..0000000
--- a/subsystem/karuta/rnt.lfe
+++ /dev/null
@@ -1,26 +0,0 @@
-;; Tuples of an RNT relation, as Translator.ml writes and reads them: a
-;; tuple is #(Type #m(Name Value ...)), its type, names and strings
-;; binaries, and its integers no wider than an OCaml int.
-(defmodule rnt
-  (export (recv 1) (send 1)))
-
-;; A tuple from the bytes RNT sent.
-(defun recv (bytes) (holdable (binary_to_term bytes '(safe))))
-
-;; The bytes RNT reads a tuple from.
-(defun send (tuple) (term_to_binary (holdable tuple)))
-
-;; The tuple, if RNT can hold it.
-(defun holdable
-  (((= (tuple type attributes) tuple)) (when (is_binary type) (is_map attributes))
-   (if (lists:all (fun attribute? 1) (maps:to_list attributes))
-     tuple
-     (error 'badarg (list tuple))))
-  ((tuple) (error 'badarg (list tuple))))
-
-(defun attribute?
-  (((tuple name value)) (when (is_binary name) (is_binary value)) 'true)
-  (((tuple name value))
-   (when (is_binary name) (is_integer value) (>= value (- (bsl 1 62))) (< value (bsl 1 62)))
-   'true)
-  ((_) 'false))