pqi-native 1.0.1.11 → 1.0.1.12
raw patch · 7 files changed
+241/−59 lines, 7 filesdep +vectordep ~pqi-conformancePVP ok
version bump matches the API change (PVP)
Dependencies added: vector
Dependency ranges changed: pqi-conformance
API changes (from Hackage documentation)
Files
- CHANGELOG.md +6/−0
- pqi-native.cabal +4/−2
- src/comms/Pqi/Native/Comms.hs +15/−0
- src/library/Pqi/Native.hs +1/−1
- src/library/Pqi/Native/Query.hs +122/−41
- src/transport/Pqi/Native/Transport.hs +75/−0
- src/transport/Pqi/Native/Transport/Message.hs +18/−15
CHANGELOG.md view
@@ -1,3 +1,9 @@+# v1.0.1.12++## Fixes++- Fixed a pipeline flow-control deadlock: a pipelined burst of commands larger than the socket buffers along the path can absorb, submitted before any result is read, used to wedge both sides permanently. `Pqi.Native.Transport.send` was a bare blocking `Network.Socket.ByteString.sendAll`, which never reads while sending; once the server has filled its output path with unread results it blocks in `ClientWrite` and stops reading the client's commands, so the client - still blocked in `send` with unsent bytes - can never finish, and neither side progresses again. libpq survives the same situation because its `pqSendSome` (`fe-misc.c`), on an incomplete send, calls `pqReadData()` to absorb incoming data and then `pqWait(true, true, …)` for read-or-write readiness, looping until its output is fully sent. `send` now mirrors that loop: whenever the socket is not writable, incoming bytes are drained into the transport's read buffer and the wait is armed for read-or-write readiness (write-biased, so a streaming server cannot starve the send); a peer closing mid-send surfaces as the same classified EOF the read side reports. On Windows the plain `sendAll` remains, as GHC's I/O manager cannot back the readiness wait there. Surfaced as hasql's `manyLargeResultsViaPipeline` benchmark hanging from 1.0.1.4 onward, after the default host resolution moved to the Unix-domain socket whose 8KB buffers expose what TCP's larger ones absorbed. Caught by the `pqi-conformance` spec `Pqi.Conformance.Operation.SendQueryParams.PipelineFlowControl`.+ # v1.0.1.11 ## Fixes
pqi-native.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: pqi-native-version: 1.0.1.11+version: 1.0.1.12 category: Database, PostgreSQL synopsis: Native (pure-Haskell) adapter for pqi description:@@ -115,6 +115,7 @@ bytestring >=0.10 && <0.13, network >=3.1 && <3.3, pqi-native:comms,+ ptr-peeker ^>=0.2, ptr-poker ^>=0.1.3, transformers >=0.5 && <0.7, @@ -143,6 +144,7 @@ pqi-native:transport, ptr-peeker ^>=0.2, ptr-poker ^>=0.1.3,+ vector >=0.12 && <0.14, if !os(windows) build-depends: unix >=2.7 && <2.9@@ -158,5 +160,5 @@ build-depends: base >=4.11 && <5, hspec >=2.11 && <2.12,- pqi-conformance ^>=1.0.11,+ pqi-conformance ^>=1.0.12, pqi-native,
src/comms/Pqi/Native/Comms.hs view
@@ -18,6 +18,7 @@ cstring, bytes, remaining,+ lengthPrefixedBytes, -- * Serializable primitives CString (..),@@ -95,6 +96,20 @@ -- | All remaining bytes of the body. remaining :: Decoder ByteString remaining = liftVariable Peeker.remainderAsByteString++-- | An @Int32@-length-prefixed byte string, with a negative length meaning+-- \"absent\" (the wire protocol's encoding for SQL @NULL@ column values).+--+-- Decoded as a single 'liftVariable' call rather than a separate 'int32'+-- (length) lift followed by a 'bytes' lift: this is the per-column-value+-- decoder, called for every cell of every row, so halving the lift count+-- here has an outsized effect on hot-path allocation.+lengthPrefixedBytes :: Decoder (Maybe ByteString)+lengthPrefixedBytes = liftVariable $ do+ len <- Peeker.fixed Peeker.beSignedInt4+ if len < 0+ then pure Nothing+ else Just <$> Peeker.fixed (Peeker.byteArrayAsByteString (fromIntegral len)) -- | A null-terminated string, as a 'Comms' building block. newtype CString = CString ByteString
src/library/Pqi/Native.hs view
@@ -9,7 +9,7 @@ ) where -import Control.Exception (IOException, catch)+import Control.Exception (catch) import qualified Data.ByteString as ByteString import qualified Data.ByteString.Char8 as ByteString.Char8 import qualified Data.Map.Strict as Map
src/library/Pqi/Native/Query.hs view
@@ -20,6 +20,7 @@ import Control.Exception (IOException, catch, mask_) import qualified Data.Map.Strict as Map import qualified Data.Sequence as Seq+import qualified Data.Vector.Mutable as MVector import Pqi (ConnStatus (..), ExecStatus (..), Format (..), PipelineStatus (..)) import Pqi.Native.Connection import Pqi.Native.Prelude@@ -245,10 +246,11 @@ else do singleRow <- readIORef (singleRowMode connection) cachedFields <- readIORef (singleRowFields connection)+ builder <- newBuilder let initBuilder = if singleRow && not (null cachedFields)- then emptyBuilder {accFields = cachedFields, accSawRowDescription = True}- else emptyBuilder+ then builder {accFields = cachedFields, accSawRowDescription = True}+ else builder go singleRow initBuilder where -- Decrement the pending-command counter and set the separator flag when in@@ -299,7 +301,9 @@ then do writeIORef (singleRowFields connection) (accFields builder) pure (Just (NativeResult SingleTuple (accFields builder) [values] Nothing Map.empty [] ""))- else go singleRow builder {accRevRows = values : (accRevRows builder)}+ else do+ pushRow (accRows builder) values+ go singleRow builder ParseComplete -> do -- Charge the @ParseComplete@ to the command that produced it by -- popping the oldest recorded origin. Only a 'sendPrepare' origin@@ -332,7 +336,7 @@ writeIORef (lastError connection) (Just "") popOriginIfPending pipeStatus builder finishCommand pipeStatus- pure (Just (commandResult builder (Just tag)))+ Just <$> commandResult builder (Just tag) EmptyQueryResponse -> do writeIORef (lastError connection) (Just "") popOriginIfPending pipeStatus builder@@ -357,14 +361,14 @@ writeIORef (lastError connection) (Just (formatResultError sql errMap)) pure (Just (NativeResult FatalError [] [] Nothing errMap [] sql)) PortalSuspended ->- pure (Just (commandResult builder Nothing))+ Just <$> commandResult builder Nothing ReadyForQuery txState -> do writeIORef (txStatus connection) txState case pipeStatus of PipelineOff -> do writeIORef (asyncPending connection) False if (accHadResponse builder)- then pure (Just (describeResult builder))+ then Just <$> describeResult builder else pure Nothing _ -> do writeIORef (pipelineStatus connection) PipelineOn@@ -452,10 +456,58 @@ message <- markConnectionLost connection err pure (NativeResult FatalError [] [] Nothing (Map.singleton 0x4d message) [] sql) +-- | A growable row buffer, replacing a cons-list-plus-reverse accumulation+-- with amortized-O(1) appends into a doubling mutable vector. A statement's+-- rows are pushed one 'DataRow' at a time and can number in the hundreds of+-- thousands; accumulating them as @row : rows@ and reversing at the end (the+-- prior representation) keeps two full-length cons-cell spines alive across+-- the whole statement, which dominates GC copy volume at scale (see+-- 'hasql-pqi-native-decode-gc-bound' memory). This buffer holds only the+-- array (which the RTS moves as one contiguous block, unlike a linked list)+-- until 'freezeRowBuffer' walks it once into the final list.+data RowBuffer+ = RowBuffer+ -- | The mutable vector of rows.+ !(IORef (MVector.IOVector [Maybe ByteString]))+ -- | Length.+ !(IORef Int)++newRowBuffer :: IO RowBuffer+newRowBuffer = do+ v <- MVector.new 64+ RowBuffer <$> newIORef v <*> newIORef 0++pushRow :: RowBuffer -> [Maybe ByteString] -> IO ()+pushRow (RowBuffer vRef lenRef) row = do+ v <- readIORef vRef+ len <- readIORef lenRef+ v' <-+ if len >= MVector.length v+ then do+ grown <- MVector.grow v (MVector.length v)+ writeIORef vRef grown+ pure grown+ else pure v+ MVector.write v' len row+ writeIORef lenRef (len + 1)++-- | Walk the buffer back-to-front once, consing onto an accumulator, which+-- yields the rows in their original order without a separate reverse pass.+freezeRowBuffer :: RowBuffer -> IO [[Maybe ByteString]]+freezeRowBuffer (RowBuffer vRef lenRef) = do+ v <- readIORef vRef+ len <- readIORef lenRef+ let go !i acc+ | i < 0 = pure acc+ | otherwise = do+ x <- MVector.read v i+ go (i - 1) (x : acc)+ go (len - 1) []+ -- accumulator for a result under construction data Builder = Builder { accFields :: [FieldDescription],- accRevRows :: [[Maybe ByteString]],+ accRows :: RowBuffer, accParamOids :: [Word32], accSawRowDescription :: Bool, accHadResponse :: Bool,@@ -464,31 +516,41 @@ accOriginPopped :: Bool } -emptyBuilder :: Builder-emptyBuilder = Builder [] [] [] False False False+newBuilder :: IO Builder+newBuilder = do+ rows <- newRowBuffer+ pure (Builder [] rows [] False False False) -- | Collect the (possibly several) results of a simple query, up to -- @ReadyForQuery@. The last is what @PQexec@ returns. -- @CopyInResponse@ and @CopyOutResponse@ terminate the loop immediately, -- returning a synthetic result so the caller can enter the copy sub-protocol. collectSimple :: Connection -> ByteString -> IO [NativeResult]-collectSimple connection sql = go emptyBuilder []+collectSimple connection sql = do+ builder0 <- newBuilder+ go builder0 [] where go builder acc = do message <- nextMessage connection case message of RowDescription fs -> go builder {accFields = fs, accSawRowDescription = True} acc- DataRow values -> go builder {accRevRows = values : (accRevRows builder)} acc+ DataRow values -> do+ pushRow (accRows builder) values+ go builder acc CommandComplete tag -> do writeIORef (lastError connection) (Just "")- go emptyBuilder (commandResult builder (Just tag) : acc)+ result <- commandResult builder (Just tag)+ builder' <- newBuilder+ go builder' (result : acc) EmptyQueryResponse -> do writeIORef (lastError connection) (Just "")- go emptyBuilder (NativeResult EmptyQuery [] [] Nothing Map.empty [] "" : acc)+ builder' <- newBuilder+ go builder' (NativeResult EmptyQuery [] [] Nothing Map.empty [] "" : acc) ErrorResponse fs -> do let errMap = Map.fromList fs writeIORef (lastError connection) (Just (formatResultError sql errMap))- go emptyBuilder (NativeResult FatalError [] [] Nothing errMap [] sql : acc)+ builder' <- newBuilder+ go builder' (NativeResult FatalError [] [] Nothing errMap [] sql : acc) CopyInResponse _ formats -> let fields = map copyField formats in pure (reverse (NativeResult CopyIn fields [] Nothing Map.empty [] "" : acc))@@ -502,7 +564,9 @@ -- | Collect the single result of an extended-protocol command. collectExtended :: Connection -> ByteString -> IO NativeResult-collectExtended connection sql = go emptyBuilder Nothing+collectExtended connection sql = do+ builder0 <- newBuilder+ go builder0 Nothing where go builder finished = do message <- nextMessage connection@@ -510,51 +574,68 @@ RowDescription fs -> go builder {accFields = fs, accSawRowDescription = True} finished ParameterDescription oids -> go builder {accParamOids = oids} finished NoData -> go builder finished- DataRow values -> go builder {accRevRows = values : (accRevRows builder)} finished+ DataRow values -> do+ pushRow (accRows builder) values+ go builder finished ParseComplete -> go builder finished BindComplete -> go builder finished CloseComplete -> go builder finished- PortalSuspended -> go emptyBuilder (finished <|> Just (commandResult builder Nothing))+ PortalSuspended -> do+ finished' <- case finished of+ Just _ -> pure finished+ Nothing -> Just <$> commandResult builder Nothing+ builder' <- newBuilder+ go builder' finished' CommandComplete tag -> do writeIORef (lastError connection) (Just "")- go emptyBuilder (Just (commandResult builder (Just tag)))+ result <- commandResult builder (Just tag)+ builder' <- newBuilder+ go builder' (Just result) EmptyQueryResponse -> do writeIORef (lastError connection) (Just "")- go emptyBuilder (Just (NativeResult EmptyQuery [] [] Nothing Map.empty [] ""))+ builder' <- newBuilder+ go builder' (Just (NativeResult EmptyQuery [] [] Nothing Map.empty [] "")) ErrorResponse fs -> do let errMap = Map.fromList fs writeIORef (lastError connection) (Just (formatResultError sql errMap))- go emptyBuilder (Just (NativeResult FatalError [] [] Nothing errMap [] sql))+ builder' <- newBuilder+ go builder' (Just (NativeResult FatalError [] [] Nothing errMap [] sql)) ReadyForQuery txState -> do writeIORef (txStatus connection) txState- pure (fromMaybe (describeResult builder) finished)+ case finished of+ Just result -> pure result+ Nothing -> describeResult builder _ -> go builder finished -- | A result terminated by @CommandComplete@\/@PortalSuspended@: 'TuplesOk' if a -- row description was seen, else 'CommandOk'.-commandResult :: Builder -> Maybe ByteString -> NativeResult-commandResult builder tag =- NativeResult- (if (accSawRowDescription builder) then TuplesOk else CommandOk)- (accFields builder)- (reverse (accRevRows builder))- tag- Map.empty- (accParamOids builder)- ""+commandResult :: Builder -> Maybe ByteString -> IO NativeResult+commandResult builder tag = do+ rows <- freezeRowBuffer (accRows builder)+ pure+ $ NativeResult+ (if (accSawRowDescription builder) then TuplesOk else CommandOk)+ (accFields builder)+ rows+ tag+ Map.empty+ (accParamOids builder)+ "" -- | A result with no command completion (a @Describe@\/@Parse@-only flow): -- 'CommandOk', carrying any column descriptions and parameter OIDs.-describeResult :: Builder -> NativeResult-describeResult builder =- NativeResult- CommandOk- (accFields builder)- (reverse (accRevRows builder))- Nothing- Map.empty- (accParamOids builder)- ""+describeResult :: Builder -> IO NativeResult+describeResult builder = do+ rows <- freezeRowBuffer (accRows builder)+ pure+ $ NativeResult+ CommandOk+ (accFields builder)+ rows+ Nothing+ Map.empty+ (accParamOids builder)+ "" lastMaybe :: [a] -> Maybe a lastMaybe = foldl (\_ x -> Just x) Nothing
src/transport/Pqi/Native/Transport.hs view
@@ -1,3 +1,5 @@+{-# LANGUAGE CPP #-}+ -- | The byte-level transport: a TCP socket with a read buffer, plus the framing -- that turns the stream into discrete @[type byte][Int32 length][body]@ -- messages.@@ -20,6 +22,9 @@ import qualified Data.ByteString as ByteString import qualified Data.ByteString.Char8 as ByteString.Char8 import Data.IORef+#if !defined(mingw32_HOST_OS)+import GHC.Conc (atomically, orElse, threadWaitReadSTM, threadWaitWriteSTM)+#endif import qualified Network.Socket as Socket import qualified Network.Socket.ByteString as Socket.ByteString import Pqi.Native.Transport.Prelude@@ -97,8 +102,78 @@ socketFd transport = fromIntegral <$> Socket.unsafeFdSocket (socket transport) -- | Send a serialized message.+--+-- Mirrors libpq's @pqSendSome@ (@fe-misc.c@): a send that cannot complete+-- immediately must not merely block on writability. The server may itself be+-- blocked writing results this connection has not read yet, and a client+-- blocked in @send@ cannot drain them, so both sides would wait forever -+-- the exact pipeline deadlock the comment above libpq's own loop describes.+-- So whenever the socket is not writable, incoming bytes are absorbed into+-- the read buffer first and the wait is for /read-or-write/ readiness:+-- whichever side can make progress does. Sending is preferred whenever the+-- socket is writable (the @orElse@ in 'awaitWritableOrReadable' is+-- left-biased), so a server streaming a large result cannot starve the+-- send.+#if defined(mingw32_HOST_OS)+-- Windows keeps the plain blocking @sendAll@: GHC's I\/O manager has no+-- dependable @threadWaitRead@\/@threadWaitWrite@ there, which the+-- read-while-sending loop is built on. The deadlock the loop prevents is+-- not Windows-specific, but reaching it needs the kernel socket buffers to+-- fill while server output is pending, which the forwarding paths Windows+-- clients typically sit behind make unlikely. send :: Transport -> Poker.Write -> IO () send transport write = Socket.ByteString.sendAll (socket transport) (Poker.toByteString write)+#else+send :: Transport -> Poker.Write -> IO ()+send transport write = go (Poker.toByteString write)+ where+ sock = socket transport+ go bytes+ | ByteString.null bytes = pure ()+ | otherwise =+ awaitWritableOrReadable sock >>= \case+ True -> do+ sent <- Socket.ByteString.send sock bytes+ go (ByteString.drop sent bytes)+ False -> do+ drainIncoming transport+ go bytes++-- | Wait until the socket is writable or, failing that, readable, reporting+-- which side woke ('True' for writable). Both waits are armed at once+-- because a socket that stays unwritable can still keep receiving data the+-- caller must absorb for the server's output path to keep draining.+awaitWritableOrReadable :: Socket.Socket -> IO Bool+awaitWritableOrReadable sock = do+ fd <- Socket.unsafeFdSocket sock+ (writable, cancelWritable) <- threadWaitWriteSTM (fromIntegral fd)+ (readable, cancelReadable) <- threadWaitReadSTM (fromIntegral fd)+ outcome <- atomically (fmap (\_ -> True) writable `orElse` fmap (\_ -> False) readable)+ cancelReadable+ cancelWritable+ pure outcome++-- | Absorb one chunk of incoming data into the read buffer - the analogue+-- of libpq calling @pqReadData()@ inside its send loop. Only called once+-- the socket has reported readable, so the @recv@ cannot block. The chunk+-- is recorded under 'mask_' for the same reason as in 'fillTo': an async+-- exception must never land between @recv@ returning and its bytes being+-- recorded, or those bytes would be lost and the stream desynced. An empty+-- chunk is the peer closing the connection mid-send, reported like every+-- other transport-level EOF.+drainIncoming :: Transport -> IO ()+drainIncoming transport = do+ closed <-+ mask_ do+ chunk <- Socket.ByteString.recv (socket transport) 65536+ if ByteString.null 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)+#endif -- | Ensure the read buffer holds at least @n@ bytes, pulling from the socket -- as needed. Throws on EOF before @n@ bytes are available.
src/transport/Pqi/Native/Transport/Message.hs view
@@ -35,6 +35,7 @@ import qualified Data.ByteString as ByteString import Pqi.Native.Comms import Pqi.Native.Transport.Prelude+import qualified PtrPeeker as Peeker import qualified PtrPoker.Write as Poker -- * Framing helpers@@ -220,7 +221,7 @@ case typeByte of 0x52 -> runDecoder authentication body 0x53 -> runDecoder (ParameterStatus <$> cstring <*> cstring) body- 0x4b -> runDecoder (BackendKeyData <$> int32 <*> int32) body+ 0x4b -> runDecoder (liftFixed (BackendKeyData <$> Peeker.beSignedInt4 <*> Peeker.beSignedInt4)) body 0x5a -> runDecoder (ReadyForQuery <$> word8) body 0x54 -> runDecoder (RowDescription <$> repeatedInt16 fieldDescription) body 0x44 -> runDecoder (DataRow <$> repeatedInt16 columnValue) body@@ -248,23 +249,25 @@ count <- int16 replicateM (fromIntegral count) element +-- | Six consecutive fixed-size fields, composed and lifted once instead of+-- separately (see 'Pqi.Native.Comms.lengthPrefixedBytes'\'s haddock for why+-- this matters; here it applies per column of every 'RowDescription'). fieldDescription :: Decoder FieldDescription-fieldDescription =- FieldDescription- <$> cstring- <*> word32- <*> int16- <*> word32- <*> int16- <*> int32- <*> int16+fieldDescription = do+ fieldName <- cstring+ liftFixed+ $ ( \tableOid_ columnAttributeNumber_ typeOid_ typeSize_ typeModifier_ formatCode_ ->+ FieldDescription fieldName tableOid_ columnAttributeNumber_ typeOid_ typeSize_ typeModifier_ formatCode_+ )+ <$> Peeker.beUnsignedInt4+ <*> Peeker.beSignedInt2+ <*> Peeker.beUnsignedInt4+ <*> Peeker.beSignedInt2+ <*> Peeker.beSignedInt4+ <*> Peeker.beSignedInt2 columnValue :: Decoder (Maybe ByteString)-columnValue = do- len <- int32- if len < 0- then pure Nothing- else Just <$> bytes (fromIntegral len)+columnValue = lengthPrefixedBytes -- | The fields of an @ErrorResponse@\/@NoticeResponse@: @(code, value)@ pairs, -- terminated by a zero code byte.