haskell - 通过网络进行高效的二进制 I/O

Question

我正在尝试编写一个使用二进制网络协议(protocol)的小型 Haskell 程序，但我遇到了令人惊讶的困难。

很明显，二进制数据应该存储为ByteString。

问题:我应该只是 hGet/hPut 单独的多字节整数，还是构建一个大的 ByteString 的性能更高？整个事情并使用它？

看起来binary包在这里应该有用。但是，binary 仅处理惰性 ByteString 值。

问题:lazy ByteString 上的 hGet 是否实际上严格读取指定数量的字节？或者它是否尝试执行某种惰性 I/O？ (我不想要惰性 I/O!)

问题:为什么文档没有指定这一点？

代码看起来将包含很多“获取下一个整数，将其与该值进行比较，如果没有则抛出错误，否则继续下一步......”我不知道如何干净地构建它，而无需编写意大利面条代码。

总之，我想做的事情非常简单，但我似乎正在努力寻找一种使代码看起来简单的方法。也许我只是想太多了，错过了一些明显的东西......

Answer 1

TCP 要求应用程序提供自己的消息边界标记。标记消息边界的简单协议(protocol)是发送数据 block 的长度、数据 block 以及是否存在属于同一消息的剩余 block 。保存消息边界信息的 header 的最佳大小取决于消息大小的分布。

开发我们自己的小消息协议(protocol)，我们将使用两个字节作为 header 。字节中的最高有效位(被视为 Word16)将保存消息中是否存在剩余 block 。剩余的 15 位将保存消息的长度(以字节为单位)。这将允许 block 大小高达 32k，大于典型的 TCP 数据包。如果消息通常非常小，特别是小于 127 字节，则两字节 header 将不是最佳选择。

我们将使用network-simple对于我们代码的网络部分。我们将使用 binary 序列化或反序列化消息对惰性 ByteString 进行编码和解码的包。

import qualified Data.ByteString.Lazy as L
import qualified Data.ByteString as B

import Network.Simple.TCP 
import Data.Bits
import Data.Binary
import Data.Functor
import Control.Monad.IO.Class

我们需要的第一个实用程序是将 Word16 header 写入严格的 ByteString 并再次将其读回的能力。我们将以大端顺序写入它们。或者，这些可以根据 Word16 的 Binary 实例编写。

writeBE :: Word16 -> B.ByteString
writeBE x = B.pack . map fromIntegral $ [(x .&. 0xFF00) `shiftR` 8, x .&. 0xFF]

readBE :: B.ByteString -> Maybe Word16
readBE s =
    case map fromIntegral . B.unpack $ s of
        [w1, w0] -> Just $ w1 `shiftL` 8 .|. w0
        _        -> Nothing

主要的挑战是发送和接收二进制包强加给我们的惰性ByteString。由于我们一次最多只能发送 32k 字节，因此我们需要能够将惰性字节串重新分块为已知总长度不超过最大值的 block 。单个 block 可能已经超过最大值；任何不适合我们的新 block 的 block 都会被分成多个 block 。

rechunk :: Int -> [B.ByteString] -> [(Int, [B.ByteString])]
rechunk n = go [] 0 . filter (not . B.null)
    where
        go acc l []     = [(l, reverse acc)]
        go acc l (x:xs) =
            let
                lx = B.length x
                l' = lx + l
            in
                if l' <= n
                then go (x:acc) l' xs
                else
                    let (x0, x1) = B.splitAt (n-l) x
                    in (n, reverse (x0:acc)) : go [] 0 (x1:xs)

recvExactly 将循环，直到收到我们请求的所有字节。

recvExactly :: MonadIO m => Socket -> Int -> m (Maybe [B.ByteString])
recvExactly s toRead = go [] toRead
    where
        go acc toRead = do
            body <- recv s toRead
            maybe (return Nothing) (go' acc toRead) body
        go' acc toRead body =
            if B.length body < toRead
            then go (body:acc) (toRead - B.length body)
            else return . Just . reverse $ acc

发送惰性ByteString包括将其分成我们知道可以发送的大小的 block ，并发送每个 block 以及包含大小的 header 以及是否还有更多 block 。

sendLazyBS :: (MonadIO m) => Socket -> L.ByteString -> m ()
sendLazyBS s = go . rechunk maxChunk . L.toChunks
    where
        maxChunk = 0x7FFF
        go [] = return ()
        go ((li, ss):xs) = do
            let l = fromIntegral li
            let h = writeBE $ if null xs then l else l .|. 0x8000
            sendMany s (h:ss)
            go xs

接收惰性ByteString包括读取两个字节 header ，读取 header 指示的大小的 block ，并且只要 header 指示还有更多 block 就继续读取。

recvLazyBS :: (MonadIO m, Functor m) => Socket -> m (Maybe L.ByteString)
recvLazyBS s = fmap L.fromChunks <$> go [] 
    where
        go acc = do
            header <- recvExactly s 2
            maybe (return Nothing) (go' acc) (header >>= readBE . B.concat)
        go' acc h = do
            body <- recvExactly s . fromIntegral $ h .&. 0x7FFF
            let next = if h .&. 0x8000 /= 0
                       then go
                       else return . Just . concat . reverse
            maybe (return Nothing) (next . (:acc) ) body     

发送或接收具有Binary实例的消息只是发送encoded惰性ByteString或接收惰性ByteString 并对其进行解码。

sendBinary :: (MonadIO m, Binary a) => Socket -> a -> m ()
sendBinary s = sendLazyBS s . encode

recvBinary :: (MonadIO m, Binary a, Functor m) => Socket -> m (Maybe a)
recvBinary s = d . fmap decodeOrFail <$> recvLazyBS s
    where
        d (Just (Right (_, _, x))) = Just x
        d _                        = Nothing