diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/pqi-native.cabal b/pqi-native.cabal --- a/pqi-native.cabal +++ b/pqi-native.cabal @@ -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, diff --git a/src/comms/Pqi/Native/Comms.hs b/src/comms/Pqi/Native/Comms.hs --- a/src/comms/Pqi/Native/Comms.hs +++ b/src/comms/Pqi/Native/Comms.hs @@ -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 diff --git a/src/library/Pqi/Native.hs b/src/library/Pqi/Native.hs --- a/src/library/Pqi/Native.hs +++ b/src/library/Pqi/Native.hs @@ -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 diff --git a/src/library/Pqi/Native/Query.hs b/src/library/Pqi/Native/Query.hs --- a/src/library/Pqi/Native/Query.hs +++ b/src/library/Pqi/Native/Query.hs @@ -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 diff --git a/src/transport/Pqi/Native/Transport.hs b/src/transport/Pqi/Native/Transport.hs --- a/src/transport/Pqi/Native/Transport.hs +++ b/src/transport/Pqi/Native/Transport.hs @@ -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. diff --git a/src/transport/Pqi/Native/Transport/Message.hs b/src/transport/Pqi/Native/Transport/Message.hs --- a/src/transport/Pqi/Native/Transport/Message.hs +++ b/src/transport/Pqi/Native/Transport/Message.hs @@ -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.