Эффективный двоичный ввод-вывод по сети

Ответ 1

Re вопрос 1...

Если дескриптор настроен с помощью NoBuffering, каждый вызов hPutStr будет генерировать системный вызов записи. Это приведет к огромному штрафу за производительность для большого количества небольших записей. См., Например, этот ответ SO для некоторого бенчмаркинга: fooobar.com/questions/550961/...

С другой стороны, если ручка включена с буферизацией, вам необходимо явно очистить дескриптор, чтобы обеспечить отправку буферизованных данных.

Я предполагаю, что вы используете протокол потоковой передачи, такой как TCP. С UDP вы, очевидно, должны формировать и отправлять каждое сообщение как атомную единицу.

Re question # 2...

Чтение кода, который кажется, что hGet для ленивых байтов будет считываться с дескриптора в кусках defaultChunkSize, который составляет около 32k.

Обновление: Похоже, что hGet не выполняет ленивый ввод-вывод в этом случае. Вот какой код проверить это. Питание:

#!/usr/bin/env perl
$| = 1;
my $c = 0;
my $k = "1" x 1024;
while (1) {
  syswrite(STDOUT, $k);
  $c++;
  print STDERR "wrote 1k count = $c\n";
}

Test.hs:

import qualified Data.ByteString.Lazy as LBS
import System.IO

main = do
  s <- LBS.hGet stdin 320000
  let s2 = LBS.take 10 s
  print $ ("Length s2 = ", s2)

Запуск perl feed | runhaskell Test.hs ясно, что программа Haskell требует, чтобы все 320 тыс. из программы perl, хотя оно использует только первые 10 байтов.

Ответ 2

TCP требует, чтобы приложение предоставляло свои собственные маркеры границы сообщений. Простым протоколом отмечать границы сообщений является отправка длины фрагмента данных, фрагмента данных и оставшихся фрагментов, которые являются частью одного и того же сообщения. Оптимальный размер заголовка, который содержит информацию о границе сообщения, зависит от распределения размеров сообщений.

Разрабатывая собственный протокол сообщений, мы будем использовать два байта для наших заголовков. Самый старший бит из байтов (обработанный как Word16) будет содержать то, остались или нет оставшиеся фрагменты в сообщении. Остальные 15 бит будут содержать длину сообщения в байтах. Это позволит размер блоков до 32 тыс., Что больше, чем типичные пакеты TCP. Заголовок из двух байтов будет менее оптимальным, если сообщения, как правило, очень малы, особенно если они меньше 127 байт.

Мы собираемся использовать network-simple для сетевой части нашего кода. Мы будем сериализовать или десериализовать сообщения с помощью binary пакета encode и decode от lazy ByteString s.

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 в строгий ByteString и снова читать их. Мы напишем их по-крупному. В качестве альтернативы они могут быть записаны в терминах Binary для Word16.

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 байт за один раз, мы должны иметь возможность rechunk ленивого байта в куски с общей известной длиной не более нашего максимума. Один кусок уже может быть больше максимального; любой кусок, который не вписывается в наши новые куски, разбивается на несколько кусков.

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 состоит в том, чтобы разбить его на куски размера, который мы знаем, мы можем отправлять и отправлять каждый фрагмент вместе с заголовком, содержащим размер, и есть ли еще куски.

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 состоит в чтении двух байтового заголовка, чтении фрагмента размера, указанного заголовком, и продолжении чтения до тех пор, пока заголовок указывает, что есть больше фрагментов.

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 просто отправляет encode d lazy ByteString или получает ленивые ByteString и decode ing.

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