1-- The event provider's `io.programs` (D479, D487): a child is started by
2-- the system handle this provider is built over, which does not wait,
3-- and the parent's ends of its pipes are made non-blocking. Reading and
4-- writing them park the fibre on the loop as any other file's do, and
5-- waiting for the child's end parks it too: on a pidfd on Linux, and on
6-- kqueue's EVFILT_PROC with `note_exit` on Darwin and FreeBSD. The child
7-- is then reaped with POSIX's `waitpid`, which no longer waits.
8
9import core/io
10import core/text
11import hosted/system
12import platform/linux
13import platform/linux/numbers
14import platform/posix
15
16events_spawn: (self: host, path: text.host_string,
17 arguments: []text.host_string, environment: []io.variable,
18 streams: io.child_streams)
19 -> (started: io.child)
20 ! io.not_found | io.no_access | io.io_failed =
21 started = try system.start(services(self), path, arguments, environment,
22 streams)
23 -- A pipe left blocking still works, since every read and write waits
24 -- for readiness first; a non-blocking one never holds the thread.
25 if started.input_open then
26 posix.set_non_blocking(system.descriptor_of(started.input), true)
27 else none
28 end if
29 if started.output_open then
30 posix.set_non_blocking(system.descriptor_of(started.output), true)
31 else none
32 end if
33 if started.errors_open then
34 posix.set_non_blocking(system.descriptor_of(started.errors), true)
35 else none
36 end if
37end events_spawn
38
39-- Close one of a child's pipes, waking whoever waits on it first; it is
40-- closed even when the system reports a failure.
41close_pipe: (self: host, opened: io.file) -> none ! io.io_failed =
42 closing: posix.descriptor = system.descriptor_of(opened)
43 forget(self, closing)
44 posix.close(closing) else (problem)
45 record(self, problem)
46 fail io.io_failed
47 end
48end close_pipe
49
50events_write_child: (self: host, inout started: io.child, bytes: []u8)
51 -> none ! io.io_failed =
52 cleared(self)
53 fail io.io_failed when not started.input_open
54 try write_all(self, started.input, bytes)
55end events_write_child
56
57events_close_input: (self: host, inout started: io.child)
58 -> none ! io.io_failed =
59 cleared(self)
60 fail io.io_failed when not started.input_open
61 started.input_open = false
62 try close_pipe(self, started.input)
63end events_close_input
64
65-- One read of `opened` that does not wait: `waits` when nothing is ready
66-- yet, else the count, zero at its end.
67read_now: (self: host, opened: io.file, into: []mut u8)
68 -> (count: usize, waits: bool) ! io.io_failed =
69 descriptor: posix.descriptor = system.descriptor_of(opened)
70 count = 0
71 waits = true
72 return when not ready_now(descriptor, false)
73 loop do
74 got: usize = posix.read(descriptor, into) else (problem)
75 continue when problem == posix.eintr
76 return when problem == posix.eagain
77 record(self, problem)
78 fail io.io_failed
79 end
80 fail io.io_failed when got > lenof into
81 count = got
82 waits = false
83 return
84 end loop
85end read_now
86
87-- The output when it has anything, its end included, and the errors when
88-- they do; when neither has, the fibre waits for whichever is ready
89-- first, as hosted/blocking's poll of both does.
90events_read_child: (self: host, inout started: io.child, into: []mut u8)
91 -> (taken: io.child_read) ! io.io_failed =
92 cleared(self)
93 taken = (count: 0, stream: io.child_output,
94 finished: not started.output_open and not started.errors_open)
95 return when lenof into == 0
96 loop do
97 if not started.output_open and not started.errors_open then
98 taken = (count: 0, stream: io.child_output, finished: true)
99 return
100 end if
101 if started.output_open then
102 (count, waits) := try read_now(self, started.output, into)
103 if not waits then
104 if count > 0 then
105 taken = (count: count, stream: io.child_output,
106 finished: false)
107 return
108 end if
109 started.output_open = false
110 try close_pipe(self, started.output)
111 continue
112 end if
113 end if
114 if started.errors_open then
115 (count, waits) := try read_now(self, started.errors, into)
116 if not waits then
117 if count > 0 then
118 taken = (count: count, stream: io.child_errors,
119 finished: false)
120 return
121 end if
122 started.errors_open = false
123 try close_pipe(self, started.errors)
124 continue
125 end if
126 end if
127 mut output: waiting = on_descriptor(-1, false, 0)
128 mut errors: waiting = on_descriptor(-1, false, 0)
129 if started.output_open then
130 output.descriptor = i32(system.descriptor_of(started.output))
131 end if
132 if started.errors_open then
133 errors.descriptor = i32(system.descriptor_of(started.errors))
134 end if
135 answer: u8 = park_either(self, addr output, addr errors)
136 else (problem)
137 record(self, problem)
138 fail io.io_failed
139 end
140 fail io.io_failed when answer == failed
141 end loop
142end events_read_child
143
144-- Close every pipe the child still has, keeping the first failure.
145close_pipes: (self: host, started: io.child) -> (closed: bool) =
146 closed = true
147 if started.input_open then
148 close_pipe(self, started.input) else (problem)
149 _ = problem
150 closed = false
151 end
152 end if
153 if started.output_open then
154 close_pipe(self, started.output) else (problem)
155 _ = problem
156 closed = false
157 end
158 end if
159 if started.errors_open then
160 close_pipe(self, started.errors) else (problem)
161 _ = problem
162 closed = false
163 end
164 end if
165end close_pipes
166
167-- Reap the child, which has ended, again when a signal interrupts it.
168waited: (self: host, child: posix.process_id)
169 -> (status: posix.child_status) ! io.io_failed =
170 loop do
171 status = posix.wait(child) else (problem)
172 continue when problem == posix.eintr
173 record(self, problem)
174 fail io.io_failed
175 end
176 return
177 end loop
178end waited
179
180-- The pipes are closed before the wait, so a child still writing to one
181-- meets a closed pipe rather than waiting for a reader that never comes.
182events_wait_child: (self: host, sink started: io.child)
183 -> (ended: io.ending) ! io.io_failed =
184 cleared(self)
185 closed: bool = close_pipes(self, started)
186 child: posix.process_id = system.process_of(started)
187 try child_ended(self, child)
188 status: posix.child_status = try waited(self, child)
189 ended = system.ended(services(self), status)
190 fail io.io_failed when not closed
191end events_wait_child
192
193host is io.programs
194 (spawn: events_spawn, write_child: events_write_child,
195 close_input: events_close_input, read_child: events_read_child,
196 wait: events_wait_child)
197
198fixed if compiler.os == linux then
199 -- Wait for the child's end on a pidfd, which is readable once the
200 -- child has ended. A system without `pidfd_open`, before Linux 5.3,
201 -- leaves the reaping to wait the thread.
202 child_ended: (self: host, child: posix.process_id) -> none ! io.io_failed =
203 made: i64 = linux.system_call(numbers.pidfd_open_number,
204 i64(i32(child)), 0, 0, 0, 0, 0)
205 return when made < 0
206 descriptor: posix.descriptor = posix.descriptor(i32(made))
207 answer: u8 = await(self, descriptor, false, 0) else (problem)
208 posix.close(descriptor) else none
209 record(self, problem)
210 fail io.io_failed
211 end
212 forget(self, descriptor)
213 posix.close(descriptor) else none
214 fail io.io_failed when answer == failed
215 end child_ended
216elsif compiler.os == darwin or compiler.os == freebsd then
217 -- Wait for the child's end on kqueue; one that has ended already, which
218 -- the queue refuses with ESRCH, is reaped at once.
219 child_ended: (self: host, child: posix.process_id) -> none ! io.io_failed =
220 mut entry: waiting = (next: 0, descriptor: -1, output: false,
221 child: i32(child), signalled: false,
222 deadline: 0, answer: pending)
223 answer: u8 = park(self, addr entry) else (problem)
224 return when problem == posix.esrch
225 record(self, problem)
226 fail io.io_failed
227 end
228 fail io.io_failed when answer == failed
229 end child_ended
230end if