diff options
Diffstat (limited to 'subsystem/karuta')
| -rw-r--r-- | subsystem/karuta/Beam.ml | 341 | ||||
| -rw-r--r-- | subsystem/karuta/Translator.ml | 19 | ||||
| -rw-r--r-- | subsystem/karuta/rnt-node.lfe | 86 | ||||
| -rw-r--r-- | subsystem/karuta/rnt.lfe | 26 |
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)) |