aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--bin/server.ml32
-rw-r--r--lib/scl/ast.ml7
-rw-r--r--lib/scl/executor.ml77
-rw-r--r--lib/scl/parser.ml19
-rw-r--r--lib/scl/sublanguage.ml26
-rw-r--r--lib/session.ml82
-rw-r--r--lib/sublanguage.ml7
7 files changed, 249 insertions, 1 deletions
diff --git a/bin/server.ml b/bin/server.ml
index 5a59e2e..2b9e651 100644
--- a/bin/server.ml
+++ b/bin/server.ml
@@ -99,8 +99,24 @@ let dbms_dispatch = Dispatch.create [
(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
@@ -114,6 +130,8 @@ let execute_command storage db_ref cmd =
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)
@@ -223,6 +241,9 @@ 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
@@ -231,12 +252,16 @@ let () =
|> 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;
@@ -247,5 +272,10 @@ let () =
while true do
let (client_fd, _addr) = Unix.accept sock in
handle_client storage db_ref client_fd;
- (try Unix.close client_fd with _ -> ())
+ (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
diff --git a/lib/scl/ast.ml b/lib/scl/ast.ml
new file mode 100644
index 0000000..043f544
--- /dev/null
+++ b/lib/scl/ast.ml
@@ -0,0 +1,7 @@
+open Sexplib.Std
+
+type statement =
+ | Begin of { query : Drl.Ast.query; limit : int option }
+ | Fetch of { cursor : string; limit : int option }
+ | Close of { cursor : string }
+[@@deriving sexp]
diff --git a/lib/scl/executor.ml b/lib/scl/executor.ml
new file mode 100644
index 0000000..6cb17b8
--- /dev/null
+++ b/lib/scl/executor.ml
@@ -0,0 +1,77 @@
+let default_batch = 50
+
+type cursor_batch = {
+ cursor_id : string;
+ rows : Tuple.materialized list;
+ has_more : bool;
+}
+
+type exec_result =
+ | Batch of cursor_batch
+ | Closed of Management.Database.t
+
+let sessions : Session.t option ref = ref None
+
+let set_sessions s = sessions := Some s
+
+module Make (Storage : Management.Physical.S) = struct
+ module DrlExec = Drl.Executor.Make(Storage)
+ module Alg = Algebra.Make(Storage)
+
+ type error =
+ | ParseError of string
+ | QueryError of DrlExec.error
+ | CursorError of string
+
+ let sexp_of_error e =
+ let open Sexplib.Sexp in
+ match e with
+ | ParseError s -> List [Atom "parse-error"; Atom s]
+ | QueryError e -> DrlExec.sexp_of_error e
+ | CursorError s -> List [Atom "cursor-error"; Atom s]
+
+ let get_sessions () =
+ match !sessions with
+ | Some s -> Ok s
+ | None -> Error (CursorError "The session registry has not been initialised; ensure `set_sessions` is called at startup.")
+
+ let ( let* ) = Result.bind
+
+ let execute
+ (storage : Storage.t)
+ (db : Management.Database.t)
+ (stmt : Ast.statement)
+ : (exec_result, error) result =
+ match stmt with
+ | Ast.Begin { query; limit } ->
+ let* reg = get_sessions () in
+ let* rel =
+ DrlExec.execute storage db query
+ |> Result.map_error (fun e -> QueryError e)
+ in
+ let gen = Alg.to_generator storage rel in
+ let query_str = Drl.Parser.to_string query in
+ let cur = Session.register reg ~db ~query:query_str ~generator:gen in
+ let batch = Option.value ~default:default_batch limit in
+ let* (rows, has_more) =
+ Session.fetch reg ~id:cur.Session.id ~limit:batch
+ |> Result.map_error (fun s -> CursorError s)
+ in
+ Ok (Batch { cursor_id = cur.Session.id; rows; has_more })
+
+ | Ast.Fetch { cursor; limit } ->
+ let* reg = get_sessions () in
+ let batch = Option.value ~default:default_batch limit in
+ let* (rows, has_more) =
+ Session.fetch reg ~id:cursor ~limit:batch
+ |> Result.map_error (fun s -> CursorError s)
+ in
+ Ok (Batch { cursor_id = cursor; rows; has_more })
+
+ | Ast.Close { cursor } ->
+ let* reg = get_sessions () in
+ Session.close reg ~id:cursor;
+ Ok (Closed db)
+end
+
+module Memory = Make(Management.Physical.Memory)
diff --git a/lib/scl/parser.ml b/lib/scl/parser.ml
new file mode 100644
index 0000000..3e9c4ed
--- /dev/null
+++ b/lib/scl/parser.ml
@@ -0,0 +1,19 @@
+type error = ParseError of string
+
+let validate = function
+ | (Ast.Begin { limit = Some limit; _ } | Ast.Fetch { limit = Some limit; _ })
+ when limit <= 0 ->
+ Error (ParseError "The provision of `limit` is expected to be positive.")
+ | stmt -> Ok stmt
+
+let of_sexp sexp =
+ match Ast.statement_of_sexp sexp with
+ | exception exn -> Error (ParseError (Printexc.to_string exn))
+ | stmt -> validate stmt
+
+let of_string s =
+ match Sexplib.Sexp.of_string s |> Ast.statement_of_sexp with
+ | exception exn -> Error (ParseError (Printexc.to_string exn))
+ | stmt -> validate stmt
+
+let to_string stmt = Ast.sexp_of_statement stmt |> Sexplib.Sexp.to_string_hum
diff --git a/lib/scl/sublanguage.ml b/lib/scl/sublanguage.ml
new file mode 100644
index 0000000..7098ced
--- /dev/null
+++ b/lib/scl/sublanguage.ml
@@ -0,0 +1,26 @@
+module Make (Storage : Management.Physical.S) = struct
+ module Exec = Executor.Make(Storage)
+
+ type storage = Storage.t
+ type ast = Ast.statement
+ type error = Exec.error
+
+ let name = "scl"
+
+ let parse_sexp sexp =
+ match Parser.of_sexp sexp with
+ | Ok r -> Ok r
+ | Error (Parser.ParseError s) -> Error (Exec.ParseError s)
+
+ let execute storage db ast =
+ Exec.execute storage db ast
+ |> Result.map (function
+ | Executor.Batch { cursor_id; rows; has_more } ->
+ Sublanguage.Cursor { cursor_id; rows; has_more }
+ | Executor.Closed db ->
+ Sublanguage.Transition (db, "cursor closed"))
+
+ let sexp_of_error = Exec.sexp_of_error
+end
+
+module Memory = Make(Management.Physical.Memory)
diff --git a/lib/session.ml b/lib/session.ml
new file mode 100644
index 0000000..272a419
--- /dev/null
+++ b/lib/session.ml
@@ -0,0 +1,82 @@
+(** Cursor registry for streaming tuple consumption.
+
+ Each cursor wraps a [Generator.t] produced by a DRL query and
+ tracks the current position so that successive [fetch] calls
+ return the next batch of rows without replaying the query. *)
+
+type cursor = {
+ id : string;
+ mutable generator : Generator.t;
+ mutable position : int;
+ db_snapshot : Management.Database.t;
+ query_sexp : string;
+ created_at : float;
+}
+
+type t = {
+ tbl : (string, cursor) Hashtbl.t;
+ mutable counter : int;
+}
+
+let create () : t =
+ { tbl = Hashtbl.create 16; counter = 0 }
+
+let register (reg : t) ~(db : Management.Database.t) ~(query : string)
+ ~(generator : Generator.t) : cursor =
+ let raw = string_of_int reg.counter ^ query ^ db.Management.Database.hash in
+ let id = Conventions.Hash.hash_text raw in
+ reg.counter <- reg.counter + 1;
+ let cur = {
+ id;
+ generator;
+ position = 0;
+ db_snapshot = db;
+ query_sexp = query;
+ created_at = Unix.gettimeofday ();
+ } in
+ Hashtbl.replace reg.tbl id cur;
+ cur
+
+let fetch (reg : t) ~(id : string) ~(limit : int)
+ : (Tuple.materialized list * bool, string) result =
+ match Hashtbl.find_opt reg.tbl id with
+ | None -> Error ("The cursor with identifier `" ^ id ^ "` was not found in the session registry.")
+ | Some cur ->
+ let rec go gen pos acc count =
+ if count >= limit then (List.rev acc, gen, pos, true)
+ else
+ match gen (Some pos) with
+ | Generator.Done -> (List.rev acc, gen, pos, false)
+ (* Generator errors are treated as exhaustion: the partial batch is
+ returned with has_more = false and the cursor is auto-removed.
+ The caller receives no indication that the stream terminated
+ abnormally. When fetch gains a richer return type this arm
+ should surface the error message instead. *)
+ | Generator.Error _ -> (List.rev acc, gen, pos, false)
+ | Generator.Value (t, next) ->
+ begin match t with
+ | Tuple.Materialized m ->
+ go next (pos + 1) (m :: acc) (count + 1)
+ | Tuple.NonMaterialized _ ->
+ go next (pos + 1) acc count
+ end
+ in
+ let (rows, gen', pos', has_more) =
+ go cur.generator cur.position [] 0
+ in
+ cur.generator <- gen';
+ cur.position <- pos';
+ if not has_more then Hashtbl.remove reg.tbl id;
+ Ok (rows, has_more)
+
+let close (reg : t) ~(id : string) : unit =
+ Hashtbl.remove reg.tbl id
+
+let gc (reg : t) ~(max_age : float) : unit =
+ let now = Unix.gettimeofday () in
+ let to_remove =
+ Hashtbl.fold (fun k cur acc ->
+ if now -. cur.created_at >= max_age then k :: acc else acc)
+ reg.tbl []
+ in
+ List.iter (Hashtbl.remove reg.tbl) to_remove
diff --git a/lib/sublanguage.ml b/lib/sublanguage.ml
index 3df91f1..d92358e 100644
--- a/lib/sublanguage.ml
+++ b/lib/sublanguage.ml
@@ -14,9 +14,16 @@
(Turing-complete; breaks the termination guarantee of DRL)
- ACL -> a functor restriction: narrows the admissible natural
transformations based on authorization context *)
+type cursor_result = {
+ cursor_id : string;
+ rows : Tuple.materialized list;
+ has_more : bool;
+}
+
type result =
| Query of Relation.t
| Transition of Management.Database.t * string
+ | Cursor of cursor_result
module type S = sig
type storage
generated by cgit v1.2.3 (git 2.25.1) at 2026-11-04 03:48:01 +0000