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
50 , writeMaybesWithBufferOf
53 , putBytesWithBufferOf
59 , writeChunksWithBufferOf
61 -- reading/writing datagrams
65import Control.Concurrent (threadWaitWrite, rtsSupportsBoundThreads)
66import Control.Exception (onException)
67import Control.Monad.Catch (MonadCatch, finally, MonadMask)
68import Control.Monad.IO.Class (MonadIO(..))
69import Control.Monad (forM_, when)
70import Data.Maybe (isNothing, fromJust)
71import Data.Word (Word8)
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
91 (fromForeignPtrUnsafe, touch)
92import Streamly.Internal.Data.Array.Foreign.Type (Array(..))
93import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)
94import Streamly.Internal.Data.Fold (Fold)
95import Streamly.Internal.Data.Stream.IsStream.Type
96 (IsStream, mkStream, fromStreamD)
97import Streamly.Internal.Data.Stream.Serial (SerialT)
98import Streamly.Internal.Data.Unfold.Type (Unfold(..))
99-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
100import Streamly.Internal.System.IO (defaultChunkSize)
102import qualified Streamly.Internal.Data.Array.Foreign as A
103import qualified Streamly.Internal.Data.Array.Foreign.Type as A
104import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
105import qualified Streamly.Internal.Data.Fold as FL
106import qualified Streamly.Internal.Data.Stream.IsStream as S
107import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
108import qualified Streamly.Internal.Data.Unfold as UF
110-- | @'forSocketM' action socket@ runs the monadic computation @action@ passing
111-- the socket handle to it. The handle will be closed on exit from
112-- 'forSocketM', whether by normal termination or by raising an exception. If
113-- closing the handle raises an exception, then this exception will be raised
114-- by 'forSocketM' rather than any exception raised by 'action'.
117{-# INLINE forSocketM #-}
118forSocketM :: (MonadMask m, MonadIO m) => (Socket -> m ()) -> Socket -> m ()
119forSocketM f sk = finally (f sk) (liftIO (Net.close sk))
121-- | Like 'forSocketM' but runs a streaming computation instead of a monadic
124-- /Inhibits stream fusion/
127{-# INLINE withSocket #-}
128withSocket :: (IsStream t, MonadAsync m, MonadCatch m)
129 => Socket -> (Socket -> t m a) -> t m a
130withSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)
132-------------------------------------------------------------------------------
134-------------------------------------------------------------------------------
136-- XXX Protocol specific socket options should be separated from socket level
139-- | Specify the socket protocol details.
140data SockSpec = SockSpec
142 sockFamily :: !Family
143 , sockType :: !SocketType
144 , sockProto :: !ProtocolNumber
145 , sockOpts :: ![(SocketOption, Int)]
148initListener :: Int -> SockSpec -> SockAddr -> IO Socket
149initListener listenQLen SockSpec{..} addr =
151 sock <- socket sockFamily sockType sockProto
152 use sock `onException` Net.close sock
158 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
160 Net.listen sock listenQLen
162{-# INLINE listenTuples #-}
163listenTuples :: MonadIO m
164 => Unfold m (Int, SockSpec, SockAddr) (Socket, SockAddr)
165listenTuples = Unfold step inject
167 inject (listenQLen, spec, addr) =
168 liftIO $ initListener listenQLen spec addr
171 r <- liftIO (Net.accept listener `onException` Net.close listener)
172 return $ D.Yield r listener
174-- | Unfold a three tuple @(listenQLen, spec, addr)@ into a stream of connected
175-- protocol sockets corresponding to incoming connections. @listenQLen@ is the
176-- maximum number of pending connections in the backlog. @spec@ is the socket
177-- protocol and options specification and @addr@ is the protocol address where
178-- the server listens for incoming connections.
182accept :: MonadIO m => Unfold m (Int, SockSpec, SockAddr) Socket
183accept = UF.map fst listenTuples
185{-# INLINE connectCommon #-}
186connectCommon :: SockSpec -> Maybe SockAddr -> SockAddr -> IO Socket
187connectCommon SockSpec{..} local remote = withSocketsDo $ do
188 sock <- socket sockFamily sockType sockProto
189 use sock `onException` Net.close sock
195 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
196 forM_ local (bind sock)
197 Net.connect sock remote
199-- | Connect to a remote host using the given socket specification and remote
200-- address. Returns a connected socket or throws an exception.
204{-# INLINE connect #-}
205connect :: SockSpec -> SockAddr -> IO Socket
206connect spec = connectCommon spec Nothing
208-- | Connect to a remote host using the given socket specification, a local
209-- address to bind to and a remote address to connect to. Returns a connected
210-- socket or throws an exception.
214{-# INLINE connectFrom #-}
215connectFrom :: SockSpec -> SockAddr -> SockAddr -> IO Socket
216connectFrom spec local = connectCommon spec (Just local)
218-------------------------------------------------------------------------------
220-------------------------------------------------------------------------------
222{-# INLINE recvConnectionTuplesWith #-}
223recvConnectionTuplesWith :: MonadAsync m
224 => Int -> SockSpec -> SockAddr -> SerialT m (Socket, SockAddr)
225recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing
228 listener <- liftIO $ initListener tcpListenQ spec addr
229 r <- liftIO (Net.accept listener `onException` Net.close listener)
230 return $ Just (r, Just listener)
232 step (Just listener) = do
233 r <- liftIO (Net.accept listener `onException` Net.close listener)
234 return $ Just (r, Just listener)
236-- | Start a TCP stream server that listens for connections on the supplied
237-- server address specification (address family, local interface IP address and
238-- port). The server generates a stream of connected sockets. The first
239-- argument is the maximum number of pending connections in the backlog.
242{-# INLINE connections #-}
243connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> SerialT m Socket
244connections tcpListenQ spec addr =
245 fst <$> recvConnectionTuplesWith tcpListenQ spec addr
247-------------------------------------------------------------------------------
249-------------------------------------------------------------------------------
251{-# INLINABLE readArrayUptoWith #-}
253 :: (h -> Ptr Word8 -> Int -> IO Int)
257readArrayUptoWith f size h = do
258 ptr <- mallocPlainForeignPtrBytes size
259 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
260 unsafeWithForeignPtr ptr $ \p -> do
262 let v = A.unsafeFreeze
263 $ fromForeignPtrUnsafe ptr (p `plusPtr` n) (p `plusPtr` size)
265 -- XXX shrink only if the diff is significant
269-- | Read a byte array from a file handle up to a maximum of the requested
270-- size. If no data is available on the handle it blocks until some data
271-- becomes available. If data is available then it immediately returns that
272-- data without blocking.
275{-# INLINABLE readChunk #-}
276readChunk :: Int -> Socket -> IO (Array Word8)
277readChunk = readArrayUptoWith recvBuf
279-------------------------------------------------------------------------------
281-------------------------------------------------------------------------------
283waitWhen0 :: Int -> Socket -> IO ()
284waitWhen0 0 s = when rtsSupportsBoundThreads $
285#if MIN_VERSION_network(3,1,0)
286 withFdSocket s $ \fd -> threadWaitWrite $ fromIntegral fd
287#elif MIN_VERSION_network(3,0,0)
288 fdSocket s >>= threadWaitWrite . fromIntegral
290 let fd = fdSocket s in threadWaitWrite $ fromIntegral fd
292waitWhen0 _ _ = return ()
294sendAll :: Socket -> Ptr Word8 -> Int -> IO ()
295sendAll _ _ len | len <= 0 = return ()
297 sent <- sendBuf s p len
299 -- assert (sent <= len)
300 when (sent >= 0) $ sendAll s (p `plusPtr` sent) (len - sent)
302{-# INLINABLE writeArrayWith #-}
303writeArrayWith :: Storable a
304 => (h -> Ptr Word8 -> Int -> IO ())
308writeArrayWith _ _ arr | A.length arr == 0 = return ()
309writeArrayWith f h Array{..} =
310 f h (castPtr arrStart) aLen >> touch arrContents
314 aLen = aEnd `minusPtr` arrStart
316-- | Write an Array to a file handle.
319{-# INLINABLE writeChunk #-}
320writeChunk :: Storable a => Socket -> Array a -> IO ()
321writeChunk = writeArrayWith sendAll
323-------------------------------------------------------------------------------
324-- Stream of Arrays IO
325-------------------------------------------------------------------------------
327{-# INLINABLE _readChunksUptoWith #-}
328_readChunksUptoWith :: (IsStream t, MonadIO m)
329 => (Int -> h -> IO (Array Word8))
330 -> Int -> h -> t m (Array Word8)
331_readChunksUptoWith f size h = go
333 -- XXX use cons/nil instead
334 go = mkStream $ \_ yld _ stp -> do
335 arr <- liftIO $ f size h
340-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
341-- The maximum size of a single array is limited to @size@.
342-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
344{-# INLINE_NORMAL toChunksWithBufferOf #-}
345toChunksWithBufferOf :: (IsStream t, MonadIO m)
346 => Int -> Socket -> t m (Array Word8)
347-- toChunksWithBufferOf = _readChunksUptoWith readChunk
348toChunksWithBufferOf size h = fromStreamD (D.Stream step ())
350 {-# INLINE_LATE step #-}
352 arr <- liftIO $ readChunk size h
358-- XXX read 'Array a' instead of Word8
360-- | @toChunks h@ reads a stream of arrays from socket handle @h@.
361-- The maximum size of a single array is limited to @defaultChunkSize@.
364{-# INLINE toChunks #-}
365toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)
366toChunks = toChunksWithBufferOf defaultChunkSize
368-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.
369-- Read requests to the socket are performed using a buffer of size @bufsize@.
370-- The size of an array in the resulting stream is always less than or equal to
374{-# INLINE_NORMAL readChunksWithBufferOf #-}
375readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
376readChunksWithBufferOf = Unfold step return
378 {-# INLINE_LATE step #-}
380 arr <- liftIO $ readChunk size h
384 _ -> D.Yield arr (size, h)
386-- | Unfolds a socket into a stream of 'Word8' arrays. Requests to the socket
387-- are performed using a buffer of size
388-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'. The
389-- size of arrays in the resulting stream are therefore less than or equal to
390-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'.
393{-# INLINE readChunks #-}
394readChunks :: MonadIO m => Unfold m Socket (Array Word8)
395readChunks = UF.supplyFirst defaultChunkSize readChunksWithBufferOf
397-------------------------------------------------------------------------------
398-- Read File to Stream
399-------------------------------------------------------------------------------
401-- TODO for concurrent streams implement readahead IO. We can send multiple
402-- read requests at the same time. For serial case we can use async IO. We can
403-- also control the read throughput in mbps or IOPS.
406-- | @readWithBufferOf bufsize handle@ reads a byte stream from a file
407-- handle, reads are performed in chunks of up to @bufsize@. The stream ends
408-- as soon as EOF is encountered.
410{-# INLINE readWithBufferOf #-}
411readWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
412readWithBufferOf chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h
416-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
418-- > read = 'readByChunks' defaultChunkSize
419-- | Generate a stream of elements of the given type from a socket. The
420-- stream ends when EOF is encountered.
423{-# INLINE toBytes #-}
424toBytes :: (IsStream t, MonadIO m) => Socket -> t m Word8
425toBytes = AS.concat . toChunks
427-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests
428-- to the socket are performed using buffers of @bufsize@.
431{-# INLINE readWithBufferOf #-}
432readWithBufferOf :: MonadIO m => Unfold m (Int, Socket) Word8
433readWithBufferOf = UF.many readChunksWithBufferOf A.read
435-- | Unfolds a 'Socket' into a byte stream. IO requests to the socket are
436-- performed in sizes of
437-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'.
441read :: MonadIO m => Unfold m Socket Word8
442read = UF.supplyFirst defaultChunkSize readWithBufferOf
444-------------------------------------------------------------------------------
446-------------------------------------------------------------------------------
448-- | Write a stream of arrays to a handle.
451{-# INLINE putChunks #-}
452putChunks :: (MonadIO m, Storable a)
453 => Socket -> SerialT m (Array a) -> m ()
454putChunks h = S.mapM_ (liftIO . writeChunk h)
456-- | Write a stream of arrays to a socket. Each array in the stream is written
457-- to the socket as a separate IO request.
460{-# INLINE writeChunks #-}
461writeChunks :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()
462writeChunks h = FL.drainBy (liftIO . writeChunk h)
464-- | @writeChunksWithBufferOf bufsize socket@ writes a stream of arrays to
465-- @socket@ after coalescing the adjacent arrays in chunks of @bufsize@.
466-- Multiple arrays are coalesed as long as the total size remains below the
467-- specified size. It never splits an array, if a single array is bigger than
468-- the specified size it emitted as it is.
471{-# INLINE writeChunksWithBufferOf #-}
472writeChunksWithBufferOf :: (MonadIO m, Storable a)
473 => Int -> Socket -> Fold m (Array a) ()
474writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
476-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
479-- Note that if you use a chunk size less than 8K (GHC's default buffer
480-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
481-- do not want buffering to occur at GHC level as well. Same thing applies to
484-- | Like 'write' but provides control over the write buffer. Output will
485-- be written to the IO device as soon as we collect the specified number of
489{-# INLINE putBytesWithBufferOf #-}
490putBytesWithBufferOf :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()
491putBytesWithBufferOf n h m = putChunks h $ AS.arraysOf n m
493-- | Write a byte stream to a socket. Accumulates the input in chunks of
494-- specified number of bytes before writing.
497{-# INLINE writeWithBufferOf #-}
498writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
499writeWithBufferOf n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)
501-- | Write a stream of 'Maybe' values. Keep buffering the 'Just' values in an
502-- array. Write the array to the 'Handle' as soon as a 'Nothing' is encountered
503-- or the buffer size exceeds the specified limit.
506{-# INLINE writeMaybesWithBufferOf #-}
507writeMaybesWithBufferOf :: (MonadIO m )
508 => Int -> Socket -> Fold m (Maybe Word8) ()
509writeMaybesWithBufferOf n h =
510 let writeNJusts = FL.lmap fromJust $ A.writeN n
511 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts
512 in FL.many writeOnNothing (writeChunks h)
514-- > write = 'writeWithBufferOf' defaultChunkSize
516-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
517-- up to 'defaultChunkSize' before writing. Note that the write behavior
518-- depends on the 'IOMode' and the current seek position of the handle.
521{-# INLINE putBytes #-}
522putBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()
523putBytes = putBytesWithBufferOf defaultChunkSize
525-- | Write a byte stream to a socket. Accumulates the input in chunks of
526-- up to 'defaultChunkSize' bytes before writing.
529-- write = 'writeWithBufferOf' 'defaultChunkSize'
534write :: MonadIO m => Socket -> Fold m Word8 ()
535write = writeWithBufferOf defaultChunkSize
539write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
540write = toHandleWith defaultChunkSize
543-------------------------------------------------------------------------------
544-- IO with encoding/decoding Unicode characters
545-------------------------------------------------------------------------------
549-- > readUtf8 = decodeUtf8 . read
551-- Read a UTF8 encoded stream of unicode characters from a file handle.
554{-# INLINE readUtf8 #-}
555readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
556readUtf8 = decodeUtf8 . read
559-- > writeUtf8 h s = write h $ encodeUtf8 s
561-- Encode a stream of unicode characters to UTF8 and write it to the given file
562-- handle. Default block buffering applies to the writes.
565{-# INLINE writeUtf8 #-}
566writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
567writeUtf8 h s = write h $ encodeUtf8 s
569-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
570-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
571-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
572-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
576{-# INLINE writeUtf8ByLines #-}
577writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
578writeUtf8ByLines = undefined
580-- | Read UTF-8 lines from a file handle and apply the specified fold to each
581-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
584{-# INLINE readLines #-}
585readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
586readLines h f = foldLines (readUtf8 h) f
588-------------------------------------------------------------------------------
589-- Framing on a sequence
590-------------------------------------------------------------------------------
592-- | Read a stream from a file handle and split it into frames delimited by
593-- the specified sequence of elements. The supplied fold is applied on each
597{-# INLINE readFrames #-}
598readFrames :: (IsStream t, MonadIO m, Storable a)
599 => Array a -> Handle -> Fold m a b -> t m b
600readFrames = undefined -- foldFrames . read
602-- | Write a stream to the given file handle buffering up to frames separated
603-- by the given sequence or up to a maximum of @defaultChunkSize@.
606{-# INLINE writeByFrames #-}
607writeByFrames :: (IsStream t, MonadIO m, Storable a)
608 => Array a -> Handle -> t m a -> m ()
609writeByFrames = 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