Source file lwt_engines.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
open ExtLib

module U = ExtUnix.Specific
module Ev = Libevent

class poll =
let readmask = U.Poll.(pollin + pollerr + pollhup + pollpri + pollrdhup) in
let writemask = U.Poll.(pollout + pollerr + pollhup) in
let convert (fd,i,o) = fd, U.Poll.((if i then pollin else none) + (if o then pollout else none)) in
object
val mutable buffer = [||]
inherit Lwt_engine.poll_based
method poll fds timeout =
(*
  let show = Stre.list (fun (fd,i,o) -> sprintf "%d%s%s" (U.int_of_file_descr fd) (if i then "r" else "") (if o then "w" else "")) in
  log #info "lwt poll %f %s" timeout (show fds);
*)
  let nfds = List.length fds in
  if nfds <= Array.length buffer && nfds * 2 > Array.length buffer then
  begin
    List.iteri (fun i x -> buffer.(i) <- convert x) fds;
  end
  else
    buffer <- Array.of_list @@ List.map convert fds;

  let timeout = if timeout < 0. then - 1. /. 1000. else timeout in

  let l = U.poll buffer ~n:nfds timeout |> List.map (fun (fd,f) -> fd, U.Poll.is_inter f readmask, U.Poll.is_inter f writemask) in
(*   log #info "lwt poll done %s" (show l); *)
  l
end

type Lwt_engine.engine_id += Engine_id__Devkit_libevent

(** libevent-based engine for lwt *)
class libevent =
let once_block = Ev.[ONCE] in
let once_nonblock = Ev.[ONCE;NONBLOCK] in
object(self)
  inherit Lwt_engine.abstract

  (* suppressing warning 7: for compatibility with both lwt.5.10 and lwt.6
     once we move fully to lwt.6 and drop compat with 5.10, we can remove the attribute *)
  method[@warning "-7"] id = Engine_id__Devkit_libevent
  val events_ = Ev.init ()
  val mutable pid = Unix.getpid ()
  method events =
    if Unix.getpid () <> pid then (pid <- Unix.getpid (); Ev.reinit events_);
    events_

  method private cleanup = Ev.free events_

  method iter block =
    try
      Ev.(loops self#events (if block then once_block else once_nonblock))
    with
      exn -> Exn.fail ~exn "Lwt_engines.libevent#iter"

  method private register_readable fd f =
    let ev = Ev.create () in
    Ev.set self#events ev fd [Ev.READ] ~persist:true (fun _ _ -> f ());
    Ev.add ev None;
    lazy (Ev.del ev)

  method private register_writable fd f =
    let ev = Ev.create () in
    Ev.set self#events ev fd [Ev.WRITE] ~persist:true (fun _ _ -> f ());
    Ev.add ev None;
    lazy (Ev.del ev)

  method private register_timer delay repeat f =
    let ev = Ev.create () in
    let stop = ref false in
    Ev.set_timer self#events ev ~persist:false begin fun () ->
      if not !stop then f ();
      if repeat && not !stop then Ev.add ev (Some delay);
    end;
    Ev.add ev (Some delay);
    lazy (stop := true; Ev.del ev)

end