package nanoev

  1. Overview
  2. Docs
Legend:
Page
Library
Module
Module type
Parameter
Class
Class type
Source

Source file nanoev_unix.ml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
open struct
  module Trace_ = Nanoev.Trace_

  let ( let@ ) = ( @@ )
  let now_ : unit -> float = Unix.gettimeofday
end

(** Callback list *)
type cbs =
  | Nil
  | Sub : 'a * 'b * (closed:bool -> 'a -> 'b -> unit) * cbs -> cbs

let[@inline] cb_is_empty = function
  | Nil -> true
  | Sub _ -> false

type timer_ev =
  | Timer : {
      deadline: float;
      x: 'a;
      y: 'b;
      f: 'a -> 'b -> unit;
    }
      -> timer_ev

type per_fd = {
  fd: Unix.file_descr;
  mutable r: cbs;
  mutable w: cbs;
}

type st = {
  timer: timer_ev Heap.t;
  fds: (Unix.file_descr, per_fd) Hashtbl.t;
  mutable sub_r: Unix.file_descr list;
  mutable sub_w: Unix.file_descr list;
  mutable sub_up_to_date: bool;
      (** are [sub_r] and [sub_w] faithful reflections of [fds]? *)
  wakeup_rd: Unix.file_descr;
  wakeup_wr: Unix.file_descr;
  wakeup_triggered: bool Atomic.t;
      (** Make [wakeup_from_outside] idempotent within an iteration of [step] *)
  in_select: bool Atomic.t;
  lock: Mutex.t;
}

let rec perform_cbs ~closed = function
  | Nil -> ()
  | Sub (x, y, f, tail) ->
    f ~closed x y;
    perform_cbs ~closed tail

let rec perform_cbs_closed ~closed = function
  | Nil -> ()
  | Sub (x, y, f, tail) ->
    f ~closed x y;
    perform_cbs_closed ~closed tail

let leq_timer (Timer a) (Timer b) = a.deadline <= b.deadline

let create_st () : st =
  let wakeup_rd, wakeup_wr = Unix.pipe () in
  Unix.set_nonblock wakeup_rd;
  {
    timer = Heap.create ~leq:leq_timer ();
    fds = Hashtbl.create 16;
    sub_r = [];
    sub_w = [];
    sub_up_to_date = true;
    wakeup_rd;
    wakeup_wr;
    wakeup_triggered = Atomic.make false;
    in_select = Atomic.make false;
    lock = Mutex.create ();
  }

let[@inline] with_lock_ (self : st) f =
  Mutex.lock self.lock;
  match f self with
  | exception e ->
    Mutex.unlock self.lock;
    raise e
  | res ->
    Mutex.unlock self.lock;
    res

let clear (self : st) =
  let@ self = with_lock_ self in
  Heap.clear self.timer;
  Hashtbl.clear self.fds;
  self.sub_r <- [];
  self.sub_w <- [];
  self.sub_up_to_date <- true;
  ()

let wakeup_from_outside (self : st) : unit =
  if not (Atomic.exchange self.wakeup_triggered true) then
    let@ _sp =
      Trace_.with_span ~__FILE__ ~__LINE__ "nanoev.wakeup-from-outside"
    in
    let b = Bytes.make 1 '!' in
    ignore (Unix.write self.wakeup_wr b 0 1 : int)

let get_fd_ (self : st) fd : per_fd =
  match Hashtbl.find self.fds fd with
  | per_fd -> per_fd
  | exception Not_found ->
    let per_fd = { fd; r = Nil; w = Nil } in
    Hashtbl.add self.fds fd per_fd;
    per_fd

let close self fd : unit =
  let@ _sp = Trace_.with_span ~__FILE__ ~__LINE__ "nanoev.close" in
  let r, w =
    let@ self = with_lock_ self in
    match Hashtbl.find self.fds fd with
    | per_fd ->
      Hashtbl.remove self.fds fd;
      self.sub_up_to_date <- false;
      if Atomic.get self.in_select then wakeup_from_outside self;
      per_fd.r, per_fd.w
    | exception Not_found ->
      invalid_arg "File descriptor is not known to Nanoev"
  in

  (* call callbacks outside of the lock *)
  perform_cbs_closed ~closed:true r;
  perform_cbs_closed ~closed:true w;
  ()

let on_readable self fd x y f : unit =
  let@ _sp = Trace_.with_span ~__FILE__ ~__LINE__ "nanoev.on-readable" in
  let@ self = with_lock_ self in
  let per_fd = get_fd_ self fd in
  per_fd.r <- Sub (x, y, f, per_fd.r);
  self.sub_up_to_date <- false;
  if Atomic.get self.in_select then wakeup_from_outside self

