Duncan Coutts pushed to branch wip/dcoutts/io-manager-tidy at Glasgow Haskell Compiler / GHC

Commits:

5 changed files:

Changes:

  • rts/IOManager.c
    ... ... @@ -745,7 +745,7 @@ bool awaitCompletedTimeoutsOrIO(CapIOManager *iomgr)
    745 745
    746 746 
     #if defined(IOMGR_ENABLED_POLL)
    747 747 
             case IO_MANAGER_POLL:
    748 
    -          awaitCompletedTimeoutsOrIOPoll(iomgr);
    748 
    +          completed = awaitCompletedTimeoutsOrIOPoll(iomgr);
    749 749 
               break;
    750 750 
     #endif
    751 751
    ... ... @@ -784,6 +784,12 @@ void interruptIOManager(CapIOManager *iomgr)
    784 784 
                 break;
    785 785 
     #endif
    786 786
    787 
    +#if defined(IOMGR_ENABLED_POLL)
    788 
    +        case IO_MANAGER_POLL:
    789 
    +            interruptIOManagerPoll(iomgr);
    790 
    +            break;
    791 
    +#endif
    792 
    +
    787 793 
             default:
    788 794 
                 break;
    789 795 
         }

  • rts/IOManagerInternals.h
    ... ... @@ -46,10 +46,12 @@ struct _CapIOManager {
    46 46 
         StgTSO *sleeping_queue;
    47 47 
     #endif
    48 48
    49 
    -#if defined(IOMGR_ENABLED_SELECT)
    50 
    -    /* FDs for interrupting up the I/O manager when it is blocked waiting */
    49 
    +#if defined(IOMGR_ENABLED_SELECT) || defined(IOMGR_ENABLED_POLL)
    50 
    +#if defined(HAVE_PREEMPTION)
    51 
    +    /* FDs for waking up the I/O manager when it is blocked waiting */
    51 52 
         int interrupt_fd_r, interrupt_fd_w;
    52 53 
     #endif
    54 
    +#endif
    53 55
    54 56 
     #if defined(IOMGR_ENABLED_POLL)
    55 57 
         /* AIOP and timeout collections shared by several I/O manager impls */
    ... ... @@ -58,8 +60,11 @@ struct _CapIOManager {
    58 60 
     #endif
    59 61
    60 62 
     #if defined(IOMGR_ENABLED_POLL)
    61 
    -    /* Auxiliary table with size and indexes matching the aiop_table */
    62 
    -    struct pollfd *aiop_poll_table;
    63 
    +    /* Auxiliary table with size and indexes matching the aiop_table. This is
    64 
    +     * aliased to the tail of the full poll table, which has a head entry for
    65 
    +     * the wakeup_fd_r above, so we can also poll that fd.
    66 
    +     */
    67 
    +    struct pollfd *aiop_poll_table, *full_poll_table;
    63 68 
     #endif
    64 69
    65 70 
     #if defined(IOMGR_ENABLED_WIN32_LEGACY)

  • rts/posix/FdWakeup.h
    ... ... @@ -29,12 +29,14 @@
    29 29
    30 30 
     #include "BeginPrivate.h"
    31 31
    32 
    +#if defined(HAVE_PREEMPTION)
    32 33 
     void newFdWakeup(int *fd_r, int *fd_w);
    33 34 
     void closeFdWakeup(int fd_r, int fd_w);
    34 35
    35 36 
     /* This is safe to use from a signal handler */
    36 37 
     void sendFdWakeup(int fd_w);
    37 38 
     void collectFdWakeup(int fd_r);
    39 
    +#endif
    38 40
    39 41 
     #include "EndPrivate.h"
    40 42

  • 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 interrupt_fd_r, interrupt_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 interrupt_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 
    +interrupt_fd_r. This means the aiop_poll_table indicies match up exactly with
    129 
    +the aiop_table, but still allows the full_poll_table to have an extra entry.
    130 
    +
    120 131 
     ******************************************************************************/
    121 132
    122 133 
     /* Forward declarations */
    ... ... @@ -129,16 +140,34 @@ 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 
    +#if defined(HAVE_PREEMPTION)
    146 
    +    newFdWakeup(&iomgr->interrupt_fd_r, &iomgr->interrupt_fd_w);
    147 
    +#endif
    148 
    +
    149 
    +    iomgr->full_poll_table = stgMallocBytes(sizeof(struct pollfd) /* size 1 */,
    150 
    +                                            "initCapabilityIOManagerPoll");
    151 
    +    iomgr->full_poll_table[0] = (struct pollfd) {
    152 
    +#if defined(HAVE_PREEMPTION)
    153 
    +                                  .fd      = iomgr->interrupt_fd_r,
    154 
    +                                  .events  = POLLIN,
    155 
    +#else
    156 
    +                                  .fd      = -1, // unused
    157 
    +                                  .events  = 0,  // unused
    158 
    +#endif
    159 
    +                                  .revents = 0
    160 
    +                                };
    161 
    +    iomgr->aiop_poll_table = iomgr->full_poll_table+1; /* hence empty */
    134 162 
     }
    135 163
    136 164
    137 165 
     void freeCapabilityIOManagerPoll(CapIOManager *iomgr)
    138 166 
     {
    139 
    -    if (iomgr->aiop_poll_table) {
    140 
    -        stgFree(iomgr->aiop_poll_table);
    141 
    -    }
    167 
    +    stgFree(iomgr->full_poll_table);
    168 
    +#if defined(HAVE_PREEMPTION)
    169 
    +    closeFdWakeup(iomgr->interrupt_fd_r, iomgr->interrupt_fd_w);
    170 
    +#endif
    142 171 
     }
    143 172
    144 173
    ... ... @@ -295,7 +324,7 @@ static void notifyIOCompletion(CapIOManager *iomgr, StgAsyncIOOp *aiop)
    295 324 
     }
    296 325
    297 326
    298 
    -static void processIOCompletions(CapIOManager *iomgr, int ncompletions)
    327 
    +static bool processIOCompletions(CapIOManager *iomgr, int ncompletions)
    299 328 
     {
    300 329 
         /* The scheme we use with poll is that we have a dense poll table, and a
    301 330 
          * corresponding table that maps to the closure table index. The poll
    ... ... @@ -305,6 +334,19 @@ static void processIOCompletions(CapIOManager *iomgr, int ncompletions)
    305 334 
          */
    306 335 
         debugTrace(DEBUG_iomanager, "processIOCompletions(ncompletions = %d)",
    307 336 
                                     ncompletions);
    337 
    +
    338 
    +    bool interrupt = false;
    339 
    +#if defined(HAVE_PREEMPTION)
    340 
    +    /* If the interrupt_fd_r is ready, collect it */
    341 
    +    if (iomgr->full_poll_table[0].revents) {
    342 
    +        ASSERT(iomgr->full_poll_table[0].fd == iomgr->interrupt_fd_r);
    343 
    +        collectFdWakeup(iomgr->interrupt_fd_r);
    344 
    +        ncompletions--;
    345 
    +        interrupt = true;
    346 
    +        debugTrace(DEBUG_iomanager, "Received interrupt in poll I/O manager");
    347 
    +    }
    348 
    +#endif
    349 
    +
    308 350 
         struct pollfd *aiop_poll_table = iomgr->aiop_poll_table;
    309 351 
         int n = ncompletions;
    310 352 
         int i = 0;
    ... ... @@ -357,11 +399,14 @@ static void processIOCompletions(CapIOManager *iomgr, int ncompletions)
    357 399 
                 i++;
    358 400 
             }
    359 401 
         }
    402 
    +    return interrupt;
    360 403 
     }
    361 404
    362 405
    363 406 
     void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
    364 407 
     {
    408 
    +    ASSERT(iomgr->aiop_poll_table == iomgr->full_poll_table+1);
    409 
    +
    365 410 
         if (!isEmptyTimeoutQueue(iomgr->timeout_queue)) {
    366 411 
             Time now = getProcessElapsedTime();
    367 412 
             processTimeoutCompletions(iomgr, now);
    ... ... @@ -369,20 +414,28 @@ void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
    369 414
    370 415 
         if (!isEmptyClosureTable(&iomgr->aiop_table)) {
    371 416
    372 
    -        nfds_t nfds = sizeClosureTable(&iomgr->aiop_table);
    417 
    +        nfds_t nfds = sizeClosureTable(&iomgr->aiop_table) + 1;
    418 
    +
    419 
    +#if defined(HAVE_PREEMPTION)
    420 
    +        /* the full_poll_table includes interrupt_fd_r */
    421 
    +        struct pollfd *poll_table = iomgr->full_poll_table;
    422 
    +#else
    423 
    +        /* the aiop_poll_table does not include interrupt_fd_r */
    424 
    +        struct pollfd *poll_table = iomgr->aiop_poll_table;
    425 
    +#endif
    373 426
    374 427 
             /* Poll for I/O readiness, without waiting. */
    375 428 
     #if defined(HAVE_DECL_PPOLL) && HAVE_DECL_PPOLL == 1
    376 429 
             /* We could use poll here, since we use no timeout, but for
    377 430 
                consistency we use the same syscall as at the other call site. */
    378 431 
             struct timespec tv = (struct timespec) { .tv_sec = 0, .tv_nsec = 0 };
    379 
    -        int res = ppoll(iomgr->aiop_poll_table, nfds, &tv, NULL);
    432 
    +        int res = ppoll(poll_table, nfds, &tv, NULL);
    380 433
    381 434 
             debugTrace(DEBUG_iomanager,
    382 435 
                        "ppoll(nfds = %d, timeout.sec = 0, timeout.nsec = 0) = %d",
    383 436 
                        nfds, res);
    384 437 
     #else
    385 
    -        int res = poll(iomgr->aiop_poll_table, nfds, 0);
    438 
    +        int res = poll(poll_table, nfds, 0);
    386 439
    387 440 
             debugTrace(DEBUG_iomanager,
    388 441 
                        "poll(nfds = %d, timeout_ms = 0) = %d",
    ... ... @@ -408,8 +461,12 @@ void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
    408 461 
     }
    409 462
    410 463
    411 
    -void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
    464 
    +bool awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
    412 465 
     {
    466 
    +    bool interrupt = false; /* got woken up via interruptIOManager */
    467 
    +
    468 
    +    ASSERT(iomgr->aiop_poll_table == iomgr->full_poll_table+1);
    469 
    +
    413 470 
         /* Loop until we've woken up some threads. This loop is needed because the
    414 471 
          * poll() timing isn't accurate, we sometimes sleep for a while but not
    415 472 
          * long enough to wake up a thread in a threadDelay. Or we may need to
    ... ... @@ -431,6 +488,14 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
    431 488 
              */
    432 489 
             bool wait = emptyRunQueue(iomgr->cap);
    433 490
    491 
    +#if defined(HAVE_PREEMPTION)
    492 
    +        /* the full_poll_table includes interrupt_fd_r */
    493 
    +        struct pollfd *poll_table = iomgr->full_poll_table;
    494 
    +#else
    495 
    +        /* the aiop_poll_table does not include interrupt_fd_r */
    496 
    +        struct pollfd *poll_table = iomgr->aiop_poll_table;
    497 
    +#endif
    498 
    +
    434 499 
             /* Decide if we are going to wait if no I/O is ready, either:
    435 500 
              * poll only, wait indefinitely, or wait until a timeout.
    436 501 
              */
    ... ... @@ -442,9 +507,9 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
    442 507 
     #endif
    443 508
    444 509 
             /* Check for I/O readiness, possibly waiting. */
    445 
    -        nfds_t nfds = sizeClosureTable(&iomgr->aiop_table);
    510 
    +        nfds_t nfds = sizeClosureTable(&iomgr->aiop_table) + 1;
    446 511 
     #if defined(HAVE_DECL_PPOLL) && HAVE_DECL_PPOLL == 1
    447 
    -        int res = ppoll(iomgr->aiop_poll_table, nfds, timeout_ns, NULL);
    512 
    +        int res = ppoll(poll_table, nfds, timeout_ns, NULL);
    448 513
    449 514 
             debugTrace(DEBUG_iomanager,
    450 515 
                        "ppoll(nfds = %d, timeout.sec = %d, timeout.nsec = %d) = %d",
    ... ... @@ -452,7 +517,7 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
    452 517 
                              timeout_ns == NULL ?  0 : timeout_ns->tv_nsec,
    453 518 
                        res);
    454 519 
     #else
    455 
    -        int res = poll(iomgr->aiop_poll_table, nfds, timeout_ms);
    520 
    +        int res = poll(poll_table, nfds, timeout_ms);
    456 521
    457 522 
             debugTrace(DEBUG_iomanager,
    458 523 
                        "poll(nfds = %d, timeout_ms = %d) = %d",
    ... ... @@ -474,7 +539,7 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
    474 539 
             } else if (res > 0) {
    475 540 
                 int ncompletions = res;
    476 541 
                 ASSERT(ncompletions <= (int)nfds);
    477 
    -            processIOCompletions(iomgr, ncompletions);
    542 
    +            interrupt = processIOCompletions(iomgr, ncompletions);
    478 543 
                 // FIXME: do we also need to check for timeout completions now?
    479 544 
                 // we have a non-empty queue, but if !wait then we have also moved
    480 545 
                 // on and so we sould check for timeouts.
    ... ... @@ -502,7 +567,9 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
    502 567 
             }
    503 568
    504 569 
         } while (emptyRunQueue(iomgr->cap)
    570 
    +         && !interrupt
    505 571 
              && (getSchedState() == SCHED_RUNNING));
    572 
    +    return !interrupt;
    506 573 
     }
    507 574
    508 575 
     static void reportPollError(int res, nfds_t nfds)
    ... ... @@ -521,6 +588,14 @@ static void reportPollError(int res, nfds_t nfds)
    521 588 
     }
    522 589
    523 590
    591 
    +void interruptIOManagerPoll(CapIOManager *iomgr)
    592 
    +{
    593 
    +#if defined(HAVE_PREEMPTION)
    594 
    +    sendFdWakeup(iomgr->interrupt_fd_w);
    595 
    +#endif
    596 
    +}
    597 
    +
    598 
    +
    524 599 
     /* Helper function to double the size of the aiop_table and aiop_poll_table.
    525 600 
      */
    526 601 
     static bool enlargeTables(CapIOManager *iomgr)
    ... ... @@ -531,13 +606,17 @@ static bool enlargeTables(CapIOManager *iomgr)
    531 606 
         bool ok = enlargeClosureTable(iomgr->cap, &iomgr->aiop_table, newcapacity);
    532 607 
         if (RTS_UNLIKELY(!ok)) return false;
    533 608
    534 
    -    /* Update the auxiliary aiop_poll_table to match */
    535 
    -    struct pollfd *aiop_poll_table;
    536 
    -    aiop_poll_table = stgReallocBytes(iomgr->aiop_poll_table,
    537 
    -                                      sizeof(struct pollfd) * newcapacity,
    538 
    -                                      "Poll.c: enlargeTables");
    539 
    -    iomgr->aiop_poll_table = aiop_poll_table;
    609 
    +    /* Update the auxiliary aiop_poll_table to match. The full_poll_table is
    610 
    +     * one bigger than the aiop_poll_table, since it has an extra entry at the
    611 
    +     * front for interrupt_fd_r, with no corresponding aiop. */
    612 
    +    iomgr->full_poll_table =
    613 
    +        stgReallocBytes(iomgr->full_poll_table,
    614 
    +                        sizeof(struct pollfd) * (newcapacity+1),
    615 
    +                        "Poll.c: enlargeTables");
    616 
    +    iomgr->aiop_poll_table = iomgr->full_poll_table+1;
    617 
    +
    540 618 
         /* Initialise the new part of the aiop_poll_table */
    619 
    +    struct pollfd *aiop_poll_table = iomgr->aiop_poll_table;
    541 620 
         for (int i = oldcapacity; i < newcapacity; i++) {
    542 621 
             aiop_poll_table[i] = (struct pollfd) {
    543 622 
                                    .fd      = -1,

  • rts/posix/Poll.h
    ... ... @@ -32,7 +32,8 @@ void asyncIOCancelPoll(CapIOManager *iomgr, StgAsyncIOOp *aiop);
    32 32 
     /* Scheduler operations */
    33 33 
     bool anyPendingTimeoutsOrIOPoll(CapIOManager *iomgr);
    34 34 
     void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr);
    35 
    -void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr);
    35 
    +bool awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr);
    36 
    +void interruptIOManagerPoll(CapIOManager *iomgr);
    36 37
    37 38 
     #endif /* IOMGR_ENABLED_POLL */
    38 39

    • —
    View it on GitLab.
    You're receiving this email because of your account on gitlab.haskell.org. Manage all notifications · Help Notification message regarding https://gitlab.haskell.org/ghc/ghc/-/commit/3ed3922c26d258eb76b33d1ab4256ca843b37a62 at 1784113753