Merge pull request #24 from Cjen1/conspire

Conspire

.github/workflows/workflow.yml

@@ -8,41 +8,26 @@ on:
 
 jobs:
    build:
-    strategy:
-      fail-fast: false
-      matrix:
-        os:
-          - ubuntu-latest
-        ocaml-version:
-          - 4.10.1
-
-    runs-on: ${{ matrix.os }}
-
-    # Steps represent a sequence of tasks that will be executed as part of the job
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@v2
-
-      - name: Use OCaml ${{ matrix.ocaml-version }}
-        uses: avsm/setup-ocaml@v1
-        with:
-          ocaml-version: ${{ matrix.ocaml-version }}
-
-      - name: Opam setup
-        run: |
-          opam pin add ocons-core.dev . --no-action
-          opam depext ocons-core --yes --with-doc --with-test
-          opam pin add ocons-variants.dev . --no-action
-          opam depext ocons-variants --yes --with-doc --with-test
-          opam install . --deps-only --with-doc --with-test
-
-      - name: Dune build
-        run: opam exec -- dune build
-
-      - name: Run tests
-        run: make test
-
-      - name: Format
-        run: |
-          opam install ocamlformat --yes --with-doc --with-test
-          opam exec -- dune build @fmt
+     name: Check build
+     runs-on: ubuntu-22.04
+     steps:
+      - uses: actions/checkout@v3
+      - uses: DeterminateSystems/nix-installer-action@main
+      - uses: DeterminateSystems/magic-nix-cache-action@main
+      - run: nix develop -c dune build
+   format:
+     name: Check formatting
+     runs-on: ubuntu-22.04
+     steps:
+      - uses: actions/checkout@v3
+      - uses: DeterminateSystems/nix-installer-action@main
+      - uses: DeterminateSystems/magic-nix-cache-action@main
+      - run: nix develop -c dune build @fmt
+   runtest:
+     name: Run tests
+     runs-on: ubuntu-22.04
+     steps:
+      - uses: actions/checkout@v3
+      - uses: DeterminateSystems/nix-installer-action@main
+      - uses: DeterminateSystems/magic-nix-cache-action@main
+      - run: nix develop -c dune runtest

.gitignore

@@ -17,3 +17,5 @@ libpaxos/
 
 perf.data
 trace*
+
+result

README.md

@@ -21,15 +21,21 @@ dune exec paxos/main.exe -- ID 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:502
 ```
 
 ```
-EIO_BACKEND=linux dune exec impl/main.exe -- paxos 0 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5000 -q 5001 -t 1
-EIO_BACKEND=linux dune exec impl/main.exe -- paxos 1 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5010 -q 5011 -t 1
-EIO_BACKEND=linux dune exec impl/main.exe -- paxos 2 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5020 -q 5021 -t 1
-
-EIO_BACKEND=linux dune exec impl/main.exe -- raft 0 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5000 -q 5001 -t 1
-EIO_BACKEND=linux dune exec impl/main.exe -- raft 1 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5010 -q 5011 -t 1
-EIO_BACKEND=linux dune exec impl/main.exe -- raft 2 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5020 -q 5021 -t 1
+dune exec impl/main.exe -- paxos 0 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5000 -q 5001 -t 1
+dune exec impl/main.exe -- paxos 1 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5010 -q 5011 -t 1
+dune exec impl/main.exe -- paxos 2 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5020 -q 5021 -t 1
+
+dune exec impl/main.exe -- raft 0 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5000 -q 5001 -t 1
+dune exec impl/main.exe -- raft 1 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5010 -q 5011 -t 1
+dune exec impl/main.exe -- raft 2 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020 -p 5020 -q 5021 -t 1
+
+dune exec impl/main.exe -- conspire 0 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020,3:127.0.0.1:5030 -p 5000 -q 5001 -t 1
+dune exec impl/main.exe -- conspire 1 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020,3:127.0.0.1:5030 -p 5010 -q 5011 -t 1
+dune exec impl/main.exe -- conspire 2 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020,3:127.0.0.1:5030 -p 5020 -q 5021 -t 1
+dune exec impl/main.exe -- conspire 3 0:127.0.0.1:5000,1:127.0.0.1:5010,2:127.0.0.1:5020,3:127.0.0.1:5030 -p 5030 -q 5031 -t 1
 ```
 
 ```
 dune exec bin/cli.exe -- write 1 1 0 127.0.0.1:5001,127.0.0.1:5011,127.0.0.1:5021 -r 10
