parse and eval for beam evaluation

1 files changed, 112 insertions, 7 deletions

diff --git a/subsystem/karuta/Beam.ml b/subsystem/karuta/Beam.ml
index f46a131..8b63dc3 100644
--- a/subsystem/karuta/Beam.ml
+++ b/subsystem/karuta/Beam.ml
@@ -4,11 +4,23 @@ 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 unimplemented () =
-    condition "beam-unimplemented" "The beam evaluator cannot do this yet" empty
+  let malformed () =
+    condition "malformed-beam-program"
+      "A beam program is not {Module, Function, #{Name => Domain}} of binaries" empty
 
-  let taken () = condition "beam-taken" "An evaluator is already registered as karuta:beam" 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
 end
 
 module type TRANSPORT = sig
@@ -60,23 +72,116 @@ type program =
 
 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 3 source with
+  | Some ([module_; function_], 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_; heading})
+      else Error (Error.malformed ())
+  | _ -> Error (Error.malformed ())
+
 module Make (T : TRANSPORT) = struct
+  module Poll = struct
+    let invoke_message id {module_; function_; _} =
+      let buffer = Buffer.create 64 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.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
+
+    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'}
+
+    let await transport =
+      let open Utilities.Result in
+      let* message = T.receive transport in
+      match if message = "" then ' ' else message.[0] with
+      | 'b' -> batch_of message
+      | 'e' -> Error (Error.failed (String.sub message 1 (String.length message - 1)))
+      | _ -> Error (Error.unexpected ())
+
+    class t transport =
+      object (self)
+        inherit Kernel.Lifecycle.counted
+        inherit Kernel.Identity.of_id
+        method to_string = name
+
+        method fetch limit =
+          Result.bind
+            (T.send transport (fetch_message (Oo.id self) limit))
+            (fun () -> await transport)
+
+        method destroy = ignore (T.send transport (close_message (Oo.id self)))
+        method protocols : Protocols.Handle.protocol list = Protocols.[Cursor.make self]
+      end
+
+    let start transport program =
+      let poll = new t transport in
+      T.send transport (invoke_message (Oo.id poll) program)
+      |> Result.map (fun () -> Protocols.Handle.make poll)
+  end
+
   class evaluator (transport : T.t) =
     object (self)
       inherit Kernel.Lifecycle.counted
       inherit Kernel.Identity.of_id
       method to_string = name
       method destroy = T.stop transport
-
-      method parse : string -> (Kernel.Program.kind, Concepts.Condition.condition) result =
-        fun _ -> Error (Error.unimplemented ())
+      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 _ _ ~within:_ -> Error (Error.unimplemented ())
+        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 transport program) )
+          | Beam _, _ -> Error (Error.inputs ())
+          | _ -> Error (Error.not_a_program ())
 
       method protocols : Protocols.Handle.protocol list = [Protocols.Evaluator.make self]
     end