Source file miou_unix.ml

module Logs = Miou.Logs.Make (struct
  let src = "unix"
end)

type elt = { time: float; syscall: Miou.syscall; mutable cancelled: bool }

module Heapq = Miou.Pqueue.Make (struct
  type t = elt

  let dummy = { time= 0.0; syscall= Obj.magic (); cancelled= false }
  let compare { time= a; _ } { time= b; _ } = Float.compare a b
end)

let rec drop_heapq heapq =
  try Heapq.delete_min_exn heapq; drop_heapq heapq with _ -> ()

let rec sleeper sleepers =
  match Heapq.find_min_exn sleepers with
  | exception Heapq.Empty -> None
  | { cancelled= true; _ } ->
      Heapq.delete_min_exn sleepers;
      sleeper sleepers
  | { time; _ } -> Some time

let in_the_past t = t = 0. || t <= Unix.gettimeofday ()

let rec collect sleepers signals =
  match Heapq.find_min_exn sleepers with
  | exception Heapq.Empty -> signals
  | { cancelled= true; _ } ->
      Heapq.delete_min_exn sleepers;
      collect sleepers signals
  | { time; syscall; _ } when in_the_past time ->
      Heapq.delete_min_exn sleepers;
      collect sleepers (Miou.signal syscall :: signals)
  | _ -> signals
module Bitv = Miou_bitv
module Epoll = Miou_epoll

type kind = STREAM | DGRAM
type file_descr = { fd: Unix.file_descr; kind: kind option; non_blocking: bool }

type bigstring =
  (char, Bigarray.int8_unsigned_elt, Bigarray.c_layout) Bigarray.Array1.t

let () = assert (Obj.is_int (Obj.repr Unix.stdout))
let int_of_file_descr : Unix.file_descr -> int = Obj.magic

module File_descrs = struct
  type t = { mutable contents: Miou.syscall list array }

  let create n = { contents= Array.make (Stdlib.max n 16) [] }

  let grow tbl needed =
    let len = Array.length tbl.contents in
    let len' = ref len in
    while !len' <= needed do
      len' := !len' * 2
    done;
    let contents = Array.make !len' [] in
    Array.blit tbl.contents 0 contents 0 len;
    tbl.contents <- contents

  let get tbl fd =
    let i = int_of_file_descr fd in
    if i >= Array.length tbl.contents then [] else tbl.contents.(i)

  let set tbl fd syscalls =
    let i = int_of_file_descr fd in
    if i >= Array.length tbl.contents then grow tbl i;
    tbl.contents.(i) <- syscalls

  let clear tbl = Array.fill tbl.contents 0 (Array.length tbl.contents) []
end

(* NOTE(dinosaure): [epoll] has a rather unusual instruction that is of
   interest to us here: [EPOLLONESHOT]. The idea is to let the kernel disables a
   file descriptor from [epoll] if it has been signalled (whether IN or OUT, it
   doesn’t matter). Note that the registration of the file-descriptor is kept
   but no further event is reported until we {i re-arm} it. We will therefore
   take advantage of this feature so as not to have to call [Epoll.del]
   systematically when we want to stop monitoring a [file_descr] (after having
   signalled it).

   We therefore maintain an [armed] table for the current domain, which should
   mirror what the kernel observes regarding file descriptors. This flag table
   is of size [_SC_OPEN_MAX] (we should not have a [file_descr]/[int] larger
   than this number). Based on this table and depending on what we wish to
   monitor, we always add [EPOLLONESHOT]. Finally, when the user considers that
   their [file_descr] should be released (using [Miou_unix]), we {i forget} its
   flag in [armed] so that [armed] is always synchronised with what the kernel
   observes.

   Finally, there is a final table, [drained], which notifies us that the
   kernel's buffer of the given socket is empty. A next [Unix.read] call will
   terminate directly with a [EAGAIN] and we would like to save us from such
   useless syscall. So, if [is_drained fd], we fallbacks directly to a
   [blocking_read] without a confirmation from the kernel (via [Unix.read]) that
   its buffer is actually empty. We are sure that such buffer is empty because
   the last [Unix.read] gives to us few bytes ([> 0]) but not enough to fill
   our buffer ([< len]). *)