+dune exec bin/bench.exe -- 127.0.0.1:5001,127.0.0.1:5011,127.0.0.1:5021,127.0.0.1:5031 -r 1000 -n 20000
 ```

bin/bench.ml

@@ -5,23 +5,26 @@ module Cli = Ocons_core.Client
 module MT = Eio.Time.Mono
 open Eio.Std
 
-let pitcher ~sw mclock n rate cmgr (dispatch : Mtime.t array) :
+let pitcher ~sw nid mclock n rate cmgr (dispatch : (int, Mtime.t) Hashtbl.t) :
     unit Eio.Promise.t =
   let t, u = Promise.create () in
   let period =
     Mtime.Span.to_float_ns Mtime.Span.s /. rate
     |> Mtime.Span.of_float_ns |> Option.get
   in
-  let req_reporter = O.Utils.InternalReporter.rate_reporter 0 "req_dispatch" in
+  let req_reporter, should_run =
+    O.Utils.InternalReporter.rate_reporter "req_dispatch"
+  in
+  should_run := true ;
+  let prime = Utils.prime nid in
   let rec aux = function
     | i, _ when i >= n ->
         ()
     | i, prev ->
+        let id = i * prime in
         let cmd =
           Command.
-            { op= Write ("asdf", "asdf")
-            ; id= i
-            ; trace_start= Unix.gettimeofday () }
+            {op= Write ("asdf", "asdf"); id; trace_start= Unix.gettimeofday ()}
         in
         let target = Mtime.add_span prev period |> Option.get in
         if Mtime.is_later target ~than:(MT.now mclock) then
@@ -31,7 +34,7 @@ let pitcher ~sw mclock n rate cmgr (dispatch : Mtime.t array) :
             traceln "Failed to dispatch %a" Fmt.exn_backtrace
               (e, Printexc.get_raw_backtrace ()) ) ;
         req_reporter () ;
-        Array.set dispatch i (Eio.Time.Mono.now mclock) ;
+        Hashtbl.replace dispatch id (Eio.Time.Mono.now mclock) ;
         aux (i + 1, target)
   in
   Fiber.fork ~sw (fun () ->
@@ -51,15 +54,19 @@ let to_float_s =
     let s = Mtime.span min s in
     Mtime.Span.to_float_ns s /. Mtime.Span.to_float_ns Mtime.Span.s
 
-let latency_reporter = O.Utils.InternalReporter.avg_reporter Fun.id "lat"
-
-let catcher_iter mclock requests responses (_, (cid, _, _)) =
-  if Array.get responses cid |> Option.is_none then (
-    let t = Eio.Time.Mono.now mclock in
-    let latency = Mtime.span t (Array.get requests cid) |> to_float_ms in
-    if latency > 500. then Magic_trace.take_snapshot () ;
-    Array.set responses cid (Some t) ;
-    latency_reporter latency )
+let catcher_iter mclock requests responses =
+  let reporter, should_run =
+    O.Utils.InternalReporter.avg_reporter Fun.id "lat"
+  in
+  should_run := true ;
+  fun (_, (cid, _, tr)) ->
+    if not @@ Hashtbl.mem responses cid then (
+      Ocons_core.Utils.TRACE.ex_cli tr ;
+      let t = Eio.Time.Mono.now mclock in
+      let latency = Mtime.span t (Hashtbl.find requests cid) |> to_float_ms in
+      if latency > 500. then Magic_trace.take_snapshot () ;
+      Hashtbl.add responses cid t ;
+      reporter latency )
 
 let pp_stats ppf s =
   let s =
@@ -107,9 +114,9 @@ let pp_stats ppf s =
   Fmt.pf ppf "%a" pp_stats s
 
 let run sockaddrs id n rate outfile debug =
-  let dispatch = Array.init n (fun _ -> Mtime.of_uint64_ns Int64.zero) in
-  let response = Array.init n (fun _ -> None) in
-  let ( / ) = Eio.Path.( / ) in
+  Ocons_core.Utils.TRACE.run_ex_cli := true ;
+  let dispatch = Hashtbl.create n in
+  let response = Hashtbl.create n in
   if debug then Ocons_conn_mgr.set_debug_flag () ;
   let main env =
     Switch.run
@@ -119,8 +126,9 @@ let run sockaddrs id n rate outfile debug =
       sockaddrs
       |> List.mapi (fun idx addr ->
              ( idx
-             , fun sw -> (Eio.Net.connect ~sw env#net addr :> Eio.Flow.two_way)
-             ) )
+             , fun sw ->
+                 ( Eio.Net.connect ~sw env#net addr
+                   :> Eio.Flow.two_way_ty Eio.Flow.two_way ) ) )
     in
     traceln "Creating conns to: %a"
       Fmt.(braces @@ list ~sep:comma Eio.Net.Sockaddr.pp)
@@ -132,23 +140,24 @@ let run sockaddrs id n rate outfile debug =
         ~sw con_ress id
         (fun () -> Eio.Time.sleep env#clock 1.)
     in
-    let complete = pitcher ~sw env#mono_clock n rate cmgr dispatch in
+    Eio.Time.sleep env#clock 2. ;
+    let complete = pitcher ~sw id env#mono_clock n rate cmgr dispatch in
     Promise.await complete ;
-    Eio.Time.sleep env#clock 1. ;
+    Eio.Time.sleep env#clock 2. ;
+    traceln "Test complete" ;
     (* End of test *)
     let request_response_pairs =
-      let open Iter.Infix in
-      0 -- (n - 1)
-      |> Iter.filter (fun i -> Array.get response i |> Option.is_some)
+      response |> Hashtbl.to_seq_keys |> Iter.of_seq
       |> Iter.map (fun i ->
-             (i, (Array.get dispatch i, Array.get response i |> Option.get)) )
+             (i, (Hashtbl.find dispatch i, Hashtbl.find response i)) )
     in
     let responses =
       request_response_pairs |> Iter.to_seq_persistent |> Array.of_seq
     in
     traceln "Results: %a" pp_stats responses ;
     outfile
     |> Option.iter (fun path ->
+           let open Eio.Path in
            let path = Eio.Stdenv.cwd env / path in
            Eio.Path.with_open_out ~create:(`If_missing 0o777) path
            @@ fun out ->
