GitLab

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

Commits:

dc4d1c48

by Zubin Duggal at 2026-05-19T18:21:00+05:30

rts: Retry SIGPIPE delivery for interruptible FFI calls (#27113)

Closes a race in the threaded RTS where a SIGPIPE fired by throwTo to
cancel an interruptible foreign call may land in the user-space prologue
of the FFI wrapper, between suspendThread releasing the capability and
the kernel syscall starting.  GHC's empty SIGPIPE handler swallows the
signal and the syscall then blocks forever.

The original retry implementation walked cap->suspended_ccalls from the
ticker and read TSO heap fields (why_blocked, blocked_exceptions, flags)
to decide whether to re-fire interruptOSThread.  Review (review.md)
identified two real bugs in that design:

  H1.  TSO reads from outside cap ownership race with moving GC's
       evacuation of suspended_ccalls TSOs and with descheduling between
       the load of incall->suspended_tso and the subsequent field reads.
       Spurious-driven SIGPIPE EINTRs unrelated unsafe syscalls on the
       same OS thread once the worker has moved on.
  H2.  blocked_exceptions \!= END is not a live-throw predicate.  A
       MessageThrowTo overwritten with MSG_NULL by doneWithMsgThrowTo
       persists on the queue under mask_, causing an EINTR storm in
       throwErrnoIfMinus1Retry-style wrappers.

This commit reworks the fix to address both:

  * A per-InCall atomic flag `interrupt_pending` is set by throwToMsg
    after it fires interruptOSThread and cleared by resumeThread (before
    recoverSuspendedTask, so the ticker cannot observe a stale set flag
    after the InCall has been unlinked).
  * throwToMsg's BlockedOnCCall_Interruptible search is extended to
    walk every capability's suspended_ccalls (under cap->lock), so the
    flag is set even when the target migrated to a different cap from
    the throwing thread.
  * doneWithMsgThrowTo's revoke path (removeFromQueues') re-scans the
    target's blocked_exceptions for any non-MSG_NULL entries; if none
    remain, clears the InCall's interrupt_pending so the retry stops.
    Fixes H2.
  * The ticker now reads only InCall fields (interrupt_pending,
    task->id) — never TSO heap state — and fires interruptOSThread
    while interrupt_pending is set.  Fixes H1.
  * The retry hook in handle_tick is gated on timer_disabled, matching
    the context-switch path (review M1).
  * T27113_c.c switched to int64_t to avoid 32-bit `long` overflow on
    32-bit POSIX (review L1).
  * Adds T27113c, a regression test for H2: a MessageThrowTo against
    an interruptible-FFI target is revoked (via killing the throwing
    thread), and the target's masked EINTR-retry loop is checked for
    forward progress (no EINTR storm).

Build: T27113 and T27113c pass on ghci/threaded1/threaded2 ways; no
regressions in conc014/015/015a/016/017/017a/058/foreignInterruptible
on threaded2.

11 changed files:

rts/RaiseAsync.c
rts/Schedule.c
rts/Schedule.h
rts/Task.c
rts/Task.h
rts/Timer.c
testsuite/tests/concurrent/should_run/T27113.hs
testsuite/tests/concurrent/should_run/T27113_c.c
+ testsuite/tests/concurrent/should_run/T27113b.hs
+ testsuite/tests/concurrent/should_run/T27113c.hs
testsuite/tests/concurrent/should_run/all.T

Changes:

rts/RaiseAsync.c

@@ -36,6 +36,11 @@ static void throwToSendMsg (Capability *cap USED_IF_THREADS,
                              Capability *target_cap USED_IF_THREADS,
                              MessageThrowTo *msg USED_IF_THREADS);
 +#if defined(THREADED_RTS)
 +static void clear_interrupt_pending_if_no_live_throw (Capability *cap,
 +                                                      StgTSO *target);
 +#endif
