pqi-native 1.0.1.16 → 1.0.1.17

raw patch · 4 files changed

+100/−5 lines, 4 filesdep ~pqi-conformancePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: pqi-conformance

API changes (from Hackage documentation)

Files

• CHANGELOG.md +6/−0
• pqi-native.cabal +16/−2
• src/test/NonThreadedSpec.hs +15/−0
• src/transport/Pqi/Native/Transport.hs +63/−3

@@ -1,3 +1,9 @@+# v1.0.1.17++## Fixes++- Fixed a query sent under `uninterruptibleMask` hanging forever on the non-threaded RTS ([#17](https://github.com/nikita-volkov/pqi-native/issues/17)). `Transport.awaitWritableOrReadable`, the read-or-write readiness wait of the send loop, armed both waits with base's `threadWaitReadSTM`/`threadWaitWriteSTM`. On the non-threaded RTS those fork a helper thread with a bare `forkIO` and cancel it with `killThread`, and a forked thread inherits the caller's masking state: from under an `uninterruptibleMask` - such as a `safe-exceptions` `bracket`'s release action - the cancellation of the readable wait could never be delivered while the socket stayed writable-but-not-readable, the steady state of an idle connection being sent a query, so the send, and with it the whole query, never reached the server. libpq's send loop (`pqSendSome` in `fe-misc.c`) waits with a plain `select` and has no masking state to inherit. On the non-threaded RTS the wait helper is now forked with `forkIOWithUnmask`, whose body runs unmasked whatever the caller's state (it is `forkIO (io unsafeUnmask)`), so the cancellation always delivers; on the threaded RTS the waits remain base's event-manager registrations, which have no helper thread. Caught by the `pqi-conformance` spec `Pqi.Conformance.Operation.Exec.UninterruptibleMask` (`pqi-conformance` 1.0.18.0), run by the new `native-test-nonthreaded` test suite: the suite must be built without `-threaded` for the divergence to manifest.+ # v1.0.1.16  ## Fixes

@@ -1,6 +1,6 @@ cabal-version: 3.0 name: pqi-native-version: 1.0.1.16+version: 1.0.1.17 category: Database, PostgreSQL synopsis: Native (pure-Haskell) adapter for pqi description:@@ -160,5 +160,19 @@   build-depends:     base >=4.11 && <5,     hspec >=2.11 && <2.12,-    pqi-conformance ^>=1.0.17,+    pqi-conformance ^>=1.0.18,+    pqi-native,++-- The conformance specs that only reproduce on the non-threaded RTS+-- ('Pqi.Conformance.nonThreadedSpecs'): on the threaded RTS they pass+-- vacuously. Built without @-threaded@, unlike 'native-test'.+test-suite native-test-nonthreaded+  import: base+  type: exitcode-stdio-1.0+  hs-source-dirs: src/test+  main-is: NonThreadedSpec.hs+  build-depends:+    base >=4.11 && <5,+    hspec >=2.11 && <2.12,+    pqi-conformance ^>=1.0.18,     pqi-native,

@@ -0,0 +1,15 @@+-- | The conformance specs that only reproduce on the non-threaded RTS: a+-- delegate to 'Pqi.Conformance.nonThreadedSpecs', which brings up a+-- throwaway PostgreSQL container and runs the same differential comparisons+-- as the main battery ('Spec'), but on the RTS without the threaded+-- scheduler, which is the configuration some divergences live in. Built+-- without @-threaded@.+module Main (main) where++import Pqi.Conformance (nonThreadedSpecs)+import qualified Pqi.Native+import Prelude+import Test.Hspec++main :: IO ()+main = hspec (nonThreadedSpecs Pqi.Native.adapter)

