The unpac monorepo manager self-hosting as a monorepo using unpac
working barebones cp
+418
-356
+15
LICENSE.md
+15
LICENSE.md
···
1
+
/*
2
+
* Copyright (C) 2020-2021 Anil Madhavapeddy
3
+
*
4
+
* Permission to use, copy, modify, and distribute this software for any
5
+
* purpose with or without fee is hereby granted, provided that the above
6
+
* copyright notice and this permission notice appear in all copies.
7
+
*
8
+
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
9
+
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
10
+
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
11
+
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
12
+
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
13
+
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
14
+
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
15
+
*/
+7
-1
README.md
+7
-1
README.md
···
1
-
This is a work in progress, it may never work!
1
+
# ocaml-uring -- bindings to Linux io_uring
2
+
3
+
These are OCaml bindings for the Linux io_uring stack.
4
+
They are intended to eventually be used with the multicore OCaml stack,
5
+
but may also be useful for sequential code too.
6
+
7
+
- Status: work in progress, unreleased.
-15
dune
-15
dune
···
13
13
(extra_deps include/liburing/compat.h)
14
14
))
15
15
16
-
(executable
17
-
(name uring_test)
18
-
(modules uring_test)
19
-
(libraries bigstringaf unix uring))
20
-
21
-
(executable
22
-
(name urcat)
23
-
(modules urcat)
24
-
(libraries bigstringaf unix uring))
25
-
26
-
(executable
27
-
(name urcp)
28
-
(modules urcp)
29
-
(libraries bigstringaf unix uring))
30
-
31
16
(rule
32
17
(deps
33
18
(source_tree liburing))
+11
-1
dune-project
+11
-1
dune-project
···
1
1
(lang dune 2.7)
2
-
(name uring)
2
+
(name uring)
3
+
(generate_opam_files true)
4
+
(source (github ocaml-multicore/ocaml-uring))
5
+
(license ISC)
6
+
(authors "Anil Madhavapeddy" "Sadiq Jaffer")
7
+
(maintainers "anil@recoil.org")
8
+
(package
9
+
(name uring)
10
+
(synopsis "OCaml bindings for Linux io_uring")
11
+
(description "Bindings to the Linux io_uring kernel IO interfaces.")
12
+
(depends (fmt (>= 0.8.4)) bigstringaf (alcotest :with-test)))
+11
examples/dune
+11
examples/dune
+161
examples/urcp.ml
+161
examples/urcp.ml
···
1
+
(* cp(1) built with liburing. Queues up as many reads as the queue
2
+
* depth allows and then queues up corresponding writes.
3
+
OCaml version of https://unixism.net/loti/tutorial/cp_liburing.html *)
4
+
5
+
let get_file_size fd =
6
+
Unix.handle_unix_error Unix.fstat fd |>
7
+
fun {Unix.st_size; _} -> st_size
8
+
(* TODO make this work with ST_ISBLK *)
9
+
10
+
type t = {
11
+
mutable insize: int;
12
+
mutable offset: int;
13
+
mutable reads: int;
14
+
mutable writes: int;
15
+
mutable write_left: int;
16
+
mutable read_left: int;
17
+
block_size: int;
18
+
infd: Unix.file_descr;
19
+
outfd: Unix.file_descr;
20
+
}
21
+
22
+
let pp ppf {insize;offset;reads;writes;write_left; read_left;_} =
23
+
Fmt.pf ppf "insize %d offset %d reads %d writes %d rleft %d wleft %d"
24
+
insize offset reads writes read_left write_left
25
+
26
+
type req = {
27
+
op: [`R | `W ];
28
+
iov: Uring.Iovec.t;
29
+
len: int;
30
+
fileoff: int;
31
+
mutable off: int;
32
+
t : t;
33
+
}
34
+
35
+
let pp_req ppf {op; len; off; fileoff; t; _ } =
36
+
Fmt.pf ppf "[%s fileoff %d len %d off %d] [%a]" (match op with |`R -> "r" |`W -> "w") fileoff len off pp t
37
+
38
+
let empty_req t = { op=`R; iov=Uring.Iovec.empty; len=0; off=0; fileoff=0; t}
39
+
40
+
(* Perform a complete read into bufs. *)
41
+
let queue_read uring t len =
42
+
let ba = Uring.Iovec.alloc_buf len in
43
+
let iov = Uring.Iovec.alloc [|ba|] in
44
+
let req = { op=`R; iov; fileoff=t.offset; len; off=0; t } in
45
+
Fmt.epr "queue_read: %a\n%!" pp_req req;
46
+
Uring.readv uring ~offset:t.offset t.infd iov req;
47
+
t.offset <- t.offset + len;
48
+
t.read_left <- t.read_left - len;
49
+
t.reads <- t.reads + 1
50
+
51
+
(* TODO compile time check *)
52
+
let eagain = -11
53
+
let eintr = -4
54
+
55
+
(* Check that a read has completely finished, and if not
56
+
* queue it up for completing the remaining amount *)
57
+
let handle_read_completion uring req res =
58
+
Fmt.epr "read_completion: res=%d %a\n%!" res pp_req req;
59
+
let bytes_to_read = req.len - req.off in
60
+
match res with
61
+
| 0 ->
62
+
Fmt.epr "eof %a\n%!" pp_req req
63
+
| n when n = eagain || n = eintr ->
64
+
(* requeue the request *)
65
+
Uring.readv ~offset:req.fileoff uring req.t.infd req.iov req;
66
+
Fmt.epr "requeued eintr read: %a\n%!" pp_req req
67
+
| n when n < 0 ->
68
+
raise (Failure ("unix errorno " ^ (string_of_int n)))
69
+
| n when n < bytes_to_read ->
70
+
(* handle short read so new iovec and resubmit *)
71
+
Uring.Iovec.advance req.iov ~idx:0 ~adj:n;
72
+
req.off <-req.off + n;
73
+
Uring.readv ~offset:req.off uring req.t.infd req.iov req;
74
+
Fmt.epr "requeued short read: %a\n%!" pp_req req
75
+
| n when n = bytes_to_read ->
76
+
(* Read is complete, all bytes are read, turn it into a write *)
77
+
req.t.reads <- req.t.reads - 1;
78
+
req.t.writes <- req.t.writes + 1;
79
+
(* reset the iovec *)
80
+
Uring.Iovec.advance req.iov ~idx:0 ~adj:(req.off * -1);
81
+
let req = { req with op=`W; off=0 } in
82
+
Uring.writev uring ~offset:req.fileoff req.t.outfd req.iov req;
83
+
Fmt.epr "queued write: %a\n%!" pp_req req
84
+
| n -> raise (Failure (Printf.sprintf "unexpected readv result %d > %d " bytes_to_read n))
85
+
86
+
let handle_write_completion uring req res =
87
+
Fmt.epr "write_completion: res=%d %a\n%!" res pp_req req;
88
+
let bytes_to_write = req.len - req.off in
89
+
match res with
90
+
| 0 -> raise End_of_file
91
+
| n when n = eagain || n = eintr ->
92
+
(* requeue the request *)
93
+
Uring.writev ~offset:req.fileoff uring req.t.infd req.iov req;
94
+
Fmt.epr "requeued eintr read: %a\n%!" pp_req req
95
+
| n when n < bytes_to_write ->
96
+
(* handle short write so new iovec and resubmit *)
97
+
Uring.Iovec.advance req.iov ~idx:0 ~adj:n;
98
+
req.off <-req.off + n;
99
+
Uring.writev ~offset:req.off uring req.t.infd req.iov req;
100
+
Fmt.epr "requeued write read: %a\n%!" pp_req req
101
+
| n when n = bytes_to_write ->
102
+
req.t.writes <- req.t.writes - 1;
103
+
req.t.write_left <- req.t.write_left - req.len;
104
+
Fmt.epr "write done: %a\n%!" pp_req req;
105
+
Uring.Iovec.free req.iov
106
+
| n -> raise (Failure (Printf.sprintf "unexpected writev result %d > %d " bytes_to_write n))
107
+
108
+
let handle_completion uring req res =
109
+
match req.op with
110
+
|`R -> handle_read_completion uring req res
111
+
|`W -> handle_write_completion uring req res
112
+
113
+
let copy_file uring t =
114
+
(* Create a set of read requests that we will turn into write requests
115
+
* up until the queue depth *)
116
+
while t.write_left > 0 || t.read_left > 0 do
117
+
let rec submit_reads () =
118
+
if t.read_left > 0 then begin
119
+
if t.reads + t.writes < (Uring.queue_depth uring) then begin
120
+
let size = min t.block_size t.read_left in
121
+
queue_read uring t size;
122
+
submit_reads ()
123
+
end
124
+
end;
125
+
in
126
+
submit_reads ();
127
+
let num = Uring.submit uring in
128
+
Fmt.(epr "%a: %d\n%!" (styled `Yellow string) "submit") num;
129
+
(* Queue now full, find at least one completion *)
130
+
let got_completion = ref false in
131
+
let rec handle_completions () =
132
+
if t.write_left > 0 then begin
133
+
let check_q = if !got_completion then Uring.peek uring else Uring.wait uring in
134
+
match check_q with
135
+
|None -> Fmt.epr "completions: retry so finishing loop\n%!"
136
+
|Some (req, res) ->
137
+
handle_completion uring req res;
138
+
got_completion := true;
139
+
handle_completions ();
140
+
end
141
+
in
142
+
handle_completions ();
143
+
let num = Uring.submit uring in
144
+
Fmt.(epr "%a: %d\n%!" (styled `Yellow string) "submit") num;
145
+
done
146
+
147
+
let () =
148
+
let infile = Sys.argv.(1) in
149
+
let outfile = Sys.argv.(2) in
150
+
let infd = Unix.(handle_unix_error (openfile infile [O_RDONLY]) 0) in
151
+
let outfd = Unix.(handle_unix_error (openfile outfile [O_WRONLY; O_CREAT; O_TRUNC]) 0o644) in
152
+
let insize = get_file_size infd in
153
+
let block_size = 32 * 1024 in
154
+
let queue_depth = 64 in
155
+
let t = { block_size; insize; offset=0; reads=0; writes=0; write_left=insize; read_left=insize; infd; outfd } in
156
+
Fmt.epr "\nstarting: %a bs=%d qd=%d\n%!" pp t block_size queue_depth;
157
+
let uring = Uring.create ~queue_depth ~default:(empty_req t) () in
158
+
copy_file uring t;
159
+
Unix.close infd;
160
+
Unix.close outfd;
161
+
Uring.exit uring
+16
iovec.ml
+16
iovec.ml
···
1
+
(*
2
+
* Copyright (C) 2020-2021 Anil Madhavapeddy
3
+
*
4
+
* Permission to use, copy, modify, and distribute this software for any
5
+
* purpose with or without fee is hereby granted, provided that the above
6
+
* copyright notice and this permission notice appear in all copies.
7
+
*
8
+
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
9
+
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
10
+
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
11
+
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
12
+
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
13
+
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
14
+
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
15
+
*)
16
+
1
17
type buf = (char, Bigarray.int8_unsigned_elt, Bigarray.c_layout) Bigarray.Array1.t
2
18
3
19
type iovec
+16
iovec.mli
+16
iovec.mli
···
1
+
(*
2
+
* Copyright (C) 2020-2021 Anil Madhavapeddy
3
+
*
4
+
* Permission to use, copy, modify, and distribute this software for any
5
+
* purpose with or without fee is hereby granted, provided that the above
6
+
* copyright notice and this permission notice appear in all copies.
7
+
*
8
+
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
9
+
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
10
+
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
11
+
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
12
+
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
13
+
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
14
+
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
15
+
*)
16
+
1
17
type buf = (char, Bigarray.int8_unsigned_elt, Bigarray.c_layout) Bigarray.Array1.t
2
18
type iovec
3
19
type t
+4
tests/dune
+4
tests/dune
+1
-1
urcat.ml
examples/urcat.ml
+1
-1
urcat.ml
examples/urcat.ml
···
9
9
(* TODO make this work with ST_ISBLK *)
10
10
11
11
let get_completion_and_print uring =
12
-
let iov, len = Uring.wait uring in
12
+
let iov, len = match Uring.wait uring with Some v -> v | None -> failwith "retry" in
13
13
let bufs = Uring.Iovec.bufs iov in
14
14
let remaining = ref len in
15
15
Printf.eprintf "%d bytes read\n%!" len;
-103
urcp.ml
-103
urcp.ml
···
1
-
(* cp(1) built with liburing. Queues up as many reads as the queue
2
-
* depth allows and then queues up corresponding writes.
3
-
OCaml version of https://unixism.net/loti/tutorial/cp_liburing.html *)
4
-
5
-
let queue_depth = 64
6
-
let block_size = 32*1024
7
-
8
-
let get_file_size fd =
9
-
Unix.handle_unix_error Unix.fstat fd |>
10
-
fun {Unix.st_size; _} -> st_size
11
-
(* TODO make this work with ST_ISBLK *)
12
-
13
-
type t = {
14
-
mutable insize: int;
15
-
mutable offset: int;
16
-
mutable reads: int;
17
-
mutable writes: int;
18
-
mutable write_left: int;
19
-
mutable read_left: int;
20
-
infd: Unix.file_descr;
21
-
outfd: Unix.file_descr;
22
-
}
23
-
24
-
type req = {
25
-
op: [`R | `W ];
26
-
iov: Uring.Iovec.t;
27
-
len: int;
28
-
mutable off: int;
29
-
t : t;
30
-
}
31
-
32
-
let empty_req t = { op=`R; iov=Uring.Iovec.empty; len=0; off=0; t}
33
-
34
-
(* Perform a complete read into bufs. *)
35
-
let queue_read uring t len =
36
-
let ba = Uring.Iovec.alloc_buf block_size in
37
-
let iov = Uring.Iovec.alloc [|ba|] in
38
-
let req = { op=`R; iov; len; off=t.offset; t } in
39
-
Uring.readv uring t.infd iov req;
40
-
t.offset <- t.offset + len;
41
-
t.read_left <- t.read_left + len;
42
-
t.reads <- t.reads + 1
43
-
44
-
(* TODO compile time check *)
45
-
let eagain = -11
46
-
let eintr = -4
47
-
48
-
(* Check that a read has completely finished, and if not
49
-
* queue it up for completing the remaining amount *)
50
-
let handle_read_completion uring req res =
51
-
let bytes_to_read = req.len - req.off in
52
-
match res with
53
-
| 0 -> raise End_of_file
54
-
| n when n = eagain || n = eintr ->
55
-
(* requeue the request *)
56
-
Uring.readv ~offset:req.off uring req.t.infd req.iov req
57
-
| n when n < 0 ->
58
-
raise (Failure ("unix errorno " ^ (string_of_int n)))
59
-
| n when n < bytes_to_read ->
60
-
(* handle short read so new iovec and resubmit *)
61
-
Uring.Iovec.advance req.iov ~idx:0 ~adj:n;
62
-
req.off <-req.off + n;
63
-
Uring.readv ~offset:req.off uring req.t.infd req.iov req
64
-
| n when n = bytes_to_read ->
65
-
(* Read is complete, all bytes are read, turn it into a write *)
66
-
req.t.reads <- req.t.reads - 1;
67
-
req.t.writes <- req.t.writes + 1;
68
-
(* reset the iovec *)
69
-
Uring.Iovec.advance req.iov ~idx:0 ~adj:(req.off * -1);
70
-
let req = { req with op=`W; off=0 } in
71
-
Uring.writev ~offset:0 uring req.t.outfd req.iov req
72
-
| n -> raise (Failure ("unexpected readv result > len " ^ (string_of_int n)))
73
-
74
-
let copy_file uring t =
75
-
(* Create a set of read requests that we will turn into write requests
76
-
* up until the queue depth *)
77
-
while t.write_left > 0 || t.read_left > 0 do
78
-
let need_submit = ref false in
79
-
let submit () =
80
-
if t.read_left > 0 then begin
81
-
if t.reads + t.writes < queue_depth then begin
82
-
let size = min block_size t.read_left in
83
-
queue_read uring t size;
84
-
need_submit := true;
85
-
end
86
-
end;
87
-
if !need_submit then
88
-
let _ = Uring.submit uring in ()
89
-
90
-
in
91
-
submit ()
92
-
done
93
-
94
-
let () =
95
-
let infile = Sys.argv.(1) in
96
-
let outfile = Sys.argv.(2) in
97
-
let infd = Unix.(handle_unix_error (openfile infile [O_RDONLY]) 0) in
98
-
let outfd = Unix.(handle_unix_error (openfile outfile [O_WRONLY; O_CREAT; O_TRUNC]) 0o644) in
99
-
let insize = get_file_size infd in
100
-
let t = { insize; offset=0; reads=0; writes=0; write_left=insize; read_left=insize; infd; outfd } in
101
-
let uring = Uring.create ~queue_depth ~default:(empty_req t) () in
102
-
copy_file uring t
103
-
(* TOD fd close and iouring exit *)
+49
-17
uring.ml
+49
-17
uring.ml
···
1
+
(*
2
+
* Copyright (C) 2020-2021 Anil Madhavapeddy
3
+
*
4
+
* Permission to use, copy, modify, and distribute this software for any
5
+
* purpose with or without fee is hereby granted, provided that the above
6
+
* copyright notice and this permission notice appear in all copies.
7
+
*
8
+
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
9
+
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
10
+
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
11
+
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
12
+
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
13
+
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
14
+
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
15
+
*)
16
+
1
17
module Iovec = Iovec
2
18
3
19
type uring
···
11
27
external uring_submit_writev : uring -> Unix.file_descr -> id -> Iovec.t -> int -> unit = "ocaml_uring_submit_writev"
12
28
13
29
external uring_wait_cqe : uring -> id * int = "ocaml_uring_wait_cqe"
30
+
external uring_peek_cqe : uring -> id * int = "ocaml_uring_peek_cqe"
31
+
14
32
15
33
type 'a t = {
16
34
uring: uring;
17
35
iobuf: Iovec.buf;
18
36
mutable id_freelist: int list;
19
37
user_data: 'a array;
38
+
queue_depth: int;
39
+
mutable dirty: bool; (* has outstanding requests that need to be submitted *)
20
40
}
21
41
22
42
let default_iobuf_len = 1024 * 1024 (* 1MB *)
···
29
49
Gc.finalise uring_exit uring;
30
50
let id_freelist = List.init queue_depth (fun i -> i) in
31
51
let user_data = Array.init queue_depth (fun _ -> default) in
32
-
{ uring; iobuf; id_freelist; user_data }
52
+
{ uring; iobuf; id_freelist; user_data; dirty=false; queue_depth }
53
+
54
+
let exit {uring;_} = uring_exit uring
33
55
34
56
let get_id t =
35
57
match t.id_freelist with
···
42
64
let readv t ?(offset=0) fd iovec user_data =
43
65
let id = get_id t in
44
66
uring_submit_readv t.uring fd id iovec offset;
67
+
t.dirty <- true;
45
68
t.user_data.(id) <- user_data
46
69
47
70
let writev t ?(offset=0) fd iovec user_data =
48
71
let id = get_id t in
49
72
uring_submit_writev t.uring fd id iovec offset;
73
+
t.dirty <- true;
50
74
t.user_data.(id) <- user_data
51
75
52
-
let submit {uring;_} =
53
-
uring_submit uring
76
+
let submit t =
77
+
if t.dirty then begin
78
+
t.dirty <- false;
79
+
uring_submit t.uring
80
+
end else
81
+
0
54
82
55
-
let wait t =
56
-
let id, res = uring_wait_cqe t.uring in
57
-
let data = t.user_data.(id) in
58
-
put_id t id;
59
-
data, res
83
+
(* TODO use unixsupport.h *)
84
+
let errno_is_retry = function -11 | -4 -> true |_ -> false
85
+
86
+
let fn_on_ring fn t =
87
+
let id, res = fn t.uring in
88
+
match id, res with
89
+
| -1, res when errno_is_retry res ->
90
+
None
91
+
| -1, res when res < 0 ->
92
+
failwith ("wait error " ^ (string_of_int res))
93
+
(* TODO switch to unixsupport.h to raise Unix_error *)
94
+
| id, res ->
95
+
let data = t.user_data.(id) in
96
+
put_id t id;
97
+
Some (data, res)
60
98
61
-
(*
62
-
external ring_queue_write_full : t -> Unix.file_descr -> (Bigstringaf.t -> int -> unit) -> Bigstringaf.t -> int -> unit = "ring_queue_write_full"
63
-
external ring_queue_read : t -> Unix.file_descr -> (Bigstringaf.t -> int -> unit) -> Bigstringaf.t -> int -> unit = "ring_queue_read"
64
-
external ring_queue_accept : t -> Unix.file_descr -> (Unix.file_descr -> unit) -> unit = "ring_queue_accept"
65
-
external ring_queue_close : t -> Unix.file_descr -> unit = "ring_queue_close"
66
-
external ring_submit : t -> int = "ring_submit"
67
-
external ring_exit : t -> unit = "ring_exit"
68
-
external ring_wait : t -> unit = "ring_wait"
69
-
*)
99
+
let peek t = fn_on_ring uring_peek_cqe t
100
+
let wait t = fn_on_ring uring_wait_cqe t
101
+
let queue_depth {queue_depth;_} = queue_depth
+20
-1
uring.mli
+20
-1
uring.mli
···
1
+
(*
2
+
* Copyright (C) 2020-2021 Anil Madhavapeddy
3
+
*
4
+
* Permission to use, copy, modify, and distribute this software for any
5
+
* purpose with or without fee is hereby granted, provided that the above
6
+
* copyright notice and this permission notice appear in all copies.
7
+
*
8
+
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
9
+
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
10
+
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
11
+
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
12
+
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
13
+
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
14
+
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
15
+
*)
16
+
1
17
module Iovec = Iovec
2
18
3
19
type 'a t
4
20
5
21
val create : queue_depth:int -> default:'a -> unit -> 'a t
22
+
val queue_depth : 'a t -> int
23
+
val exit : 'a t -> unit
6
24
7
25
val readv : 'a t -> ?offset:int -> Unix.file_descr -> Iovec.t -> 'a -> unit
8
26
val writev : 'a t -> ?offset:int -> Unix.file_descr -> Iovec.t -> 'a -> unit
9
27
val submit : 'a t -> int
10
28
11
-
val wait : 'a t -> 'a * int
29
+
val wait : 'a t -> ('a * int) option
30
+
val peek : 'a t -> ('a * int) option
+31
uring.opam
+31
uring.opam
···
1
+
# This file is generated by dune, edit dune-project instead
2
+
opam-version: "2.0"
3
+
synopsis: "OCaml bindings for Linux io_uring"
4
+
description: "Bindings to the Linux io_uring kernel IO interfaces."
5
+
maintainer: ["anil@recoil.org"]
6
+
authors: ["Anil Madhavapeddy" "Sadiq Jaffer"]
7
+
license: "ISC"
8
+
homepage: "https://github.com/ocaml-multicore/ocaml-uring"
9
+
bug-reports: "https://github.com/ocaml-multicore/ocaml-uring/issues"
10
+
depends: [
11
+
"dune" {>= "2.7"}
12
+
"fmt" {>= "0.8.4"}
13
+
"bigstringaf"
14
+
"alcotest" {with-test}
15
+
"odoc" {with-doc}
16
+
]
17
+
build: [
18
+
["dune" "subst"] {dev}
19
+
[
20
+
"dune"
21
+
"build"
22
+
"-p"
23
+
name
24
+
"-j"
25
+
jobs
26
+
"@install"
27
+
"@runtest" {with-test}
28
+
"@doc" {with-doc}
29
+
]
30
+
]
31
+
dev-repo: "git+https://github.com/ocaml-multicore/ocaml-uring.git"
+69
-216
uring_stubs.c
+69
-216
uring_stubs.c
···
1
+
/*
2
+
* Copyright (C) 2020-2021 Anil Madhavapeddy
3
+
* Copyright (C) 2020-2021 Sadiq Jaffer
4
+
*
5
+
* Permission to use, copy, modify, and distribute this software for any
6
+
* purpose with or without fee is hereby granted, provided that the above
7
+
* copyright notice and this permission notice appear in all copies.
8
+
*
9
+
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
10
+
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
11
+
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
12
+
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
13
+
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
14
+
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
15
+
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
16
+
*/
17
+
1
18
#include <liburing.h>
2
19
#include <caml/alloc.h>
3
20
#include <caml/bigarray.h>
···
7
24
#include <caml/memory.h>
8
25
#include <caml/mlvalues.h>
9
26
#include <caml/signals.h>
27
+
#include <caml/unixsupport.h>
10
28
#include <string.h>
11
29
12
-
#define Ring_val(v) *((struct io_uring**)Data_custom_val(v))
30
+
#undef URING_DEBUG
31
+
#ifdef URING_DEBUG
32
+
#define dprintf(fmt, ...) fprintf(stderr, fmt, ##__VA_ARGS__)
33
+
#else
34
+
#define dprintf(fmt, ...) ((void)0)
35
+
#endif
13
36
14
-
#define EVENT_TYPE_READ 0
15
-
#define EVENT_TYPE_WRITE 1
16
-
#define EVENT_TYPE_ACCEPT 2
17
-
#define EVENT_TYPE_CLOSE 3
37
+
#define Ring_val(v) *((struct io_uring**)Data_custom_val(v))
18
38
19
39
static struct custom_operations ring_ops = {
20
40
"uring.ring",
···
51
71
value ring_custom = caml_alloc_custom_mem(&ring_ops, sizeof(struct io_uring*), sizeof(struct io_uring));
52
72
*((struct io_uring**)Data_custom_val(ring_custom)) = ring;
53
73
CAMLreturn(ring_custom);
54
-
} else {
55
-
caml_failwith(strerror(-status));
56
-
}
74
+
} else
75
+
unix_error(-status, "io_uring_queue_init", Nothing);
57
76
}
58
77
59
78
// TODO also add an unregister ba
···
65
84
struct iovec iov[1];
66
85
iov[0].iov_base = Caml_ba_data_val(v_ba);
67
86
iov[0].iov_len = Caml_ba_array_val(v_ba)->dim[0];
68
-
fprintf(stderr,"uring %p: registering iobuf base %p len %lu\n", ring, iov[0].iov_base, iov[0].iov_len);
87
+
dprintf("uring %p: registering iobuf base %p len %lu\n", ring, iov[0].iov_base, iov[0].iov_len);
69
88
int ret = io_uring_register_buffers(ring, iov, 1);
70
89
if (ret)
71
-
caml_failwith(strerror(-ret));
90
+
unix_error(-ret, "io_uring_register_buffers", Nothing);
72
91
CAMLreturn(Val_unit);
73
92
}
74
93
75
94
value ocaml_uring_exit(value v_uring) {
76
95
CAMLparam1(v_uring);
77
96
struct io_uring *ring = Ring_val(v_uring);
78
-
fprintf(stderr, "uring %p: exit\n", ring);
97
+
dprintf("uring %p: exit\n", ring);
79
98
io_uring_queue_exit(ring);
80
99
caml_stat_free(ring);
100
+
ring = NULL;
81
101
CAMLreturn(Val_unit);
82
102
}
83
103
···
90
110
value v_ba = Field(v_ba_arr,i);
91
111
iovs[i].iov_base = Caml_ba_data_val(v_ba);
92
112
iovs[i].iov_len = Caml_ba_array_val(v_ba)->dim[0];
93
-
fprintf(stderr, "iov %d: %p %lu\n", i, iovs[i].iov_base, iovs[i].iov_len);
113
+
dprintf( "iov %d: %p %lu\n", i, iovs[i].iov_base, iovs[i].iov_len);
94
114
}
95
115
if (((uintptr_t) iovs & 1) == 1) caml_failwith("unaligned alloc??");
96
116
CAMLreturn ((value) iovs | 1);
···
125
145
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
126
146
if (!sqe)
127
147
caml_failwith("unable to allocate SQE");
128
-
fprintf(stderr, "submit_readv: %d ents off %d\n", len, Int_val(v_off));
148
+
dprintf("submit_readv: %d ents len[0] %lu off %d\n", len, iovs[0].iov_len, Int_val(v_off));
129
149
io_uring_prep_readv(sqe, Int_val(v_fd), iovs, len, Int_val(v_off)); /* TODO add offset to intf */
130
150
io_uring_sqe_set_data(sqe, (void *)(uintptr_t)Int_val(v_id)); /* TODO sort out cast */
131
151
CAMLreturn(Val_unit);
···
140
160
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
141
161
if (!sqe)
142
162
caml_failwith("unable to allocate SQE");
143
-
fprintf(stderr, "submit_writev: %d ents off %d\n", len, Int_val(v_off));
163
+
dprintf("submit_writev: %d ents len[0] %lu off %d\n", len, iovs[0].iov_len, Int_val(v_off));
144
164
io_uring_prep_writev(sqe, Int_val(v_fd), iovs, len, Int_val(v_off)); /* TODO add offset to intf */
145
165
io_uring_sqe_set_data(sqe, (void *)(uintptr_t)Int_val(v_id)); /* TODO sort out cast */
146
166
CAMLreturn(Val_unit);
···
161
181
long id;
162
182
struct io_uring *ring = Ring_val(v_uring);
163
183
struct io_uring_cqe *cqe;
164
-
fprintf(stderr, "cqe: waiting\n");
165
-
io_uring_wait_cqe(ring, &cqe);
166
-
if (cqe->res < 0)
167
-
caml_failwith(strerror(-cqe->res));
168
-
fprintf(stderr, "cqe %p: res=%d\n", cqe, cqe->res);
169
-
id = (long)io_uring_cqe_get_data(cqe);
170
-
io_uring_cqe_seen(ring, cqe);
171
-
v_ret = caml_alloc(2, 0);
172
-
Store_field(v_ret, 0, Val_int(id));
173
-
Store_field(v_ret, 1, Val_int(cqe->res));
184
+
int res;
185
+
dprintf("cqe: waiting\n");
186
+
res = io_uring_wait_cqe(ring, &cqe);
187
+
if (res < 0) {
188
+
v_ret = caml_alloc(2, 0);
189
+
Store_field(v_ret, 0, Val_int(-1));
190
+
Store_field(v_ret, 1, Val_int(res));
191
+
} else {
192
+
id = (long)io_uring_cqe_get_data(cqe);
193
+
io_uring_cqe_seen(ring, cqe);
194
+
v_ret = caml_alloc(2, 0);
195
+
Store_field(v_ret, 0, Val_int(id));
196
+
Store_field(v_ret, 1, Val_int(cqe->res));
197
+
}
174
198
CAMLreturn(v_ret);
175
199
}
176
-
#if 0
177
200
178
-
void ring_queue_write_full(value ring_custom, value fd, value callback, value buffer_bigarray, value nbytes) {
179
-
CAMLparam5(ring_custom, fd, callback, buffer_bigarray, nbytes);
180
-
181
-
struct io_uring* ring = Ring_val(ring_custom);
182
-
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
183
-
184
-
char* buf = (char*)Caml_ba_data_val(buffer_bigarray);
185
-
186
-
struct request* req = (struct request*)caml_stat_alloc(sizeof(struct request));
187
-
188
-
req->fd = Int_val(fd);
189
-
req->write_length = Long_val(nbytes);
190
-
req->written_length = 0;
191
-
req->iov.iov_base = buf;
192
-
req->iov.iov_len = req->write_length;
193
-
194
-
io_uring_prep_writev(sqe, req->fd, &req->iov, 1, 0);
195
-
196
-
req->event_type = EVENT_TYPE_WRITE;
197
-
req->callback = callback;
198
-
req->buffer = buffer_bigarray;
199
-
200
-
caml_register_generational_global_root(&req->buffer);
201
-
caml_register_generational_global_root(&req->callback);
202
-
203
-
io_uring_sqe_set_data(sqe, req);
204
-
205
-
CAMLreturn0;
206
-
}
207
-
208
-
// For now we use readv instead because it's available in 5.1
209
-
void ring_queue_read(value ring_custom, value fd, value callback, value buffer_bigarray, value offset) {
210
-
CAMLparam5(ring_custom, fd, callback, buffer_bigarray, offset);
211
-
212
-
struct io_uring* ring = Ring_val(ring_custom);
213
-
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
214
-
215
-
char* buf = (char*)Caml_ba_data_val(buffer_bigarray);
216
-
size_t buf_len = Caml_ba_array_val(buffer_bigarray)->dim[0];
217
-
218
-
printf("buf_len: %ld\n", buf_len);
219
-
220
-
struct request* req = (struct request*)caml_stat_alloc(sizeof(struct request));
221
-
222
-
req->fd = Int_val(fd);
223
-
req->iov.iov_base = buf;
224
-
req->iov.iov_len = buf_len;
225
-
226
-
io_uring_prep_readv(sqe, req->fd, &req->iov, 1, Long_val(offset));
227
-
228
-
req->event_type = EVENT_TYPE_READ;
229
-
req->callback = callback;
230
-
req->buffer = buffer_bigarray;
231
-
232
-
caml_register_generational_global_root(&req->buffer);
233
-
caml_register_generational_global_root(&req->callback);
234
-
235
-
io_uring_sqe_set_data(sqe, req);
236
-
237
-
CAMLreturn0;
238
-
}
239
-
240
-
241
-
void ring_queue_close(value ring_custom, value fd) {
242
-
CAMLparam2(ring_custom, fd);
243
-
244
-
struct io_uring* ring = Ring_val(ring_custom);
245
-
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
246
-
247
-
io_uring_prep_close(sqe, Int_val(fd));
248
-
249
-
struct request* req = (struct request*)caml_stat_alloc(sizeof(struct request));
250
-
251
-
req->event_type = EVENT_TYPE_CLOSE;
252
-
253
-
io_uring_sqe_set_data(sqe, req);
254
-
255
-
CAMLreturn0;
256
-
}
257
-
258
-
void ring_queue_accept(value ring_custom, value fd, value callback) {
259
-
CAMLparam3(ring_custom, fd, callback);
260
-
261
-
struct io_uring* ring = Ring_val(ring_custom);
262
-
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
263
-
264
-
struct request* req = (struct request*)caml_stat_alloc(sizeof(struct request));
265
-
req->sockaddr = (struct sockaddr*)caml_stat_alloc(sizeof(struct sockaddr));
266
-
req->socklen = sizeof(struct sockaddr);
267
-
268
-
io_uring_prep_accept(sqe, Int_val(fd), req->sockaddr, &req->socklen, 0);
269
-
270
-
req->event_type = EVENT_TYPE_ACCEPT;
271
-
req->callback = callback;
272
-
273
-
caml_register_generational_global_root(&req->caml_callback);
274
-
275
-
io_uring_sqe_set_data(sqe, req);
276
-
277
-
CAMLreturn0;
278
-
}
279
-
280
-
void ring_wait(value ring_custom) {
281
-
CAMLparam1(ring_custom);
282
-
283
-
struct io_uring* ring = Ring_val(ring_custom);
284
-
struct io_uring_cqe *cqe;
285
-
286
-
caml_enter_blocking_section();
287
-
int ret = io_uring_wait_cqe(ring, &cqe);
288
-
caml_leave_blocking_section();
289
-
290
-
printf("got event! ret: %d, cqe->res: %d\n", ret, cqe->res);
291
-
292
-
if( ret < 0 ) {
293
-
caml_failwith(strerror(-ret));
294
-
}
295
-
296
-
struct request* req = io_uring_cqe_get_data(cqe);
297
-
298
-
if( cqe->res < 0 ) {
299
-
caml_failwith(strerror(-cqe->res));
300
-
}
301
-
302
-
int cleanup_req = 0;
303
-
304
-
if( req->event_type == EVENT_TYPE_READ || req->event_type == EVENT_TYPE_WRITE ) {
305
-
switch( req->event_type ) {
306
-
case EVENT_TYPE_READ:
307
-
caml_callback2(req->callback, req->buffer, Val_long(cqe->res));
308
-
309
-
cleanup_req = 1;
310
-
break;
311
-
case EVENT_TYPE_WRITE:
312
-
/* check we actually wrote the full length we tried to write */
313
-
if(cqe->res + req->written_length == req->write_length) {
314
-
/* call the callback if it exists */
315
-
if( Is_block(req->callback) ) {
316
-
caml_callback2(req->callback, req->buffer, req->write_length);
317
-
}
318
-
319
-
cleanup_req = 1;
320
-
} else {
321
-
// Here we wrote less than the amount we requested
322
-
323
-
// Store how much we wrote
324
-
req->written_length += cqe->res;
325
-
326
-
// Now we queue up a new write
327
-
req->iov.iov_base = req->iov.iov_base + cqe->res;
328
-
req->iov.iov_len = req->write_length - req->written_length;
329
-
330
-
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
331
-
332
-
io_uring_prep_writev(sqe, req->fd, &req->iov, 1, 0);
333
-
}
334
-
}
335
-
336
-
if( cleanup_req ) {
337
-
caml_remove_generational_global_root(&req->callback);
338
-
caml_remove_generational_global_root(&req->buffer);
339
-
}
340
-
}
341
-
else if( req->event_type == EVENT_TYPE_ACCEPT ) {
342
-
caml_callback(req->callback, Val_int(cqe->res));
343
-
344
-
free(req->sockaddr);
345
-
346
-
caml_remove_generational_global_root(&req->callback);
347
-
}
348
-
201
+
value ocaml_uring_peek_cqe(value v_uring)
202
+
{
203
+
CAMLparam1(v_uring);
204
+
CAMLlocal1(v_ret);
205
+
long id;
206
+
struct io_uring *ring = Ring_val(v_uring);
207
+
struct io_uring_cqe *cqe;
208
+
int res;
209
+
dprintf("cqe: peeking\n");
210
+
res = io_uring_peek_cqe(ring, &cqe);
211
+
if (res < 0) {
212
+
v_ret = caml_alloc(2, 0);
213
+
Store_field(v_ret, 0, Val_int(-1));
214
+
Store_field(v_ret, 1, Val_int(res));
215
+
} else {
216
+
id = (long)io_uring_cqe_get_data(cqe);
349
217
io_uring_cqe_seen(ring, cqe);
350
-
351
-
if( cleanup_req ) {
352
-
caml_stat_free(req);
353
-
}
354
-
355
-
CAMLreturn0;
218
+
v_ret = caml_alloc(2, 0);
219
+
Store_field(v_ret, 0, Val_int(id));
220
+
Store_field(v_ret, 1, Val_int(cqe->res));
221
+
}
222
+
CAMLreturn(v_ret);
356
223
}
357
-
358
-
value ring_submit(value ring_custom) {
359
-
CAMLparam1(ring_custom);
360
-
361
-
struct io_uring* ring = Ring_val(ring_custom);
362
-
363
-
int submitted = io_uring_submit(ring);
364
-
365
-
printf("submitted: %d\n", submitted);
366
-
367
-
CAMLreturn(Val_int(submitted));
368
-
}
369
-
370
-
#endif
+7
-1
uring_test.ml
tests/basic_file_read.ml
+7
-1
uring_test.ml
tests/basic_file_read.ml
···
7
7
Uring.readv t fd iov ();
8
8
let res = Uring.submit t in
9
9
Printf.eprintf "submitted %d\n%!" res;
10
-
let (), res = Uring.wait t in
10
+
let (), res =
11
+
let rec retry () =
12
+
match Uring.wait t with
13
+
| None -> retry ()
14
+
| Some v -> v
15
+
in retry ()
16
+
in
11
17
Uring.Iovec.free iov;
12
18
Printf.eprintf "res %d\n%!" res;
13
19
Printf.eprintf "%s -- %s\n%!" (Bigstringaf.to_string b1) (Bigstringaf.to_string b2);