++
  /* -----------------------------------------------------------------------------
     throwToSingleThreaded
@@ -338,7 +343,14 @@ check_target:
+         }
          // nobody else can wake up this TSO after we claim the message
 +#if defined(THREADED_RTS)
 +        // #27113: snapshot m->target before doneWithMsgThrowTo nulls m.
 +        StgTSO *m_target = m->target;
 +#endif
          doneWithMsgThrowTo(cap, m);
 +#if defined(THREADED_RTS)
 +        clear_interrupt_pending_if_no_live_throw(cap, m_target);
 +#endif
          raiseAsync(cap, target, msg->exception, false, NULL);
          return THROWTO_SUCCESS;
@@ -433,11 +445,12 @@ check_target:
  #if defined(THREADED_RTS)
+     {
          Task *task = NULL;
 -        // walk suspended_ccalls to find the correct worker thread
 +        InCall *target_incall = NULL;
          InCall *incall;
          for (incall = cap->suspended_ccalls; incall != NULL; incall = incall->next) {
              if (incall->suspended_tso == target) {
                  task = incall->task;
 +                target_incall = incall;
                  break;
+             }
+         }
@@ -446,6 +459,8 @@ check_target:
              if (!((target->flags & TSO_BLOCKEX) &&
                    ((target->flags & TSO_INTERRUPTIBLE) == 0))) {
                  interruptWorkerTask(task);
 +                // #27113: ticker re-fires interruptOSThread while this is set.
 +                RELEASE_STORE(&target_incall->interrupt_pending, 1);
+             }
              return THROWTO_BLOCKED;
          } else {
@@ -511,6 +526,9 @@ throwToSendMsg (Capability *cap STG_UNUSED,
  // Block a throwTo message on the target TSO's blocked_exceptions
  // queue.  The current Capability must own the target TSO in order to
  // modify the blocked_exceptions queue.
 +//
 +// #27113: cap->lock serialises the queue mutation with the cross-cap revoke
 +// walk in clear_interrupt_pending_if_no_live_throw.
  void
  blockedThrowTo (Capability *cap, StgTSO *target, MessageThrowTo *msg)
+ {
@@ -520,8 +538,14 @@ blockedThrowTo (Capability *cap, StgTSO *target, MessageThrowTo *msg)
      ASSERT(target->cap == cap);
      dirty_TSO(cap,target); // we will modify the blocked_exceptions queue
 +#if defined(THREADED_RTS)
 +    ACQUIRE_LOCK(&cap->lock);
 +#endif
      msg->link = target->blocked_exceptions;
 -    target->blocked_exceptions = msg;
 +    RELEASE_STORE(&target->blocked_exceptions, msg);
 +#if defined(THREADED_RTS)
 +    RELEASE_LOCK(&cap->lock);
 +#endif
+ }
  /* -----------------------------------------------------------------------------
@@ -581,6 +605,8 @@ maybePerformBlockedException (Capability *cap, StgTSO *tso)
          throwToSingleThreaded(cap, msg->target, msg->exception);
          source = msg->source;
 +        // #27113: msg->target == tso; its own resumeThread already cleared
 +        // (or is about to clear) interrupt_pending, so no helper call here.
          doneWithMsgThrowTo(cap, msg);
          tryWakeupThread(cap, source);
          return 1;
@@ -603,6 +629,7 @@ awakenBlockedExceptionQueue (Capability *cap, StgTSO *tso)
          i = lockClosure((StgClosure *)msg);
          if (i != &stg_MSG_NULL_info) {
              source = msg->source;
 +            // #27113: msg->target == tso; its own resumeThread clears the flag.
              doneWithMsgThrowTo(cap, msg);
              tryWakeupThread(cap, source);
          } else {
@@ -664,6 +691,54 @@ removeFromMVarBlockedQueue (StgTSO *tso)
      tso->_link = END_TSO_QUEUE;
+ }
 +#if defined(THREADED_RTS)
 +// #27113: clear an interruptible-FFI target's interrupt_pending flag once
 +// a revoke leaves no live (non-MSG_NULL) entry on its blocked_exceptions
 +// queue.  target->cap is stable while target is suspended in an interruptible
 +// FFI call (no migration off the run queue), so the InCall lives on
 +// target_cap->suspended_ccalls.  target_cap->lock is taken unconditionally:
 +// it serialises us with blockedThrowTo, suspendThread's re-arm, and
 +// resumeThread's unlink — and also with the ticker's TRY_ACQUIRE_LOCK, so
 +// the clear is not raced by a ticker reading the pre-clear flag.
 +static void
 +clear_interrupt_pending_if_no_live_throw (Capability *cap, StgTSO *target)
 +{
 +    if (target == NULL) return;
 +    if (ACQUIRE_LOAD(&target->why_blocked) != BlockedOnCCall_Interruptible)
 +        return;
++
 +    Capability *target_cap = ACQUIRE_LOAD(&target->cap);
 +    ACQUIRE_LOCK(&target_cap->lock);
++
 +    // Find the InCall on target_cap.  If target has already resumed (or
 +    // wasn't really suspended on this cap), do nothing.
 +    InCall *found = NULL;
 +    for (InCall *ic = target_cap->suspended_ccalls; ic != NULL; ic = ic->next) {
 +        if (ic->suspended_tso == target) { found = ic; break; }
 +    }
 +    if (found == NULL) {
 +        RELEASE_LOCK(&target_cap->lock);
 +        return;
 +    }
++
 +    bool live = false;
 +    for (MessageThrowTo *t = ACQUIRE_LOAD(&target->blocked_exceptions);
 +         t != END_BLOCKED_EXCEPTIONS_QUEUE;
 +         t = (MessageThrowTo*)t->link) {
 +        if (RELAXED_LOAD((StgWord*)&t->header.info)
 +            != (StgWord)&stg_MSG_NULL_info) {
 +            live = true; break;
 +        }
 +    }
 +    if (!live) {
 +        RELEASE_STORE(&found->interrupt_pending, 0);
 +    }
++
 +    RELEASE_LOCK(&target_cap->lock);
 +    (void)cap;
 +}
 +#endif
++
  static void
  removeFromQueues(Capability *cap, StgTSO *tso)
+ {
@@ -699,8 +774,17 @@ removeFromQueues(Capability *cap, StgTSO *tso)
        // capabilities.
        // ASSERT(m->header.info == &stg_WHITEHOLE_info);
 +#if defined(THREADED_RTS)
 +      // #27113: snapshot m->target before doneWithMsgThrowTo nulls m.
 +      StgTSO *m_target = m->target;
 +#endif
++
        // unlock and revoke it at the same time
        doneWithMsgThrowTo(cap, m);
++
 +#if defined(THREADED_RTS)
 +      clear_interrupt_pending_if_no_live_throw(cap, m_target);
 +#endif
        break;
+   }

rts/Schedule.c

@@ -2536,6 +2536,30 @@ suspendThread (StgRegTable *reg, bool interruptible)
    suspendTask(cap,task);
    cap->in_haskell = false;
++
 +#if defined(THREADED_RTS)
 +  // #27113: re-arm interrupt_pending if entering an interruptible call with
 +  // a live throw already queued.  Covers under-mask EINTR-retry: throwToMsg
 +  // set the flag, resumeThread cleared it, and we're now re-entering pause.
 +  // Under cap->lock, after suspendTask, so the scan and store are serialised
 +  // with the revoke walk.
 +  if (interruptible
 +      && !((tso->flags & TSO_BLOCKEX) &&
 +           ((tso->flags & TSO_INTERRUPTIBLE) == 0))) {
 +      MessageThrowTo *head = ACQUIRE_LOAD(&tso->blocked_exceptions);
 +      bool live = false;
 +      for (MessageThrowTo *m = head;
 +           m != END_BLOCKED_EXCEPTIONS_QUEUE;
 +           m = (MessageThrowTo*)m->link) {
 +          if (RELAXED_LOAD((StgWord*)&m->header.info)
 +              != (StgWord)&stg_MSG_NULL_info) { live = true; break; }
 +      }
 +      if (live) {
 +          RELEASE_STORE(&task->incall->interrupt_pending, 1);
 +      }
 +  }
 +#endif
++
    releaseCapability_(cap,false);
    RELEASE_LOCK(&cap->lock);
@@ -2574,8 +2598,14 @@ resumeThread (void *task_)
      // entry on the suspended_ccalls list will also have been
      // migrated.
 -    // Remove the thread from the suspended list
 +    // #27113: clear interrupt_pending and unlink under cap->lock so the
 +    // ticker's TRY_ACQUIRE_LOCK serialises with us.  Without the lock, the
 +    // ticker can read a stale flag and SIGPIPE the worker after it has
 +    // already returned from the FFI call onto an unrelated syscall.
 +    ACQUIRE_LOCK(&cap->lock);
 +    RELEASE_STORE(&incall->interrupt_pending, 0);
      recoverSuspendedTask(cap,task);
 +    RELEASE_LOCK(&cap->lock);
      tso = incall->suspended_tso;
      incall->suspended_tso = NULL;
@@ -2614,6 +2644,49 @@ resumeThread (void *task_)
      return &cap->r;
+ }
 +#if defined(THREADED_RTS)
 +// #27113: re-fire interruptOSThread for InCalls whose interrupt_pending is
 +// still set.  Reads only InCall fields, never TSO heap state, so no GC race.
 +//
 +// cap->lock is held across the OS interrupt call.  This pairs with the
 +// same lock taken by resumeThread around its clear+unlink: when we fire
 +// interruptOSThread, resumeThread is either blocked behind us on the lock
 +// (signal lands harmlessly on a thread inside pthread sync) or has not yet
 +// run (signal lands inside the FFI call where it belongs).  Releasing the
 +// lock before signalling would let resumeThread complete and re-enter
 +// Haskell first, so SIGPIPE / CancelSynchronousIo would hit an unrelated
 +// later syscall.
 +//
 +// The per-OS-call cost is small (~µs for pthread_kill on POSIX, ~tens of
 +// µs for OpenThread+CancelSynchronousIo+CloseHandle on Windows), but the
 +// per-tick lock-hold time scales linearly with the number of pending
 +// interruptible calls on this cap.  For typical workloads this is well
 +// below the tick interval; FFI-heavy workloads with many concurrent
 +// interruptible calls per cap may want empirical measurement.
 +void
 +retryInterruptibleSignals(void)
 +{
 +    bool any_pending = false;
 +    uint32_t n = getNumCapabilities();
 +    for (uint32_t i = 0; i < n; i++) {
 +        Capability *cap = getCapability(i);
 +        if (RELAXED_LOAD(&cap->n_suspended_ccalls) == 0) continue;
 +        if (TRY_ACQUIRE_LOCK(&cap->lock) != 0) continue;
 +        for (InCall *ic = cap->suspended_ccalls; ic != NULL; ic = ic->next) {
 +            if (ACQUIRE_LOAD(&ic->interrupt_pending) == 0) continue;
 +            // ic->task is set by newInCall and never cleared.
 +            if (ic->task != NULL) {
 +                any_pending = true;
 +                interruptOSThread(ic->task->id);
 +            }
 +        }
 +        RELEASE_LOCK(&cap->lock);
 +    }
 +    // Keep the timer alive; handle_tick's idle path would otherwise stop it.
 +    if (any_pending) setRecentActivity(ACTIVITY_YES);
 +}
 +#endif
++
  /* ---------------------------------------------------------------------------
   * scheduleThread()
+  *

rts/Schedule.h

@@ -43,6 +43,10 @@ void scheduleThreadOn(Capability *cap, StgWord cpu, StgTSO *tso);
  void wakeUpRts(void);
  #endif
 +#if defined(THREADED_RTS)
 +void retryInterruptibleSignals(void);
 +#endif
++
  /* raiseExceptionHelper */
  StgWord raiseExceptionHelper (StgRegTable *reg, StgTSO *tso, StgClosure *exception);

