1{-# OPTIONS_HADDOCK hide #-}
3{-# LANGUAGE BangPatterns #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE MagicHash #-}
6{-# LANGUAGE RecordWildCards #-}
7{-# LANGUAGE UnboxedTuples #-}
12-- Module : Streamly.Internal.Network.Socket
13-- Copyright : (c) 2018 Composewell Technologies
16-- Maintainer : streamly@composewell.com
17-- Stability : experimental
20module Streamly.Internal.Network.Socket
27 -- * Accept connections
31 -- * Read from connection
44 , readChunksWithBufferOf
47 , toChunksWithBufferOf
51 -- * Write to connection
59 , fromBytesWithBufferOf
66 -- reading/writing datagrams
70import Control.Concurrent (threadWaitWrite, rtsSupportsBoundThreads)
71import Control.Monad.Catch (MonadCatch, finally, MonadMask)
72import Control.Monad.IO.Class (MonadIO(..))
73import Control.Monad (when)
74import Data.Word (Word8)
75import Foreign.ForeignPtr (withForeignPtr)
76import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
77import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)
78import Foreign.Storable (Storable(..))
79import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
80import Network.Socket (sendBuf, recvBuf)
82 (Socket, SocketOption(..), Family(..), SockAddr(..),
83 ProtocolNumber, withSocketsDo, SocketType(..), socket, bind,
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 (MonadAsync)
95import Streamly.Internal.Data.Unfold.Types (Unfold(..))
96import Streamly.Internal.Memory.Array.Types (Array(..))
97import Streamly.Streams.Serial (SerialT)
98import Streamly.Streams.StreamK.Type (IsStream, mkStream)
99import Streamly.Data.Fold (Fold)
100-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
102import qualified Streamly.Data.Fold as FL
103import qualified Streamly.Internal.Data.Fold.Types as FL
104import qualified Streamly.Internal.Data.Unfold as UF
105import qualified Streamly.Memory.Array as A
106import qualified Streamly.Internal.Memory.ArrayStream as AS
107import qualified Streamly.Internal.Memory.Array.Types as A
108import qualified Streamly.Prelude as S
109import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
111-- | @'useSocketM' socket act@ runs the monadic computation @act@ passing the
112-- socket handle to it. The handle will be closed on exit from 'useSocketM',
113-- whether by normal termination or by raising an exception. If closing the
114-- handle raises an exception, then this exception will be raised by
115-- 'useSocketM' rather than any exception raised by 'act'.
118{-# INLINE useSocketM #-}
119useSocketM :: (MonadMask m, MonadIO m) => Socket -> (Socket -> m ()) -> m ()
120useSocketM sk f = finally (liftIO (Net.close sk)) (f sk)
122-- | Like 'useSocketM' but runs a streaming computation instead of a monadic
126{-# INLINE useSocket #-}
127useSocket :: (IsStream t, MonadCatch m, MonadIO m)
128 => Socket -> (Socket -> t m a) -> t m a
129useSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)
131-------------------------------------------------------------------------------
133-------------------------------------------------------------------------------
135-- XXX Protocol specific socket options should be separated from socket level
138-- | Specify the socket protocol details.
139data SockSpec = SockSpec
141 sockFamily :: !Family
142 , sockType :: !SocketType
143 , sockProto :: !ProtocolNumber
144 , sockOpts :: ![(SocketOption, Int)]
147initListener :: Int -> SockSpec -> SockAddr -> IO Socket
148initListener listenQLen SockSpec{..} addr =
150 sock <- socket sockFamily sockType sockProto
151 mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts
153 Net.listen sock listenQLen
156{-# INLINE listenTuples #-}
157listenTuples :: MonadIO m
158 => Unfold m (Int, SockSpec, SockAddr) (Socket, SockAddr)
159listenTuples = Unfold step inject
161 inject (listenQLen, spec, addr) = do
162 listener <- liftIO $ initListener listenQLen spec addr
166 r <- liftIO $ Net.accept listener
167 -- XXX error handling
168 return $ D.Yield r listener
170-- | Unfold a three tuple @(listenQLen, spec, addr)@ into a stream of connected
171-- protocol sockets corresponding to incoming connections. @listenQLen@ is the
172-- maximum number of pending connections in the backlog. @spec@ is the socket
173-- protocol and options specification and @addr@ is the protocol address where
174-- the server listens for incoming connections.
178accept :: MonadIO m => Unfold m (Int, SockSpec, SockAddr) Socket
179accept = UF.map fst listenTuples
181-------------------------------------------------------------------------------
183-------------------------------------------------------------------------------
185{-# INLINE recvConnectionTuplesWith #-}
186recvConnectionTuplesWith :: MonadAsync m
187 => Int -> SockSpec -> SockAddr -> SerialT m (Socket, SockAddr)
188recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing
191 listener <- liftIO $ initListener tcpListenQ spec addr
192 r <- liftIO $ Net.accept listener
193 -- XXX error handling
194 return $ Just (r, Just listener)
196 step (Just listener) = do
197 r <- liftIO $ Net.accept listener
198 -- XXX error handling
199 return $ Just (r, Just listener)
201-- | Start a TCP stream server that listens for connections on the supplied
202-- server address specification (address family, local interface IP address and
203-- port). The server generates a stream of connected sockets. The first
204-- argument is the maximum number of pending connections in the backlog.
207{-# INLINE connections #-}
208connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> SerialT m Socket
209connections tcpListenQ spec addr = fmap fst $
210 recvConnectionTuplesWith tcpListenQ spec addr
212-------------------------------------------------------------------------------
214-------------------------------------------------------------------------------
216{-# INLINABLE readArrayUptoWith #-}
218 :: (h -> Ptr Word8 -> Int -> IO Int)
222readArrayUptoWith f size h = do
223 ptr <- mallocPlainForeignPtrBytes size
224 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
225 withForeignPtr ptr $ \p -> do
229 , aEnd = p `plusPtr` n
230 , aBound = p `plusPtr` size
232 -- XXX shrink only if the diff is significant
236-- | Read a 'ByteArray' from a file handle. If no data is available on the
237-- handle it blocks until some data becomes available. If data is available
238-- then it immediately returns that data without blocking. It reads a maximum
239-- of up to the size requested.
240{-# INLINABLE readArrayOf #-}
241readArrayOf :: Int -> Socket -> IO (Array Word8)
242readArrayOf = readArrayUptoWith recvBuf
244-------------------------------------------------------------------------------
246-------------------------------------------------------------------------------
248waitWhen0 :: Int -> Socket -> IO ()
249waitWhen0 0 s = when rtsSupportsBoundThreads $
250#if MIN_VERSION_network(3,1,0)
251 withFdSocket s $ \fd -> threadWaitWrite $ fromIntegral fd
253 let fd = fdSocket s in threadWaitWrite $ fromIntegral fd
255waitWhen0 _ _ = return ()
257sendAll :: Socket -> Ptr Word8 -> Int -> IO ()
258sendAll _ _ len | len <= 0 = return ()
260 sent <- sendBuf s p len
262 -- assert (sent <= len)
263 when (sent >= 0) $ sendAll s (p `plusPtr` sent) (len - sent)
265{-# INLINABLE writeArrayWith #-}
266writeArrayWith :: Storable a
267 => (h -> Ptr Word8 -> Int -> IO ())
271writeArrayWith _ _ arr | A.length arr == 0 = return ()
272writeArrayWith f h Array{..} = withForeignPtr aStart $ \p ->
276 let p = unsafeForeignPtrToPtr aStart
279-- | Write an Array to a file handle.
282{-# INLINABLE writeArray #-}
283writeArray :: Storable a => Socket -> Array a -> IO ()
284writeArray = writeArrayWith sendAll
286-------------------------------------------------------------------------------
287-- Stream of Arrays IO
288-------------------------------------------------------------------------------
290{-# INLINABLE readChunksUptoWith #-}
291readChunksUptoWith :: (IsStream t, MonadIO m)
292 => (Int -> h -> IO (Array Word8))
293 -> Int -> h -> t m (Array Word8)
294readChunksUptoWith f size h = go
296 -- XXX use cons/nil instead
297 go = mkStream $ \_ yld _ stp -> do
298 arr <- liftIO $ f size h
303-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
304-- The maximum size of a single array is limited to @size@.
305-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
307{-# INLINABLE toChunksWithBufferOf #-}
308toChunksWithBufferOf :: (IsStream t, MonadIO m)
309 => Int -> Socket -> t m (Array Word8)
310toChunksWithBufferOf = readChunksUptoWith readArrayOf
312-- XXX read 'Array a' instead of Word8
314-- | @toChunks h@ reads a stream of arrays from socket handle @h@.
315-- The maximum size of a single array is limited to @defaultChunkSize@.
318{-# INLINE toChunks #-}
319toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)
320toChunks = toChunksWithBufferOf A.defaultChunkSize
322-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.
323-- Read requests to the socket are performed using a buffer of size @bufsize@.
324-- The size of an array in the resulting stream is always less than or equal to
328{-# INLINE_NORMAL readChunksWithBufferOf #-}
329readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
330readChunksWithBufferOf = Unfold step return
332 {-# INLINE_LATE step #-}
334 arr <- liftIO $ readArrayOf size h
338 _ -> D.Yield arr (size, h)
340-- | Unfolds a socket into a stream of 'Word8' arrays. Requests to the socket
341-- are performed using a buffer of size
342-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'. The
343-- size of arrays in the resulting stream are therefore less than or equal to
344-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
347{-# INLINE readChunks #-}
348readChunks :: MonadIO m => Unfold m Socket (Array Word8)
349readChunks = UF.supplyFirst readChunksWithBufferOf A.defaultChunkSize
351-------------------------------------------------------------------------------
352-- Read File to Stream
353-------------------------------------------------------------------------------
355-- TODO for concurrent streams implement readahead IO. We can send multiple
356-- read requests at the same time. For serial case we can use async IO. We can
357-- also control the read throughput in mbps or IOPS.
360-- | @readWithBufferOf bufsize handle@ reads a byte stream from a file
361-- handle, reads are performed in chunks of up to @bufsize@. The stream ends
362-- as soon as EOF is encountered.
364{-# INLINE readWithBufferOf #-}
365readWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
366readWithBufferOf chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h
370-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
372-- > read = 'readByChunks' A.defaultChunkSize
373-- | Generate a stream of elements of the given type from a socket. The
374-- stream ends when EOF is encountered.
377{-# INLINE toBytes #-}
378toBytes :: (IsStream t, MonadIO m) => Socket -> t m Word8
379toBytes = AS.concat . toChunks
381-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests
382-- to the socket are performed using buffers of @bufsize@.
385{-# INLINE readWithBufferOf #-}
386readWithBufferOf :: MonadIO m => Unfold m (Int, Socket) Word8
387readWithBufferOf = UF.concat readChunksWithBufferOf A.read
389-- | Unfolds a 'Socket' into a byte stream. IO requests to the socket are
390-- performed in sizes of
391-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
395read :: MonadIO m => Unfold m Socket Word8
396read = UF.supplyFirst readWithBufferOf A.defaultChunkSize
398-------------------------------------------------------------------------------
400-------------------------------------------------------------------------------
402-- | Write a stream of arrays to a handle.
405{-# INLINE fromChunks #-}
406fromChunks :: (MonadIO m, Storable a)
407 => Socket -> SerialT m (Array a) -> m ()
408fromChunks h m = S.mapM_ (liftIO . writeArray h) m
410-- | Write a stream of arrays to a socket. Each array in the stream is written
411-- to the socket as a separate IO request.
414{-# INLINE writeChunks #-}
415writeChunks :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()
416writeChunks h = FL.drainBy (liftIO . writeArray h)
418-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
421-- Note that if you use a chunk size less than 8K (GHC's default buffer
422-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
423-- do not want buffering to occur at GHC level as well. Same thing applies to
426-- | Like 'write' but provides control over the write buffer. Output will
427-- be written to the IO device as soon as we collect the specified number of
431{-# INLINE fromBytesWithBufferOf #-}
432fromBytesWithBufferOf :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()
433fromBytesWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
435-- | Write a byte stream to a socket. Accumulates the input in chunks of
436-- specified number of bytes before writing.
439{-# INLINE writeWithBufferOf #-}
440writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
441writeWithBufferOf n h = FL.lchunksOf n (A.writeNUnsafe n) (writeChunks h)
443-- > write = 'writeWithBufferOf' A.defaultChunkSize
445-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
446-- up to 'A.defaultChunkSize' before writing. Note that the write behavior
447-- depends on the 'IOMode' and the current seek position of the handle.
450{-# INLINE fromBytes #-}
451fromBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()
452fromBytes = fromBytesWithBufferOf A.defaultChunkSize
454-- | Write a byte stream to a socket. Accumulates the input in chunks of
455-- up to 'A.defaultChunkSize' bytes before writing.
458-- write = 'writeWithBufferOf' 'A.defaultChunkSize'
463write :: MonadIO m => Socket -> Fold m Word8 ()
464write = writeWithBufferOf A.defaultChunkSize
468write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
469write = toHandleWith A.defaultChunkSize
472-------------------------------------------------------------------------------
473-- IO with encoding/decoding Unicode characters
474-------------------------------------------------------------------------------
478-- > readUtf8 = decodeUtf8 . read
480-- Read a UTF8 encoded stream of unicode characters from a file handle.
483{-# INLINE readUtf8 #-}
484readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
485readUtf8 = decodeUtf8 . read
488-- > writeUtf8 h s = write h $ encodeUtf8 s
490-- Encode a stream of unicode characters to UTF8 and write it to the given file
491-- handle. Default block buffering applies to the writes.
494{-# INLINE writeUtf8 #-}
495writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
496writeUtf8 h s = write h $ encodeUtf8 s
498-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
499-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
500-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
501-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
505{-# INLINE writeUtf8ByLines #-}
506writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
507writeUtf8ByLines = undefined
509-- | Read UTF-8 lines from a file handle and apply the specified fold to each
510-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
513{-# INLINE readLines #-}
514readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
515readLines h f = foldLines (readUtf8 h) f
517-------------------------------------------------------------------------------
518-- Framing on a sequence
519-------------------------------------------------------------------------------
521-- | Read a stream from a file handle and split it into frames delimited by
522-- the specified sequence of elements. The supplied fold is applied on each
526{-# INLINE readFrames #-}
527readFrames :: (IsStream t, MonadIO m, Storable a)
528 => Array a -> Handle -> Fold m a b -> t m b
529readFrames = undefined -- foldFrames . read
531-- | Write a stream to the given file handle buffering up to frames separated
532-- by the given sequence or up to a maximum of @defaultChunkSize@.
535{-# INLINE writeByFrames #-}
536writeByFrames :: (IsStream t, MonadIO m, Storable a)
537 => Array a -> Handle -> t m a -> m ()
538writeByFrames = 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