8000 More documentatin for loopExtractDataRows · postgres-haskell/postgres-wire@e7e19f4 · GitHub
[go: up one dir, main page]
Skip to content
8000

Commit e7e19f4

Browse files
VyacheslavHashovVyacheslavHashov
committed
More documentatin for loopExtractDataRows
1 parent 4cbad53 commit e7e19f4

File tree

3 files changed

+33
-40
lines changed

3 files changed

+33
-40
lines changed

‎src/Database/PostgreSQL/Driver/Connection.hs

Lines changed: 0 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -239,36 +239,6 @@ receiverThread rawConn dataChan =
239239
(\bs -> (bs <>) <$> rReceive rawConn 4096)
240240
(writeChan dataChan . Right)
241241

242-
where
243-
receiveLoop
244-
:: Maybe Header
245-
-> B.ByteString
246-
-> [B.ByteString] -> IO ()
247-
-- Parsing header
248-
receiveLoop Nothing bs acc
249-
| B.length bs < 5 = do
250-
r <- rReceive rawConn 4096
251-
receiveLoop Nothing (bs <> r) acc
252-
| otherwise = case runDecode decodeHeader bs of
253-
-- TODO handle error
254-
Left reason -> undefined
255-
-- reportReceiverError dataChan allChan
256-
-- $ DecodeError $ BS.pack reason
257-
Right (rest, h) -> receiveLoop (Just h) rest acc
258-
-- Parsing body
259-
receiveLoop (Just h@(Header _ len)) bs acc
260-
| B.length bs < len = do
261-
r <- rReceive rawConn 4096
262-
receiveLoop (Just h) (bs <> r) acc
263-
| otherwise = case runDecode (decodeServerMessage h) bs of
264-
-- TODO handle error
265-
Left reason -> undefined
266-
-- reportReceiverError dataChan allChan
267-
-- $ DecodeError $ BS.pack reason
268-
Right (rest, v) -> do
269-
newAcc <- dispatchExtended dataChan v acc
270-
receiveLoop Nothing rest newAcc
271-
272242
-- | Any exception prevents thread from future work
273243
receiverThreadCommon
274244
:: RawConnection

‎src/Database/PostgreSQL/Protocol/Decoders.hs

Lines changed: 31 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -30,60 +30,80 @@ import Database.PostgreSQL.Protocol.Types
3030
import Database.PostgreSQL.Protocol.Store.Decode
3131
import Database.PostgreSQL.Protocol.Utils
3232

33-
-- Extracts DataRows
34-
--
33+
-- Optimized loop for extracting chunks of DataRows.
34+
-- Ignores all messages from database that do not relate to data.
35+
-- Does not throw exceptions.
3536
loopExtractDataRows
36-
:: (B.ByteString -> IO B.ByteString) -- read more action
37-
-> (DataMessage -> IO ()) -- callback on every DataMessage
37+
-- Action that returs more data with `ByteString` prepended.
38+
:: (B.ByteString -> IO B.ByteString)
39+
-- Will be called on every DataMessage.
40+
-> (DataMessage -> IO ())
3841
-> IO ()
3942
loopExtractDataRows readMoreAction callback = go "" ""
4043
where
4144
go :: B.ByteString -> BL.ByteString -> IO ()
4245
go bs acc
46+
-- 5 - header size, defined by PostgreSQL
4347
| B.length bs < 5 = readMoreAndGo bs acc
4448
| otherwise = do
4549
ScanRowResult ch rest r <- scanDataRows bs
4650
-- We should force accumulator
51+
-- `BL.chunk` does not prepend empty bytestring as chunk.
4752
let !newAcc = BL.chunk ch acc
4853

4954
case r of
55+
-- Following happened:
56+
-- not enough bytes to read header
57+
-- or header is for `DataRow`, not enough bytes to read body
5058
1 -> readMoreAndGo rest newAcc
59+
-- Header was read, it is not for `DataRow`. We can safely
60+
-- call `parseHeader`, because scanDataRows already checked
61+
-- that there are enough bytes to read header.
5162
2 -> do
5263
Header mt len <- parseHeader rest
5364
dispatchHeader mt len (B.drop 5 rest) newAcc
5465

