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.Unboxed.Stream (lpackArraysChunksOf)
96import Streamly.Internal.Data.Fold (Fold)
97import Streamly.Internal.Data.Stream.Type (Stream)
98import Streamly.Internal.Data.Unfold.Type (Unfold(..))
99-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
100import Streamly.Internal.System.IO (defaultChunkSize)
102import qualified Streamly.Data.Array.Unboxed as A (read, length, writeN)
103import qualified Streamly.Data.Fold as FL
104import qualified Streamly.Internal.Data.Array.Unboxed.Type as A
105 (unsafeFreeze, asPtrUnsafe, byteLength, writeNUnsafe)
106import qualified Streamly.Internal.Data.Array.Unboxed.Mut as MArray
107 (Array(..), newPinnedBytes, asPtrUnsafe)
108import qualified Streamly.Internal.Data.Array.Unboxed.Stream as AS
109import qualified Streamly.Internal.Data.Stream as S
110import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
111 (Stream(..), Step(..))
112import qualified Streamly.Data.Unfold as UF
113import qualified Streamly.Internal.Data.Unfold as UF (first, map)
114import qualified Streamly.Internal.Data.Stream.StreamK.Type as K (mkStream)
116-- | @'forSocketM' action socket@ runs the monadic computation @action@ passing
117-- the socket handle to it. The handle will be closed on exit from
118-- 'forSocketM', whether by normal termination or by raising an exception. If
119-- closing the handle raises an exception, then this exception will be raised
120-- by 'forSocketM' rather than any exception raised by 'action'.
123{-# INLINE forSocketM #-}
124forSocketM :: (MonadMask m, MonadIO m) => (Socket -> m ()) -> Socket -> m ()
125forSocketM f sk = finally (f sk) (liftIO (Net.close sk))
127-- | Like 'forSocketM' but runs a streaming computation instead of a monadic
130-- /Inhibits stream fusion/
133{-# INLINE withSocket #-}
134withSocket :: (MonadAsync m, MonadCatch m)
135 => Socket -> (Socket -> Stream m a) -> Stream m a
136withSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)
138-------------------------------------------------------------------------------
140-------------------------------------------------------------------------------
142-- XXX Protocol specific socket options should be separated from socket level
145-- | Specify the socket protocol details.
146data SockSpec = SockSpec
148 sockFamily :: !Family
149 , sockType :: !SocketType
150 , sockProto :: !ProtocolNumber
151 , sockOpts :: ![(SocketOption, Int)]
154initListener :: Int -> SockSpec -> SockAddr -> IO Socket
155initListener listenQLen SockSpec{..} addr =
157 sock <- socket sockFamily sockType sockProto
158 use sock `onException` Net.close sock
164 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
166 Net.listen sock listenQLen
168{-# INLINE listenTuples #-}
169listenTuples :: MonadIO m
170 => Unfold m (Int, SockSpec, SockAddr) (Socket, SockAddr)
171listenTuples = Unfold step inject
173 inject (listenQLen, spec, addr) =
174 liftIO $ initListener listenQLen spec addr
177 r <- liftIO (Net.accept listener `onException` Net.close listener)
178 return $ D.Yield r listener
180-- | Unfold a three tuple @(listenQLen, spec, addr)@ into a stream of connected
181-- protocol sockets corresponding to incoming connections. @listenQLen@ is the
182-- maximum number of pending connections in the backlog. @spec@ is the socket
183-- protocol and options specification and @addr@ is the protocol address where
184-- the server listens for incoming connections.
188accept :: MonadIO m => Unfold m (Int, SockSpec, SockAddr) Socket
189accept = UF.map fst listenTuples
191{-# INLINE connectCommon #-}
192connectCommon :: SockSpec -> Maybe SockAddr -> SockAddr -> IO Socket
193connectCommon SockSpec{..} local remote = withSocketsDo $ do
194 sock <- socket sockFamily sockType sockProto
195 use sock `onException` Net.close sock
201 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
202 forM_ local (bind sock)
203 Net.connect sock remote
205-- | Connect to a remote host using the given socket specification and remote
206-- address. Returns a connected socket or throws an exception.
210{-# INLINE connect #-}
211connect :: SockSpec -> SockAddr -> IO Socket
212connect spec = connectCommon spec Nothing
214-- | Connect to a remote host using the given socket specification, a local
215-- address to bind to and a remote address to connect to. Returns a connected
216-- socket or throws an exception.
220{-# INLINE connectFrom #-}
221connectFrom :: SockSpec -> SockAddr -> SockAddr -> IO Socket
222connectFrom spec local = connectCommon spec (Just local)
224-------------------------------------------------------------------------------
226-------------------------------------------------------------------------------
228{-# INLINE recvConnectionTuplesWith #-}
229recvConnectionTuplesWith :: MonadAsync m
230 => Int -> SockSpec -> SockAddr -> Stream m (Socket, SockAddr)
231recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing
234 listener <- liftIO $ initListener tcpListenQ spec addr
235 r <- liftIO (Net.accept listener `onException` Net.close listener)
236 return $ Just (r, Just listener)
238 step (Just listener) = do
239 r <- liftIO (Net.accept listener `onException` Net.close listener)
240 return $ Just (r, Just listener)
242-- | Start a TCP stream server that listens for connections on the supplied
243-- server address specification (address family, local interface IP address and
244-- port). The server generates a stream of connected sockets. The first
245-- argument is the maximum number of pending connections in the backlog.
248{-# INLINE connections #-}
249connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> Stream m Socket
250connections tcpListenQ spec addr =
251 fst <$> recvConnectionTuplesWith tcpListenQ spec addr
253-------------------------------------------------------------------------------
255-------------------------------------------------------------------------------
257{-# INLINABLE readArrayUptoWith #-}
259 :: (h -> Ptr Word8 -> Int -> IO Int)
263readArrayUptoWith f size h = do
264 arr <- MArray.newPinnedBytes size
265 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
266 MArray.asPtrUnsafe arr $ \p -> do
268 let v = A.unsafeFreeze
269 $ arr { MArray.arrEnd = n, MArray.arrBound = size }
271 -- XXX shrink only if the diff is significant
275-- | Read a byte array from a file handle up to a maximum of the requested
276-- size. If no data is available on the handle it blocks until some data
277-- becomes available. If data is available then it immediately returns that
278-- data without blocking.
281{-# INLINABLE readChunk #-}
282readChunk :: Int -> Socket -> IO (Array Word8)
283readChunk = readArrayUptoWith recvBuf
285-------------------------------------------------------------------------------
287-------------------------------------------------------------------------------
289waitWhen0 :: Int -> Socket -> IO ()
290waitWhen0 0 s = when rtsSupportsBoundThreads $
291#if MIN_VERSION_network(3,1,0)
292 withFdSocket s $ \fd -> threadWaitWrite $ fromIntegral fd
293#elif MIN_VERSION_network(3,0,0)
294 fdSocket s >>= threadWaitWrite . fromIntegral
296 let fd = fdSocket s in threadWaitWrite $ fromIntegral fd
298waitWhen0 _ _ = return ()
300sendAll :: Socket -> Ptr Word8 -> Int -> IO ()
301sendAll _ _ len | len <= 0 = return ()
303 sent <- sendBuf s p len
305 -- assert (sent <= len)
306 when (sent >= 0) $ sendAll s (p `plusPtr` sent) (len - sent)
308{-# INLINABLE writeArrayWith #-}
309writeArrayWith :: Unboxed a
310 => (h -> Ptr Word8 -> Int -> IO ())
314writeArrayWith _ _ arr | A.length arr == 0 = return ()
315writeArrayWith f h arr = A.asPtrUnsafe arr $ \ptr -> f h (castPtr ptr) aLen
319 aLen = A.byteLength arr
321-- | Write an Array to a file handle.
324{-# INLINABLE writeChunk #-}
325writeChunk :: Unboxed a => Socket -> Array a -> IO ()
326writeChunk = writeArrayWith sendAll
328-------------------------------------------------------------------------------
329-- Stream of Arrays IO
330-------------------------------------------------------------------------------
332{-# INLINE _readChunksUptoWith #-}
333_readChunksUptoWith :: (MonadIO m)
334 => (Int -> h -> IO (Array Word8))
335 -> Int -> h -> Stream m (Array Word8)
336_readChunksUptoWith f size h = S.fromStreamK go
338 -- XXX use cons/nil instead
339 go = K.mkStream $ \_ yld _ stp -> do
340 arr <- liftIO $ f size h
345-- | @toChunksWith size h@ reads a stream of arrays from file handle @h@.
346-- The maximum size of a single array is limited to @size@.
347-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
349{-# INLINE_NORMAL toChunksWith #-}
350toChunksWith :: (MonadIO m)
351 => Int -> Socket -> Stream m (Array Word8)
352-- toChunksWith = _readChunksUptoWith readChunk
353toChunksWith size h = S.fromStreamD (D.Stream step ())
355 {-# INLINE_LATE step #-}
357 arr <- liftIO $ readChunk size h
363-- XXX read 'Array a' instead of Word8
365-- | @toChunks h@ reads a stream of arrays from socket handle @h@.
366-- The maximum size of a single array is limited to @defaultChunkSize@.
369{-# INLINE toChunks #-}
370toChunks :: (MonadIO m) => Socket -> Stream m (Array Word8)
371toChunks = toChunksWith defaultChunkSize
373-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.
374-- Read requests to the socket are performed using a buffer of size @bufsize@.
375-- The size of an array in the resulting stream is always less than or equal to
379{-# INLINE_NORMAL readChunksWith #-}
380readChunksWith :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
381readChunksWith = Unfold step return
383 {-# INLINE_LATE step #-}
385 arr <- liftIO $ readChunk size h
389 _ -> D.Yield arr (size, h)
391-- | Same as 'readChunksWith'
394{-# DEPRECATED readChunksWithBufferOf "Please use 'readChunksWith' instead" #-}
395{-# INLINE_NORMAL readChunksWithBufferOf #-}
396readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
397readChunksWithBufferOf = readChunksWith
399-- | Unfolds a socket into a stream of 'Word8' arrays. Requests to the socket
400-- are performed using a buffer of size
401-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'. The
402-- size of arrays in the resulting stream are therefore less than or equal to
403-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'.
406{-# INLINE readChunks #-}
407readChunks :: MonadIO m => Unfold m Socket (Array Word8)
408readChunks = UF.first defaultChunkSize readChunksWith
410-------------------------------------------------------------------------------
411-- Read File to Stream
412-------------------------------------------------------------------------------
414-- TODO for concurrent streams implement readahead IO. We can send multiple
415-- read requests at the same time. For serial case we can use async IO. We can
416-- also control the read throughput in mbps or IOPS.
419-- | @readWith bufsize handle@ reads a byte stream from a file
420-- handle, reads are performed in chunks of up to @bufsize@. The stream ends
421-- as soon as EOF is encountered.
423{-# INLINE readWith #-}
424readWith :: (MonadIO m) => Int -> Handle -> Stream m Word8
425readWith chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h
429-- read :: (MonadIO m, Unboxed a) => Handle -> Stream m a
431-- > read = 'readByChunks' defaultChunkSize
432-- | Generate a stream of elements of the given type from a socket. The
433-- stream ends when EOF is encountered.
436{-# INLINE toBytes #-}
437toBytes :: MonadIO m => Socket -> Stream m Word8
438toBytes = AS.concat . toChunks
440-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests
441-- to the socket are performed using buffers of @bufsize@.
444{-# INLINE readWith #-}
445readWith :: MonadIO m => Unfold m (Int, Socket) Word8
446readWith = UF.many A.read readChunksWith
448-- | Same as 'readWith'
451{-# DEPRECATED readWithBufferOf "Please use 'readWith' instead" #-}
452{-# INLINE readWithBufferOf #-}
453readWithBufferOf :: MonadIO m => Unfold m (Int, Socket) Word8
454readWithBufferOf = readWith
456-- | Unfolds a 'Socket' into a byte stream. IO requests to the socket are
457-- performed in sizes of
458-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'.
462read :: MonadIO m => Unfold m Socket Word8
463read = UF.first defaultChunkSize readWith
465-------------------------------------------------------------------------------
467-------------------------------------------------------------------------------
469-- | Write a stream of arrays to a handle.
472{-# INLINE putChunks #-}
473putChunks :: (MonadIO m, Unboxed a)
474 => Socket -> Stream m (Array a) -> m ()
475putChunks h = S.fold (FL.drainBy (liftIO . writeChunk h))
477-- | Write a stream of arrays to a socket. Each array in the stream is written
478-- to the socket as a separate IO request.
481{-# INLINE writeChunks #-}
482writeChunks :: (MonadIO m, Unboxed a) => Socket -> Fold m (Array a) ()
483writeChunks h = FL.drainBy (liftIO . writeChunk h)
485-- | @writeChunksWith bufsize socket@ writes a stream of arrays to
486-- @socket@ after coalescing the adjacent arrays in chunks of @bufsize@.
487-- Multiple arrays are coalesed as long as the total size remains below the
488-- specified size. It never splits an array, if a single array is bigger than
489-- the specified size it emitted as it is.
492{-# INLINE writeChunksWith #-}
493writeChunksWith :: (MonadIO m, Unboxed a)
494 => Int -> Socket -> Fold m (Array a) ()
495writeChunksWith n h = lpackArraysChunksOf n (writeChunks h)
497-- | Same as 'writeChunksWith'
500{-# DEPRECATED writeChunksWithBufferOf "Please use 'writeChunksWith' instead" #-}
501{-# INLINE writeChunksWithBufferOf #-}
502writeChunksWithBufferOf :: (MonadIO m, Unboxed a)
503 => Int -> Socket -> Fold m (Array a) ()
504writeChunksWithBufferOf = writeChunksWith
506-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
509-- Note that if you use a chunk size less than 8K (GHC's default buffer
510-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
511-- do not want buffering to occur at GHC level as well. Same thing applies to
514-- | Like 'write' but provides control over the write buffer. Output will
515-- be written to the IO device as soon as we collect the specified number of
519{-# INLINE putBytesWith #-}
520putBytesWith :: MonadIO m => Int -> Socket -> Stream m Word8 -> m ()
521putBytesWith n h m = putChunks h $ AS.arraysOf n m
523-- | Write a byte stream to a socket. Accumulates the input in chunks of
524-- specified number of bytes before writing.
527{-# INLINE writeWith #-}
528writeWith :: MonadIO m => Int -> Socket -> Fold m Word8 ()
529writeWith n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)
531-- | Same as 'writeWith'
534{-# DEPRECATED writeWithBufferOf "Please use 'writeWith' instead" #-}
535{-# INLINE writeWithBufferOf #-}
536writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
537writeWithBufferOf = writeWith
539-- | Write a stream of 'Maybe' values. Keep buffering the 'Just' values in an
540-- array. Write the array to the 'Handle' as soon as a 'Nothing' is encountered
541-- or the buffer size exceeds the specified limit.
544{-# INLINE writeMaybesWith #-}
545writeMaybesWith :: (MonadIO m )
546 => Int -> Socket -> Fold m (Maybe Word8) ()
548 let writeNJusts = FL.lmap fromJust $ A.writeN n
549 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts
550 in FL.many writeOnNothing (writeChunks h)
552-- > write = 'writeWith' defaultChunkSize
554-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
555-- up to 'defaultChunkSize' before writing. Note that the write behavior
556-- depends on the 'IOMode' and the current seek position of the handle.
559{-# INLINE putBytes #-}
560putBytes :: MonadIO m => Socket -> Stream m Word8 -> m ()
561putBytes = putBytesWith defaultChunkSize
563-- | Write a byte stream to a socket. Accumulates the input in chunks of
564-- up to 'defaultChunkSize' bytes before writing.
567-- write = 'writeWith' 'defaultChunkSize'
572write :: MonadIO m => Socket -> Fold m Word8 ()
573write = writeWith defaultChunkSize
577write :: (MonadIO m, Unboxed a) => Handle -> Stream m a -> m ()
578write = toHandleWith defaultChunkSize
581-------------------------------------------------------------------------------
582-- IO with encoding/decoding Unicode characters
583-------------------------------------------------------------------------------
587-- > readUtf8 = decodeUtf8 . read
589-- Read a UTF8 encoded stream of unicode characters from a file handle.
592{-# INLINE readUtf8 #-}
593readUtf8 :: MonadIO m => Handle -> Stream m Char
594readUtf8 = decodeUtf8 . read
597-- > writeUtf8 h s = write h $ encodeUtf8 s
599-- Encode a stream of unicode characters to UTF8 and write it to the given file
600-- handle. Default block buffering applies to the writes.
603{-# INLINE writeUtf8 #-}
604writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()
605writeUtf8 h s = write h $ encodeUtf8 s
607-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
608-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
609-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
610-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
614{-# INLINE writeUtf8ByLines #-}
615writeUtf8ByLines :: MonadIO m => Handle -> Stream m Char -> m ()
616writeUtf8ByLines = undefined
618-- | Read UTF-8 lines from a file handle and apply the specified fold to each
619-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
622{-# INLINE readLines #-}
623readLines :: MonadIO m => Handle -> Fold m Char b -> Stream m b
624readLines h f = foldLines (readUtf8 h) f
626-------------------------------------------------------------------------------
627-- Framing on a sequence
628-------------------------------------------------------------------------------
630-- | Read a stream from a file handle and split it into frames delimited by
631-- the specified sequence of elements. The supplied fold is applied on each
635{-# INLINE readFrames #-}
636readFrames :: (MonadIO m, Unboxed a)
637 => Array a -> Handle -> Fold m a b -> Stream m b
638readFrames = undefined -- foldFrames . read
640-- | Write a stream to the given file handle buffering up to frames separated
641-- by the given sequence or up to a maximum of @defaultChunkSize@.
644{-# INLINE writeByFrames #-}
645writeByFrames :: (MonadIO m, Unboxed a)
646 => Array a -> Handle -> Stream m a -> m ()
647writeByFrames = 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