4-- Module : Streamly.Internal.Network.Socket
5-- Copyright : (c) 2018 Composewell Technologies
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12module Streamly.Internal.Network.Socket
19 -- * Accept connections
25 -- * Read from connection
38 , readChunksWithBufferOf
40 , toChunksWithBufferOf
44 -- * Write to connection
52 , putBytesWithBufferOf
58 , writeChunksWithBufferOf
60 -- reading/writing datagrams
64import Control.Concurrent (threadWaitWrite, rtsSupportsBoundThreads)
65import Control.Exception (onException)
66import Control.Monad.Catch (MonadCatch, finally, MonadMask)
67import Control.Monad.IO.Class (MonadIO(..))
68import Control.Monad (forM_, when)
69import Data.Word (Word8)
71import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
72import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)
73import Foreign.Storable (Storable(..))
74import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
76 (Socket, SocketOption(..), Family(..), SockAddr(..),
77 ProtocolNumber, withSocketsDo, SocketType(..), socket, bind,
78 setSocketOption, sendBuf, recvBuf)
79#if MIN_VERSION_network(3,1,0)
80import Network.Socket (withFdSocket)
82import Network.Socket (fdSocket)
84import Prelude hiding (read)
86import qualified Network.Socket as Net
88import Streamly.Internal.BaseCompat
89import Streamly.Internal.Control.Concurrent (MonadAsync)
90import Streamly.Internal.Data.Array.Foreign.Mut.Type (mutableArray)
91import Streamly.Internal.Data.Array.Foreign.Type (Array(..))
92import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)
93import Streamly.Internal.Data.Fold (Fold)
94import Streamly.Internal.Data.Stream.IsStream.Type
95 (IsStream, mkStream, fromStreamD)
96import Streamly.Internal.Data.Stream.Serial (SerialT)
97import Streamly.Internal.Data.Unfold.Type (Unfold(..))
98-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
99import Streamly.Internal.System.IO (defaultChunkSize)
101import qualified Streamly.Internal.Data.Array.Foreign as A
102import qualified Streamly.Internal.Data.Array.Foreign.Type as A
103import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
104import qualified Streamly.Internal.Data.Fold as FL
105import qualified Streamly.Internal.Data.Stream.IsStream as S
106import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
107import qualified Streamly.Internal.Data.Unfold as UF
109-- | @'forSocketM' action socket@ runs the monadic computation @action@ passing
110-- the socket handle to it. The handle will be closed on exit from
111-- 'forSocketM', whether by normal termination or by raising an exception. If
112-- closing the handle raises an exception, then this exception will be raised
113-- by 'forSocketM' rather than any exception raised by 'action'.
116{-# INLINE forSocketM #-}
117forSocketM :: (MonadMask m, MonadIO m) => (Socket -> m ()) -> Socket -> m ()
118forSocketM f sk = finally (f sk) (liftIO (Net.close sk))
120-- | Like 'forSocketM' but runs a streaming computation instead of a monadic
123-- /Inhibits stream fusion/
126{-# INLINE withSocket #-}
127withSocket :: (IsStream t, MonadAsync m, MonadCatch m)
128 => Socket -> (Socket -> t m a) -> t m a
129withSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)
131-------------------------------------------------------------------------------
133-------------------------------------------------------------------------------
135-- XXX Protocol specific socket options should be separated from socket level
138-- | Specify the socket protocol details.
139data SockSpec = SockSpec
141 sockFamily :: !Family
142 , sockType :: !SocketType
143 , sockProto :: !ProtocolNumber
144 , sockOpts :: ![(SocketOption, Int)]
147initListener :: Int -> SockSpec -> SockAddr -> IO Socket
148initListener listenQLen SockSpec{..} addr =
150 sock <- socket sockFamily sockType sockProto
151 use sock `onException` Net.close sock
157 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
159 Net.listen sock listenQLen
161{-# INLINE listenTuples #-}
162listenTuples :: MonadIO m
163 => Unfold m (Int, SockSpec, SockAddr) (Socket, SockAddr)
164listenTuples = Unfold step inject
166 inject (listenQLen, spec, addr) =
167 liftIO $ initListener listenQLen spec addr
170 r <- liftIO (Net.accept listener `onException` Net.close listener)
171 return $ D.Yield r listener
173-- | Unfold a three tuple @(listenQLen, spec, addr)@ into a stream of connected
174-- protocol sockets corresponding to incoming connections. @listenQLen@ is the
175-- maximum number of pending connections in the backlog. @spec@ is the socket
176-- protocol and options specification and @addr@ is the protocol address where
177-- the server listens for incoming connections.
181accept :: MonadIO m => Unfold m (Int, SockSpec, SockAddr) Socket
182accept = UF.map fst listenTuples
184{-# INLINE connectCommon #-}
185connectCommon :: SockSpec -> Maybe SockAddr -> SockAddr -> IO Socket
186connectCommon SockSpec{..} local remote = withSocketsDo $ do
187 sock <- socket sockFamily sockType sockProto
188 use sock `onException` Net.close sock
194 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
195 forM_ local (bind sock)
196 Net.connect sock remote
198-- | Connect to a remote host using the given socket specification and remote
199-- address. Returns a connected socket or throws an exception.
203{-# INLINE connect #-}
204connect :: SockSpec -> SockAddr -> IO Socket
205connect spec = connectCommon spec Nothing
207-- | Connect to a remote host using the given socket specification, a local
208-- address to bind to and a remote address to connect to. Returns a connected
209-- socket or throws an exception.
213{-# INLINE connectFrom #-}
214connectFrom :: SockSpec -> SockAddr -> SockAddr -> IO Socket
215connectFrom spec local = connectCommon spec (Just local)
217-------------------------------------------------------------------------------
219-------------------------------------------------------------------------------
221{-# INLINE recvConnectionTuplesWith #-}
222recvConnectionTuplesWith :: MonadAsync m
223 => Int -> SockSpec -> SockAddr -> SerialT m (Socket, SockAddr)
224recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing
227 listener <- liftIO $ initListener tcpListenQ spec addr
228 r <- liftIO (Net.accept listener `onException` Net.close listener)
229 return $ Just (r, Just listener)
231 step (Just listener) = do
232 r <- liftIO (Net.accept listener `onException` Net.close listener)
233 return $ Just (r, Just listener)
235-- | Start a TCP stream server that listens for connections on the supplied
236-- server address specification (address family, local interface IP address and
237-- port). The server generates a stream of connected sockets. The first
238-- argument is the maximum number of pending connections in the backlog.
241{-# INLINE connections #-}
242connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> SerialT m Socket
243connections tcpListenQ spec addr =
244 fst <$> recvConnectionTuplesWith tcpListenQ spec addr
246-------------------------------------------------------------------------------
248-------------------------------------------------------------------------------
250{-# INLINABLE readArrayUptoWith #-}
252 :: (h -> Ptr Word8 -> Int -> IO Int)
256readArrayUptoWith f size h = do
257 ptr <- mallocPlainForeignPtrBytes size
258 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
259 unsafeWithForeignPtr ptr $ \p -> do
261 let v = A.unsafeFreeze
262 $ mutableArray ptr (p `plusPtr` n) (p `plusPtr` size)
264 -- XXX shrink only if the diff is significant
268-- | Read a byte array from a file handle up to a maximum of the requested
269-- size. If no data is available on the handle it blocks until some data
270-- becomes available. If data is available then it immediately returns that
271-- data without blocking.
274{-# INLINABLE readChunk #-}
275readChunk :: Int -> Socket -> IO (Array Word8)
276readChunk = readArrayUptoWith recvBuf
278-------------------------------------------------------------------------------
280-------------------------------------------------------------------------------
282waitWhen0 :: Int -> Socket -> IO ()
283waitWhen0 0 s = when rtsSupportsBoundThreads $
284#if MIN_VERSION_network(3,1,0)
285 withFdSocket s $ \fd -> threadWaitWrite $ fromIntegral fd
286#elif MIN_VERSION_network(3,0,0)
287 fdSocket s >>= threadWaitWrite . fromIntegral
289 let fd = fdSocket s in threadWaitWrite $ fromIntegral fd
291waitWhen0 _ _ = return ()
293sendAll :: Socket -> Ptr Word8 -> Int -> IO ()
294sendAll _ _ len | len <= 0 = return ()
296 sent <- sendBuf s p len
298 -- assert (sent <= len)
299 when (sent >= 0) $ sendAll s (p `plusPtr` sent) (len - sent)
301{-# INLINABLE writeArrayWith #-}
302writeArrayWith :: Storable a
303 => (h -> Ptr Word8 -> Int -> IO ())
307writeArrayWith _ _ arr | A.length arr == 0 = return ()
308writeArrayWith f h Array{..} = unsafeWithForeignPtr aStart $ \p ->
312 let p = unsafeForeignPtrToPtr aStart
315-- | Write an Array to a file handle.
318{-# INLINABLE writeChunk #-}
319writeChunk :: Storable a => Socket -> Array a -> IO ()
320writeChunk = writeArrayWith sendAll
322-------------------------------------------------------------------------------
323-- Stream of Arrays IO
324-------------------------------------------------------------------------------
326{-# INLINABLE _readChunksUptoWith #-}
327_readChunksUptoWith :: (IsStream t, MonadIO m)
328 => (Int -> h -> IO (Array Word8))
329 -> Int -> h -> t m (Array Word8)
330_readChunksUptoWith f size h = go
332 -- XXX use cons/nil instead
333 go = mkStream $ \_ yld _ stp -> do
334 arr <- liftIO $ f size h
339-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
340-- The maximum size of a single array is limited to @size@.
341-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
343{-# INLINE_NORMAL toChunksWithBufferOf #-}
344toChunksWithBufferOf :: (IsStream t, MonadIO m)
345 => Int -> Socket -> t m (Array Word8)
346-- toChunksWithBufferOf = _readChunksUptoWith readChunk
347toChunksWithBufferOf size h = fromStreamD (D.Stream step ())
349 {-# INLINE_LATE step #-}
351 arr <- liftIO $ readChunk size h
357-- XXX read 'Array a' instead of Word8
359-- | @toChunks h@ reads a stream of arrays from socket handle @h@.
360-- The maximum size of a single array is limited to @defaultChunkSize@.
363{-# INLINE toChunks #-}
364toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)
365toChunks = toChunksWithBufferOf defaultChunkSize
367-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.
368-- Read requests to the socket are performed using a buffer of size @bufsize@.
369-- The size of an array in the resulting stream is always less than or equal to
373{-# INLINE_NORMAL readChunksWithBufferOf #-}
374readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
375readChunksWithBufferOf = Unfold step return
377 {-# INLINE_LATE step #-}
379 arr <- liftIO $ readChunk size h
383 _ -> D.Yield arr (size, h)
385-- | Unfolds a socket into a stream of 'Word8' arrays. Requests to the socket
386-- are performed using a buffer of size
387-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'. The
388-- size of arrays in the resulting stream are therefore less than or equal to
389-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'.
392{-# INLINE readChunks #-}
393readChunks :: MonadIO m => Unfold m Socket (Array Word8)
394readChunks = UF.supplyFirst defaultChunkSize readChunksWithBufferOf
396-------------------------------------------------------------------------------
397-- Read File to Stream
398-------------------------------------------------------------------------------
400-- TODO for concurrent streams implement readahead IO. We can send multiple
401-- read requests at the same time. For serial case we can use async IO. We can
402-- also control the read throughput in mbps or IOPS.
405-- | @readWithBufferOf bufsize handle@ reads a byte stream from a file
406-- handle, reads are performed in chunks of up to @bufsize@. The stream ends
407-- as soon as EOF is encountered.
409{-# INLINE readWithBufferOf #-}
410readWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
411readWithBufferOf chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h
415-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
417-- > read = 'readByChunks' defaultChunkSize
418-- | Generate a stream of elements of the given type from a socket. The
419-- stream ends when EOF is encountered.
422{-# INLINE toBytes #-}
423toBytes :: (IsStream t, MonadIO m) => Socket -> t m Word8
424toBytes = AS.concat . toChunks
426-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests
427-- to the socket are performed using buffers of @bufsize@.
430{-# INLINE readWithBufferOf #-}
431readWithBufferOf :: MonadIO m => Unfold m (Int, Socket) Word8
432readWithBufferOf = UF.many readChunksWithBufferOf A.read
434-- | Unfolds a 'Socket' into a byte stream. IO requests to the socket are
435-- performed in sizes of
436-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'.
440read :: MonadIO m => Unfold m Socket Word8
441read = UF.supplyFirst defaultChunkSize readWithBufferOf
443-------------------------------------------------------------------------------
445-------------------------------------------------------------------------------
447-- | Write a stream of arrays to a handle.
450{-# INLINE putChunks #-}
451putChunks :: (MonadIO m, Storable a)
452 => Socket -> SerialT m (Array a) -> m ()
453putChunks h = S.mapM_ (liftIO . writeChunk h)
455-- | Write a stream of arrays to a socket. Each array in the stream is written
456-- to the socket as a separate IO request.
459{-# INLINE writeChunks #-}
460writeChunks :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()
461writeChunks h = FL.drainBy (liftIO . writeChunk h)
463-- | @writeChunksWithBufferOf bufsize socket@ writes a stream of arrays to
464-- @socket@ after coalescing the adjacent arrays in chunks of @bufsize@.
465-- Multiple arrays are coalesed as long as the total size remains below the
466-- specified size. It never splits an array, if a single array is bigger than
467-- the specified size it emitted as it is.
470{-# INLINE writeChunksWithBufferOf #-}
471writeChunksWithBufferOf :: (MonadIO m, Storable a)
472 => Int -> Socket -> Fold m (Array a) ()
473writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
475-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
478-- Note that if you use a chunk size less than 8K (GHC's default buffer
479-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
480-- do not want buffering to occur at GHC level as well. Same thing applies to
483-- | Like 'write' but provides control over the write buffer. Output will
484-- be written to the IO device as soon as we collect the specified number of
488{-# INLINE putBytesWithBufferOf #-}
489putBytesWithBufferOf :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()
490putBytesWithBufferOf n h m = putChunks h $ AS.arraysOf n m
492-- | Write a byte stream to a socket. Accumulates the input in chunks of
493-- specified number of bytes before writing.
496{-# INLINE writeWithBufferOf #-}
497writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
498writeWithBufferOf n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)
500-- > write = 'writeWithBufferOf' defaultChunkSize
502-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
503-- up to 'defaultChunkSize' before writing. Note that the write behavior
504-- depends on the 'IOMode' and the current seek position of the handle.
507{-# INLINE putBytes #-}
508putBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()
509putBytes = putBytesWithBufferOf defaultChunkSize
511-- | Write a byte stream to a socket. Accumulates the input in chunks of
512-- up to 'defaultChunkSize' bytes before writing.
515-- write = 'writeWithBufferOf' 'defaultChunkSize'
520write :: MonadIO m => Socket -> Fold m Word8 ()
521write = writeWithBufferOf defaultChunkSize
525write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
526write = toHandleWith defaultChunkSize
529-------------------------------------------------------------------------------
530-- IO with encoding/decoding Unicode characters
531-------------------------------------------------------------------------------
535-- > readUtf8 = decodeUtf8 . read
537-- Read a UTF8 encoded stream of unicode characters from a file handle.
540{-# INLINE readUtf8 #-}
541readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
542readUtf8 = decodeUtf8 . read
545-- > writeUtf8 h s = write h $ encodeUtf8 s
547-- Encode a stream of unicode characters to UTF8 and write it to the given file
548-- handle. Default block buffering applies to the writes.
551{-# INLINE writeUtf8 #-}
552writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
553writeUtf8 h s = write h $ encodeUtf8 s
555-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
556-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
557-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
558-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
562{-# INLINE writeUtf8ByLines #-}
563writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
564writeUtf8ByLines = undefined
566-- | Read UTF-8 lines from a file handle and apply the specified fold to each
567-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
570{-# INLINE readLines #-}
571readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
572readLines h f = foldLines (readUtf8 h) f
574-------------------------------------------------------------------------------
575-- Framing on a sequence
576-------------------------------------------------------------------------------
578-- | Read a stream from a file handle and split it into frames delimited by
579-- the specified sequence of elements. The supplied fold is applied on each
583{-# INLINE readFrames #-}
584readFrames :: (IsStream t, MonadIO m, Storable a)
585 => Array a -> Handle -> Fold m a b -> t m b
586readFrames = undefined -- foldFrames . read
588-- | Write a stream to the given file handle buffering up to frames separated
589-- by the given sequence or up to a maximum of @defaultChunkSize@.
592{-# INLINE writeByFrames #-}
593writeByFrames :: (IsStream t, MonadIO m, Storable a)
594 => Array a -> Handle -> t m a -> m ()
595writeByFrames = 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