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:

12 changed files:

Changes:

  • rts/Capability.c
    ... ... @@ -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) {
    

  • rts/IOManager.c
    ... ... @@ -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
     
    

  • rts/IOManager.h
    ... ... @@ -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
     
    

  • rts/IOManagerInternals.h
    ... ... @@ -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)
    

  • rts/Schedule.c
    ... ... @@ -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));
    

  • rts/posix/FdWakeup.c
    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
    +}

  • rts/posix/FdWakeup.h
    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
    +

  • rts/posix/Poll.c
    ... ... @@ -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,
    

  • rts/posix/Poll.h
    ... ... @@ -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,
    

  • rts/posix/Select.c
    ... ... @@ -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 */

  • rts/posix/Select.h
    ... ... @@ -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"
    

  • rts/rts.cabal
    ... ... @@ -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