5566
{-# INLINE dispatchHeader #-}
5667
dispatchHeader :: Word8 -> Int -> B.ByteString -> BL.ByteString -> IO ()
5768
dispatchHeader mt len bs acc = case mt of
58-
-- 'C'
69+
-- 'C' - CommandComplete.
70+
-- Command is completed, return the result.
5971
67 -> do
6072
callback $
6173
DataMessage . DataRows $
6274
BL.foldlChunks (flip BL.chunk) BL.empty acc
6375

6476
newBs <- skipBytes bs len
6577
go newBs BL.empty
66-
-- 'I'
78+
79+
-- 'I' - EmptyQueryResponse.
80+
-- PostgreSQL sends this if query string was empty and datarows
81+
-- should be empty, but anyway we return data collected in `acc`.
6782
73 -> do
6883
callback $
6984
DataMessage . DataRows $
7085
BL.foldlChunks (flip BL.chunk) BL.empty acc
7186

7287
go bs BL.empty
73-
-- 'E"
88+
89+
-- 'E' - ErrorResponse.
90+
-- On ErrorResponse we should discard all the collected datarows.
7491
69 -> do
7592
(b, newBs) <- readAtLeast bs len
7693
desc <- eitherToDecode $ parseErrorDesc b
7794
callback (DataError desc)
7895

7996
go newBs BL.empty
80-
-- 'Z'
97+
98+
-- 'Z' - ReadyForQuery.
99+
-- To know when command processing is finished
81100
90 -> do
82101
callback DataReady
83102

84103
newBs <- skipBytes bs len
85104
go newBs acc
86105

106+
-- Skip any other message.
87107
_ -> do
88108
newBs <- skipBytes bs len
89109
go newBs acc
@@ -94,6 +114,7 @@ loopExtractDataRows readMoreAction callback = go "" ""
94114
newBs <- readMoreAction bs
95115
go newBs acc
96116

117+
-- | Returns a bytestring that contain exactly @len@ bytes and the rest.
97118
{-# INLINE readAtLeast #-}
98119
readAtLeast :: B.ByteString -> Int -> IO (B.ByteString, B.ByteString)
99120
readAtLeast bs len
@@ -102,6 +123,7 @@ loopExtractDataRows readMoreAction callback = go "" ""
102123
newBs <- readMoreAction bs
103124
readAtLeast newBs len
104125

126+
-- | Skips exactly @toSkip@ bytes.
105127
{-# INLINE skipBytes #-}
106128
skipBytes :: B.ByteString -> Int -> IO B.ByteString
107129
skipBytes bs toSkip
@@ -113,7 +135,7 @@ loopExtractDataRows readMoreAction callback = go "" ""
113135

114136
{-# INLINE parseHeader #-}
115137
parseHeader :: B.ByteString -> IO Header
116-
parseHeader !bs =
138+
parseHeader bs =
117139
B.unsafeUseAsCStringLen bs $ \(ptr, len) -> do
118140
b <- peek (castPtr ptr)
119141
w <- byteSwap32 <$> peekByteOff (castPtr ptr) 1

‎src/Database/PostgreSQL/Protocol/Utils.hs

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,12 @@ import qualified Data.ByteString as B
77
import qualified Data.ByteString.Unsafe as B
88

99
data ScanRowResult = ScanRowResult
10-
{-# UNPACK #-} !B.ByteString -- chunk of datarows
10+
{-# UNPACK #-} !B.ByteString -- chunk of datarows, may be empty
1111
{-# UNPACK #-} !B.ByteString -- the rest of string
1212
{-# UNPACK #-} !Int -- reason code
1313

1414
{-# INLINE scanDataRows #-}
15+
-- | Scans `ByteString` for a chunk of `DataRow`s.
1516
scanDataRows :: B.ByteString -> IO ScanRowResult
1617
scanDataRows bs =
1718
alloca $ \reasonPtr ->

0 commit comments

Comments
 (0)
0