The unpac monorepo manager self-hosting as a monorepo using unpac
wip
+231
-41
+6
dune
+6
dune
+43
urcat.ml
+43
urcat.ml
···
1
+
(* cat(1) built with liburing.
2
+
OCaml version of https://unixism.net/loti/tutorial/cat_liburing.html *)
3
+
4
+
let block_size = 1024
5
+
6
+
let get_file_size fd =
7
+
Unix.handle_unix_error Unix.fstat fd |>
8
+
fun {Unix.st_size; _} -> st_size
9
+
(* TODO make this work with ST_ISBLK *)
10
+
11
+
let get_completion_and_print uring =
12
+
let bufs, len = Uring.wait_cqe uring in
13
+
let remaining = ref len in
14
+
Printf.eprintf "%d bytes read\n%!" len;
15
+
Array.iter (fun buf ->
16
+
let buflen = Bigstringaf.length buf in
17
+
if !remaining > 0 then begin
18
+
if buflen <= !remaining then begin
19
+
print_string (Bigstringaf.to_string buf);
20
+
remaining := !remaining - buflen;
21
+
end else begin
22
+
print_string (Bigstringaf.substring ~off:0 ~len:!remaining buf);
23
+
remaining := 0;
24
+
end
25
+
end
26
+
) bufs
27
+
28
+
let submit_read_request fname uring =
29
+
let fd = Unix.(handle_unix_error (openfile fname [O_RDONLY]) 0) in
30
+
let file_sz = get_file_size fd in
31
+
let blocks = if file_sz mod block_size <> 0 then (file_sz / block_size)+1 else file_sz/block_size in
32
+
let bufs = Array.init blocks (fun _ -> Uring.iobuf_alloc block_size) in
33
+
Uring.submit_readv uring fd bufs;
34
+
let numreq = Uring.submit uring in
35
+
assert(numreq=1);
36
+
()
37
+
38
+
let () =
39
+
let fname = Sys.argv.(1) in
40
+
let uring = Uring.create ~queue_depth:1 () in
41
+
submit_read_request fname uring;
42
+
get_completion_and_print uring
43
+
+56
-3
uring.ml
+56
-3
uring.ml
···
1
-
type t
1
+
type uring
2
+
3
+
type iobuf = (char, Bigarray.int8_unsigned_elt, Bigarray.c_layout) Bigarray.Array1.t
4
+
5
+
let iobuf_alloc len = Bigarray.(Array1.create char c_layout len)
6
+
7
+
(** [ring_setup i] allocates a new io_uring with queue depth [i].
8
+
@raise [Failure] *)
9
+
external uring_create : int -> uring = "ocaml_uring_setup"
10
+
external uring_exit : uring -> unit = "ocaml_uring_exit"
11
+
(* external uring_register_bigarray : uring -> iobuf -> unit = "ocaml_uring_register_ba" *)
12
+
external uring_submit : uring -> int = "ocaml_uring_submit"
13
+
14
+
type iovecs
15
+
external uring_alloc_iovecs : iobuf array -> iovecs = "ocaml_uring_alloc_iovecs"
16
+
external uring_free_iovecs : iovecs -> unit = "ocaml_uring_free_iovecs"
17
+
18
+
external uring_submit_readv : uring -> Unix.file_descr -> iovecs -> int -> unit = "ocaml_uring_submit_readv"
19
+
external uring_wait_cqe : uring -> iovecs * int = "ocaml_uring_wait_cqe"
20
+
21
+
type t = {
22
+
uring: uring;
23
+
iobuf: iobuf;
24
+
pending: (iovecs, iobuf array) Hashtbl.t;
25
+
}
26
+
27
+
let default_iobuf_len = 1024 * 1024 (* 1MB *)
28
+
29
+
let create ~queue_depth () =
30
+
let uring = uring_create queue_depth in
31
+
(* TODO posix memalign this to page *)
32
+
let iobuf = Bigarray.(Array1.create char c_layout default_iobuf_len) in
33
+
(* uring_register_bigarray uring iobuf; *)
34
+
Gc.finalise uring_exit uring;
35
+
let pending = Hashtbl.create 1 in
36
+
{ uring; iobuf; pending }
37
+
38
+
let submit_readv {uring;pending;_} fd bufs =
39
+
let len = Array.length bufs in
40
+
let iovs = uring_alloc_iovecs bufs in
41
+
uring_submit_readv uring fd iovs len;
42
+
Hashtbl.add pending iovs bufs;
43
+
()
44
+
45
+
let submit {uring;_} =
46
+
uring_submit uring
2
47
3
-
external ring_setup : int -> t = "ring_setup"
48
+
let wait_cqe {uring;pending;_} =
49
+
let iovecs, len = uring_wait_cqe uring in
50
+
let bas = Hashtbl.find pending iovecs in
51
+
uring_free_iovecs iovecs;
52
+
bas, len
53
+
54
+
55
+
(*
4
56
external ring_queue_write_full : t -> Unix.file_descr -> (Bigstringaf.t -> int -> unit) -> Bigstringaf.t -> int -> unit = "ring_queue_write_full"
5
57
external ring_queue_read : t -> Unix.file_descr -> (Bigstringaf.t -> int -> unit) -> Bigstringaf.t -> int -> unit = "ring_queue_read"
6
58
external ring_queue_accept : t -> Unix.file_descr -> (Unix.file_descr -> unit) -> unit = "ring_queue_accept"
7
59
external ring_queue_close : t -> Unix.file_descr -> unit = "ring_queue_close"
8
60
external ring_submit : t -> int = "ring_submit"
9
61
external ring_exit : t -> unit = "ring_exit"
10
-
external ring_wait : t -> unit = "ring_wait"
62
+
external ring_wait : t -> unit = "ring_wait"
63
+
*)
+7
-8
uring.mli
+7
-8
uring.mli
···
1
+
1
2
type t
3
+
type iobuf = (char, Bigarray.int8_unsigned_elt, Bigarray.c_layout) Bigarray.Array1.t
2
4
3
-
val ring_setup : int -> t
4
-
val ring_queue_write_full : t -> Unix.file_descr -> (Bigstringaf.t -> int -> unit) -> Bigstringaf.t -> int -> unit
5
-
val ring_queue_read : t -> Unix.file_descr -> (Bigstringaf.t -> int -> unit) -> Bigstringaf.t -> int -> unit
6
-
val ring_queue_accept : t -> Unix.file_descr -> (Unix.file_descr -> unit) -> unit
7
-
val ring_queue_close : t -> Unix.file_descr -> unit
8
-
val ring_submit : t -> int
9
-
val ring_exit : t -> unit
10
-
val ring_wait : t -> unit
5
+
val iobuf_alloc : int -> iobuf
6
+
val create : queue_depth:int -> unit -> t
7
+
val submit_readv : t -> Unix.file_descr -> iobuf array -> unit
8
+
val submit : t -> int
9
+
val wait_cqe : t -> iobuf array * int
+102
-26
uring_stubs.c
+102
-26
uring_stubs.c
···
9
9
#include <caml/signals.h>
10
10
#include <string.h>
11
11
12
-
#define Ring_val(v) *((struct io_uring**)Data_custom_val(v));
12
+
#define Ring_val(v) *((struct io_uring**)Data_custom_val(v))
13
13
14
14
#define EVENT_TYPE_READ 0
15
15
#define EVENT_TYPE_WRITE 1
···
39
39
socklen_t socklen; // used by ACCEPT
40
40
};
41
41
42
-
value ring_setup(value entries) {
43
-
CAMLparam1(entries);
44
42
45
-
struct io_uring* ring = (struct io_uring*)malloc(sizeof(struct io_uring));
43
+
value ocaml_uring_setup(value entries) {
44
+
CAMLparam1(entries);
46
45
47
-
int status = io_uring_queue_init(Long_val(entries), ring, 0);
46
+
struct io_uring* ring = (struct io_uring*)caml_stat_alloc(sizeof(struct io_uring));
48
47
49
-
if( !status ) {
50
-
// Everything was set up fine
51
-
value ring_custom = caml_alloc_custom_mem(&ring_ops, sizeof(struct io_uring*), sizeof(struct io_uring));
48
+
int status = io_uring_queue_init(Long_val(entries), ring, 0);
52
49
53
-
*((struct io_uring**)Data_custom_val(ring_custom)) = ring;
50
+
if (status == 0) {
51
+
value ring_custom = caml_alloc_custom_mem(&ring_ops, sizeof(struct io_uring*), sizeof(struct io_uring));
52
+
*((struct io_uring**)Data_custom_val(ring_custom)) = ring;
53
+
CAMLreturn(ring_custom);
54
+
} else {
55
+
caml_failwith(strerror(-status));
56
+
}
57
+
}
54
58
55
-
CAMLreturn(ring_custom);
56
-
} else {
57
-
// Something did not go well, raise an exception
58
-
char* error_msg = strerror(status);
59
-
caml_failwith(error_msg);
60
-
}
59
+
// TODO also add an unregister ba
60
+
61
+
value ocaml_uring_register_ba(value v_uring, value v_ba) {
62
+
CAMLparam2(v_uring, v_ba);
63
+
struct io_uring *ring = Ring_val(v_uring);
64
+
struct iovec iov[1];
65
+
iov[0].iov_base = Caml_ba_data_val(v_ba);
66
+
iov[0].iov_len = Caml_ba_array_val(v_ba)->dim[0];
67
+
fprintf(stderr,"uring %p: registering iobuf base %p len %lu\n", ring, iov[0].iov_base, iov[0].iov_len);
68
+
int ret = io_uring_register_buffers(ring, iov, 1);
69
+
if (ret)
70
+
caml_failwith(strerror(-ret));
71
+
CAMLreturn(Val_unit);
61
72
}
73
+
74
+
value ocaml_uring_exit(value v_uring) {
75
+
CAMLparam1(v_uring);
76
+
struct io_uring *ring = Ring_val(v_uring);
77
+
fprintf(stderr, "uring %p: exit\n", ring);
78
+
io_uring_queue_exit(ring);
79
+
caml_stat_free(ring);
80
+
CAMLreturn(Val_unit);
81
+
}
82
+
83
+
value
84
+
ocaml_uring_alloc_iovecs(value v_ba_arr) {
85
+
CAMLparam1(v_ba_arr);
86
+
size_t len = Wosize_val(v_ba_arr);
87
+
struct iovec *iovs = caml_stat_alloc(len * sizeof(struct iovec));
88
+
for (int i=0; i<len; i++) {
89
+
value v_ba = Field(v_ba_arr,i);
90
+
iovs[i].iov_base = Caml_ba_data_val(v_ba);
91
+
iovs[i].iov_len = Caml_ba_array_val(v_ba)->dim[0];
92
+
fprintf(stderr, "iov %d: %p %lu\n", i, iovs[i].iov_base, iovs[i].iov_len);
93
+
}
94
+
if (((uintptr_t) iovs & 1) == 1) caml_failwith("unaligned alloc??");
95
+
CAMLreturn ((value) iovs | 1);
96
+
}
97
+
98
+
value
99
+
ocaml_uring_free_iovecs(value iovecs)
100
+
{
101
+
struct iovec *iovs = (struct iovec *) (iovecs & ~1);
102
+
free (iovs);
103
+
iovs = NULL;
104
+
return(Val_unit);
105
+
}
106
+
107
+
value
108
+
ocaml_uring_submit_readv(value v_uring, value v_fd, value iovecs, value v_len) {
109
+
CAMLparam1(v_uring);
110
+
struct io_uring *ring = Ring_val(v_uring);
111
+
struct iovec *iovs = (struct iovec *) (iovecs & ~1);
112
+
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
113
+
if (!sqe)
114
+
caml_failwith("unable to allocate SQE");
115
+
io_uring_prep_readv(sqe, Int_val(v_fd), iovs, Int_val(v_len), 0);
116
+
io_uring_sqe_set_data(sqe, iovs);
117
+
CAMLreturn(Val_unit);
118
+
}
119
+
120
+
value ocaml_uring_submit(value v_uring)
121
+
{
122
+
CAMLparam1(v_uring);
123
+
struct io_uring *ring = Ring_val(v_uring);
124
+
int num = io_uring_submit(ring);
125
+
CAMLreturn(Val_int(num));
126
+
}
127
+
128
+
value ocaml_uring_wait_cqe(value v_uring)
129
+
{
130
+
CAMLparam1(v_uring);
131
+
CAMLlocal1(v_ret);
132
+
struct io_uring *ring = Ring_val(v_uring);
133
+
struct io_uring_cqe *cqe;
134
+
fprintf(stderr, "cqe: waiting\n");
135
+
io_uring_wait_cqe(ring, &cqe);
136
+
fprintf(stderr, "cqe: %p res = %d\n", cqe, cqe->res);
137
+
if (cqe->res < 0)
138
+
caml_failwith(strerror(-cqe->res));
139
+
fprintf(stderr, "ceq %p: res=%d\n", cqe, cqe->res);
140
+
struct iovec *iovs = io_uring_cqe_get_data(cqe);
141
+
io_uring_cqe_seen(ring, cqe);
142
+
v_ret = caml_alloc(3, 0);
143
+
Store_field(v_ret, 0, (value)iovs | 1);
144
+
Store_field(v_ret, 1, Val_int(cqe->res));
145
+
CAMLreturn(v_ret);
146
+
}
147
+
#if 0
62
148
63
149
void ring_queue_write_full(value ring_custom, value fd, value callback, value buffer_bigarray, value nbytes) {
64
150
CAMLparam5(ring_custom, fd, callback, buffer_bigarray, nbytes);
···
252
338
CAMLreturn(Val_int(submitted));
253
339
}
254
340
255
-
void ring_exit(value ring_custom) {
256
-
CAMLparam1(ring_custom);
257
-
258
-
struct io_uring* ring = Ring_val(ring_custom);
259
-
260
-
io_uring_queue_exit(ring);
261
-
262
-
caml_stat_free(ring);
263
-
264
-
CAMLreturn0;
265
-
}
341
+
#endif
+17
-4
uring_test.ml
+17
-4
uring_test.ml
···
1
-
let ring = Uring.ring_setup 1024
1
+
let () =
2
+
let t = Uring.create ~queue_depth:1 () in
3
+
let fd = Unix.(handle_unix_error (openfile "test.txt" [O_RDONLY]) 0) in
4
+
let b1 = Uring.iobuf_alloc 3 in
5
+
let b2 = Uring.iobuf_alloc 5 in
6
+
Uring.submit_readv t fd [|b1;b2|];
7
+
let res = Uring.submit t in
8
+
Printf.eprintf "submitted %d\n%!" res;
9
+
let _,res = Uring.wait_cqe t in
10
+
Printf.eprintf "res %d\n%!" res;
11
+
Printf.eprintf "%s -- %s\n%!" (Bigstringaf.to_string b1) (Bigstringaf.to_string b2);
12
+
()
2
13
14
+
(*
3
15
let server_fd = Unix.socket PF_INET SOCK_STREAM 0
4
16
5
17
let client_fd = ref None
···
10
22
| Some(client) ->
11
23
let exit_loop = ref false in
12
24
let read_buf = Bigstringaf.create 4096 in
13
-
Uring.ring_queue_read ring client (fun buf len ->
14
-
if len == 0 then
25
+
Uring.ring_queue_read ring client (fun buf len ->
26
+
if len = 0 then
15
27
begin
16
28
Printf.printf "Client disconnected!";
17
29
exit_loop := true
···
36
48
let (new_client_fd, _) = Unix.accept server_fd in
37
49
Printf.printf "Go connection!\n%!";
38
50
client_fd := Some(new_client_fd);
39
-
server_loop ()
51
+
server_loop ()
52
+
*)