diff options
| -rw-r--r-- | bin/main.ml | 24 | ||||
| -rw-r--r-- | flake.nix | 6 | ||||
| -rw-r--r-- | lib/disk.ml | 64 | ||||
| -rw-r--r-- | lib/dune | 1 | ||||
| -rw-r--r-- | test/test_relational_engine.ml | 16 |
5 files changed, 100 insertions, 11 deletions
diff --git a/bin/main.ml b/bin/main.ml index a21c63e..8250c44 100644 --- a/bin/main.ml +++ b/bin/main.ml @@ -1,4 +1,28 @@ +open Relational_engine.Disk + +let x() = + let command_write1: Command.t = + {kind = Command.WRITE; timestamp = 10.0; hash = "bca39cc55b19274133ce57d0bea1ff794589f025d4bcf670c44d5d2811837830"; content = "asd"; filename = "trash"} in + let command_write2: Command.t = + {kind = Command.WRITE; timestamp = 10.0; hash = "bca39cc55b19274133ce57d0bea1ff794589f025d4bcf670c44d5d2811837831"; content = "qwe"; filename = "trash2"} in + let command_write3: Command.t = + {kind = Command.WRITE; timestamp = 10.0; hash = "bca39cc55b19274133ce57d0bea1ff794589f025d4bcf670c44d5d2811837832"; content = "fgh"; filename = "trash"} in + let command_write4: Command.t = + {kind = Command.WRITE; timestamp = 10.0; hash = "bca39cc55b19274133ce57d0bea1ff794589f025d4bcf670c44d5d2811837833"; content = "rty"; filename = "trash2"} in + let command_write5: Command.t = + {kind = Command.WRITE; timestamp = 10.0; hash = "bca39cc55b19274133ce57d0bea1ff794589f025d4bcf670c44d5d2811837834"; content = "jkl"; filename = "trash"} in + let command_read: Command.t = {kind = Command.READ; timestamp = 0.0; hash = ""; content = ""; filename = "trash"} in + let stream: Executor.history = [{state = Executor._EMPTY_SHA_HASH_; files = Executor.StringMap.empty}] in + let locations: Executor.locations = Executor.StringMap.empty in + let open Relational_engine.Extensions.Result in + let+ (stream, locations) = Command.commit_and_perform stream locations command_write1 in + let+ (stream, locations) = Command.commit_and_perform stream locations command_write2 in + let+ (stream, locations) = Command.commit_and_perform stream locations command_write3 in + let+ (stream, locations) = Command.commit_and_perform stream locations command_write4 in + let+ (stream, locations) = Command.commit_and_perform stream locations command_write5 in + Command.commit_and_perform (List.tl stream) locations command_read let () = begin + let _ = x() in () end @@ -164,6 +164,7 @@ ocamlPackages.ocaml legacyPackages.ocamlformat ocamlPackages.ppx_inline_test + ocamlPackages.ppx_deriving ]; } '' echo "checking dune and ocaml formatting" @@ -233,7 +234,7 @@ legacyPackages.nixpkgs-fmt legacyPackages.ocamlformat # For `dune build --watch ...` - legacyPackages.fswatch + # legacyPackages.fswatch # For `dune build @doc` ocamlPackages.odoc # OCaml editor support @@ -248,9 +249,10 @@ ocamlPackages.ctypes-foreign ocamlPackages.data-encoding ocamlPackages.ppx_inline_test + ocamlPackages.ppx_deriving legacyPackages.cmake legacyPackages.gcc - legacyPackages.nixfmt-classic + # legacyPackages.nixfmt-classic ]; shellHook = '' diff --git a/lib/disk.ml b/lib/disk.ml index fc99607..b29ea09 100644 --- a/lib/disk.ml +++ b/lib/disk.ml @@ -22,6 +22,7 @@ module type FileSystem = sig | Storage | Transaction val append: target -> bytes -> (int*int64, string) result + val read: target -> int64 -> int -> (bytes, string) result end module FileSystem (C: Configuration): FileSystem = struct @@ -29,21 +30,44 @@ module FileSystem (C: Configuration): FileSystem = struct | Storage | Transaction let target_to_string = function - | Storage -> "storage" - | Transaction -> "transaction" + | Storage -> C.storage ^ "storage.re" + | Transaction -> C.transactions ^ "transaction.re" (** [append target content] writes content to file in append only format, returning [content_length * offset_written] *) let append target content = let register channel = - let offset_written = Out_channel.length channel in - Out_channel.seek channel offset_written; + let offset_written = + try + let stats = Unix.stat (target_to_string target) in + Int64.of_int stats.Unix.st_size + with _ -> 0L + in + (* Out_channel.seek channel offset_written; *) + (* let offset_written = Out_channel.length channel in *) Out_channel.output_bytes channel content; Out_channel.flush channel; Out_channel.close channel; Ok (Bytes.length content, offset_written) in - try Out_channel.with_open_bin (target_to_string target) register - with _ -> Error (Printf.sprintf "Error on appending to target '%s'" (target_to_string target)) + try + Out_channel.with_open_gen [Open_append; Open_binary; Open_creat] 0o666 (target_to_string target) register + with e -> + Error (Printf.sprintf "Error on appending to target '%s': %s" (target_to_string target) (Printexc.to_string e)) + + let read target (offset: int64) size = + print_endline ("READ_OFFSET: " ^ Int64.to_string offset); + print_endline ("READ_SIZE: " ^ (string_of_int ((Int64.to_int offset) + size))); + let register (channel: in_channel) = + let buffer = Bytes.create size in + In_channel.seek channel offset; + let _ = In_channel.really_input channel buffer 0 size in + print_bytes buffer; + print_newline (); + In_channel.close channel; + Ok buffer + in + try In_channel.with_open_gen [Open_binary; Open_rdonly] 0o666 (target_to_string target) register + with e -> Error (Printf.sprintf "Error on reading from target '%s': %s" (target_to_string target) (Printexc.to_string e)) end module Executor = struct @@ -54,11 +78,14 @@ module Executor = struct { offset: int64; size: int; hash: string } + [@@deriving show] end module Hashes = struct type t = { values: string list; hash: string } + [@@deriving show] type history = t list + [@@deriving show] end type commit = { state: string; files: Hashes.history StringMap.t } type history = commit list @@ -104,6 +131,7 @@ module Executor = struct in let open Extensions.Result in let+ (size, offset) = FS.append Storage content in + let () = print_endline @@ "OFFSET: " ^ (Int64.to_string offset) in let locations = StringMap.add computed_hash ({size; offset; hash = computed_hash}: Location.t) locations in let commit = {state = compose_new_state files; files} in Ok (commit, locations) @@ -120,6 +148,7 @@ module Executor = struct let files = StringMap.update filename update_fun files in let open Extensions.Result in let+ (size, offset) = FS.append Storage content in + let () = print_endline @@ "OFFSET: " ^ (Int64.to_string offset) in let locations = StringMap.add computed_hash ({size; offset; hash = computed_hash}: Location.t) locations in let commit = {state = compose_new_state files; files} in Ok (commit, locations) @@ -141,6 +170,14 @@ module Executor = struct | None -> Ok (commit, locations) end + + let read ({files; _} as _commit: commit) (locations: locations) ~filename = + let history: Hashes.history = StringMap.find filename files in + let {values = location_hashes; _}: Hashes.t = List.hd history in + let physical_locations = List.map (fun hash -> StringMap.find hash locations) location_hashes in + (* Terrible! Here I read everything opening a new IO every time. We should be able to read with only one call, but changing the offset every time *) + let content = List.map (fun ({offset; size; _}: Location.t) -> match FS.read FS.Storage offset size with | Ok x -> x | Error err -> failwith err) physical_locations + in Bytes.concat Bytes.empty @@ List.rev content end module Startup = struct @@ -191,13 +228,22 @@ module Command = struct (** TODO: This does not write atomically. If the system crashes while it attempts to write, it will corrupt. Solve this with a temporary file and move later. *) - let commit_and_perform (commit: Executor.commit) (locations: Executor.locations) command = + let commit_and_perform (stream: Executor.history) (locations: Executor.locations) command = let open Extensions.Result in let+ serialized_command = Result.map_error (fun _ -> "Failed to serialize command to binary format.") @@ Binary.to_bytes command_encoding command in let+ _ = FS.append FS.Transaction serialized_command in match command.kind with - | WRITE -> Executor.write commit locations ~filename:command.filename @@ Bytes.of_string command.content - | _ -> Ok (commit, locations) + | WRITE -> + let open Extensions.Result in + let+ (commit, locations) = Executor.write (List.hd stream) locations ~filename:command.filename @@ Bytes.of_string command.content in + let updated_stream = commit::stream in + Ok (updated_stream, locations) + | READ -> + let content = Executor.read (List.hd stream) locations ~filename:command.filename in + print_string "CONTENT: "; + print_endline @@ Bytes.to_string content; + Ok (stream, locations) + | _ -> Ok (stream, locations) end @@ -1,6 +1,7 @@ (library (name relational_engine) (libraries ctypes ctypes-foreign data-encoding) + (preprocess (pps ppx_deriving.show ppx_deriving.ord)) ;; (c_library_flags -lssl -lcrypto -L %{env:LIBRESSL_LIB_PATH=}) ;; (foreign_stubs ;; (language cxx) diff --git a/test/test_relational_engine.ml b/test/test_relational_engine.ml index 69a7df2..cc9652b 100644 --- a/test/test_relational_engine.ml +++ b/test/test_relational_engine.ml @@ -6,3 +6,19 @@ let%test _ = let (new_commit, _) = Result.get_ok @@ write (List.hd stream) locations ~filename:"trash" @@ Bytes.of_string "abc" in let expected_state_hash = "bca39cc55b19274133ce57d0bea1ff794589f025d4bcf670c44d5d2811837830" in new_commit.state = expected_state_hash + +let%test _ = + let open Relational_engine.Disk.Executor in + let stream: history = [{state = _EMPTY_SHA_HASH_; files = StringMap.empty}] in + let locations: locations = StringMap.empty in + let (new_commit, new_locations) = Result.get_ok @@ write (List.hd stream) locations ~filename:"trash" @@ Bytes.of_string "abc" in + let result_test = + let open Relational_engine.Extensions.Option in + let+ new_file_content = StringMap.find_opt "trash" new_commit.files in + let { values; _ }: Hashes.t = List.hd new_file_content in + StringMap.find_opt (List.hd values) new_locations + in match result_test with + | Some location -> + print_endline @@ Location.show location; + location.size = 3 && location.offset = 0L + | None -> false |