rts/Task.c

@@ -267,6 +267,7 @@ newInCall (Task *task)
      incall->next = NULL;
      incall->prev = NULL;
      incall->prev_stack = task->incall;
 +    incall->interrupt_pending = 0;
      task->incall = incall;
+ }

rts/Task.h

@@ -106,6 +106,15 @@ typedef struct InCall_ {
      // Links InCalls onto suspended_ccalls, spare_incalls
      struct InCall_ *prev;
      struct InCall_ *next;
++
 +    // #27113.  Set (atomically) by throwToMsg once it has fired
 +    // interruptOSThread for this call's worker; cleared by resumeThread
 +    // (when the call returns) and by the revoke paths in
 +    // doneWithMsgThrowTo (when all queued throws against the target are
 +    // MSG_NULL).  The ticker reads this and re-fires interruptOSThread
 +    // while it is set.  Not GC-managed — lives in malloc'd memory with
 +    // the Task that owns this InCall.
 +    StgWord interrupt_pending;
  } InCall;
  typedef struct Task_ {

rts/Timer.c

@@ -131,6 +131,11 @@ handle_tick(int unused STG_UNUSED)
            flushEventLog(NULL);
+       }
+   }
 +  // #27113 retry; gated on timer_disabled like the context-switch path
 +  // above so that setNumCapabilities's resize pause is respected (#17289).
 +  if (SEQ_CST_LOAD_ALWAYS(&timer_disabled) == 0) {
 +      retryInterruptibleSignals();
 +  }
  #endif
    /*

testsuite/tests/concurrent/should_run/T27113.hs

@@ -10,7 +10,7 @@ import System.IO
  import System.Timeout
  foreign import ccall interruptible "cpu_then_pause"
 -    c_cpu_then_pause :: CLong -> IO ()
 +    c_cpu_then_pause :: CLLong -> IO ()
  killDelay :: Int
  killDelay = 200_000
@@ -18,7 +18,7 @@ killDelay = 200_000
  joinTimeout :: Int
  joinTimeout = 3_000_000
 -runCase :: String -> CLong -> IO ()
 +runCase :: String -> CLLong -> IO ()
  runCase name spinNs = do
      tid <- forkIO $
          c_cpu_then_pause spinNs

testsuite/tests/concurrent/should_run/T27113_c.c

  #include <errno.h>
 +#include <stdint.h>
  #include <time.h>
  #include <unistd.h>
 -static long now_ns(void) {
 +static int64_t now_ns(void) {
      struct timespec t;
      while (clock_gettime(CLOCK_MONOTONIC, &t) != 0) {
          if (errno != EINTR) return 0;
+     }
 -    return (long)t.tv_sec * 1000000000L + (long)t.tv_nsec;
 +    return (int64_t)t.tv_sec * 1000000000LL + (int64_t)t.tv_nsec;
+ }
 -int cpu_then_pause(long spin_ns) {
 -    long deadline = now_ns() + spin_ns;
 +int cpu_then_pause(int64_t spin_ns) {
 +    int64_t deadline = now_ns() + spin_ns;
      while (now_ns() < deadline) {
+     }
      return pause();

testsuite/tests/concurrent/should_run/T27113b.hs

 +{-# LANGUAGE InterruptibleFFI #-}
 +{-# LANGUAGE ScopedTypeVariables #-}
++
 +module Main where
++
 +import Control.Concurrent
 +import Control.Exception
 +import Foreign.C.Types
 +import System.IO
 +import System.Timeout
++
 +foreign import ccall interruptible "cpu_then_pause"
 +    c_cpu_then_pause :: CLLong -> IO ()
++
 +-- Recommended pattern: under mask, with allowInterrupt as the explicit
 +-- interruption point.  After a SIGPIPE-driven pause EINTR, control
 +-- returns to Haskell still under mask, so the queued throw isn't
 +-- delivered until allowInterrupt.  For the kill to complete, the SIGPIPE
 +-- retry must fire when the initial SIGPIPE is lost in the user-space
 +-- prologue (the spin loop) — otherwise pause blocks forever and
 +-- killThread hangs.
 +myCall :: CLLong -> IO ()
 +myCall spinNs = mask_ $ do
 +    allowInterrupt
 +    c_cpu_then_pause spinNs
 +    allowInterrupt
++
 +main :: IO ()
 +main = do
 +    hSetBuffering stdout NoBuffering
 +    started <- newEmptyMVar
 +    done <- newEmptyMVar
 +    -- mask_ around the forkIOWithUnmask ensures the child inherits
 +    -- MaskedInterruptible, so finally is set up before our killThread
 +    -- below can be delivered.
 +    tid <- mask_ $ forkIOWithUnmask $ \unmask ->
 +        flip finally (putMVar done ()) $ do
 +            putMVar started ()
 +            unmask (myCall 2_000_000_000)
 +                `catch` (\(_ :: SomeException) -> return ())
 +    takeMVar started
 +    threadDelay 200_000
 +    result <- timeout 5_000_000 (killThread tid >> takeMVar done)
 +    case result of
 +        Just () -> putStrLn "ok"
 +        Nothing -> do
 +            hPutStrLn stderr "FAIL: killThread hung"
 +            error "hung"

testsuite/tests/concurrent/should_run/T27113c.hs

 +{-# LANGUAGE InterruptibleFFI #-}
 +{-# LANGUAGE ScopedTypeVariables #-}
++
 +module Main where
++
 +import Control.Concurrent
 +import Control.Exception
 +import Control.Monad
 +import Data.IORef
 +import Foreign.C.Types
 +import System.Exit
 +import System.IO
++
 +foreign import ccall interruptible "pause" c_pause :: IO CInt
++
 +postRevokeWindow :: Int
 +postRevokeWindow = 2_000_000
++
 +postRevokeThreshold :: Int
 +postRevokeThreshold = 5
++
 +main :: IO ()
 +main = do
 +    hSetBuffering stdout NoBuffering
 +    iters <- newIORef (0 :: Int)
++
 +    workerStarted <- newEmptyMVar
 +    worker <- forkIO $ mask_ $ do
 +        putMVar workerStarted ()
 +        (forever $ do
 +            atomicModifyIORef' iters (\n -> (n + 1, ()))
 +            _ <- c_pause
 +            return ())
 +          `catch` (\(_ :: SomeException) -> return ())
++
 +    takeMVar workerStarted
 +    threadDelay 100_000
++
 +    killer1Ready <- newEmptyMVar
 +    killer1 <- forkIO $ do
 +        putMVar killer1Ready ()
 +        killThread worker
++
 +    takeMVar killer1Ready
 +    threadDelay 100_000
++
 +    preRevoke <- readIORef iters
++
 +    killThread killer1
++
 +    threadDelay postRevokeWindow
++
 +    postRevoke <- readIORef iters
 +    let storm = postRevoke - preRevoke
 +    if storm > postRevokeThreshold
 +        then do
 +            hPutStrLn stderr ("FAIL: post-revoke EINTR storm, iterations="
 +                              ++ show storm
 +                              ++ " (pre=" ++ show preRevoke
 +                              ++ " post=" ++ show postRevoke ++ ")")
 +            exitFailure
 +        else
 +            putStrLn ("ok (post-revoke iterations=" ++ show storm ++ ")")

testsuite/tests/concurrent/should_run/all.T

@@ -208,11 +208,24 @@ test('foreignInterruptible', [when(fast(), skip),
  test('T27113',
       [when(opsys('mingw32'), skip),
 +      only_threaded_ways,
 +      req_c],
 +     compile_and_run, ['T27113_c.c'])
++
 +test('T27113b',
 +     [extra_files(['T27113_c.c']),
 +      when(opsys('mingw32'), skip),
        only_threaded_ways,
        req_c,
 -      expect_broken(27113)],
 +      ignore_stdout],
       compile_and_run, ['T27113_c.c'])
 +test('T27113c',
 +     [when(opsys('mingw32'), skip),
 +      only_threaded_ways,
 +      ignore_stdout],
 +     compile_and_run, [''])
++
  test('conc037', only_ways(['threaded1', 'threaded2', 'nonmoving_thr']), compile_and_run, [''])
  test('conc038', only_ways(['threaded1', 'threaded2', 'nonmoving_thr']), compile_and_run, [''])