diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,16 @@ # Revision history for Z-IO +## 0.8.0.0 -- 2020-04-25 + +This is an experimental version to test new 'BIO' module. + +* Rewrite `Z.IO.BIO` module, now `BIO` is push based. +* Remove `>|>`, `>~>`, `>!>`, now `BIO` nodes can be composed via funtion composition `(.)`! +* Remove `zipSource/zipBIO`, add `stepBIO/stepBIO_/runBIO_`. +* Add `zipBIO` to `Z.IO.BIO.Concurrent`, which run two BIO nodes concurrently. +* Add `ungroupingNode`, change `newGroupingNode` to use `Vector`. +* Rename `EOF` exception to `UnexpectedEOF` to avoid the clash with `EOF` pattern. + ## 0.7.1.0 -- 2020-03-16 * Use `CPtr` from Z-Data instead of `ForeignPtr`. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -5,6 +5,7 @@ [![MacOS Build Status](https://github.com/ZHaskell/z-io/workflows/osx-ci/badge.svg)](https://github.com/ZHaskell/z-io/actions) [![Windows Build Status](https://github.com/ZHaskell/z-io/workflows/win-ci/badge.svg)](https://github.com/ZHaskell/z-io/actions) [![Docker Build Status](https://github.com/ZHaskell/z-io/workflows/docker-ci/badge.svg)](https://github.com/ZHaskell/z-io/actions) +[![Gitter chat](https://badges.gitter.im/gitterHQ/gitter.svg)](https://gitter.im/Z-Haskell/community) This package is part of [Z.Haskell](https://z.haskell.world) project, provides basic IO operations: @@ -39,10 +40,8 @@ > :{ let addr = ipv4 "13.107.21.200" 80 in withResource (initTCPClient defaultTCPClientConfig{ tcpRemoteAddr = addr}) $ \ tcp -> do - i <- newBufferedInput tcp - o <- newBufferedOutput tcp - writeBuffer o "GET http://www.bing.com HTTP/1.1\r\nHost: www.bing.com\r\n\r\n" - flushBuffer o + (i, o) <- newBufferedIO tcp + writeBuffer' o "GET http://www.bing.com HTTP/1.1\r\nHost: www.bing.com\r\n\r\n" readBuffer i >>= pure . T.validate :} "HTTP/1.1 200 OK\r\nDate: Sat, 19 Sep 2020 06:11:08 GMT\r\nContent-Length: 0\r\n\r\n" diff --git a/Z-IO.cabal b/Z-IO.cabal --- a/Z-IO.cabal +++ b/Z-IO.cabal @@ -1,6 +1,6 @@ cabal-version: 2.4 name: Z-IO -version: 0.7.1.0 +version: 0.8.0.0 synopsis: Simple and high performance IO toolkit for Haskell description: Simple and high performance IO toolkit for Haskell, including @@ -17,6 +17,7 @@ bug-reports: https://github.com/ZHaskell/Z-IO/issues extra-source-files: ChangeLog.md + include/fs_shared.hs LICENSE README.md third_party/cwalk/include/cwalk.h @@ -64,7 +65,6 @@ third_party/zlib/inftrees.h third_party/zlib/trees.h third_party/zlib/zutil.h - Z/IO/FileSystem/_Shared.hs source-repository head type: git @@ -113,19 +113,18 @@ Z.IO.UV.FFI_Env Z.IO.UV.Manager Z.IO.UV.UVStream - other-modules: - Z.IO.UV.Win + other-modules: Z.IO.UV.Win build-depends: , base >=4.12 && <5.0 , containers ^>=0.6 , exceptions ^>=0.10 , primitive >=0.7.1 && <0.7.2 , stm ^>=2.5 - , time >=1.9 && <=2.0 - , unix-time >=0.4.7 && <=0.5 + , time >=1.9 && <2.0 + , unix-time >=0.4.7 && <0.5 , unordered-containers ^>=0.2 - , Z-Data >=0.7.1 && <0.8 + , Z-Data >=0.8.1 && <0.9 default-language: Haskell2010 default-extensions: diff --git a/Z/IO/BIO.hs b/Z/IO/BIO.hs --- a/Z/IO/BIO.hs +++ b/Z/IO/BIO.hs @@ -1,7 +1,6 @@ -{-# OPTIONS_GHC -Wno-missing-fields #-} {-| Module : Z.IO.BIO -Description : Buffered IO interface +Description : Composable IO Loops Copyright : (c) Dong Han, 2017-2020 License : BSD Maintainer : winterland1989@gmail.com @@ -37,7 +36,7 @@ withResource (initSourceFromFile origin) $ \ src -> withResource (initSinkToFile target) $ \ sink -> - runBIO $ src >|> base64Enc >|> zlibCompressor >|> sink + runBIO_ $ src . base64Enc . zlibCompressor . sink > base64AndCompressFile "test" "test.gz" -- run 'zcat "test.gz" | base64 -d' will give you original file @@ -46,27 +45,28 @@ -} module Z.IO.BIO ( -- * The BIO type - BIO(..), Source, Sink + BIO, pattern EOF, Source, Sink -- ** Basic combinators - , (>|>), (>~>), (>!>), appendSource - , concatSource, zipSource, zipBIO + , appendSource, concatSource, concatSource' , joinSink, fuseSink -- * Run BIO chain - , runBIO - , runSource, runSource_ + , discard + , stepBIO, stepBIO_ + , runBIO, runBIO_ , runBlock, runBlock_, unsafeRunBlock , runBlocks, runBlocks_, unsafeRunBlocks -- * Make new BIO , pureBIO, ioBIO -- ** Source + , initSourceFromFile + , initSourceFromFile' , sourceFromIO , sourceFromList - , initSourceFromFile , sourceFromBuffered , sourceTextFromBuffered , sourceJSONFromBuffered , sourceParserFromBuffered - , sourceParseChunksFromBuffered + , sourceParseChunkFromBuffered -- ** Sink , sinkToIO , sinkToList @@ -74,22 +74,25 @@ , sinkToBuffered , sinkBuilderToBuffered -- ** Bytes specific - , newParserNode, newReChunk, newUTF8Decoder, newMagicSplitter, newLineSplitter + , newReChunk + , newUTF8Decoder + , newParserNode, newMagicSplitter, newLineSplitter , newBase64Encoder, newBase64Decoder , hexEncoder, newHexDecoder -- ** Generic BIO - , newCounterNode - , newSeqNumNode + , counterNode + , seqNumNode , newGroupingNode + , ungroupingNode + , consumedNode ) where +import Control.Concurrent.MVar import Control.Monad import Control.Monad.IO.Class import Data.Bits ((.|.)) import Data.IORef import qualified Data.List as List -import Data.Sequence (Seq (..)) -import qualified Data.Sequence as Seq import Data.Void import Data.Word import System.IO.Unsafe (unsafePerformIO) @@ -112,13 +115,13 @@ -- | A 'BIO'(blocked IO) node. -- --- A 'BIO' node consist of two functions: 'push' and 'pull'. It can be used to describe different kinds of IO +-- A 'BIO' node is a push based stream transformer. It can be used to describe different kinds of IO -- devices: -- -- * @BIO inp out@ describe an IO state machine(e.g. z_stream in zlib), -- which takes some input in block, then outputs. -- * @type Source out = BIO Void out@ described an IO source, which never takes input, --- but gives output until EOF when 'pull'ed. +-- but gives output until EOF by looping. -- * @type Sink inp = BIO inp Void@ described an IO sink, which takes input and perform some IO effects, -- such as writing to terminal or files. -- @@ -127,10 +130,11 @@ -- -- You can run a 'BIO' node in different ways: -- --- * 'runBIO' will continuously pull value from source, push to sink until source reaches EOF. --- * 'runSource' will continuously pull value from source, and perform effects along the way. --- * 'runBlock' will supply a single block of input as whole input, and return output if there's any. --- * 'runBlocks' will supply a list of blocks as whole input, and return a list of output blocks. +-- * 'stepBIO'\/'stepBIO_' to supply a single chunk of input and step the BIO node. +-- * 'runBIO'\/'runBIO_' will supply EOF directly, which will effectively pull all values from source, +-- and push to sink until source reaches EOF. +-- * 'runBlock'\/'runBlock_' will supply a single block of input as whole input and run the BIO node. +-- * 'runBlocks'\/'runBlocks_' will supply a list of blocks as whole input and run the BIO node. -- -- Note 'BIO' usually contains some IO states, you can consider it as an opaque 'IORef': -- @@ -140,232 +144,139 @@ -- 'BIO' is simply a convenient way to construct single-thread streaming computation, to use 'BIO' -- in multiple threads, check "Z.IO.BIO.Concurrent" module. -- -data BIO inp out = BIO - { push :: inp -> IO (Maybe out) - -- ^ Push a block of input, perform some effect, and return output, - -- if input is not enough to produce any output yet, return 'Nothing'. - , pull :: IO (Maybe out) - -- ^ When input reaches EOF, there may be a finalize stage to output - -- trailing output blocks. return 'Nothing' to indicate current node - -- reaches EOF too. - } +type BIO inp out = (Maybe out -> IO ()) -- ^ Pass 'EOF' to indicate current node reaches EOF + -> Maybe inp -- ^ 'EOF' indicates upstream reaches EOF + -> IO () +-- | Patterns for more meaningful pattern matching. +pattern EOF :: Maybe a +pattern EOF = Nothing + -- | Type alias for 'BIO' node which never takes input. -- --- 'push' is not available by type system, and 'pull' return 'Nothing' when --- reaches EOF. -type Source out = BIO Void out +-- Note when implement a 'Source', you should assume 'EOF' argument is supplied only once, and you +-- should loop to call downstream continuation with all available chunks, then write a final 'EOF' +-- to indicate EOF. +type Source x = BIO Void x -- | Type alias for 'BIO' node which only takes input and perform effects. -- --- 'push' doesn't produce any meaningful output, and 'pull' usually does a flush. -type Sink inp = BIO inp Void - -instance Functor (BIO inp) where - {-# INLINABLE fmap #-} - fmap f BIO{..} = BIO push_ pull_ - where - push_ inp = do - r <- push inp - return $! fmap f r - pull_ = do - r <- pull - return $! fmap f r - -infixl 3 >|> -infixl 3 >~> - --- | Connect two 'BIO' nodes, feed left one's output to right one's input. -(>|>) :: HasCallStack => BIO a b -> BIO b c -> BIO a c -{-# INLINE (>|>) #-} -BIO pushA pullA >|> BIO pushB pullB = BIO push_ pull_ - where - push_ inp = do - x <- pushA inp - case x of Just x' -> pushB x' - _ -> return Nothing - pull_ = do - x <- pullA - case x of - Just x' -> do - y <- pushB x' - case y of Nothing -> pull_ -- draw input from A until there's an output from B - _ -> return y - _ -> pullB - --- | Flipped 'fmap' for easier chaining. -(>~>) :: BIO a b -> (b -> c) -> BIO a c -{-# INLINE (>~>) #-} -(>~>) = flip fmap - --- | Connect BIO to an effectful function. -(>!>) :: HasCallStack => BIO a b -> (b -> IO c) -> BIO a c -{-# INLINE (>!>) #-} -(>!>) BIO{..} f = BIO push_ pull_ - where - push_ x = push x >>= \ r -> - case r of Just r' -> Just <$!> f r' - _ -> return Nothing - pull_ = pull >>= \ r -> - case r of Just r' -> Just <$!> f r' - _ -> return Nothing +-- Note when implement a 'Sink', you should assume 'EOF' argument is supplied only once(when upstream +-- reaches EOF), you do not need to call downstream continuation before EOF, and +-- do a flush(also write a final 'EOF') when upstream reach EOF. +type Sink x = BIO x Void --- | Connect two 'BIO' source, after first reach EOF, draw element from second. -appendSource :: HasCallStack => Source a -> Source a -> IO (Source a) +-- | Connect two 'BIO' source, after first reach EOF, draw elements from second. +appendSource :: HasCallStack => Source a -> Source a -> Source a {-# INLINE appendSource #-} -b1 `appendSource` b2 = concatSource [b1, b2] +b1 `appendSource` b2 = \ k _ -> + b1 (\ y -> + case y of Just _ -> k y + _ -> b2 k EOF) EOF -- | Fuse two 'BIO' sinks, i.e. everything written to the fused sink will be written to left and right sink. -- -- Flush result 'BIO' will effectively flush both sink. joinSink :: HasCallStack => Sink out -> Sink out -> Sink out {-# INLINE joinSink #-} -b1 `joinSink` b2 = fuseSink [b1, b2] +b1 `joinSink` b2 = \ k mx -> + case mx of + Just _ -> do + b1 discard mx + b2 discard mx + _ -> do + b1 discard EOF + b2 discard EOF + k EOF -- | Fuse a list of 'BIO' sinks, everything written to the fused sink will be written to every sink in the list. -- -- Flush result 'BIO' will effectively flush every sink in the list. fuseSink :: HasCallStack => [Sink out] -> Sink out {-# INLINABLE fuseSink #-} -fuseSink ss = BIO push_ pull_ - where - push_ inp = forM_ ss (\ b -> push b inp) >> return Nothing - pull_ = mapM_ pull ss >> return Nothing +fuseSink ss = \ k mx -> + case mx of + Just _ -> mapM_ (\ s -> s discard mx) ss + _ -> do + mapM_ (\ s -> s discard mx) ss + k EOF -- | Connect list of 'BIO' sources, after one reach EOF, draw element from next. -concatSource :: HasCallStack => [Source a] -> IO (Source a) +concatSource :: HasCallStack => [Source a] -> Source a {-# INLINABLE concatSource #-} -concatSource ss0 = newIORef ss0 >>= \ ref -> return (BIO{ pull = loop ref}) - where - loop ref = do - ss <- readIORef ref - case ss of - [] -> return Nothing - (s:rest) -> do - r <- pull s - case r of - Just _ -> return r - _ -> writeIORef ref rest >> loop ref - --- | Zip two 'BIO' source into one, reach EOF when either one reached EOF. -zipSource :: HasCallStack => Source a -> Source b -> IO (Source (a,b)) -{-# INLINABLE zipSource #-} -zipSource (BIO _ pullA) (BIO _ pullB) = do - finRef <- newIORef False - return $ BIO { pull = do - fin <- readIORef finRef - if fin - then return Nothing - else do - mA <- pullA - mB <- pullB - let r = (,) <$> mA <*> mB - case r of - Just _ -> return r - _ -> writeIORef finRef True >> return Nothing - } +concatSource = List.foldl' appendSource emptySource --- | Zip two 'BIO' node into one, reach EOF when either one reached EOF. --- --- The output item number should match, unmatched output will be discarded. -zipBIO :: HasCallStack => BIO a b -> BIO a c -> IO (BIO a (b, c)) -{-# INLINABLE zipBIO #-} -zipBIO (BIO pushA pullA) (BIO pushB pullB) = do - finRef <- newIORef False - aSeqRef <- newIORef Seq.Empty - bSeqRef <- newIORef Seq.Empty - return (BIO (push_ aSeqRef bSeqRef) (pull_ finRef aSeqRef bSeqRef)) - where - push_ aSeqRef bSeqRef x = do - ma <- pushA x - mb <- pushB x - forM_ ma (\ a -> modifyIORef' aSeqRef (a :<|)) - forM_ mb (\ b -> modifyIORef' bSeqRef (b :<|)) - aSeq <- readIORef aSeqRef - bSeq <- readIORef bSeqRef - case aSeq of - (!as :|> a) -> case bSeq of - (!bs :|> b) -> do - writeIORef aSeqRef as - writeIORef bSeqRef bs - return (Just (a, b)) - _ -> return Nothing - _ -> return Nothing +-- | A 'Source' directly write EOF to downstream. +emptySource :: Source a +{-# INLINABLE emptySource #-} +emptySource = \ k _ -> k EOF - pull_ finRef aSeqRef bSeqRef = do - fin <- readIORef finRef - if fin - then return Nothing - else do - aSeq <- readIORef aSeqRef - bSeq <- readIORef bSeqRef - ma <- case aSeq of (_ :|> a) -> return (Just a) - _ -> pullA - mb <- case bSeq of (_ :|> b) -> return (Just b) - _ -> pullB - case ma of - Just a -> case mb of - Just b -> return (Just (a, b)) - _ -> writeIORef finRef True >> return Nothing - _ -> writeIORef finRef True >> return Nothing +-- | Connect list of 'BIO' sources, after one reach EOF, draw element from next. +concatSource' :: HasCallStack => Source (Source a) -> Source a +{-# INLINABLE concatSource' #-} +concatSource' ssrc = \ k _ -> ssrc (\ msrc -> + case msrc of + Just src -> src (\ mx -> + case mx of Just _ -> k mx + _ -> return ()) EOF + _ -> k EOF) EOF ------------------------------------------------------------------------------- -- Run BIO --- | Run a 'BIO' loop (source >|> ... >|> sink). -runBIO :: HasCallStack => BIO Void Void -> IO () -{-# INLINABLE runBIO #-} -runBIO BIO{..} = pull >> return () +-- | Discards a value. +discard :: a -> IO () +{-# INLINABLE discard #-} +discard _ = return () --- | Drain a 'BIO' source into a List in memory. -runSource :: HasCallStack => Source x -> IO [x] -{-# INLINABLE runSource #-} -runSource BIO{..} = loop pull [] - where - loop f acc = do - r <- f - case r of Just r' -> loop f (r':acc) - _ -> return (List.reverse acc) +-- | Supply a single chunk of input to a 'BIO' and collect result. +stepBIO :: HasCallStack => BIO inp out -> inp -> IO [out] +{-# INLINABLE stepBIO #-} +stepBIO bio inp = do + accRef <- newIORef [] + bio (mapM_ $ \ x -> modifyIORef' accRef (x:)) (Just inp) + reverse <$> readIORef accRef --- | Drain a source without collecting result. -runSource_ :: HasCallStack => Source x -> IO () -{-# INLINABLE runSource_ #-} -runSource_ BIO{..} = loop pull - where - loop f = do - r <- f - case r of Just _ -> loop f - _ -> return () +-- | Supply a single chunk of input to a 'BIO' without collecting result. +stepBIO_ :: HasCallStack => BIO inp out -> inp -> IO () +{-# INLINABLE stepBIO_ #-} +stepBIO_ bio = bio discard . Just --- | Supply a single block of input, then run BIO node until EOF. +-- | Run a 'BIO' loop without providing input. -- --- Note many 'BIO' node will be closed or not be able to take new input after drained. +-- When used on 'Source', it starts the streaming loop. +-- When used on 'Sink', it performs a flush. +runBIO_ :: HasCallStack => BIO inp out -> IO () +{-# INLINABLE runBIO_ #-} +runBIO_ bio = bio discard EOF + +-- | Run a 'BIO' loop without providing input, and collect result. +-- +-- When used on 'Source', it will collect all input chunks. +runBIO :: HasCallStack => BIO inp out -> IO [out] +{-# INLINABLE runBIO #-} +runBIO bio = do + accRef <- newIORef [] + bio (mapM_ $ \ x -> modifyIORef' accRef (x:)) EOF + reverse <$> readIORef accRef + +-- | Run a 'BIO' loop with a single chunk of input and EOF, and collect result. +-- runBlock :: HasCallStack => BIO inp out -> inp -> IO [out] {-# INLINABLE runBlock #-} -runBlock BIO{..} inp = do - x <- push inp - let acc = case x of Just x' -> [x'] - _ -> [] - loop pull acc - where - loop f acc = do - r <- f - case r of Just r' -> loop f (r':acc) - _ -> return (List.reverse acc) +runBlock bio inp = do + accRef <- newIORef [] + bio (mapM_ $ \ x -> modifyIORef' accRef (x:)) (Just inp) + bio (mapM_ $ \ x -> modifyIORef' accRef (x:)) EOF + reverse <$> readIORef accRef --- | Supply a single block of input, then run BIO node until EOF with collecting result. +-- | Run a 'BIO' loop with a single chunk of input and EOF, without collecting result. -- --- Note many 'BIO' node will be closed or not be able to take new input after drained. runBlock_ :: HasCallStack => BIO inp out -> inp -> IO () {-# INLINABLE runBlock_ #-} -runBlock_ BIO{..} inp = do - _ <- push inp - loop pull - where - loop f = do - r <- f - case r of Just _ -> loop f - _ -> return () +runBlock_ bio inp = do + bio discard (Just inp) + bio discard EOF -- | Wrap 'runBlock' into a pure interface. -- @@ -375,38 +286,25 @@ {-# INLINABLE unsafeRunBlock #-} unsafeRunBlock new inp = unsafePerformIO (new >>= \ bio -> runBlock bio inp) --- | Supply blocks of input, then run BIO node until EOF. +-- | Supply blocks of input and EOF to a 'BIO', and collect results. -- -- Note many 'BIO' node will be closed or not be able to take new input after drained. runBlocks :: HasCallStack => BIO inp out -> [inp] -> IO [out] {-# INLINABLE runBlocks #-} -runBlocks BIO{..} = loop [] - where - loop acc (inp:inps) = do - r <- push inp - case r of - Just r' -> loop (r':acc) inps - _ -> loop acc inps - loop acc [] = loop' acc - loop' acc = do - r <- pull - case r of - Just r' -> loop' (r':acc) - _ -> return (List.reverse acc) +runBlocks bio inps = do + accRef <- newIORef [] + forM_ inps $ bio (mapM_ $ \ x -> modifyIORef' accRef (x:)) . Just + bio (mapM_ $ \ x -> modifyIORef' accRef (x:)) EOF + reverse <$> readIORef accRef --- | Supply blocks of input, then run BIO node until EOF with collecting result. +-- | Supply blocks of input and EOF to a 'BIO', without collecting results. -- -- Note many 'BIO' node will be closed or not be able to take new input after drained. runBlocks_ :: HasCallStack => BIO inp out -> [inp] -> IO () {-# INLINABLE runBlocks_ #-} -runBlocks_ bio (inp:inps) = push bio inp >> runBlocks_ bio inps -runBlocks_ bio [] = loop - where - loop = do - r <- pull bio - case r of - Just _ -> loop - _ -> return () +runBlocks_ bio inps = do + forM_ inps $ bio discard . Just + bio discard EOF -- | Wrap 'runBlocks' into a pure interface. -- @@ -418,66 +316,72 @@ ------------------------------------------------------------------------------- -- Source --- | Source a list from memory. +-- | Source a list(or any 'Foldable') from memory. -- -sourceFromList :: [a] -> IO (Source a) -sourceFromList xs0 = do - xsRef <- newIORef xs0 - return BIO{ pull = popper xsRef } - where - popper xsRef = do - xs <- readIORef xsRef - case xs of - (x:xs') -> do - writeIORef xsRef xs' - return (Just x) - _ -> return Nothing +sourceFromList :: Foldable f => f a -> Source a +sourceFromList xs0 = \ k _ -> do + mapM_ (k . Just) xs0 + k EOF --- | Turn a 'BufferedInput' into 'BIO' source, map EOF to Nothing. +-- | Turn a 'BufferedInput' into 'BIO' source, map EOF to EOF. -- sourceFromBuffered :: HasCallStack => BufferedInput -> Source V.Bytes {-# INLINABLE sourceFromBuffered #-} -sourceFromBuffered i = BIO{ pull = do - readBuffer i >>= \ x -> if V.null x then return Nothing - else return (Just x)} +sourceFromBuffered i = \ k _ -> + let loop = readBuffer i >>= \ x -> + if V.null x then k EOF else k (Just x) >> loop + in loop -- | Turn a `IO` action into 'Source' sourceFromIO :: HasCallStack => IO (Maybe a) -> Source a {-# INLINABLE sourceFromIO #-} -sourceFromIO io = BIO{ pull = io } +sourceFromIO io = \ k _ -> + let loop = io >>= \ x -> + case x of + Just _ -> k x >> loop + _ -> k EOF + in loop --- | Turn a UTF8 encoded 'BufferedInput' into 'BIO' source, map EOF to Nothing. +-- | Turn a UTF8 encoded 'BufferedInput' into 'BIO' source, map EOF to EOF. -- sourceTextFromBuffered :: HasCallStack => BufferedInput -> Source T.Text {-# INLINABLE sourceTextFromBuffered #-} -sourceTextFromBuffered i = BIO{ pull = do - readBufferText i >>= \ x -> if T.null x then return Nothing - else return (Just x)} +sourceTextFromBuffered i = \ k _ -> + let loop = readBufferText i >>= \ x -> + if T.null x then k EOF else k (Just x) >> loop + in loop -- | Turn a 'JSON' encoded 'BufferedInput' into 'BIO' source, ignoring any --- whitespaces bewteen JSON objects. If EOF reached, then return Nothing. +-- whitespaces bewteen JSON objects. If EOF reached, then return 'EOF'. -- Throw 'OtherError' with name "EJSON" if JSON value is not parsed or converted. +-- sourceJSONFromBuffered :: forall a. (JSON.JSON a, HasCallStack) => BufferedInput -> Source a {-# INLINABLE sourceJSONFromBuffered #-} -sourceJSONFromBuffered = sourceParseChunksFromBuffered JSON.decodeChunks +sourceJSONFromBuffered = sourceParseChunkFromBuffered JSON.decodeChunk -- | Turn buffered input device into a packet source, throw 'OtherError' with name @EPARSE@ if parsing fail. sourceParserFromBuffered :: HasCallStack => P.Parser a -> BufferedInput -> Source a {-# INLINABLE sourceParserFromBuffered #-} -sourceParserFromBuffered p = sourceParseChunksFromBuffered (P.parseChunks p) +sourceParserFromBuffered p = sourceParseChunkFromBuffered (P.parseChunk p) -- | Turn buffered input device into a packet source, throw 'OtherError' with name @EPARSE@ if parsing fail. -sourceParseChunksFromBuffered :: (HasCallStack, T.Print e) => P.ParseChunks IO V.Bytes e a -> BufferedInput -> Source a -{-# INLINABLE sourceParseChunksFromBuffered #-} -sourceParseChunksFromBuffered cp bi = BIO{ pull = do - bs <- readBuffer bi - if V.null bs - then return Nothing - else do - (rest, r) <- cp (readBuffer bi) bs - unReadBuffer rest bi - case r of Right v -> return (Just v) - Left e -> throwOtherError "EPARSE" (T.toText e) } +sourceParseChunkFromBuffered :: (HasCallStack, T.Print e) + => (V.Bytes -> P.Result e a) -> BufferedInput -> Source a +{-# INLINABLE sourceParseChunkFromBuffered #-} +sourceParseChunkFromBuffered pc bi = \ k _ -> + let loopA = do + bs <- readBuffer bi + if V.null bs + then k EOF + else loopB bs + loopB bs = do + (rest, r) <- P.parseChunks pc (readBuffer bi) bs + case r of Right v -> k (Just v) + Left e -> throwOtherError "EPARSE" (T.toText e) + if V.null rest + then loopA + else loopB rest + in loopA -- | Turn a file into a 'V.Bytes' source. initSourceFromFile :: HasCallStack => CBytes -> Resource (Source V.Bytes) @@ -486,25 +390,32 @@ f <- FS.initFile p FS.O_RDONLY FS.DEFAULT_FILE_MODE liftIO (sourceFromBuffered <$> newBufferedInput f) +-- | Turn a file into a 'V.Bytes' source with given chunk size. +initSourceFromFile' :: HasCallStack => CBytes -> Int -> Resource (Source V.Bytes) +{-# INLINABLE initSourceFromFile' #-} +initSourceFromFile' p bufSiz = do + f <- FS.initFile p FS.O_RDONLY FS.DEFAULT_FILE_MODE + liftIO (sourceFromBuffered <$> newBufferedInput' bufSiz f) + -------------------------------------------------------------------------------- -- Sink -- | Turn a 'BufferedOutput' into a 'V.Bytes' sink. sinkToBuffered :: HasCallStack => BufferedOutput -> Sink V.Bytes {-# INLINABLE sinkToBuffered #-} -sinkToBuffered bo = BIO push_ pull_ - where - push_ inp = writeBuffer bo inp >> pure Nothing - pull_ = flushBuffer bo >> pure Nothing +sinkToBuffered bo = \ k mbs -> + case mbs of + Just bs -> writeBuffer bo bs + _ -> flushBuffer bo >> k EOF -- | Turn a 'BufferedOutput' into a 'B.Builder' sink. -- sinkBuilderToBuffered :: HasCallStack => BufferedOutput -> Sink (B.Builder a) {-# INLINABLE sinkBuilderToBuffered #-} -sinkBuilderToBuffered bo = BIO push_ pull_ - where - push_ inp = writeBuilder bo inp >> pure Nothing - pull_ = flushBuffer bo >> pure Nothing +sinkBuilderToBuffered bo = \ k mbs -> + case mbs of + Just bs -> writeBuilder bo bs + _ -> flushBuffer bo >> k EOF -- | Turn a file into a 'V.Bytes' sink. -- @@ -518,25 +429,33 @@ -- | Turn an `IO` action into 'BIO' sink. -- --- 'push' will call `IO` action with input chunk, `pull` has no effect. sinkToIO :: HasCallStack => (a -> IO ()) -> Sink a {-# INLINABLE sinkToIO #-} -sinkToIO f = BIO push_ pull_ - where - push_ x = f x >> pure Nothing - pull_ = pure Nothing +sinkToIO f = \ k ma -> + case ma of + Just a -> f a + _ -> k EOF --- | Sink to a list in memory. +-- | Turn an `IO` action(and a flush action), into 'BIO' sink. -- --- The list's 'IORef' is not thread safe here, --- and list items are in reversed order during sinking(will be reversed when flushed, i.e. pulled), --- Please don't use it in multiple thread. +sinkToIO' :: HasCallStack => (a -> IO ()) -> IO () -> Sink a +{-# INLINABLE sinkToIO' #-} +sinkToIO' f flush = \ k ma -> + case ma of + Just a -> f a + _ -> flush >> k EOF + +-- | Sink to a list in memory. -- -sinkToList :: IO (IORef [a], Sink a) +-- The 'MVar' will be empty during sinking, and will be filled after sink receives an EOF. +sinkToList :: IO (MVar [a], Sink a) sinkToList = do xsRef <- newIORef [] - return (xsRef, BIO (\ x -> modifyIORef xsRef (x:) >> return Nothing) - (modifyIORef xsRef reverse >> return Nothing)) + rRef <- newEmptyMVar + return (rRef, sinkToIO' (\ x -> modifyIORef xsRef (x:)) + (do modifyIORef xsRef reverse + xs <- readIORef xsRef + putMVar rRef xs)) -------------------------------------------------------------------------------- -- Nodes @@ -545,14 +464,18 @@ -- -- BIO node made with this funtion are stateless, thus can be reused across chains. pureBIO :: (a -> b) -> BIO a b -pureBIO f = BIO (\ x -> let !r = f x in return (Just r)) (return Nothing) +{-# INLINE pureBIO #-} +pureBIO f = \ k x -> k (f <$> x) -- | BIO node from an IO function. -- -- BIO node made with this funtion may not be stateless, it depends on if the IO function use -- IO state. -ioBIO :: (HasCallStack => a -> IO b) -> BIO a b -ioBIO f = BIO (\ x -> Just <$!> f x) (return Nothing) +ioBIO :: HasCallStack => (a -> IO b) -> BIO a b +{-# INLINE ioBIO #-} +ioBIO f = \ k x -> + case x of Just x' -> f x' >>= k . Just + _ -> k EOF -- | Make a chunk size divider. -- @@ -563,75 +486,55 @@ {-# INLINABLE newReChunk #-} newReChunk n = do trailingRef <- newIORef V.empty - return (BIO (push_ trailingRef) (pull_ trailingRef)) - where - push_ trailingRef bs = do - trailing <- readIORef trailingRef - let chunk = trailing `V.append` bs - l = V.length chunk - if l >= n - then do - let l' = l - (l `rem` n) - (chunk', rest) = V.splitAt l' chunk - writeIORef trailingRef rest - return (Just chunk') - else do - writeIORef trailingRef chunk - return Nothing - pull_ trailingRef = do - trailing <- readIORef trailingRef - if V.null trailing - then return Nothing - else do - writeIORef trailingRef V.empty - return (Just trailing) + return $ \ k mbs -> + case mbs of + Just bs -> do + trailing <- readIORef trailingRef + let chunk = trailing `V.append` bs + l = V.length chunk + if l >= n + then do + let l' = l - (l `rem` n) + (chunk', rest) = V.splitAt l' chunk + writeIORef trailingRef rest + k (Just chunk') + else writeIORef trailingRef chunk + _ -> do + trailing <- readIORef trailingRef + unless (V.null trailing) $ do + writeIORef trailingRef V.empty + k (Just trailing) + k EOF -- | Read buffer and parse with 'Parser'. -- --- This function will continuously draw data from input before parsing finish. --- Unconsumed bytes will be returned to buffer. +-- This function will turn a 'Parser' into a 'BIO', throw 'OtherError' with name @EPARSE@ if parsing fail. -- --- Return 'Nothing' if reach EOF before parsing, throw 'OtherError' with name @EPARSE@ if parsing fail. newParserNode :: HasCallStack => P.Parser a -> IO (BIO V.Bytes a) {-# INLINABLE newParserNode #-} newParserNode p = do - -- type LastParseState = Either V.Bytes (V.Bytes -> P.Result) - resultRef <- newIORef (Left V.empty) - return (BIO (push_ resultRef) (pull_ resultRef)) - where - push_ resultRef bs = do - lastResult <- readIORef resultRef - let (chunk, f) = case lastResult of - Left trailing -> (trailing `V.append` bs, P.parseChunk p) - Right x -> (bs, x) - case f chunk of - P.Success a trailing' -> do - writeIORef resultRef (Left trailing') - return (Just a) - P.Failure e _ -> - throwOtherError "EPARSE" (T.toText e) - P.Partial f' -> do - writeIORef resultRef (Right f') - return Nothing - - pull_ resultRef = do + -- type LastParseState = Maybe (V.Bytes -> P.Result) + resultRef <- newIORef EOF + return $ \ k mbs -> do + let loop f chunk = case f chunk of + P.Success a trailing -> do + k (Just a) + unless (V.null trailing) (loop f trailing) + P.Partial f' -> + writeIORef resultRef (Just f') + P.Failure e _ -> + throwOtherError "EPARSE" (T.toText e) lastResult <- readIORef resultRef - case lastResult of - Left trailing -> - if V.null trailing - then return Nothing - else lastChunk resultRef (P.parseChunk p) trailing - Right f -> lastChunk resultRef f V.empty - - lastChunk resultRef f chunk = - case f chunk of - P.Success a trailing' -> do - writeIORef resultRef (Left trailing') - return (Just a) - P.Failure e _ -> - throwOtherError "EPARSE" (T.toText e) - P.Partial _ -> - throwOtherError "EPARSE" "last chunk partial parse" + case mbs of + Just bs -> do + let f = case lastResult of + Just x -> x + _ -> P.parseChunk p + loop f bs + _ -> + case lastResult of + Just f -> loop f V.empty + _ -> k EOF -- | Make a new UTF8 decoder, which decode bytes streams into text streams. -- @@ -647,34 +550,32 @@ {-# INLINABLE newUTF8Decoder #-} newUTF8Decoder = do trailingRef <- newIORef V.empty - return (BIO (push_ trailingRef) (pull_ trailingRef)) - where - push_ trailingRef bs = do - trailing <- readIORef trailingRef - let chunk = trailing `V.append` bs - (V.PrimVector arr s l) = chunk - if l > 0 && T.decodeCharLen arr s <= l - then do - let (i, _) = V.findR (\ w -> w >= 0b11000000 || w <= 0b01111111) chunk - if (i == -1) - then throwOtherError "EINVALIDUTF8" "invalid UTF8 bytes" - else do - if T.decodeCharLen arr (s + i) > l - i + return $ \ k mbs -> do + case mbs of + Just bs -> do + trailing <- readIORef trailingRef + let chunk = trailing `V.append` bs + (V.PrimVector arr s l) = chunk + if l > 0 && T.decodeCharLen arr s <= l then do - writeIORef trailingRef (V.fromArr arr (s+i) (l-i)) - return (Just (T.validate (V.fromArr arr s i))) - else do - writeIORef trailingRef V.empty - return (Just (T.validate chunk)) - else do - writeIORef trailingRef chunk - return Nothing + let (i, _) = V.findR (\ w -> w >= 0b11000000 || w <= 0b01111111) chunk + if (i == -1) + then throwOtherError "EINVALIDUTF8" "invalid UTF8 bytes" + else do + if T.decodeCharLen arr (s + i) > l - i + then do + writeIORef trailingRef (V.fromArr arr (s+i) (l-i)) + k (Just (T.validate (V.fromArr arr s i))) + else do + writeIORef trailingRef V.empty + k (Just (T.validate chunk)) + else writeIORef trailingRef chunk - pull_ trailingRef = do - trailing <- readIORef trailingRef - if V.null trailing - then return Nothing - else throwOtherError "EINVALIDUTF8" "invalid UTF8 bytes" + _ -> do + trailing <- readIORef trailingRef + if V.null trailing + then k EOF + else throwOtherError "EINVALIDUTF8" "invalid UTF8 bytes" -- | Make a new stream splitter based on magic byte. -- @@ -682,32 +583,24 @@ {-# INLINABLE newMagicSplitter #-} newMagicSplitter magic = do trailingRef <- newIORef V.empty - return (BIO (push_ trailingRef) (pull_ trailingRef)) - where - push_ trailingRef bs = do - trailing <- readIORef trailingRef - let chunk = trailing `V.append` bs - case V.elemIndex magic chunk of - Just i -> do - let (line, rest) = V.splitAt (i+1) chunk - writeIORef trailingRef rest - return (Just line) - _ -> do - writeIORef trailingRef chunk - return Nothing - - pull_ trailingRef = do - chunk <- readIORef trailingRef - if V.null chunk - then return Nothing - else case V.elemIndex magic chunk of - Just i -> do - let (line, rest) = V.splitAt (i+1) chunk - writeIORef trailingRef rest - return (Just line) + return $ \ k mx -> + case mx of + Just bs -> do + trailing <- readIORef trailingRef + let loop chunk = case V.elemIndex magic chunk of + Just i -> do + -- TODO: looping + let (line, rest) = V.splitAt (i+1) chunk + k (Just line) + loop rest + _ -> writeIORef trailingRef chunk + loop (trailing `V.append` bs) _ -> do - writeIORef trailingRef V.empty - return (Just chunk) + chunk <- readIORef trailingRef + unless (V.null chunk) $ do + writeIORef trailingRef V.empty + k (Just chunk) + k EOF -- | Make a new stream splitter based on linefeed(@\r\n@ or @\n@). -- @@ -716,7 +609,7 @@ {-# INLINABLE newLineSplitter #-} newLineSplitter = do s <- newMagicSplitter 10 - return (s >~> dropLineEnd) + return (s . pureBIO dropLineEnd) where dropLineEnd bs@(V.PrimVector arr s l) = case bs `V.indexMaybe` (l-2) of @@ -730,14 +623,14 @@ {-# INLINABLE newBase64Encoder #-} newBase64Encoder = do re <- newReChunk 3 - return (re >~> base64Encode) + return (re . pureBIO base64Encode) -- | Make a new base64 decoder node. newBase64Decoder :: HasCallStack => IO (BIO V.Bytes V.Bytes) {-# INLINABLE newBase64Decoder #-} newBase64Decoder = do re <- newReChunk 4 - return (re >~> base64Decode') + return (re . pureBIO base64Decode') -- | Make a hex encoder node. -- @@ -752,69 +645,82 @@ {-# INLINABLE newHexDecoder #-} newHexDecoder = do re <- newReChunk 2 - return (re >~> hexDecode') + return (re . pureBIO hexDecode') -- | Make a new BIO node which counts items flow throught it. -- --- Returned 'Counter' is increased atomically, it's safe to read \/ reset the counter from other threads. -newCounterNode :: IO (Counter, BIO a a) -{-# INLINABLE newCounterNode #-} -newCounterNode = do - c <- newCounter 0 - return (c, BIO (push_ c) (return Nothing)) +-- 'Counter' is increased atomically, it's safe to read \/ reset the counter from other threads. +counterNode :: Counter -> BIO a a +{-# INLINABLE counterNode #-} +counterNode c = ioBIO inc where - push_ c x = do + inc x = do atomicAddCounter_ c 1 - return (Just x) + return x -- | Make a new BIO node which counts items, and label item with a sequence number. -- --- Returned 'Counter' is increased atomically, it's safe to read \/ reset the counter from other threads. -newSeqNumNode :: IO (Counter, BIO a (Int, a)) -{-# INLINABLE newSeqNumNode #-} -newSeqNumNode = do - c <- newCounter 0 - return (c, BIO (push_ c) (return Nothing)) +-- 'Counter' is increased atomically, it's safe to read \/ reset the counter from other threads. +seqNumNode :: Counter -> BIO a (Int, a) +{-# INLINABLE seqNumNode #-} +seqNumNode c = ioBIO inc where - push_ c x = do - !i <- atomicAddCounter c 1 - return (Just (i, x)) + inc x = do + i <- atomicAddCounter c 1 + return (i, x) -- | Make a BIO node grouping items into fixed size arrays. -- -newGroupingNode :: Int -> IO (BIO a (A.SmallArray a)) +-- Trailing items are directly returned. +newGroupingNode :: Int -> IO (BIO a (V.Vector a)) {-# INLINABLE newGroupingNode #-} newGroupingNode n | n < 1 = newGroupingNode 1 | otherwise = do c <- newCounter 0 arrRef <- newIORef =<< A.newArr n - return (BIO (push_ c arrRef) (pull_ c arrRef)) - where - push_ c arrRef x = do - i <- readPrimIORef c - if i == n - 1 - then do - marr <- readIORef arrRef - A.writeArr marr i x - writePrimIORef c 0 - writeIORef arrRef =<< A.newArr n - return . Just =<< A.unsafeFreezeArr marr - else do - marr <- readIORef arrRef - A.writeArr marr i x - writePrimIORef c (i+1) - return Nothing - pull_ c arrRef = do - i <- readPrimIORef c - if i /= 0 - then do - marr <- readIORef arrRef -#if MIN_VERSION_base(4,14,0) - A.shrinkMutableArr marr i - return . Just =<< A.unsafeFreezeArr marr -#else - marr' <- A.resizeMutableArr marr i - return . Just =<< A.unsafeFreezeArr marr' -#endif - else return Nothing + return $ \ k mx -> + case mx of + Just x -> do + i <- readPrimIORef c + if i == n - 1 + then do + marr <- readIORef arrRef + A.writeArr marr i x + writePrimIORef c 0 + writeIORef arrRef =<< A.newArr n + arr <- A.unsafeFreezeArr marr + k . Just $! V.fromArr arr 0 n + else do + marr <- readIORef arrRef + A.writeArr marr i x + writePrimIORef c (i+1) + _ -> do + i <- readPrimIORef c + if i /= 0 + then do + writePrimIORef c 0 + marr <- readIORef arrRef + A.shrinkMutableArr marr i + arr <- A.unsafeFreezeArr marr + k . Just $! V.fromArr arr 0 i + else k EOF + +-- | A BIO node flatten items. +-- +ungroupingNode :: BIO (V.Vector a) a +{-# INLINABLE ungroupingNode #-} +ungroupingNode = \ k mx -> + case mx of + Just x -> V.traverseVec_ (k . Just) x + _ -> k EOF + +-- | A BIO node which write 'True' to 'IORef' when 'EOF' is reached. +consumedNode :: IORef Bool -> BIO a a +{-# INLINABLE consumedNode #-} +consumedNode ref = \ k mx -> case mx of + Just _ -> k mx + _ -> do writeIORef ref True + k EOF + + diff --git a/Z/IO/BIO/Concurrent.hs b/Z/IO/BIO/Concurrent.hs --- a/Z/IO/BIO/Concurrent.hs +++ b/Z/IO/BIO/Concurrent.hs @@ -36,7 +36,7 @@ forkIO $ do ... - (runBIO $ ... >|> sink) -- producer using BIO + (runBIO $ ... . sink) -- producer using BIO `onException` (pull sink) -------------------------------------------------------------------------------- @@ -46,11 +46,11 @@ ... r <- pull src -- consumer using pull case r of Just r' -> ... - _ -> ... -- Nothing indicate all producers reached EOF + _ -> ... -- EOF indicate all producers reached EOF forkIO $ do ... - runBIO $ src >|> ... -- consumer using BIO + runBIO $ src . ... -- consumer using BIO @ -} @@ -58,12 +58,51 @@ module Z.IO.BIO.Concurrent where import Control.Monad +import Control.Concurrent import Control.Concurrent.STM +import qualified Data.Sequence as Seq +import Data.Sequence (Seq((:<|),(:|>))) import GHC.Natural import Z.IO.BIO import Z.Data.PrimRef import Z.IO.Exception +-- | Zip two BIO node by running them concurrently. +-- +-- This implementation use 'MVar' to synchronize two BIO's output, which has some implications: +-- +-- * Two node should output same numebr of results. +-- * If the number differs, one node maybe +-- +zipBIO :: BIO a b -> BIO a c -> BIO a (b,c) +zipBIO b1 b2 = \ k mx -> do + bEOF <- newTVarIO False + cEOF <- newTVarIO False + bBuf <- newTVarIO Seq.empty + cBuf <- newTVarIO Seq.empty + _ <- forkIO (b1 (f bBuf bEOF) mx) + _ <- forkIO (b2 (f cBuf cEOF) mx) + loop k bBuf cBuf bEOF cEOF + where + f xBuf xEOF = \ mx -> + case mx of + Just x -> atomically $ modifyTVar' xBuf (:|> x) + _ -> atomically $ writeTVar xEOF True + + loop k bBuf cBuf bEOF cEOF = join . atomically $ do + bs <- readTVar bBuf + cs <- readTVar cBuf + beof <- readTVar bEOF + ceof <- readTVar cEOF + case bs of + b :<| bs' -> case cs of + c :<| cs' -> do + writeTVar bBuf bs' + writeTVar cBuf cs' + return (k (Just (b, c)) >> loop k bBuf cBuf bEOF cEOF) + _ -> if ceof then return (k EOF) else retry + _ -> if beof then return (k EOF) else retry + -- | Make an unbounded queue and a pair of sink and souce connected to it. newTQueueNode :: Int -- ^ number of producers -> IO (Sink a, Source a) @@ -71,15 +110,21 @@ q <- newTQueueIO ec <- newCounter 0 return - ( BIO (\ x -> atomically (writeTQueue q (Just x)) >> return Nothing) - (do i <- atomicAddCounter' ec 1 - when (i == n) (atomically (writeTQueue q Nothing)) - return Nothing) - , BIO { pull = uninterruptibleMask $ \ restore -> do + ( \ k mx -> case mx of + Just _ -> atomically (writeTQueue q mx) + _ -> do + i <- atomicAddCounter' ec 1 + when (i == n) $ do + atomically (writeTQueue q EOF) + k EOF + + , \ k _ -> + let loop = uninterruptibleMask $ \ restore -> do x <- restore $ atomically (readTQueue q) - case x of Just _ -> return x - _ -> do atomically (unGetTQueue q Nothing) - return Nothing}) + case x of Just _ -> k x >> loop + _ -> do atomically (unGetTQueue q EOF) + k EOF + in loop) -- | Make an bounded queue and a pair of sink and souce connected to it. newTBQueueNode :: Int -- ^ number of producers @@ -89,15 +134,21 @@ q <- newTBQueueIO bound ec <- newCounter 0 return - ( BIO (\ x -> atomically (writeTBQueue q (Just x)) >> return Nothing) - (do i <- atomicAddCounter' ec 1 - when (i == n) (atomically (writeTBQueue q Nothing)) - return Nothing) - , BIO { pull = uninterruptibleMask $ \ restore -> do + ( \ k mx -> case mx of + Just _ -> atomically (writeTBQueue q mx) + _ -> do + i <- atomicAddCounter' ec 1 + when (i == n) $ do + atomically (writeTBQueue q EOF) + k EOF + + , \ k _ -> + let loop = uninterruptibleMask $ \ restore -> do x <- restore $ atomically (readTBQueue q) - case x of Just _ -> return x - _ -> do atomically (unGetTBQueue q Nothing) - return Nothing}) + case x of Just _ -> k x >> loop + _ -> do atomically (unGetTBQueue q EOF) + k EOF + in loop) -- | Make a broadcast chan and a sink connected to it, and a function return sources to receive broadcast message. newBroadcastTChanNode :: Int -- ^ number of producers @@ -107,12 +158,17 @@ ec <- newCounter 0 let dupSrc = do c <- atomically (dupTChan b) - return (BIO { pull = do - x <- atomically (readTChan c) - case x of Just _ -> return x - _ -> return Nothing }) - return ( BIO (\ x -> atomically (writeTChan b (Just x)) >> return Nothing) - (do i <- atomicAddCounter' ec 1 - when (i == n) (atomically (writeTChan b Nothing)) - return Nothing) - , dupSrc) + return $ \ k _ -> + let loop = do + x <- atomically (readTChan c) + case x of Just _ -> k x >> loop + _ -> k EOF + in loop + + return + (\ k mx -> case mx of + Just _ -> atomically (writeTChan b mx) + _ -> do i <- atomicAddCounter' ec 1 + when (i == n) (atomically (writeTChan b EOF)) + k EOF + , dupSrc) diff --git a/Z/IO/BIO/Zlib.hsc b/Z/IO/BIO/Zlib.hsc --- a/Z/IO/BIO/Zlib.hsc +++ b/Z/IO/BIO/Zlib.hsc @@ -12,7 +12,7 @@ @ -- add compressor to your BIO chain to compress streaming blocks of 'V.Bytes'. (_, zlibCompressor) <- newCompress defaultCompressConfig{compressWindowBits = 31} -runBIO $ src >|> zlibCompressor >|> sink +runBIO $ src . zlibCompressor . sink @ -} @@ -54,7 +54,6 @@ import Control.Monad import Data.IORef -import qualified Data.List as List import Data.Word import Foreign hiding (void) import Foreign.C @@ -126,9 +125,8 @@ defaultMemLevel V.empty Z_DEFAULT_STRATEGY V.defaultChunkSize -- | A foreign pointer to a zlib\'s @z_stream_s@ struct. -data ZStream = ZStream - {-# UNPACK #-} !(CPtr ZStream) - {-# UNPACK #-} !(IORef Bool) +newtype ZStream = ZStream (CPtr ZStream) deriving (Eq, Ord, Show) + deriving newtype Print -- | Make a new compress node. -- @@ -137,10 +135,10 @@ => CompressConfig -> IO (ZStream, BIO V.Bytes V.Bytes) newCompress (CompressConfig level windowBits memLevel dict strategy bufSiz) = do - (zs, _) <- newCPtrUnsafe - (\ mba## -> do - ps <- throwOOMIfNull (create_z_stream mba##) - throwZlibIfMinus_ $ deflate_init2 ps level windowBits memLevel strategy) + zs <- newCPtr' + (do ps <- throwOOMIfNull create_z_stream + throwZlibIfMinus_ $ deflate_init2 ps level windowBits memLevel strategy + return ps) free_z_stream_deflate unless (V.null dict) . withCPtr zs $ \ ps -> do @@ -148,60 +146,51 @@ deflate_set_dictionary ps pdict off (fromIntegral $ len) buf <- A.newPinnedPrimArray bufSiz - set_avail_out zs buf bufSiz bufRef <- newIORef buf - finRef <- newIORef False - return (ZStream zs finRef, BIO (zwrite zs bufRef) (zflush finRef zs bufRef [])) - where - zwrite zs bufRef input = do - set_avail_in zs input (V.length input) - zloop zs bufRef [] + set_avail_out zs buf bufSiz - zloop zs bufRef acc = do - oavail :: CUInt <- withCPtr zs $ \ ps -> do - throwZlibIfMinus_ (deflate ps (#const Z_NO_FLUSH)) - (#peek struct z_stream_s, avail_out) ps - if oavail == 0 - then do - oarr <- A.unsafeFreezeArr =<< readIORef bufRef + let newOutBuffer = do buf' <- A.newPinnedPrimArray bufSiz - set_avail_out zs buf' bufSiz writeIORef bufRef buf' - zloop zs bufRef (V.PrimVector oarr 0 bufSiz : acc) - else do - let output = V.concat (List.reverse acc) - if V.null output then return Nothing - else return (Just output) + set_avail_out zs buf' bufSiz - zflush finRef zs bufRef acc = do - fin <- readIORef finRef - if fin - then return Nothing - else do - buf <- readIORef bufRef - (r, osiz) <- withCPtr zs $ \ ps -> do - r <- throwZlibIfMinus (deflate ps (#const Z_FINISH)) - oavail :: CUInt <- (#peek struct z_stream_s, avail_out) ps - return (r, bufSiz - fromIntegral oavail) - if (r /= (#const Z_STREAM_END) && osiz /= 0) - then do - oarr <- A.unsafeFreezeArr buf - buf' <- A.newPinnedPrimArray bufSiz - set_avail_out zs buf' bufSiz - writeIORef bufRef buf' - zflush finRef zs bufRef (V.PrimVector oarr 0 osiz : acc) - else do - oarr <- A.unsafeFreezeArr buf - let trailing = V.concat . List.reverse $ V.PrimVector oarr 0 osiz : acc - -- stream ends - writeIORef finRef True - if V.null trailing then return Nothing else return (Just trailing) + return (ZStream zs, \ k mbs -> case mbs of + Just bs -> do + set_avail_in zs bs (V.length bs) + let loop = do + oavail :: CUInt <- withCPtr zs $ \ ps -> do + throwZlibIfMinus_ (deflate ps (#const Z_NO_FLUSH)) + (#peek struct z_stream_s, avail_out) ps + when (oavail == 0) $ do + oarr <- A.unsafeFreezeArr =<< readIORef bufRef + k (Just (V.PrimVector oarr 0 bufSiz)) + newOutBuffer + loop + loop + _ -> + let loop = do + (r, osiz) <- withCPtr zs $ \ ps -> do + r <- throwZlibIfMinus (deflate ps (#const Z_FINISH)) + oavail :: CUInt <- (#peek struct z_stream_s, avail_out) ps + return (r, bufSiz - fromIntegral oavail) + if (r /= (#const Z_STREAM_END) && osiz /= 0) + then do + oarr <- A.unsafeFreezeArr =<< readIORef bufRef + k (Just (V.PrimVector oarr 0 osiz)) + newOutBuffer + loop + else do + -- stream ends + when (osiz /= 0) $ do + oarr <- A.unsafeFreezeArr =<< readIORef bufRef + k (Just (V.PrimVector oarr 0 osiz)) + k EOF + in loop) -- | Reset compressor's state so that related 'BIO' can be reused. compressReset :: ZStream -> IO () -compressReset (ZStream fp finRef) = do +compressReset (ZStream fp) = do throwZlibIfMinus_ (withCPtr fp deflateReset) - writeIORef finRef False -- | Compress some bytes. compress :: HasCallStack => CompressConfig -> V.Bytes -> V.Bytes @@ -226,82 +215,76 @@ -- The returned 'BIO' node can be reused only if you call 'decompressReset' on the 'ZStream'. newDecompress :: DecompressConfig -> IO (ZStream, BIO V.Bytes V.Bytes) newDecompress (DecompressConfig windowBits dict bufSiz) = do - (zs, _) <- newCPtrUnsafe - (\ mba## -> do - ps <- throwOOMIfNull (create_z_stream mba##) - throwZlibIfMinus_ $ inflate_init2 ps windowBits) + zs <- newCPtr' + (do ps <- throwOOMIfNull create_z_stream + throwZlibIfMinus_ $ inflate_init2 ps windowBits + return ps) free_z_stream_inflate buf <- A.newPinnedPrimArray bufSiz - set_avail_out zs buf bufSiz bufRef <- newIORef buf - finRef <- newIORef False - return (ZStream zs finRef, BIO (zwrite zs bufRef) (zflush finRef zs bufRef [])) - where - zwrite zs bufRef input = do - set_avail_in zs input (V.length input) - zloop zs bufRef [] + set_avail_out zs buf bufSiz - zloop zs bufRef acc = do - oavail :: CUInt <- withCPtr zs $ \ ps -> do - r <- throwZlibIfMinus (inflate ps (#const Z_NO_FLUSH)) - when (r == (#const Z_NEED_DICT)) $ - if V.null dict - then throwIO (ZlibException "Z_NEED_DICT" callStack) - else do - throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len -> - inflate_set_dictionary ps pdict off (fromIntegral len) - throwZlibIfMinus_ (inflate ps (#const Z_NO_FLUSH)) - (#peek struct z_stream_s, avail_out) ps - if oavail == 0 - then do - oarr <- A.unsafeFreezeArr =<< readIORef bufRef + let newOutBuffer = do buf' <- A.newPinnedPrimArray bufSiz - set_avail_out zs buf' bufSiz writeIORef bufRef buf' - zloop zs bufRef (V.PrimVector oarr 0 bufSiz : acc) - else do - let output = V.concat (List.reverse acc) - if V.null output then return Nothing - else return (Just output) + set_avail_out zs buf' bufSiz - zflush finRef zs bufRef acc = do - fin <- readIORef finRef - if fin - then return Nothing - else do - buf <- readIORef bufRef - (r, osiz) <- withCPtr zs $ \ ps -> do - r <- throwZlibIfMinus (inflate ps (#const Z_FINISH)) - r' <- if r == (#const Z_NEED_DICT) - then if V.null dict - then throwIO (ZlibException "Z_NEED_DICT" callStack) + return (ZStream zs, \ k mbs -> case mbs of + Just bs -> do + set_avail_in zs bs (V.length bs) + + let loop = do + oavail :: CUInt <- withCPtr zs $ \ ps -> do + r <- throwZlibIfMinus (inflate ps (#const Z_NO_FLUSH)) + when (r == (#const Z_NEED_DICT)) $ + if V.null dict + then throwIO (ZlibException "Z_NEED_DICT" callStack) + else do + throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len -> + inflate_set_dictionary ps pdict off (fromIntegral len) + throwZlibIfMinus_ (inflate ps (#const Z_NO_FLUSH)) + (#peek struct z_stream_s, avail_out) ps + + when (oavail == 0) $ do + oarr <- A.unsafeFreezeArr =<< readIORef bufRef + k (Just (V.PrimVector oarr 0 bufSiz)) + newOutBuffer + loop + loop + + _ -> + let loop = do + (r, osiz) <- withCPtr zs $ \ ps -> do + r <- throwZlibIfMinus (inflate ps (#const Z_FINISH)) + r' <- if r == (#const Z_NEED_DICT) + then if V.null dict + then throwIO (ZlibException "Z_NEED_DICT" callStack) + else do + throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len -> + inflate_set_dictionary ps pdict off (fromIntegral len) + throwZlibIfMinus (inflate ps (#const Z_FINISH)) + else return r + oavail :: CUInt <- (#peek struct z_stream_s, avail_out) ps + return (r', bufSiz - fromIntegral oavail) + if (r /= (#const Z_STREAM_END) && osiz /= 0) + then do + oarr <- A.unsafeFreezeArr =<< readIORef bufRef + k (Just (V.PrimVector oarr 0 osiz)) + newOutBuffer + loop else do - throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len -> - inflate_set_dictionary ps pdict off (fromIntegral len) - throwZlibIfMinus (inflate ps (#const Z_FINISH)) - else return r - oavail :: CUInt <- (#peek struct z_stream_s, avail_out) ps - return (r', bufSiz - fromIntegral oavail) - if (r /= (#const Z_STREAM_END) && osiz /= 0) - then do - oarr <- A.unsafeFreezeArr buf - buf' <- A.newPinnedPrimArray bufSiz - set_avail_out zs buf' bufSiz - writeIORef bufRef buf' - zflush finRef zs bufRef (V.PrimVector oarr 0 osiz : acc) - else do - oarr <- A.unsafeFreezeArr buf - let trailing = V.concat . List.reverse $ V.PrimVector oarr 0 osiz : acc - -- stream ends - writeIORef finRef True - if V.null trailing then return Nothing else return (Just trailing) + -- stream ends + when (osiz /= 0) $ do + oarr <- A.unsafeFreezeArr =<< readIORef bufRef + k (Just (V.PrimVector oarr 0 osiz)) + k EOF + in loop) -- | Reset decompressor's state so that related 'BIO' can be reused. decompressReset :: ZStream -> IO () -decompressReset (ZStream fp finRef) = do +decompressReset (ZStream fp) = do throwZlibIfMinus_ (withCPtr fp inflateReset) - writeIORef finRef False -- | Decompress some bytes. decompress :: HasCallStack => DecompressConfig -> V.Bytes -> V.Bytes @@ -342,7 +325,7 @@ throwZlibIfMinus_ = void . throwZlibIfMinus foreign import ccall unsafe - create_z_stream :: MBA## (Ptr ZStream) -> IO (Ptr ZStream) + create_z_stream :: IO (Ptr ZStream) foreign import ccall unsafe "hs_zlib.c &free_z_stream_inflate" free_z_stream_inflate :: FunPtr (Ptr ZStream -> IO ()) diff --git a/Z/IO/Buffered.hs b/Z/IO/Buffered.hs --- a/Z/IO/Buffered.hs +++ b/Z/IO/Buffered.hs @@ -14,7 +14,7 @@ module Z.IO.Buffered ( -- * Input & Output device - Input(..), Output(..) + Input(..), Output(..), IODev -- * Buffered Input , BufferedInput, bufInput , newBufferedInput @@ -23,7 +23,7 @@ , unReadBuffer , clearInputBuffer , readParser - , readParseChunks + , readParseChunk , readExactly , readToMagic , readLine @@ -32,10 +32,13 @@ , BufferedOutput, bufOutput , newBufferedOutput , newBufferedOutput' - , writeBuffer + , writeBuffer, writeBuffer' , writeBuilder , flushBuffer , clearOutputBuffer + -- * Buffered Input and Output + , newBufferedIO + , newBufferedIO' -- * common buffer size , V.defaultChunkSize , V.smallChunkSize @@ -74,6 +77,15 @@ class Output o where writeOutput :: o -> Ptr Word8 -> Int -> IO () +-- | Input and Output device +-- +-- 'readInput' should return 0 on EOF. +-- +-- 'writeOutput' should not return until all data are written (may not +-- necessarily flushed to hardware, that should be done in device specific way). +-- +type IODev io = (Input io, Output io) + -- | Input device with buffer, NOT THREAD SAFE! -- -- * A 'BufferedInput' should not be used in multiple threads, there's no locking mechanism to protect @@ -135,7 +147,19 @@ inputBuffer <- newIORef buf return (BufferedInput (readInput i) pb inputBuffer) +-- | Open a new buffered input and output with 'V.defaultChunkSize' as buffer size. +newBufferedIO :: IODev dev => dev -> IO (BufferedInput, BufferedOutput) +{-# INLINE newBufferedIO #-} +newBufferedIO dev = newBufferedIO' dev V.defaultChunkSize V.defaultChunkSize +-- | Open a new buffered input and output with given buffer size, e.g. 'V.defaultChunkSize'. +newBufferedIO' :: IODev dev => dev -> Int -> Int -> IO (BufferedInput, BufferedOutput) +{-# INLINE newBufferedIO' #-} +newBufferedIO' dev inSize outSize = do + i <- newBufferedInput' inSize dev + o <- newBufferedOutput' outSize dev + pure (i, o) + -- | Request bytes chunk from 'BufferedInput'. -- -- The buffering logic is quite simple: @@ -287,19 +311,22 @@ unReadBuffer pb' BufferedInput{..} = unless (V.null pb') $ do modifyIORef' bufPushBack (\ pb -> pb' `V.append` pb) --- | Read buffer and parse with 'P.ParseChunks'. +-- | Read buffer and parse with 'P.parseChunk' style function. -- -- This function will continuously draw data from input before parsing finish. Unconsumed -- bytes will be returned to buffer. -- -- Throw 'OtherError' with name @EPARSE@ if parsing failed. -readParseChunks :: (T.Print e, HasCallStack) => P.ParseChunks IO V.Bytes e a -> BufferedInput -> IO a -{-# INLINABLE readParseChunks #-} -readParseChunks cp i = do - bs <- readBuffer i - (rest, r) <- cp (readBuffer i) bs - unReadBuffer rest i - unwrap "EPARSE" r +readParseChunk :: (T.Print e, HasCallStack) => (V.Bytes -> P.Result e a) -> BufferedInput -> IO a +{-# INLINABLE readParseChunk #-} +readParseChunk pc i = loop pc + where + loop f = do + bs <- readBuffer i + case f bs of + P.Success v rest -> unReadBuffer rest i >> return v + P.Failure e rest -> unReadBuffer rest i >> throwOtherError "EPARSE" (T.toText e) + P.Partial f' -> loop f' -- | Read buffer and parse with 'P.Parser'. -- @@ -309,7 +336,7 @@ -- Throw 'OtherError' with name @EPARSE@ if parsing failed. readParser :: HasCallStack => P.Parser a -> BufferedInput -> IO a {-# INLINABLE readParser #-} -readParser = readParseChunks . P.parseChunks +readParser = readParseChunk . P.parseChunk {-| Read until reach a magic bytes, return bytes(including the magic bytes). @@ -397,6 +424,17 @@ copyPrimArray outputBuffer i ba s l -- copy to buffer writePrimIORef bufIndex l -- update index +-- | Write 'V.Bytes' into buffered handle then flush the buffer into output device (if buffer is not empty). +-- +-- * If buffer is empty and bytes are larger than half of buffer, directly write bytes, +-- otherwise copy bytes to buffer. +-- +-- * If buffer is not empty, then copy bytes to buffer if it can hold, otherwise +-- write buffer first, then try again. +-- +writeBuffer' :: HasCallStack => BufferedOutput -> V.Bytes -> IO () +{-# INLINE writeBuffer' #-} +writeBuffer' bo o = writeBuffer bo o >> flushBuffer bo -- | Directly write 'B.Builder' into buffered handle. -- diff --git a/Z/IO/Exception.hs b/Z/IO/Exception.hs --- a/Z/IO/Exception.hs +++ b/Z/IO/Exception.hs @@ -43,7 +43,7 @@ , NoSuchThing(..) , ResourceBusy(..) , ResourceExhausted(..) - , EOF(..) + , UnexpectedEOF(..) , IllegalOperation(..) , PermissionDenied(..) , UnsatisfiedConstraints(..) @@ -119,7 +119,7 @@ IOE(NoSuchThing) IOE(ResourceBusy) IOE(ResourceExhausted) -IOE(EOF) +IOE(UnexpectedEOF) IOE(IllegalOperation) IOE(PermissionDenied) IOE(UnsatisfiedConstraints) @@ -261,7 +261,7 @@ throwUVError :: CInt -> IOEInfo -> IO a {-# INLINABLE throwUVError #-} throwUVError e info = case e of - UV_EOF -> throwIO (EOF info) + UV_EOF -> throwIO (UnexpectedEOF info) UV_E2BIG -> throwIO (ResourceExhausted info) UV_EACCES -> throwIO (PermissionDenied info) UV_EADDRINUSE -> throwIO (ResourceBusy info) diff --git a/Z/IO/FileSystem/Base.hs b/Z/IO/FileSystem/Base.hs --- a/Z/IO/FileSystem/Base.hs +++ b/Z/IO/FileSystem/Base.hs @@ -132,7 +132,7 @@ import Z.IO.Resource import Z.IO.UV.FFI -#include "_Shared.hs" +#include "fs_shared.hs" -------------------------------------------------------------------------------- -- File @@ -405,8 +405,8 @@ allocaBytes uvStatSize $ \ s -> do r <- fromIntegral <$> hs_uv_fs_stat p s if | r == UV_ENOENT -> return Nothing - | r < 0 -> throwUV r - | otherwise -> Just <$> peekUVStat s + | r < 0 -> throwUV r + | otherwise -> Just <$> peekUVStat s -- | Equivalent to -- @@ -416,8 +416,8 @@ allocaBytes uvStatSize $ \ s -> do r <- fromIntegral <$> hs_uv_fs_lstat p s if | r == UV_ENOENT -> return Nothing - | r < 0 -> throwUV r - | otherwise -> Just <$> peekUVStat s + | r < 0 -> throwUV r + | otherwise -> Just <$> peekUVStat s -- | Equivalent to fstat :: HasCallStack => File -> IO FStat diff --git a/Z/IO/FileSystem/Threaded.hs b/Z/IO/FileSystem/Threaded.hs --- a/Z/IO/FileSystem/Threaded.hs +++ b/Z/IO/FileSystem/Threaded.hs @@ -145,7 +145,7 @@ import Z.IO.UV.FFI import Z.IO.UV.Manager -#include "_Shared.hs" +#include "fs_shared.hs" -------------------------------------------------------------------------------- -- File @@ -444,8 +444,8 @@ uvm <- getUVManager withUVRequest' uvm (hs_uv_fs_stat_threaded p s) $ \ r -> if | r == fromIntegral UV_ENOENT -> pure Nothing - | r < 0 -> throwUV r - | otherwise -> Just <$> peekUVStat s + | r < 0 -> throwUV r + | otherwise -> Just <$> peekUVStat s -- | Equivalent to -- @@ -457,8 +457,8 @@ uvm <- getUVManager withUVRequest' uvm (hs_uv_fs_lstat_threaded p s) $ \ r -> if | r == fromIntegral UV_ENOENT -> pure Nothing - | r < 0 -> throwUV r - | otherwise -> Just <$> peekUVStat s + | r < 0 -> throwUV r + | otherwise -> Just <$> peekUVStat s -- | Equivalent to fstat :: HasCallStack => File -> IO FStat diff --git a/Z/IO/FileSystem/Watch.hs b/Z/IO/FileSystem/Watch.hs --- a/Z/IO/FileSystem/Watch.hs +++ b/Z/IO/FileSystem/Watch.hs @@ -16,7 +16,7 @@ -- dup a file event source src <- srcf -- print event to stdout - runBIO $ src >|> sinkToIO printStd + runBIO_ $ src . sinkToIO printStd @ -} @@ -31,7 +31,9 @@ import Data.Bits import qualified Data.HashMap.Strict as HM import Data.IORef +#if defined(linux_HOST_OS) import qualified Data.List as List +#endif import Data.Primitive.PrimArray import Data.Word import GHC.Generics @@ -66,7 +68,7 @@ watchDirs dirs rec callback = do withResource (initWatchDirs dirs rec) $ \ srcf -> do src <- srcf - runBIO $ src >|> sinkToIO callback + runBIO_ $ src . sinkToIO callback -- | Start watching a list of given directories, stream version. initWatchDirs :: [CBytes] -- ^ watching list @@ -108,7 +110,7 @@ cleanUpWatcher mRef sink = do m <- takeMVar mRef forM_ m killThread - void (pull sink) + void (sink discard EOF) watchThread mRef dir sink = do -- IORef store temp events to de-duplicated @@ -205,7 +207,7 @@ case me of Just e -> (Nothing, Just e) _ -> (Nothing, Nothing) - forM_ me' (push sink) + forM_ me' (stepBIO_ sink) me' <- atomicModifyIORef' eRef $ \ me -> case me of @@ -213,4 +215,4 @@ then (me, Nothing) else (Just event, Just e) _ -> (Just event, Nothing) - forM_ me' (push sink) + forM_ me' (stepBIO_ sink) diff --git a/Z/IO/FileSystem/_Shared.hs b/Z/IO/FileSystem/_Shared.hs deleted file mode 100644 --- a/Z/IO/FileSystem/_Shared.hs +++ /dev/null @@ -1,156 +0,0 @@ --- This file should be included from both base and threaded FS module - --- | File bundled with offset. --- --- Reading or writing using 'Input' \/ 'Output' instance will automatically increase offset. --- 'FilePtr' and its operations are NOT thread safe, use 'MVar' 'FilePtr' in multiple threads. --- --- The notes on linux 'writeFileP' applied to 'FilePtr' too. -data FilePtr = FilePtr {-# UNPACK #-} !File - {-# UNPACK #-} !(PrimIORef Int64) - --- | Create a file offset bundle from an 'File'. --- -newFilePtr :: File -- ^ the file we're reading - -> Int64 -- ^ initial offset - -> IO FilePtr -newFilePtr uvf off = FilePtr uvf <$> newPrimIORef off - --- | Get current offset. -getFilePtrOffset :: FilePtr -> IO Int64 -getFilePtrOffset (FilePtr _ offsetRef) = readPrimIORef offsetRef - --- | Change current offset. -setFilePtrOffset :: FilePtr -> Int64 -> IO () -setFilePtrOffset (FilePtr _ offsetRef) = writePrimIORef offsetRef - -instance Input FilePtr where - readInput (FilePtr file offsetRef) buf bufSiz = - readPrimIORef offsetRef >>= \ off -> do - l <- readFileP file buf bufSiz off - writePrimIORef offsetRef (off + fromIntegral l) - return l - -instance Output FilePtr where - writeOutput (FilePtr file offsetRef) buf bufSiz = - readPrimIORef offsetRef >>= \ off -> do - writeFileP file buf bufSiz off - writePrimIORef offsetRef (off + fromIntegral bufSiz) - --- | Quickly open a file and read its content. -readFile :: HasCallStack => CBytes -> IO V.Bytes -readFile filename = do - withResource (initFile filename O_RDONLY DEFAULT_FILE_MODE) $ \ file -> do - readAll' =<< newBufferedInput file - --- | Quickly open a file and read its content as UTF8 text. -readTextFile :: HasCallStack => CBytes -> IO T.Text -readTextFile filename = T.validate <$> readFile filename - --- | Quickly open a file and write some content. -writeFile :: HasCallStack => CBytes -> V.Bytes -> IO () -writeFile filename content = do - withResource (initFile filename (O_WRONLY .|. O_CREAT) DEFAULT_FILE_MODE) $ \ file -> do - withPrimVectorSafe content (writeOutput file) - --- | Quickly open a file and write some content as UTF8 text. -writeTextFile :: HasCallStack => CBytes -> T.Text -> IO () -writeTextFile filename content = writeFile filename (T.getUTF8Bytes content) - --- | Quickly open a file and read its content as a JSON value. --- Throw 'OtherError' with name @EPARSE@ if JSON value is not parsed. -readJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> IO a -readJSONFile filename = unwrap "EPARSE" . JSON.decode' =<< readFile filename - --- | Quickly open a file and write a JSON Value. -writeJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> a -> IO () -writeJSONFile filename x = writeFile filename (JSON.encode x) - --------------------------------------------------------------------------------- - --- | Find all files and directories within a given directory with a predicator. --- --- @ --- import Z.IO.FileSystem.FilePath (splitExtension) --- -- find all haskell source file within current dir --- scandirRecursively "." (\\ p _ -> (== ".hs") . snd \<$\> splitExtension p) --- @ -scandirRecursively :: HasCallStack => CBytes -> (CBytes -> DirEntType -> IO Bool) -> IO [CBytes] -scandirRecursively dir p = loop [] =<< P.normalize dir - where - loop acc0 pdir = - foldM (\ acc (d,t) -> do - d' <- pdir `P.join` d - r <- p d' t - let acc' = if r then (d':acc) else acc - if (t == DirEntDir) - then loop acc' d' - else return acc' - ) acc0 =<< scandir pdir - --------------------------------------------------------------------------------- - --- | Does given path exist? --- -doesPathExist :: CBytes -> IO Bool -doesPathExist path = maybe False (const True) <$> stat' path - --- | Returns 'True' if the argument file exists and is either a file or a --- symbolic link to a file, and 'False' otherwise. -doesFileExist :: CBytes -> IO Bool -doesFileExist path = maybe False isFileSt <$> stat' path - --- | Returns 'True' if the argument directory exists and is either a directory or a --- symbolic link to a directory, and 'False' otherwise. -doesDirExist :: CBytes -> IO Bool -doesDirExist path = maybe False isDirSt <$> stat' path - --------------------------------------------------------------------------------- - --- | If given path is a symbolic link? -isLink :: HasCallStack => CBytes -> IO Bool -isLink = fmap isLinkSt . lstat - --- | If given path is a directory or a symbolic link to a directory? -isDir :: HasCallStack => CBytes -> IO Bool -isDir = fmap isDirSt . stat - --- | If given path is a file or a symbolic link to a file? -isFile :: HasCallStack => CBytes -> IO Bool -isFile = fmap isFileSt . stat - --- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFLNK@ --- --- Note you should use 'lstat' to get the link's stat. -isLinkSt :: FStat -> Bool -isLinkSt st = stMode st .&. S_IFMT == S_IFLNK - --- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFDIR@ -isDirSt :: FStat -> Bool -isDirSt st = stMode st .&. S_IFMT == S_IFDIR - --- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFREG@ -isFileSt :: FStat -> Bool -isFileSt st = stMode st .&. S_IFMT == S_IFREG - -------------------------------------------------------------------------------- - --- | Make a temporary file under system 'Env.getTempDir' and automatically clean after used. --- --- >>> withResource (initTempFile "foo") $ printStd --- File 13 --- -initTempFile :: CBytes -> Resource File -initTempFile prefix = - initResource initAction unlink >>= (\f -> initFile f O_RDWR DEFAULT_FILE_MODE) - where - initAction = Env.getTempDir >>= (`P.join` prefix) >>= mkstemp - --- | Make a temporary directory under system 'Env.getTempDir' and automatically clean after used. --- --- >>> withResource (initTempDir "foo") $ printStd --- "/tmp/fooxfWR0L" --- -initTempDir :: CBytes -> Resource CBytes -initTempDir prefix = - initResource (Env.getTempDir >>= (`P.join` prefix) >>= mkdtemp) rmrf diff --git a/Z/IO/Network/DNS.hsc b/Z/IO/Network/DNS.hsc --- a/Z/IO/Network/DNS.hsc +++ b/Z/IO/Network/DNS.hsc @@ -28,7 +28,7 @@ import Foreign.C.Types import Foreign.Marshal.Utils import Foreign.Ptr -import Foreign.Storable +import Foreign.Storable import GHC.Generics import Z.Data.CBytes as CBytes import Z.Data.Text.Print (Print(..)) @@ -321,7 +321,7 @@ getNameInfo flags doHost doService addr = withUVInitDo $ do (host, (service, _)) <- allocCBytes (fromIntegral h_len) $ \ ptr_h -> allocCBytes (fromIntegral s_len) $ \ ptr_s -> - withSocketAddr addr $ \ ptr_addr -> + withSocketAddr addr $ \ ptr_addr -> throwUVIfMinus_ $ hs_getnameinfo ptr_addr addr_len ptr_h h_len ptr_s s_len cflag return (host, service) where @@ -357,7 +357,7 @@ ----------------------------------------------------------------------------- foreign import ccall safe "hs_getaddrinfo" - hs_getaddrinfo :: Ptr Word8 -- ^ host + hs_getaddrinfo :: Ptr Word8 -- ^ host -> Ptr Word8 -- ^ service -> Ptr AddrInfo -- ^ hints -> Ptr (Ptr AddrInfo) -- ^ output addrinfo linked list @@ -368,7 +368,7 @@ foreign import ccall safe "hs_getnameinfo" hs_getnameinfo :: Ptr SocketAddr -> CSize - -> CString -- ^ output host + -> CString -- ^ output host -> CSize -> CString -- ^ output service -> CSize diff --git a/Z/IO/Network/SocketAddr.hsc b/Z/IO/Network/SocketAddr.hsc --- a/Z/IO/Network/SocketAddr.hsc +++ b/Z/IO/Network/SocketAddr.hsc @@ -44,6 +44,12 @@ -- * port numbber , PortNumber(..) , portAny + , defaultPortNumberHTTP + , defaultPortNumberHTTPS + , defaultPortNumberSMTP + , defaultPortNumberPOP3 + , defaultPortNumberIMAP + , defaultPortNumberIRC -- * family, type, protocol -- ** SocketFamily , SocketFamily @@ -90,7 +96,7 @@ import Z.IO.Exception import Z.Foreign -#include "hs_uv.h" +#include "hs_uv.h" #if defined(i386_HOST_ARCH) && defined(mingw32_HOST_OS) #let CALLCONV = "stdcall" @@ -109,7 +115,7 @@ #endif -- | IPv4 or IPv6 socket address, i.e. the `sockaddr_in` or `sockaddr_in6` struct. --- +-- -- Example on JSON instance: -- -- @ @@ -118,7 +124,7 @@ -- > JSON.encodeText $ ipv4 "128.14.32.1" 9090 -- "{\"addr\":[128,14,32,1],\"port\":9090}" -- @ -data SocketAddr +data SocketAddr = SocketAddrIPv4 {-# UNPACK #-} !IPv4 -- sin_addr (ditto) {-# UNPACK #-} !PortNumber -- sin_port (network byte order) @@ -129,7 +135,7 @@ {-# UNPACK #-} !ScopeID -- sin6_scope_id (ditto) deriving (Eq, Ord, Generic) -instance JSON SocketAddr where +instance JSON SocketAddr where {-# INLINE encodeJSON #-} encodeJSON (SocketAddrIPv4 addr port) = T.curly $ do "addr" `B.kv` encodeJSON addr @@ -145,11 +151,11 @@ "scope" `B.kv` encodeJSON scope {-# INLINE toValue #-} - toValue (SocketAddrIPv4 addr port) = JSON.Object . V.pack $ + toValue (SocketAddrIPv4 addr port) = JSON.Object . V.pack $ [ ("addr", toValue addr) , ("number", toValue port) ] - toValue (SocketAddrIPv6 addr port flow scope) = JSON.Object . V.pack $ + toValue (SocketAddrIPv6 addr port flow scope) = JSON.Object . V.pack $ [ ("addr", toValue addr) , ("number", toValue port) , ("flow", toValue flow) @@ -189,7 +195,7 @@ = T.toUTF8Builder addr >> T.char7 ':' >> T.toUTF8Builder port toUTF8BuilderP _ (SocketAddrIPv6 addr port _ _) = do T.square (T.toUTF8Builder addr) - T.char7 ':' + T.char7 ':' T.toUTF8Builder port sockAddrFamily :: SocketAddr -> SocketFamily @@ -222,7 +228,7 @@ -- JSON instance encode ipv4 address into an array with 4 'Word8' octets. newtype IPv4 = IPv4 { getIPv4Addr :: Word32 } deriving (Eq, Ord, Generic) - + instance JSON IPv4 where {-# INLINE encodeJSON #-} encodeJSON = encodeJSON . ipv4AddrToTuple @@ -235,13 +241,13 @@ instance Print IPv4 where toUTF8BuilderP _ ia = do let (a,b,c,d) = ipv4AddrToTuple ia - T.int a - T.char7 '.' - T.int b - T.char7 '.' + T.int a + T.char7 '.' + T.int b + T.char7 '.' T.int c - T.char7 '.' - T.int d + T.char7 '.' + T.int d -- | @0.0.0.0@ ipv4Any :: IPv4 @@ -273,7 +279,7 @@ instance Storable IPv4 where sizeOf _ = 4 - alignment _ = alignment (undefined :: Word32) + alignment _ = alignment (undefined :: Word32) peek p = (IPv4 . ntohl) `fmap` peekByteOff p 0 poke p (IPv4 ia) = pokeByteOff p 0 (htonl ia) @@ -282,7 +288,7 @@ pokeMBA p off x = pokeMBA p off (htonl (getIPv4Addr x)) peekMBA p off = IPv4 . ntohl <$> peekMBA p off indexBA p off = IPv4 (ntohl (indexBA p off)) - + -- | Converts 'IPv4' to representation-independent IPv4 quadruple. -- For example for @127.0.0.1@ the function will return @(127, 0, 0, 1)@ -- regardless of host endianness. @@ -307,7 +313,7 @@ data IPv6 = IPv6 {-# UNPACK #-}!Word32 {-# UNPACK #-}!Word32 {-# UNPACK #-}!Word32 - {-# UNPACK #-}!Word32 + {-# UNPACK #-}!Word32 deriving (Eq, Ord, Generic) instance JSON IPv6 where @@ -415,7 +421,7 @@ instance Unaligned IPv6 where unalignedSize = (#size struct in6_addr) - indexBA p off = + indexBA p off = let a = indexBA p (off + s6_addr_offset + 0) b = indexBA p (off + s6_addr_offset + 4) c = indexBA p (off + s6_addr_offset + 8) @@ -489,8 +495,8 @@ (#size struct sockaddr_in) (#alignment struct sockaddr_in) $ \ p -> pokeSocketAddr p sa >> f p withSocketAddr sa@(SocketAddrIPv6 _ _ _ _) f = do - allocaBytesAligned - (#size struct sockaddr_in6) + allocaBytesAligned + (#size struct sockaddr_in6) (#alignment struct sockaddr_in6) $ \ p -> pokeSocketAddr p sa >> f p -- | Pass 'SocketAddr' to FFI as pointer. @@ -499,12 +505,12 @@ -- withSocketAddrUnsafe :: SocketAddr -> (MBA## SocketAddr -> IO a) -> IO a withSocketAddrUnsafe sa@(SocketAddrIPv4 _ _) f = do - (MutableByteArray p) <- newByteArray (#size struct sockaddr_in) - pokeSocketAddrMBA p sa + (MutableByteArray p) <- newByteArray (#size struct sockaddr_in) + pokeSocketAddrMBA p sa f p withSocketAddrUnsafe sa@(SocketAddrIPv6 _ _ _ _) f = do - (MutableByteArray p) <- newByteArray (#size struct sockaddr_in6) - pokeSocketAddrMBA p sa + (MutableByteArray p) <- newByteArray (#size struct sockaddr_in6) + pokeSocketAddrMBA p sa f p sizeOfSocketAddr :: SocketAddr -> CSize @@ -524,7 +530,7 @@ -- withSocketAddrStorageUnsafe :: (MBA## SocketAddr -> IO ()) -> IO SocketAddr withSocketAddrStorageUnsafe f = do - (MutableByteArray p) <- newByteArray (#size struct sockaddr_storage) + (MutableByteArray p) <- newByteArray (#size struct sockaddr_storage) f p peekSocketAddrMBA p @@ -536,14 +542,14 @@ family <- peekMBA p (#offset struct sockaddr, sa_family) case family :: CSaFamily of (#const AF_INET) -> do - addr <- peekMBA p (#offset struct sockaddr_in, sin_addr) - port <- peekMBA p (#offset struct sockaddr_in, sin_port) + addr <- peekMBA p (#offset struct sockaddr_in, sin_addr) + port <- peekMBA p (#offset struct sockaddr_in, sin_port) return (SocketAddrIPv4 addr port) (#const AF_INET6) -> do - port <- peekMBA p (#offset struct sockaddr_in6, sin6_port) - flow <- peekMBA p (#offset struct sockaddr_in6, sin6_flowinfo) - addr <- peekMBA p (#offset struct sockaddr_in6, sin6_addr) - scope <- peekMBA p (#offset struct sockaddr_in6, sin6_scope_id) + port <- peekMBA p (#offset struct sockaddr_in6, sin6_port) + flow <- peekMBA p (#offset struct sockaddr_in6, sin6_flowinfo) + addr <- peekMBA p (#offset struct sockaddr_in6, sin6_addr) + scope <- peekMBA p (#offset struct sockaddr_in6, sin6_scope_id) return (SocketAddrIPv6 addr port flow scope) _ -> do let errno = UV_EAI_ADDRFAMILY name <- uvErrName errno @@ -578,7 +584,7 @@ -- Port Numbers -- | Port number. --- +-- -- Use the @Num@ instance (i.e. use a literal) to create a -- @PortNumber@ value. -- -- >>> 1 :: PortNumber @@ -593,7 +599,7 @@ -- True -- >>> 50000 + (10000 :: PortNumber) -- 60000 -newtype PortNumber = PortNumber Word16 +newtype PortNumber = PortNumber Word16 deriving (Eq, Ord, Enum, Generic) deriving newtype (Show, Print, Read, Num, Bounded, Real, Integral, JSON) @@ -601,6 +607,24 @@ portAny :: PortNumber portAny = PortNumber 0 +defaultPortNumberHTTP :: PortNumber +defaultPortNumberHTTP = 80 + +defaultPortNumberHTTPS :: PortNumber +defaultPortNumberHTTPS = 443 + +defaultPortNumberSMTP :: PortNumber +defaultPortNumberSMTP = 25 + +defaultPortNumberPOP3 :: PortNumber +defaultPortNumberPOP3 = 110 + +defaultPortNumberIMAP :: PortNumber +defaultPortNumberIMAP = 143 + +defaultPortNumberIRC :: PortNumber +defaultPortNumberIRC = 194 + instance Storable PortNumber where sizeOf _ = sizeOf (0 :: Word16) alignment _ = alignment (0 :: Word16) @@ -612,7 +636,7 @@ indexBA p off = PortNumber . ntohs $ indexBA p off pokeMBA p off (PortNumber po) = pokeMBA p off (htons po) peekMBA p off = PortNumber . ntohs <$> peekMBA p off - + -------------------------------------------------------------------------------- type SocketFamily = CInt diff --git a/Z/IO/Resource.hs b/Z/IO/Resource.hs --- a/Z/IO/Resource.hs +++ b/Z/IO/Resource.hs @@ -129,8 +129,8 @@ -- | Create a new resource and run some computation, resource is guarantee to -- be closed. -- --- Be care don't leak the resource through computation return value, because --- after the computation finishes, the resource is closed already. +-- Be careful, don't leak the resource through the computation return value +-- because after the computation finishes, the resource is already closed. -- withResource :: (MonadCatch.MonadMask m, MonadIO m, HasCallStack) => Resource a -> (a -> m b) -> m b diff --git a/cbits/hs_zlib.c b/cbits/hs_zlib.c --- a/cbits/hs_zlib.c +++ b/cbits/hs_zlib.c @@ -32,9 +32,8 @@ #include #include -z_stream* create_z_stream(z_stream** streamp){ +z_stream* create_z_stream(){ z_stream* stream = malloc(sizeof(z_stream)); - *streamp = stream; if (stream) { stream->zalloc = Z_NULL; stream->zfree = Z_NULL; diff --git a/include/fs_shared.hs b/include/fs_shared.hs new file mode 100644 --- /dev/null +++ b/include/fs_shared.hs @@ -0,0 +1,156 @@ +-- This file should be included from both base and threaded FS module + +-- | File bundled with offset. +-- +-- Reading or writing using 'Input' \/ 'Output' instance will automatically increase offset. +-- 'FilePtr' and its operations are NOT thread safe, use 'MVar' 'FilePtr' in multiple threads. +-- +-- The notes on linux 'writeFileP' applied to 'FilePtr' too. +data FilePtr = FilePtr {-# UNPACK #-} !File + {-# UNPACK #-} !(PrimIORef Int64) + +-- | Create a file offset bundle from an 'File'. +-- +newFilePtr :: File -- ^ the file we're reading + -> Int64 -- ^ initial offset + -> IO FilePtr +newFilePtr uvf off = FilePtr uvf <$> newPrimIORef off + +-- | Get current offset. +getFilePtrOffset :: FilePtr -> IO Int64 +getFilePtrOffset (FilePtr _ offsetRef) = readPrimIORef offsetRef + +-- | Change current offset. +setFilePtrOffset :: FilePtr -> Int64 -> IO () +setFilePtrOffset (FilePtr _ offsetRef) = writePrimIORef offsetRef + +instance Input FilePtr where + readInput (FilePtr file offsetRef) buf bufSiz = + readPrimIORef offsetRef >>= \ off -> do + l <- readFileP file buf bufSiz off + writePrimIORef offsetRef (off + fromIntegral l) + return l + +instance Output FilePtr where + writeOutput (FilePtr file offsetRef) buf bufSiz = + readPrimIORef offsetRef >>= \ off -> do + writeFileP file buf bufSiz off + writePrimIORef offsetRef (off + fromIntegral bufSiz) + +-- | Quickly open a file and read its content. +readFile :: HasCallStack => CBytes -> IO V.Bytes +readFile filename = do + withResource (initFile filename O_RDONLY DEFAULT_FILE_MODE) $ \ file -> do + readAll' =<< newBufferedInput file + +-- | Quickly open a file and read its content as UTF8 text. +readTextFile :: HasCallStack => CBytes -> IO T.Text +readTextFile filename = T.validate <$> readFile filename + +-- | Quickly open a file and write some content. +writeFile :: HasCallStack => CBytes -> V.Bytes -> IO () +writeFile filename content = do + withResource (initFile filename (O_WRONLY .|. O_CREAT) DEFAULT_FILE_MODE) $ \ file -> do + withPrimVectorSafe content (writeOutput file) + +-- | Quickly open a file and write some content as UTF8 text. +writeTextFile :: HasCallStack => CBytes -> T.Text -> IO () +writeTextFile filename content = writeFile filename (T.getUTF8Bytes content) + +-- | Quickly open a file and read its content as a JSON value. +-- Throw 'OtherError' with name @EPARSE@ if JSON value is not parsed. +readJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> IO a +readJSONFile filename = unwrap "EPARSE" . JSON.decode' =<< readFile filename + +-- | Quickly open a file and write a JSON Value. +writeJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> a -> IO () +writeJSONFile filename x = writeFile filename (JSON.encode x) + +-------------------------------------------------------------------------------- + +-- | Find all files and directories within a given directory with a predicator. +-- +-- @ +-- import Z.IO.FileSystem.FilePath (splitExtension) +-- -- find all haskell source file within current dir +-- scandirRecursively "." (\\ p _ -> (== ".hs") . snd \<$\> splitExtension p) +-- @ +scandirRecursively :: HasCallStack => CBytes -> (CBytes -> DirEntType -> IO Bool) -> IO [CBytes] +scandirRecursively dir p = loop [] =<< P.normalize dir + where + loop acc0 pdir = + foldM (\ acc (d,t) -> do + d' <- pdir `P.join` d + r <- p d' t + let acc' = if r then (d':acc) else acc + if (t == DirEntDir) + then loop acc' d' + else return acc' + ) acc0 =<< scandir pdir + +-------------------------------------------------------------------------------- + +-- | Does given path exist? +-- +doesPathExist :: CBytes -> IO Bool +doesPathExist path = maybe False (const True) <$> stat' path + +-- | Returns 'True' if the argument file exists and is either a file or a +-- symbolic link to a file, and 'False' otherwise. +doesFileExist :: CBytes -> IO Bool +doesFileExist path = maybe False isFileSt <$> stat' path + +-- | Returns 'True' if the argument directory exists and is either a directory or a +-- symbolic link to a directory, and 'False' otherwise. +doesDirExist :: CBytes -> IO Bool +doesDirExist path = maybe False isDirSt <$> stat' path + +-------------------------------------------------------------------------------- + +-- | If given path is a symbolic link? +isLink :: HasCallStack => CBytes -> IO Bool +isLink = fmap isLinkSt . lstat + +-- | If given path is a directory or a symbolic link to a directory? +isDir :: HasCallStack => CBytes -> IO Bool +isDir = fmap isDirSt . stat + +-- | If given path is a file or a symbolic link to a file? +isFile :: HasCallStack => CBytes -> IO Bool +isFile = fmap isFileSt . stat + +-- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFLNK@ +-- +-- Note you should use 'lstat' to get the link's stat. +isLinkSt :: FStat -> Bool +isLinkSt st = stMode st .&. S_IFMT == S_IFLNK + +-- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFDIR@ +isDirSt :: FStat -> Bool +isDirSt st = stMode st .&. S_IFMT == S_IFDIR + +-- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFREG@ +isFileSt :: FStat -> Bool +isFileSt st = stMode st .&. S_IFMT == S_IFREG + +------------------------------------------------------------------------------- + +-- | Make a temporary file under system 'Env.getTempDir' and automatically clean after used. +-- +-- >>> withResource (initTempFile "foo") $ printStd +-- File 13 +-- +initTempFile :: CBytes -> Resource File +initTempFile prefix = + initResource initAction unlink >>= (\f -> initFile f O_RDWR DEFAULT_FILE_MODE) + where + initAction = Env.getTempDir >>= (`P.join` prefix) >>= mkstemp + +-- | Make a temporary directory under system 'Env.getTempDir' and automatically clean after used. +-- +-- >>> withResource (initTempDir "foo") $ printStd +-- "/tmp/fooxfWR0L" +-- +initTempDir :: CBytes -> Resource CBytes +initTempDir prefix = + initResource (Env.getTempDir >>= (`P.join` prefix) >>= mkdtemp) rmrf diff --git a/test/Z/IO/BIO/ConcurrentSpec.hs b/test/Z/IO/BIO/ConcurrentSpec.hs --- a/test/Z/IO/BIO/ConcurrentSpec.hs +++ b/test/Z/IO/BIO/ConcurrentSpec.hs @@ -1,15 +1,18 @@ {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} module Z.IO.BIO.ConcurrentSpec where import Control.Concurrent import Control.Monad +import Control.Monad.IO.Class import Data.IORef import Z.IO.BIO.Concurrent import Z.IO.BIO import Test.QuickCheck import Test.QuickCheck.Function import Test.QuickCheck.Property +import Test.QuickCheck.Monadic as QM import Test.Hspec import Test.Hspec.QuickCheck import Test.HUnit @@ -19,6 +22,10 @@ spec :: Spec spec = describe "BIO.Concurrent" $ do + prop "zipBIO works like zip on Sources" $ \ xs ys -> monadicIO $ do + (zs :: [(Int, Int)]) <- liftIO . runBIO $ zipBIO (sourceFromList xs) (sourceFromList ys) + QM.assert (zs == zip xs ys) + it "TQueueNode works as expected" $ do let content = [1..1000] @@ -27,12 +34,12 @@ let producter = do src' <- sourceListWithDelay content - runBIO (src' >|> sink) + runBIO_ (src' . sink) let consumer = do (rRef, sink') <- sinkToList - runBIO (src >|> sink') - r <- readIORef rRef + runBIO_ (src . sink') + r <- takeMVar rRef atomicModifyIORef' sumRef $ \ x -> (x + sum r, ()) forkIO $ consumer @@ -54,12 +61,12 @@ let producter = do src' <- sourceListWithDelay content - runBIO (src' >|> sink) + runBIO_ (src' . sink) let consumer = do (rRef, sink') <- sinkToList - runBIO (src >|> sink') - r <- readIORef rRef + runBIO_ (src . sink') + r <- takeMVar rRef atomicModifyIORef' sumRef $ \ x -> (x + sum r, ()) forkIO $ consumer @@ -81,13 +88,13 @@ let producter = do src' <- sourceListWithDelay content - runBIO (src' >|> sink) + runBIO_ (src' . sink) let consumer = do (rRef, sink') <- sinkToList src <- srcf - runBIO (src >|> sink') - r <- readIORef rRef + runBIO_ (src . sink') + r <- takeMVar rRef atomicModifyIORef' sumRef $ \ x -> (x + sum r, ()) forkIO $ consumer @@ -103,15 +110,9 @@ sourceListWithDelay :: [Int] -> IO (Source Int) -sourceListWithDelay xs0 = do - xsRef <- newIORef xs0 - return BIO{ pull = popper xsRef } - where - popper xsRef = do - xs <- readIORef xsRef - case xs of - (x:xs') -> do - writeIORef xsRef xs' - threadDelay x - return (Just x) - _ -> return Nothing +sourceListWithDelay xs = do + return $ \ k _ -> do + forM_ xs $ \ x -> do + threadDelay x + k (Just x) + k Nothing diff --git a/test/Z/IO/BIO/ZlibSpec.hs b/test/Z/IO/BIO/ZlibSpec.hs --- a/test/Z/IO/BIO/ZlibSpec.hs +++ b/test/Z/IO/BIO/ZlibSpec.hs @@ -48,22 +48,22 @@ B.concat vs @=? BL.toStrict vs' - prop "compress >|> decompress" $ \ xss -> do + prop "compress . decompress" $ \ xss -> do (_, c) <- newCompress defaultCompressConfig (_, d) <- newDecompress defaultDecompressConfig let vs = Prelude.map V.pack xss - vs' <- runBlocks (c >|> d) vs + vs' <- runBlocks (c . d) vs V.concat vs @=? V.concat vs' - prop "compress >|> decompress (with dict)" $ \ xss -> do + prop "compress . decompress (with dict)" $ \ xss -> do let dict = "aabbccdd" (_, c) <- newCompress defaultCompressConfig{compressDictionary = dict} (_, d) <- newDecompress defaultDecompressConfig{decompressDictionary = dict} let vs = Prelude.map V.pack xss - vs' <- runBlocks (c >|> d) vs + vs' <- runBlocks (c . d) vs V.concat vs @=? V.concat vs' diff --git a/test/Z/IO/BIOSpec.hs b/test/Z/IO/BIOSpec.hs --- a/test/Z/IO/BIOSpec.hs +++ b/test/Z/IO/BIOSpec.hs @@ -2,6 +2,7 @@ module Z.IO.BIOSpec where +import Control.Concurrent import Control.Monad import qualified Codec.Compression.Zlib as TheZlib import Data.IORef @@ -23,22 +24,22 @@ describe "decode . encode === id(Base64)" $ prop "Base64" $ \ xs -> let r = unsafePerformIO $ do - src <- sourceFromList xs + let src = sourceFromList xs (rRef, sink) <- sinkToList enc <- newBase64Encoder dec <- newBase64Decoder - runBIO $ src >|> enc >|> dec >|> sink - readIORef rRef + runBIO $ src . enc . dec . sink + takeMVar rRef in V.concat r === V.concat xs describe "decode . encode === id(Hex)" $ do prop "Hex" $ \ xs upper -> let r = unsafePerformIO $ do - src <- sourceFromList xs + let src = sourceFromList xs (rRef, sink) <- sinkToList let enc = hexEncoder upper dec <- newHexDecoder - runBIO $ src >|> enc >|> dec >|> sink - readIORef rRef + runBIO $ src . enc . dec . sink + takeMVar rRef in V.concat r === V.concat xs