@@ -27,7 +27,21 @@ import Foreign.Marshal.Utils (copyBytes) import Foreign.Ptr (plusPtr) #if !defined(mingw32_HOST_OS)-import GHC.Conc (atomically, orElse, threadWaitReadSTM, threadWaitWriteSTM)+import Control.Concurrent (forkIOWithUnmask, killThread, rtsSupportsBoundThreads)+import GHC.Conc+  ( STM,+    atomically,+    newTVarIO,+    orElse,+    readTVar,+    retry,+    threadWaitRead,+    threadWaitReadSTM,+    threadWaitWrite,+    threadWaitWriteSTM,+    writeTVar,+  )+import System.Posix.Types (Fd) #endif import qualified Network.Socket as Socket import qualified Network.Socket.ByteString as Socket.ByteString@@ -229,12 +243,58 @@ awaitWritableOrReadable :: Socket.Socket -> IO Bool awaitWritableOrReadable sock = do   fd <- Socket.unsafeFdSocket sock-  (writable, cancelWritable) <- threadWaitWriteSTM (fromIntegral fd)-  (readable, cancelReadable) <- threadWaitReadSTM (fromIntegral fd)+  (writable, cancelWritable) <- awaitWritableSTM (fromIntegral fd)+  (readable, cancelReadable) <- awaitReadableSTM (fromIntegral fd)   outcome <- atomically (fmap (\_ -> True) writable `orElse` fmap (\_ -> False) readable)   cancelReadable   cancelWritable   pure outcome++-- | The socket's writable wait with its cancellation: base's+-- 'threadWaitWriteSTM', whose threaded-RTS implementation is an+-- event-manager registration, on the threaded RTS; 'forkedWaitSTM' on the+-- non-threaded one.+awaitWritableSTM :: Fd -> IO (STM (), IO ())+awaitWritableSTM fd+  | rtsSupportsBoundThreads = threadWaitWriteSTM fd+  | otherwise = forkedWaitSTM threadWaitWrite fd++-- | The socket's readable wait with its cancellation, as 'awaitWritableSTM'.+awaitReadableSTM :: Fd -> IO (STM (), IO ())+awaitReadableSTM fd+  | rtsSupportsBoundThreads = threadWaitReadSTM fd+  | otherwise = forkedWaitSTM threadWaitRead fd++-- | Arm one fd-readiness wait on the non-threaded RTS, as @(wait, cancel)@:+-- what base's non-threaded 'threadWaitReadSTM' does, except that the helper+-- thread is forked with 'forkIOWithUnmask'.+--+-- The difference matters because base forks its helper with a bare+-- 'forkIO', and a forked thread inherits the caller's masking state: a+-- caller inside 'Control.Exception.uninterruptibleMask' - such as a+-- @safe-exceptions@ @bracket@'s release action - could then never deliver+-- the 'killThread' of the cancellation to a helper blocked on a wait that+-- never fires, and the send would hang forever: a query sent under the mask+-- never even reached the server+-- (<https://github.com/nikita-volkov/pqi-native/issues/17>). The body of+-- @forkIOWithUnmask@ runs unmasked whatever the caller's state (it is+-- @forkIO (io unsafeUnmask)@), so the kill always delivers, and the caller+-- is never blocked in it for longer than the fd needs to report readiness+-- or the helper to die. On the threaded RTS base's implementation arms the+-- wait through the event manager and has no helper thread at all, which is+-- why only the non-threaded one is replaced.+--+-- The race between the helper's write and the kill is harmless: whichever+-- wins, the write goes to a 'TVar' nobody reads anymore, and a 'killThread'+-- of an already-finished helper is a no-op.+forkedWaitSTM :: (Fd -> IO ()) -> Fd -> IO (STM (), IO ())+forkedWaitSTM wait fd = do+  ready <- newTVarIO False+  waiter <- forkIOWithUnmask \unmask -> unmask (wait fd) >> atomically (writeTVar ready True)+  pure+    ( readTVar ready >>= \flag -> if flag then pure () else retry,+      killThread waiter+    )  -- | Absorb one chunk of incoming data into the read buffer - the analogue -- of libpq calling @pqReadData()@ inside its send loop. Only called once