lfe with a beam evaluator for karuta

5 files changed, 412 insertions, 28 deletions

diff --git a/subsystem/dune b/subsystem/dune
index 9ff6f50..0148f5e 100644
--- a/subsystem/dune
+++ b/subsystem/dune
@@ -3,4 +3,4 @@
 (library
  (name subsystem)
  (public_name rnt.subsystem)
- (libraries rnt batteries))
+ (libraries rnt batteries unix))
diff --git a/subsystem/karuta/Beam.ml b/subsystem/karuta/Beam.ml
new file mode 100644
index 0000000..2e1fc0f
--- /dev/null
+++ b/subsystem/karuta/Beam.ml
@@ -0,0 +1,311 @@
+(* 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
+     both  b Exhausted:8 (Size:32 Tuple)...                     a batch
+     both  e Message                                            a failure
+
+   fetch and next are answered with a batch or a failure, invoke and
+   close not at all, so a function that cannot be applied fails the
+   first fetch. Calls nest: until its answer comes, whoever waits serves
+   the calls that come instead. *)
+
+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 unheld_input () =
+    condition "unheld-input" "An input could not be held for as long as the program needs it" empty
+end
+
+module type TRANSPORT = sig
+  type t
+
+  val send : t -> string -> (unit, Concepts.Condition.condition) result
+  val receive : t -> (string, Concepts.Condition.condition) result
+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}
+
+  let send {output; _} message =
+    let header = Bytes.create 4 in
+    Bytes.set_int32_be header 0 (Int32.of_int (String.length message));
+    try
+      output_bytes output header;
+      output_string output message;
+      Ok (flush output)
+    with Sys_error _ -> Error (Error.closed ())
+
+  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)))
+    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 -> \
+       io:format(standard_error, \"~p:~p~n~p~n\", [C, R, S]), halt(1) end."
+    in
+    let input, output =
+      Unix.open_process_args erl
+        (Array.of_list ([erl; "-noshell"; "-noinput"; "-eval"; boot; "-extra"] @ directories))
+    in
+    {input; output}
+
+  let stop {input; output} = ignore (Unix.close_process (input, output))
+end
+
+let subsystem = "karuta"
+let language = "beam"
+let name = subsystem ^ ":" ^ language
+
+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 failure_message condition = "e" ^ Concepts.Condition.to_string_hum condition
+
+let batch_of message =
+  let length = String.length message in
+  let rec tuples at read =
+    if at = length then Ok (List.rev read)
+    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
+        Result.bind
+          (Translator.decode (String.sub message (at + 4) size))
+          (fun tuple -> tuples (at + 4 + size) (tuple :: read))
+  in
+  Result.map
+    (fun read ->
+      {Protocols.Cursor.tuples= BatFingerTree.of_list read; exhausted= message.[1] <> '\000'} )
+    (tuples 2 [])
+
+module Make (T : TRANSPORT) = struct
+  (* The cursors over its inputs that 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 = {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 rec await node =
+    let open Utilities.Result in
+    let* message = T.receive node.transport 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 ->
+        let answer =
+          match next node message with
+          | Ok batch -> batch_message batch
+          | Error c -> failure_message c
+        in
+        let* () = T.send node.transport answer in
+        await node
+    | _ -> 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
+
+  class cursor node id readers =
+    object (self)
+      inherit Kernel.Lifecycle.counted
+      inherit Kernel.Identity.of_id
+      method to_string = "beam-cursor"
+
+      method fetch limit =
+        Result.bind (T.send node.transport (fetch_message id limit)) (fun () -> await node)
+
+      method destroy =
+        ignore (T.send node.transport (close_message id));
+        forget node readers
+
+      method protocols : Protocols.Handle.protocol list = Protocols.[Cursor.make self]
+    end
+
+  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
+
+  let invoke node program inputs =
+    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))
+    | 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
+
+  (* 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 =
+    let open Utilities.Result in
+    let* inputs = hold inputs in
+    Ok
+      (Kernel.EphemeralRelation.instantiate ~inputs program.heading
+         ~modes:(fun () -> Ok [Concepts.Mode.mode [] Concepts.Cardinality.Finite])
+         (Kernel.EphemeralRelation.temporary program.heading)
+         (fun () -> invoke node program inputs) )
+
+  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
+        | _ -> Error (Error.not_a_program ())
+
+      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 = {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)
diff --git a/subsystem/karuta/rnt-node.lfe b/subsystem/karuta/rnt-node.lfe
new file mode 100644
index 0000000..b7bffc8
--- /dev/null
+++ b/subsystem/karuta/rnt-node.lfe
@@ -0,0 +1,74 @@
+;; 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.
+(defmodule rnt-node
+  (export (start 0)))
+
+(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))
+
+;; A function that cannot be applied leaves a relation that fails when read.
+(defun serve
+  (((binary #\i (cursor (size 32)) (m (size 16)) (module binary (size m))
+            (f (size 16)) (func binary (size f)) (readers binary)))
+   (put (tuple 'cursor cursor)
+        (try
+          (apply (binary_to_existing_atom module) (binary_to_existing_atom func)
+                 (lc ((<= (binary (reader (size 32))) readers)) (relation reader)))
+          (catch
+            ((tuple class reason stack) (lambda (_) (erlang:raise class reason stack)))))))
+  (((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)))))
+
+;; An input, 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, 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 #\e (message binary)) (error (tuple 'rnt message)))
+    (call (serve call) (answer))))
diff --git a/subsystem/karuta/rnt.erl b/subsystem/karuta/rnt.erl
deleted file mode 100644
index c94a01c..0000000
--- a/subsystem/karuta/rnt.erl
+++ /dev/null
@@ -1,27 +0,0 @@
-%% Tuples of an RNT relation, as Translator.ml writes and reads them:
-%% a tuple is {Type, #{Name => Value}}, its type, names, bitstrings
-%% and its integers no wider than an OCaml int.
--module(rnt).
--export([recv/1, send/1]).
-
--define(MIN_INT, -(1 bsl 62)).
--define(MAX_INT, (1 bsl 62) - 1).
-
-%% A tuple from the bytes RNT sent
-recv(Bytes) -> holdable(binary_to_term(Bytes, [safe])).
-
-%% The bytes RNT reads a tuple from
-send(Tuple) -> term_to_binary(holdable(Tuple)).
-
-%% The tuple, if RNT can hold it
-holdable({Type, Attributes} = Tuple) when is_binary(Type), is_map(Attributes) ->
-    case lists:all(fun attribute/1, maps:to_list(Attributes)) of
-        true -> Tuple;
-        false -> error(badarg, [Tuple])
-    end;
-holdable(Tuple) -> error(badarg, [Tuple]).
-
-attribute({Name, Value}) when is_binary(Name), is_binary(Value) -> true;
-attribute({Name, Value}) when is_binary(Name), is_integer(Value) ->
-    Value >= ?MIN_INT andalso Value =< ?MAX_INT;
-attribute(_) -> false.
diff --git a/subsystem/karuta/rnt.lfe b/subsystem/karuta/rnt.lfe
new file mode 100644
index 0000000..dc0fbb7
--- /dev/null
+++ b/subsystem/karuta/rnt.lfe
@@ -0,0 +1,26 @@
+;; 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))