type domain = {
    readers: File_descrs.t
  ; writers: File_descrs.t
  ; sleepers: Heapq.t
  ; revert: (Miou.uid, Unix.file_descr) Hashtbl.t
  ; mutable epoll: Epoll.t
  ; events: Epoll.events
  ; drained: bytes
  ; armed: bytes
}

external max_open_files : unit -> int = "miou_unix_epoll_max_open_files"
[@@noalloc]

let flags_by_fd () = Bytes.make (max_open_files ()) '\000'

let is_drained domain fd =
  let idx = int_of_file_descr fd in
  idx < Bytes.length domain.drained
  && Bytes.unsafe_get domain.drained idx <> '\000'

let set_drained domain fd v =
  let idx = int_of_file_descr fd in
  if idx < Bytes.length domain.drained then
    Bytes.unsafe_set domain.drained idx (if v then '\001' else '\000')

let is_armed domain fd =
  let idx = int_of_file_descr fd in
  idx < Bytes.length domain.armed && Bytes.unsafe_get domain.armed idx <> '\000'

let set_armed domain fd v =
  let idx = int_of_file_descr fd in
  if idx < Bytes.length domain.armed then
    Bytes.unsafe_set domain.armed idx (if v then '\001' else '\000')

let arm domain fd =
  let epoll = domain.epoll in
  let flags =
    let rfd = File_descrs.get domain.readers fd in
    let wfd = File_descrs.get domain.writers fd in
    match (rfd, wfd) with
    | [], [] -> Epoll.Flags.empty
    | _ :: _, [] -> Epoll.Flags.epollin
    | [], _ :: _ -> Epoll.Flags.epollout
    | _ :: _, _ :: _ -> Epoll.Flags.(epollin + epollout)
  in
  if flags <> Epoll.Flags.empty then begin
    let flags = Epoll.Flags.(flags + epolloneshot) in
    if is_armed domain fd then
      begin if Epoll.upd epoll fd flags <> 0 then
        if Epoll.add epoll fd flags = 0 then set_armed domain fd true
        else set_armed domain fd false
      end
    else if Epoll.add epoll fd flags = 0 then set_armed domain fd true
    else if Epoll.upd epoll fd flags = 0 then set_armed domain fd true
  end

let forget domain fd =
  let idx = int_of_file_descr fd in
  if idx < Bytes.length domain.armed then
    Bytes.unsafe_set domain.armed idx '\000';
  if idx < Bytes.length domain.drained then
    Bytes.unsafe_set domain.drained idx '\000'

