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.Data.Array.Unboxed as A (read, length, writeN)
105import qualified Streamly.Data.Fold as FL
106import qualified Streamly.Internal.Data.Array.Unboxed.Type as A
107 (unsafeFreeze, asPtrUnsafe, byteLength, writeNUnsafe)
108import qualified Streamly.Internal.Data.Array.Unboxed.Mut as MArray
109 (Array(..), newPinnedArrayBytes, asPtrUnsafe)
110import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
111import qualified Streamly.Internal.Data.Stream.IsStream as S
112import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
113 (Stream(..), Step(..))
114import qualified Streamly.Data.Unfold as UF
115import qualified Streamly.Internal.Data.Unfold as UF (first, map)
117-- | @'forSocketM' action socket@ runs the monadic computation @action@ passing
118-- the socket handle to it. The handle will be closed on exit from
119-- 'forSocketM', whether by normal termination or by raising an exception. If
120-- closing the handle raises an exception, then this exception will be raised
121-- by 'forSocketM' rather than any exception raised by 'action'.
124{-# INLINE forSocketM #-}
125forSocketM :: (MonadMask m, MonadIO m) => (Socket -> m ()) -> Socket -> m ()
126forSocketM f sk = finally (f sk) (liftIO (Net.close sk))
128-- | Like 'forSocketM' but runs a streaming computation instead of a monadic
131-- /Inhibits stream fusion/
134{-# INLINE withSocket #-}
135withSocket :: (IsStream t, MonadAsync m, MonadCatch m)
136 => Socket -> (Socket -> t m a) -> t m a
137withSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)
139-------------------------------------------------------------------------------
141-------------------------------------------------------------------------------
143-- XXX Protocol specific socket options should be separated from socket level
146-- | Specify the socket protocol details.
147data SockSpec = SockSpec
149 sockFamily :: !Family
150 , sockType :: !SocketType
151 , sockProto :: !ProtocolNumber
152 , sockOpts :: ![(SocketOption, Int)]
155initListener :: Int -> SockSpec -> SockAddr -> IO Socket
156initListener listenQLen SockSpec{..} addr =
158 sock <- socket sockFamily sockType sockProto
159 use sock `onException` Net.close sock
165 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
167 Net.listen sock listenQLen
169{-# INLINE listenTuples #-}
170listenTuples :: MonadIO m
171 => Unfold m (Int, SockSpec, SockAddr) (Socket, SockAddr)
172listenTuples = Unfold step inject
174 inject (listenQLen, spec, addr) =
175 liftIO $ initListener listenQLen spec addr
178 r <- liftIO (Net.accept listener `onException` Net.close listener)
179 return $ D.Yield r listener
181-- | Unfold a three tuple @(listenQLen, spec, addr)@ into a stream of connected
182-- protocol sockets corresponding to incoming connections. @listenQLen@ is the
183-- maximum number of pending connections in the backlog. @spec@ is the socket
184-- protocol and options specification and @addr@ is the protocol address where
185-- the server listens for incoming connections.
189accept :: MonadIO m => Unfold m (Int, SockSpec, SockAddr) Socket
190accept = UF.map fst listenTuples
192{-# INLINE connectCommon #-}
193connectCommon :: SockSpec -> Maybe SockAddr -> SockAddr -> IO Socket
194connectCommon SockSpec{..} local remote = withSocketsDo $ do
195 sock <- socket sockFamily sockType sockProto
196 use sock `onException` Net.close sock
202 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
203 forM_ local (bind sock)
204 Net.connect sock remote
206-- | Connect to a remote host using the given socket specification and remote
207-- address. Returns a connected socket or throws an exception.
211{-# INLINE connect #-}
212connect :: SockSpec -> SockAddr -> IO Socket
213connect spec = connectCommon spec Nothing
215-- | Connect to a remote host using the given socket specification, a local
216-- address to bind to and a remote address to connect to. Returns a connected
217-- socket or throws an exception.
221{-# INLINE connectFrom #-}
222connectFrom :: SockSpec -> SockAddr -> SockAddr -> IO Socket
223connectFrom spec local = connectCommon spec (Just local)
225-------------------------------------------------------------------------------
227-------------------------------------------------------------------------------
229{-# INLINE recvConnectionTuplesWith #-}
230recvConnectionTuplesWith :: MonadAsync m
231 => Int -> SockSpec -> SockAddr -> Stream m (Socket, SockAddr)
232recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing
235 listener <- liftIO $ initListener tcpListenQ spec addr
236 r <- liftIO (Net.accept listener `onException` Net.close listener)
237 return $ Just (r, Just listener)
239 step (Just listener) = do
240 r <- liftIO (Net.accept listener `onException` Net.close listener)
241 return $ Just (r, Just listener)
243-- | Start a TCP stream server that listens for connections on the supplied
244-- server address specification (address family, local interface IP address and
245-- port). The server generates a stream of connected sockets. The first
246-- argument is the maximum number of pending connections in the backlog.
249{-# INLINE connections #-}
250connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> Stream m Socket
251connections tcpListenQ spec addr =
252 fst <$> recvConnectionTuplesWith tcpListenQ spec addr
254-------------------------------------------------------------------------------
256-------------------------------------------------------------------------------
258{-# INLINABLE readArrayUptoWith #-}
260 :: (h -> Ptr Word8 -> Int -> IO Int)
264readArrayUptoWith f size h = do
265 arr <- MArray.newPinnedArrayBytes size
266 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
267 MArray.asPtrUnsafe arr $ \p -> do
269 let v = A.unsafeFreeze
270 $ arr { MArray.arrEnd = n, MArray.arrBound = size }
272 -- XXX shrink only if the diff is significant
276-- | Read a byte array from a file handle up to a maximum of the requested
277-- size. If no data is available on the handle it blocks until some data
278-- becomes available. If data is available then it immediately returns that
279-- data without blocking.
282{-# INLINABLE readChunk #-}
283readChunk :: Int -> Socket -> IO (Array Word8)
284readChunk = readArrayUptoWith recvBuf
286-------------------------------------------------------------------------------
288-------------------------------------------------------------------------------
290waitWhen0 :: Int -> Socket -> IO ()
291waitWhen0 0 s = when rtsSupportsBoundThreads $
292#if MIN_VERSION_network(3,1,0)
293 withFdSocket s $ \fd -> threadWaitWrite $ fromIntegral fd
294#elif MIN_VERSION_network(3,0,0)
295 fdSocket s >>= threadWaitWrite . fromIntegral
297 let fd = fdSocket s in threadWaitWrite $ fromIntegral fd
299waitWhen0 _ _ = return ()
301sendAll :: Socket -> Ptr Word8 -> Int -> IO ()
302sendAll _ _ len | len <= 0 = return ()
304 sent <- sendBuf s p len
306 -- assert (sent <= len)
307 when (sent >= 0) $ sendAll s (p `plusPtr` sent) (len - sent)
309{-# INLINABLE writeArrayWith #-}
310writeArrayWith :: Unboxed a
311 => (h -> Ptr Word8 -> Int -> IO ())
315writeArrayWith _ _ arr | A.length arr == 0 = return ()
316writeArrayWith f h arr = A.asPtrUnsafe arr $ \ptr -> f h (castPtr ptr) aLen
320 aLen = A.byteLength arr
322-- | Write an Array to a file handle.
325{-# INLINABLE writeChunk #-}
326writeChunk :: Unboxed a => Socket -> Array a -> IO ()
327writeChunk = writeArrayWith sendAll
329-------------------------------------------------------------------------------
330-- Stream of Arrays IO
331-------------------------------------------------------------------------------
333{-# INLINE _readChunksUptoWith #-}
334_readChunksUptoWith :: (IsStream t, MonadIO m)
335 => (Int -> h -> IO (Array Word8))
336 -> Int -> h -> t m (Array Word8)
337_readChunksUptoWith f size h = go
339 -- XXX use cons/nil instead
340 go = mkStream $ \_ yld _ stp -> do
341 arr <- liftIO $ f size h
346-- | @toChunksWith size h@ reads a stream of arrays from file handle @h@.
347-- The maximum size of a single array is limited to @size@.
348-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
350{-# INLINE_NORMAL toChunksWith #-}
351toChunksWith :: (IsStream t, MonadIO m)
352 => Int -> Socket -> t m (Array Word8)
353-- toChunksWith = _readChunksUptoWith readChunk
354toChunksWith size h = fromStreamD (D.Stream step ())
356 {-# INLINE_LATE step #-}
358 arr <- liftIO $ readChunk size h
364-- XXX read 'Array a' instead of Word8
366-- | @toChunks h@ reads a stream of arrays from socket handle @h@.
367-- The maximum size of a single array is limited to @defaultChunkSize@.
370{-# INLINE toChunks #-}
371toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)
372toChunks = toChunksWith defaultChunkSize
374-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.
375-- Read requests to the socket are performed using a buffer of size @bufsize@.
376-- The size of an array in the resulting stream is always less than or equal to
380{-# INLINE_NORMAL readChunksWith #-}
381readChunksWith :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
382readChunksWith = Unfold step return
384 {-# INLINE_LATE step #-}
386 arr <- liftIO $ readChunk size h
390 _ -> D.Yield arr (size, h)
392-- | Same as 'readChunksWith'
395{-# DEPRECATED readChunksWithBufferOf "Please use 'readChunksWith' instead" #-}
396{-# INLINE_NORMAL readChunksWithBufferOf #-}
397readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
398readChunksWithBufferOf = readChunksWith
400-- | Unfolds a socket into a stream of 'Word8' arrays. Requests to the socket
401-- are performed using a buffer of size
402-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'. The
403-- size of arrays in the resulting stream are therefore less than or equal to
404-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'.
407{-# INLINE readChunks #-}
408readChunks :: MonadIO m => Unfold m Socket (Array Word8)
409readChunks = UF.first defaultChunkSize readChunksWith
411-------------------------------------------------------------------------------
412-- Read File to Stream
413-------------------------------------------------------------------------------
415-- TODO for concurrent streams implement readahead IO. We can send multiple
416-- read requests at the same time. For serial case we can use async IO. We can
417-- also control the read throughput in mbps or IOPS.
420-- | @readWith bufsize handle@ reads a byte stream from a file
421-- handle, reads are performed in chunks of up to @bufsize@. The stream ends
422-- as soon as EOF is encountered.
424{-# INLINE readWith #-}
425readWith :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
426readWith chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h
430-- read :: (IsStream t, MonadIO m, Unboxed a) => Handle -> t m a
432-- > read = 'readByChunks' defaultChunkSize
433-- | Generate a stream of elements of the given type from a socket. The
434-- stream ends when EOF is encountered.
437{-# INLINE toBytes #-}
438toBytes :: (IsStream t, MonadIO m) => Socket -> t m Word8
439toBytes = AS.concat . toChunks
441-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests
442-- to the socket are performed using buffers of @bufsize@.
445{-# INLINE readWith #-}
446readWith :: MonadIO m => Unfold m (Int, Socket) Word8
447readWith = UF.many A.read readChunksWith
449-- | Same as 'readWith'
452{-# DEPRECATED readWithBufferOf "Please use 'readWith' instead" #-}
453{-# INLINE readWithBufferOf #-}
454readWithBufferOf :: MonadIO m => Unfold m (Int, Socket) Word8
455readWithBufferOf = readWith
457-- | Unfolds a 'Socket' into a byte stream. IO requests to the socket are
458-- performed in sizes of
459-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'.
463read :: MonadIO m => Unfold m Socket Word8
464read = UF.first defaultChunkSize readWith
466-------------------------------------------------------------------------------
468-------------------------------------------------------------------------------
470-- | Write a stream of arrays to a handle.
473{-# INLINE putChunks #-}
474putChunks :: (MonadIO m, Unboxed a)
475 => Socket -> Stream m (Array a) -> m ()
476putChunks h = S.mapM_ (liftIO . writeChunk h)
478-- | Write a stream of arrays to a socket. Each array in the stream is written
479-- to the socket as a separate IO request.
482{-# INLINE writeChunks #-}
483writeChunks :: (MonadIO m, Unboxed a) => Socket -> Fold m (Array a) ()
484writeChunks h = FL.drainBy (liftIO . writeChunk h)
486-- | @writeChunksWith bufsize socket@ writes a stream of arrays to
487-- @socket@ after coalescing the adjacent arrays in chunks of @bufsize@.
488-- Multiple arrays are coalesed as long as the total size remains below the
489-- specified size. It never splits an array, if a single array is bigger than
490-- the specified size it emitted as it is.
493{-# INLINE writeChunksWith #-}
494writeChunksWith :: (MonadIO m, Unboxed a)
495 => Int -> Socket -> Fold m (Array a) ()
496writeChunksWith n h = lpackArraysChunksOf n (writeChunks h)
498-- | Same as 'writeChunksWith'
501{-# DEPRECATED writeChunksWithBufferOf "Please use 'writeChunksWith' instead" #-}
502{-# INLINE writeChunksWithBufferOf #-}
503writeChunksWithBufferOf :: (MonadIO m, Unboxed a)
504 => Int -> Socket -> Fold m (Array a) ()
505writeChunksWithBufferOf = writeChunksWith
507-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
510-- Note that if you use a chunk size less than 8K (GHC's default buffer
511-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
512-- do not want buffering to occur at GHC level as well. Same thing applies to
515-- | Like 'write' but provides control over the write buffer. Output will
516-- be written to the IO device as soon as we collect the specified number of
520{-# INLINE putBytesWith #-}
521putBytesWith :: MonadIO m => Int -> Socket -> Stream m Word8 -> m ()
522putBytesWith n h m = putChunks h $ AS.arraysOf n m
524-- | Write a byte stream to a socket. Accumulates the input in chunks of
525-- specified number of bytes before writing.
528{-# INLINE writeWith #-}
529writeWith :: MonadIO m => Int -> Socket -> Fold m Word8 ()
530writeWith n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)
532-- | Same as 'writeWith'
535{-# DEPRECATED writeWithBufferOf "Please use 'writeWith' instead" #-}
536{-# INLINE writeWithBufferOf #-}
537writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
538writeWithBufferOf = writeWith
540-- | Write a stream of 'Maybe' values. Keep buffering the 'Just' values in an
541-- array. Write the array to the 'Handle' as soon as a 'Nothing' is encountered
542-- or the buffer size exceeds the specified limit.
545{-# INLINE writeMaybesWith #-}
546writeMaybesWith :: (MonadIO m )
547 => Int -> Socket -> Fold m (Maybe Word8) ()
549 let writeNJusts = FL.lmap fromJust $ A.writeN n
550 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts
551 in FL.many writeOnNothing (writeChunks h)
553-- > write = 'writeWith' defaultChunkSize
555-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
556-- up to 'defaultChunkSize' before writing. Note that the write behavior
557-- depends on the 'IOMode' and the current seek position of the handle.
560{-# INLINE putBytes #-}
561putBytes :: MonadIO m => Socket -> Stream m Word8 -> m ()
562putBytes = putBytesWith defaultChunkSize
564-- | Write a byte stream to a socket. Accumulates the input in chunks of
565-- up to 'defaultChunkSize' bytes before writing.
568-- write = 'writeWith' 'defaultChunkSize'
573write :: MonadIO m => Socket -> Fold m Word8 ()
574write = writeWith defaultChunkSize
578write :: (MonadIO m, Unboxed a) => Handle -> Stream m a -> m ()
579write = toHandleWith defaultChunkSize
582-------------------------------------------------------------------------------
583-- IO with encoding/decoding Unicode characters
584-------------------------------------------------------------------------------
588-- > readUtf8 = decodeUtf8 . read
590-- Read a UTF8 encoded stream of unicode characters from a file handle.
593{-# INLINE readUtf8 #-}
594readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
595readUtf8 = decodeUtf8 . read
598-- > writeUtf8 h s = write h $ encodeUtf8 s
600-- Encode a stream of unicode characters to UTF8 and write it to the given file
601-- handle. Default block buffering applies to the writes.
604{-# INLINE writeUtf8 #-}
605writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()
606writeUtf8 h s = write h $ encodeUtf8 s
608-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
609-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
610-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
611-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
615{-# INLINE writeUtf8ByLines #-}
616writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
617writeUtf8ByLines = undefined
619-- | Read UTF-8 lines from a file handle and apply the specified fold to each
620-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
623{-# INLINE readLines #-}
624readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
625readLines h f = foldLines (readUtf8 h) f
627-------------------------------------------------------------------------------
628-- Framing on a sequence
629-------------------------------------------------------------------------------
631-- | Read a stream from a file handle and split it into frames delimited by
632-- the specified sequence of elements. The supplied fold is applied on each
636{-# INLINE readFrames #-}
637readFrames :: (IsStream t, MonadIO m, Unboxed a)
638 => Array a -> Handle -> Fold m a b -> t m b
639readFrames = undefined -- foldFrames . read
641-- | Write a stream to the given file handle buffering up to frames separated
642-- by the given sequence or up to a maximum of @defaultChunkSize@.
645{-# INLINE writeByFrames #-}
646writeByFrames :: (IsStream t, MonadIO m, Unboxed a)
647 => Array a -> Handle -> t m a -> m ()
648writeByFrames = 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