Zubin pushed to branch wip/27113-retry at Glasgow Haskell Compiler / GHC

Commits:

11 changed files:

Changes:

  • rts/RaiseAsync.c
    ... ... @@ -36,6 +36,11 @@ static void throwToSendMsg (Capability *cap USED_IF_THREADS,
    36 36 
                                 Capability *target_cap USED_IF_THREADS,
    37 37 
                                 MessageThrowTo *msg USED_IF_THREADS);
    38 38
    39 
    +#if defined(THREADED_RTS)
    40 
    +static void clear_interrupt_pending_if_no_live_throw (Capability *cap,
    41 
    +                                                      StgTSO *target);
    42 
    +#endif
    43 
    +
    39 44 
     /* -----------------------------------------------------------------------------
    40 45 
        throwToSingleThreaded
    41 46
    ... ... @@ -338,7 +343,13 @@ check_target:
    338 343 
             }
    339 344
    340 345 
             // nobody else can wake up this TSO after we claim the message
    346 
    +#if defined(THREADED_RTS)
    347 
    +        StgTSO *m_target = m->target;
    348 
    +#endif
    341 349 
             doneWithMsgThrowTo(cap, m);
    350 
    +#if defined(THREADED_RTS)
    351 
    +        clear_interrupt_pending_if_no_live_throw(cap, m_target);
    352 
    +#endif
    342 353
    343 354 
             raiseAsync(cap, target, msg->exception, false, NULL);
    344 355 
             return THROWTO_SUCCESS;
    ... ... @@ -433,11 +444,12 @@ check_target:
    433 444 
     #if defined(THREADED_RTS)
    434 445 
         {
    435 446 
             Task *task = NULL;
    436 
    -        // walk suspended_ccalls to find the correct worker thread
    447 
    +        InCall *target_incall = NULL;
    437 448 
             InCall *incall;
    438 449 
             for (incall = cap->suspended_ccalls; incall != NULL; incall = incall->next) {
    439 450 
                 if (incall->suspended_tso == target) {
    440 451 
                     task = incall->task;
    452 
    +                target_incall = incall;
    441 453 
                     break;
    442 454 
                 }
    443 455 
             }
    ... ... @@ -446,6 +458,7 @@ check_target:
    446 458 
                 if (!((target->flags & TSO_BLOCKEX) &&
    447 459 
                       ((target->flags & TSO_INTERRUPTIBLE) == 0))) {
    448 460 
                     interruptWorkerTask(task);
    461 
    +                RELEASE_STORE(&target_incall->interrupt_pending, 1);
    449 462 
                 }
    450 463 
                 return THROWTO_BLOCKED;
    451 464 
             } else {
    ... ... @@ -520,8 +533,14 @@ blockedThrowTo (Capability *cap, StgTSO *target, MessageThrowTo *msg)
    520 533 
         ASSERT(target->cap == cap);
    521 534
    522 535 
         dirty_TSO(cap,target); // we will modify the blocked_exceptions queue
    536 
    +#if defined(THREADED_RTS)
    537 
    +    ACQUIRE_LOCK(&cap->lock);
    538 
    +#endif
    523 539 
         msg->link = target->blocked_exceptions;
    524 
    -    target->blocked_exceptions = msg;
    540 
    +    RELEASE_STORE(&target->blocked_exceptions, msg);
    541 
    +#if defined(THREADED_RTS)
    542 
    +    RELEASE_LOCK(&cap->lock);
    543 
    +#endif
    525 544 
     }
    526 545
    527 546 
     /* -----------------------------------------------------------------------------
    ... ... @@ -664,6 +683,46 @@ removeFromMVarBlockedQueue (StgTSO *tso)
    664 683 
         tso->_link = END_TSO_QUEUE;
    665 684 
     }
    666 685
    686 
    +#if defined(THREADED_RTS)
    687 
    +static void
    688 
    +clear_interrupt_pending_if_no_live_throw (Capability *cap, StgTSO *target)
    689 
    +{
    690 
    +    if (target == NULL) return;
    691 
    +    if (ACQUIRE_LOAD(&target->why_blocked) != BlockedOnCCall_Interruptible)
    692 
    +        return;
    693 
    +
    694 
    +    Capability *target_cap = ACQUIRE_LOAD(&target->cap);
    695 
    +    ACQUIRE_LOCK(&target_cap->lock);
    696 
    +
    697 
    +    // Find the InCall on target_cap.  If target has already resumed (or
    698 
    +    // wasn't really suspended on this cap), do nothing.
    699 
    +    InCall *found = NULL;
    700 
    +    for (InCall *ic = target_cap->suspended_ccalls; ic != NULL; ic = ic->next) {
    701 
    +        if (ic->suspended_tso == target) { found = ic; break; }
    702 
    +    }
    703 
    +    if (found == NULL) {
    704 
    +        RELEASE_LOCK(&target_cap->lock);
    705 
    +        return;
    706 
    +    }
    707 
    +
    708 
    +    bool live = false;
    709 
    +    for (MessageThrowTo *t = ACQUIRE_LOAD(&target->blocked_exceptions);
    710 
    +         t != END_BLOCKED_EXCEPTIONS_QUEUE;
    711 
    +         t = (MessageThrowTo*)t->link) {
    712 
    +        if (RELAXED_LOAD((StgWord*)&t->header.info)
    713 
    +            != (StgWord)&stg_MSG_NULL_info) {
    714 
    +            live = true; break;
    715 
    +        }
    716 
    +    }
    717 
    +    if (!live) {
    718 
    +        RELEASE_STORE(&found->interrupt_pending, 0);
    719 
    +    }
    720 
    +
    721 
    +    RELEASE_LOCK(&target_cap->lock);
    722 
    +    (void)cap;
    723 
    +}
    724 
    +#endif
    725 
    +
    667 726 
     static void
    668 727 
     removeFromQueues(Capability *cap, StgTSO *tso)
    669 728 
     {
    ... ... @@ -699,8 +758,16 @@ removeFromQueues(Capability *cap, StgTSO *tso)
    699 758 
           // capabilities.
    700 759 
           // ASSERT(m->header.info == &stg_WHITEHOLE_info);
    701 760
    761 
    +#if defined(THREADED_RTS)
    762 
    +      StgTSO *m_target = m->target;
    763 
    +#endif
    764 
    +
    702 765 
           // unlock and revoke it at the same time
    703 766 
           doneWithMsgThrowTo(cap, m);
    767 
    +
    768 
    +#if defined(THREADED_RTS)
    769 
    +      clear_interrupt_pending_if_no_live_throw(cap, m_target);
    770 
    +#endif
    704 771 
           break;
    705 772 
       }
    706 773

  • rts/Schedule.c
    ... ... @@ -2536,6 +2536,25 @@ suspendThread (StgRegTable *reg, bool interruptible)
    2536 2536
    2537 2537 
       suspendTask(cap,task);
    2538 2538 
       cap->in_haskell = false;
    2539 
    +
    2540 
    +#if defined(THREADED_RTS)
    2541 
    +  if (interruptible
    2542 
    +      && !((tso->flags & TSO_BLOCKEX) &&
    2543 
    +           ((tso->flags & TSO_INTERRUPTIBLE) == 0))) {
    2544 
    +      MessageThrowTo *head = ACQUIRE_LOAD(&tso->blocked_exceptions);
    2545 
    +      bool live = false;
    2546 
    +      for (MessageThrowTo *m = head;
    2547 
    +           m != END_BLOCKED_EXCEPTIONS_QUEUE;
    2548 
    +           m = (MessageThrowTo*)m->link) {
    2549 
    +          if (RELAXED_LOAD((StgWord*)&m->header.info)
    2550 
    +              != (StgWord)&stg_MSG_NULL_info) { live = true; break; }
    2551 
    +      }
    2552 
    +      if (live) {
    2553 
    +          RELEASE_STORE(&task->incall->interrupt_pending, 1);
    2554 
    +      }
    2555 
    +  }
    2556 
    +#endif
    2557 
    +
    2539 2558 
       releaseCapability_(cap,false);
    2540 2559
    2541 2560 
       RELEASE_LOCK(&cap->lock);
    ... ... @@ -2574,8 +2593,10 @@ resumeThread (void *task_)
    2574 2593 
         // entry on the suspended_ccalls list will also have been
    2575 2594 
         // migrated.
    2576 2595
    2577 
    -    // Remove the thread from the suspended list
    2596 
    +    ACQUIRE_LOCK(&cap->lock);
    2597 
    +    RELEASE_STORE(&incall->interrupt_pending, 0);
    2578 2598 
         recoverSuspendedTask(cap,task);
    2599 
    +    RELEASE_LOCK(&cap->lock);
    2579 2600
    2580 2601 
         tso = incall->suspended_tso;
    2581 2602 
         incall->suspended_tso = NULL;
    ... ... @@ -2614,6 +2635,31 @@ resumeThread (void *task_)
    2614 2635 
         return &cap->r;
    2615 2636 
     }
    2616 2637
    2638 
    +#if defined(THREADED_RTS)
    2639 
    +void
    2640 
    +retryInterruptibleSignals(void)
    2641 
    +{
    2642 
    +    bool any_pending = false;
    2643 
    +    uint32_t n = getNumCapabilities();
    2644 
    +    for (uint32_t i = 0; i < n; i++) {
    2645 
    +        Capability *cap = getCapability(i);
    2646 
    +        if (RELAXED_LOAD(&cap->n_suspended_ccalls) == 0) continue;
    2647 
    +        if (TRY_ACQUIRE_LOCK(&cap->lock) != 0) continue;
    2648 
    +        for (InCall *ic = cap->suspended_ccalls; ic != NULL; ic = ic->next) {
    2649 
    +            if (ACQUIRE_LOAD(&ic->interrupt_pending) == 0) continue;
    2650 
    +            // ic->task is set by newInCall and never cleared.
    2651 
    +            if (ic->task != NULL) {
    2652 
    +                any_pending = true;
    2653 
    +                interruptOSThread(ic->task->id);
    2654 
    +            }
    2655 
    +        }
    2656 
    +        RELEASE_LOCK(&cap->lock);
    2657 
    +    }
    2658 
    +    // Keep the timer alive; handle_tick's idle path would otherwise stop it.
    2659 
    +    if (any_pending) setRecentActivity(ACTIVITY_YES);
    2660 
    +}
    2661 
    +#endif
    2662 
    +
    2617 2663 
     /* ---------------------------------------------------------------------------
    2618 2664 
      * scheduleThread()
    2619 2665 
      *

  • rts/Schedule.h
    ... ... @@ -43,6 +43,10 @@ void scheduleThreadOn(Capability *cap, StgWord cpu, StgTSO *tso);
    43 43 
     void wakeUpRts(void);
    44 44 
     #endif
    45 45
    46 
    +#if defined(THREADED_RTS)
    47 
    +void retryInterruptibleSignals(void);
    48 
    +#endif
    49 
    +
    46 50 
     /* raiseExceptionHelper */
    47 51 
     StgWord raiseExceptionHelper (StgRegTable *reg, StgTSO *tso, StgClosure *exception);
    48 52

  • rts/Task.c
    ... ... @@ -267,6 +267,7 @@ newInCall (Task *task)
    267 267 
         incall->next = NULL;
    268 268 
         incall->prev = NULL;
    269 269 
         incall->prev_stack = task->incall;
    270 
    +    incall->interrupt_pending = 0;
    270 271 
         task->incall = incall;
    271 272 
     }
    272 273

  • rts/Task.h
    ... ... @@ -106,6 +106,8 @@ typedef struct InCall_ {
    106 106 
         // Links InCalls onto suspended_ccalls, spare_incalls
    107 107 
         struct InCall_ *prev;
    108 108 
         struct InCall_ *next;
    109 
    +
    110 
    +    StgWord interrupt_pending;
    109 111 
     } InCall;
    110 112
    111 113 
     typedef struct Task_ {

  • rts/Timer.c
    ... ... @@ -131,6 +131,9 @@ handle_tick(int unused STG_UNUSED)
    131 131 
               flushEventLog(NULL);
    132 132 
           }
    133 133 
       }
    134 
    +  if (SEQ_CST_LOAD_ALWAYS(&timer_disabled) == 0) {
    135 
    +      retryInterruptibleSignals();
    136 
    +  }
    134 137 
     #endif
    135 138
    136 139 
       /*

  • testsuite/tests/concurrent/should_run/T27113.hs
    ... ... @@ -10,7 +10,7 @@ import System.IO
    10 10 
     import System.Timeout
    11 11
    12 12 
     foreign import ccall interruptible "cpu_then_pause"
    13 
    -    c_cpu_then_pause :: CLong -> IO ()
    13 
    +    c_cpu_then_pause :: CLLong -> IO ()
    14 14
    15 15 
     killDelay :: Int
    16 16 
     killDelay = 200_000
    ... ... @@ -18,7 +18,7 @@ killDelay = 200_000
    18 18 
     joinTimeout :: Int
    19 19 
     joinTimeout = 3_000_000
    20 20
    21 
    -runCase :: String -> CLong -> IO ()
    21 
    +runCase :: String -> CLLong -> IO ()
    22 22 
     runCase name spinNs = do
    23 23 
         tid <- forkIO $
    24 24 
             c_cpu_then_pause spinNs

  • testsuite/tests/concurrent/should_run/T27113_c.c
    1 1 
     #include <errno.h>
    2 
    +#include <stdint.h>
    2 3 
     #include <time.h>
    3 4 
     #include <unistd.h>
    4 5
    5 
    -static long now_ns(void) {
    6 
    +static int64_t now_ns(void) {
    6 7 
         struct timespec t;
    7 8 
         while (clock_gettime(CLOCK_MONOTONIC, &t) != 0) {
    8 9 
             if (errno != EINTR) return 0;
    9 10 
         }
    10 
    -    return (long)t.tv_sec * 1000000000L + (long)t.tv_nsec;
    11 
    +    return (int64_t)t.tv_sec * 1000000000LL + (int64_t)t.tv_nsec;
    11 12 
     }
    12 13
    13 
    -int cpu_then_pause(long spin_ns) {
    14 
    -    long deadline = now_ns() + spin_ns;
    14 
    +int cpu_then_pause(int64_t spin_ns) {
    15 
    +    int64_t deadline = now_ns() + spin_ns;
    15 16 
         while (now_ns() < deadline) {
    16 17 
         }
    17 18 
         return pause();

  • testsuite/tests/concurrent/should_run/T27113b.hs
    1 
    +{-# LANGUAGE InterruptibleFFI #-}
    2 
    +{-# LANGUAGE ScopedTypeVariables #-}
    3 
    +
    4 
    +module Main where
    5 
    +
    6 
    +import Control.Concurrent
    7 
    +import Control.Exception
    8 
    +import Foreign.C.Types
    9 
    +import System.IO
    10 
    +import System.Timeout
    11 
    +
    12 
    +foreign import ccall interruptible "cpu_then_pause"
    13 
    +    c_cpu_then_pause :: CLLong -> IO ()
    14 
    +
    15 
    +myCall :: CLLong -> IO ()
    16 
    +myCall spinNs = mask_ $ do
    17 
    +    allowInterrupt
    18 
    +    c_cpu_then_pause spinNs
    19 
    +    allowInterrupt
    20 
    +
    21 
    +main :: IO ()
    22 
    +main = do
    23 
    +    hSetBuffering stdout NoBuffering
    24 
    +    started <- newEmptyMVar
    25 
    +    done <- newEmptyMVar
    26 
    +    tid <- mask_ $ forkIOWithUnmask $ \unmask ->
    27 
    +        flip finally (putMVar done ()) $ do
    28 
    +            putMVar started ()
    29 
    +            unmask (myCall 2_000_000_000)
    30 
    +                `catch` (\(_ :: SomeException) -> return ())
    31 
    +    takeMVar started
    32 
    +    threadDelay 200_000
    33 
    +    result <- timeout 5_000_000 (killThread tid >> takeMVar done)
    34 
    +    case result of
    35 
    +        Just () -> putStrLn "ok"
    36 
    +        Nothing -> do
    37 
    +            hPutStrLn stderr "FAIL: killThread hung"
    38 
    +            error "hung"

  • testsuite/tests/concurrent/should_run/T27113c.hs
    1 
    +{-# LANGUAGE InterruptibleFFI #-}
    2 
    +{-# LANGUAGE ScopedTypeVariables #-}
    3 
    +
    4 
    +module Main where
    5 
    +
    6 
    +import Control.Concurrent
    7 
    +import Control.Exception
    8 
    +import Control.Monad
    9 
    +import Data.IORef
    10 
    +import Foreign.C.Types
    11 
    +import System.Exit
    12 
    +import System.IO
    13 
    +
    14 
    +foreign import ccall interruptible "pause" c_pause :: IO CInt
    15 
    +
    16 
    +postRevokeWindow :: Int
    17 
    +postRevokeWindow = 2_000_000
    18 
    +
    19 
    +postRevokeThreshold :: Int
    20 
    +postRevokeThreshold = 5
    21 
    +
    22 
    +main :: IO ()
    23 
    +main = do
    24 
    +    hSetBuffering stdout NoBuffering
    25 
    +    iters <- newIORef (0 :: Int)
    26 
    +
    27 
    +    workerStarted <- newEmptyMVar
    28 
    +    worker <- forkIO $ mask_ $ do
    29 
    +        putMVar workerStarted ()
    30 
    +        (forever $ do
    31 
    +            atomicModifyIORef' iters (\n -> (n + 1, ()))
    32 
    +            _ <- c_pause
    33 
    +            return ())
    34 
    +          `catch` (\(_ :: SomeException) -> return ())
    35 
    +
    36 
    +    takeMVar workerStarted
    37 
    +    threadDelay 100_000
    38 
    +
    39 
    +    killer1Ready <- newEmptyMVar
    40 
    +    killer1 <- forkIO $ do
    41 
    +        putMVar killer1Ready ()
    42 
    +        killThread worker
    43 
    +
    44 
    +    takeMVar killer1Ready
    45 
    +    threadDelay 100_000
    46 
    +
    47 
    +    preRevoke <- readIORef iters
    48 
    +
    49 
    +    killThread killer1
    50 
    +
    51 
    +    threadDelay postRevokeWindow
    52 
    +
    53 
    +    postRevoke <- readIORef iters
    54 
    +    let storm = postRevoke - preRevoke
    55 
    +    if storm > postRevokeThreshold
    56 
    +        then do
    57 
    +            hPutStrLn stderr ("FAIL: post-revoke EINTR storm, iterations="
    58 
    +                              ++ show storm
    59 
    +                              ++ " (pre=" ++ show preRevoke
    60 
    +                              ++ " post=" ++ show postRevoke ++ ")")
    61 
    +            exitFailure
    62 
    +        else
    63 
    +            putStrLn ("ok (post-revoke iterations=" ++ show storm ++ ")")

  • testsuite/tests/concurrent/should_run/all.T
    ... ... @@ -208,11 +208,24 @@ test('foreignInterruptible', [when(fast(), skip),
    208 208
    209 209 
     test('T27113',
    210 210 
          [when(opsys('mingw32'), skip),
    211 
    +      only_threaded_ways,
    212 
    +      req_c],
    213 
    +     compile_and_run, ['T27113_c.c'])
    214 
    +
    215 
    +test('T27113b',
    216 
    +     [extra_files(['T27113_c.c']),
    217 
    +      when(opsys('mingw32'), skip),
    211 218 
           only_threaded_ways,
    212 219 
           req_c,
    213 
    -      expect_broken(27113)],
    220 
    +      ignore_stdout],
    214 221 
          compile_and_run, ['T27113_c.c'])
    215 222
    223 
    +test('T27113c',
    224 
    +     [when(opsys('mingw32'), skip),
    225 
    +      only_threaded_ways,
    226 
    +      ignore_stdout],
    227 
    +     compile_and_run, [''])
    228 
    +
    216 229 
     test('conc037', only_ways(['threaded1', 'threaded2', 'nonmoving_thr']), compile_and_run, [''])
    217 230 
     test('conc038', only_ways(['threaded1', 'threaded2', 'nonmoving_thr']), compile_and_run, [''])
    218 231

    • —
    View it on GitLab.
    You're receiving this email because of your account on gitlab.haskell.org. Manage all notifications · Help