beam subsystem hack for karuta execution under internal nodeHEADmaster

1 files changed, 414 insertions, 0 deletions

diff --git a/subsystem/karuta/Beam.ml b/subsystem/karuta/Beam.ml
new file mode 100644
index 0000000..82ae827
--- /dev/null
+++ b/subsystem/karuta/Beam.ml
@@ -0,0 +1,414 @@
+open Rnt
+
+module Error = struct
+  open Concepts.Condition
+
+  let closed () = condition "beam-closed" "The BEAM node is gone" empty
+  let taken () = condition "beam-taken" "An evaluator is already registered as karuta:beam" empty
+
+  let malformed () =
+    condition "malformed-beam-program"
+      "A beam program is not {Module, Function, Beam, #{Name => Domain}} of binaries" empty
+
+  let not_a_program () =
+    condition "not-a-beam-program" "The evaluator was given a program of another kind" empty
+
+  let inputs () = condition "beam-program-inputs" "A beam program takes no inputs" empty
+
+  let failed message =
+    condition "beam-failed" "The BEAM node failed to answer a call"
+      ("message" |=| Concepts.Value.String message)
+
+  let unexpected () =
+    condition "unexpected-beam-message" "The BEAM node said something out of turn" empty
+
+  let unnamed () = condition "unnamed-beam-program" "A program was defined at an empty path" empty
+
+  let program_taken path =
+    condition "beam-program-taken" "A program is already defined at the path"
+      ("path" |=| Concepts.Value.String (Kernel.Path.to_string path))
+
+  let unknown_cursor id =
+    condition "unknown-beam-cursor" "The BEAM node named a cursor it does not hold"
+      ("cursor" |=| Concepts.Value.Integer id)
+end
+
+module type TRANSPORT = sig
+  type t
+
+  val send : t -> string -> (unit, Concepts.Condition.condition) result
+  val receive : t -> (string, Concepts.Condition.condition) result
+  val stop : t -> unit
+end
+
+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) land 0xffff_ffff))
+    with End_of_file | Sys_error _ -> Error (Error.closed ())
+
+  let start ?(erl = "erl") directories =
+    let boot =
+      "try [{module, _} = code:load_abs(filename:rootname(F)) || D <- init:get_plain_arguments(), \
+       F <- filelib:wildcard(D ++ \"/*.beam\")], rnt:start() catch C:R:S -> \
+       io:format(standard_error, \"~p:~p~n~p~n\", [C, R, S]), halt(1) end."
+    in
+    match
+      Unix.open_process_args erl
+        (Array.of_list ([erl; "-noshell"; "-noinput"; "-eval"; boot; "-extra"] @ directories))
+    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 name = "karuta"
+
+type program =
+  { module_: string;
+    function_: string;
+    beam: string;
+    heading: Protocols.Schematics.relation_description }
+
+type Kernel.Program.kind += Beam of program
+
+let decode source =
+  let domain _ = function
+    | Concepts.Value.String domain -> Some {Protocols.Schematics.domain; provenance= []}
+    | Concepts.Value.Integer _ -> None
+  in
+  match Translator.read 4 source with
+  | Some ([module_; function_; beam], attributes)
+    when String.length module_ <= 0xffff && String.length function_ <= 0xffff ->
+      let heading = BatMap.String.filter_map domain attributes in
+      if BatMap.String.cardinal heading = BatMap.String.cardinal attributes then
+        Ok (Beam {module_; function_; beam; heading})
+      else Error (Error.malformed ())
+  | _ -> Error (Error.malformed ())
+
+(* Both ends speak over one pipe, and each serves the other's requests
+   while it waits for its own answer, so a BEAM program RNT is reading
+   can call back into RNT, and so on, as deep as the calls nest.
+
+   RNT asks the node:
+   - h                                 run every loaded rnt_init/0  -> o | e
+   - i Id:64 M:16 Module F:16 Function B:32 Beam
+                                       start Module:Function() from Beam
+   - f Id:64 Limit:32                  the next batch of it         -> b | e
+   - c Id:64                           forget it
+   The node asks RNT:
+   - D Path Evaluator:16 Source        define a program at Path     -> o | e
+   - X Path                            invoke the program at Path   -> r Id:64 | e
+   - E Evaluator:16 Source             invoke a program of no path  -> r Id:64 | e
+   - N Id:64 Limit:32                  the next batch of its result -> b | e
+   - C Id:64                           release that result
+   A Path is Count:16 then each segment as Length:16 Segment, and a
+   batch b is Exhausted:8 then each tuple as Size:32 term_to_binary. *)
+module Make (T : TRANSPORT) = struct
+  (* [root] is borrowed, not held: the evaluator lives under it, so
+     holding it would keep both alive forever. [cursors] are the
+     results the node invoked and has not finished reading. *)
+  type node =
+    { transport: T.t;
+      root: Protocols.Handle.t;
+      cursors: (int, Protocols.Handle.t * Protocols.Handle.t) Hashtbl.t;
+      mutable next: int }
+
+  let node transport root = {transport; root; cursors= Hashtbl.create 8; next= 0}
+
+  module Request = struct
+    type t =
+      | Define of string list * string * string
+      | Invoke of string list
+      | Evaluate of string * string
+      | Next of int * int
+      | Close of int
+
+    exception Short
+
+    let of_message message =
+      let length = String.length message in
+      let at = ref 1 in
+      let take n =
+        if !at + n > length then raise Short;
+        let p = !at in
+        at := p + n;
+        p
+      in
+      let u16 () = String.get_uint16_be message (take 2) in
+      let u32 () = Int32.to_int (String.get_int32_be message (take 4)) land 0xffff_ffff in
+      let u64 () = Int64.to_int (String.get_int64_be message (take 8)) in
+      let sized () =
+        let n = u16 () in
+        String.sub message (take n) n
+      in
+      let path () = List.init (u16 ()) (fun _ -> sized ()) in
+      let finished request = if !at = length then Some request else None in
+      match message.[0] with
+      | 'D' ->
+          let path = path () in
+          let evaluator = sized () in
+          Some (Define (path, evaluator, String.sub message !at (length - !at)))
+      | 'E' ->
+          let evaluator = sized () in
+          Some (Evaluate (evaluator, String.sub message !at (length - !at)))
+      | 'X' -> finished (Invoke (path ()))
+      | 'N' ->
+          let id = u64 () in
+          finished (Next (id, u32 ()))
+      | 'C' -> finished (Close (u64 ()))
+      | _ -> None
+      | exception Short -> None
+  end
+
+  let batch_message {Protocols.Cursor.tuples; exhausted} =
+    let buffer = Buffer.create 64 in
+    Buffer.add_char buffer 'b';
+    Buffer.add_char buffer (if exhausted then '\001' else '\000');
+    BatFingerTree.iter
+      (fun tuple ->
+        let term = Translator.encode tuple in
+        Buffer.add_int32_be buffer (Int32.of_int (String.length term));
+        Buffer.add_string buffer term )
+      tuples;
+    Buffer.contents buffer
+
+  let cursor_message id =
+    let buffer = Buffer.create 9 in
+    Buffer.add_char buffer 'r';
+    Buffer.add_int64_be buffer (Int64.of_int id);
+    Buffer.contents buffer
+
+  let forget node id =
+    Hashtbl.find_opt node.cursors id
+    |> Option.iter (fun (relation, cursor) ->
+        Hashtbl.remove node.cursors id;
+        Protocols.Handle.release cursor;
+        Protocols.Handle.release relation )
+
+  let define node path evaluator source =
+    let open Utilities.Result in
+    match List.rev path with
+    | [] -> Error (Error.unnamed ())
+    | key :: directory -> (
+        let* program = Kernel.Program.load node.root ~evaluator ~source in
+        match
+          Kernel.Path.update node.root
+            (Kernel.Path.of_list (List.rev directory))
+            key None (Some program)
+        with
+        | Ok true -> Ok "o"
+        | Ok false ->
+            Protocols.Handle.release program;
+            Error (Error.program_taken (Kernel.Path.of_list path))
+        | Error c -> Protocols.Handle.release program; Error c )
+
+  (* Runs [program], which it releases, keeping a cursor on the result
+     for the node to read. *)
+  let invoke node program =
+    let open Utilities.Result in
+    let* relation =
+      Protocols.Handle.releasing program (fun () ->
+          let* executable = Protocols.Executable.require program in
+          Protocols.Executable.invoke executable [] ~within:node.root )
+    in
+    match Protocols.Enumerable.cursor relation with
+    | Error c -> Protocols.Handle.release relation; Error c
+    | Ok cursor ->
+        let id = node.next in
+        node.next <- id + 1;
+        Hashtbl.replace node.cursors id (relation, cursor);
+        Ok (cursor_message id)
+
+  let fetch node id limit =
+    let open Utilities.Result in
+    let* _, cursor =
+      Hashtbl.find_opt node.cursors id |> Option.to_result ~none:(Error.unknown_cursor id)
+    in
+    let* scan = Protocols.Cursor.require cursor in
+    let* batch = Protocols.Cursor.fetch scan limit in
+    if batch.exhausted then forget node id;
+    Ok (batch_message batch)
+
+  let serve node request =
+    let answer = function
+      | Ok message -> T.send node.transport message
+      | Error c -> T.send node.transport ("e" ^ Concepts.Condition.to_string_hum c)
+    in
+    match request with
+    | Request.Define (path, evaluator, source) -> answer (define node path evaluator source)
+    | Request.Invoke path ->
+        answer (Result.bind (Kernel.Path.lookup node.root (Kernel.Path.of_list path)) (invoke node))
+    | Request.Evaluate (evaluator, source) ->
+        answer (Result.bind (Kernel.Program.load node.root ~evaluator ~source) (invoke node))
+    | Request.Next (id, limit) -> answer (fetch node id limit)
+    | Request.Close id -> Ok (forget node id)
+
+  let batch_of message =
+    let open Utilities.Result in
+    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
+          let* tuple = Translator.decode (String.sub message (at + 4) size) in
+          tuples (at + 4 + size) (tuple :: read)
+    in
+    if length < 2 then Error (Error.unexpected ())
+    else
+      let* read = tuples 2 [] in
+      Ok {Protocols.Cursor.tuples= BatFingerTree.of_list read; exhausted= message.[1] <> '\000'}
+
+  (* The node's answer to what RNT last asked, serving whatever the
+     node asks in between. *)
+  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' -> Result.map (fun batch -> `Batch batch) (batch_of message)
+    | 'o' when message = "o" -> Ok `Done
+    | 'e' -> Error (Error.failed (String.sub message 1 (String.length message - 1)))
+    | _ -> (
+      match Request.of_message message with
+      | None -> Error (Error.unexpected ())
+      | Some request ->
+          let* () = serve node request in
+          await node )
+
+  (* Lets the node's rnt_init/0 functions define their programs. *)
+  let handshake node =
+    let open Utilities.Result in
+    let* () = T.send node.transport "h" in
+    let* reply = await node in
+    match reply with `Done -> Ok () | `Batch _ -> Error (Error.unexpected ())
+
+  module Poll = struct
+    let invoke_message id {module_; function_; beam; _} =
+      let buffer = Buffer.create (64 + String.length beam) in
+      Buffer.add_char buffer 'i';
+      Buffer.add_int64_be buffer (Int64.of_int id);
+      List.iter
+        (fun name ->
+          Buffer.add_uint16_be buffer (String.length name);
+          Buffer.add_string buffer name )
+        [module_; function_];
+      Buffer.add_int32_be buffer (Int32.of_int (String.length beam));
+      Buffer.add_string buffer beam;
+      Buffer.contents buffer
+
+    let fetch_message id limit =
+      let buffer = Buffer.create 13 in
+      Buffer.add_char buffer 'f';
+      Buffer.add_int64_be buffer (Int64.of_int id);
+      Buffer.add_int32_be buffer (Int32.of_int (max 0 (min limit 0xffff_ffff)));
+      Buffer.contents buffer
+
+    let close_message id =
+      let buffer = Buffer.create 9 in
+      Buffer.add_char buffer 'c';
+      Buffer.add_int64_be buffer (Int64.of_int id);
+      Buffer.contents buffer
+
+    class t node =
+      object (self)
+        inherit Kernel.Lifecycle.counted
+        inherit Kernel.Identity.of_id
+        method to_string = name
+
+        method fetch limit =
+          let open Utilities.Result in
+          let* () = T.send node.transport (fetch_message (Oo.id self) limit) in
+          let* reply = await node in
+          match reply with `Batch batch -> Ok batch | `Done -> Error (Error.unexpected ())
+
+        method destroy = ignore (T.send node.transport (close_message (Oo.id self)))
+        method protocols : Protocols.Handle.protocol list = Protocols.[Cursor.make self]
+      end
+
+    let start node program =
+      let poll = new t node in
+      T.send node.transport (invoke_message (Oo.id poll) program)
+      |> Result.map (fun () -> Protocols.Handle.make poll)
+  end
+
+  class evaluator node =
+    object (self)
+      inherit Kernel.Lifecycle.counted
+      inherit Kernel.Identity.of_id
+      method to_string = name
+
+      method destroy =
+        Hashtbl.to_seq_keys node.cursors |> List.of_seq |> List.iter (forget node);
+        T.stop node.transport
+
+      method parse source = decode source
+
+      method eval :
+          Kernel.Program.kind ->
+          Protocols.Handle.t list ->
+          within:Protocols.Handle.t ->
+          (Protocols.Handle.t, Concepts.Condition.condition) result =
+        fun kind inputs ~within:_ ->
+          match kind, inputs with
+          | Beam program, [] ->
+              Ok
+                (Kernel.EphemeralRelation.instantiate program.heading
+                   ~modes:(fun () -> Ok [Concepts.Mode.mode [] Concepts.Cardinality.Countable])
+                   (Kernel.EphemeralRelation.temporary program.heading)
+                   (fun () -> Poll.start node program) )
+          | Beam _, _ -> Error (Error.inputs ())
+          | _ -> Error (Error.not_a_program ())
+
+      method protocols : Protocols.Handle.protocol list = [Protocols.Evaluator.make self]
+    end
+end
+
+module Node = Make (Port)
+
+(* \karuta\program, where the node can define programs of its own. *)
+let programs root =
+  let open Utilities.Result in
+  let* found = Kernel.Path.find root Kernel.Path.(name @/ "program" @/ this) in
+  match found with
+  | Some programs -> Ok (Protocols.Handle.release programs)
+  | None -> (
+      let programs = Kernel.Namespace.make () in
+      match Kernel.Path.update root Kernel.Path.(name @/ this) "program" None (Some programs) with
+      | Ok true -> Ok ()
+      | Ok false -> Ok (Protocols.Handle.release programs)
+      | Error c -> Protocols.Handle.release programs; Error c )
+
+let register root directories =
+  let open Utilities.Result in
+  let* port = Port.start directories in
+  let node = Node.node port root in
+  let evaluator = Protocols.Handle.make (new Node.evaluator node) in
+  let bound =
+    let* () = Kernel.Initialization.mount root name in
+    let* () = programs root in
+    Kernel.Path.update root Kernel.Path.(name @/ "evaluator" @/ this) "beam" None (Some evaluator)
+  in
+  match bound with
+  | Ok true -> Node.handshake node
+  | Ok false ->
+      Protocols.Handle.release evaluator;
+      Error (Error.taken ())
+  | Error c ->
+      Protocols.Handle.release evaluator;
+      Error c