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
44 -- * Write to connection
61 -- reading/writing datagrams
65 , readChunksWithBufferOf
67 , writeChunksWithBufferOf
71import Control.Concurrent (threadWaitWrite, rtsSupportsBoundThreads)
72import Control.Exception (onException)
73import Control.Monad.Catch (MonadCatch, finally, MonadMask)
74import Control.Monad.IO.Class (MonadIO(..))
75import Control.Monad (forM_, when)
76import Data.Maybe (isNothing, fromJust)
77import Data.Word (Word8)
78import Foreign.Ptr (plusPtr, Ptr, castPtr)
79import Streamly.Internal.Data.Unboxed (Unboxed)
81 (Socket, SocketOption(..), Family(..), SockAddr(..),
82 ProtocolNumber, withSocketsDo, SocketType(..), socket, bind,
83 setSocketOption, sendBuf, recvBuf)
84#if MIN_VERSION_network(3,1,0)
85import Network.Socket (withFdSocket)
87import Network.Socket (fdSocket)
89import Prelude hiding (read)
91import qualified Network.Socket as Net
93import Streamly.Internal.Control.Concurrent (MonadAsync)
94import Streamly.Internal.Data.Array.Unboxed.Type (Array(..))
95import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)
96import Streamly.Internal.Data.Fold (Fold)
97import Streamly.Internal.Data.Stream.IsStream.Type
98 (IsStream, mkStream, fromStreamD)
99import Streamly.Internal.Data.Stream.Type (Stream)
100import Streamly.Internal.Data.Unfold.Type (Unfold(..))
101-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
102import Streamly.Internal.System.IO (defaultChunkSize)
104import qualified Streamly.Internal.Data.Array.Unboxed as A
105import qualified Streamly.Internal.Data.Array.Unboxed.Type as A
106import qualified Streamly.Internal.Data.Array.Unboxed.Mut.Type as MArray
107import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
108import qualified Streamly.Internal.Data.Fold as FL
109import qualified Streamly.Internal.Data.Stream.IsStream as S
110import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
111import qualified Streamly.Internal.Data.Unfold as UF
113-- | @'forSocketM' action socket@ runs the monadic computation @action@ passing
114-- the socket handle to it. The handle will be closed on exit from
115-- 'forSocketM', whether by normal termination or by raising an exception. If
116-- closing the handle raises an exception, then this exception will be raised
117-- by 'forSocketM' rather than any exception raised by 'action'.
120{-# INLINE forSocketM #-}
121forSocketM :: (MonadMask m, MonadIO m) => (Socket -> m ()) -> Socket -> m ()
122forSocketM f sk = finally (f sk) (liftIO (Net.close sk))
124-- | Like 'forSocketM' but runs a streaming computation instead of a monadic
127-- /Inhibits stream fusion/
130{-# INLINE withSocket #-}
131withSocket :: (IsStream t, MonadAsync m, MonadCatch m)
132 => Socket -> (Socket -> t m a) -> t m a
133withSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)
135-------------------------------------------------------------------------------
137-------------------------------------------------------------------------------
139-- XXX Protocol specific socket options should be separated from socket level
142-- | Specify the socket protocol details.
143data SockSpec = SockSpec
145 sockFamily :: !Family
146 , sockType :: !SocketType
147 , sockProto :: !ProtocolNumber
148 , sockOpts :: ![(SocketOption, Int)]
151initListener :: Int -> SockSpec -> SockAddr -> IO Socket
152initListener listenQLen SockSpec{..} addr =
154 sock <- socket sockFamily sockType sockProto
155 use sock `onException` Net.close sock
161 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
163 Net.listen sock listenQLen
165{-# INLINE listenTuples #-}
166listenTuples :: MonadIO m
167 => Unfold m (Int, SockSpec, SockAddr) (Socket, SockAddr)
168listenTuples = Unfold step inject
170 inject (listenQLen, spec, addr) =
171 liftIO $ initListener listenQLen spec addr
174 r <- liftIO (Net.accept listener `onException` Net.close listener)
175 return $ D.Yield r listener
177-- | Unfold a three tuple @(listenQLen, spec, addr)@ into a stream of connected
178-- protocol sockets corresponding to incoming connections. @listenQLen@ is the
179-- maximum number of pending connections in the backlog. @spec@ is the socket
180-- protocol and options specification and @addr@ is the protocol address where
181-- the server listens for incoming connections.
185accept :: MonadIO m => Unfold m (Int, SockSpec, SockAddr) Socket
186accept = UF.map fst listenTuples
188{-# INLINE connectCommon #-}
189connectCommon :: SockSpec -> Maybe SockAddr -> SockAddr -> IO Socket
190connectCommon SockSpec{..} local remote = withSocketsDo $ do
191 sock <- socket sockFamily sockType sockProto
192 use sock `onException` Net.close sock
198 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
199 forM_ local (bind sock)
200 Net.connect sock remote
202-- | Connect to a remote host using the given socket specification and remote
203-- address. Returns a connected socket or throws an exception.
207{-# INLINE connect #-}
208connect :: SockSpec -> SockAddr -> IO Socket
209connect spec = connectCommon spec Nothing
211-- | Connect to a remote host using the given socket specification, a local
212-- address to bind to and a remote address to connect to. Returns a connected
213-- socket or throws an exception.
217{-# INLINE connectFrom #-}
218connectFrom :: SockSpec -> SockAddr -> SockAddr -> IO Socket
219connectFrom spec local = connectCommon spec (Just local)
221-------------------------------------------------------------------------------
223-------------------------------------------------------------------------------
225{-# INLINE recvConnectionTuplesWith #-}
226recvConnectionTuplesWith :: MonadAsync m
227 => Int -> SockSpec -> SockAddr -> Stream m (Socket, SockAddr)
228recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing
231 listener <- liftIO $ initListener tcpListenQ spec addr
232 r <- liftIO (Net.accept listener `onException` Net.close listener)
233 return $ Just (r, Just listener)
235 step (Just listener) = do
236 r <- liftIO (Net.accept listener `onException` Net.close listener)
237 return $ Just (r, Just listener)
239-- | Start a TCP stream server that listens for connections on the supplied
240-- server address specification (address family, local interface IP address and
241-- port). The server generates a stream of connected sockets. The first
242-- argument is the maximum number of pending connections in the backlog.
245{-# INLINE connections #-}
246connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> Stream m Socket
247connections tcpListenQ spec addr =
248 fst <$> recvConnectionTuplesWith tcpListenQ spec addr
250-------------------------------------------------------------------------------
252-------------------------------------------------------------------------------
254{-# INLINABLE readArrayUptoWith #-}
256 :: (h -> Ptr Word8 -> Int -> IO Int)
260readArrayUptoWith f size h = do
261 arr <- MArray.newPinnedArrayBytes size
262 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
263 MArray.asPtrUnsafe arr $ \p -> do
265 let v = A.unsafeFreeze
266 $ arr { MArray.arrEnd = n, MArray.arrBound = size }
268 -- XXX shrink only if the diff is significant
272-- | Read a byte array from a file handle up to a maximum of the requested
273-- size. If no data is available on the handle it blocks until some data
274-- becomes available. If data is available then it immediately returns that
275-- data without blocking.
278{-# INLINABLE readChunk #-}
279readChunk :: Int -> Socket -> IO (Array Word8)
280readChunk = readArrayUptoWith recvBuf
282-------------------------------------------------------------------------------
284-------------------------------------------------------------------------------
286waitWhen0 :: Int -> Socket -> IO ()
287waitWhen0 0 s = when rtsSupportsBoundThreads $
288#if MIN_VERSION_network(3,1,0)
289 withFdSocket s $ \fd -> threadWaitWrite $ fromIntegral fd
290#elif MIN_VERSION_network(3,0,0)
291 fdSocket s >>= threadWaitWrite . fromIntegral
293 let fd = fdSocket s in threadWaitWrite $ fromIntegral fd
295waitWhen0 _ _ = return ()
297sendAll :: Socket -> Ptr Word8 -> Int -> IO ()
298sendAll _ _ len | len <= 0 = return ()
300 sent <- sendBuf s p len
302 -- assert (sent <= len)
303 when (sent >= 0) $ sendAll s (p `plusPtr` sent) (len - sent)
305{-# INLINABLE writeArrayWith #-}
306writeArrayWith :: Unboxed a
307 => (h -> Ptr Word8 -> Int -> IO ())
311writeArrayWith _ _ arr | A.length arr == 0 = return ()
312writeArrayWith f h arr = A.asPtrUnsafe arr $ \ptr -> f h (castPtr ptr) aLen
316 aLen = A.byteLength arr
318-- | Write an Array to a file handle.
321{-# INLINABLE writeChunk #-}
322writeChunk :: Unboxed a => Socket -> Array a -> IO ()
323writeChunk = writeArrayWith sendAll
325-------------------------------------------------------------------------------
326-- Stream of Arrays IO
327-------------------------------------------------------------------------------
329{-# INLINE _readChunksUptoWith #-}
330_readChunksUptoWith :: (IsStream t, MonadIO m)
331 => (Int -> h -> IO (Array Word8))
332 -> Int -> h -> t m (Array Word8)
333_readChunksUptoWith f size h = go
335 -- XXX use cons/nil instead
336 go = mkStream $ \_ yld _ stp -> do
337 arr <- liftIO $ f size h
342-- | @toChunksWith size h@ reads a stream of arrays from file handle @h@.
343-- The maximum size of a single array is limited to @size@.
344-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
346{-# INLINE_NORMAL toChunksWith #-}
347toChunksWith :: (IsStream t, MonadIO m)
348 => Int -> Socket -> t m (Array Word8)
349-- toChunksWith = _readChunksUptoWith readChunk
350toChunksWith size h = fromStreamD (D.Stream step ())
352 {-# INLINE_LATE step #-}
354 arr <- liftIO $ readChunk size h
360-- XXX read 'Array a' instead of Word8
362-- | @toChunks h@ reads a stream of arrays from socket handle @h@.
363-- The maximum size of a single array is limited to @defaultChunkSize@.
366{-# INLINE toChunks #-}
367toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)
368toChunks = toChunksWith defaultChunkSize
370-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.
371-- Read requests to the socket are performed using a buffer of size @bufsize@.
372-- The size of an array in the resulting stream is always less than or equal to
376{-# INLINE_NORMAL readChunksWith #-}
377readChunksWith :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
378readChunksWith = Unfold step return
380 {-# INLINE_LATE step #-}
382 arr <- liftIO $ readChunk size h
386 _ -> D.Yield arr (size, h)
388-- | Same as 'readChunksWith'
391{-# DEPRECATED readChunksWithBufferOf "Please use 'readChunksWith' instead" #-}
392{-# INLINE_NORMAL readChunksWithBufferOf #-}
393readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
394readChunksWithBufferOf = readChunksWith
396-- | Unfolds a socket into a stream of 'Word8' arrays. Requests to the socket
397-- are performed using a buffer of size
398-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'. The
399-- size of arrays in the resulting stream are therefore less than or equal to
400-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'.
403{-# INLINE readChunks #-}
404readChunks :: MonadIO m => Unfold m Socket (Array Word8)
405readChunks = UF.first defaultChunkSize readChunksWith
407-------------------------------------------------------------------------------
408-- Read File to Stream
409-------------------------------------------------------------------------------
411-- TODO for concurrent streams implement readahead IO. We can send multiple
412-- read requests at the same time. For serial case we can use async IO. We can
413-- also control the read throughput in mbps or IOPS.
416-- | @readWith bufsize handle@ reads a byte stream from a file
417-- handle, reads are performed in chunks of up to @bufsize@. The stream ends
418-- as soon as EOF is encountered.
420{-# INLINE readWith #-}
421readWith :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
422readWith chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h
426-- read :: (IsStream t, MonadIO m, Unboxed a) => Handle -> t m a
428-- > read = 'readByChunks' defaultChunkSize
429-- | Generate a stream of elements of the given type from a socket. The
430-- stream ends when EOF is encountered.
433{-# INLINE toBytes #-}
434toBytes :: (IsStream t, MonadIO m) => Socket -> t m Word8
435toBytes = AS.concat . toChunks
437-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests
438-- to the socket are performed using buffers of @bufsize@.
441{-# INLINE readWith #-}
442readWith :: MonadIO m => Unfold m (Int, Socket) Word8
443readWith = UF.many A.read readChunksWith
445-- | Same as 'readWith'
448{-# DEPRECATED readWithBufferOf "Please use 'readWith' instead" #-}
449{-# INLINE readWithBufferOf #-}
450readWithBufferOf :: MonadIO m => Unfold m (Int, Socket) Word8
451readWithBufferOf = readWith
453-- | Unfolds a 'Socket' into a byte stream. IO requests to the socket are
454-- performed in sizes of
455-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'.
459read :: MonadIO m => Unfold m Socket Word8
460read = UF.first defaultChunkSize readWith
462-------------------------------------------------------------------------------
464-------------------------------------------------------------------------------
466-- | Write a stream of arrays to a handle.
469{-# INLINE putChunks #-}
470putChunks :: (MonadIO m, Unboxed a)
471 => Socket -> Stream m (Array a) -> m ()
472putChunks h = S.mapM_ (liftIO . writeChunk h)
474-- | Write a stream of arrays to a socket. Each array in the stream is written
475-- to the socket as a separate IO request.
478{-# INLINE writeChunks #-}
479writeChunks :: (MonadIO m, Unboxed a) => Socket -> Fold m (Array a) ()
480writeChunks h = FL.drainBy (liftIO . writeChunk h)
482-- | @writeChunksWith bufsize socket@ writes a stream of arrays to
483-- @socket@ after coalescing the adjacent arrays in chunks of @bufsize@.
484-- Multiple arrays are coalesed as long as the total size remains below the
485-- specified size. It never splits an array, if a single array is bigger than
486-- the specified size it emitted as it is.
489{-# INLINE writeChunksWith #-}
490writeChunksWith :: (MonadIO m, Unboxed a)
491 => Int -> Socket -> Fold m (Array a) ()
492writeChunksWith n h = lpackArraysChunksOf n (writeChunks h)
494-- | Same as 'writeChunksWith'
497{-# DEPRECATED writeChunksWithBufferOf "Please use 'writeChunksWith' instead" #-}
498{-# INLINE writeChunksWithBufferOf #-}
499writeChunksWithBufferOf :: (MonadIO m, Unboxed a)
500 => Int -> Socket -> Fold m (Array a) ()
501writeChunksWithBufferOf = writeChunksWith
503-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
506-- Note that if you use a chunk size less than 8K (GHC's default buffer
507-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
508-- do not want buffering to occur at GHC level as well. Same thing applies to
511-- | Like 'write' but provides control over the write buffer. Output will
512-- be written to the IO device as soon as we collect the specified number of
516{-# INLINE putBytesWith #-}
517putBytesWith :: MonadIO m => Int -> Socket -> Stream m Word8 -> m ()
518putBytesWith n h m = putChunks h $ AS.arraysOf n m
520-- | Write a byte stream to a socket. Accumulates the input in chunks of
521-- specified number of bytes before writing.
524{-# INLINE writeWith #-}
525writeWith :: MonadIO m => Int -> Socket -> Fold m Word8 ()
526writeWith n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)
528-- | Same as 'writeWith'
531{-# DEPRECATED writeWithBufferOf "Please use 'writeWith' instead" #-}
532{-# INLINE writeWithBufferOf #-}
533writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
534writeWithBufferOf = writeWith
536-- | Write a stream of 'Maybe' values. Keep buffering the 'Just' values in an
537-- array. Write the array to the 'Handle' as soon as a 'Nothing' is encountered
538-- or the buffer size exceeds the specified limit.
541{-# INLINE writeMaybesWith #-}
542writeMaybesWith :: (MonadIO m )
543 => Int -> Socket -> Fold m (Maybe Word8) ()
545 let writeNJusts = FL.lmap fromJust $ A.writeN n
546 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts
547 in FL.many writeOnNothing (writeChunks h)
549-- > write = 'writeWith' defaultChunkSize
551-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
552-- up to 'defaultChunkSize' before writing. Note that the write behavior
553-- depends on the 'IOMode' and the current seek position of the handle.
556{-# INLINE putBytes #-}
557putBytes :: MonadIO m => Socket -> Stream m Word8 -> m ()
558putBytes = putBytesWith defaultChunkSize
560-- | Write a byte stream to a socket. Accumulates the input in chunks of
561-- up to 'defaultChunkSize' bytes before writing.
564-- write = 'writeWith' 'defaultChunkSize'
569write :: MonadIO m => Socket -> Fold m Word8 ()
570write = writeWith defaultChunkSize
574write :: (MonadIO m, Unboxed a) => Handle -> Stream m a -> m ()
575write = toHandleWith defaultChunkSize
578-------------------------------------------------------------------------------
579-- IO with encoding/decoding Unicode characters
580-------------------------------------------------------------------------------
584-- > readUtf8 = decodeUtf8 . read
586-- Read a UTF8 encoded stream of unicode characters from a file handle.
589{-# INLINE readUtf8 #-}
590readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
591readUtf8 = decodeUtf8 . read
594-- > writeUtf8 h s = write h $ encodeUtf8 s
596-- Encode a stream of unicode characters to UTF8 and write it to the given file
597-- handle. Default block buffering applies to the writes.
600{-# INLINE writeUtf8 #-}
601writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()
602writeUtf8 h s = write h $ encodeUtf8 s
604-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
605-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
606-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
607-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
611{-# INLINE writeUtf8ByLines #-}
612writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
613writeUtf8ByLines = undefined
615-- | Read UTF-8 lines from a file handle and apply the specified fold to each
616-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
619{-# INLINE readLines #-}
620readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
621readLines h f = foldLines (readUtf8 h) f
623-------------------------------------------------------------------------------
624-- Framing on a sequence
625-------------------------------------------------------------------------------
627-- | Read a stream from a file handle and split it into frames delimited by
628-- the specified sequence of elements. The supplied fold is applied on each
632{-# INLINE readFrames #-}
633readFrames :: (IsStream t, MonadIO m, Unboxed a)
634 => Array a -> Handle -> Fold m a b -> t m b
635readFrames = undefined -- foldFrames . read
637-- | Write a stream to the given file handle buffering up to frames separated
638-- by the given sequence or up to a maximum of @defaultChunkSize@.
641{-# INLINE writeByFrames #-}
642writeByFrames :: (IsStream t, MonadIO m, Unboxed a)
643 => Array a -> Handle -> t m a -> m ()
644writeByFrames = 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