1-- The hosted provider whose waiting operations switch fibres over the
2-- system's event mechanism (D476), `host`: epoll and signalfd on Linux,
3-- kqueue on Darwin and FreeBSD (D481). It runs its fibres on a round of
4-- turns, core/fibre/stacks' `round`, whose wait when nobody can run is
5-- this module's: a wait for events whose time limit is the earliest
6-- deadline any fibre waits for (D480). Reading and writing are here,
7-- running programs in programs.ldn, core/net's network in network.ldn,
8-- the monotonic clock and its sleep in time.ldn, core/io/signal's
9-- signals in signals.ldn, and what does not wait is the hosted/system
10-- handle's, in delegated.ldn, as hosted/blocking hands it.
11--
12-- Every operation that waits is completion-shaped (D486): it fills a
13-- `waiting` record on its own stack, submits it, and its fibre takes
14-- turns until the loop answers it, ready, expired or failed; then it
15-- makes the call again. Readiness is how the loop answers today; a
16-- completion interface would answer with the call's own result, behind
17-- the same entries. Nothing here allocates: the round's records and
18-- table, the loop's state and the last-error cells are the caller's.
19-- Import this module only at a hosted authority root.
20
21import core/fibre
22import core/fibre/stacks
23import core/io
24import core/mem
25import core/text
26import hosted/system
27import platform/kqueue
28import platform/libc
29import platform/linux
30import platform/posix
31
32-- What the loop answered a waiting record: nothing yet, its descriptor,
33-- child or signal is ready, its deadline passed, or what it waited on
34-- was shut or closed through the provider, or its signals stopped.
35pending: u8 = 0
36ready: u8 = 1
37expired: u8 = 2
38failed: u8 = 3
39
40-- One operation waiting, kept on the stack of the fibre that waits, which
41-- does not return from its call until the loop has answered it: what it
42-- waits for, a descriptor in a direction, a child process (kqueue's), a
43-- signal, a deadline on the monotonic clock in nanoseconds, or several,
44-- and the answer. `next` links the loop's list of them by address.
45waiting: type = struct
46 next: usize
47 descriptor: i32
48 output: bool
49 child: i32
50 signalled: bool
51 deadline: u64
52 answer: u8
53end waiting
54
55-- A record waiting for `descriptor` in one direction until `deadline`,
56-- zero for none.
57on_descriptor: (descriptor: i32, output: bool, deadline: u64)
58 -> (entry: waiting) =
59 entry = (next: 0, descriptor: descriptor, output: output, child: 0,
60 signalled: false, deadline: deadline, answer: pending)
61end on_descriptor
62
63-- How many reports one wait takes at most; the rest wait for the next.
64report_count: usize = 16
65
66-- How many signals `signal.kind` names.
67kinds: usize = 6
68
69-- The loop: its queue, the list of waiting records, whether a fibre took
70-- a turn of its own since the last wait, and the signals: Linux's
71-- signalfd, the watched numbers as a mask, whether watching stopped, and
72-- the arrivals not yet read, by `signal.kind`'s index, in arrival order.
73loop_state: type = struct
74 queue: i32
75 first: usize
76 turned: bool
77 reader: i32
78 watched: u64
79 stopped: bool
80 arrived: [6]u8
81 queued: usize
82 reports: [16]report
83end loop_state
84
85-- The round's idler, the wait for events, which keeps the address of
86-- the loop's state beside it in the caller's `space`.
87idle_value: type = struct
88 address: usize
89end idle_value
90
91--- Caller-owned state of an event provider: the round of turns its
92--- fibres take, and its event loop. Declare it `zeroed` and give it to
93--- `new`, which fills it; it must outlive the handle.
94public space: type = struct
95 round: stacks.round_space
96 events: loop_state
97 idling: idle_value
98end space
99
100events_value: type = struct
101 state: ptr mut loop_state
102 round: stacks.round
103end events_value
104
105--- The hosted provider whose waiting operations switch fibres: `io.world`,
106--- `io.programs`, `net.network`, the monotonic `time.clock`,
107--- `fibre.scheduler` and `fibre.turns`, and `signal.signals`, each of whose
108--- waits lets the provider's other fibres run, and every concept
109--- `hosted/system` supplies, which it hands the system's handle it is
110--- built over: `io.directories`, `io.environment`, `io.file_system`,
111--- `io.terminals`, `io.entropy` and `time.wall_clock`. A handle: copies
112--- share one round and one loop. Its files are the system's, as
113--- `hosted/system`'s `descriptor_of`, `file_of` and `socket_of` give
114--- them. Its entries that may switch are `docs/library.md`'s list (D446).
115public host: type = events_value
116
117--- Return errno captured for the terminal failing provider call, as
118--- `hosted/system`'s `last_errno` does: the cell is the process's, which
119--- this provider's own calls report through too.
120public last_errno: (handle: host) -> (number: i32) =
121 number = system.last_errno(services(handle))
122end last_errno
123
124--- Return the number of the signal that ended the child this handle last
125--- saw end through `io.wait`, as `hosted/system`'s `last_signal` does.
126public last_signal: (handle: host) -> (number: i32) =
127 number = system.last_signal(services(handle))
128end last_signal
129
130--- Fill `state` and return a handle to the event provider over it and
131--- over `entries`: a round of turns over that table of the fibres that
132--- may be unfinished at once, one more than a group's count, an event
133--- queue, closed on exec, and the process's `hosted/system` handle, whose
134--- last-error and last-signal cells it reports through. Fails
135--- `mem.out_of_memory` when `entries` holds fewer than two,
136--- and `io.io_failed` when the system makes no queue, with its errno in
137--- the process's cell. Calling this before the hosted entry has
138--- established the process's arguments traps, as `hosted/system`'s `new`
139--- does. The queue stays open until the process ends.
140public new: (inout state: space, entries: []mut usize)
141 -> (handle: host from state, entries)
142 ! mem.out_of_memory | io.io_failed =
143 system.clear_errno(system.new())
144 state.events.queue = -1
145 state.events.first = 0
146 state.events.turned = false
147 state.events.reader = -1
148 state.events.watched = 0
149 state.events.stopped = false
150 state.events.queued = 0
151 state.idling = (address: usize(addr state.events))
152 round: stacks.round =
153 try stacks.round_over(state.round, entries, any(addr state.idling))
154 state.events.queue = made_queue() else (problem)
155 system.record_errno(system.new(), problem)
156 fail io.io_failed
157 end
158 handle = (state: addr state.events, round: round)
159end new
160
161-- The hosted/system handle beneath: the process's, which borrows
162-- nothing, so a handle `from` its caller's state carries no storage of
163-- the system's.
164services: (self: host) -> (handle: system.host) =
165 _ = self
166 handle = system.new()
167end services
168
169-- Keep a failure's errno in the system handle's cell.
170record: (self: host, problem: posix.errno) -> none =
171 system.record_errno(services(self), problem)
172end record
173
174-- Clear the cell before a call asks the system.
175cleared: (self: host) -> none =
176 system.clear_errno(services(self))
177end cleared
178
179-- The waiting record at `address`.
180entry_at: (address: usize) -> (entry: ptr mut waiting) =
181 -- [0810]: a list entry is the address of a record on the stack of a
182 -- fibre that waits in it.
183 entry = ptr(address)
184end entry_at
185
186-- Put `entry` on the loop's list.
187link: (state: ptr mut loop_state, entry: ptr mut waiting) -> none =
188 entry.val.next = state.val.first
189 state.val.first = usize(entry)
190end link
191
192-- Take `entry` off the loop's list, where it may not be.
193unlink: (state: ptr mut loop_state, entry: ptr mut waiting) -> none =
194 sought: usize = usize(entry)
195 if state.val.first == sought then
196 state.val.first = entry.val.next
197 return
198 end if
199 mut at: usize = state.val.first
200 while at <> 0 do
201 here: ptr mut waiting = entry_at(at)
202 if here.val.next == sought then
203 here.val.next = entry.val.next
204 return
205 end if
206 at = here.val.next
207 end while
208end unlink
209
210-- Answer `entry`, taking it off the list; its fibre sees the answer at
211-- its next turn.
212settle: (state: ptr mut loop_state, entry: ptr mut waiting, answer: u8)
213 -> none =
214 unlink(state, entry)
215 entry.val.answer = answer
216end settle
217
218-- Answer every record waiting on `descriptor` for input when `input`,
219-- and for output when `output`.
220answer_descriptor: (state: ptr mut loop_state, descriptor: i32, input: bool,
221 output: bool, answer: u8) -> none =
222 mut at: usize = state.val.first
223 while at <> 0 do
224 here: ptr mut waiting = entry_at(at)
225 at = here.val.next
226 if here.val.descriptor == descriptor
227 and ((here.val.output and output)
228 or (not here.val.output and input)) then
229 settle(state, here, answer)
230 end if
231 end while
232end answer_descriptor
233
234-- Whether any record waits on `descriptor` for input, and for output.
235wanted_on: (state: ptr mut loop_state, descriptor: i32)
236 -> (input: bool, output: bool) =
237 input = false
238 output = false
239 mut at: usize = state.val.first
240 while at <> 0 do
241 here: ptr mut waiting = entry_at(at)
242 if here.val.descriptor == descriptor then
243 if here.val.output then output = true else input = true end if
244 end if
245 at = here.val.next
246 end while
247end wanted_on
248
249-- Answer every record waiting for a signal.
250answer_signals: (state: ptr mut loop_state, answer: u8) -> none =
251 mut at: usize = state.val.first
252 while at <> 0 do
253 here: ptr mut waiting = entry_at(at)
254 at = here.val.next
255 if here.val.signalled then settle(state, here, answer) end if
256 end while
257end answer_signals
258
259-- Answer every record waiting for the child `child` to end.
260answer_child: (state: ptr mut loop_state, child: i32) -> none =
261 mut at: usize = state.val.first
262 while at <> 0 do
263 here: ptr mut waiting = entry_at(at)
264 at = here.val.next
265 if child <> 0 and here.val.child == child then
266 settle(state, here, ready)
267 end if
268 end while
269end answer_child
270
271-- Answer every record whose deadline the clock has reached.
272answer_expired: (state: ptr mut loop_state, now: u64) -> none =
273 mut at: usize = state.val.first
274 while at <> 0 do
275 here: ptr mut waiting = entry_at(at)
276 at = here.val.next
277 if here.val.deadline <> 0 and here.val.deadline <= now then
278 settle(state, here, expired)
279 end if
280 end while
281end answer_expired
282
283-- The earliest deadline any record waits for, zero for none.
284earliest: (state: ptr mut loop_state) -> (deadline: u64) =
285 deadline = 0
286 mut at: usize = state.val.first
287 while at <> 0 do
288 here: ptr mut waiting = entry_at(at)
289 if here.val.deadline <> 0
290 and (deadline == 0 or here.val.deadline < deadline) then
291 deadline = here.val.deadline
292 end if
293 at = here.val.next
294 end while
295end earliest
296
297-- The longest a wait sleeps at once, a day, in milliseconds; the loop
298-- looks at its deadlines again after it.
299longest_wait: u64 = 86400000
300
301-- The wait for events, when every fibre of the round has taken a turn
302-- since anything last started or ended: with no limit when every fibre
303-- waits on the loop and none has a deadline, until the earliest
304-- deadline, a part of a millisecond counting as a whole one, and not at
305-- all when a fibre took a turn of its own, which is runnable work. It
306-- answers what the reports say is ready, then every record whose
307-- deadline has passed. With nothing waiting it returns at once.
308loop_idle: (self: idle_value) -> none =
309 -- [0810]: the address of the loop's state `new` kept.
310 state: ptr mut loop_state = ptr(self.address)
311 turned: bool = state.val.turned
312 state.val.turned = false
313 return when state.val.first == 0 and not turned
314 mut limit: i32 = -1
315 if turned then
316 limit = 0
317 else
318 soonest: u64 = earliest(state)
319 if soonest <> 0 then
320 now: u64 = reading()
321 if soonest <= now then
322 limit = 0
323 else
324 mut left: u64 = (soonest - now + 999999) / 1000000
325 if left > longest_wait then left = longest_wait end if
326 limit = i32(left)
327 end if
328 end if
329 end if
330 count: usize = wait_events(state, limit)
331 dispatch(state, count)
332 answer_expired(state, reading())
333end loop_idle
334
335idle_value is stacks.idler (idle: loop_idle)
336
337-- Submit `entry` and take turns until the loop answers it: the answer,
338-- or the errno of a submission the system refused, with nothing left
339-- waiting. The record is the caller's local, and the caller is this
340-- fibre, which does not return from here until the record is off the
341-- list.
342park: (self: host, entry: ptr mut waiting) -> (answer: u8) ! posix.errno =
343 state: ptr mut loop_state = self.state
344 entry.val.answer = pending
345 link(state, entry)
346 arm(state, entry) else (problem)
347 unlink(state, entry)
348 fail problem
349 end
350 while entry.val.answer == pending do
351 turn(self.round)
352 end while
353 answer = entry.val.answer
354end park
355
356-- Submit two records and take turns until the loop answers either; the
357-- other is withdrawn. The answer is the first's when it was answered,
358-- else the second's.
359park_either: (self: host, first: ptr mut waiting, second: ptr mut waiting)
360 -> (answer: u8) ! posix.errno =
361 state: ptr mut loop_state = self.state
362 first.val.answer = pending
363 second.val.answer = pending
364 link(state, first)
365 arm(state, first) else (problem)
366 unlink(state, first)
367 fail problem
368 end
369 link(state, second)
370 arm(state, second) else (problem)
371 unlink(state, first)
372 unlink(state, second)
373 fail problem
374 end
375 while first.val.answer == pending and second.val.answer == pending do
376 turn(self.round)
377 end while
378 unlink(state, first)
379 unlink(state, second)
380 answer = if first.val.answer <> pending then first.val.answer
381 else second.val.answer end if
382end park_either
383
384-- Wake every fibre waiting on `descriptor`, failing its call, and stop
385-- watching it: the descriptor is about to be closed through the
386-- provider (D449).
387forget: (self: host, descriptor: posix.descriptor) -> none =
388 answer_descriptor(self.state, i32(descriptor), true, true, failed)
389 withdraw(self.state, i32(descriptor))
390end forget
391
392-- Whether `descriptor` is ready in the direction asked now, without
393-- waiting: a regular file always is.
394ready_now: (descriptor: posix.descriptor, output: bool) -> (yes: bool) =
395 asked: posix.poll_events = if output then posix.poll_output
396 else posix.poll_input end if
397 mut entries: [1]posix.poll_entry = [posix.watch(descriptor, asked)]
398 found: usize = posix.poll(entries[0..<1], 0) else (problem)
399 -- A descriptor poll refuses is left to the call to fail on.
400 _ = problem
401 yes = true
402 return
403 end
404 yes = found > 0
405end ready_now
406
407-- Wait for `descriptor` in one direction until `deadline`, zero for
408-- none: the loop's answer.
409await: (self: host, descriptor: posix.descriptor, output: bool,
410 deadline: u64) -> (answer: u8) ! posix.errno =
411 mut entry: waiting = on_descriptor(i32(descriptor), output, deadline)
412 answer = try park(self, addr entry)
413end await
414
415-- The descriptor ready in the direction asked, waiting for it as a
416-- fibre does: failed when it was shut or closed through the provider.
417until_ready: (self: host, descriptor: posix.descriptor, output: bool)
418 -> none ! io.io_failed =
419 return when ready_now(descriptor, output)
420 answer: u8 = await(self, descriptor, output, 0) else (problem)
421 record(self, problem)
422 fail io.io_failed
423 end
424 fail io.io_failed when answer == failed
425end until_ready
426
427-- Write some of `bytes` to `opened`: a send on a socket, a write
428-- otherwise.
429put_some: (opened: io.file, bytes: []u8) -> (count: usize) ! posix.errno =
430 if system.is_socket(opened) then
431 count = try posix.send(system.descriptor_of(opened), bytes)
432 else
433 count = try posix.write(system.descriptor_of(opened), bytes)
434 end if
435end put_some
436
437-- Read into `into` from `opened`, waiting for it to be ready as a fibre
438-- does; a regular file is always ready, and is read at once.
439events_read: (self: host, opened: io.file, into: []mut u8)
440 -> (count: usize) ! io.io_failed =
441 cleared(self)
442 count = 0
443 return when lenof into == 0
444 descriptor: posix.descriptor = system.descriptor_of(opened)
445 loop do
446 try until_ready(self, descriptor, false)
447 got: usize = posix.read(descriptor, into) else (problem)
448 -- Nothing was read: a signal came first, or another fibre
449 -- read what was ready.
450 continue when problem == posix.eintr or problem == posix.eagain
451 record(self, problem)
452 fail io.io_failed
453 end
454 fail io.io_failed when got > lenof into
455 count = got
456 return
457 end loop
458end events_read
459
460-- Write some of `bytes` once `opened` has room, waiting for it as a fibre
461-- does: a positive count no greater than the slice's.
462written_some: (self: host, opened: io.file, bytes: []u8)
463 -> (count: usize) ! io.io_failed =
464 descriptor: posix.descriptor = system.descriptor_of(opened)
465 loop do
466 try until_ready(self, descriptor, true)
467 accepted: usize = put_some(opened, bytes) else (problem)
468 continue when problem == posix.eintr or problem == posix.eagain
469 record(self, problem)
470 fail io.io_failed
471 end
472 fail io.io_failed when accepted == 0 or accepted > lenof bytes
473 count = accepted
474 return
475 end loop
476end written_some
477
478write_all: (self: host, opened: io.file, bytes: []u8)
479 -> none ! io.io_failed =
480 mut written: usize = 0
481 while written < lenof bytes do
482 written += try written_some(self, opened,
483 bytes[written..<lenof bytes])
484 end while
485end write_all
486
487events_write: (self: host, opened: io.file, bytes: []u8)
488 -> none ! io.io_failed =
489 cleared(self)
490 return when lenof bytes == 0
491 try write_all(self, opened, bytes)
492end events_write
493
494events_write_some: (self: host, opened: io.file, bytes: []u8)
495 -> (count: usize) ! io.io_failed =
496 cleared(self)
497 if lenof bytes == 0 then
498 count = 0
499 return
500 end if
501 mut requested: usize = lenof bytes
502 if requested > 65536 then requested = 65536 end if
503 count = try written_some(self, opened, bytes[0..<requested])
504end events_write_some
505
506-- What does not wait is the system's: its files, streams and arguments.
507-- Closing a file wakes every fibre waiting on it first (D449).
508events_open_read: (self: host, path: text.host_string)
509 -> (opened: io.file)
510 ! io.not_found | io.no_access | io.io_failed =
511 opened = try io.open_read(services(self), path)
512end events_open_read
513
514events_open_write: (self: host, path: text.host_string)
515 -> (opened: io.file)
516 ! io.not_found | io.no_access | io.io_failed =
517 opened = try io.open_write(services(self), path)
518end events_open_write
519
520events_same_file: (self: host, left: text.host_string,
521 right: text.host_string)
522 -> (same: bool) ! io.io_failed =
523 same = try io.same_file(services(self), left, right)
524end events_same_file
525
526events_close: (self: host, sink opened: io.file) -> none ! io.io_failed =
527 forget(self, system.descriptor_of(opened))
528 try io.close(services(self), opened)
529end events_close
530
531events_input: (self: host) -> (stream: io.file) =
532 stream = io.input(services(self))
533end events_input
534
535events_out: (self: host) -> (stream: io.file) =
536 stream = io.out(services(self))
537end events_out
538
539events_err: (self: host) -> (stream: io.file) =
540 stream = io.err(services(self))
541end events_err
542
543events_argument_count: (self: host) -> (count: usize) =
544 count = io.argument_count(services(self))
545end events_argument_count
546
547events_argument: (self: host, index: usize)
548 -> (value: text.host_string from self) ! io.no_argument =
549 value = try io.argument_at(services(self), index)
550end events_argument
551
552host is io.reader (read: events_read)
553host is io.writer (write: events_write, write_some: events_write_some)
554host is io.files
555 (open_read: events_open_read, open_write: events_open_write,
556 same_file: events_same_file, close: events_close)
557host is io.process
558 (input: events_input, out: events_out, err: events_err,
559 argument_count: events_argument_count, argument: events_argument)
560host is io.world ()
561
562-- The round's entries, with the loop told when a fibre takes a turn of
563-- its own, which keeps the next wait for events from sleeping.
564-- The round's own entries, through its evidence: a round's
565-- representation is core/fibre/stacks'.
566turn: (provider: type is fibre.turns, runs: provider) -> none =
567 provider.yield(runs)
568end turn
569
570started: (provider: type is fibre.turns, runs: provider,
571 escaping record: ptr mut fibre.space, escaping stack: []mut u8,
572 escaping work: any fibre.task) -> none ! mem.out_of_memory =
573 try provider.start(runs, record, stack, work)
574end started
575
576ended: (provider: type is fibre.turns, runs: provider, record: fibre.space)
577 -> (yes: bool) =
578 yes = provider.finished(runs, record)
579end ended
580
581waited_for: (provider: type is fibre.turns, runs: provider) -> none =
582 provider.wait(runs)
583end waited_for
584
585events_start: (self: host, escaping record: ptr mut fibre.space,
586 escaping stack: []mut u8, escaping work: any fibre.task)
587 -> none ! mem.out_of_memory =
588 try started(self.round, record, stack, work)
589end events_start
590
591events_finished: (self: host, record: fibre.space) -> (yes: bool) =
592 yes = ended(self.round, record)
593end events_finished
594
595events_wait: (self: host) -> none =
596 waited_for(self.round)
597end events_wait
598
599events_yield: (self: host) -> none =
600 self.state.val.turned = true
601 turn(self.round)
602end events_yield
603
604host is fibre.scheduler (start: events_start, finished: events_finished,
605 wait: events_wait)
606host is fibre.turns (yield: events_yield)
607
608-- Each system's event mechanism beneath the loop: epoll and a signalfd
609-- on Linux, kqueue on Darwin and FreeBSD. Each supplies the record a
610-- wait reports, the queue, arming a waiting record, withdrawing a
611-- descriptor about to close, the wait and its dispatch, and the signal
612-- numbers by `signal.kind`'s order.
613fixed if compiler.os == linux then
614 report: type = linux.epoll_record
615
616 -- SIGINT, SIGTERM, SIGHUP, SIGCHLD, SIGUSR1 and SIGUSR2, as Linux
617 -- numbers them on every architecture.
618 signal_numbers: [6]i32 = [2, 15, 1, 17, 10, 12]
619
620 made_queue: () -> (queue: i32) ! posix.errno =
621 queue = linux.epoll_create1(linux.epoll_cloexec)
622 fail posix.errno_of(-queue) when queue < 0
623 end made_queue
624
625 -- Watch a record's descriptor for what every record waiting on it
626 -- wants, once: a report disarms it until it is armed again. A
627 -- record waiting only for a signal or a deadline arms nothing; the
628 -- signalfd is watched for as long as signals are.
629 arm: (state: ptr mut loop_state, entry: ptr mut waiting)
630 -> none ! posix.errno =
631 return when entry.val.descriptor < 0
632 try armed(state, entry.val.descriptor)
633 end arm
634
635 armed: (state: ptr mut loop_state, descriptor: i32)
636 -> none ! posix.errno =
637 (input, output) := wanted_on(state, descriptor)
638 return when not input and not output
639 mut asked: linux.epoll_record = zeroed
640 asked.events = linux.epolloneshot
641 if input then asked.events = asked.events | linux.epollin end if
642 if output then asked.events = asked.events | linux.epollout end if
643 asked.data = u64(u32(descriptor))
644 mut result: i32 = linux.epoll_ctl(state.val.queue,
645 linux.epoll_ctl_mod, descriptor,
646 addr asked)
647 if result == -2 then
648 result = linux.epoll_ctl(state.val.queue, linux.epoll_ctl_add,
649 descriptor, addr asked)
650 end if
651 fail posix.errno_of(-result) when result < 0
652 end armed
653
654 withdraw: (state: ptr mut loop_state, descriptor: i32) -> none =
655 mut unused: linux.epoll_record = zeroed
656 _ = linux.epoll_ctl(state.val.queue, linux.epoll_ctl_del, descriptor,
657 addr unused)
658 end withdraw
659
660 wait_events: (state: ptr mut loop_state, milliseconds: i32)
661 -> (count: usize) =
662 count = 0
663 result: i32 = linux.epoll_pwait(state.val.queue,
664 addr state.val.reports[0],
665 i32(report_count), milliseconds,
666 linux.no_mask)
667 -- EINTR, or a failure no wait of the loop's own makes: the loop
668 -- looks again.
669 return when result <= 0
670 count = usize(result)
671 end wait_events
672
673 dispatch: (state: ptr mut loop_state, count: usize) -> none =
674 for index in usize(0)..<count do
675 got: linux.epoll_record = state.val.reports[index]
676 descriptor: i32 = i32(u32(got.data))
677 if descriptor == state.val.reader then
678 collect(state)
679 answer_signals(state, ready)
680 else
681 broken: u32 = linux.epollerr | linux.epollhup
682 input: bool =
683 (got.events & (linux.epollin | linux.epollrdhup | broken))
684 <> 0
685 output: bool = (got.events & (linux.epollout | broken)) <> 0
686 answer_descriptor(state, descriptor, input, output, ready)
687 -- One report disarmed it for every record: arm it again
688 -- for those still waiting.
689 armed(state, descriptor) else (problem)
690 _ = problem
691 answer_descriptor(state, descriptor, true, true, ready)
692 end
693 end if
694 end for
695 end dispatch
696
697 -- Read every signal the signalfd holds into the arrivals.
698 collect: (state: ptr mut loop_state) -> none =
699 return when state.val.reader < 0
700 loop do
701 mut taken: linux.signalfd_record = zeroed
702 into: ptr mut u8 = ptr(usize(addr taken))
703 got: isize = linux.read(state.val.reader, into,
704 sizeof linux.signalfd_record)
705 break when got <> isize(sizeof linux.signalfd_record)
706 arrive(state, i32(taken.signal))
707 end loop
708 end collect
709
710 -- Take `number` from its default disposition: blocked, and read
711 -- through the loop's signalfd, which the loop watches from the first.
712 take_signal: (state: ptr mut loop_state, number: i32)
713 -> none ! posix.errno =
714 mut adding: u64 = linux.signal_bit(number)
715 result: i32 = linux.rt_sigprocmask(linux.sig_block, addr adding,
716 linux.no_mask)
717 fail posix.errno_of(-result) when result < 0
718 mut mask: u64 = state.val.watched | adding
719 flags: i32 = linux.sfd_cloexec | linux.sfd_nonblock
720 made: i32 = linux.signalfd4(state.val.reader, addr mask, flags)
721 fail posix.errno_of(-made) when made < 0
722 if state.val.reader < 0 then
723 mut asked: linux.epoll_record = zeroed
724 asked.events = linux.epollin
725 asked.data = u64(u32(made))
726 added: i32 = linux.epoll_ctl(state.val.queue,
727 linux.epoll_ctl_add, made,
728 addr asked)
729 if added < 0 then
730 _ = linux.close(made)
731 fail posix.errno_of(-added)
732 end if
733 state.val.reader = made
734 end if
735 end take_signal
736
737 -- Give every watched signal back: read what arrived, so that none is
738 -- delivered as its default when it is unblocked, close the signalfd
739 -- and unblock them.
740 give_signals: (state: ptr mut loop_state) -> none ! posix.errno =
741 collect(state)
742 if state.val.reader >= 0 then
743 withdraw(state, state.val.reader)
744 _ = linux.close(state.val.reader)
745 state.val.reader = -1
746 end if
747 mut mask: u64 = state.val.watched
748 result: i32 = linux.rt_sigprocmask(linux.sig_unblock, addr mask,
749 linux.no_mask)
750 fail posix.errno_of(-result) when result < 0
751 end give_signals
752elsif compiler.os == darwin or compiler.os == freebsd then
753 report: type = kqueue.event_record
754
755 -- SIGINT, SIGTERM, SIGHUP, SIGCHLD, SIGUSR1 and SIGUSR2, as Darwin
756 -- and FreeBSD number them.
757 signal_numbers: [6]i32 = [2, 15, 1, 20, 30, 31]
758
759 -- The errno the C library left.
760 failure: () -> (problem: posix.errno) =
761 cell: ptr mut i32 = libc.errno_cell()
762 problem = posix.errno_of(cell.val)
763 end failure
764
765 made_queue: () -> (queue: i32) ! posix.errno =
766 queue = kqueue.kqueue()
767 fail failure() when queue < 0
768 if libc.fcntl(queue, libc.f_setfd, libc.fd_cloexec) <> 0 then
769 problem: posix.errno = failure()
770 _ = libc.close(queue)
771 fail problem
772 end if
773 end made_queue
774
775 -- Make one change to the queue, without waiting.
776 changed: (state: ptr mut loop_state, change: kqueue.event_record)
777 -> none ! posix.errno =
778 no_wait: libc.timespec = (seconds: 0, nanoseconds: 0)
779 result: i32 = kqueue.kevent(state.val.queue, addr change, 1,
780 kqueue.none_given, 0, addr no_wait)
781 fail failure() when result < 0
782 end changed
783
784 -- Watch a record's descriptor in its direction, or its child for its
785 -- end, once. ESRCH for a child that has ended already.
786 arm: (state: ptr mut loop_state, entry: ptr mut waiting)
787 -> none ! posix.errno =
788 oneshot: u16 = kqueue.ev_add | kqueue.ev_oneshot
789 if entry.val.descriptor >= 0 then
790 filter: i16 = if entry.val.output then kqueue.evfilt_write
791 else kqueue.evfilt_read end if
792 try changed(state, kqueue.change(usize(u32(entry.val.descriptor)),
793 filter, oneshot, 0, 0))
794 elsif entry.val.child > 0 then
795 try changed(state, kqueue.change(usize(u32(entry.val.child)),
796 kqueue.evfilt_proc, oneshot,
797 kqueue.note_exit, 0))
798 end if
799 end arm
800
801 -- Closing a descriptor removes its events from the queue.
802 withdraw: (state: ptr mut loop_state, descriptor: i32) -> none =
803 _ = state
804 _ = descriptor
805 end withdraw
806
807 wait_events: (state: ptr mut loop_state, milliseconds: i32)
808 -> (count: usize) =
809 count = 0
810 limit: libc.timespec = (seconds: i64(milliseconds / 1000),
811 nanoseconds: i64(milliseconds % 1000)
812 * 1000000)
813 bound: kqueue.wait_limit = if milliseconds < 0 then kqueue.none_given
814 else addr limit end if
815 result: i32 = kqueue.kevent(state.val.queue, kqueue.none_given, 0,
816 addr state.val.reports[0],
817 i32(report_count), bound)
818 return when result <= 0
819 count = usize(result)
820 end wait_events
821
822 dispatch: (state: ptr mut loop_state, count: usize) -> none =
823 for index in usize(0)..<count do
824 got: kqueue.event_record = state.val.reports[index]
825 identifier: i32 = i32(u32(got.identifier))
826 if got.filter == kqueue.evfilt_read then
827 answer_descriptor(state, identifier, true, false, ready)
828 elsif got.filter == kqueue.evfilt_write then
829 answer_descriptor(state, identifier, false, true, ready)
830 elsif got.filter == kqueue.evfilt_signal then
831 arrive(state, identifier)
832 answer_signals(state, ready)
833 elsif got.filter == kqueue.evfilt_proc then
834 answer_child(state, identifier)
835 end if
836 end for
837 end dispatch
838
839 -- kqueue reports a signal itself; nothing waits to be read.
840 collect: (state: ptr mut loop_state) -> none =
841 _ = state
842 end collect
843
844 -- Set `number`'s disposition to `handler`.
845 disposed: (number: i32, handler: u64) -> none ! posix.errno =
846 mut action: libc.signal_action = zeroed
847 action.handler = handler
848 fail failure() when libc.sigaction(number, addr action,
849 libc.no_address) <> 0
850 end disposed
851
852 -- Take `number` from its default disposition: ignored, so that its
853 -- arrival does nothing but the queue's report, which counts it
854 -- whatever its disposition. SIGCHLD keeps its default, which
855 -- ignores it already: ignoring it outright would reap every child
856 -- before `wait` could.
857 take_signal: (state: ptr mut loop_state, number: i32)
858 -> none ! posix.errno =
859 if number <> signal_numbers[3] then
860 try disposed(number, libc.sig_ign)
861 end if
862 try changed(state, kqueue.change(usize(u32(number)),
863 kqueue.evfilt_signal, kqueue.ev_add,
864 0, 0))
865 end take_signal
866
867 -- Give every watched signal back: no longer reported, and its
868 -- default disposition again.
869 give_signals: (state: ptr mut loop_state) -> none ! posix.errno =
870 for index in usize(0)..<kinds do
871 number: i32 = signal_numbers[index]
872 if (state.val.watched & (u64(1) << u64(number - 1))) <> 0 then
873 changed(state, kqueue.change(usize(u32(number)),
874 kqueue.evfilt_signal,
875 kqueue.ev_delete, 0, 0))
876 else none
877 if index <> 3 then
878 try disposed(number, libc.sig_dfl)
879 end if
880 end if
881 end for
882 end give_signals
883end if