The unpac monorepo manager self-hosting as a monorepo using unpac
Merge pull request #64 from talex5/fix-sketch
Add a readv benchmark and fix sketch buffer GC
authored by
Christiano Haesbaert
and committed by
GitHub
af3811ca
3846a702
+171
-13
+6
bench/dune
+6
bench/dune
+62
bench/readv.ml
+62
bench/readv.ml
···
1
+
(* This benchmark uses [Uring.readv] to read from /dev/zero in a loop.
2
+
This is intended to stress the sketch buffer. *)
3
+
4
+
let buffer_size = 100 (* Use a small buffer to stress the system more *)
5
+
let n_concurrent = 16 (* How many requests to have active at once *)
6
+
let n_iters = 1_000_000 (* How many times to accept and resubmit *)
7
+
8
+
let rec wait t handle =
9
+
match Uring.peek t with
10
+
| Some { result; data = buf } -> handle result buf
11
+
| None ->
12
+
match Uring.wait t with
13
+
| None -> wait t handle
14
+
| Some { result; data = buf } -> handle result buf
15
+
16
+
let run_bechmark ~polling_timeout fd =
17
+
let got = ref 0 in
18
+
(* For polling mode, [queue_depth] needs to be slightly larger than [n_concurrent] or submission
19
+
occasionally fails for some reason. *)
20
+
let t = Uring.create ?polling_timeout ~queue_depth:(n_concurrent * 2) () in
21
+
(* We start by submitting [n_concurrent] reads. *)
22
+
for _ = 1 to n_concurrent do
23
+
let buf = Cstruct.create buffer_size in
24
+
let _job : _ Uring.job = Uring.readv t fd [buf] ~file_offset:Optint.Int63.zero [buf] |> Option.get in
25
+
()
26
+
done;
27
+
assert (Uring.submit t = n_concurrent);
28
+
(* Then we wait for reads to complete. Each time a read returns, we immediately submit another. *)
29
+
let t0 = Unix.gettimeofday () in
30
+
for _ = 1 to n_iters do
31
+
wait t (fun result bufs ->
32
+
if result < 0 then (
33
+
raise (Unix.Unix_error (Uring.error_of_errno result, "readv", ""))
34
+
) else (
35
+
got := !got + result;
36
+
let _job : _ Uring.job = Uring.readv t fd bufs ~file_offset:Optint.Int63.zero bufs |> Option.get in
37
+
()
38
+
)
39
+
)
40
+
done;
41
+
(* Get a snapshot of the stats before letting things finish. *)
42
+
let stats = Uring.get_debug_stats t in
43
+
let t1 = Unix.gettimeofday () in
44
+
let time = t1 -. t0 in
45
+
let got = float !got /. (1024. *. 1024.) in
46
+
let polling = polling_timeout <> None in
47
+
Fmt.pr "@[<v2>Read %.2f MB in %.2f seconds (%.2f MB/second) # buffer_size=%d, polling=%b@,%a@]@."
48
+
got time (got /. time) buffer_size polling
49
+
Uring.Stats.pp stats;
50
+
(* Finally, drain the remaining reads and shut down the ring. *)
51
+
for _ = 1 to n_concurrent do
52
+
wait t (fun _result _buf -> ())
53
+
done;
54
+
Uring.exit t
55
+
56
+
let () =
57
+
let fd = Unix.openfile "/dev/zero" Unix.[O_RDONLY] 0 in
58
+
run_bechmark fd ~polling_timeout:None;
59
+
run_bechmark fd ~polling_timeout:(Some 1000);
60
+
run_bechmark fd ~polling_timeout:None;
61
+
run_bechmark fd ~polling_timeout:(Some 1000);
62
+
Unix.close fd
+57
-10
lib/uring/uring.ml
+57
-10
lib/uring/uring.ml
···
133
133
134
134
let create_buffer len = Bigarray.(Array1.create char c_layout len)
135
135
136
-
let create len =
137
-
{ buffer = create_buffer len; off = 0; old_buffers = [] }
136
+
let create () =
137
+
{ buffer = Cstruct.empty.buffer; off = 0; old_buffers = [] }
138
138
139
139
let length t = Bigarray.Array1.size_in_bytes t.buffer
140
140
···
146
146
let alloc t alloc_len =
147
147
let alloc_len = round alloc_len in
148
148
if alloc_len > avail t then begin
149
-
let new_buffer = create_buffer ((length t) + alloc_len) in
149
+
(* At least 64 bytes, at least twice the previous size, and
150
+
at least big enough for the new allocation. *)
151
+
let new_size = max 64 (max (2 * length t) alloc_len) in
152
+
let new_buffer = create_buffer new_size in
150
153
t.old_buffers <- t.buffer :: t.old_buffers;
151
154
t.off <- 0;
152
155
t.buffer <- new_buffer;
···
288
291
let data = Heap.create queue_depth in
289
292
let id = object end in
290
293
let fixed_iobuf = Cstruct.empty.buffer in
291
-
let sketch = Sketch.create 0 in
294
+
let sketch = Sketch.create () in
292
295
let t = { id; uring; sketch; fixed_iobuf; data; dirty=false; queue_depth } in
293
296
register_gc_root t;
294
297
t
···
413
416
ignore (Heap.ptr job : Uring.id); (* Check it's still valid *)
414
417
with_id t (fun id -> Uring.submit_cancel t.uring id (Heap.ptr job)) user_data
415
418
419
+
(* Free stale entries in the sketch buffer, if possible.
420
+
This isn't quite right: a busy system might never have 0 unsubmitted entries.
421
+
We should probably track how many requests need to be submitted before each
422
+
of [t.sketch.old_buffers] can be released, but this will do for now. *)
423
+
let gc_sketch t =
424
+
if Uring.sq_ready t.uring = 0 then Sketch.release t.sketch
425
+
416
426
let submit t =
417
427
let v =
418
428
if t.dirty then begin
···
421
431
end else
422
432
0
423
433
in
424
-
if Uring.sq_ready t.uring = 0 then
425
-
Sketch.release t.sketch;
434
+
(* In non-polling mode, we will almost always be able to free the sketch buffer here.
435
+
However, in polling mode it's unlikely the entries have been consumed by the kernel yet,
436
+
and we must rely on other GC points. *)
437
+
gc_sketch t;
426
438
v
427
439
428
440
type 'a completion_option =
···
436
448
let data = Heap.free t.data user_data_id in
437
449
Some { result = res; data }
438
450
439
-
let peek t = fn_on_ring Uring.peek_cqe t
451
+
let peek t =
452
+
gc_sketch t;
453
+
fn_on_ring Uring.peek_cqe t
440
454
441
455
let wait ?timeout t =
442
-
match timeout with
443
-
| None -> fn_on_ring Uring.wait_cqe t
444
-
| Some timeout -> fn_on_ring (Uring.wait_cqe_timeout timeout) t
456
+
let r =
457
+
match timeout with
458
+
| None -> fn_on_ring Uring.wait_cqe t
459
+
| Some timeout -> fn_on_ring (Uring.wait_cqe_timeout timeout) t
460
+
in
461
+
(* In polling mode, this is a good time to GC the sketch buffer, because the
462
+
kernel has probably consumed all the enties while we were blocking. *)
463
+
gc_sketch t;
464
+
r
445
465
446
466
let queue_depth {queue_depth;_} = queue_depth
447
467
let buf {fixed_iobuf;_} = fixed_iobuf
448
468
449
469
let error_of_errno e =
450
470
Uring.error_of_errno (abs e)
471
+
472
+
module Stats = struct
473
+
type t = {
474
+
sqe_ready : int;
475
+
active_ops : int;
476
+
sketch_buffer_size : int;
477
+
sketch_used : int;
478
+
sketch_old_buffers : int;
479
+
}
480
+
481
+
let pp f { sqe_ready; active_ops; sketch_used; sketch_buffer_size; sketch_old_buffers } =
482
+
Fmt.pf f "@[<v>SQEs ready: %d@,\
483
+
Operations active: %d@,\
484
+
Sketch buffer: %d/%d (plus %d old buffers)@]"
485
+
sqe_ready
486
+
active_ops
487
+
sketch_used sketch_buffer_size sketch_old_buffers
488
+
end
489
+
490
+
let get_debug_stats t =
491
+
{ Stats.
492
+
sqe_ready = Uring.sq_ready t.uring;
493
+
active_ops = Heap.in_use t.data;
494
+
sketch_used = t.sketch.off;
495
+
sketch_buffer_size = Bigarray.Array1.dim t.sketch.buffer;
496
+
sketch_old_buffers = List.length t.sketch.old_buffers;
497
+
}
+15
lib/uring/uring.mli
+15
lib/uring/uring.mli
···
271
271
val error_of_errno : int -> Unix.error
272
272
(** [error_of_errno e] converts the error code [abs e] to a Unix error type. *)
273
273
274
+
module Stats : sig
275
+
type t = {
276
+
sqe_ready : int;
277
+
active_ops : int;
278
+
sketch_buffer_size : int;
279
+
sketch_used : int;
280
+
sketch_old_buffers : int;
281
+
}
282
+
283
+
val pp : t Fmt.t
284
+
end
285
+
286
+
val get_debug_stats : _ t -> Stats.t
287
+
(** [get_debug_stats t] collects some metrics about the internal state of [t]. *)
288
+
274
289
module Private : sig
275
290
module Heap = Heap
276
291
end
+1
-1
lib/uring/uring_stubs.c
+1
-1
lib/uring/uring_stubs.c
+30
-2
tests/main.md
+30
-2
tests/main.md
···
153
153
```ocaml
154
154
# let queue_depth = 5;;
155
155
val queue_depth : int = 5
156
+
156
157
# let t = Uring.create ~queue_depth ();;
157
158
val t : '_weak1 Uring.t = <abstr>
159
+
160
+
# Fmt.pr "%a@." Uring.Stats.pp (Uring.get_debug_stats t);;
161
+
SQEs ready: 0
162
+
Operations active: 0
163
+
Sketch buffer: 0/0 (plus 0 old buffers)
164
+
- : unit = ()
165
+
158
166
# for i = 1 to queue_depth do
159
167
assert (Option.is_some (Uring.noop t i));
160
-
done;
161
-
Uring.submit t;;
168
+
done;;
169
+
- : unit = ()
170
+
171
+
# Fmt.pr "%a@." Uring.Stats.pp (Uring.get_debug_stats t);;
172
+
SQEs ready: 5
173
+
Operations active: 5
174
+
Sketch buffer: 0/0 (plus 0 old buffers)
175
+
- : unit = ()
176
+
177
+
# Uring.submit t;;
162
178
- : int = 5
163
179
180
+
# Fmt.pr "%a@." Uring.Stats.pp (Uring.get_debug_stats t);;
181
+
SQEs ready: 0
182
+
Operations active: 5
183
+
Sketch buffer: 0/0 (plus 0 old buffers)
184
+
- : unit = ()
185
+
164
186
# for i = 1 to queue_depth do
165
187
let tkn, res = consume t in
166
188
traceln "%d returned %d" tkn res;
···
170
192
3 returned 0
171
193
4 returned 0
172
194
5 returned 0
195
+
- : unit = ()
196
+
197
+
# Fmt.pr "%a@." Uring.Stats.pp (Uring.get_debug_stats t);;
198
+
SQEs ready: 0
199
+
Operations active: 0
200
+
Sketch buffer: 0/0 (plus 0 old buffers)
173
201
- : unit = ()
174
202
175
203
# Uring.exit t;;