pqi-native 1.0.1.2 → 1.0.1.3
raw patch · 4 files changed
+75/−36 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 +12/−0
- pqi-native.cabal +2/−2
- src/library/Pqi/Native/Query.hs +13/−1
- src/transport/Pqi/Native/Transport.hs +48/−33
CHANGELOG.md view
@@ -1,3 +1,15 @@+# v1.0.1.3++## Fixes++- Fixed an async exception around an aborted pipeline leaving a connection permanently stuck, with no timer able to reclaim it. Two changes, which are only a fix together:++ - `Query.getNextResult` now runs `mask_`ed. The connection's result bookkeeping - the pending-command counter, the separator flag, the `ParseComplete` origin FIFO - lives in separate `IORef`s that one logical transition updates in sequence. An interrupt landing between two of those updates left them inconsistent, and an inconsistent pair sends the next `getNextResult` off to wait for a message the backend has already decided not to send.++ - `Transport.receiveFrame` no longer runs `uninterruptibleMask_`ed. It buffers a whole frame before consuming any of it and takes it out of the buffer in one atomic step, so the framing 1.0.1.2 set out to protect stays intact - but the blocking wait is masked only across moving bytes off the socket, not across waiting for them. The 1.0.1.2 shape made every wait unabandonable, which is what turned the stall above into a deadlock `System.Timeout.timeout` could not break.++ Found via a hang in `hasql`'s `Integration.Sharing.Connection.Use.PipelineAbortedInterruptionCleanup`, which wedged only under concurrent load and only on this adapter.+ # v1.0.1.2 ## Fixes
pqi-native.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: pqi-native-version: 1.0.1.2+version: 1.0.1.3 category: Database, PostgreSQL synopsis: Native (pure-Haskell) adapter for pqi description:@@ -158,5 +158,5 @@ build-depends: base >=4.11 && <5, hspec >=2.11 && <2.12,- pqi-conformance ^>=1.0.3.0,+ pqi-conformance ^>=1.0.4, pqi-native,
src/library/Pqi/Native/Query.hs view
@@ -17,6 +17,7 @@ ) where +import Control.Exception (mask_) import qualified Data.Map.Strict as Map import qualified Data.Sequence as Seq import Pqi (ConnStatus (..), ExecStatus (..), Format (..), PipelineStatus (..))@@ -168,8 +169,19 @@ -- result set, and a 'PipelineSync' result is returned for each @Sync@ -- boundary. In single-row mode each data row is delivered as a separate -- 'SingleTuple' result followed by a final 'TuplesOk' with no rows.+--+-- Runs 'mask_'ed. The connection's result bookkeeping - the pending-command+-- counter, the separator flag, the @ParseComplete@ origin FIFO, the pipeline+-- status - lives in separate 'IORef's that a single logical transition+-- updates one after another. An async exception landing between two of those+-- updates leaves the pair inconsistent, and an inconsistent pair is not merely+-- wrong: it sends the next 'getNextResult' down the @readAndProcess@ path to+-- wait for a message the backend has already decided not to send, which is a+-- stall with no timer behind it. Masking keeps each transition atomic. The+-- blocking read inside stays interruptible (see 'Transport.receiveFrame'), so+-- this costs no abandonability. getNextResult :: Connection -> IO (Maybe NativeResult)-getNextResult connection = do+getNextResult connection = mask_ do pending <- readIORef (asyncPending connection) if not pending then pure Nothing
src/transport/Pqi/Native/Transport.hs view
@@ -14,7 +14,7 @@ ) where -import Control.Exception (uninterruptibleMask_)+import Control.Exception (mask_) import qualified Data.ByteString as ByteString import qualified Data.ByteString.Char8 as ByteString.Char8 import Data.IORef@@ -64,45 +64,60 @@ send :: Transport -> Poker.Write -> IO () send transport write = Socket.ByteString.sendAll (socket transport) (Poker.toByteString write) --- | Read exactly @n@ bytes, looping over @recv@ (which yields up to @n@) and--- buffering any overshoot. Throws on EOF before @n@ bytes arrive.-receiveExactly :: Transport -> Int -> IO ByteString-receiveExactly transport n = do- buffered <- readIORef (readBuffer transport)- go buffered+-- | Ensure the read buffer holds at least @n@ bytes, pulling from the socket+-- as needed. Throws on EOF before @n@ bytes are available.+--+-- The wait for bytes is deliberately left interruptible. Nothing has been+-- consumed at this point, so a caller that gives up here loses nothing - and,+-- crucially, is /able/ to give up. A caller blocked on a message the server+-- will never send (an aborted pipeline whose bookkeeping has drifted, say)+-- must stay abandonable by 'System.Timeout.timeout'; masking the wait+-- uninterruptibly turns that stall into a deadlock no timer can break.+--+-- Only the step that moves bytes off the socket and into the buffer is masked,+-- which is all the atomicity the framing needs: an async exception can never+-- land in the gap between @recv@ returning and its bytes being recorded, so+-- bytes are never dropped. @recv@ itself stays interruptible inside 'mask_',+-- so the blocking wait keeps its abandonability.+fillTo :: Transport -> Int -> IO ()+fillTo transport n = go where- go accumulated- | ByteString.length accumulated >= n = do- let (result, rest) = ByteString.splitAt n accumulated- writeIORef (readBuffer transport) rest- pure result- | otherwise = do- chunk <- Socket.ByteString.recv (socket transport) (max 4096 (n - ByteString.length accumulated))+ go = do+ buffered <- readIORef (readBuffer transport)+ let missing = n - ByteString.length buffered+ when (missing > 0) do+ closed <- mask_ do+ chunk <- Socket.ByteString.recv (socket transport) (max 4096 missing) if ByteString.null chunk- then ioError (mkIOError eofErrorType "pqi-native: connection closed by server" Nothing Nothing)- else go (accumulated <> chunk)+ then pure True+ else do+ modifyIORef' (readBuffer transport) (<> chunk)+ pure False+ when closed do+ ioError (mkIOError eofErrorType "pqi-native: connection closed by server" Nothing Nothing)+ go -- | Receive one framed message: its type byte and its body (the length prefix, -- which counts itself, is consumed). ----- Runs fully 'uninterruptibleMask_'ed: 'Socket.ByteString.recv' is--- interruptible even under a plain 'mask', so an async exception (e.g. from--- 'System.Timeout.timeout') landing between a @recv@ returning and its bytes--- being folded into 'readBuffer' would otherwise drop them - desyncing the--- connection's framing for every subsequent read. Deferring delivery across--- the whole frame mirrors how @pqi-ffi@'s @safe@ FFI call into @libpq@ is--- immune to the same hazard.+-- Buffering the whole frame before consuming any of it keeps the framing+-- atomic without masking the wait: an async exception landing while the frame+-- is still incomplete leaves the buffer untouched, and the frame is taken out+-- of the buffer in a single 'atomicModifyIORef'' step. The earlier shape -+-- consuming the header, then blocking again for the body - is what made a+-- mid-read interrupt desync the connection, and what an outer+-- 'Control.Exception.uninterruptibleMask_' was papering over at the cost of+-- making every stall permanent. receiveFrame :: Transport -> IO (Word8, ByteString)-receiveFrame transport = uninterruptibleMask_ do- header <- receiveExactly transport 5- let typeByte = ByteString.head header- frameLength = decodeInt32BE (ByteString.drop 1 header)- bodyLength = frameLength - 4- body <-- if bodyLength > 0- then receiveExactly transport bodyLength- else pure ByteString.empty- pure (typeByte, body)+receiveFrame transport = do+ fillTo transport 5+ header <- ByteString.take 5 <$> readIORef (readBuffer transport)+ let frameLength = decodeInt32BE (ByteString.drop 1 header)+ frameSize = 5 + max 0 (frameLength - 4)+ fillTo transport frameSize+ atomicModifyIORef' (readBuffer transport) \buffered ->+ let (frame, rest) = ByteString.splitAt frameSize buffered+ in (rest, (ByteString.head frame, ByteString.drop 5 frame)) -- | Read bytes until the peer closes the connection (EOF), discarding them. -- Used by the cancel path to mirror libpq's behaviour: keep the socket open