@@ -1,18 +1,9 @@
2−{-# LANGUAGE BangPatterns #-}
3−{-# LANGUAGE FlexibleContexts #-}
4−{-# LANGUAGE MagicHash #-}
5−{-# LANGUAGE RecordWildCards #-}
6−{-# LANGUAGE UnboxedTuples #-}
11-- Module : Streamly.Network.Socket
2-- Module : Streamly.Network.Socket
12−-- Copyright : (c) 2018 Harendra Kumar
3+-- Copyright : (c) 2018 Composewell Technologies
14-- License : BSD3
5-- License : BSD3
15−-- Maintainer : harendra.kumar@gmail.com
6+-- Maintainer : streamly@composewell.com
16-- Stability : experimental
7-- Stability : experimental
17-- Portability : GHC
8-- Portability : GHC
@@ -66,346 +57,5 @@ module Streamly.Network.Socket
69−import Control.Concurrent (threadWaitWrite, rtsSupportsBoundThreads)
70−import Control.Monad.Catch (MonadCatch, onException)
71−import Control.Monad.IO.Class (MonadIO(..))
72−import Control.Monad (when)
73−import Data.Word (Word8)
74−import Foreign.ForeignPtr (withForeignPtr)
75−import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
76−import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)
77−import Foreign.Storable (Storable(..))
78−import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
79−import Network.Socket (Socket, sendBuf, recvBuf)
80−#if MIN_VERSION_network(3,1,0)
81−import Network.Socket (withFdSocket)
83−import Network.Socket (fdSocket)
60+import Streamly.Internal.Network.Socket
85import Prelude hiding (read)
61import Prelude hiding (read)
87−import qualified Network.Socket as Net
89−import Streamly.Internal.Memory.Array.Types (Array(..))
90−import Streamly.Streams.Serial (SerialT)
91−import Streamly.Streams.StreamK.Type (IsStream, mkStream)
92−import Streamly.Data.Fold (Fold)
93−-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
95−import qualified Streamly.Data.Fold as FL
96−import qualified Streamly.Internal.Data.Fold.Types as FL
97−import qualified Streamly.Memory.Array as A
98−import qualified Streamly.Internal.Memory.ArrayStream as AS
99−import qualified Streamly.Internal.Memory.Array.Types as A
100−import qualified Streamly.Prelude as S
102−-- | @'withSocket' socket act@ runs the monadic computation @act@ passing the
103−-- socket handle to it. The handle will be closed on exit from 'withSocket',
104−-- whether by normal termination or by raising an exception. If closing the
105−-- handle raises an exception, then this exception will be raised by
106−-- 'withSocket' rather than any exception raised by 'act'.
109−{-# INLINE withSocket #-}
110−withSocket :: (MonadCatch m, MonadIO m) => Socket -> (Socket -> m ()) -> m ()
112− f sk `onException` liftIO (Net.close sk)
113− liftIO (Net.close sk)
115−-- XXX bracket performs 2x better than 'finally'. That's perhaps because of
116−-- better fusion of "A.flattenArrays . readArrays"?
118−-- withSocketS :: (IsStream t, MonadCatch m, MonadIO m) => Socket -> t m Word8
119−-- withSocketS sk = A.flattenArrays $
120−-- S.finally (liftIO (Net.close sk)) (readArrays sk)
122−-- | Like 'withSocket' but runs a streaming computation instead of a monadic
126−{-# INLINE withSocketS #-}
127−withSocketS :: (IsStream t, MonadCatch m, MonadIO m)
128− => Socket -> (Socket -> t m a) -> t m a
129−withSocketS sk = S.bracket (return sk) (liftIO . Net.close)
131−-------------------------------------------------------------------------------
133−-------------------------------------------------------------------------------
135−{-# INLINABLE readArrayUptoWith #-}
137− :: (h -> Ptr Word8 -> Int -> IO Int)
141−readArrayUptoWith f size h = do
142− ptr <- mallocPlainForeignPtrBytes size
143− -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
144− withForeignPtr ptr $ \p -> do
148− , aEnd = p `plusPtr` n
149− , aBound = p `plusPtr` size
151− -- XXX shrink only if the diff is significant
155−-- | Read a 'ByteArray' from a file handle. If no data is available on the
156−-- handle it blocks until some data becomes available. If data is available
157−-- then it immediately returns that data without blocking. It reads a maximum
158−-- of up to the size requested.
159−{-# INLINABLE readArrayOf #-}
160−readArrayOf :: Int -> Socket -> IO (Array Word8)
161−readArrayOf = readArrayUptoWith recvBuf
163−-------------------------------------------------------------------------------
165−-------------------------------------------------------------------------------
167−waitWhen0 :: Int -> Socket -> IO ()
168−waitWhen0 0 s = when rtsSupportsBoundThreads $
169−#if MIN_VERSION_network(3,1,0)
170− withFdSocket s $ \fd -> threadWaitWrite $ fromIntegral fd
172− let fd = fdSocket s in threadWaitWrite $ fromIntegral fd
174−waitWhen0 _ _ = return ()
176−sendAll :: Socket -> Ptr Word8 -> Int -> IO ()
177−sendAll _ _ len | len <= 0 = return ()
179− sent <- sendBuf s p len
181− -- assert (sent <= len)
182− when (sent >= 0) $ sendAll s (p `plusPtr` sent) (len - sent)
184−{-# INLINABLE writeArrayWith #-}
185−writeArrayWith :: Storable a
186− => (h -> Ptr Word8 -> Int -> IO ())
190−writeArrayWith _ _ arr | A.length arr == 0 = return ()
191−writeArrayWith f h Array{..} = withForeignPtr aStart $ \p ->
192− f h (castPtr p) aLen
195− let p = unsafeForeignPtrToPtr aStart
196− in aEnd `minusPtr` p
198−-- | Write an Array to a file handle.
201−{-# INLINABLE writeArray #-}
202−writeArray :: Storable a => Socket -> Array a -> IO ()
203−writeArray = writeArrayWith sendAll
205−-------------------------------------------------------------------------------
206−-- Stream of Arrays IO
207−-------------------------------------------------------------------------------
209−{-# INLINABLE readArraysUptoWith #-}
210−readArraysUptoWith :: (IsStream t, MonadIO m)
211− => (Int -> h -> IO (Array Word8))
212− -> Int -> h -> t m (Array Word8)
213−readArraysUptoWith f size h = go
215− -- XXX use cons/nil instead
216− go = mkStream $ \_ yld _ stp -> do
217− arr <- liftIO $ f size h
218− if A.length arr == 0
222−-- | @readArraysOf size h@ reads a stream of arrays from file handle @h@.
223−-- The maximum size of a single array is limited to @size@.
224−-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
226−{-# INLINABLE readArraysOf #-}
227−readArraysOf :: (IsStream t, MonadIO m)
228− => Int -> Socket -> t m (Array Word8)
229−readArraysOf = readArraysUptoWith readArrayOf
231−-- XXX read 'Array a' instead of Word8
233−-- | @readArrays h@ reads a stream of arrays from socket handle @h@.
234−-- The maximum size of a single array is limited to @defaultChunkSize@.
237−{-# INLINE readArrays #-}
238−readArrays :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)
239−readArrays = readArraysOf A.defaultChunkSize
241−-------------------------------------------------------------------------------
242−-- Read File to Stream
243−-------------------------------------------------------------------------------
245−-- TODO for concurrent streams implement readahead IO. We can send multiple
246−-- read requests at the same time. For serial case we can use async IO. We can
247−-- also control the read throughput in mbps or IOPS.
250−-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file
251−-- handle, reads are performed in chunks of up to @chunkSize@. The stream ends
252−-- as soon as EOF is encountered.
254−{-# INLINE readInChunksOf #-}
255−readInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
256−readInChunksOf chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h
260−-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
262−-- > read = 'readByChunks' A.defaultChunkSize
263−-- | Generate a stream of elements of the given type from a socket. The
264−-- stream ends when EOF is encountered.
268−read :: (IsStream t, MonadIO m) => Socket -> t m Word8
269−read = AS.concat . readArrays
271−-------------------------------------------------------------------------------
273−-------------------------------------------------------------------------------
275−-- | Write a stream of arrays to a handle.
278−{-# INLINE writeArraysS #-}
279−writeArraysS :: (MonadIO m, Storable a)
280− => Socket -> SerialT m (Array a) -> m ()
281−writeArraysS h m = S.mapM_ (liftIO . writeArray h) m
283−-- | Write a stream of arrays to a socket.
286−{-# INLINE writeArrays #-}
287−writeArrays :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()
288−writeArrays h = FL.drainBy (liftIO . writeArray h)
290−-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
293−-- Note that if you use a chunk size less than 8K (GHC's default buffer
294−-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
295−-- do not want buffering to occur at GHC level as well. Same thing applies to
298−-- | Like 'write' but provides control over the write buffer. Output will
299−-- be written to the IO device as soon as we collect the specified number of
303−{-# INLINE writeInChunksOfS #-}
304−writeInChunksOfS :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()
305−writeInChunksOfS n h m = writeArraysS h $ AS.arraysOf n m
307−-- | Write a byte stream to a socket. Accumulates the input in chunks of
308−-- specified number of bytes before writing.
311−{-# INLINE writeInChunksOf #-}
312−writeInChunksOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
313−writeInChunksOf n h = FL.lchunksOf n (A.writeNUnsafe n) (writeArrays h)
315−-- > write = 'writeInChunksOf' A.defaultChunkSize
317−-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
318−-- up to 'A.defaultChunkSize' before writing. Note that the write behavior
319−-- depends on the 'IOMode' and the current seek position of the handle.
322−{-# INLINE _writeS #-}
323−_writeS :: MonadIO m => Socket -> SerialT m Word8 -> m ()
324−_writeS = writeInChunksOfS A.defaultChunkSize
326−-- | Write a byte stream to a socket. Accumulates the input in chunks of
327−-- up to 'A.defaultChunkSize' bytes before writing.
330−-- write = 'writeInChunksOf' 'A.defaultChunkSize'
335−write :: MonadIO m => Socket -> Fold m Word8 ()
336−write = writeInChunksOf A.defaultChunkSize
340−write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
341−write = toHandleWith A.defaultChunkSize
344−-------------------------------------------------------------------------------
345−-- IO with encoding/decoding Unicode characters
346−-------------------------------------------------------------------------------
350−-- > readUtf8 = decodeUtf8 . read
352−-- Read a UTF8 encoded stream of unicode characters from a file handle.
355−{-# INLINE readUtf8 #-}
356−readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
357−readUtf8 = decodeUtf8 . read
360−-- > writeUtf8 h s = write h $ encodeUtf8 s
362−-- Encode a stream of unicode characters to UTF8 and write it to the given file
363−-- handle. Default block buffering applies to the writes.
366−{-# INLINE writeUtf8 #-}
367−writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
368−writeUtf8 h s = write h $ encodeUtf8 s
370−-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
371−-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
372−-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
373−-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
377−{-# INLINE writeUtf8ByLines #-}
378−writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
379−writeUtf8ByLines = undefined
381−-- | Read UTF-8 lines from a file handle and apply the specified fold to each
382−-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
385−{-# INLINE readLines #-}
386−readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
387−readLines h f = foldLines (readUtf8 h) f
389−-------------------------------------------------------------------------------
390−-- Framing on a sequence
391−-------------------------------------------------------------------------------
393−-- | Read a stream from a file handle and split it into frames delimited by
394−-- the specified sequence of elements. The supplied fold is applied on each
398−{-# INLINE readFrames #-}
399−readFrames :: (IsStream t, MonadIO m, Storable a)
400− => Array a -> Handle -> Fold m a b -> t m b
401−readFrames = undefined -- foldFrames . read
403−-- | Write a stream to the given file handle buffering up to frames separated
404−-- by the given sequence or up to a maximum of @defaultChunkSize@.
407−{-# INLINE writeByFrames #-}
408−writeByFrames :: (IsStream t, MonadIO m, Storable a)
409− => Array a -> Handle -> t m a -> m ()
410−writeByFrames = undefined