1{-# OPTIONS_HADDOCK hide #-}
3{-# LANGUAGE BangPatterns #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE MagicHash #-}
6{-# LANGUAGE RecordWildCards #-}
7{-# LANGUAGE UnboxedTuples #-}
12-- Module : Streamly.Internal.Network.Socket
13-- Copyright : (c) 2018 Composewell Technologies
16-- Maintainer : streamly@composewell.com
17-- Stability : experimental
20module Streamly.Internal.Network.Socket
27 -- * Accept connections
31 -- * Read from connection
45 , readChunksWithBufferOf
48 , toChunksWithBufferOf
51 -- * Write to connection
62 -- reading/writing datagrams
66import Control.Concurrent (threadWaitWrite, rtsSupportsBoundThreads)
67import Control.Monad.Catch (MonadCatch, finally, MonadMask)
68import Control.Monad.IO.Class (MonadIO(..))
69import Control.Monad (when)
70import Data.Word (Word8)
71import Foreign.ForeignPtr (withForeignPtr)
72import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
73import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)
74import Foreign.Storable (Storable(..))
75import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
76import Network.Socket (sendBuf, recvBuf)
78 (Socket, SocketOption(..), Family(..), SockAddr(..),
79 ProtocolNumber, withSocketsDo, SocketType(..), socket, bind,
81#if MIN_VERSION_network(3,1,0)
82import Network.Socket (withFdSocket)
84import Network.Socket (fdSocket)
86import Prelude hiding (read)
88import qualified Network.Socket as Net
90import Streamly (MonadAsync)
91import Streamly.Internal.Data.Unfold.Types (Unfold(..))
92import Streamly.Internal.Memory.Array.Types (Array(..))
93import Streamly.Streams.Serial (SerialT)
94import Streamly.Streams.StreamK.Type (IsStream, mkStream)
95import Streamly.Data.Fold (Fold)
96-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
98import qualified Streamly.Data.Fold as FL
99import qualified Streamly.Internal.Data.Fold.Types as FL
100import qualified Streamly.Internal.Data.Unfold as UF
101import qualified Streamly.Memory.Array as A
102import qualified Streamly.Internal.Memory.ArrayStream as AS
103import qualified Streamly.Internal.Memory.Array.Types as A
104import qualified Streamly.Prelude as S
105import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
107-- | @'useSocketM' socket act@ runs the monadic computation @act@ passing the
108-- socket handle to it. The handle will be closed on exit from 'useSocketM',
109-- whether by normal termination or by raising an exception. If closing the
110-- handle raises an exception, then this exception will be raised by
111-- 'useSocketM' rather than any exception raised by 'act'.
114{-# INLINE useSocketM #-}
115useSocketM :: (MonadMask m, MonadIO m) => Socket -> (Socket -> m ()) -> m ()
116useSocketM sk f = finally (liftIO (Net.close sk)) (f sk)
118-- | Like 'useSocketM' but runs a streaming computation instead of a monadic
122{-# INLINE useSocket #-}
123useSocket :: (IsStream t, MonadCatch m, MonadIO m)
124 => Socket -> (Socket -> t m a) -> t m a
125useSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)
127-------------------------------------------------------------------------------
129-------------------------------------------------------------------------------
131-- XXX Protocol specific socket options should be separated from socket level
134-- | Specify the socket protocol details.
135data SockSpec = SockSpec
137 sockFamily :: !Family
138 , sockType :: !SocketType
139 , sockProto :: !ProtocolNumber
140 , sockOpts :: ![(SocketOption, Int)]
143initListener :: Int -> SockSpec -> SockAddr -> IO Socket
144initListener listenQLen SockSpec{..} addr =
146 sock <- socket sockFamily sockType sockProto
147 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
149 Net.listen sock listenQLen
152{-# INLINE listenTuples #-}
153listenTuples :: MonadIO m
154 => Unfold m (Int, SockSpec, SockAddr) (Socket, SockAddr)
155listenTuples = Unfold step inject
157 inject (listenQLen, spec, addr) = do
158 listener <- liftIO $ initListener listenQLen spec addr
162 r <- liftIO $ Net.accept listener
163 -- XXX error handling
164 return $ D.Yield r listener
166-- | Unfold a three tuple @(listenQLen, spec, addr)@ into a stream of connected
167-- protocol sockets corresponding to incoming connections. @listenQLen@ is the
168-- maximum number of pending connections in the backlog. @spec@ is the socket
169-- protocol and options specification and @addr@ is the protocol address where
170-- the server listens for incoming connections.
174accept :: MonadIO m => Unfold m (Int, SockSpec, SockAddr) Socket
175accept = UF.map fst listenTuples
177-------------------------------------------------------------------------------
179-------------------------------------------------------------------------------
181{-# INLINE recvConnectionTuplesWith #-}
182recvConnectionTuplesWith :: MonadAsync m
183 => Int -> SockSpec -> SockAddr -> SerialT m (Socket, SockAddr)
184recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing
187 listener <- liftIO $ initListener tcpListenQ spec addr
188 r <- liftIO $ Net.accept listener
189 -- XXX error handling
190 return $ Just (r, Just listener)
192 step (Just listener) = do
193 r <- liftIO $ Net.accept listener
194 -- XXX error handling
195 return $ Just (r, Just listener)
197-- | Start a TCP stream server that listens for connections on the supplied
198-- server address specification (address family, local interface IP address and
199-- port). The server generates a stream of connected sockets. The first
200-- argument is the maximum number of pending connections in the backlog.
203{-# INLINE connections #-}
204connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> SerialT m Socket
205connections tcpListenQ spec addr = fmap fst $
206 recvConnectionTuplesWith tcpListenQ spec addr
208-------------------------------------------------------------------------------
210-------------------------------------------------------------------------------
212{-# INLINABLE readArrayUptoWith #-}
214 :: (h -> Ptr Word8 -> Int -> IO Int)
218readArrayUptoWith f size h = do
219 ptr <- mallocPlainForeignPtrBytes size
220 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
221 withForeignPtr ptr $ \p -> do
225 , aEnd = p `plusPtr` n
226 , aBound = p `plusPtr` size
228 -- XXX shrink only if the diff is significant
232-- | Read a 'ByteArray' from a file handle. If no data is available on the
233-- handle it blocks until some data becomes available. If data is available
234-- then it immediately returns that data without blocking. It reads a maximum
235-- of up to the size requested.
236{-# INLINABLE readArrayOf #-}
237readArrayOf :: Int -> Socket -> IO (Array Word8)
238readArrayOf = readArrayUptoWith recvBuf
240-------------------------------------------------------------------------------
242-------------------------------------------------------------------------------
244waitWhen0 :: Int -> Socket -> IO ()
245waitWhen0 0 s = when rtsSupportsBoundThreads $
246#if MIN_VERSION_network(3,1,0)
247 withFdSocket s $ \fd -> threadWaitWrite $ fromIntegral fd
249 let fd = fdSocket s in threadWaitWrite $ fromIntegral fd
251waitWhen0 _ _ = return ()
253sendAll :: Socket -> Ptr Word8 -> Int -> IO ()
254sendAll _ _ len | len <= 0 = return ()
256 sent <- sendBuf s p len
258 -- assert (sent <= len)
259 when (sent >= 0) $ sendAll s (p `plusPtr` sent) (len - sent)
261{-# INLINABLE writeArrayWith #-}
262writeArrayWith :: Storable a
263 => (h -> Ptr Word8 -> Int -> IO ())
267writeArrayWith _ _ arr | A.length arr == 0 = return ()
268writeArrayWith f h Array{..} = withForeignPtr aStart $ \p ->
272 let p = unsafeForeignPtrToPtr aStart
275-- | Write an Array to a file handle.
278{-# INLINABLE writeArray #-}
279writeArray :: Storable a => Socket -> Array a -> IO ()
280writeArray = writeArrayWith sendAll
282-------------------------------------------------------------------------------
283-- Stream of Arrays IO
284-------------------------------------------------------------------------------
286{-# INLINABLE readChunksUptoWith #-}
287readChunksUptoWith :: (IsStream t, MonadIO m)
288 => (Int -> h -> IO (Array Word8))
289 -> Int -> h -> t m (Array Word8)
290readChunksUptoWith f size h = go
292 -- XXX use cons/nil instead
293 go = mkStream $ \_ yld _ stp -> do
294 arr <- liftIO $ f size h
299-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
300-- The maximum size of a single array is limited to @size@.
301-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
303{-# INLINABLE toChunksWithBufferOf #-}
304toChunksWithBufferOf :: (IsStream t, MonadIO m)
305 => Int -> Socket -> t m (Array Word8)
306toChunksWithBufferOf = readChunksUptoWith readArrayOf
308-- XXX read 'Array a' instead of Word8
310-- | @toChunks h@ reads a stream of arrays from socket handle @h@.
311-- The maximum size of a single array is limited to @defaultChunkSize@.
314{-# INLINE toChunks #-}
315toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)
316toChunks = toChunksWithBufferOf A.defaultChunkSize
318-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.
319-- Read requests to the socket are performed using a buffer of size @bufsize@.
320-- The size of an array in the resulting stream is always less than or equal to
324{-# INLINE_NORMAL readChunksWithBufferOf #-}
325readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
326readChunksWithBufferOf = Unfold step return
328 {-# INLINE_LATE step #-}
330 arr <- liftIO $ readArrayOf size h
334 _ -> D.Yield arr (size, h)
336-- | Unfolds a socket into a stream of 'Word8' arrays. Requests to the socket
337-- are performed using a buffer of size
338-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'. The
339-- size of arrays in the resulting stream are therefore less than or equal to
340-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
343{-# INLINE readChunks #-}
344readChunks :: MonadIO m => Unfold m Socket (Array Word8)
345readChunks = UF.supplyFirst readChunksWithBufferOf A.defaultChunkSize
347-------------------------------------------------------------------------------
348-- Read File to Stream
349-------------------------------------------------------------------------------
351-- TODO for concurrent streams implement readahead IO. We can send multiple
352-- read requests at the same time. For serial case we can use async IO. We can
353-- also control the read throughput in mbps or IOPS.
356-- | @readWithBufferOf bufsize handle@ reads a byte stream from a file
357-- handle, reads are performed in chunks of up to @bufsize@. The stream ends
358-- as soon as EOF is encountered.
360{-# INLINE readWithBufferOf #-}
361readWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
362readWithBufferOf chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h
366-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
368-- > read = 'readByChunks' A.defaultChunkSize
369-- | Generate a stream of elements of the given type from a socket. The
370-- stream ends when EOF is encountered.
373{-# INLINE toStream #-}
374toStream :: (IsStream t, MonadIO m) => Socket -> t m Word8
375toStream = AS.concat . toChunks
377-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests
378-- to the socket are performed using buffers of @bufsize@.
381{-# INLINE readWithBufferOf #-}
382readWithBufferOf :: MonadIO m => Unfold m (Int, Socket) Word8
383readWithBufferOf = UF.concat readChunksWithBufferOf A.read
385-- | Unfolds a 'Socket' into a byte stream. IO requests to the socket are
386-- performed in sizes of
387-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
391read :: MonadIO m => Unfold m Socket Word8
392read = UF.supplyFirst readWithBufferOf A.defaultChunkSize
394-------------------------------------------------------------------------------
396-------------------------------------------------------------------------------
398-- | Write a stream of arrays to a handle.
401{-# INLINE writeChunksS #-}
402writeChunksS :: (MonadIO m, Storable a)
403 => Socket -> SerialT m (Array a) -> m ()
404writeChunksS h m = S.mapM_ (liftIO . writeArray h) m
406-- | Write a stream of arrays to a socket. Each array in the stream is written
407-- to the socket as a separate IO request.
410{-# INLINE writeChunks #-}
411writeChunks :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()
412writeChunks h = FL.drainBy (liftIO . writeArray h)
414-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
417-- Note that if you use a chunk size less than 8K (GHC's default buffer
418-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
419-- do not want buffering to occur at GHC level as well. Same thing applies to
422-- | Like 'write' but provides control over the write buffer. Output will
423-- be written to the IO device as soon as we collect the specified number of
427{-# INLINE writeWithBufferOfS #-}
428writeWithBufferOfS :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()
429writeWithBufferOfS n h m = writeChunksS h $ AS.arraysOf n m
431-- | Write a byte stream to a socket. Accumulates the input in chunks of
432-- specified number of bytes before writing.
435{-# INLINE writeWithBufferOf #-}
436writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
437writeWithBufferOf n h = FL.lchunksOf n (A.writeNUnsafe n) (writeChunks h)
439-- > write = 'writeWithBufferOf' A.defaultChunkSize
441-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
442-- up to 'A.defaultChunkSize' before writing. Note that the write behavior
443-- depends on the 'IOMode' and the current seek position of the handle.
446{-# INLINE _writeS #-}
447_writeS :: MonadIO m => Socket -> SerialT m Word8 -> m ()
448_writeS = writeWithBufferOfS A.defaultChunkSize
450-- | Write a byte stream to a socket. Accumulates the input in chunks of
451-- up to 'A.defaultChunkSize' bytes before writing.
454-- write = 'writeWithBufferOf' 'A.defaultChunkSize'
459write :: MonadIO m => Socket -> Fold m Word8 ()
460write = writeWithBufferOf A.defaultChunkSize
464write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
465write = toHandleWith A.defaultChunkSize
468-------------------------------------------------------------------------------
469-- IO with encoding/decoding Unicode characters
470-------------------------------------------------------------------------------
474-- > readUtf8 = decodeUtf8 . read
476-- Read a UTF8 encoded stream of unicode characters from a file handle.
479{-# INLINE readUtf8 #-}
480readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
481readUtf8 = decodeUtf8 . read
484-- > writeUtf8 h s = write h $ encodeUtf8 s
486-- Encode a stream of unicode characters to UTF8 and write it to the given file
487-- handle. Default block buffering applies to the writes.
490{-# INLINE writeUtf8 #-}
491writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
492writeUtf8 h s = write h $ encodeUtf8 s
494-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
495-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
496-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
497-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
501{-# INLINE writeUtf8ByLines #-}
502writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
503writeUtf8ByLines = undefined
505-- | Read UTF-8 lines from a file handle and apply the specified fold to each
506-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
509{-# INLINE readLines #-}
510readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
511readLines h f = foldLines (readUtf8 h) f
513-------------------------------------------------------------------------------
514-- Framing on a sequence
515-------------------------------------------------------------------------------
517-- | Read a stream from a file handle and split it into frames delimited by
518-- the specified sequence of elements. The supplied fold is applied on each
522{-# INLINE readFrames #-}
523readFrames :: (IsStream t, MonadIO m, Storable a)
524 => Array a -> Handle -> Fold m a b -> t m b
525readFrames = undefined -- foldFrames . read
527-- | Write a stream to the given file handle buffering up to frames separated
528-- by the given sequence or up to a maximum of @defaultChunkSize@.
531{-# INLINE writeByFrames #-}
532writeByFrames :: (IsStream t, MonadIO m, Storable a)
533 => Array a -> Handle -> t m a -> m ()
534writeByFrames = undefined
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list