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 (minusPtr, plusPtr, Ptr, castPtr)
79import Streamly.Internal.Data.Unboxed (Storable)
80import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
82 (Socket, SocketOption(..), Family(..), SockAddr(..),
83 ProtocolNumber, withSocketsDo, SocketType(..), socket, bind,
84 setSocketOption, sendBuf, recvBuf)
85#if MIN_VERSION_network(3,1,0)
86import Network.Socket (withFdSocket)
88import Network.Socket (fdSocket)
90import Prelude hiding (read)
92import qualified Network.Socket as Net
94import Streamly.Internal.BaseCompat
95import Streamly.Internal.Control.Concurrent (MonadAsync)
96import Streamly.Internal.Data.Array.Foreign.Mut.Type
97 (fromForeignPtrUnsafe, touch)
98import Streamly.Internal.Data.Array.Foreign.Type (Array(..))
99import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)
100import Streamly.Internal.Data.Fold (Fold)
101import Streamly.Internal.Data.Stream.IsStream.Type
102 (IsStream, mkStream, fromStreamD)
103import Streamly.Internal.Data.Stream.Serial (SerialT)
104import Streamly.Internal.Data.Unfold.Type (Unfold(..))
105-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
106import Streamly.Internal.System.IO (defaultChunkSize)
108import qualified Streamly.Internal.Data.Array.Foreign as A
109import qualified Streamly.Internal.Data.Array.Foreign.Type as A
110import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
111import qualified Streamly.Internal.Data.Fold as FL
112import qualified Streamly.Internal.Data.Stream.IsStream as S
113import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
114import qualified Streamly.Internal.Data.Unfold as UF
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 :: (IsStream t, MonadAsync m, MonadCatch m)
135 => Socket -> (Socket -> t m a) -> t 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 -> SerialT 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 -> SerialT 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 ptr <- mallocPlainForeignPtrBytes size
265 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
266 unsafeWithForeignPtr ptr $ \p -> do
268 let v = A.unsafeFreeze
269 $ fromForeignPtrUnsafe ptr (p `plusPtr` n) (p `plusPtr` 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 :: Storable a
310 => (h -> Ptr Word8 -> Int -> IO ())
314writeArrayWith _ _ arr | A.length arr == 0 = return ()
315writeArrayWith f h Array{..} =
316 f h (castPtr arrStart) aLen >> touch arrContents
320 aLen = aEnd `minusPtr` arrStart
322-- | Write an Array to a file handle.
325{-# INLINABLE writeChunk #-}
326writeChunk :: Storable 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.Foreign.Type.defaultChunkSize'. The
403-- size of arrays in the resulting stream are therefore less than or equal to
404-- 'Streamly.Internal.Data.Array.Foreign.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, Storable 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.Foreign.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, Storable a)
475 => Socket -> SerialT 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, Storable 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, Storable 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, Storable 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 -> SerialT 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 -> SerialT 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, Storable a) => Handle -> SerialT 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 -> SerialT 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, Storable 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, Storable 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