Duncan Coutts pushed to branch wip/io-manager-deadlock-detection at Glasgow Haskell Compiler / GHC
WARNING: The push did not contain any new commits, but force pushed to delete the commits and changes below.
Deleted commits:
-
e9c4d16c
by Duncan Coutts at 2026-02-16T00:30:50+00:00
-
fcc1cdd3
by Duncan Coutts at 2026-02-16T00:30:50+00:00
-
4e4aa84f
by Duncan Coutts at 2026-02-16T00:30:50+00:00
-
6fe60774
by Duncan Coutts at 2026-02-16T00:30:50+00:00
-
24b6f621
by Duncan Coutts at 2026-02-16T00:30:50+00:00
-
bc97a19f
by Duncan Coutts at 2026-02-16T00:30:50+00:00
12 changed files:
- rts/Capability.c
- rts/IOManager.c
- rts/IOManager.h
- rts/IOManagerInternals.h
- rts/Schedule.c
- + rts/posix/FdWakeup.c
- + rts/posix/FdWakeup.h
- rts/posix/Poll.c
- rts/posix/Poll.h
- rts/posix/Select.c
- rts/posix/Select.h
- rts/rts.cabal
Changes:
| ... | ... | @@ -1280,6 +1280,7 @@ shutdownCapabilities(Task *task, bool safe) |
| 1280 | 1280 | static void
|
| 1281 | 1281 | freeCapability (Capability *cap)
|
| 1282 | 1282 | {
|
| 1283 | + freeCapabilityIOManager(cap->iomgr);
|
|
| 1283 | 1284 | stgFree(cap->mut_lists);
|
| 1284 | 1285 | stgFree(cap->saved_mut_lists);
|
| 1285 | 1286 | if (cap->current_segments) {
|
| ... | ... | @@ -343,9 +343,7 @@ void initCapabilityIOManager(CapIOManager *iomgr) |
| 343 | 343 | switch (iomgr_type) {
|
| 344 | 344 | #if defined(IOMGR_ENABLED_SELECT)
|
| 345 | 345 | case IO_MANAGER_SELECT:
|
| 346 | - iomgr->blocked_queue_hd = END_TSO_QUEUE;
|
|
| 347 | - iomgr->blocked_queue_tl = END_TSO_QUEUE;
|
|
| 348 | - iomgr->sleeping_queue = END_TSO_QUEUE;
|
|
| 346 | + initCapabilityIOManagerSelect(iomgr);
|
|
| 349 | 347 | break;
|
| 350 | 348 | #endif
|
| 351 | 349 | |
| ... | ... | @@ -373,6 +371,26 @@ void initCapabilityIOManager(CapIOManager *iomgr) |
| 373 | 371 | }
|
| 374 | 372 | |
| 375 | 373 | |
| 374 | +void freeCapabilityIOManager(CapIOManager *iomgr)
|
|
| 375 | +{
|
|
| 376 | + switch (iomgr_type) {
|
|
| 377 | +#if defined(IOMGR_ENABLED_SELECT)
|
|
| 378 | + case IO_MANAGER_SELECT:
|
|
| 379 | + freeCapabilityIOManagerSelect(iomgr);
|
|
| 380 | + break;
|
|
| 381 | +#endif
|
|
| 382 | + |
|
| 383 | +#if defined(IOMGR_ENABLED_POLL)
|
|
| 384 | + case IO_MANAGER_POLL:
|
|
| 385 | + freeCapabilityIOManagerPoll(iomgr);
|
|
| 386 | + break;
|
|
| 387 | +#endif
|
|
| 388 | + default:
|
|
| 389 | + break;
|
|
| 390 | + }
|
|
| 391 | +}
|
|
| 392 | + |
|
| 393 | + |
|
| 376 | 394 | /* Called late in the RTS initialisation
|
| 377 | 395 | */
|
| 378 | 396 | void startIOManager(void)
|
| ... | ... | @@ -546,8 +564,21 @@ exitIOManager(bool wait_threads) |
| 546 | 564 | */
|
| 547 | 565 | void wakeupIOManager(void)
|
| 548 | 566 | {
|
| 567 | + debugTrace(DEBUG_iomanager, "Sending wakeup to I/O manager...");
|
|
| 549 | 568 | switch (iomgr_type) {
|
| 550 | 569 | |
| 570 | +#if defined(IOMGR_ENABLED_SELECT)
|
|
| 571 | + case IO_MANAGER_SELECT:
|
|
| 572 | + wakeupIOManagerSelect(MainCapability.iomgr);
|
|
| 573 | + break;
|
|
| 574 | +#endif
|
|
| 575 | + |
|
| 576 | +#if defined(IOMGR_ENABLED_POLL)
|
|
| 577 | + case IO_MANAGER_POLL:
|
|
| 578 | + wakeupIOManagerPoll(MainCapability.iomgr);
|
|
| 579 | + break;
|
|
| 580 | +#endif
|
|
| 581 | + |
|
| 551 | 582 | #if defined(IOMGR_ENABLED_MIO_POSIX)
|
| 552 | 583 | case IO_MANAGER_MIO_POSIX:
|
| 553 | 584 | /* MIO Posix implementation in posix/Signals.c */
|
| ... | ... | @@ -572,8 +603,13 @@ void wakeupIOManager(void) |
| 572 | 603 | #endif
|
| 573 | 604 | break;
|
| 574 | 605 | #endif
|
| 575 | - default:
|
|
| 606 | +#if defined(IOMGR_ENABLED_WIN32_LEGACY)
|
|
| 607 | + case IO_MANAGER_WIN32_LEGACY:
|
|
| 608 | + abandonRequestWait();
|
|
| 576 | 609 | break;
|
| 610 | +#endif
|
|
| 611 | + default:
|
|
| 612 | + barf("wakeupIOManager not implemented");
|
|
| 577 | 613 | }
|
| 578 | 614 | }
|
| 579 | 615 | |
| ... | ... | @@ -782,7 +818,9 @@ void awaitCompletedTimeoutsOrIO(CapIOManager *iomgr) |
| 782 | 818 | default:
|
| 783 | 819 | barf("pollCompletedTimeoutsOrIO not implemented");
|
| 784 | 820 | }
|
| 785 | - ASSERT(!emptyRunQueue(iomgr->cap) || getSchedState() != SCHED_RUNNING);
|
|
| 821 | + // FIXME: the post condition is now more complicated. Await can now simply
|
|
| 822 | + // be interrupted by wakeupIOManager.
|
|
| 823 | + // ASSERT(!emptyRunQueue(iomgr->cap) || getSchedState() != SCHED_RUNNING);
|
|
| 786 | 824 | }
|
| 787 | 825 | |
| 788 | 826 |
| ... | ... | @@ -242,6 +242,28 @@ CapIOManager *allocCapabilityIOManager(Capability *cap); |
| 242 | 242 | */
|
| 243 | 243 | void initCapabilityIOManager(CapIOManager *iomgr);
|
| 244 | 244 | |
| 245 | +/* When shutting down a capability, or after forkProcess, free the resources
|
|
| 246 | + * held by a CapIOManager to put it back into a state in which either it can be
|
|
| 247 | + * re-initialised using initCapabilityIOManager, or the whole structure freed.
|
|
| 248 | + *
|
|
| 249 | + * Note that this does not free the CapIOManager structure itself, just the
|
|
| 250 | + * contents.
|
|
| 251 | + *
|
|
| 252 | + * This is used during capability shutdown, during RTS shutdown. It is not used
|
|
| 253 | + * when reducing the number of capabilities. Capabilities are disabled rather
|
|
| 254 | + * than freed entirely: the I/O manager keeps running but threads that become
|
|
| 255 | + * runnable are migrated away.
|
|
| 256 | + *
|
|
| 257 | + * It is also used after forkProcess.
|
|
| 258 | + */
|
|
| 259 | +void freeCapabilityIOManager(CapIOManager *iomgr);
|
|
| 260 | + |
|
| 261 | +/* CapIOManager life cycle:
|
|
| 262 | + *
|
|
| 263 | + * alloc -> init -> free -> free struct
|
|
| 264 | + * ^ |
|
|
| 265 | + * +--------+
|
|
| 266 | + */
|
|
| 245 | 267 | |
| 246 | 268 | /* Init hook: called from hs_init_ghc, very late in the startup after almost
|
| 247 | 269 | * everything else is done.
|
| ... | ... | @@ -283,19 +305,13 @@ void stopIOManager(void); |
| 283 | 305 | void exitIOManager(bool wait_threads);
|
| 284 | 306 | |
| 285 | 307 | |
| 286 | -/* Wakeup hook: called from the scheduler's wakeUpRts (currently only in
|
|
| 287 | - * threaded mode).
|
|
| 308 | +/* Wakeup hook: called from the scheduler's wakeUpRts().
|
|
| 288 | 309 | *
|
| 289 | 310 | * The I/O manager can be blocked waiting on I/O or timers. Sometimes there are
|
| 290 | 311 | * other external events where we need to wake up the I/O manager and return
|
| 291 | - * to the schedulr.
|
|
| 292 | - *
|
|
| 293 | - * At the moment, all the non-threaded I/O managers will do this automagically
|
|
| 294 | - * since a signal will interrupt any waiting system calls, so at the moment
|
|
| 295 | - * the implementation for the non-threaded I/O managers does nothing.
|
|
| 312 | + * to the scheduler.
|
|
| 296 | 313 | *
|
| 297 | - * For the I/O managers in threaded mode, this arranges to unblock the I/O
|
|
| 298 | - * manager if it waa blocked waiting.
|
|
| 314 | + * This arranges to unblock the I/O manager if it was blocked waiting.
|
|
| 299 | 315 | */
|
| 300 | 316 | void wakeupIOManager(void);
|
| 301 | 317 |
| ... | ... | @@ -46,6 +46,11 @@ struct _CapIOManager { |
| 46 | 46 | StgTSO *sleeping_queue;
|
| 47 | 47 | #endif
|
| 48 | 48 | |
| 49 | +#if defined(IOMGR_ENABLED_SELECT) || defined(IOMGR_ENABLED_POLL)
|
|
| 50 | + /* FDs for waking up the I/O manager when it is blocked waiting */
|
|
| 51 | + int wakeup_fd_r, wakeup_fd_w;
|
|
| 52 | +#endif
|
|
| 53 | + |
|
| 49 | 54 | #if defined(IOMGR_ENABLED_POLL)
|
| 50 | 55 | /* AIOP and timeout collections shared by several I/O manager impls */
|
| 51 | 56 | ClosureTable aiop_table;
|
| ... | ... | @@ -53,8 +58,11 @@ struct _CapIOManager { |
| 53 | 58 | #endif
|
| 54 | 59 | |
| 55 | 60 | #if defined(IOMGR_ENABLED_POLL)
|
| 56 | - /* Auxiliary table with size and indexes matching the aiop_table */
|
|
| 57 | - struct pollfd *aiop_poll_table;
|
|
| 61 | + /* Auxiliary table with size and indexes matching the aiop_table. This is
|
|
| 62 | + * aliased to the tail of the full poll table, which has a head entry for
|
|
| 63 | + * the wakeup_fd_r above, so we can also poll that fd.
|
|
| 64 | + */
|
|
| 65 | + struct pollfd *aiop_poll_table, *full_poll_table;
|
|
| 58 | 66 | #endif
|
| 59 | 67 | |
| 60 | 68 | #if defined(IOMGR_ENABLED_WIN32_LEGACY)
|
| ... | ... | @@ -2165,6 +2165,15 @@ forkProcess(HsStablePtr *entry |
| 2165 | 2165 | // exist.
|
| 2166 | 2166 | truncateRunQueue(cap);
|
| 2167 | 2167 | |
| 2168 | + // Reset and re-initialise the capability's I/O manager,
|
|
| 2169 | + // to get the I/O manager ready again.
|
|
| 2170 | + //
|
|
| 2171 | + // Any threads waiting on I/O or timers should have been
|
|
| 2172 | + // removed from I/O manager queues by deleteThread_ above.
|
|
| 2173 | + // TODO: but we could assert that here.
|
|
| 2174 | + freeCapabilityIOManager(cap->iomgr);
|
|
| 2175 | + initCapabilityIOManager(cap->iomgr);
|
|
| 2176 | + |
|
| 2168 | 2177 | // Any suspended C-calling Tasks are no more, their OS threads
|
| 2169 | 2178 | // don't exist now:
|
| 2170 | 2179 | cap->suspended_ccalls = NULL;
|
| ... | ... | @@ -2309,6 +2318,10 @@ setNumCapabilities (uint32_t new_n_capabilities USED_IF_THREADS) |
| 2309 | 2318 | // the capability; we don't have to worry about GC data
|
| 2310 | 2319 | // structures, the nursery, etc.
|
| 2311 | 2320 | //
|
| 2321 | + // This approach also handles threads blocked on I/O. Such threads
|
|
| 2322 | + // remain blocked, and when I/O completes and threads become runnable
|
|
| 2323 | + // then they are migrated away.
|
|
| 2324 | + //
|
|
| 2312 | 2325 | for (n = new_n_capabilities; n < enabled_capabilities; n++) {
|
| 2313 | 2326 | getCapability(n)->disabled = true;
|
| 2314 | 2327 | traceCapDisable(getCapability(n));
|
| 1 | +/* -----------------------------------------------------------------------------
|
|
| 2 | + *
|
|
| 3 | + * (c) The GHC Team 2025
|
|
| 4 | + *
|
|
| 5 | + * Utilities for a simple fd-based cross-thread wakeup mechanism.
|
|
| 6 | + *
|
|
| 7 | + * This is used in I/O managers, to provide a mechanism to wake them when they
|
|
| 8 | + * are blocked waiting on fds and timeouts. The mechanism works by including
|
|
| 9 | + * the read end fd into the set of fds the I/O manager waits on, and when a
|
|
| 10 | + * wake up is needed, the write end fd is used.
|
|
| 11 | + *
|
|
| 12 | + * This is implemented using either eventfd() or pipe().
|
|
| 13 | + *
|
|
| 14 | + * Linux 2.6.22+ and FreeBSD 13+ support eventfd. It is a single fd with a
|
|
| 15 | + * 64bit counter. It uses less resources than a pipe, and is probably a tad
|
|
| 16 | + * faster. Using write() adds to the counter, while read() reads and resets
|
|
| 17 | + * it. This gives us event combining.
|
|
| 18 | + *
|
|
| 19 | + * Otherwise we use a classic unix pipe.
|
|
| 20 | + *
|
|
| 21 | + * -------------------------------------------------------------------------*/
|
|
| 22 | + |
|
| 23 | +#include "rts/PosixSource.h"
|
|
| 24 | +#include "Rts.h"
|
|
| 25 | + |
|
| 26 | +#include "FdWakeup.h"
|
|
| 27 | + |
|
| 28 | +#include <fcntl.h>
|
|
| 29 | +#include <unistd.h>
|
|
| 30 | + |
|
| 31 | +#ifdef HAVE_SYS_EVENTFD_H
|
|
| 32 | +#include <sys/eventfd.h>
|
|
| 33 | +#endif
|
|
| 34 | + |
|
| 35 | +#if !defined(HAVE_EVENTFD) \
|
|
| 36 | + || (defined(HAVE_EVENTFD) && !(defined(EFD_CLOEXEC) && defined(EFD_NONBLOCK)))
|
|
| 37 | +static void fcntl_CLOEXEC_NONBLOCK(int fd)
|
|
| 38 | +{
|
|
| 39 | + int res1 = fcntl(fd, F_SETFD, FD_CLOEXEC);
|
|
| 40 | + int res2 = fcntl(fd, F_SETFL, O_NONBLOCK);
|
|
| 41 | + if (RTS_UNLIKELY(res1 < 0 || res2 < 0)) {
|
|
| 42 | + sysErrorBelch("newFdWakeup fcntl()");
|
|
| 43 | + stg_exit(EXIT_FAILURE);
|
|
| 44 | + }
|
|
| 45 | +}
|
|
| 46 | +#endif
|
|
| 47 | + |
|
| 48 | +void newFdWakeup(int *wakeup_fd_r, int *wakeup_fd_w)
|
|
| 49 | +{
|
|
| 50 | +#if defined(HAVE_EVENTFD)
|
|
| 51 | + int wakeup_fd;
|
|
| 52 | +#if defined(EFD_CLOEXEC) && defined(EFD_NONBLOCK)
|
|
| 53 | + wakeup_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
|
|
| 54 | +#else
|
|
| 55 | + wakeup_fd = eventfd(0, 0);
|
|
| 56 | + if (wakeup_fd >= 0) fcntl_CLOEXEC_NONBLOCK(wakeup_fd);
|
|
| 57 | +#endif
|
|
| 58 | + if (RTS_UNLIKELY(wakeup_fd < 0)) {
|
|
| 59 | + sysErrorBelch("newFdWakeup eventfd()");
|
|
| 60 | + stg_exit(EXIT_FAILURE);
|
|
| 61 | + }
|
|
| 62 | + /* eventfd uses the same fd for each end */
|
|
| 63 | + *wakeup_fd_r = wakeup_fd;
|
|
| 64 | + *wakeup_fd_w = wakeup_fd;
|
|
| 65 | +#else
|
|
| 66 | + int pipefd[2];
|
|
| 67 | + int res;
|
|
| 68 | + res = pipe(pipefd);
|
|
| 69 | + if (RTS_UNLIKELY(res < 0)) {
|
|
| 70 | + sysErrorBelch("newFdWakeup pipe");
|
|
| 71 | + stg_exit(EXIT_FAILURE);
|
|
| 72 | + }
|
|
| 73 | + fcntl_CLOEXEC_NONBLOCK(pipefd[0]);
|
|
| 74 | + fcntl_CLOEXEC_NONBLOCK(pipefd[1]);
|
|
| 75 | + *wakeup_fd_r = pipefd[0]; /* read end */
|
|
| 76 | + *wakeup_fd_w = pipefd[1]; /* write end */
|
|
| 77 | +#endif
|
|
| 78 | +}
|
|
| 79 | + |
|
| 80 | +void closeFdWakeup(int wakeup_fd_r, int wakeup_fd_w)
|
|
| 81 | +{
|
|
| 82 | +#if defined(HAVE_EVENTFD)
|
|
| 83 | + ASSERT(wakeup_fd_r == wakeup_fd_w);
|
|
| 84 | + close(wakeup_fd_r);
|
|
| 85 | +#else
|
|
| 86 | + ASSERT(wakeup_fd_r != wakeup_fd_w);
|
|
| 87 | + close(wakeup_fd_r);
|
|
| 88 | + close(wakeup_fd_w);
|
|
| 89 | +#endif
|
|
| 90 | +}
|
|
| 91 | + |
|
| 92 | +void sendFdWakeup(int wakeup_fd_w)
|
|
| 93 | +{
|
|
| 94 | + int res;
|
|
| 95 | +#if defined(HAVE_EVENTFD)
|
|
| 96 | + uint64_t val = 1;
|
|
| 97 | + res = write(wakeup_fd_w, &val, 8);
|
|
| 98 | +#else
|
|
| 99 | + unsigned char buf = 1;
|
|
| 100 | + res = write(wakeup_fd_w, &buf, 1);
|
|
| 101 | +#endif
|
|
| 102 | + if (RTS_UNLIKELY(res < 0)) {
|
|
| 103 | + /* Unlikely the pipe buffer will fill, but it would not be an error. */
|
|
| 104 | + if (errno == EAGAIN) return;
|
|
| 105 | + sysErrorBelch("sendFdWakeup write");
|
|
| 106 | + stg_exit(EXIT_FAILURE);
|
|
| 107 | + }
|
|
| 108 | +}
|
|
| 109 | + |
|
| 110 | +void collectFdWakeup(int wakeup_fd_r)
|
|
| 111 | +{
|
|
| 112 | + int res;
|
|
| 113 | +#if defined(HAVE_EVENTFD)
|
|
| 114 | + uint64_t buf;
|
|
| 115 | + /* eventfd combines events into one counter, so a single read is enough */
|
|
| 116 | + res = read(wakeup_fd_r, &buf, 8);
|
|
| 117 | +#else
|
|
| 118 | + /* Drain the pipe buffer. Multiple wakeup notifications could
|
|
| 119 | + * have been sent before we have a chance to collect them.
|
|
| 120 | + */
|
|
| 121 | + uint64_t buf;
|
|
| 122 | + do {
|
|
| 123 | + res = read(wakeup_fd_r, &buf, 8);
|
|
| 124 | + } while (res == 8);
|
|
| 125 | +#endif
|
|
| 126 | + if (RTS_UNLIKELY(res < 0)) {
|
|
| 127 | + /* After the first pipe read, it could block */
|
|
| 128 | + if (errno == EAGAIN) return;
|
|
| 129 | + sysErrorBelch("collectFdWakeup read");
|
|
| 130 | + stg_exit(EXIT_FAILURE);
|
|
| 131 | + }
|
|
| 132 | +} |
| 1 | +/* -----------------------------------------------------------------------------
|
|
| 2 | + *
|
|
| 3 | + * (c) The GHC Team 2025
|
|
| 4 | + *
|
|
| 5 | + * Utilities for a simple fd-based cross-thread wakeup mechanism.
|
|
| 6 | + *
|
|
| 7 | + * This is used in I/O managers, to provide a mechanism to wake them when they
|
|
| 8 | + * are blocked waiting on fds and timeouts. The mechanism works by including
|
|
| 9 | + * the read end fd into the set of fds the I/O manager waits on, and when a
|
|
| 10 | + * wake up is needed, the write end fd is used.
|
|
| 11 | + *
|
|
| 12 | + * Prototypes for functions in FdWakeup.c
|
|
| 13 | + *
|
|
| 14 | + * -------------------------------------------------------------------------*/
|
|
| 15 | + |
|
| 16 | +#pragma once
|
|
| 17 | + |
|
| 18 | +#include "BeginPrivate.h"
|
|
| 19 | + |
|
| 20 | +void newFdWakeup(int *fd_r, int *fd_w);
|
|
| 21 | +void closeFdWakeup(int fd_r, int fd_w);
|
|
| 22 | + |
|
| 23 | +void sendFdWakeup(int fd_w);
|
|
| 24 | +void collectFdWakeup(int fd_r);
|
|
| 25 | + |
|
| 26 | +#include "EndPrivate.h"
|
|
| 27 | + |
| ... | ... | @@ -41,6 +41,7 @@ |
| 41 | 41 | |
| 42 | 42 | #include "IOManagerInternals.h"
|
| 43 | 43 | #include "Timeout.h"
|
| 44 | +#include "FdWakeup.h"
|
|
| 44 | 45 | |
| 45 | 46 | /******************************************************************************
|
| 46 | 47 | |
| ... | ... | @@ -107,8 +108,9 @@ timeout (if any) as the poll() timeout parameter. |
| 107 | 108 | The CapIOManager structure for this I/O manager contains:
|
| 108 | 109 | |
| 109 | 110 | ClosureTable aiop_table;
|
| 110 | - struct pollfd *aiop_poll_table;
|
|
| 111 | + struct pollfd *aiop_poll_table, *full_poll_table;
|
|
| 111 | 112 | StgTimeoutQueue *timeout_queue;
|
| 113 | + int wakeup_fd_r, wakeup_fd_w;
|
|
| 112 | 114 | |
| 113 | 115 | We also support the Linux-specific ppoll API which supports higher resolution
|
| 114 | 116 | time delays -- nanoseconds rather than milliseconds as in classic poll(). It
|
| ... | ... | @@ -117,6 +119,15 @@ also allows the signal mask to be adjusted, but we do not make use of this. |
| 117 | 119 | int ppoll(struct pollfd *fds, nfds_t nfds,
|
| 118 | 120 | const struct timespec *tmo_p, const sigset_t *sigmask);
|
| 119 | 121 | |
| 122 | +We have both aiop_poll_table and full_poll_table. This is to cope with needing
|
|
| 123 | +to wait on the special extra file descriptor wakeup_fd_r. This fd is used to
|
|
| 124 | +support waking the I/O manager when we are blocked in a poll call. This
|
|
| 125 | +requires waiting on an extra fd that has no corresponding entry in the
|
|
| 126 | +aiop_table. To manage this quirk, we alias the aiop_poll_table to be the tail
|
|
| 127 | +of the full_poll_table and have the first entry of the full_poll_table be the
|
|
| 128 | +wakeup_fd_r. This means the aiop_poll_table indicies match up exactly with the
|
|
| 129 | +aiop_table, but still allows the full_poll_table to have an extra entry.
|
|
| 130 | + |
|
| 120 | 131 | ******************************************************************************/
|
| 121 | 132 | |
| 122 | 133 | /* Forward declarations */
|
| ... | ... | @@ -129,8 +140,31 @@ static void reportPollError(int res, nfds_t nfds) STG_NORETURN; |
| 129 | 140 | void initCapabilityIOManagerPoll(CapIOManager *iomgr)
|
| 130 | 141 | {
|
| 131 | 142 | initClosureTable(&iomgr->aiop_table, ClosureTableCompact);
|
| 132 | - iomgr->aiop_poll_table = NULL;
|
|
| 133 | 143 | iomgr->timeout_queue = emptyTimeoutQueue();
|
| 144 | + |
|
| 145 | + newFdWakeup(&iomgr->wakeup_fd_r, &iomgr->wakeup_fd_w);
|
|
| 146 | + |
|
| 147 | + iomgr->full_poll_table = stgMallocBytes(sizeof(struct pollfd) /* size 1 */,
|
|
| 148 | + "initCapabilityIOManagerPoll");
|
|
| 149 | + iomgr->full_poll_table[0] = (struct pollfd) {
|
|
| 150 | + .fd = iomgr->wakeup_fd_r,
|
|
| 151 | + .events = POLLIN,
|
|
| 152 | + .revents = 0
|
|
| 153 | + };
|
|
| 154 | + iomgr->aiop_poll_table = iomgr->full_poll_table+1; /* hence empty */
|
|
| 155 | +}
|
|
| 156 | + |
|
| 157 | + |
|
| 158 | +void freeCapabilityIOManagerPoll(CapIOManager *iomgr)
|
|
| 159 | +{
|
|
| 160 | + stgFree(iomgr->full_poll_table);
|
|
| 161 | + closeFdWakeup(iomgr->wakeup_fd_r, iomgr->wakeup_fd_w);
|
|
| 162 | +}
|
|
| 163 | + |
|
| 164 | + |
|
| 165 | +void wakeupIOManagerPoll(CapIOManager *iomgr)
|
|
| 166 | +{
|
|
| 167 | + sendFdWakeup(iomgr->wakeup_fd_w);
|
|
| 134 | 168 | }
|
| 135 | 169 | |
| 136 | 170 | |
| ... | ... | @@ -275,7 +309,7 @@ static void notifyIOCompletion(CapIOManager *iomgr, StgAsyncIOOp *aiop) |
| 275 | 309 | }
|
| 276 | 310 | |
| 277 | 311 | |
| 278 | -static void processIOCompletions(CapIOManager *iomgr, int ncompletions)
|
|
| 312 | +static bool processIOCompletions(CapIOManager *iomgr, int ncompletions)
|
|
| 279 | 313 | {
|
| 280 | 314 | /* The scheme we use with poll is that we have a dense poll table, and a
|
| 281 | 315 | * corresponding table that maps to the closure table index. The poll
|
| ... | ... | @@ -285,6 +319,19 @@ static void processIOCompletions(CapIOManager *iomgr, int ncompletions) |
| 285 | 319 | */
|
| 286 | 320 | debugTrace(DEBUG_iomanager, "processIOCompletions(ncompletions = %d)",
|
| 287 | 321 | ncompletions);
|
| 322 | + |
|
| 323 | + bool wakeup;
|
|
| 324 | + /* If the wakeup_fd_r is ready, collect it */
|
|
| 325 | + if (iomgr->full_poll_table[0].revents) {
|
|
| 326 | + ASSERT(iomgr->full_poll_table[0].fd == iomgr->wakeup_fd_r);
|
|
| 327 | + collectFdWakeup(iomgr->wakeup_fd_r);
|
|
| 328 | + ncompletions--;
|
|
| 329 | + wakeup = true;
|
|
| 330 | + debugTrace(DEBUG_iomanager, "Received wakeup in poll I/O manager.");
|
|
| 331 | + } else {
|
|
| 332 | + wakeup = false;
|
|
| 333 | + }
|
|
| 334 | + |
|
| 288 | 335 | struct pollfd *aiop_poll_table = iomgr->aiop_poll_table;
|
| 289 | 336 | int n = ncompletions;
|
| 290 | 337 | int i = 0;
|
| ... | ... | @@ -337,11 +384,14 @@ static void processIOCompletions(CapIOManager *iomgr, int ncompletions) |
| 337 | 384 | i++;
|
| 338 | 385 | }
|
| 339 | 386 | }
|
| 387 | + return wakeup;
|
|
| 340 | 388 | }
|
| 341 | 389 | |
| 342 | 390 | |
| 343 | 391 | void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
|
| 344 | 392 | {
|
| 393 | + ASSERT(iomgr->aiop_poll_table == iomgr->full_poll_table+1);
|
|
| 394 | + |
|
| 345 | 395 | if (!isEmptyTimeoutQueue(iomgr->timeout_queue)) {
|
| 346 | 396 | Time now = getProcessElapsedTime();
|
| 347 | 397 | processTimeoutCompletions(iomgr, now);
|
| ... | ... | @@ -349,20 +399,20 @@ void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr) |
| 349 | 399 | |
| 350 | 400 | if (!isEmptyClosureTable(&iomgr->aiop_table)) {
|
| 351 | 401 | |
| 352 | - nfds_t nfds = sizeClosureTable(&iomgr->aiop_table);
|
|
| 402 | + nfds_t nfds = sizeClosureTable(&iomgr->aiop_table) + 1;
|
|
| 353 | 403 | |
| 354 | 404 | /* Poll for I/O readiness, without waiting. */
|
| 355 | 405 | #if defined(HAVE_DECL_PPOLL) && HAVE_DECL_PPOLL == 1
|
| 356 | 406 | /* We could use poll here, since we use no timeout, but for
|
| 357 | 407 | consistency we use the same syscall as at the other call site. */
|
| 358 | 408 | struct timespec tv = (struct timespec) { .tv_sec = 0, .tv_nsec = 0 };
|
| 359 | - int res = ppoll(iomgr->aiop_poll_table, nfds, &tv, NULL);
|
|
| 409 | + int res = ppoll(iomgr->full_poll_table, nfds, &tv, NULL);
|
|
| 360 | 410 | |
| 361 | 411 | debugTrace(DEBUG_iomanager,
|
| 362 | 412 | "ppoll(nfds = %d, timeout.sec = 0, timeout.nsec = 0) = %d",
|
| 363 | 413 | nfds, res);
|
| 364 | 414 | #else
|
| 365 | - int res = poll(iomgr->aiop_poll_table, nfds, 0);
|
|
| 415 | + int res = poll(iomgr->full_poll_table, nfds, 0);
|
|
| 366 | 416 | |
| 367 | 417 | debugTrace(DEBUG_iomanager,
|
| 368 | 418 | "poll(nfds = %d, timeout_ms = 0) = %d",
|
| ... | ... | @@ -390,6 +440,10 @@ void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr) |
| 390 | 440 | |
| 391 | 441 | void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
|
| 392 | 442 | {
|
| 443 | + bool wakeup = false; /* got woken up via wakeupIOManager */
|
|
| 444 | + |
|
| 445 | + ASSERT(iomgr->aiop_poll_table == iomgr->full_poll_table+1);
|
|
| 446 | + |
|
| 393 | 447 | /* Loop until we've woken up some threads. This loop is needed because the
|
| 394 | 448 | * poll() timing isn't accurate, we sometimes sleep for a while but not
|
| 395 | 449 | * long enough to wake up a thread in a threadDelay. Or we may need to
|
| ... | ... | @@ -422,9 +476,9 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr) |
| 422 | 476 | #endif
|
| 423 | 477 | |
| 424 | 478 | /* Check for I/O readiness, possibly waiting. */
|
| 425 | - nfds_t nfds = sizeClosureTable(&iomgr->aiop_table);
|
|
| 479 | + nfds_t nfds = sizeClosureTable(&iomgr->aiop_table) + 1;
|
|
| 426 | 480 | #if defined(HAVE_DECL_PPOLL) && HAVE_DECL_PPOLL == 1
|
| 427 | - int res = ppoll(iomgr->aiop_poll_table, nfds, timeout_ns, NULL);
|
|
| 481 | + int res = ppoll(iomgr->full_poll_table, nfds, timeout_ns, NULL);
|
|
| 428 | 482 | |
| 429 | 483 | debugTrace(DEBUG_iomanager,
|
| 430 | 484 | "ppoll(nfds = %d, timeout.sec = %d, timeout.nsec = %d) = %d",
|
| ... | ... | @@ -432,7 +486,7 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr) |
| 432 | 486 | timeout_ns == NULL ? 0 : timeout_ns->tv_nsec,
|
| 433 | 487 | res);
|
| 434 | 488 | #else
|
| 435 | - int res = poll(iomgr->aiop_poll_table, nfds, timeout_ms);
|
|
| 489 | + int res = poll(iomgr->full_poll_table, nfds, timeout_ms);
|
|
| 436 | 490 | |
| 437 | 491 | debugTrace(DEBUG_iomanager,
|
| 438 | 492 | "poll(nfds = %d, timeout_ms = %d) = %d",
|
| ... | ... | @@ -454,7 +508,7 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr) |
| 454 | 508 | } else if (res > 0) {
|
| 455 | 509 | int ncompletions = res;
|
| 456 | 510 | ASSERT(ncompletions <= (int)nfds);
|
| 457 | - processIOCompletions(iomgr, ncompletions);
|
|
| 511 | + wakeup = processIOCompletions(iomgr, ncompletions);
|
|
| 458 | 512 | |
| 459 | 513 | } else if (errno == EINTR) {
|
| 460 | 514 | /* We got interrupted by a signal. In the non-threaded RTS, if the
|
| ... | ... | @@ -479,6 +533,7 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr) |
| 479 | 533 | }
|
| 480 | 534 | |
| 481 | 535 | } while (emptyRunQueue(iomgr->cap)
|
| 536 | + && !wakeup
|
|
| 482 | 537 | && (getSchedState() == SCHED_RUNNING));
|
| 483 | 538 | }
|
| 484 | 539 | |
| ... | ... | @@ -508,13 +563,17 @@ static bool enlargeTables(CapIOManager *iomgr) |
| 508 | 563 | bool ok = enlargeClosureTable(iomgr->cap, &iomgr->aiop_table, newcapacity);
|
| 509 | 564 | if (RTS_UNLIKELY(!ok)) return false;
|
| 510 | 565 | |
| 511 | - /* Update the auxiliary aiop_poll_table to match */
|
|
| 512 | - struct pollfd *aiop_poll_table;
|
|
| 513 | - aiop_poll_table = stgReallocBytes(iomgr->aiop_poll_table,
|
|
| 514 | - sizeof(struct pollfd) * newcapacity,
|
|
| 515 | - "Poll.c: enlargeTables");
|
|
| 516 | - iomgr->aiop_poll_table = aiop_poll_table;
|
|
| 566 | + /* Update the auxiliary aiop_poll_table to match. The full_poll_table is
|
|
| 567 | + * one bigger than the aiop_poll_table, since it has an extra entry at the
|
|
| 568 | + * front for wakeup_fd_r, with no corresponding aiop. */
|
|
| 569 | + iomgr->full_poll_table =
|
|
| 570 | + stgReallocBytes(iomgr->full_poll_table,
|
|
| 571 | + sizeof(struct pollfd) * (newcapacity+1),
|
|
| 572 | + "Poll.c: enlargeTables");
|
|
| 573 | + iomgr->aiop_poll_table = iomgr->full_poll_table+1;
|
|
| 574 | + |
|
| 517 | 575 | /* Initialise the new part of the aiop_poll_table */
|
| 576 | + struct pollfd *aiop_poll_table = iomgr->aiop_poll_table;
|
|
| 518 | 577 | for (int i = oldcapacity; i < newcapacity; i++) {
|
| 519 | 578 | aiop_poll_table[i] = (struct pollfd) {
|
| 520 | 579 | .fd = -1,
|
| ... | ... | @@ -17,6 +17,8 @@ |
| 17 | 17 | #if defined(IOMGR_ENABLED_POLL)
|
| 18 | 18 | |
| 19 | 19 | void initCapabilityIOManagerPoll(CapIOManager *iomgr);
|
| 20 | +void freeCapabilityIOManagerPoll(CapIOManager *iomgr);
|
|
| 21 | +void wakeupIOManagerPoll(CapIOManager *iomgr);
|
|
| 20 | 22 | |
| 21 | 23 | /* Synchronous I/O and timer operations */
|
| 22 | 24 | bool syncIOWaitReadyPoll(CapIOManager *iomgr, StgTSO *tso,
|
| ... | ... | @@ -22,6 +22,7 @@ |
| 22 | 22 | #include "IOManagerInternals.h"
|
| 23 | 23 | #include "Stats.h"
|
| 24 | 24 | #include "GetTime.h"
|
| 25 | +#include "FdWakeup.h"
|
|
| 25 | 26 | |
| 26 | 27 | # if defined(HAVE_SYS_SELECT_H)
|
| 27 | 28 | # include <sys/select.h>
|
| ... | ... | @@ -54,6 +55,25 @@ |
| 54 | 55 | #define TimeToLowResTimeRoundUp(t) (t)
|
| 55 | 56 | #endif
|
| 56 | 57 | |
| 58 | +void initCapabilityIOManagerSelect(CapIOManager *iomgr)
|
|
| 59 | +{
|
|
| 60 | + iomgr->blocked_queue_hd = END_TSO_QUEUE;
|
|
| 61 | + iomgr->blocked_queue_tl = END_TSO_QUEUE;
|
|
| 62 | + iomgr->sleeping_queue = END_TSO_QUEUE;
|
|
| 63 | + |
|
| 64 | + newFdWakeup(&iomgr->wakeup_fd_r, &iomgr->wakeup_fd_w);
|
|
| 65 | +}
|
|
| 66 | + |
|
| 67 | +void freeCapabilityIOManagerSelect(CapIOManager *iomgr)
|
|
| 68 | +{
|
|
| 69 | + closeFdWakeup(iomgr->wakeup_fd_r, iomgr->wakeup_fd_w);
|
|
| 70 | +}
|
|
| 71 | + |
|
| 72 | +void wakeupIOManagerSelect(CapIOManager *iomgr)
|
|
| 73 | +{
|
|
| 74 | + sendFdWakeup(iomgr->wakeup_fd_w);
|
|
| 75 | +}
|
|
| 76 | + |
|
| 57 | 77 | /*
|
| 58 | 78 | * Return the time since the program started, in LowResTime,
|
| 59 | 79 | * rounded down.
|
| ... | ... | @@ -225,6 +245,7 @@ awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait) |
| 225 | 245 | bool seen_bad_fd = false;
|
| 226 | 246 | struct timeval tv, *ptv;
|
| 227 | 247 | LowResTime now;
|
| 248 | + bool wakeup = false; /* got woken up via wakeupIOManager */
|
|
| 228 | 249 | |
| 229 | 250 | IF_DEBUG(scheduler,
|
| 230 | 251 | debugBelch("scheduler: checking for threads blocked on I/O");
|
| ... | ... | @@ -252,6 +273,13 @@ awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait) |
| 252 | 273 | FD_ZERO(&rfd);
|
| 253 | 274 | FD_ZERO(&wfd);
|
| 254 | 275 | |
| 276 | + /* We're always interested in our wakeup fd */
|
|
| 277 | + {
|
|
| 278 | + int fd = iomgr->wakeup_fd_r;
|
|
| 279 | + maxfd = (fd > maxfd) ? fd : maxfd;
|
|
| 280 | + FD_SET(fd, &rfd);
|
|
| 281 | + }
|
|
| 282 | + |
|
| 255 | 283 | for(tso = iomgr->blocked_queue_hd;
|
| 256 | 284 | tso != END_TSO_QUEUE;
|
| 257 | 285 | tso = next) {
|
| ... | ... | @@ -376,6 +404,13 @@ awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait) |
| 376 | 404 | }
|
| 377 | 405 | }
|
| 378 | 406 | |
| 407 | + /* If the wakeup_fd_r is ready, collect it */
|
|
| 408 | + if (FD_ISSET(iomgr->wakeup_fd_r, &rfd)) {
|
|
| 409 | + collectFdWakeup(iomgr->wakeup_fd_r);
|
|
| 410 | + wakeup = true;
|
|
| 411 | + debugTrace(DEBUG_iomanager, "Received wakeup in select I/O manager.");
|
|
| 412 | + }
|
|
| 413 | + |
|
| 379 | 414 | /* Step through the waiting queue, unblocking every thread that now has
|
| 380 | 415 | * a file descriptor in a ready state.
|
| 381 | 416 | */
|
| ... | ... | @@ -458,7 +493,8 @@ awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait) |
| 458 | 493 | }
|
| 459 | 494 | |
| 460 | 495 | } while (wait && getSchedState() == SCHED_RUNNING
|
| 461 | - && emptyRunQueue(iomgr->cap));
|
|
| 496 | + && emptyRunQueue(iomgr->cap)
|
|
| 497 | + && !wakeup);
|
|
| 462 | 498 | }
|
| 463 | 499 | |
| 464 | 500 | #endif /* IOMGR_ENABLED_SELECT */ |
| ... | ... | @@ -15,6 +15,10 @@ typedef StgWord LowResTime; |
| 15 | 15 | |
| 16 | 16 | LowResTime getDelayTarget (HsInt us);
|
| 17 | 17 | |
| 18 | +void initCapabilityIOManagerSelect(CapIOManager *iomgr);
|
|
| 19 | +void freeCapabilityIOManagerSelect(CapIOManager *iomgr);
|
|
| 20 | +void wakeupIOManagerSelect(CapIOManager *iomgr);
|
|
| 21 | + |
|
| 18 | 22 | void awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait);
|
| 19 | 23 | |
| 20 | 24 | #include "EndPrivate.h"
|
| ... | ... | @@ -569,6 +569,7 @@ library |
| 569 | 569 | wasm/OSThreads.c
|
| 570 | 570 | wasm/JSFFI.c
|
| 571 | 571 | wasm/JSFFIGlobals.c
|
| 572 | + posix/FdWakeup.c
|
|
| 572 | 573 | posix/Select.c
|
| 573 | 574 | posix/Poll.c
|
| 574 | 575 | posix/Timeout.c
|
| ... | ... | @@ -581,6 +582,7 @@ library |
| 581 | 582 | posix/Ticker.c
|
| 582 | 583 | posix/OSMem.c
|
| 583 | 584 | posix/OSThreads.c
|
| 585 | + posix/FdWakeup.c
|
|
| 584 | 586 | posix/MIO.c
|
| 585 | 587 | posix/Poll.c
|
| 586 | 588 | posix/Select.c
|