git.iko.soy/streamly
on 43dda21 help sign in
streamly/src/Streamly/Internal/System/IOVec.hs
126 lines 4.7 kB history raw
1-- |
2-- Module : Streamly.Internal.System.IOVec
3-- Copyright : (c) 2019 Composewell Technologies
4--
5-- License : BSD3
6-- Maintainer : streamly@composewell.com
7-- Stability : experimental
8-- Portability : GHC
9--
10-- Low level IO routines interfacing the operating system.
11--
13module Streamly.Internal.System.IOVec
14 ( IOVec(..)
15 , c_writev
16 , c_safe_writev
17#if !defined(mingw32_HOST_OS)
18 , groupIOVecsOf
19 , groupIOVecsOfMut
20#endif
21 )
22where
24#include "inline.hs"
26#if !defined(mingw32_HOST_OS)
27import Control.Monad (when)
28import Control.Monad.IO.Class (MonadIO(..))
29import Foreign.Ptr (castPtr)
30import Streamly.Internal.Data.Array.Foreign.Mut.Type (length)
31import Streamly.Internal.Data.SVar.Type (adaptState)
32import Streamly.Internal.Data.Array.Foreign.Mut.Type (Array(..))
34import qualified Streamly.Internal.Data.Array.Foreign.Type as Array
35import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MArray
36import qualified Streamly.Internal.Data.Stream.StreamD as D
37#endif
39import Streamly.Internal.System.IOVec.Type
41import Prelude hiding (length)
43#if !defined(mingw32_HOST_OS)
44data GatherState s arr
45 = GatherInitial s
46 | GatherBuffering s arr Int
47 | GatherYielding arr (GatherState s arr)
48 | GatherFinish
50-- | @groupIOVecsOf maxBytes maxEntries@ groups arrays in the incoming stream
51-- to create a stream of 'IOVec' arrays with a maximum of @maxBytes@ bytes in
52-- each array and a maximum of @maxEntries@ entries in each array.
53--
54-- @since 0.7.0
55{-# INLINE_NORMAL groupIOVecsOfMut #-}
56groupIOVecsOfMut :: MonadIO m
57 => Int -> Int -> D.Stream m (Array a) -> D.Stream m (Array IOVec)
58groupIOVecsOfMut n maxIOVLen (D.Stream step state) =
59 D.Stream step' (GatherInitial state)
61 where
63 {-# INLINE_LATE step' #-}
64 step' gst (GatherInitial st) = do
65 when (n <= 0) $
66 -- XXX we can pass the module string from the higher level API
67 error $ "Streamly.Internal.Data.Array.Foreign.Mut.Type.groupIOVecsOf: the size of "
68 ++ "groups [" ++ show n ++ "] must be a natural number"
69 when (maxIOVLen <= 0) $
70 -- XXX we can pass the module string from the higher level API
71 error $ "Streamly.Internal.Data.Array.Foreign.Mut.Type.groupIOVecsOf: the number of "
72 ++ "IOVec entries [" ++ show n ++ "] must be a natural number"
73 r <- step (adaptState gst) st
74 case r of
75 D.Yield arr s -> do
76 let p = arrStart arr
77 len = MArray.byteLength arr
78 iov <- liftIO $ MArray.newArray maxIOVLen
79 iov' <- liftIO $ MArray.snocUnsafe iov (IOVec (castPtr p)
80 (fromIntegral len))
81 if len >= n
82 then return $ D.Skip (GatherYielding iov' (GatherInitial s))
83 else return $ D.Skip (GatherBuffering s iov' len)
84 D.Skip s -> return $ D.Skip (GatherInitial s)
85 D.Stop -> return D.Stop
87 step' gst (GatherBuffering st iov len) = do
88 r <- step (adaptState gst) st
89 case r of
90 D.Yield arr s -> do
91 let p = arrStart arr
92 alen = MArray.byteLength arr
93 len' = len + alen
94 if len' > n || length iov >= maxIOVLen
95 then do
96 iov' <- liftIO $ MArray.newArray maxIOVLen
97 iov'' <- liftIO $ MArray.snocUnsafe iov' (IOVec (castPtr p)
98 (fromIntegral alen))
99 return $ D.Skip (GatherYielding iov
100 (GatherBuffering s iov'' alen))
101 else do
102 iov' <- liftIO $ MArray.snocUnsafe iov (IOVec (castPtr p)
103 (fromIntegral alen))
104 return $ D.Skip (GatherBuffering s iov' len')
105 D.Skip s -> return $ D.Skip (GatherBuffering s iov len)
106 D.Stop -> return $ D.Skip (GatherYielding iov GatherFinish)
108 step' _ GatherFinish = return D.Stop
110 step' _ (GatherYielding iov next) = return $ D.Yield iov next
112-- | @groupIOVecsOf maxBytes maxEntries@ groups arrays in the incoming stream
113-- to create a stream of 'IOVec' arrays with a maximum of @maxBytes@ bytes in
114-- each array and a maximum of @maxEntries@ entries in each array.
115--
116-- @since 0.7.0
117{-# INLINE_NORMAL groupIOVecsOf #-}
118groupIOVecsOf :: MonadIO m
119 => Int
120 -> Int
121 -> D.Stream m (Array.Array a) -> D.Stream m (Array.Array IOVec)
122groupIOVecsOf n maxIOVLen str =
123 D.map Array.unsafeFreeze
124 $ groupIOVecsOfMut n maxIOVLen
125 $ D.map Array.unsafeThaw str
126#endif