diff options
Diffstat (limited to 'bin')
| -rw-r--r-- | bin/main.ml | 17 | ||||
| -rw-r--r-- | bin/server.ml | 287 |
2 files changed, 44 insertions, 260 deletions
diff --git a/bin/main.ml b/bin/main.ml deleted file mode 100644 index 6b2a068..0000000 --- a/bin/main.ml +++ /dev/null @@ -1,17 +0,0 @@ -let main () = - let digest = - Sha256.to_hex (Sha256.string "sakura") - in - print_endline digest - -let () = - let tuple = - Relational_engine.Tuple.make_materialized ~relation:"natural_less_than" - ~attributes: - (Relational_engine.Tuple.AttributeMap.of_list - [ ("left", Obj.magic 0); ("right", Obj.magic 1) ]) - in - print_endline @@ string_of_bool - @@ Relational_engine.Prelude.Standard.less_than_natural.membership_criteria - tuple; - main () diff --git a/bin/server.ml b/bin/server.ml index 82965dc..1d4d5b3 100644 --- a/bin/server.ml +++ b/bin/server.ml @@ -1,160 +1,21 @@ open Relational_engine -let default_port = 7777 -let default_limit = 50 +module B = Management.Physical.Memory +module T = Transport.TCP +module Listener = Listener.Make(T)(B) -(* TODO: AbstractValue carries no type tag of its own; we recover the OCaml - runtime tag here via Obj introspection. This is fragile — it breaks the - moment a domain stores a boxed integer or a custom block. The right fix is - a typed value representation in Conventions. *) - -(* TODO: Generator.Error is silently treated as end-of-stream here, hiding - real failures from the client. *) -let materialize_generator gen limit = - let rec go gen pos acc count = - if count >= limit then (List.rev acc, true) - else - match gen (Some pos) with - | Generator.Done -> (List.rev acc, false) - | Generator.Error _ -> (List.rev acc, false) - | Generator.Value (t, next) -> ( - let mat = - match t with - | Tuple.Materialized m -> Some m - | Tuple.NonMaterialized _ -> None - in - match mat with - | None -> go next (pos + 1) acc count - | Some m -> go next (pos + 1) (m :: acc) (count + 1)) - in - go gen 0 [] 0 - -let materialize_relation storage (rel : Relation.t) limit = - let gen = Algebra.Memory.to_generator storage rel in - materialize_generator gen limit - -let short_hash (h : string) = String.sub h 0 (min 8 (String.length h)) - -module BranchOps = Management.Branch.Make (Management.Physical.Memory) -module Dispatch = Dbms_language.Memory - -let get_branch storage = - match BranchOps.get_head storage with Ok (Some name) -> name | _ -> "--" - -let tuple_to_sexp (t : Tuple.materialized) = - let open Sexplib.Sexp in - List - (Tuple.AttributeMap.bindings t.Tuple.attributes - |> List.map (fun (k, attr) -> - List - [ Atom k; Conventions.AbstractValue.sexp_of_t attr.Attribute.value ]) - ) - -let relation_to_sexp storage db_name db_hash (rel : Relation.t) limit = - let open Sexplib.Sexp in - let schema_sexp = - List (List.map (fun (a, d) -> List [ Atom a; Atom d ]) rel.Relation.schema) - in - let tuples, truncated = materialize_relation storage rel limit in - let rows_sexp = List (List.map tuple_to_sexp tuples) in - to_string - @@ List - [ - Atom "relation"; - List [ Atom "name"; Atom rel.Relation.name ]; - List [ Atom "schema"; schema_sexp ]; - List [ Atom "rows"; rows_sexp ]; - List [ Atom "row_count"; Atom (string_of_int (List.length tuples)) ]; - List [ Atom "truncated"; Atom (string_of_bool truncated) ]; - List [ Atom "db_hash"; Atom (short_hash db_hash) ]; - List [ Atom "db_name"; Atom db_name ]; - List [ Atom "branch"; Atom (get_branch storage) ]; - ] - -let ok_response storage db_name db_hash msg = - let open Sexplib.Sexp in - to_string - @@ List - [ - Atom "ok"; - List [ Atom "message"; Atom msg ]; - List [ Atom "db_hash"; Atom (short_hash db_hash) ]; - List [ Atom "db_name"; Atom db_name ]; - List [ Atom "branch"; Atom (get_branch storage) ]; - ] - -let error_response storage db_name db_hash err_sexp = - let open Sexplib.Sexp in - to_string - @@ List - [ - Atom "error"; - err_sexp; - List [ Atom "db_hash"; Atom (short_hash db_hash) ]; - List [ Atom "db_name"; Atom db_name ]; - List [ Atom "branch"; Atom (get_branch storage) ]; - ] - -let advance_head_branch storage new_hash = - match BranchOps.get_head storage with - | Ok (Some branch_name) -> - ignore (BranchOps.update_tip storage ~name:branch_name ~tip:new_hash) - | _ -> () - -let dbms_dispatch = - Dispatch.create - [ - (module Drl.Sublanguage.Memory : Dispatch.SubS); - (module Ddl.Sublanguage.Memory : Dispatch.SubS); - (module Dml.Sublanguage.Memory : Dispatch.SubS); - (module Icl.Sublanguage.Memory : Dispatch.SubS); - (module Dcl.Sublanguage.Memory : Dispatch.SubS); - (module Scl.Sublanguage.Memory : Dispatch.SubS); - ] - -let cursor_to_sexp storage db_name db_hash cursor_id rows has_more = - let open Sexplib.Sexp in - let row_sexps = List.map tuple_to_sexp rows in - let rows_sexp = List row_sexps in - to_string - @@ List - [ - Atom "cursor"; - List [ Atom "id"; Atom cursor_id ]; - List [ Atom "rows"; rows_sexp ]; - List [ Atom "row_count"; Atom (string_of_int (List.length row_sexps)) ]; - List [ Atom "has_more"; Atom (string_of_bool has_more) ]; - List [ Atom "db_hash"; Atom (short_hash db_hash) ]; - List [ Atom "db_name"; Atom db_name ]; - List [ Atom "branch"; Atom (get_branch storage) ]; - ] - -let execute_command storage db_ref cmd = - let db = !db_ref in - let h = db.Management.Database.hash in - let name = db.Management.Database.name in - let ok = ok_response storage name in - let err = error_response storage name in - match Dispatch.execute dbms_dispatch storage db cmd with - | Ok (Sublanguage.Query rel) -> - relation_to_sexp storage name h rel default_limit - | Ok (Sublanguage.Transition (new_db, msg)) -> - db_ref := new_db; - advance_head_branch storage new_db.Management.Database.hash; - ok new_db.Management.Database.hash msg - | Ok (Sublanguage.Cursor { cursor_id; rows; has_more }) -> - cursor_to_sexp storage name h cursor_id rows has_more - | Error e -> err h (Dispatch.sexp_of_dispatch_error e) +module Memory = Manipulation.Make (B) (* TODO: registration failures are silently ignored; a broken prelude relation leaves the catalog in a partially-seeded state with no indication to the user. *) +(* TODO: think of a better way to initialize new databases *) let register_prelude_relations storage db = let open Prelude.Standard in List.fold_left (fun db (rel : Relation.t) -> match - Manipulation.Memory.create_immutable_relation storage db ~name:rel.name + Memory.create_immutable_relation storage db ~name:rel.name ~schema:rel.schema ~generator:(Option.get rel.generator) ~membership_criteria:rel.membership_criteria ~cardinality:rel.cardinality @@ -163,7 +24,7 @@ let register_prelude_relations storage db = | Error e -> Printf.eprintf "Warning: failed to register prelude relation %s: %s\n%!" rel.name - (Manipulation.string_of_error e); + (Sexplib.Sexp.to_string (Error.sexp_of_error e)); db) db [ @@ -179,43 +40,27 @@ let register_prelude_relations storage db = divide_natural; ] -let print_with_time str = - let now = Unix.gettimeofday () in - let tm = Unix.localtime now in - let orange = "\027[38;5;208m" in - let reset = "\027[0m" in - let formatted_time = - Printf.sprintf "%04d-%02d-%02d %02d:%02d:%02d" (tm.Unix.tm_year + 1900) - (tm.Unix.tm_mon + 1) tm.Unix.tm_mday tm.Unix.tm_hour tm.Unix.tm_min - tm.Unix.tm_sec - in - print_endline @@ Printf.sprintf "%s[%s]%s %s" orange formatted_time reset str - -(* TODO: single-threaded accept loop — one slow or hung client blocks all - others. Should use threads or non-blocking I/O. *) -let handle_client storage db_ref fd = - let ic = Unix.in_channel_of_descr fd in - let oc = Unix.out_channel_of_descr fd in - try - while true do - let line = input_line ic in - let line = String.trim line in - if line <> "" then begin - let response = - try execute_command storage db_ref line - with e -> - let db = !db_ref in - error_response storage db.Management.Database.name - db.Management.Database.hash - Sexplib.Sexp.( - List [ Atom "uncaught-error"; Atom (Printexc.to_string e) ]) - in - print_with_time response; - output_string oc (response ^ "\n"); - flush oc - end - done - with End_of_file | Unix.Unix_error _ -> () +(* let register_branch_relations storage db = *) +(* (\* Register ephemeral sakura:branch and sakura:head relations backed by *) +(* the raw branch registry — no stored tuples, no circularity. *\) *) +(* let register db rel = *) +(* match *) +(* Memory.create_immutable_relation storage db *) +(* ~name:rel.Relation.name ~schema:rel.Relation.schema *) +(* ~generator:(Option.get rel.Relation.generator) *) +(* ~membership_criteria:rel.Relation.membership_criteria *) +(* ~cardinality:rel.Relation.cardinality *) +(* with *) +(* | Ok (db, _) -> db *) +(* | Error e -> *) +(* Printf.eprintf "Warning: failed to register %s: %s\n%!" *) +(* rel.Relation.name *) +(* (Manipulation.Error.string_of_error e); *) +(* db *) +(* in *) +(* db *) +(* |> Fun.flip register (BranchOps.branch_relation storage) *) +(* |> Fun.flip register (BranchOps.head_relation storage) *) let setup () = let ( let* ) = Result.bind in @@ -224,71 +69,27 @@ let setup () = |> Result.map_error (Fun.const "Failed to create storage") in let* db = - Manipulation.Memory.create_database storage ~name:"sakura" + Memory.create_database storage ~name:"sakura" |> Result.map_error (Fun.const "Failed to create initial database") in Ok (storage, db) -let register_branch_relations storage db = - (* Register ephemeral sakura:branch and sakura:head relations backed by - the raw branch registry — no stored tuples, no circularity. *) - let register db rel = - match - Manipulation.Memory.create_immutable_relation storage db - ~name:rel.Relation.name ~schema:rel.Relation.schema - ~generator:(Option.get rel.Relation.generator) - ~membership_criteria:rel.Relation.membership_criteria - ~cardinality:rel.Relation.cardinality - with - | Ok (db, _) -> db - | Error e -> - Printf.eprintf "Warning: failed to register %s: %s\n%!" - rel.Relation.name - (Manipulation.string_of_error e); - db - in - db - |> Fun.flip register (BranchOps.branch_relation storage) - |> Fun.flip register (BranchOps.head_relation storage) - -let warn_on_error fmt = function Ok () -> () | Error e -> Printf.eprintf fmt e -let cursor_gc_max_age = 300.0 (* 5 minutes *) -let cursor_gc_interval = 60.0 (* run GC at most once per minute *) +let address = "127.0.0.1" +let port = 7777 let () = match setup () with - | Error msg -> failwith msg + | Error msg -> Printf.eprintf "Couldn't initialize: %s" msg | Ok (storage, db) -> - let db = - db - |> register_prelude_relations storage - |> register_branch_relations storage - in - (* Initialise cursor session registry for SCL *) - let session_reg = Session.create () in - Scl.Executor.set_sessions session_reg; - (* Create default master branch in raw registry and set HEAD *) - BranchOps.create storage ~name:"master" ~tip:db.Management.Database.hash - |> warn_on_error "Warning: failed to create master branch: %s\n%!"; - BranchOps.checkout storage "master" - |> warn_on_error "Warning: failed to checkout master: %s\n%!"; - let db_ref = ref db in - let last_gc = ref 0.0 in - let sock = Unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in - Unix.setsockopt sock Unix.SO_REUSEADDR true; - Unix.setsockopt sock Unix.SO_REUSEPORT true; - Unix.bind sock (Unix.ADDR_INET (Unix.inet_addr_loopback, default_port)); - Unix.listen sock 5; - Printf.printf "Sakura server listening on 127.0.0.1:%d\n" default_port; - Printf.printf "Database hash: %s\n%!" - (short_hash db.Management.Database.hash); - while true do - let client_fd, _addr = Unix.accept sock in - handle_client storage db_ref client_fd; - (try Unix.close client_fd with _ -> ()); - let now = Unix.gettimeofday () in - if now -. !last_gc >= cursor_gc_interval then begin - Session.gc session_reg ~max_age:cursor_gc_max_age; - last_gc := now - end - done + let db = + db + |> register_prelude_relations storage + (* |> register_branch_relations storage *) + in + let transport = + T.create + { address_family = Unix.PF_INET; + address = Unix.ADDR_INET (Unix.inet_addr_of_string address, port) } + in + Printf.printf "Listening on %s:%d" address port; print_newline (); + Listener.run transport db storage |