let clean domain uids =
  let clean uid' (fd : Unix.file_descr) tbl =
    match File_descrs.get tbl fd with
    | [] -> ()
    | syscalls ->
        File_descrs.set tbl fd
          (List.filter (fun s -> uid' <> Miou.uid s) syscalls);
        arm domain fd
  in
  let clean uid =
    match Hashtbl.find domain.revert uid with
    | fd ->
        clean uid fd domain.readers;
        clean uid fd domain.writers
    | exception Not_found -> ()
  in
  List.iter clean uids;
  List.iter (Hashtbl.remove domain.revert) uids;
  let clean ({ syscall; _ } as elt) =
    if List.exists (( = ) (Miou.uid syscall)) uids then elt.cancelled <- true
  in
  Heapq.iter clean domain.sleepers

let domain =
  (* NOTE(dinosaure): see [miou_unix.poll.ml.in] for an explanation about
     [split_from_parent]. *)
  let rec split_from_parent v =
    File_descrs.clear v.readers;
    File_descrs.clear v.writers;
    drop_heapq v.sleepers;
    Hashtbl.clear v.revert;
    Bytes.fill v.drained 0 (Bytes.length v.drained) '\000';
    for i = 0 to Bytes.length v.armed - 1 do
      if Bytes.unsafe_get v.armed i <> '\000' then begin
        ignore (Epoll.del v.epoll (Obj.magic i : Unix.file_descr));
        Bytes.unsafe_set v.armed i '\000'
      end
    done;
    make ()
  and make () =
    {
      readers= File_descrs.create 0x100
    ; writers= File_descrs.create 0x100
    ; sleepers= Heapq.create ()
    ; revert= Hashtbl.create 0x100
    ; epoll= Epoll.invalid_fd
    ; events= Epoll.events 256
    ; drained= flags_by_fd ()
    ; armed= flags_by_fd ()
    }
  in
  let key = Stdlib.Domain.DLS.new_key ~split_from_parent make in
  fun () -> Stdlib.Domain.DLS.get key

let append tbl fd syscall =
  File_descrs.set tbl fd (syscall :: File_descrs.get tbl fd)

let blocking_read ?(name = "read") fd =
  let syscall = Miou.syscall ~name () in
  let uid = Miou.uid syscall in
  let domain = domain () in
  let fn () =
    Hashtbl.replace domain.revert uid fd;
    append domain.readers fd syscall;
    arm domain fd
  in
  Miou.suspend ~fn syscall

let blocking_write ?(name = "write") fd =
  let syscall = Miou.syscall ~name () in
  let uid = Miou.uid syscall in
  let domain = domain () in
  let fn () =
    Hashtbl.replace domain.revert uid fd;
    append domain.writers fd syscall;
    arm domain fd
  in
  Miou.suspend ~fn syscall

let rec unsafe_read ({ fd; non_blocking; _ } as file_descr) off len buf =
  if non_blocking then begin
    let domain = domain () in
    if is_drained domain fd then begin
      set_drained domain fd false;
      (* NOTE(dinosaure): here, we would like to do an [Unix.read] which will
         probably return EAGAIN, as the previous read gave us a few bytes but
         not everything we’d like to have. We therefore assume that we should
         wait for an [epoll_wait2]/[select] to notify us that the read has
         completed, rather than trying and failing. *)
      blocking_read fd;
      unsafe_read file_descr off len buf
    end
    else
      match Unix.read fd buf off len with
      | exception Unix.(Unix_error (EINTR, _, _)) ->
          unsafe_read file_descr off len buf
      | exception Unix.(Unix_error ((EAGAIN | EWOULDBLOCK), _, _)) ->
          blocking_read fd;
          unsafe_read file_descr off len buf
      | value ->
          if value > 0 && value < len then set_drained domain fd true;
          value
  end
  else
    let rec go () =
      match Unix.read fd buf off len with
      | exception Unix.(Unix_error (EINTR, _, _)) -> go ()
      | value -> value
    in
    blocking_read fd; go ()

let rec unsafe_read_bigstring ({ fd; non_blocking; _ } as file_descr) off len
    bstr =
  if non_blocking then begin
    let domain = domain () in
    if is_drained domain fd then begin
      set_drained domain fd false;
      blocking_read fd;
      unsafe_read_bigstring file_descr off len bstr
    end
    else
      match Unix.read_bigarray fd bstr off len with
      | exception Unix.(Unix_error (EINTR, _, _)) ->
          unsafe_read_bigstring file_descr off len bstr
      | exception Unix.(Unix_error ((EAGAIN | EWOULDBLOCK), _, _)) ->
          blocking_read fd;
          unsafe_read_bigstring file_descr off len bstr
      | value ->
          if value > 0 && value < len then set_drained domain fd true;
          value
  end
  else
    let rec go () =
      match Unix.read_bigarray fd bstr off len with
      | exception Unix.(Unix_error (EINTR, _, _)) -> go ()
      | value -> value
    in
    blocking_read fd; go ()

let close { fd; _ } =
  forget (domain ()) fd;
  Unix.close fd

let intr fd =
  let buf = Bytes.create 0x100 in
  match Unix.read fd buf 0 (Bytes.length buf) with
  | _ -> ()
  | exception Unix.(Unix_error (EAGAIN, _, _)) -> ()

let transmit_and_clean domain syscall =
  Hashtbl.remove domain.revert (Miou.uid syscall);
  Miou.signal syscall

let wakeup domain tbl fd =
  match File_descrs.get tbl fd with
  | [] -> []
  | syscalls ->
      File_descrs.set tbl fd [];
      List.rev_map (transmit_and_clean domain) syscalls

let select _uid ic ~block cancelled_syscalls =
  let domain = domain () in
  clean domain cancelled_syscalls;
  let timeout =
    match (sleeper domain.sleepers, block) with
    | None, true -> Epoll.Infinite
    | (None | Some _), false -> Epoll.No_wait
    | Some value, true ->
        let value = value -. Unix.gettimeofday () in
        let value = Float.max value 0.0 *. 1e9 in
        Epoll.Nanoseconds (Int64.of_float value)
  in
  match Epoll.wait domain.epoll domain.events timeout with
  | exception Unix.(Unix_error (EINTR, _, _)) -> collect domain.sleepers []
  | nready ->
      let acc = ref (collect domain.sleepers []) in
      let fn fd flags =
        if fd == ic then intr ic
        else begin
          let err = Epoll.Flags.(mem flags (epollerr + epollhup)) in
          let readable = err || Epoll.Flags.mem flags Epoll.Flags.epollin in
          let writable = err || Epoll.Flags.mem flags Epoll.Flags.epollout in
          if readable || writable then begin
            let x = if readable then wakeup domain domain.readers fd else [] in
            let y = if writable then wakeup domain domain.writers fd else [] in
            (* NOTE(dinosaure): [EPOLLONESHOT] already disarmed it kernel-side;
               anything still waiting on it is re-armed here. *)
            arm domain fd;
            acc := List.rev_append y (List.rev_append x !acc)
          end
        end
      in
      Epoll.iter domain.events nready fn;
      !acc

let signal = Bytes.make 1 '\000'

let interrupt oc () =
  match Unix.single_write oc signal 0 1 with
  | n -> assert (n = 1)
  | exception Unix.(Unix_error (EAGAIN, _, _)) -> ()
  | exception Unix.(Unix_error (EBADF, _, _)) -> ()

let events uid =
  let domain = domain () in
  domain.epoll <- Epoll.create ();
  let ic, oc = Unix.pipe ~cloexec:true () in
  Unix.set_nonblock ic;
  Unix.set_nonblock oc;
  (* NOTE(dinosaure): the interrupt pipe stays armed for the lifetime of the
     domain, so it is registered without [EPOLLONESHOT] and never re-armed. *)
  ignore (Epoll.add domain.epoll ic Epoll.Flags.epollin);
  let select = select uid ic in
  let finaliser () =
    Unix.close ic;
    Unix.close oc;
    Epoll.close domain.epoll;
    domain.epoll <- Epoll.invalid_fd;
    Bytes.fill domain.armed 0 (Bytes.length domain.armed) '\000';
    Bytes.fill domain.drained 0 (Bytes.length domain.drained) '\000'
  in
  { Miou.interrupt= interrupt oc; select; finaliser }

let run ?quanta ?poll ?g ?domains ?handler fn =
  Miou.run ?quanta ?poll ~events ?g ?domains ?handler fn
let is_dgram { kind; _ } = match kind with Some DGRAM -> true | _ -> false

let of_file_descr ?(non_blocking = true) fd =
  if non_blocking then Unix.set_nonblock fd else Unix.clear_nonblock fd;
  match Unix.getsockopt_int fd Unix.SO_TYPE with
  | 1 -> { fd; kind= Some STREAM; non_blocking }
  | 2 -> { fd; kind= Some DGRAM; non_blocking }
  | _ ->
      failwith
        "Miou_unix.of_file_descr: invalid file-descriptor (neither TCP nor UDP)"
  | exception _exn -> { fd; kind= None; non_blocking }

let to_file_descr { fd; _ } = fd

let nonblocking_stream fam =
  let open Unix in
  let fd = socket fam SOCK_STREAM 0 in
  set_nonblock fd;
  { fd; non_blocking= true; kind= Some STREAM }

let nonblocking_dgram fam =
  let open Unix in
  let fd = socket fam SOCK_DGRAM 0 in
  set_nonblock fd;
  { fd; non_blocking= true; kind= Some DGRAM }

let unix_socket () = nonblocking_stream Unix.PF_UNIX
let tcpv4 () = nonblocking_stream Unix.PF_INET
let tcpv6 () = nonblocking_stream Unix.PF_INET6
let udpv4 () = nonblocking_dgram Unix.PF_INET
let udpv6 () = nonblocking_dgram Unix.PF_INET6

let bind_and_listen ?(backlog = 64) ?(reuseaddr = true) ?(reuseport = true)
    ({ fd; _ } as file_descr) sockaddr =
  let open Unix in
  setsockopt fd SO_REUSEADDR reuseaddr;
  setsockopt fd SO_REUSEPORT reuseport;
  bind fd sockaddr;
  (* NOTE(dinosaure): [listen] is meaningless (and fails with [EOPNOTSUPP]) on a
     datagram socket. We still allow [bind_and_listen] to bind such a socket so
     that UDP servers can share the same entry-point. *)
  if not (is_dgram file_descr) then listen fd backlog

let read file_descr ?(off = 0) ?len buf =
  let len = match len with None -> Bytes.length buf - off | Some len -> len in
  if off < 0 || len < 0 || off > Bytes.length buf - len then
    invalid_arg "Miou_unix.read";
  if is_dgram file_descr then invalid_arg "Miou_unix.read: invalid file-descr";
  unsafe_read file_descr off len buf

let read_bigstring file_descr ?(off = 0) ?len bstr =
  let len =
    match len with None -> Bigarray.Array1.dim bstr - off | Some len -> len
  in
  if off < 0 || len < 0 || off > Bigarray.Array1.dim bstr - len then
    invalid_arg "Miou_unix.read_bigstring";
  if is_dgram file_descr then
    invalid_arg "Miou_unix.read_bigstring: invalid file-descr";
  unsafe_read_bigstring file_descr off len bstr

let rec unsafe_recvfrom ({ fd; non_blocking; _ } as file_descr) off len buf
    flags =
  if non_blocking then
    match Unix.recvfrom fd buf off len flags with
    | exception Unix.(Unix_error (EINTR, _, _)) ->
        unsafe_recvfrom file_descr off len buf flags
    | exception Unix.(Unix_error ((EAGAIN | EWOULDBLOCK), _, _)) ->
        blocking_read fd;
        unsafe_recvfrom file_descr off len buf flags
    | value -> value
  else
    let rec go () =
      match Unix.recvfrom fd buf off len flags with
      | exception Unix.(Unix_error (EINTR, _, _)) -> go ()
      | value -> value
    in
    blocking_read fd; go ()

let recvfrom file_descr ?(off = 0) ?len buf flags =
  let len = match len with None -> Bytes.length buf - off | Some len -> len in
  if off < 0 || len < 0 || off > Bytes.length buf - len then
    invalid_arg "Miou_unix.recvfrom";
  if not (is_dgram file_descr) then
    invalid_arg "Miou_unix.recvfrom: invalid file-descr";
  unsafe_recvfrom file_descr off len buf flags

let rec really_read_go file_descr off len buf =
  let len' = unsafe_read file_descr off len buf in
  if len' == 0 then raise End_of_file
  else if len - len' > 0 then
    really_read_go file_descr (off + len') (len - len') buf

let really_read file_descr ?(off = 0) ?len buf =
  let len = match len with None -> Bytes.length buf - off | Some len -> len in
  if off < 0 || len < 0 || off > Bytes.length buf - len then
    invalid_arg "Miou_unix.really_read";
  if is_dgram file_descr then invalid_arg "Miou_unix.read: invalid file-descr";
  if len > 0 then really_read_go file_descr off len buf

let rec unsafe_write ({ fd; non_blocking; _ } as file_descr) off len str =
  if non_blocking then
    match Unix.write fd (Bytes.unsafe_of_string str) off len with
    | exception Unix.(Unix_error (EINTR, _, _)) ->
        unsafe_write file_descr off len str
    | exception Unix.(Unix_error ((EAGAIN | EWOULDBLOCK), _, _)) ->
        blocking_write fd;
        unsafe_write file_descr off len str
    | len' when len' < len ->
        unsafe_write file_descr (off + len') (len - len') str
    | _ -> ()
  else
    let rec go () =
      match Unix.write fd (Bytes.unsafe_of_string str) off len with
      | exception Unix.(Unix_error (EINTR, _, _)) -> go ()
      | len' when len' < len ->
          unsafe_write file_descr (off + len') (len - len') str
      | _ -> ()
    in
    blocking_write fd; go ()

let write file_descr ?(off = 0) ?len str =
  let len =
    match len with None -> String.length str - off | Some len -> len
  in
  if off < 0 || len < 0 || off > String.length str - len then
    invalid_arg "Miou_unix.write";
  if is_dgram file_descr then invalid_arg "Miou_unix.write: invalid file-descr";
  unsafe_write file_descr off len str

let rec unsafe_write_bigstring ({ fd; non_blocking; _ } as file_descr) off len
    bstr =
  if len > 0 then
    begin if non_blocking then
      match Unix.single_write_bigarray fd bstr off len with
      | exception Unix.(Unix_error (EINTR, _, _)) ->
          unsafe_write_bigstring file_descr off len bstr
      | exception Unix.(Unix_error ((EAGAIN | EWOULDBLOCK), _, _)) ->
          blocking_write fd;
          unsafe_write_bigstring file_descr off len bstr
      | written ->
          let off = off + written in
          let len = len - written in
          unsafe_write_bigstring file_descr off len bstr
    else
      let rec go off len =
        match Unix.single_write_bigarray fd bstr off len with
        | exception Unix.(Unix_error (EINTR, _, _)) -> go off len
        | written ->
            let off = off + written in
            let len = len - written in
            blocking_write fd; go off len
      in
      blocking_write fd; go off len
    end

let write_bigstring file_descr ?(off = 0) ?len bstr =
  let len =
    match len with None -> Bigarray.Array1.dim bstr - off | Some len -> len
  in
  if off < 0 || len < 0 || off > Bigarray.Array1.dim bstr - len then
    invalid_arg "Miou_unix.write_bigstring";
  if is_dgram file_descr then
    invalid_arg "Miou_unix.write_bigstring: invalid file-descr";
  unsafe_write_bigstring file_descr off len bstr

let rec unsafe_sendto ({ fd; non_blocking; _ } as file_descr) off len str flags
    sockaddr =
  if non_blocking then
    match Unix.sendto_substring fd str off len flags sockaddr with
    | exception Unix.(Unix_error (EINTR, _, _)) ->
        unsafe_sendto file_descr off len str flags sockaddr
    | exception Unix.(Unix_error ((EAGAIN | EWOULDBLOCK), _, _)) ->
        blocking_write fd;
        unsafe_sendto file_descr off len str flags sockaddr
    | value -> value
  else
    let rec go () =
      match Unix.sendto_substring fd str off len flags sockaddr with
      | exception Unix.(Unix_error (EINTR, _, _)) -> go ()
      | value -> value
    in
    blocking_write fd; go ()

let sendto file_descr ?(off = 0) ?len str flags sockaddr =
  let len =
    match len with None -> String.length str - off | Some len -> len
  in
  if off < 0 || len < 0 || off > String.length str - len then
    invalid_arg "Miou_unix.sendto";
  if not (is_dgram file_descr) then
    invalid_arg "Miou_unix.sendto: invalid file-descr";
  unsafe_sendto file_descr off len str flags sockaddr

let rec accept ?cloexec ({ fd; non_blocking; _ } as file_descr) =
  if is_dgram file_descr then invalid_arg "Miou_unix.accept: invalid file-descr";
  if non_blocking then (
    match Unix.accept ?cloexec fd with
    | exception Unix.(Unix_error (EINTR, _, _)) -> accept ?cloexec file_descr
    | exception Unix.(Unix_error ((EAGAIN | EWOULDBLOCK), _, _)) ->
        blocking_read ~name:"accept" fd;
        accept ?cloexec file_descr
    | fd, sockaddr ->
        Unix.set_nonblock fd;
        let file_descr = { fd; non_blocking= true; kind= Some STREAM } in
        (file_descr, sockaddr))
  else
    let rec go () =
      match Unix.accept ?cloexec fd with
      | exception Unix.(Unix_error (EINTR, _, _)) -> go ()
      | fd, sockaddr ->
          Unix.set_nonblock fd;
          let file_descr = { fd; non_blocking= true; kind= Some STREAM } in
          (file_descr, sockaddr)
    in
    blocking_read fd; go ()

let rec connect ({ fd; non_blocking; _ } as file_descr) sockaddr =
  if not non_blocking then
    invalid_arg
      "Miou_unix.connect: we expect a file descriptor in the non-blocking mode";
  if is_dgram file_descr then
    invalid_arg "Miou_unix.connect: invalid file-descr";
  match Unix.connect fd sockaddr with
  | () -> ()
  | exception Unix.(Unix_error (EINTR, _, _)) -> connect file_descr sockaddr
  | exception Unix.(Unix_error (EINPROGRESS, _, _)) -> (
      blocking_write ~name:"connect" fd;
      match Unix.getsockopt_error fd with
      | None -> ()
      | Some err -> raise (Unix.Unix_error (err, "connect", "")))

let sleep until =
  let syscall = Miou.syscall () in
  let domain = domain () in
  let elt =
    { time= Unix.gettimeofday () +. until; syscall; cancelled= false }
  in
  let fn () = Heapq.insert elt domain.sleepers in
  Miou.suspend ~fn syscall

module Ownership = struct
  type old = file_descr

  type file_descr = {
      fd: Unix.file_descr
    ; non_blocking: bool
    ; resource: Miou.Ownership.t
    ; kind: kind option
  }

  let bind_and_listen ?backlog { fd; non_blocking; resource; kind } sockaddr =
    Miou.Ownership.check resource;
    bind_and_listen ?backlog { fd; non_blocking; kind } sockaddr

  let read { fd; non_blocking; resource; kind } ?off ?len buf =
    Miou.Ownership.check resource;
    read { fd; non_blocking; kind } ?off ?len buf

  let really_read { fd; non_blocking; resource; kind } ?off ?len buf =
    Miou.Ownership.check resource;
    really_read { fd; non_blocking; kind } ?off ?len buf

  let write { fd; non_blocking; resource; kind } ?off ?len str =
    Miou.Ownership.check resource;
    write { fd; non_blocking; kind } ?off ?len str

  let accept ?cloexec { fd; non_blocking; resource; kind } =
    Miou.Ownership.check resource;
    let ({ fd; non_blocking; kind } : old), sockaddr =
      accept ?cloexec { fd; non_blocking; kind }
    in
    let resource = Miou.Ownership.create ~finally:Unix.close fd in
    Miou.Ownership.own resource;
    ({ fd; non_blocking; resource; kind }, sockaddr)

  let connect { fd; non_blocking; resource; kind } sockaddr =
    Miou.Ownership.check resource;
    connect { fd; non_blocking; kind } sockaddr

  let close { fd; non_blocking; resource; kind } =
    Miou.Ownership.disown resource;
    close { fd; non_blocking; kind }

  let of_file_descr ?(non_blocking = true) fd =
    let resource = Miou.Ownership.create ~finally:Unix.close fd in
    if non_blocking then Unix.set_nonblock fd else Unix.clear_nonblock fd;
    Miou.Ownership.own resource;
    match Unix.getsockopt_int fd Unix.SO_TYPE with
    | 1 -> { fd; non_blocking; resource; kind= Some STREAM }
    | 2 -> { fd; non_blocking; resource; kind= Some DGRAM }
    | _ ->
        invalid_arg
          "Miou_unix.Ownership.of_file_descr: invalid file-descriptor (neither \
           TCP nor UDP)"
    | exception _ -> { fd; non_blocking; resource; kind= None }

  let to_file_descr { fd; _ } = fd
  let resource { resource; _ } = resource

  let tcpv4 () =
    let fd = Unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
    let resource = Miou.Ownership.create ~finally:Unix.close fd in
    Unix.set_nonblock fd;
    Miou.Ownership.own resource;
    { fd; non_blocking= true; resource; kind= Some STREAM }

  let tcpv6 () =
    let fd = Unix.socket Unix.PF_INET6 Unix.SOCK_STREAM 0 in
    let resource = Miou.Ownership.create ~finally:Unix.close fd in
    Unix.set_nonblock fd;
    Miou.Ownership.own resource;
    { fd; non_blocking= true; resource; kind= Some STREAM }

  let udpv4 () =
    let fd = Unix.socket Unix.PF_INET Unix.SOCK_DGRAM 0 in
    let resource = Miou.Ownership.create ~finally:Unix.close fd in
    Unix.set_nonblock fd;
    Miou.Ownership.own resource;
    { fd; non_blocking= true; resource; kind= Some DGRAM }

  let udpv6 () =
    let fd = Unix.socket Unix.PF_INET6 Unix.SOCK_DGRAM 0 in
    let resource = Miou.Ownership.create ~finally:Unix.close fd in
    Unix.set_nonblock fd;
    Miou.Ownership.own resource;
    { fd; non_blocking= true; resource; kind= Some DGRAM }

  let recvfrom { fd; non_blocking; resource; kind } ?off ?len buf flags =
    Miou.Ownership.check resource;
    recvfrom { fd; non_blocking; kind } ?off ?len buf flags

  let sendto { fd; non_blocking; resource; kind } ?off ?len str flags sockaddr =
    Miou.Ownership.check resource;
    sendto { fd; non_blocking; kind } ?off ?len str flags sockaddr
end