Source file affect_unix__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
(*---------------------------------------------------------------------------
   Copyright (c) 2026 The affect programmers. All rights reserved.
   SPDX-License-Identifier: ISC
  ---------------------------------------------------------------------------*)

open Affect
open Affect__base
open Affect.Action.Private

module Signal = Affect_unix__signal

(* Unblocking the cooperants *)

module Fd_unblocker =
  (* XXX eventually plug that that with system dependent fd unblocker
     modules. *)
  Affect_unix__select

module Unblocker = struct
  type t =
    { timeline : Affect_unix__timeline.t;
      signal_unblocker : Signal.Unblocker.t;
      fd_unblocker : Fd_unblocker.t; }

  let make () =
    let timeline = Affect_unix__timeline.make () in
    let signal_unblocker = Signal.Unblocker.make () in
    let fd_unblocker = Fd_unblocker.make () in
    { timeline; signal_unblocker; fd_unblocker }

  let nonblock = Some 0L
  let unblock u ~block =
    let timer_did_unblock, timeout_ns =
      Affect_unix__timeline.progress_to_next_deadline u.timeline
    in
    let signal_did_unblock = Signal.Unblocker.unblock u.signal_unblocker in
    let did_unblock = timer_did_unblock || signal_did_unblock in
    let timeout_ns =
      if not block || did_unblock then nonblock else timeout_ns
    in
    let fd_did_unblock = Fd_unblocker.unblock u.fd_unblocker ~timeout_ns in
    did_unblock || fd_did_unblock

  let set_block_bypass u () = Fd_unblocker.set_block_bypass u.fd_unblocker

  let domain_local u =
    let unblock = unblock u in
    let set_block_bypass = set_block_bypass u in
    Action.Unblocker.Domain_local.make ~unblock ~set_block_bypass ()

  let init u = Signal.Unblocker.register u.signal_unblocker
  let deinit u =
    Signal.Unblocker.unregister u.signal_unblocker;
    Fd_unblocker.dispose u.fd_unblocker

  let domain_install u =
    Affect_unix__timeline.set_domain_local u.timeline;
    Signal.Unblocker.set_domain_local u.signal_unblocker;
    Fd_unblocker.set_domain_local u.fd_unblocker

  let domain_uninstall _u =
    Affect_unix__timeline.clear_domain_local ();
    Signal.Unblocker.clear_domain_local ();
    Fd_unblocker.clear_domain_local ()
end

let unblocker () =
  (* We share a single unblocker among all domain *)
  let u = Unblocker.make () in
  let local = Unblocker.domain_local u in
  let init ~domain_count:_ = Unblocker.init u in
  let domain_local _index = local in
  let domain_install _local = Unblocker.domain_install u in
  let domain_uninstall _local = Unblocker.domain_uninstall u in
  let deinit () = Unblocker.deinit u in
  Action.Unblocker.make
    ~init ~domain_local ~domain_install ~domain_uninstall ~deinit ()

let main
    ?(sigpipe = Signal.Ignore) ?domain_spawn ?domain_count ?schedule ?handler f
  =
  Signal.set_and_restore Sys.sigpipe sigpipe @@ fun () ->
  let unblocker = unblocker () in
  Fun.Async.main ~unblocker ?domain_spawn ?domain_count ?schedule ?handler f

(* Exending and overriding Unix *)

include Unix

(* File descriptor actions *)

let wait_readable_meta = Action.Meta.make ~name:"Unix.wait_readable" ()
let wait_readable fd tag =
  let poll = Action.Primitive.poll_is_none in
  let block ~blocked =
    let blocked = Action.Blocked.Value.make tag blocked in
    let unblocker = Fd_unblocker.get_domain_local () in
    Fd_unblocker.add_wait_readable unblocker fd ~blocked
  in
  Action.Primitive.make ~meta:wait_readable_meta ~poll ~block

let wait_writable_meta = Action.Meta.make ~name:"Unix.wait_writable" ()
let wait_writable fd tag =
  let poll = Action.Primitive.poll_is_none in
  let block ~blocked =
    let blocked = Action.Blocked.Value.make tag blocked in
    let unblocker = Fd_unblocker.get_domain_local () in
    Fd_unblocker.add_wait_writable unblocker fd ~blocked
  in
  Action.Primitive.make ~meta:wait_writable_meta ~poll ~block

(* File descriptor operations *)

let close_noerr fd = try Unix.close fd with Unix.Unix_error _ -> ()

(* Sockets *)

let socket ?(cloexec = true) dom typ proto =
  let fd = Unix.socket ~cloexec dom typ proto in
  Unix.set_nonblock fd; fd

let rec accept ?cloexec fd = try Unix.accept ?cloexec fd with
| Unix.Unix_error ((EWOULDBLOCK | EAGAIN), _, _) ->
    Action.invoke (wait_readable fd ()); accept ?cloexec fd

let rec connect fd addr = try Unix.connect fd addr with
| Unix.Unix_error (EINPROGRESS, _, _) ->
    Action.invoke (wait_writable fd ());
    match Unix.getsockopt_error fd with
    | Some error -> raise (Unix.Unix_error (error, "connect", ""))
    | None -> ()

(* Reads *)

let rec read fd buf first length = try Unix.read fd buf first length with
| Unix.Unix_error ((EWOULDBLOCK | EAGAIN), _, _) ->
    Action.invoke (wait_readable fd ()); read fd buf first length

let rec read_bigarray fd buf first length =
  try Unix.read_bigarray fd buf first length with
  | Unix.Unix_error ((EWOULDBLOCK | EAGAIN), _, _) ->
      Action.invoke (wait_readable fd ()); read_bigarray fd buf first length

(* Writes *)

let rec write fd buf first length =
  try Unix.write fd buf first length with
  | Unix.Unix_error ((EWOULDBLOCK | EAGAIN), _, _) ->
      Action.invoke (wait_writable fd ()); write fd buf first length

let rec write_bigarray fd buf first length =
  try Unix.write_bigarray fd buf first length with
  | Unix.Unix_error ((EWOULDBLOCK | EAGAIN), _, _) ->
      Action.invoke (wait_writable fd ()); write_bigarray fd buf first length

let rec write_substring fd buf first length =
  try Unix.write_substring fd buf first length with
  | Unix.Unix_error ((EWOULDBLOCK | EAGAIN), _, _) ->
      Action.invoke (wait_writable fd ()); write_substring fd buf first length

let rec single_write fd buf first length =
  try Unix.single_write fd buf first length with
  | Unix.Unix_error ((EWOULDBLOCK | EAGAIN), _, _) ->
      Action.invoke (wait_writable fd ()); single_write fd buf first length

let rec single_write_bigarray fd buf first length =
  try Unix.single_write_bigarray fd buf first length with
  | Unix.Unix_error ((EWOULDBLOCK | EAGAIN), _, _) ->
      Action.invoke (wait_writable fd ());
      single_write_bigarray fd buf first length

let rec single_write_substring fd buf first length =
  try Unix.single_write_substring fd buf first length with
  | Unix.Unix_error ((EWOULDBLOCK | EAGAIN), _, _) ->
      Action.invoke (wait_writable fd ());
      single_write_substring fd buf first length