diff options
Diffstat (limited to '')
| -rw-r--r-- | bin/server.ml | 339 |
1 files changed, 176 insertions, 163 deletions
diff --git a/bin/server.ml b/bin/server.ml index 2b9e651..82965dc 100644 --- a/bin/server.ml +++ b/bin/server.ml @@ -15,16 +15,17 @@ let materialize_generator gen limit = 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)) + | 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 @@ -34,106 +35,116 @@ let materialize_relation storage (rel : Relation.t) 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 +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 - | _ -> "--" + 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])) + 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) + 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 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)]; - ] + 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)]; - ] + 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)]; - ] + 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) + 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 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)]; - ] + 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 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 + 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 + 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 + 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) + cursor_to_sexp storage name h cursor_id rows has_more + | Error e -> err h (Dispatch.sexp_of_dispatch_error e) (* TODO: registration failures are silently ignored; a broken prelude relation leaves the catalog in a partially-seeded state with no @@ -142,29 +153,30 @@ 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 - ~schema:rel.schema - ~generator:(Option.get rel.generator) - ~membership_criteria:rel.membership_criteria - ~cardinality:rel.cardinality + match + Manipulation.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 with | Ok (new_db, _) -> new_db - | Error e -> - Printf.eprintf "Warning: failed to register prelude relation %s: %s\n%!" - rel.name (Manipulation.string_of_error e); - db) + | Error e -> + Printf.eprintf + "Warning: failed to register prelude relation %s: %s\n%!" rel.name + (Manipulation.string_of_error e); + db) db - [ less_than_natural - ; less_than_or_equal_natural - ; greater_than_natural - ; greater_than_or_equal_natural - ; equal_natural - ; not_equal_natural - ; plus_natural - ; times_natural - ; minus_natural - ; divide_natural + [ + less_than_natural; + less_than_or_equal_natural; + greater_than_natural; + greater_than_or_equal_natural; + equal_natural; + not_equal_natural; + plus_natural; + times_natural; + minus_natural; + divide_natural; ] let print_with_time str = @@ -173,9 +185,9 @@ let print_with_time str = 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 + 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 @@ -184,25 +196,26 @@ let print_with_time str = 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 _ -> ()) + 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 setup () = let ( let* ) = Result.bind in @@ -220,62 +233,62 @@ 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 + 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 + 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 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 () = match setup () with | Error msg -> failwith 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 + (* 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 |