aboutsummaryrefslogtreecommitdiffstats
path: root/subsystem
diff options
context:
space:
mode:
Diffstat (limited to 'subsystem')
-rw-r--r--subsystem/dune2
-rw-r--r--subsystem/karuta/Beam.ml311
-rw-r--r--subsystem/karuta/rnt-node.lfe74
-rw-r--r--subsystem/karuta/rnt.erl27
-rw-r--r--subsystem/karuta/rnt.lfe26
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))
generated by cgit v1.2.3 (git 2.25.1) at 2026-11-04 04:42:24 +0000