@@ -170,8 +179,9 @@ let run sockaddrs id n rate outfile debug =
     Cli.Cmgr.close cmgr ;
     traceln "Closed everything, done bench"
   in
-  Eio_unix.Ctf.with_tracing "trace.ctf"
-  @@ fun () ->
+  (*Eio_unix.Ctf.with_tracing "trace.ctf"
+    @@ fun () ->
+  *)
   try Eio_main.run main
   with e -> Fmt.pr "%a" Fmt.exn_backtrace (e, Printexc.get_raw_backtrace ())
 
@@ -215,7 +225,8 @@ let cmd =
   let id_t =
     Arg.(
       required
-      & pos 0 (some int) None (info ~docv:"ID" ~doc:"The id of the client" []) )
+      & pos 0 (some int) None
+          (info ~docv:"ID" ~doc:"The id of the client (max 5)" []) )
   in
   let sockaddrs_t =
     Arg.(

bin/cli.ml

@@ -13,7 +13,7 @@ let run op sockaddrs id retry_timeout =
     sockaddrs
     |> List.mapi (fun idx addr ->
            ( idx
-           , fun sw -> (Eio.Net.connect ~sw env#net addr :> Eio.Flow.two_way) ) )
+           , fun sw -> (Eio.Net.connect ~sw env#net addr :> Eio.Flow.two_way_ty Eio.Flow.two_way) ) )
   in
   Eio.traceln "Creating conns to: %a"
     Fmt.(braces @@ list ~sep:comma Eio.Net.Sockaddr.pp)

bin/dune

@@ -1,4 +1,16 @@
-(executables
- (names cli bench echo)
- (modules cli bench echo)
- (libraries ocons.core cmdliner))
+(executable
+  (name cli)
+  (modules cli)
+  (libraries ocons.core cmdliner))
+(executable
+  (name bench)
+  (modules bench)
+  (libraries ocons.core cmdliner))
+(executable
+  (name echo)
+  (modules echo)
+  (libraries ocons.core cmdliner))
+(executable
+  (name test_alloc)
+  (modules test_alloc)
+  (libraries core core_bench core_unix.command_unix))

bin/echo.ml

@@ -1,11 +1,15 @@
 open! Eio.Std
 
-let set_nodelay sock =
-  match sock |> Eio_unix.FD.peek_opt with
-  | None ->
-      Fmt.invalid_arg "Could not get underlying file descriptor"
+let set_nodelay ?(should_warn = true) sock =
+  match sock |> Eio_unix.Resource.fd_opt with
   | Some fd ->
-      Unix.setsockopt fd Unix.TCP_NODELAY true
+      Eio_unix.Fd.use ~if_closed:ignore fd (fun fd ->
+          Unix.setsockopt fd Unix.TCP_NODELAY true )
+  | None ->
+      if should_warn then
+        Eio.traceln
+          "WARNING: unable to set TCP_NODELAY, higher than required latencies \
+           may be experienced"
 
 let server addr =
   Eio_main.run

bin/test_alloc.ml

@@ -0,0 +1,17 @@
+open Core
+open Core_bench
+
+type data = Nil | Node of data
+
+let rec make_data = function 0 -> Nil | n -> Node (make_data (n - 1))
+
+let alloc_test =
+  let test `init =
+    let large = make_data 1000 in
+    let extern = ref None in
+    fun () -> 
+      extern := Some large
+  in
+  Bench.Test.create_with_initialization ~name:"alloc_with_reference" test
+
+let () = Command_unix.run (Bench.make_command [alloc_test])

dune-project

@@ -11,7 +11,7 @@
   (name ocons)
   (synopsis "The infrastructure for pluggable consensus implementations")
   (depends 
-    (ocaml (= 5.0.0))
+    (ocaml (= 5.1.0))
     cmdliner
     iter
     accessor
@@ -24,8 +24,11 @@
     base
     core
     core_unix
-    eio
+    (eio (>= 0.11))
     eio_main
     eio_linux
     tdigest
+    irmin
+    irmin-containers
+    core_bench
     ))