let on_writable self fd x y f : unit =
  let@ _sp = Trace_.with_span ~__FILE__ ~__LINE__ "nanoev.on-writable" in
  let@ self = with_lock_ self in
  let per_fd = get_fd_ self fd in
  per_fd.w <- Sub (x, y, f, per_fd.w);
  self.sub_up_to_date <- false;
  if Atomic.get self.in_select then wakeup_from_outside self

let run_after_s self time x y f : unit =
  let@ _sp = Trace_.with_span ~__FILE__ ~__LINE__ "nanoev.run-after-s" in
  let@ self = with_lock_ self in
  let deadline = now_ () +. time in
  Heap.insert self.timer (Timer { deadline; x; y; f });
  if Atomic.get self.in_select then wakeup_from_outside self

let recompute_if_needed (self : st) =
  if not self.sub_up_to_date then (
    let@ _sp = Trace_.with_span ~__FILE__ ~__LINE__ "recompute-if-needed" in
    self.sub_up_to_date <- true;
    self.sub_r <- [];
    self.sub_w <- [];
    Hashtbl.iter
      (fun fd per_fd ->
        if cb_is_empty per_fd.r && cb_is_empty per_fd.w then
          Hashtbl.remove self.fds fd;
        if not (cb_is_empty per_fd.r) then self.sub_r <- fd :: self.sub_r;
        if not (cb_is_empty per_fd.w) then self.sub_w <- fd :: self.sub_w)
      self.fds
  )

let next_deadline_ (self : st) : float option =
  match Heap.peek_min_exn self.timer with
  | exception Heap.Empty -> None
  | Timer t -> Some t.deadline

let step (self : st) : unit =
  let@ _sp = Trace_.with_span ~__FILE__ ~__LINE__ "nanoev.unix.step" in
  (* gather the subscriptions and timeout *)
  let now = now_ () in
  let timeout, sub_r, sub_w =
    let@ self = with_lock_ self in
    recompute_if_needed self;
    let timeout =
      match next_deadline_ self with
      | None -> 30.
      | Some d -> max 0. (d -. now)
    in
    timeout, self.sub_r, self.sub_w
  in

  (* run timers *)
  while
    if Heap.is_empty self.timer then
      false
    else (
      let (Timer t) = Heap.peek_min_exn self.timer in
      if t.deadline <= now then (
        ignore (Heap.pop_min_exn self.timer : timer_ev);
        t.f t.x t.y;
        true
      ) else
        false
    )
  do
    ()
  done;

  (* enter [select] *)
  Atomic.set self.in_select true;
  let r_reads, r_writes, _ =
    let@ _sp =
      Trace_.with_span ~__FILE__ ~__LINE__ "select" ~data:(fun () ->
          [
            "timeout", `Float timeout;
            "reads", `Int (List.length sub_r);
            "writes", `Int (List.length sub_w);
          ])
    in
    Unix.select (self.wakeup_rd :: sub_r) sub_w [] timeout
  in
  Atomic.set self.in_select false;

  (* drain pipe *)
  if Atomic.exchange self.wakeup_triggered false then (
    let b1 = Bytes.create 1 in
    while try Unix.read self.wakeup_rd b1 0 1 > 0 with _ -> false do
      ()
    done
  );

  (* gather the [per_fd] that are ready *)
  let ready_r = ref [] in
  let ready_w = ref [] in

  (* gather the [per_fd] that have updates *)
  (let@ self = with_lock_ self in
   if r_reads != [] || r_writes != [] then self.sub_up_to_date <- false;

   List.iter
     (fun fd ->
       if fd != self.wakeup_rd then (
         let per_fd = Hashtbl.find self.fds fd in
         ready_r := per_fd :: !ready_r
       ))
     r_reads;
   List.iter
     (fun fd ->
       let per_fd = Hashtbl.find self.fds fd in
       ready_w := per_fd :: !ready_w)
     r_writes);

  (* call callbacks *)
  List.iter
    (fun fd ->
      perform_cbs ~closed:false fd.r;
      fd.r <- Nil)
    !ready_r;
  List.iter
    (fun fd ->
      perform_cbs ~closed:false fd.w;
      fd.w <- Nil)
    !ready_w;

  ()

(* limit for select is fixed and known *)
let max_fds _ = 1024

let ops : st Nanoev.Impl.ops =
  {
    step;
    close;
    on_readable;
    on_writable;
    max_fds;
    run_after_s;
    wakeup_from_outside;
    clear;
  }

include Nanoev

let create () : t = Impl.build ops (create_st ())
OCaml

Innovation. Community. Security.