Z-IO 0.1.7.0 → 0.1.8.0

raw patch · 36 files changed

+2360/−865 lines, 36 filesdep +unordered-containersdep ~Z-DataPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: unordered-containers

Dependency ranges changed: Z-Data

API changes (from Hackage documentation)

Files

• ChangeLog.md +8/−0
• Z-IO.cabal +17/−6
• Z/Compression/Zlib.hsc +0/−345
• Z/IO.hs +3/−1
• Z/IO/BIO.hs +765/−0
• Z/IO/BIO/Concurrent.hs +111/−0
• Z/IO/BIO/Zlib.hsc +370/−0
• Z/IO/Buffered.hs +147/−204
• Z/IO/Environment.hs +0/−1
• Z/IO/Exception.hs +2/−6
• Z/IO/FileSystem.hs +92/−11
• Z/IO/FileSystem/FilePath.hsc +12/−10
• Z/IO/FileSystem/Threaded.hs +136/−51
• Z/IO/FileSystem/Watch.hs +205/−0
• Z/IO/Logger.hs +7/−7
• Z/IO/LowResTimer.hs +9/−9
• Z/IO/Network/IPC.hs +6/−6
• Z/IO/Network/SocketAddr.hsc +0/−1
• Z/IO/Network/TCP.hs +6/−6
• Z/IO/Network/UDP.hs +11/−18
• Z/IO/Process.hsc +1/−1
• Z/IO/StdStream.hs +17/−9
• Z/IO/UV/FFI.hsc +69/−39
• Z/IO/UV/Manager.hs +18/−5
• Z/IO/UV/UVStream.hs +4/−2
• cbits/hs_uv_base.c +20/−0
• cbits/hs_uv_fs_event.c +72/−0
• cbits/hs_uv_stream.c +0/−17
• cbits/hs_uv_udp.c +0/−17
• include/hs_uv.h +12/−6
• test/Z/Compression/ZlibSpec.hs +0/−77
• test/Z/IO/BIO/ConcurrentSpec.hs +117/−0
• test/Z/IO/BIO/ZlibSpec.hs +69/−0
• test/Z/IO/BIOSpec.hs +44/−0
• test/Z/IO/FileSystem/ThreadedSpec.hs +9/−9
• third_party/libuv/src/unix/linux-inotify.c +1/−1

@@ -1,5 +1,13 @@ # Revision history for Z-IO +## 0.1.8.0  -- 2020-10-24++* Remove type index from `BufferedInput`, `BufferedOutput`.+* Add `Z.IO.BIO` module to facilitate streaming process, and `Z.IO.BIO.Concurrent` to facilitate producer-consumer model.+* Remove streamming related functions from `Z.IO.Buffered`, use `Z.IO.BIO` instead.+* Move `Z.IO.Compression.Zlib` to `Z.IO.BIO.Zlib`, change API to `BIO` style.+* Add `Z.IO.FileSystem.Watch` module, provides cross-platform filesystem watching.+ ## 0.1.7.0  -- 2020-10-24  * Add `iso8016DateFormat`, change logger's default time format to include time zone.

@@ -1,6 +1,6 @@ cabal-version:              2.4 name:                       Z-IO-version:                    0.1.7.0+version:                    0.1.8.0 synopsis:                   Simple and high performance IO toolkit for Haskell description:                Simple and high performance IO toolkit for Haskell, including                             file system, network, ipc and more!@@ -85,9 +85,15 @@  library     exposed-modules:        Z.IO+                            Z.IO.BIO+                            Z.IO.BIO.Zlib+                            Z.IO.BIO.Concurrent+                             Z.IO.FileSystem                             Z.IO.FileSystem.FilePath                             Z.IO.FileSystem.Threaded+                            Z.IO.FileSystem.Watch+                             Z.IO.Network.SocketAddr                             Z.IO.Network.DNS                             Z.IO.Network.TCP@@ -112,18 +118,20 @@                              Z.IO.Environment -                            Z.Compression.Zlib+                            -- Z.Compression.Zlib      build-depends:          base                    >= 4.12 && < 5.0-                          , Z-Data                  >= 0.1.7.2 && < 0.1.9-                          , primitive               >= 0.7.1 && < 0.7.2                           , exceptions              == 0.10.*+                          , primitive               >= 0.7.1 && < 0.7.2+                          , stm                     == 2.5.*                           , time                    >= 1.8 && < 2.0                           , unix-time               >= 0.4.7 && < 0.5-                          , stm                     == 2.5.*+                          , unordered-containers    == 0.2.*+                          , Z-Data                  >= 0.1.9.1 && < 0.2.0      default-language:       Haskell2010     default-extensions:     BangPatterns+                            BinaryLiterals                             CApiFFI                             ConstraintKinds                             CPP @@ -180,6 +188,7 @@     c-sources:              cbits/hs_uv_base.c                             cbits/hs_uv_dns.c                             cbits/hs_uv_file.c+                            cbits/hs_uv_fs_event.c                             cbits/hs_uv_stream.c                             cbits/hs_uv_process.c                             cbits/hs_uv_udp.c@@ -360,7 +369,9 @@                           , zlib                           , bytestring -    other-modules:          Z.Compression.ZlibSpec+    other-modules:          Z.IO.BIOSpec+                            Z.IO.BIO.ZlibSpec+                            Z.IO.BIO.ConcurrentSpec                             Z.IO.FileSystem.ThreadedSpec                             Z.IO.FileSystemSpec                             Z.IO.Network.UDPSpec

@@ -1,345 +0,0 @@-{-|-Module      : Z.Compression.Zlib-Description : The zlib-Copyright   : (c) Dong Han, 2017-2018-License     : BSD-Maintainer  : winterland1989@gmail.com-Stability   : experimental-Portability : non-portable-This module provides <https://zlib.net zlib> bindings.--}--module Z.Compression.Zlib(-  -- * Compression-    CompressConfig(..)-  , defaultCompressConfig-  , compress-  , compressSink-  , WindowBits-  , defaultWindowBits-  , MemLevel-  , defaultMemLevel-  -- * Decompression-  , DecompressConfig(..)-  , defaultDecompressConfig-  , decompress-  , decompressSource-  -- * Constants-  -- ** Strategy -  , Strategy -  , pattern Z_FILTERED-  , pattern Z_HUFFMAN_ONLY-  , pattern Z_RLE-  , pattern Z_FIXED-  , pattern Z_DEFAULT_STRATEGY-  -- ** CompressLevel-  , CompressLevel-  , pattern Z_BEST_SPEED-  , pattern Z_BEST_COMPRESSION-  , pattern Z_DEFAULT_COMPRESSION-  ) where--import           Control.Monad-import           Data.IORef-import           Data.Typeable-import           Data.Word-import           Foreign            hiding (void)-import           Foreign.C-import           GHC.Generics-import           System.IO.Unsafe   (unsafePerformIO)-import           Z.Data.Array       as A-import           Z.Data.CBytes      as CBytes-import           Z.Data.Vector.Base as V-import           Z.Data.Text.ShowT  (ShowT)-import           Z.Foreign-import           Z.IO.Buffered-import           Z.IO.Exception--#include "zlib.h"--type Strategy = CInt --pattern Z_FILTERED           :: Strategy-pattern Z_HUFFMAN_ONLY       :: Strategy-pattern Z_RLE                :: Strategy-pattern Z_FIXED              :: Strategy-pattern Z_DEFAULT_STRATEGY   :: Strategy-pattern Z_FILTERED           = #const Z_FILTERED        -pattern Z_HUFFMAN_ONLY       = #const Z_HUFFMAN_ONLY    -pattern Z_RLE                = #const Z_RLE             -pattern Z_FIXED              = #const Z_FIXED           -pattern Z_DEFAULT_STRATEGY   = #const Z_DEFAULT_STRATEGY--type CompressLevel = CInt ---- pattern Z_NO_COMPRESSION       =  CompressLevel (#const Z_NO_COMPRESSION     )-pattern Z_BEST_SPEED          :: CompressLevel-pattern Z_BEST_COMPRESSION    :: CompressLevel-pattern Z_DEFAULT_COMPRESSION :: CompressLevel-pattern Z_BEST_SPEED          = #const Z_BEST_SPEED         -pattern Z_BEST_COMPRESSION    = #const Z_BEST_COMPRESSION   -pattern Z_DEFAULT_COMPRESSION = #const Z_DEFAULT_COMPRESSION--{- | The 'WindowBits' is the base two logarithm of the maximum window size (the size of the history buffer).-It should be in the range 8..15 for this version of the library. The 'defaultWindowBits' value is 15. Decompressing windowBits must be greater than or equal to the compressing windowBits. If a compressed stream with a larger window size is given as input, decompress will throw 'ZDataError'-windowBits can also be –8..–15 for raw inflate. In this case, -windowBits determines the window size. inflate() will then process raw deflate data, not looking for a zlib or gzip header, not generating a check value, and not looking for any check values for comparison at the end of the stream.-windowBits can also be greater than 15 for optional gzip decoding. Add 32 to windowBits to enable zlib and gzip decoding with automatic header detection, or add 16 to decode only the gzip format.--}-type WindowBits = CInt--defaultWindowBits :: WindowBits-defaultWindowBits = 15---- | The 'MemLevel' specifies how much memory should be allocated for the internal compression state. 1 uses minimum memory but is slow and reduces compression ratio; 9 uses maximum memory for optimal speed. The default value is 8.-type MemLevel = CInt--defaultMemLevel :: MemLevel-defaultMemLevel = 9--data CompressConfig = CompressConfig-    { compressLevel :: CompressLevel-    , compressWindowBits :: WindowBits-    , compressMemoryLevel :: MemLevel-    , compressDictionary :: V.Bytes-    , compressStrategy :: Strategy-    }   deriving (Show, Eq, Ord, Generic)-        deriving anyclass ShowT--defaultCompressConfig :: CompressConfig-defaultCompressConfig =-    CompressConfig Z_DEFAULT_COMPRESSION  defaultWindowBits-        defaultMemLevel V.empty Z_DEFAULT_STRATEGY---- | Compress all the data written to a output.----compressSink :: HasCallStack-           => CompressConfig-           -> Sink V.Bytes-           -> IO (Sink V.Bytes)-compressSink (CompressConfig level windowBits memLevel dict strategy) (write, flush) = do-    zs <- newForeignPtr free_z_stream_deflate =<< create_z_stream-    buf <- A.newPinnedPrimArray bufSiz-    set_avail_out zs buf bufSiz-    bufRef <- newIORef buf--    withForeignPtr zs $ \ ps -> do-        throwZlibIfMinus_ $ deflate_init2 ps level windowBits memLevel strategy-        unless (V.null dict) $-            throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len ->-            deflate_set_dictionary ps pdict off (fromIntegral $ len)--    return (zwrite zs bufRef, zflush zs bufRef)--  where-    bufSiz = V.defaultChunkSize--    zwrite zs bufRef input = do-        set_avail_in zs input (V.length input)-        zloop zs bufRef--    zloop zs bufRef = do-        oavail :: CUInt <- withForeignPtr 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-            buf' <- A.newPinnedPrimArray bufSiz-            set_avail_out zs buf' bufSiz-            writeIORef bufRef buf'-            write (V.PrimVector oarr 0 bufSiz)-            zloop zs bufRef--    zflush zs bufRef = do-        r :: CInt <- withForeignPtr zs $ \ ps -> do-            r <- throwZlibIfMinus (deflate ps (#const Z_FINISH))-            oavail :: CUInt <- (#peek struct z_stream_s, avail_out) ps-            when (oavail /= fromIntegral bufSiz) $ do-                oarr <- A.unsafeFreezeArr =<< readIORef bufRef-                write (V.PrimVector oarr 0 (bufSiz - fromIntegral oavail))-                flush-            return r--        when (r /= (#const Z_STREAM_END)) $ do-            buf' <- A.newPinnedPrimArray bufSiz-            set_avail_out zs buf' bufSiz-            writeIORef bufRef buf'-            zflush zs bufRef---- | Compress some bytes.-compress :: HasCallStack => CompressConfig -> V.Bytes -> V.Bytes-compress conf input = unsafePerformIO $ do-    ref <- newIORef []-    (write, flush) <- compressSink conf (\ x -> modifyIORef' ref (x:), return ())-    write input-    flush-    V.concat . reverse <$> readIORef ref---{--compressBuilderStream :: HasCallStack-                      => CompressConfig-                      -> (B.Builder a -> IO ())-                      -> IO (B.Builder a -> IO ())----}--data DecompressConfig = DecompressConfig-    { decompressWindowBits :: WindowBits-    , decompressDictionary :: V.Bytes-    }   deriving (Show, Eq, Ord, Generic)-        deriving anyclass ShowT--defaultDecompressConfig :: DecompressConfig-defaultDecompressConfig = DecompressConfig defaultWindowBits V.empty---- | Decompress bytes from source.-decompressSource :: DecompressConfig-                 -> Source V.Bytes-                 -> IO (Source V.Bytes)-decompressSource (DecompressConfig windowBits dict) source = do -    zs <- newForeignPtr free_z_stream_inflate =<< create_z_stream-    buf <- A.newPinnedPrimArray bufSiz-    set_avail_out zs buf bufSiz-    bufRef <- newIORef buf--    withForeignPtr zs $ \ ps -> do-        throwZlibIfMinus_ $ inflate_init2 ps windowBits--    return (zread zs bufRef)-  where-    bufSiz = V.defaultChunkSize-    -    zread zs bufRef = do -        bufLen <- A.sizeofMutableArr =<< readIORef bufRef -        if bufLen == 0-        then return Nothing-        else do-            oavail :: CUInt <- withForeignPtr zs (#peek struct z_stream_s, avail_out)-            if (oavail == 0) -            then do-                oarr <- A.unsafeFreezeArr =<< readIORef bufRef-                buf' <- A.newPinnedPrimArray bufSiz-                set_avail_out zs buf' bufSiz-                writeIORef bufRef buf'-                return (Just (V.PrimVector oarr 0 bufSiz))-            else zloop zs bufRef --    zloop zs bufRef  = do-        iavail :: CUInt <- withForeignPtr zs (#peek struct z_stream_s, avail_in) -        if iavail == 0-        then do-            input <- source-            case input of-                Just input' -> do-                    set_avail_in zs input' (V.length input')-                    withForeignPtr zs $ \ ps -> do-                        r <- throwZlibIfMinus (inflate ps (#const Z_NO_FLUSH))-                        when (r == (#const Z_NEED_DICT) && not (V.null dict)) $ do-                            throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len ->-                                inflate_set_dictionary ps pdict off (fromIntegral len)-                    zread zs bufRef -                _ -> zfinish zs bufRef []-        else do-            withForeignPtr zs $ \ ps ->-                throwZlibIfMinus_ (inflate ps (#const Z_NO_FLUSH))-            zloop zs bufRef --    zfinish zs bufRef acc = do-        r <- withForeignPtr zs $ \ ps -> do-            throwZlibIfMinus (inflate ps (#const Z_FINISH))--        oavail :: CUInt <- withForeignPtr zs (#peek struct z_stream_s, avail_out) -        oarr <- A.unsafeFreezeArr =<< readIORef bufRef-        let !v = V.PrimVector oarr 0 (bufSiz - fromIntegral oavail)--        if (r == (#const Z_STREAM_END)) -        then do-            writeIORef bufRef =<< A.newArr 0-            let !v' = V.concat (reverse (v:acc))-            return (Just v')-        else do-            buf' <- A.newPinnedPrimArray bufSiz-            set_avail_out zs buf' bufSiz-            writeIORef bufRef buf'-            zfinish zs bufRef (v:acc)--    --- | Decompress some bytes.-decompress :: HasCallStack => DecompressConfig -> V.Bytes -> V.Bytes-decompress conf input = V.concat . unsafePerformIO $ do-     collectSource =<< decompressSource conf =<< sourceFromList [input]------------------------------------------------------------------------------------toZErrorMsg :: CInt -> CBytes-toZErrorMsg (#const Z_OK           ) =  "Z_OK"-toZErrorMsg (#const Z_STREAM_END   ) =  "Z_STREAM_END"-toZErrorMsg (#const Z_NEED_DICT    ) =  "Z_NEED_DICT"-toZErrorMsg (#const Z_ERRNO        ) =  "Z_ERRNO"-toZErrorMsg (#const Z_STREAM_ERROR ) =  "Z_STREAM_ERROR"-toZErrorMsg (#const Z_DATA_ERROR   ) =  "Z_DATA_ERROR"-toZErrorMsg (#const Z_MEM_ERROR    ) =  "Z_MEM_ERROR"-toZErrorMsg (#const Z_BUF_ERROR    ) =  "Z_BUF_ERROR"-toZErrorMsg (#const Z_VERSION_ERROR) =  "Z_VERSION_ERROR"-toZErrorMsg _                        =  "Z_UNEXPECTED"---- | Zlib exceptions, a sub exception type to 'SomeIOException'.-data ZlibException = ZlibException CBytes CallStack deriving (Show, Typeable)-instance Exception ZlibException where-    toException = ioExceptionToException-    fromException = ioExceptionFromException--throwZlibIfMinus :: HasCallStack => IO CInt -> IO CInt-throwZlibIfMinus f = do-    r <- f-    if r < 0 && r /= (#const Z_BUF_ERROR)-    then throwIO (ZlibException (toZErrorMsg r) callStack)-    else return r--throwZlibIfMinus_ :: HasCallStack => IO CInt -> IO ()-throwZlibIfMinus_ = void . throwZlibIfMinus--data ZStream--foreign import ccall unsafe-    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 ())--foreign import ccall unsafe "hs_zlib.c &free_z_stream_deflate"-    free_z_stream_deflate :: FunPtr (Ptr ZStream -> IO ())--foreign import ccall unsafe-    deflate_init2 :: Ptr ZStream -> CompressLevel -> WindowBits -> MemLevel -> Strategy -> IO CInt--foreign import ccall unsafe-    deflate_set_dictionary :: Ptr ZStream -> BA## Word8 -> Int -> Int -> IO CInt--foreign import ccall unsafe-    deflate :: Ptr ZStream -> CInt -> IO CInt--foreign import ccall unsafe-    inflate_init2 :: Ptr ZStream -> WindowBits -> IO CInt--foreign import ccall unsafe-    inflate_set_dictionary :: Ptr ZStream -> BA## Word8 -> Int -> Int -> IO CInt--foreign import ccall unsafe-    inflate :: Ptr ZStream -> CInt -> IO CInt--set_avail_in :: ForeignPtr ZStream -> V.Bytes -> Int -> IO ()-set_avail_in zs buf buflen = do-    withPrimVectorSafe buf $ \ pbuf _ ->-        withForeignPtr zs $ \ ps -> do-            (#poke struct z_stream_s, next_in) ps pbuf-            (#poke struct z_stream_s, avail_in) ps (fromIntegral buflen :: CUInt)--set_avail_out :: ForeignPtr ZStream -> MutablePrimArray RealWorld Word8 -> Int -> IO ()-set_avail_out zs buf bufSiz = do-    withMutablePrimArrayContents buf $ \ pbuf ->-        withForeignPtr zs $ \ ps -> do-            (#poke struct z_stream_s, next_out) ps pbuf-            (#poke struct z_stream_s, avail_out) ps (fromIntegral bufSiz :: CUInt)

@@ -13,7 +13,8 @@ -}  module Z.IO-  ( module Z.IO.Buffered+  ( module Z.IO.BIO+  , module Z.IO.Buffered   , module Z.IO.Environment   , module Z.IO.Exception   , module Z.IO.Logger@@ -21,6 +22,7 @@   , module Z.IO.StdStream   ) where +import Z.IO.BIO import Z.IO.Buffered import Z.IO.Environment import Z.IO.Exception

@@ -0,0 +1,765 @@+{-# OPTIONS_GHC -Wno-missing-fields #-}+{-|+Module      : Z.IO.BIO+Description : Buffered IO interface+Copyright   : (c) Dong Han, 2017-2018+License     : BSD+Maintainer  : winterland1989@gmail.com+Stability   : experimental+Portability : non-portable++This module provides 'BIO' (block IO) type to facilitate writing streaming programs. A 'BIO' node usually:++  * Process input in unit of block(or item).+  * Running in constant spaces, which means the memory usage won't accumulate.+  * Keep some state in IO, which is sealed in 'BIO' closure.++Some examples of such nodes are:++  * Compressor \/ decompressor, e.g. zlib, etc.+  * Codec, e.g. utf8 codec, base64 codec.+  * Ciphers.+  * Packet parsers.++We use @BIO inp out@ type to represent all the objects above, @BIO Void out@ to represent an 'IO' source,+and @BIO inp Void@ to represent an 'IO' sink, which can all be connected with '>|>' to build a larger 'BIO' node.++@+import Z.Data.CBytes    (CBytes)+import Z.IO+import Z.IO.BIO+import Z.IO.BIO.Zlib++base64AndCompressFile :: HasCallStack => CBytes -> CBytes -> IO ()+base64AndCompressFile origin target = do+    base64Enc <- newBase64Encoder+    (_, zlibCompressor) <- newCompress defaultCompressConfig{compressWindowBits = 31}++    withResource (sourceFromFile origin) $ \ src ->+        withResource (sinkToFile target) $ \ sink ->+            runBIO $ src >|> base64Enc >|> zlibCompressor >|> sink++> base64AndCompressFile "test" "test.gz"+-- run 'zcat "test.gz" | base64 -d' will give you original file+@++-}+module Z.IO.BIO (+  -- * The BIO type+    BIO(..), Source, Sink+  , (>|>), (>~>), (>!>), appendSource+  , concatSource, zipSource+  , joinSink, fuseSink+  , ParseException(..)+  -- * Run BIO chain+  , runBIO+  , runSource, runSource_+  , runBlock, runBlock_, unsafeRunBlock+  , runBlocks, runBlocks_, unsafeRunBlocks+  -- * Make new BIO+  , sourceFromList+  , sinkToList+  , sourceFromBuffered+  , sourceTextFromBuffered+  , sinkToBuffered+  , sinkBuilderToBuffered+  -- * Input & Output BIO adapters+  , sourceFromInput+  , sourceFromFile+  , sourceTextFromInput+  , sourceParsedInput+  , sinkToOutput+  , sinkToFile+  , sinkBuilderToOutput+  , sinkToIO+  -- * Bytes specific+  , newParserNode, newReChunk, newUTF8Decoder, newMagicSplitter, newLineSplitter+  , newBase64Encoder, newBase64Decoder+  , hexEncoder, newHexDecoder+  -- * Generic BIO+  , newCounterNode+  , newSeqNumNode+  , newGroupingNode+  ) where++import           Control.Monad+import           Control.Monad.IO.Class+import           Data.IORef+import qualified Data.List             as List+import           Data.Void+import           Data.Word+import           Data.Bits             ((.|.))+import           System.IO.Unsafe      (unsafePerformIO)+import qualified Z.Data.Array          as A+import qualified Z.Data.Builder        as B+import qualified Z.Data.Parser         as P+import           Z.Data.CBytes         (CBytes)+import           Z.Data.PrimRef+import qualified Z.Data.Text           as T+import qualified Z.Data.Text.UTF8Codec as T+import qualified Z.Data.Vector         as V+import qualified Z.Data.Vector.Base    as V+import           Z.Data.Vector.Base64+import           Z.Data.Vector.Hex+import           Z.IO.Buffered+import           Z.IO.Exception+import qualified Z.IO.FileSystem       as FS+import           Z.IO.Resource++-- | 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+-- 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.+--  * @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.+--+-- You can connect these 'BIO' nodes with '>|>', which connect left node's output to right node's input,+-- and return a new 'BIO' node with left node's input type and right node's output type.+--+-- 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.+--+-- Note 'BIO' usually contains some IO states, you can consider it as an opaque 'IORef':+--+--   * You shouldn't use a 'BIO' node across multiple 'BIO' chain.+--   * You shouldn't use a 'BIO' node across multiple threads.+--+-- Note 'BIO' is just 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 :: HasCallStack => 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 :: HasCallStack => 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 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++-- | 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.+(>|>) :: 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.+(>!>) :: 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++-- | BIO node from a pure function.+pureBIO :: (a -> b) -> BIO a b+pureBIO f = BIO (\ x -> return (Just (f x))) (return Nothing)++-- | Connect two 'BIO' source, after first reach EOF, draw element from second.+appendSource :: Source a -> Source a  -> IO (Source a)+{-# INLINE appendSource #-}+b1 `appendSource` b2 = concatSource [b1, b2]++-- | 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 :: Sink out -> Sink out -> Sink out+{-# INLINE joinSink #-}+b1 `joinSink` b2 = fuseSink [b1, b2]++-- | 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 :: [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++-- | Connect list of 'BIO' sources, after one reach EOF, draw element from next.+concatSource :: [Source a] -> IO (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 :: Source a -> Source b -> Source (a,b)+{-# INLINABLE zipSource #-}+zipSource (BIO _ pullA) (BIO _ pullB) = BIO { pull = do+    mA <- pullA+    mB <- pullB+    return ((,) <$> mA <*> mB)}++-- | Run a 'BIO' loop (source >|> ... >|> sink).+runBIO :: BIO Void Void -> IO ()+{-# INLINABLE runBIO #-}+runBIO BIO{..} = pull >> return ()++-- | Drain a 'BIO' source into a List in memory.+runSource :: 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)++-- | Drain a source without collecting result.+runSource_ :: 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 block of input, then run BIO node until EOF.+--+-- Note many 'BIO' node will be closed or not be able to take new input after drained.+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)+++-- | Supply a single block of input, then run BIO node until EOF with 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 ()++-- | Wrap a stream computation into a pure interface.+--+-- You can wrap a stateful BIO computation(including the creation of 'BIO' node),+-- when you can guarantee a computation is pure, e.g. compressing, decoding, etc.+unsafeRunBlock :: HasCallStack => IO (BIO inp out) -> inp -> [out]+{-# INLINABLE unsafeRunBlock #-}+unsafeRunBlock new inp = unsafePerformIO (new >>= \ bio -> runBlock bio inp)++-- | Supply blocks of input, then run BIO node until EOF.+--+-- 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)++-- | Supply blocks of input, then run BIO node until EOF with collecting result.+--+-- 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 ()++-- | Wrap a stream computation into a pure interface.+--+-- Similar to 'unsafeRunBlock', but with a list of input blocks.+unsafeRunBlocks :: HasCallStack => IO (BIO inp out) -> [inp] -> [out]+{-# INLINABLE unsafeRunBlocks #-}+unsafeRunBlocks new inps = unsafePerformIO (new >>= \ bio -> runBlocks bio inps)++--------------------------------------------------------------------------------+-- | Turn a 'BufferedOutput' into a 'V.Bytes' sink.+--+sinkToBuffered :: BufferedOutput -> Sink V.Bytes+{-# INLINABLE sinkToBuffered #-}+sinkToBuffered bo = BIO push_ pull_+  where+    push_ inp = writeBuffer bo inp >> pure Nothing+    pull_ = flushBuffer bo >> pure Nothing++-- | Turn a 'BufferedOutput' into a 'B.Builder' sink.+--+sinkBuilderToBuffered :: 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++-- | Turn an 'Output' into 'V,Bytes' sink.+--+-- 'push' will write input to buffer, and 'pull'_ will flush buffer.+sinkToOutput :: HasCallStack => Output o => o -> IO (Sink V.Bytes)+{-# INLINABLE sinkToOutput #-}+sinkToOutput o =+    newBufferedOutput o >>= \ bo -> return (BIO (push_ bo) (pull_ bo))+  where+    push_ bo inp = writeBuffer bo inp >> pure Nothing+    pull_ bo = flushBuffer bo >> pure Nothing++-- | Turn a file into a 'V.Bytes' sink.+--+-- Note the file will be opened in @'FS.O_APPEND' .|. 'FS.O_CREAT' .|. 'FS.O_WRONLY'@ mode,+-- bytes will be written after the end of the original file if there'are old bytes.+sinkToFile :: HasCallStack => CBytes -> Resource (Sink V.Bytes)+{-# INLINABLE sinkToFile #-}+sinkToFile p = do+    f <- FS.initFile p (FS.O_APPEND .|. FS.O_CREAT .|. FS.O_WRONLY) FS.DEFAULT_MODE+    liftIO (sinkToOutput f)++-- | Turn an 'Output' into 'B.Builder' sink.+--+-- 'push' will write input to buffer, and 'pull'_ will flush buffer.+sinkBuilderToOutput :: Output o => o -> IO (Sink (B.Builder ()))+{-# INLINABLE sinkBuilderToOutput #-}+sinkBuilderToOutput o =+    newBufferedOutput o >>= \ bo -> return (BIO (push_ bo) (pull_ bo))+  where+    push_ bo inp = writeBuilder bo inp >> pure Nothing+    pull_ bo = flushBuffer bo >> pure Nothing++-- | Turn an 'Output' into 'BIO' sink.+--+-- 'push' will write input to buffer then perform flush, tend to degrade performance.+sinkToIO :: (a -> IO ()) -> Sink a+{-# INLINABLE sinkToIO #-}+sinkToIO f = BIO push_ pull_+  where+    push_ x = f x >> pure Nothing+    pull_ = pure Nothing++-- | Source a list 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++-- | Sink to a list in memory.+--+-- 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.+--+sinkToList :: IO (IORef [a], Sink a)+sinkToList = do+    xsRef <- newIORef []+    return (xsRef, BIO (\ x -> modifyIORef xsRef (x:) >> return Nothing)+                       (modifyIORef xsRef reverse >> return Nothing))++-- | Turn a 'BufferedInput' into 'BIO' source, map EOF to Nothing.+--+sourceFromBuffered :: 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)}++-- | Turn a UTF8 encoded 'BufferedInput' into 'BIO' source, map EOF to Nothing.+--+sourceTextFromBuffered :: 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)}++-- | Turn an input device into a 'V.Bytes' source.+sourceFromInput :: Input i => i -> IO (Source V.Bytes)+{-# INLINABLE sourceFromInput #-}+sourceFromInput i =+    newBufferedInput i >>= return . sourceFromBuffered++-- | Turn a file into a 'V.Bytes' source.+sourceFromFile :: CBytes -> Resource (Source V.Bytes)+{-# INLINABLE sourceFromFile #-}+sourceFromFile p = do+    f <- FS.initFile p FS.O_RDONLY FS.DEFAULT_MODE+    liftIO (sourceFromInput f)++-- | Turn an input device into a 'T.Text' source.+sourceTextFromInput :: Input i => i -> IO (Source T.Text)+{-# INLINABLE sourceTextFromInput #-}+sourceTextFromInput i =+    newBufferedInput i >>= return . sourceTextFromBuffered++-- | Turn input device into a packet source, see 'sourceParsedInput''.+sourceParsedInput :: Input i => i -> P.Parser a -> IO (Source a)+{-# INLINABLE sourceParsedInput #-}+sourceParsedInput i p =+    newBufferedInput i >>= return . loopParse+  where+    loopParse bi = BIO{ pull = do+        bs <- readBuffer bi+        if V.null bs+        then return Nothing+        else do+            (rest, r) <- P.parseChunks p (readBuffer bi) bs+            unReadBuffer rest bi+            case r of Right v -> return (Just v)+                      Left e  -> throwIO (ParseException e callStack)}++--------------------------------------------------------------------------------++-- | Make a chunk size divider.+--+-- A divider size divide each chunk's size to the nearest multiplier to granularity,+-- last trailing chunk is directly returned.+newReChunk :: Int                -- ^ chunk granularity+           -> IO (BIO V.Bytes V.Bytes)+{-# 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)+++-- | Exception when parsing failed in streams.+--+-- Note this exception is a sub-type of 'SomeIOException'.+data ParseException = ParseException P.ParseError CallStack deriving Show+instance Exception ParseException where+    toException = ioExceptionToException+    fromException = ioExceptionFromException++-- | Read buffer and parse with 'Parser'.+--+-- This function will continuously draw data from input before parsing finish. Unconsumed+-- bytes will be returned to buffer.+--+-- Return 'Nothing' if reach EOF before parsing, throw 'ParseException' if parsing fail.+newParserNode :: 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 _ ->+                throwIO (ParseException e callStack)+            P.Partial f' -> do+                writeIORef resultRef (Right f')+                return Nothing++    pull_ resultRef = do+        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 _ ->+                throwIO (ParseException e callStack)+            P.Partial _ ->+                throwIO (ParseException ["last chunk partial parse"] callStack)++-- | Make a new UTF8 decoder, which decode bytes streams into text streams.+--+--  Note this node is supposed to be used with preprocess node such as compressor, decoder, etc. where bytes+--  boundary cannot be controlled, UTF8 decoder will concat trailing bytes from last block to next one.+--  Use this node directly with 'sourceFromBuffered' \/ 'sourceFromInput' will not be as efficient as directly use+--  'sourceTextFromBuffered' \/ 'sourceTextFromInput', because 'BufferedInput' provides push back capability,+--  trailing bytes can be pushde back to reading buffer and returned with next block input together.+--+newUTF8Decoder :: IO (BIO V.Bytes T.Text)+{-# 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 throwIO (T.InvalidUTF8Exception callStack)+            else do+                if T.decodeCharLen arr (s + i) > l - i+                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++    pull_ trailingRef = do+        trailing <- readIORef trailingRef+        if V.null trailing+        then return Nothing+        else throwIO (T.InvalidUTF8Exception callStack)++-- | Make a new stream splitter based on magic byte.+--+newMagicSplitter :: Word8 -> IO (BIO V.Bytes V.Bytes)+{-# 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)+            Nothing -> 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)++-- | Make a new stream splitter based on linefeed(@\r\n@ or @\n@).+--+-- The result bytes doesn't contain linefeed.+newLineSplitter :: IO (BIO V.Bytes V.Bytes)+{-# INLINABLE newLineSplitter #-}+newLineSplitter = do+    s <- newMagicSplitter 10+    return (s >~> dropLineEnd)+  where+    dropLineEnd bs@(V.PrimVector arr s l) =+        case bs `V.indexMaybe` (l-2) of+            Nothing -> V.PrimVector arr s (l-1)+            Just r | r == 13   -> V.PrimVector arr s (l-2)+                   | otherwise -> V.PrimVector arr s (l-1)++-- | Make a new base64 encoder node.+newBase64Encoder :: IO (BIO V.Bytes V.Bytes)+{-# INLINABLE newBase64Encoder #-}+newBase64Encoder = do+    re <- newReChunk 3+    return (re >~> base64Encode)++-- | Make a new base64 decoder node.+newBase64Decoder :: IO (BIO V.Bytes V.Bytes)+{-# INLINABLE newBase64Decoder #-}+newBase64Decoder = do+    re <- newReChunk 4+    return (re >~> base64Decode')++-- | Make a new hex encoder node.+hexEncoder :: Bool   -- ^ uppercase?+           -> BIO V.Bytes V.Bytes+{-# INLINABLE hexEncoder #-}+hexEncoder upper = pureBIO (hexEncode upper)++-- | Make a new hex decoder node.+newHexDecoder :: IO (BIO V.Bytes V.Bytes)+{-# INLINABLE newHexDecoder #-}+newHexDecoder = do+    re <- newReChunk 2+    return (re >~> 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))+  where+    push_ c x = do+        atomicAddCounter_ c 1+        return (Just 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))+  where+    push_ c x = do+        i <- atomicAddCounter c 1+        return (Just (i, x))++-- | Make a BIO node grouping items into fixed size arrays.+--+newGroupingNode :: Int -> IO (BIO a (A.SmallArray 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+

@@ -0,0 +1,111 @@+{-# OPTIONS_GHC -Wno-missing-fields #-}+{-# OPTIONS_GHC -Wno-incomplete-patterns #-}++{-|+Module      : Z.IO.BIO.Concurrent+Description : Base64 codec+Copyright   : (c) Dong Han, 2017-2018+License     : BSD+Maintainer  : winterland1989@gmail.com+Stability   : experimental+Portability : non-portable++This module provides some concurrent 'BIO' node, to ease the implementation of producer-consumer model.+All sources and sinks return by this module are safe to be used in multiple threads.++  * Use 'newTQueueNode' for common cases.+  * Use 'newTBQueueNode' if you have a fast producer and you don't want input get piled up in memory.+  * Use 'newBroadcastTChanNode' if you want messages get broadcasted, i.e. every message written by+    producers will be received by every consumers.++It's important to correctly set the numebr of producers, internally it keeps a counter on how many producers+reached their ends, and send EOF to all consumers when last producer ends. So it's a good idea to catch+exceptions and pull the sink(which indicate EOF) on producer side.++@+(sink, src) <- newTQueueNode 2  -- it's important to correctly set the numebr of producers++forkIO $ do+    ...+    push x sink             -- producer using push+    ...+    pull sink               -- when EOF is reached, manually pull, you may consider put it in a bracket.++forkIO $ do+    ...+    (runBIO $ ... >|> sink) -- producer using BIO+        `onException` (pull sink)+forkIO $ do+    ...+    r <- pull src           -- consumer using pull+    case r of Just r' -> ...+              _ -> ...      -- Nothing indicate all producers reached EOF++forkIO $ do+    ...+    runBIO $ src >|> ...    -- consumer using BIO+@++-}++module Z.IO.BIO.Concurrent where++import Control.Monad+import Control.Concurrent.STM+import GHC.Natural+import Z.IO.BIO+import Z.Data.PrimRef+import Z.IO.Exception++-- | Make an unbounded queue and a pair of sink and souce connected to it.+newTQueueNode :: Int -- ^ number of producers+              -> IO (Sink a, Source a)+newTQueueNode n = do+    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+                    x <- restore $ atomically (readTQueue q)+                    case x of Just _ -> return x+                              _ -> do atomically (unGetTQueue q Nothing)+                                      return Nothing})++-- | Make an bounded queue and a pair of sink and souce connected to it.+newTBQueueNode :: Int       -- ^ number of producers+               -> Natural   -- ^ queue buffer bound+               -> IO (Sink a, Source a)+newTBQueueNode n bound = do+    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+                    x <- restore $ atomically (readTBQueue q)+                    case x of Just _ -> return x+                              _      -> do atomically (unGetTBQueue q Nothing)+                                           return Nothing})++-- | Make a broadcast chan and a sink connected to it, and a function return sources to receive broadcast message.+newBroadcastTChanNode :: Int                        -- ^ number of producers+                      -> IO (Sink a, IO (Source a)) -- ^ (Sink, IO Source)+newBroadcastTChanNode n = do+    b <- newBroadcastTChanIO+    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)

@@ -0,0 +1,370 @@+{-|+Module      : Z.IO.BIO.Zlib+Description : The zlib+Copyright   : (c) Dong Han, 2017-2018+License     : BSD+Maintainer  : winterland1989@gmail.com+Stability   : experimental+Portability : non-portable++This module provides <https://zlib.net zlib> bindings using 'BIO' interface.+-}++module Z.IO.BIO.Zlib(+  -- * Compression+    CompressConfig(..)+  , defaultCompressConfig+  , newCompress, compressReset+  , compress+  , compressBlocks+  , WindowBits+  , defaultWindowBits+  , MemLevel+  , defaultMemLevel+  -- * Decompression+  , DecompressConfig(..)+  , defaultDecompressConfig+  , newDecompress, decompressReset+  , decompress+  , decompressBlocks+  -- * Constants+  -- ** Strategy+  , Strategy+  , pattern Z_FILTERED+  , pattern Z_HUFFMAN_ONLY+  , pattern Z_RLE+  , pattern Z_FIXED+  , pattern Z_DEFAULT_STRATEGY+  -- ** CompressLevel+  , CompressLevel+  , pattern Z_BEST_SPEED+  , pattern Z_BEST_COMPRESSION+  , pattern Z_DEFAULT_COMPRESSION+  ) where++import           Control.Monad+import           Data.IORef+import           Data.Typeable+import           Data.Word+import qualified Data.List          as List+import           Foreign            hiding (void)+import           Foreign.C+import           GHC.Generics+import           Z.Data.Array       as A+import           Z.Data.CBytes      as CBytes+import           Z.Data.Vector.Base as V+import           Z.Data.Text.ShowT  (ShowT)+import           Z.Foreign+import           Z.IO.BIO+import           Z.IO.Exception++#include "zlib.h"++type Strategy = CInt++pattern Z_FILTERED           :: Strategy+pattern Z_HUFFMAN_ONLY       :: Strategy+pattern Z_RLE                :: Strategy+pattern Z_FIXED              :: Strategy+pattern Z_DEFAULT_STRATEGY   :: Strategy+pattern Z_FILTERED           = #const Z_FILTERED+pattern Z_HUFFMAN_ONLY       = #const Z_HUFFMAN_ONLY+pattern Z_RLE                = #const Z_RLE+pattern Z_FIXED              = #const Z_FIXED+pattern Z_DEFAULT_STRATEGY   = #const Z_DEFAULT_STRATEGY++type CompressLevel = CInt++-- pattern Z_NO_COMPRESSION       =  CompressLevel (#const Z_NO_COMPRESSION     )+pattern Z_BEST_SPEED          :: CompressLevel+pattern Z_BEST_COMPRESSION    :: CompressLevel+pattern Z_DEFAULT_COMPRESSION :: CompressLevel+pattern Z_BEST_SPEED          = #const Z_BEST_SPEED+pattern Z_BEST_COMPRESSION    = #const Z_BEST_COMPRESSION+pattern Z_DEFAULT_COMPRESSION = #const Z_DEFAULT_COMPRESSION++{- | The 'WindowBits' is the base two logarithm of the maximum window size (the size of the history buffer).+It should be in the range 8..15 for this version of the library. The 'defaultWindowBits' value is 15. Decompressing windowBits must be greater than or equal to the compressing windowBits. If a compressed stream with a larger window size is given as input, decompress will throw 'ZDataError'+windowBits can also be –8..–15 for raw inflate. In this case, -windowBits determines the window size. inflate() will then process raw deflate data, not looking for a zlib or gzip header, not generating a check value, and not looking for any check values for comparison at the end of the stream.+windowBits can also be greater than 15 for optional gzip decoding. Add 32 to windowBits to enable zlib and gzip decoding with automatic header detection, or add 16 to decode only the gzip format.+-}+type WindowBits = CInt++defaultWindowBits :: WindowBits+defaultWindowBits = 15++-- | The 'MemLevel' specifies how much memory should be allocated for the internal compression state. 1 uses minimum memory but is slow and reduces compression ratio; 9 uses maximum memory for optimal speed. The default value is 8.+type MemLevel = CInt++defaultMemLevel :: MemLevel+defaultMemLevel = 9++data CompressConfig = CompressConfig+    { compressLevel :: CompressLevel+    , compressWindowBits :: WindowBits+    , compressMemoryLevel :: MemLevel+    , compressDictionary :: V.Bytes+    , compressStrategy :: Strategy+    , compressBufferSize :: Int+    }   deriving (Show, Eq, Ord, Generic)+        deriving anyclass ShowT++defaultCompressConfig :: CompressConfig+defaultCompressConfig =+    CompressConfig Z_DEFAULT_COMPRESSION  defaultWindowBits+        defaultMemLevel V.empty Z_DEFAULT_STRATEGY V.defaultChunkSize++data ZStream = ZStream (ForeignPtr ZStream) (IORef Bool)++-- | Compress all the data written to a output.+--+-- The returned 'BIO' node can be reused only if you call 'compressReset' on the 'ZStream'.+newCompress :: HasCallStack+            => CompressConfig+            -> IO (ZStream, BIO V.Bytes V.Bytes)+newCompress (CompressConfig level windowBits memLevel dict strategy bufSiz) = do+    zs <- newForeignPtr free_z_stream_deflate =<< create_z_stream+    buf <- A.newPinnedPrimArray bufSiz+    set_avail_out zs buf bufSiz+    bufRef <- newIORef buf++    withForeignPtr zs $ \ ps -> do+        throwZlibIfMinus_ $ deflate_init2 ps level windowBits memLevel strategy+        unless (V.null dict) $+            throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len ->+            deflate_set_dictionary ps pdict off (fromIntegral $ len)++    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 []++    zloop zs bufRef acc = do+        oavail :: CUInt <- withForeignPtr 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+            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)++    zflush finRef zs bufRef acc = do+        fin <- readIORef finRef+        if fin +        then return Nothing+        else do+            buf <- readIORef bufRef+            (r, osiz) <- withForeignPtr 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)++-- | Reset compressor's state so that related 'BIO' can be reused.+compressReset :: ZStream -> IO ()+compressReset (ZStream fp finRef) = do+    throwZlibIfMinus_ (withForeignPtr fp deflateReset)+    writeIORef finRef False++-- | Decompress some bytes.+compress :: HasCallStack => CompressConfig -> V.Bytes -> V.Bytes+compress conf = V.concat . unsafeRunBlock (snd <$> newCompress conf)++-- | Decompress some bytes list.+compressBlocks :: HasCallStack => CompressConfig -> [V.Bytes] -> [V.Bytes]+compressBlocks conf = unsafeRunBlocks (snd <$> newCompress conf)++data DecompressConfig = DecompressConfig+    { decompressWindowBits :: WindowBits+    , decompressDictionary :: V.Bytes+    , decompressBufferSize :: Int+    }   deriving (Show, Eq, Ord, Generic)+        deriving anyclass ShowT++defaultDecompressConfig :: DecompressConfig+defaultDecompressConfig = DecompressConfig defaultWindowBits V.empty V.defaultChunkSize++-- | Decompress bytes from source.+--+-- 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 <- newForeignPtr free_z_stream_inflate =<< create_z_stream+    buf <- A.newPinnedPrimArray bufSiz+    set_avail_out zs buf bufSiz+    bufRef <- newIORef buf+    withForeignPtr zs $ \ ps -> do+        throwZlibIfMinus_ $ inflate_init2 ps windowBits+    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 []++    zloop zs bufRef acc = do+        oavail :: CUInt <- withForeignPtr 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+            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)++    zflush finRef zs bufRef acc = do+        fin <- readIORef finRef+        if fin+        then return Nothing+        else do+            buf <- readIORef bufRef+            (r, osiz) <- withForeignPtr 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 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)++-- | Reset decompressor's state so that related 'BIO' can be reused.+decompressReset :: ZStream -> IO ()+decompressReset (ZStream fp finRef) = do+    throwZlibIfMinus_ (withForeignPtr fp inflateReset)+    writeIORef finRef False++-- | Decompress some bytes.+decompress :: HasCallStack => DecompressConfig -> V.Bytes -> V.Bytes+decompress conf = V.concat . unsafeRunBlock (snd <$> newDecompress conf)++-- | Decompress some bytes list.+decompressBlocks :: HasCallStack => DecompressConfig -> [V.Bytes] -> [V.Bytes]+decompressBlocks conf = unsafeRunBlocks (snd <$> newDecompress conf)++--------------------------------------------------------------------------------++toZErrorMsg :: CInt -> CBytes+toZErrorMsg (#const Z_OK           ) =  "Z_OK"+toZErrorMsg (#const Z_STREAM_END   ) =  "Z_STREAM_END"+toZErrorMsg (#const Z_NEED_DICT    ) =  "Z_NEED_DICT"+toZErrorMsg (#const Z_ERRNO        ) =  "Z_ERRNO"+toZErrorMsg (#const Z_STREAM_ERROR ) =  "Z_STREAM_ERROR"+toZErrorMsg (#const Z_DATA_ERROR   ) =  "Z_DATA_ERROR"+toZErrorMsg (#const Z_MEM_ERROR    ) =  "Z_MEM_ERROR"+toZErrorMsg (#const Z_BUF_ERROR    ) =  "Z_BUF_ERROR"+toZErrorMsg (#const Z_VERSION_ERROR) =  "Z_VERSION_ERROR"+toZErrorMsg _                        =  "Z_UNEXPECTED"++-- | Zlib exceptions, a sub exception type to 'SomeIOException'.+data ZlibException = ZlibException CBytes CallStack deriving (Show, Typeable)+instance Exception ZlibException where+    toException = ioExceptionToException+    fromException = ioExceptionFromException++throwZlibIfMinus :: HasCallStack => IO CInt -> IO CInt+throwZlibIfMinus f = do+    r <- f+    if r < 0 && r /= (#const Z_BUF_ERROR)+    then throwIO (ZlibException (toZErrorMsg r) callStack)+    else return r++throwZlibIfMinus_ :: HasCallStack => IO CInt -> IO ()+throwZlibIfMinus_ = void . throwZlibIfMinus++foreign import ccall unsafe+    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 ())++foreign import ccall unsafe "hs_zlib.c &free_z_stream_deflate"+    free_z_stream_deflate :: FunPtr (Ptr ZStream -> IO ())++foreign import ccall unsafe+    deflate_init2 :: Ptr ZStream -> CompressLevel -> WindowBits -> MemLevel -> Strategy -> IO CInt++foreign import ccall unsafe+    deflate_set_dictionary :: Ptr ZStream -> BA## Word8 -> Int -> Int -> IO CInt++foreign import ccall unsafe+    deflate :: Ptr ZStream -> CInt -> IO CInt++foreign import ccall unsafe+    deflateReset :: Ptr ZStream -> IO CInt++foreign import ccall unsafe+    inflate_init2 :: Ptr ZStream -> WindowBits -> IO CInt++foreign import ccall unsafe+    inflate_set_dictionary :: Ptr ZStream -> BA## Word8 -> Int -> Int -> IO CInt++foreign import ccall unsafe+    inflate :: Ptr ZStream -> CInt -> IO CInt++foreign import ccall unsafe+    inflateReset :: Ptr ZStream -> IO CInt++set_avail_in :: ForeignPtr ZStream -> V.Bytes -> Int -> IO ()+set_avail_in zs buf buflen = do+    withPrimVectorSafe buf $ \ pbuf _ ->+        withForeignPtr zs $ \ ps -> do+            (#poke struct z_stream_s, next_in) ps pbuf+            (#poke struct z_stream_s, avail_in) ps (fromIntegral buflen :: CUInt)++set_avail_out :: ForeignPtr ZStream -> MutablePrimArray RealWorld Word8 -> Int -> IO ()+set_avail_out zs buf bufSiz = do+    withMutablePrimArrayContents buf $ \ pbuf ->+        withForeignPtr zs $ \ ps -> do+            (#poke struct z_stream_s, next_out) ps pbuf+            (#poke struct z_stream_s, avail_out) ps (fromIntegral bufSiz :: CUInt)

@@ -7,7 +7,8 @@ Stability   : experimental Portability : non-portable -This module provides buffered IO interface and basic streaming utilities.+This module provides low level buffered IO interface, it's recommended to check higher level streaming interface+"Z.IO.BIO" first as it provides more features.  -} @@ -18,7 +19,7 @@   , BufferedInput, bufInput   , newBufferedInput   , newBufferedInput'-  , readBuffer+  , readBuffer, readBufferText   , unReadBuffer   , readParser   , readExactly,  readExactly'@@ -32,20 +33,8 @@   , writeBuffer   , writeBuilder   , flushBuffer-    -- * Stream utilities-  , Source, Sink-  , sourceBuffer-  , sinkBuffer-  , sourceFromList-  , (>+>)-  , parseSource-  , collectSource-  , concatSource-  , zipSource-  , (>>>>=)-     -- * Exceptions-  , BufferedException(..)+  , IncompleteInput(..)     -- * common buffer size   , V.defaultChunkSize   , V.smallChunkSize@@ -53,11 +42,8 @@   ) where  import           Control.Monad-import           Control.Monad.Primitive     (ioToPrim, primToIO)-import           Control.Monad.ST import           Data.IORef import           Data.Primitive.PrimArray-import           Data.Typeable import           Data.Word import           Data.Bits                 (unsafeShiftR) import           Foreign.Ptr@@ -65,6 +51,8 @@ import qualified Z.Data.Builder.Base       as B import qualified Z.Data.Parser             as P import qualified Z.Data.Vector             as V+import qualified Z.Data.Text               as T+import qualified Z.Data.Text.UTF8Codec     as T import qualified Z.Data.Vector.Base        as V import           Z.Data.PrimRef.PrimIORef import           Z.Foreign@@ -93,8 +81,8 @@ -- * A 'Input' device should only be used with a single 'BufferedInput', If multiple 'BufferedInput' s --   are opened on a same 'Input' device, the behaviour is undefined. ---data BufferedInput i = BufferedInput-    { bufInput    :: i+data BufferedInput = BufferedInput+    { bufInput    :: HasCallStack => Ptr Word8 -> Int -> IO Int     , bufPushBack :: {-# UNPACK #-} !(IORef V.Bytes)     , inputBuffer :: {-# UNPACK #-} !(IORef (MutablePrimArray RealWorld Word8))     }@@ -107,41 +95,43 @@ -- * A 'Output' device should only be used with a single 'BufferedOutput', If multiple 'BufferedOutput' s --   are opened on a same 'BufferedOutput' device, the output will be interleaved. ---data BufferedOutput o = BufferedOutput-    { bufOutput     :: o+data BufferedOutput = BufferedOutput+    { bufOutput     :: HasCallStack => Ptr Word8 -> Int -> IO ()     , bufIndex      :: {-# UNPACK #-} !Counter     , outputBuffer  :: {-# UNPACK #-} !(MutablePrimArray RealWorld Word8)     }  -- | Open a new buffered input with 'V.defaultChunkSize' as buffer size.-newBufferedInput :: i -> IO (BufferedInput i)+newBufferedInput :: Input i => i -> IO BufferedInput newBufferedInput = newBufferedInput' V.defaultChunkSize  -- | Open a new buffered output with 'V.defaultChunkSize' as buffer size.-newBufferedOutput :: o -> IO (BufferedOutput o)+newBufferedOutput :: Output o => o -> IO BufferedOutput newBufferedOutput = newBufferedOutput' V.defaultChunkSize  -- | Open a new buffered output with given buffer size, e.g. 'V.defaultChunkSize'.-newBufferedOutput' :: Int    -- ^ Output buffer size+newBufferedOutput' :: Output o+                   => Int    -- ^ Output buffer size                    -> o-                   -> IO (BufferedOutput o)+                   -> IO BufferedOutput newBufferedOutput' bufSiz o = do     index <- newPrimIORef 0     buf <- newPinnedPrimArray (max bufSiz 0)-    return (BufferedOutput o index buf)+    return (BufferedOutput (writeOutput o) index buf)  -- | Open a new buffered input with given buffer size, e.g. 'V.defaultChunkSize'.-newBufferedInput' :: Int     -- ^ Input buffer size+newBufferedInput' :: Input i+                  => Int     -- ^ Input buffer size                   -> i-                  -> IO (BufferedInput i)+                  -> IO BufferedInput newBufferedInput' bufSiz i = do     pb <- newIORef V.empty     buf <- newPinnedPrimArray (max bufSiz 0)     inputBuffer <- newIORef buf-    return (BufferedInput i pb inputBuffer)+    return (BufferedInput (readInput i) pb inputBuffer)  --- | Request bytes from 'BufferedInput'.+-- | Request bytes chunk from 'BufferedInput'. -- -- The buffering logic is quite simple: --@@ -149,15 +139,15 @@ -- If we read N bytes, and N is larger than half of the buffer size, then we freeze buffer and return, -- otherwise we copy buffer into result and reuse buffer afterward. ---readBuffer :: (Input i, HasCallStack) => BufferedInput i -> IO V.Bytes+readBuffer :: HasCallStack => BufferedInput -> IO V.Bytes readBuffer BufferedInput{..} = do     pb <- readIORef bufPushBack     if V.null pb     then do         rbuf <- readIORef inputBuffer         bufSiz <- getSizeofMutablePrimArray rbuf-        l <- readInput bufInput (mutablePrimArrayContents rbuf) bufSiz-        if l < bufSiz `quot` 2                -- read less than half size+        l <- bufInput (mutablePrimArrayContents rbuf) bufSiz+        if l < bufSiz `unsafeShiftR` 1                -- read less than half size         then do             mba <- newPrimArray l              -- copy result into new array             copyMutablePrimArray mba 0 rbuf 0 l@@ -167,17 +157,76 @@             when (bufSiz /= 0) $ do                 buf' <- newPinnedPrimArray bufSiz                 writeIORef inputBuffer buf'+            shrinkMutablePrimArray rbuf l             ba <- unsafeFreezePrimArray rbuf             return $! V.fromArr ba 0 l     else do         writeIORef bufPushBack V.empty         return pb --- | Read exactly N bytes+-- | Request UTF8 'T.Text' chunk from 'BufferedInput'. --+-- The buffer size must be larger than 4 bytes to guarantee decoding progress.+readBufferText :: HasCallStack => BufferedInput -> IO T.Text+readBufferText BufferedInput{..} = do+    pb <- readIORef bufPushBack+    rbuf <- readIORef inputBuffer+    bufSiz <- getSizeofMutablePrimArray rbuf+    if V.null pb+    then do+        l <- bufInput (mutablePrimArrayContents rbuf) bufSiz+        handleBuf l+    else do+        -- clear push back first+        writeIORef bufPushBack V.empty+        let (arr, s, delta) = V.toArr pb+        if T.decodeCharLen arr s <= delta+        -- trailing bytes still contain text+        then splitLastChar pb+        -- trailing bytes contain partial codepoint+        else do+            -- copy trailing bytes to buffer and read+            copyPrimArray rbuf 0 arr s delta+            l <- bufInput (mutablePrimArrayContents rbuf `plusPtr` delta) (bufSiz - delta)+            -- if EOF is reached, no further progress is possible+            when (l == 0) (throwIO (IncompleteInput callStack))+            handleBuf (l + delta)+  where+    handleBuf l = do+        rbuf <- readIORef inputBuffer+        bufSiz <- getSizeofMutablePrimArray rbuf+        if l < bufSiz `unsafeShiftR` 1                -- read less than half size+        then do+            mba <- newPrimArray l              -- copy result into new array+            copyMutablePrimArray mba 0 rbuf 0 l+            ba <- unsafeFreezePrimArray mba+            splitLastChar (V.PrimVector ba 0 l)+        else do                                -- freeze buf into result+            when (bufSiz /= 0) $ do+                buf' <- newPinnedPrimArray bufSiz+                writeIORef inputBuffer buf'+            shrinkMutablePrimArray rbuf l+            ba <- unsafeFreezePrimArray rbuf+            splitLastChar (V.PrimVector ba 0 l)++    splitLastChar bs@(V.toArr -> (arr, s, l))+        | l == 0 = return T.empty+        | otherwise = do+            let (i, _) = V.findR (\ w -> w >= 0b11000000 || w <= 0b01111111) bs+            if (i == -1)+            then throwIO (T.InvalidUTF8Exception callStack)+            else do+                if T.decodeCharLen arr (s + i) > l - i+                then do+                    writeIORef bufPushBack (V.fromArr arr (s+i) (l-i))+                    return (T.validate (V.fromArr arr s i))+                else return (T.validate bs)++-- | Read N bytes(may be smaller than N if EOF reached).+-- -- If EOF reached before N bytes read, trailing bytes will be returned. ---readExactly :: (HasCallStack, Input i) => Int -> BufferedInput i -> IO V.Bytes+readExactly :: HasCallStack => Int -> BufferedInput -> IO V.Bytes readExactly n0 h0 = V.concat `fmap` (go h0 n0)   where     go h n = do@@ -198,20 +247,20 @@  -- | Read exactly N bytes ----- If EOF reached before N bytes read, a 'ShortReadException' will be thrown+-- If EOF reached before N bytes read, a 'IncompleteInput' will be thrown ---readExactly' :: (HasCallStack, Input i) => Int -> BufferedInput i -> IO V.Bytes+readExactly' :: HasCallStack => Int -> BufferedInput -> IO V.Bytes readExactly' n h = do     v <- readExactly n h     if (V.length v /= n)-    then throwIO (ShortReadException callStack)+    then throwIO (IncompleteInput callStack)     else return v  -- | Read all chunks from a 'BufferedInput'. -- -- This function will loop read until meet EOF('Input' device return 'V.empty'), -- Useful for reading small file into memory.-readAll :: (HasCallStack, Input i) => BufferedInput i -> IO [V.Bytes]+readAll :: HasCallStack => BufferedInput -> IO [V.Bytes] readAll h = loop []   where     loop acc = do@@ -224,22 +273,22 @@ -- -- This function will loop read until meet EOF('Input' device return 'V.empty'), -- Useful for reading small file into memory.-readAll' :: (HasCallStack, Input i) => BufferedInput i -> IO V.Bytes+readAll' :: HasCallStack => BufferedInput -> IO V.Bytes readAll' i = V.concat <$> readAll i --- | Exceptions related to buffered IO. A sub exception type to 'SomeIOException'.-data BufferedException = ParseException P.ParseError CallStack-                       | ShortReadException CallStack deriving (Show, Typeable)--instance Exception BufferedException where+-- | Exceptions when read not enough input.+--+-- Note this exception is a sub-type of 'SomeIOException'.+data IncompleteInput = IncompleteInput CallStack deriving Show+instance Exception IncompleteInput where     toException = ioExceptionToException     fromException = ioExceptionFromException  -- | Push bytes back into buffer(if not empty). ---unReadBuffer :: (HasCallStack, Input i) => V.Bytes -> BufferedInput i -> IO ()+unReadBuffer :: HasCallStack => V.Bytes -> BufferedInput -> IO () unReadBuffer pb' BufferedInput{..} = unless (V.null pb') $ do-    modifyIORef' bufPushBack $ \ pb -> pb' `V.append` pb+    modifyIORef' bufPushBack (\ pb -> pb' `V.append` pb)  -- | Read buffer and parse with 'Parser'. --@@ -247,7 +296,7 @@ -- bytes will be returned to buffer. -- -- Either during parsing or before parsing, reach EOF will result in 'P.ParseError'.-readParser :: (HasCallStack, Input i) => P.Parser a -> BufferedInput i -> IO (Either P.ParseError a)+readParser :: HasCallStack => P.Parser a -> BufferedInput -> IO (Either P.ParseError a) readParser p i = do     bs <- readBuffer i     (rest, r) <- P.parseChunks p (readBuffer i) bs@@ -257,7 +306,7 @@ -- | Read until reach a magic bytes, return bytes(including the magic bytes) -- -- If EOF is reached before meet a magic byte, partial bytes are returned.-readToMagic :: (HasCallStack, Input i) => Word8 -> BufferedInput i -> IO V.Bytes+readToMagic :: HasCallStack => Word8 -> BufferedInput -> IO V.Bytes readToMagic magic0 h0 = V.concat `fmap` (go h0 magic0)   where     go h magic = do@@ -275,14 +324,14 @@  -- | Read until reach a magic bytes, return bytes(including the magic bytes) ----- If EOF is reached before meet a magic byte, a 'ShortReadException' will be thrown.-readToMagic' :: (HasCallStack, Input i) => Word8 -> BufferedInput i -> IO V.Bytes+-- If EOF is reached before meet a magic byte, a 'IncompleteInput' will be thrown.+readToMagic' :: HasCallStack => Word8 -> BufferedInput -> IO V.Bytes readToMagic' magic0 h0 = V.concat `fmap` (go h0 magic0)   where     go h magic = do         chunk <- readBuffer h         if V.null chunk-        then throwIO (ShortReadException callStack)+        then throwIO (IncompleteInput callStack)         else case V.elemIndex magic chunk of             Just i -> do                 let (lastChunk, rest) = V.splitAt (i+1) chunk@@ -296,7 +345,7 @@ -- -- Return bytes don't include linefeed, empty bytes indicate empty line, 'Nothing' indicate EOF. -- If EOF is reached before meet a line feed, partial line is returned.-readLine :: (HasCallStack, Input i) => BufferedInput i -> Source V.Bytes+readLine :: HasCallStack => BufferedInput -> IO (Maybe V.Bytes) readLine i = do     bs@(V.PrimVector arr s l) <- readToMagic 10 i     if l == 0@@ -309,8 +358,8 @@ -- | Read to a linefeed ('\n' or '\r\n'), return 'Bytes' before it. -- -- Return bytes don't include linefeed, empty bytes indicate empty line, 'Nothing' indicate EOF.--- If EOF reached before meet a line feed, a 'ShortReadException' will be thrown.-readLine' :: (HasCallStack, Input i) => BufferedInput i -> Source V.Bytes+-- If EOF reached before meet a line feed, a 'IncompleteInput' will be thrown.+readLine' :: HasCallStack => BufferedInput -> IO (Maybe V.Bytes) readLine' i = do     bs@(V.PrimVector arr s l) <- readToMagic' 10 i     if l == 0@@ -330,7 +379,7 @@ -- * If buffer is not empty, then copy bytes to buffer if it can hold, otherwise --   write buffer first, then try again. ---writeBuffer :: (HasCallStack, Output o) => BufferedOutput o -> V.Bytes -> IO ()+writeBuffer :: HasCallStack => BufferedOutput -> V.Bytes -> IO () writeBuffer o@BufferedOutput{..} v@(V.PrimVector ba s l) = do     i <- readPrimIORef bufIndex     bufSiz <- getSizeofMutablePrimArray outputBuffer@@ -342,13 +391,13 @@             writePrimIORef bufIndex (i+l)              -- update index         else do             -- flush the buffer first-            withMutablePrimArrayContents outputBuffer $ \ ptr -> (writeOutput bufOutput) ptr i+            withMutablePrimArrayContents outputBuffer $ \ ptr -> bufOutput ptr i             writePrimIORef bufIndex 0             -- try write to buffer again             writeBuffer o v     else-        if l > bufSiz `unsafeShiftR` 2-        then withPrimVectorSafe v (writeOutput bufOutput)+        if l > bufSiz `unsafeShiftR` 1+        then withPrimVectorSafe v bufOutput         else do             copyPrimArray outputBuffer i ba s l   -- copy to buffer             writePrimIORef bufIndex l             -- update index@@ -358,160 +407,54 @@ -- -- Run 'B.Builder' with buffer if it can hold, write to device when buffer is full. ---writeBuilder :: (HasCallStack, Output o) => BufferedOutput o -> B.Builder a -> IO ()+writeBuilder :: HasCallStack => BufferedOutput -> B.Builder a -> IO () writeBuilder BufferedOutput{..} (B.Builder b) = do     i <- readPrimIORef bufIndex     originBufSiz <- getSizeofMutablePrimArray outputBuffer-    _ <- primToIO (b (B.OneShotAction action) (lastStep originBufSiz) (B.Buffer outputBuffer i))-    return ()+    loop originBufSiz =<< b (\ _ -> return . B.Done) (B.Buffer outputBuffer i)   where-    action :: V.Bytes -> ST RealWorld ()-    action bytes = ioToPrim (withPrimVectorSafe bytes (writeOutput bufOutput))+    loop originBufSiz r = case r of+        B.Done buffer@(B.Buffer buf' i') -> do+            if sameMutablePrimArray buf' outputBuffer+            then writePrimIORef bufIndex i'+            else if i' >= originBufSiz+                then do+                    action =<< freezeBuffer buffer+                    writePrimIORef bufIndex 0+                else do+                    copyMutablePrimArray outputBuffer 0 buf' 0 i'+                    writePrimIORef bufIndex i'+        B.BufferFull buffer@(B.Buffer _ i') wantSiz k -> do+            when (i' /= 0) (action =<< freezeBuffer buffer)+            if wantSiz <= originBufSiz+            then loop originBufSiz =<< k (B.Buffer outputBuffer 0)+            else do+                tempBuf <- newPinnedPrimArray wantSiz+                loop originBufSiz =<< k (B.Buffer tempBuf 0)+        B.InsertBytes buffer@(B.Buffer _ i')  bs@(V.PrimVector arr s l) k -> do+            when (i' /= 0) (action =<< freezeBuffer buffer)+            if V.length bs < originBufSiz+            then do+                copyPrimArray outputBuffer 0 arr s l+                loop originBufSiz =<< k (B.Buffer outputBuffer l)+            else do+                action bs+                loop originBufSiz =<< k (B.Buffer outputBuffer 0) -    lastStep :: Int -> a -> B.BuildStep RealWorld-    lastStep originBufSiz _ (B.Buffer buf offset)-        | sameMutablePrimArray buf outputBuffer = ioToPrim $ do-            writePrimIORef bufIndex offset   -- record new buffer index-            return []-        | offset >= originBufSiz = ioToPrim $ do-            withMutablePrimArrayContents buf $ \ ptr -> (writeOutput bufOutput) ptr offset-            writePrimIORef bufIndex 0-            return [] -- to match 'BuildStep' return type-        | otherwise = ioToPrim $ do-            copyMutablePrimArray outputBuffer 0 buf 0 offset-            writePrimIORef bufIndex offset-            return [] -- to match 'BuildStep' return type+    action bytes = withPrimVectorSafe bytes bufOutput +    freezeBuffer (B.Buffer buf offset) = do+        -- we can't shrink buffer here, it will be reused+        -- when (offset < siz) (A.shrinkMutablePrimArray buf offset)+        !arr <- unsafeFreezePrimArray buf+        return (V.PrimVector arr 0 offset)+ -- | Flush the buffer into output device(if buffer is not empty). ---flushBuffer :: (HasCallStack, Output o) => BufferedOutput o -> IO ()+flushBuffer :: HasCallStack => BufferedOutput -> IO () flushBuffer BufferedOutput{..} = do     i <- readPrimIORef bufIndex     when (i /= 0) $ do-        withMutablePrimArrayContents outputBuffer $ \ ptr -> (writeOutput bufOutput) ptr i+        withMutablePrimArrayContents outputBuffer $ \ ptr -> bufOutput ptr i         writePrimIORef bufIndex 0 ------------------------------------------------------------------------------------- | Type alias for input stream, 'Nothing' indicate EOF.-type Source a = IO (Maybe a)---- | Type alias for output stream, contain a write & a flush function.-type Sink a = (a -> IO (), IO ())---- | Turn a 'BufferedInput' into 'Source', map EOF to Nothing.----sourceBuffer :: (HasCallStack, Input i) => BufferedInput i -> Source V.Bytes-{-# INLINABLE sourceBuffer #-}-sourceBuffer i = readBuffer i >>= \ x -> if V.null x then return Nothing-                                                     else return (Just x)---- | Turn a 'BufferedOutput' into 'Sink'.----sinkBuffer :: (HasCallStack, Output o) => BufferedOutput o -> Sink V.Bytes-{-# INLINABLE sinkBuffer #-}-sinkBuffer o = (writeBuffer o, flushBuffer o)---- | Source a list streamly.-sourceFromList :: [a] -> IO (Source a)-{-# INLINABLE sourceFromList #-}-sourceFromList xs0 = do-    xsRef <- newIORef xs0-    return (popper xsRef)-  where-    popper xsRef = do-        xs <- readIORef xsRef-        case xs of-            (x:xs') -> do-                writeIORef xsRef xs'-                return (Just x)-            _ -> return Nothing---- | Connect two streams, after first reach EOF, draw element from second.-(>+>) :: Source a -> Source a  -> IO (Source a)-{-# INLINABLE (>+>) #-}-input1 >+> input2 = concatSource [input1, input2]---- | Read all stream elements to a list.-collectSource :: Source a -> IO [a]-{-# INLINABLE collectSource #-}-collectSource input = loop []-  where-    loop acc = do-        r <- input-        case r of-            Just r' -> loop (r':acc)-            _       -> return $! reverse acc----- | Read buffer and parse with 'Parser'.------ This function will continuously draw data from input before parsing finish. Unconsumed--- bytes will be returned to buffer.------ Return 'Nothing' if reach EOF before parsing, throw 'ParseException' if parsing fail.-parseSource :: HasCallStack => P.Parser a -> Source V.Bytes -> IO (Source a)-{-# INLINABLE parseSource #-}-parseSource p source = do-    trailingRef <- newIORef V.empty-    return (go trailingRef)-  where-    go trailingRef = do-        trailing <- readIORef trailingRef-        if V.null trailing-        then do-            bs <- source-            case bs of-                Just bs' -> do-                    (rest, r) <- P.parseChunks p source' bs'-                    writeIORef trailingRef rest-                    case r of Right v -> return (Just v)-                              Left e  -> throwIO (ParseException e callStack)-                _    -> return Nothing-        else do-            (rest, r) <- P.parseChunks p source' trailing-            writeIORef trailingRef rest-            case r of Right v -> return (Just v)-                      Left e  -> throwIO (ParseException e callStack)--    source' = source >>= \ r -> case r of Just r' -> return r'-                                          _      -> return V.empty---- | Connect list of streams, after one stream reach EOF, draw element from next.-concatSource :: [Source a] -> IO (Source a)-{-# INLINABLE concatSource #-}-concatSource ss0 = newIORef ss0 >>= return . loop-  where-    loop ref = do-        ss <- readIORef ref-        case ss of-          []       -> return Nothing-          (input:rest) -> do-              chunk <- input-              case chunk of-                Just _  -> return chunk-                _       -> writeIORef ref rest >> loop ref---- | Zip two streams into one.-zipSource :: Source a -> Source b -> Source (a,b)-{-# INLINABLE zipSource #-}-zipSource inputA inputB = do-    mA <- inputA-    mB <- inputB-    case mA of Just a -> case mB of Just b -> return (Just (a, b))-                                    _ -> return Nothing-               _ -> return Nothing---- | Loop read stream and write to output, when input ends flush the output.----(>>>>=) :: Source a     -- ^ stream to write-        -> Sink a-        -> IO ()-{-# INLINABLE (>>>>=) #-}-(>>>>=) input (write, flush) = loop-  where-    loop = do-        m <- input-        case m of-            Just x' -> write x' >> loop-            _       -> flush

@@ -40,7 +40,6 @@ import Z.Data.CBytes import Z.Foreign import Z.IO.Exception-import Z.IO.UV.Errno import Z.IO.UV.Manager import Foreign.Storable import Z.IO.UV.FFI

@@ -190,14 +190,10 @@     , ioeCallStack   :: CallStack   -- ^ lightweight partial call-stack     } -instance Show IOEInfo where-    show (IOEInfo errno desc cstack) =-         "{name:" ++ T.unpack errno ++-         ", description:" ++ T.unpack desc ++-         ", callstack:" ++ prettyCallStack cstack ++ "}"+instance Show IOEInfo where show = T.toString  instance T.ShowT IOEInfo where-    toTextBuilder _ (IOEInfo errno desc cstack) = do+    toUTF8BuilderP _ (IOEInfo errno desc cstack) = do          "{name:"          T.text errno          ", description:"

@@ -18,14 +18,16 @@     -- * file offset bundle   , FilePtr, newFilePtr, getFileOffset, setFileOffset   -- * filesystem operations-  , mkdir+  , mkdir, mkdirp   , unlink   , mkdtemp-  , rmdir+  , rmdir, rmdirrf   , DirEntType(..)   , scandir+  , scandirRecursively   , FStat(..), UVTimeSpec(..)   , stat, lstat, fstat+  , isLink, isDir, isFile   , rename   , fsync, fdatasync   , ftruncate@@ -57,6 +59,11 @@   , pattern S_IXGRP   , pattern S_IRWXO   , pattern S_IROTH+  -- ** file type constant+  , pattern S_IFMT+  , pattern S_IFLNK+  , pattern S_IFDIR+  , pattern S_IFREG   -- ** FileFlag   , FileFlag   , pattern O_APPEND@@ -105,8 +112,8 @@ import           Z.Foreign import           Z.IO.Buffered import           Z.IO.Exception+import qualified Z.IO.FileSystem.FilePath       as P import           Z.IO.Resource-import           Z.IO.UV.Errno import           Z.IO.UV.FFI import           Prelude hiding (writeFile, readFile) @@ -221,12 +228,11 @@ -- -- Resource closing is thread safe, on some versions of OSX, repeatly open and close same file 'Resource' may -- result in shared memory object error, use 'O_CREAT' to avoid that.-initFile :: HasCallStack-           => CBytes-           -> FileFlag        -- ^ Opening flags, e.g. 'O_CREAT' @.|.@ 'O_RDWR'-           -> FileMode      -- ^ Sets the file mode (permission and sticky bits),-                              -- but only if the file was created, see 'DEFAULT_MODE'.-           -> Resource File+initFile :: CBytes+         -> FileFlag        -- ^ Opening flags, e.g. 'O_CREAT' @.|.@ 'O_RDWR'+         -> FileMode      -- ^ Sets the file mode (permission and sticky bits),+                            -- but only if the file was created, see 'DEFAULT_MODE'.+         -> Resource File initFile path flags mode =     initResource         (do !fd <- withCBytesUnsafe path $ \ p ->@@ -262,16 +268,41 @@  -- | Equivalent to <http://linux.die.net/man/2/mkdir mkdir(2)>. ----- Note mode is currently not implemented on Windows.+-- Note mode is currently not implemented on Windows. On unix you should set execute bit+-- if you want the directory is accessable, e.g. 0o777. mkdir :: HasCallStack => CBytes -> FileMode -> IO () mkdir path mode = throwUVIfMinus_ . withCBytesUnsafe path $ \ p ->      hs_uv_fs_mkdir p mode +-- | Equivalent to @mkdir -p@+--+-- Note mode is currently not implemented on Windows. On unix you should set execute bit+-- if you want the directory is accessable(so that child folder can be created), e.g. 0o777.+mkdirp :: HasCallStack => CBytes -> FileMode -> IO ()+mkdirp path mode = do+    r <- withCBytesUnsafe path $ \ p -> hs_uv_fs_mkdir p mode+    if fromIntegral r == UV_ENOENT+    then do+        (root, segs) <- P.splitSegments path+        case segs of+            seg:segs' -> loop segs' =<< P.join root seg+            _ -> throwUVIfMinus_ (return r)+    else throwUVIfMinus_ (return r)+  where+    loop segs p = do+        a <- access p F_OK+        case a of+            AccessOK     -> return ()+            NoExistence  -> mkdir p mode+            NoPermission -> throwUVIfMinus_ (return UV_EACCES)+        case segs of+            (nextp:ps) -> P.join p nextp >>= loop ps+            _  -> return ()+ -- | Equivalent to <http://linux.die.net/man/2/unlink unlink(2)>. unlink :: HasCallStack => CBytes -> IO () unlink path = throwUVIfMinus_ (withCBytesUnsafe path hs_uv_fs_unlink) - -- | Equivalent to <mkdtemp http://linux.die.net/man/3/mkdtemp> -- -- Creates a temporary directory in the most secure manner possible.@@ -292,9 +323,23 @@         return p'  -- | Equivalent to <http://linux.die.net/man/2/rmdir rmdir(2)>.+--+-- Note this function may inherent OS limitations such as argument must be an empty folder. rmdir :: HasCallStack => CBytes -> IO () rmdir path = throwUVIfMinus_ (withCBytesUnsafe path hs_uv_fs_rmdir) +-- | Equivalent to @rmdir -rf@+--+-- This function will try to remove folder and files contained by it.+rmdirrf :: HasCallStack => CBytes -> IO ()+rmdirrf path = do+    ds <- scandir path+    forM_ ds $ \ (d, t) -> do+        if t /= DirEntDir+        then unlink d+        else rmdirrf =<< path `P.join` d+    rmdir path+ -- | Equivalent to <http://linux.die.net/man/3/scandir scandir(3)>. -- -- Note Unlike scandir(3), this function does not return the “.” and “..” entries.@@ -314,6 +359,26 @@             !p' <- fromCString p             return (p', typ')) +-- | 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+ --------------------------------------------------------------------------------  -- | Equivalent to <http://linux.die.net/man/2/stat stat(2)>@@ -337,6 +402,18 @@         throwUVIfMinus_ (hs_uv_fs_fstat fd s)         peekUVStat s +-- | If given path is a symbolic link?+isLink :: HasCallStack => CBytes -> IO Bool+isLink p = lstat p >>= \ st -> return (stMode st .&. S_IFMT == S_IFLNK)++-- | If given path is a directory or a symbolic link to a directory?+isDir :: HasCallStack => CBytes -> IO Bool+isDir p = stat p >>= \ st -> return (stMode st .&. S_IFMT == S_IFDIR)++-- | If given path is a file or a symbolic link to a file?+isFile :: HasCallStack => CBytes -> IO Bool+isFile p = stat p >>= \ st -> return (stMode st .&. S_IFMT == S_IFREG)+ --------------------------------------------------------------------------------  -- | Equivalent to <http://linux.die.net/man/2/rename rename(2)>.@@ -479,3 +556,7 @@ -- | Equivalent to <http://linux.die.net/man/2/lchown lchown(2)>. lchown :: HasCallStack => CBytes -> UID -> GID -> IO () lchown path uid gid = throwUVIfMinus_ . withCBytesUnsafe path $ \ p -> hs_uv_fs_lchown p uid gid++--------------------------------------------------------------------------------+-- high level utilities+

@@ -251,20 +251,22 @@  -- | Split file path into (root, segments, basename) tuple. ----- Root and basename may be empty, segments are separated by 'pathSeparator' and never be empty if any.+-- Root may be empty, segments are separated by 'pathSeparator' and never be empty if any. splitSegments :: CBytes-              -> IO (CBytes, [CBytes], CBytes) -- ^ return (root, segments, basename)+              -> IO (CBytes, [CBytes]) -- ^ return (root, segments) {-# INLINABLE splitSegments #-} splitSegments p = do-    (root, rest) <- splitRoot p-    (CB seg, basename) <- splitBaseName rest+    (root, CB seg) <- splitRoot =<< normalize p     sty <- getPathStyle-    let segs = case sty of-            UnixStyle ->-                V.splitWith (== (#const SLASH)) seg-            _ ->-                V.splitWith (\ w -> w == (#const SLASH) || w == (#const BACKSLASH)) seg-    return (root, (map CB (filter (not . V.null) segs)), basename)+    let segs =+            if V.null seg+            then []+            else case sty of+                UnixStyle ->+                    V.splitWith (== (#const SLASH)) seg+                _ ->+                    V.splitWith (\ w -> w == (#const SLASH) || w == (#const BACKSLASH)) seg+    return (root, (map CB segs))  -- | Determine whether the path is absolute or not. --

@@ -7,7 +7,7 @@ Stability   : experimental Portability : non-portable -This module provides IO operations related to filesystem, operations are implemented using libuv's threadpool to achieve non-block behavior (non-block here meaning won't block other haskell threads), which should be prefered when the operations' estimated time is long enough(>1ms) or running with a non-threaded haskell runtime, such as accessing network filesystem or scan a very large directory. Otherwise you may block RTS's capability thus all the other haskell threads live on it.+This module provides filesystem API exactly same with `Z.IO.FileSystem`, operations are implemented using libuv's threadpool to achieve non-block behavior (non-block here meaning won't block other haskell threads), which would be prefered when the operations' estimated time is long(>1ms) or running with a non-threaded haskell runtime, such as accessing network filesystem or scan a very large directory. Otherwise you may block RTS's capability thus all the other haskell threads live on it.  The threadpool version operations have overheads similar to safe FFI, but provide same adventages: @@ -15,23 +15,27 @@   * The threadpool version works with non-threaded runtime, which doesn't have safe FFI available.   * The threadpool version won't relinquish current HEC (Haskell Execution Context) a.k.a. capability. +Most of the time you don't need this module though, since modern hardware(SSD) are sufficiently fast.+ -}  module Z.IO.FileSystem.Threaded   ( -- * regular file devices-    FileT, initFileT, readFileT, writeFileT, getFileTFD+    File, initFile, readFile, writeFile, getFileFD   , quickReadFile, quickReadTextFile, quickWriteFile, quickWriteTextFile     -- * file offset bundle   , FilePtrT, newFilePtrT, getFileOffset, setFileOffset   -- * filesystem operations-  , mkdir+  , mkdir, mkdirp   , unlink   , mkdtemp-  , rmdir+  , rmdir, rmdirrf   , DirEntType(..)   , scandir+  , scandirRecursively   , FStat(..), UVTimeSpec(..)   , stat, lstat, fstat+  , isLink, isDir, isFile   , rename   , fsync, fdatasync   , ftruncate@@ -63,6 +67,11 @@   , pattern S_IXGRP   , pattern S_IRWXO   , pattern S_IROTH+  -- ** file type constant+  , pattern S_IFMT+  , pattern S_IFLNK+  , pattern S_IFDIR+  , pattern S_IFREG   -- ** FileFlag   , FileFlag   , pattern O_APPEND@@ -104,76 +113,77 @@ import           Foreign.Ptr import           Foreign.Storable               (peekElemOff) import           Foreign.Marshal.Alloc          (allocaBytes)-import           Z.Data.CBytes                 as CBytes+import           Z.Data.CBytes                  as CBytes import           Z.Data.PrimRef.PrimIORef import qualified Z.Data.Text                    as T import qualified Z.Data.Vector                  as V import           Z.Foreign import           Z.IO.Buffered import           Z.IO.Exception+import qualified Z.IO.FileSystem.FilePath       as P import           Z.IO.Resource-import           Z.IO.UV.Errno import           Z.IO.UV.FFI import           Z.IO.UV.Manager+import           Prelude hiding (writeFile, readFile)  -------------------------------------------------------------------------------- -- File --- | 'FileT' and its operations are NOT thread safe, use 'MVar' 'FileT' in multiple threads.+-- | 'File' and its operations are NOT thread safe, use 'MVar' 'File' in multiple threads. -- -- Note this is a differet data type from "Z.IO.FileSystem" \'s one, the 'Input' -- and 'Output' instance use thread pool version functions. -- -- libuv implements read and write method with both implict and explict offset capable. -- Implict offset interface is provided by 'Input' \/ 'Output' instances.--- Explict offset interface is provided by 'readFileT' \/ 'writeFileT'.+-- Explict offset interface is provided by 'readFile' \/ 'writeFile'. ---data FileT =  FileT  {-# UNPACK #-} !FD      -- ^ the file+data File =  File  {-# UNPACK #-} !FD      -- ^ the file                      {-# UNPACK #-} !(IORef Bool)  -- ^ closed flag --- | Return FileT fd.-getFileTFD :: FileT -> IO FD-getFileTFD (FileT fd closedRef) = do+-- | Return File fd.+getFileFD :: File -> IO FD+getFileFD (File fd closedRef) = do     closed <- readIORef closedRef     if closed then throwECLOSED else return fd  -- | If fd is -1 (closed), throw 'ResourceVanished' ECLOSED.-checkFileTClosed :: HasCallStack => FileT -> (FD -> IO a) -> IO a-checkFileTClosed (FileT fd closedRef) f = do+checkFileClosed :: HasCallStack => File -> (FD -> IO a) -> IO a+checkFileClosed (File fd closedRef) f = do     closed <- readIORef closedRef     if closed then throwECLOSED else f fd -instance Input FileT where-    readInput f buf bufSiz = readFileT f buf bufSiz (-1)+instance Input File where+    readInput f buf bufSiz = readFile f buf bufSiz (-1)  -- | Read file with given offset -- -- Read length may be smaller than buffer size.-readFileT :: HasCallStack-          => FileT+readFile :: HasCallStack+          => File           -> Ptr Word8 -- ^ buffer           -> Int       -- ^ buffer size           -> Int64     -- ^ file offset, pass -1 to use default(system) offset           -> IO Int    -- ^ read length-readFileT uvf buf bufSiz off =-    checkFileTClosed uvf  $ \ fd -> do+readFile uvf buf bufSiz off =+    checkFileClosed uvf  $ \ fd -> do         uvm <- getUVManager         withUVRequest uvm (hs_uv_fs_read_threaded fd buf bufSiz off) -instance Output FileT where-    writeOutput f buf bufSiz = writeFileT f buf bufSiz (-1)+instance Output File where+    writeOutput f buf bufSiz = writeFile f buf bufSiz (-1)  -- | Write buffer to file -- -- This function will loop until all bytes are written.-writeFileT :: HasCallStack-           => FileT+writeFile :: HasCallStack+           => File            -> Ptr Word8 -- ^ buffer            -> Int       -- ^ buffer size            -> Int64     -- ^ file offset, pass -1 to use default(system) offset            -> IO ()-writeFileT uvf buf0 bufSiz0 off0 =-    checkFileTClosed uvf $ \ fd -> do+writeFile uvf buf0 bufSiz0 off0 =+    checkFileClosed uvf $ \ fd -> do              (if off0 == -1 then go fd buf0 bufSiz0                             else go' fd buf0 bufSiz0 off0)   where@@ -199,12 +209,12 @@ -- Reading or writing using 'Input' \/ 'Output' instance will automatically increase offset. -- 'FilePtrT' and its operations are NOT thread safe, use 'MVar' 'FilePtrT' in multiple threads. ---data FilePtrT = FilePtrT {-# UNPACK #-} !FileT+data FilePtrT = FilePtrT {-# UNPACK #-} !File                          {-# UNPACK #-} !(PrimIORef Int64)  -- |  Create a file offset bundle from an 'File'. ---newFilePtrT :: FileT      -- ^ the file we're reading+newFilePtrT :: File      -- ^ the file we're reading             -> Int64      -- ^ initial offset             -> IO FilePtrT newFilePtrT uvf off = FilePtrT uvf <$> newPrimIORef off@@ -220,14 +230,14 @@ instance Input FilePtrT where     readInput (FilePtrT file offsetRef) buf bufSiz =         readPrimIORef offsetRef >>= \ off -> do-            l <- readFileT file buf bufSiz off+            l <- readFile file buf bufSiz off             writePrimIORef offsetRef (off + fromIntegral l)             return l  instance Output FilePtrT where     writeOutput (FilePtrT file offsetRef) buf bufSiz =         readPrimIORef offsetRef >>= \ off -> do-            writeFileT file buf bufSiz off+            writeFile file buf bufSiz off             writePrimIORef offsetRef (off + fromIntegral bufSiz)  --------------------------------------------------------------------------------@@ -239,19 +249,18 @@ -- be a problem if you are using multiple readers or writers in multiple threads. -- In that case you have to stop all reading or writing thread if you don't want to -- block the resource thread.-initFileT :: HasCallStack-          => CBytes+initFile :: CBytes           -> FileFlag        -- ^ Opening flags, e.g. 'O_CREAT' @.|.@ 'O_RDWR'           -> FileMode        -- ^ Sets the file mode (permission and sticky bits),                                -- but only if the file was created, see 'DEFAULT_MODE'.-          -> Resource FileT-initFileT path flags mode =+          -> Resource File+initFile path flags mode =     initResource         (do uvm <- getUVManager             fd <- withCBytesUnsafe path $ \ p ->                 withUVRequest uvm (hs_uv_fs_open_threaded p flags mode)-            FileT (fromIntegral fd) <$> newIORef False)-        (\ (FileT fd closedRef) -> do+            File (fromIntegral fd) <$> newIORef False)+        (\ (File fd closedRef) -> do             closed <- readIORef closedRef             unless closed $ do                 throwUVIfMinus_ (hs_uv_fs_close fd)@@ -260,7 +269,7 @@ -- | Quickly open a file and read its content. quickReadFile :: HasCallStack => CBytes -> IO V.Bytes quickReadFile filename = do-    withResource (initFileT filename O_RDONLY DEFAULT_MODE) $ \ file -> do+    withResource (initFile filename O_RDONLY DEFAULT_MODE) $ \ file -> do         readAll' =<< newBufferedInput file  -- | Quickly open a file and read its content as UTF8 text.@@ -270,7 +279,7 @@ -- | Quickly open a file and write some content. quickWriteFile :: HasCallStack => CBytes -> V.Bytes -> IO () quickWriteFile filename content = do-    withResource (initFileT filename (O_WRONLY .|. O_CREAT) DEFAULT_MODE) $ \ file -> do+    withResource (initFile filename (O_WRONLY .|. O_CREAT) DEFAULT_MODE) $ \ file -> do         withPrimVectorSafe content (writeOutput file)  -- | Quickly open a file and write some content as UTF8 text.@@ -281,13 +290,43 @@  -- | Equivalent to <http://linux.die.net/man/2/mkdir mkdir(2)>. ----- Note mode is currently not implemented on Windows.+-- Note mode is currently not implemented on Windows. On unix you should set execute bit+-- if you want the directory is accessable, e.g. 0o777. mkdir :: HasCallStack => CBytes -> FileMode -> IO () mkdir path mode = do     uvm <- getUVManager     withCBytesUnsafe path $ \ p ->         withUVRequest_ uvm (hs_uv_fs_mkdir_threaded p mode) +-- | Equivalent to @mkdir -p@+--+-- Note mode is currently not implemented on Windows. On unix you should set execute bit+-- if you want the directory is accessable(so that child folder can be created), e.g. 0o777.+mkdirp :: HasCallStack => CBytes -> FileMode -> IO ()+mkdirp path mode = do+    uvm <- getUVManager+    r <- withCBytesUnsafe path $ \ p ->+        withUVRequest' uvm (hs_uv_fs_mkdir_threaded p mode)+            (\ r -> do+                when (r < 0 && fromIntegral r /= UV_ENOENT)+                    (throwUVIfMinus_ (return r))+                return r)+    when (fromIntegral r == UV_ENOENT) $ do+        (root, segs) <- P.splitSegments path+        case segs of+            seg:segs' -> loop segs' =<< P.join root seg+            _ -> throwUVIfMinus_ (return r)+  where+    loop segs p = do+        a <- access p F_OK+        case a of+            AccessOK     -> return ()+            NoExistence  -> mkdir p mode+            NoPermission -> throwUVIfMinus_ (return UV_EACCES)+        case segs of+            (nextp:ps) -> P.join p nextp >>= loop ps+            _  -> return ()+ -- | Equivalent to <http://linux.die.net/man/2/unlink unlink(2)>. unlink :: HasCallStack => CBytes -> IO () unlink path = do@@ -315,11 +354,25 @@         return p''  -- | Equivalent to <http://linux.die.net/man/2/rmdir rmdir(2)>.+--+-- Note this function may inherent OS limitations such as argument must be an empty folder. rmdir :: HasCallStack => CBytes -> IO () rmdir path = do     uvm <- getUVManager     withCBytesUnsafe path (\ p -> void . withUVRequest uvm $ hs_uv_fs_rmdir_threaded p) +-- | Equivalent to @rmdir -rf@+--+-- This function will try to remove folder and files contained by it.+rmdirrf :: HasCallStack => CBytes -> IO ()+rmdirrf path = do+    ds <- scandir path+    forM_ ds $ \ (d, t) -> do+        if t /= DirEntDir+        then unlink d+        else rmdirrf =<< path `P.join` d+    rmdir path+ --------------------------------------------------------------------------------  -- | Equivalent to <http://linux.die.net/man/3/scandir scandir(3)>.@@ -344,6 +397,26 @@             !p' <- fromCString p             return (p', typ')) +-- | 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+ --------------------------------------------------------------------------------  -- | Equivalent to <http://linux.die.net/man/2/stat stat(2)>@@ -365,13 +438,25 @@             peekUVStat s  -- | Equivalent to <http://linux.die.net/man/2/fstat fstat(2)>-fstat :: HasCallStack => FileT -> IO FStat-fstat uvf = checkFileTClosed uvf $ \ fd ->+fstat :: HasCallStack => File -> IO FStat+fstat uvf = checkFileClosed uvf $ \ fd ->      (allocaBytes uvStatSize $ \ s -> do         uvm <- getUVManager         withUVRequest_ uvm (hs_uv_fs_fstat_threaded fd s)         peekUVStat s) +-- | If given path is a symbolic link?+isLink :: HasCallStack => CBytes -> IO Bool+isLink p = lstat p >>= \ st -> return (stMode st .&. S_IFMT == S_IFLNK)++-- | If given path is a directory or a symbolic link to a directory?+isDir :: HasCallStack => CBytes -> IO Bool+isDir p = stat p >>= \ st -> return (stMode st .&. S_IFMT == S_IFDIR)++-- | If given path is a file or a symbolic link to a file?+isFile :: HasCallStack => CBytes -> IO Bool+isFile p = stat p >>= \ st -> return (stMode st .&. S_IFMT == S_IFREG)+ --------------------------------------------------------------------------------  -- | Equivalent to <http://linux.die.net/man/2/rename rename(2)>.@@ -385,20 +470,20 @@             withUVRequest_ uvm (hs_uv_fs_rename_threaded p p')  -- | Equivalent to <http://linux.die.net/man/2/fsync fsync(2)>.-fsync :: HasCallStack => FileT -> IO ()-fsync uvf = checkFileTClosed uvf $ \ fd -> do+fsync :: HasCallStack => File -> IO ()+fsync uvf = checkFileClosed uvf $ \ fd -> do     uvm <- getUVManager     withUVRequest_ uvm (hs_uv_fs_fsync_threaded fd)  -- | Equivalent to <http://linux.die.net/man/2/fdatasync fdatasync(2)>.-fdatasync :: HasCallStack => FileT -> IO ()-fdatasync uvf = checkFileTClosed uvf $ \ fd -> do+fdatasync :: HasCallStack => File -> IO ()+fdatasync uvf = checkFileClosed uvf $ \ fd -> do     uvm <- getUVManager     withUVRequest_ uvm (hs_uv_fs_fdatasync_threaded fd)  -- | Equivalent to <http://linux.die.net/man/2/ftruncate ftruncate(2)>.-ftruncate :: HasCallStack => FileT -> Int64 -> IO ()-ftruncate uvf off = checkFileTClosed uvf $ \ fd -> do+ftruncate :: HasCallStack => File -> Int64 -> IO ()+ftruncate uvf off = checkFileClosed uvf $ \ fd -> do     uvm <- getUVManager     withUVRequest_ uvm (hs_uv_fs_ftruncate_threaded fd off) @@ -437,8 +522,8 @@         withUVRequest_ uvm (hs_uv_fs_chmod_threaded p mode)  -- | Equivalent to <http://linux.die.net/man/2/fchmod fchmod(2)>.-fchmod :: HasCallStack => FileT -> FileMode -> IO ()-fchmod uvf mode = checkFileTClosed uvf $ \ fd -> do+fchmod :: HasCallStack => File -> FileMode -> IO ()+fchmod uvf mode = checkFileClosed uvf $ \ fd -> do     uvm <- getUVManager     withUVRequest_ uvm (hs_uv_fs_fchmod_threaded fd mode) @@ -459,8 +544,8 @@ -- | Equivalent to <https://man7.org/linux/man-pages/man3/futimes.3.html futime(3)>. -- -- Same precision notes with 'utime'.-futime :: HasCallStack => FileT -> Double -> Double -> IO ()-futime uvf atime mtime = checkFileTClosed uvf $ \ fd -> do+futime :: HasCallStack => File -> Double -> Double -> IO ()+futime uvf atime mtime = checkFileClosed uvf $ \ fd -> do     uvm <- getUVManager     withUVRequest_ uvm (hs_uv_fs_futime_threaded fd atime mtime) 

@@ -0,0 +1,205 @@+{-|+Module      : Z.IO.FileSystem.Watch+Description : cross-platform recursive fs watcher+Copyright   : (c) Dong Han, 2017~2019+License     : BSD+Maintainer  : winterland1989@gmail.com+Stability   : experimental+Portability : non-portable++This module provides fs watcher based on libuv's fs_event, we also maintain watch list if target OS doesn't+support recursive watch(Linux's inotify).++@+-- start watching threads, use returned close function to cleanup watching threads.+(close, srcf) <- watchDir "fold_to_be_watch"+-- dup a file event source+src <- srcf+-- print event to stdout+runBIO $ src >|> sinkToIO printLineStd+@+-}++module Z.IO.FileSystem.Watch (+    FileEvent(..)+  , watchDirs+  , watchDirsRecursively+  ) where++import           Control.Concurrent+import           Control.Monad+import           Data.Bits+import           Data.IORef+import qualified Data.HashMap.Strict      as HM+import qualified Data.List                as List+import           Data.Word+import           GHC.Generics+import           Data.Primitive.PrimArray+import           Z.Data.Array+import           Z.Data.Array.Unaligned+import           Z.Data.CBytes            (CBytes)+import qualified Z.Data.CBytes            as CBytes+import           Z.Data.JSON              (EncodeJSON, FromValue, ToValue)+import           Z.Data.Text.ShowT        (ShowT)+import           Z.Data.Vector            (defaultChunkSize)+import           Z.Foreign+import           Z.IO.BIO+import           Z.IO.BIO.Concurrent+import           Z.IO.Exception+import           Z.IO.FileSystem+import qualified Z.IO.FileSystem.FilePath as P+import           Z.IO.UV.FFI+import           Z.IO.UV.Manager+import           Z.IO.LowResTimer++-- | File event with path info.+data FileEvent = FileAdd CBytes | FileRemove CBytes | FileModify CBytes+  deriving (Show, Read, Ord, Eq, Generic)+  deriving anyclass (ShowT, FromValue, ToValue, EncodeJSON)+++-- | Start watching a list of given directories.+--+watchDirs :: [CBytes] -> IO (IO (), IO (Source FileEvent))+watchDirs dirs =  do+    forM_ dirs $ \ dir -> do+        b <- isDir dir+        unless b (throwUVIfMinus_ (return UV_ENOTDIR))+    watch_ 0 dirs++-- | Start watching a list of given directories recursively.+--+watchDirsRecursively :: [CBytes] -> IO (IO (), IO (Source FileEvent))+watchDirsRecursively dirs = do+#if defined(linux_HOST_OS)+    -- inotify doesn't support recursive watch, so we manually maintain watch list+    subDirs <- forM dirs (\ dir -> scandirRecursively dir (\ _ t -> return (t == DirEntDir)))+    watch_ UV_FS_EVENT_RECURSIVE (List.concat (dirs:subDirs))+#else+    watch_ UV_FS_EVENT_RECURSIVE dirs+#endif++-- Internal function to start watching+watch_ :: CUInt -> [CBytes] -> IO (IO (), IO (Source FileEvent))+watch_ flag dirs = do+    -- HashMap to store all watchers+    mRef <- newMVar HM.empty+    -- IORef store temp events to de-duplicated+    eRef <- newIORef Nothing+    -- there's only one place to pull the sink, that is cleanUpWatcher+    (sink, srcf) <- newBroadcastTChanNode 1+    -- lock UVManager first+    (forM_ dirs $ \ dir -> do+        dir' <- P.normalize dir+        tid <- forkIO $ watchThread mRef eRef dir' sink+        modifyMVar_ mRef $ \ m ->+            return $! HM.insert dir' tid m) `onException` cleanUpWatcher mRef sink+    return (cleanUpWatcher mRef sink, srcf)+  where+    eventBufSiz = defaultChunkSize++    cleanUpWatcher mRef sink = do+        m <- takeMVar mRef+        forM_ m killThread+        void (pull sink)++    watchThread mRef eRef dir sink = do+        uvm <- getUVManager+        (bracket+            (do withUVManager uvm $ \ loop -> do+                    hdl <- hs_uv_handle_alloc loop+                    slot <- getUVSlot uvm (peekUVHandleData hdl)+                    -- init uv struct+                    throwUVIfMinus_ (uv_fs_event_init loop hdl)++                    buf <- newPinnedPrimArray eventBufSiz :: IO (MutablePrimArray RealWorld Word8)++                    check <- throwOOMIfNull $ hs_uv_check_alloc+                    throwUVIfMinus_ (hs_uv_check_init check hdl)++                    withMutablePrimArrayContents buf $ \ p -> do+                        pokeBufferTable uvm slot (castPtr p) eventBufSiz+                        -- init uv_check_t must come after poking buffer+                        throwUVIfMinus_ $ hs_uv_fs_event_check_start check++                    return (hdl, slot, buf, check))++            (\ (hdl,_,_,check) -> hs_uv_handle_close hdl >> hs_uv_check_close check)++            (\ (hdl, slot, buf, _) -> do+                m <- getBlockMVar uvm slot+                withUVManager' uvm $ do+                    _ <- tryTakeMVar m+                    pokeBufferSizeTable uvm slot eventBufSiz+                    CBytes.withCBytesUnsafe dir $ \ p ->+                        throwUVIfMinus_ (hs_uv_fs_event_start hdl p flag)++                forever $ do++                    _ <- takeMVar m `onException` (do+                            _ <- withUVManager' uvm $ uv_fs_event_stop hdl+                            void (tryTakeMVar m))++                    (PrimArray buf#) <- withUVManager' uvm $ do+                        _ <- tryTakeMVar m+                        r <- peekBufferSizeTable uvm slot+                        pokeBufferSizeTable uvm slot eventBufSiz++                        let eventSiz = eventBufSiz - r+                        buf' <- newPrimArray eventSiz+                        copyMutablePrimArray buf' 0 buf r eventSiz+                        unsafeFreezePrimArray buf'++                    forkIO $ processEvent dir mRef eRef sink =<< loopReadFileEvent buf# 0 [])+            ) `catch`+                -- when a directory is removed, either watcher is killed+                -- or hs_uv_fs_event_start return ENOENT+                (\ (_ :: NoSuchThing) -> return ())++    loopReadFileEvent buf# i acc+        | i >= siz = return acc+        | otherwise =+            let !event  = indexBA buf# i+                !path   = indexBA buf# (i + 1)+            in loopReadFileEvent buf# (i + CBytes.length path + 2) ((event,path):acc)+      where siz = sizeofPrimArray (PrimArray buf# :: PrimArray Word8)++    processEvent pdir mRef eRef sink = mapM_ $ \ (e, path) -> do+        f <- pdir `P.join` path+        if (e .&. UV_RENAME) /= 0+        then catch+            (do _s <- lstat f+#if defined(linux_HOST_OS)+                when ((stMode _s .&. S_IFMT == S_IFDIR) && (flag .&. UV_FS_EVENT_RECURSIVE /= 0)) $ do+                    modifyMVar_ mRef $ \ m -> do+                        case HM.lookup f m of+                            Just _ -> return m+                            _ -> do+                                ds <- scandirRecursively f (\ _ t -> return (t == DirEntDir))+                                foldM (\ m' d -> do+                                    tid <- forkIO $ watchThread mRef eRef d sink+                                    return $! HM.insert d tid m') m (f:ds)+#endif+                pushDedup eRef sink (FileAdd f))+            (\ (_ :: NoSuchThing) -> do+                modifyMVar_ mRef $ \ m -> do+                    forM_ (HM.lookup f m) killThread+                    return (HM.delete f m)+                pushDedup eRef sink (FileRemove f))+        else pushDedup eRef sink (FileModify f)++    pushDedup eRef sink event = do+        registerLowResTimer_ 1 $ do+            me' <- atomicModifyIORef' eRef $ \ me ->+                case me of+                    Just e -> (Nothing, Just e)+                    _ -> (Nothing, Nothing)+            forM_ me' (push sink)++        me' <- atomicModifyIORef' eRef $ \ me ->+            case me of+                Just e -> if (e == event)+                    then (me, Nothing)+                    else (Just event, Just e)+                _ -> (Just event, Nothing)+        forM_ me' (push sink)

@@ -108,8 +108,9 @@     , loggerTSCache        :: IO (B.Builder ())     -- ^ An IO action return a formatted date\/time string     , loggerFmt            :: LogFormatter-    -- ^ Output logs if level is equal or higher than this value.+    -- ^ Log formatter     , loggerLevel          :: {-# UNPACK #-} !Level+    -- ^ Output logs if level is equal or higher than this value.     }  -- | Logger config type used in this module.@@ -142,9 +143,8 @@  -- | Make a new simple logger. ---newLogger :: Output o-          => LoggerConfig-          -> MVar (BufferedOutput o)+newLogger :: LoggerConfig+          -> MVar BufferedOutput           -> IO Logger newLogger LoggerConfig{..} oLock = do     logsRef <- newIORef []@@ -187,7 +187,7 @@     let !bs = B.buildBytesWith loggerLineBufSize b     unless (V.null bs) $ atomicModifyIORef' logsRef (\ bss -> (bs:bss, ())) -flushLogIORef :: (HasCallStack, Output o) => MVar (BufferedOutput o) -> IORef [V.Bytes] -> IO ()+flushLogIORef :: HasCallStack => MVar BufferedOutput -> IORef [V.Bytes] -> IO () flushLogIORef oLock logsRef =     withMVar oLock $ \ o -> do         bss <- atomicModifyIORef' logsRef (\ bss -> ([], bss))@@ -247,11 +247,11 @@ getDefaultLogger :: IO Logger getDefaultLogger = readIORef globalLogger --- | Manually flush stderr logger.+-- | Manually flush global logger. flushDefaultLogger :: IO () flushDefaultLogger = getDefaultLogger >>= flushLogger --- | Flush stderr logger when program exits.+-- | Flush global logger when program exits. withDefaultLogger :: IO () -> IO () withDefaultLogger = (`finally` flushDefaultLogger) 

@@ -1,13 +1,13 @@ {-| Module      : Z.IO.LowResTimer-Description : Low resolution (0.1s) timing wheel+Description : Low resolution (decisecond) timing wheel Copyright   : (c) Dong Han, 2017-2018 License     : BSD Maintainer  : winterland1989@gmail.com Stability   : experimental Portability : non-portable -This module provide low resolution (0.1s) timers using a timing wheel of size 128 per capability,+This module provide low resolution (decisecond, i.e. 0.1s) timers using a timing wheel of size 128 per capability, each timer thread will automatically started or stopped based on demannd. register or cancel a timeout is O(1), and each step only need scan n\/128 items given timers are registered in an even fashion. @@ -141,7 +141,7 @@ --   registerLowResTimer 100 (forkIO $ killThread tid) -- @ ---registerLowResTimer :: Int          -- ^ timeout in unit of 0.1s+registerLowResTimer :: Int          -- ^ timeout in unit of decisecond(0.1s)                     -> IO ()        -- ^ the action you want to perform, it should not block                     -> IO LowResTimer registerLowResTimer t action = do@@ -149,7 +149,7 @@     registerLowResTimerOn lrtm t action  -- | 'void' ('registerLowResTimer' t action)-registerLowResTimer_ :: Int          -- ^ timeout in unit of 0.1s+registerLowResTimer_ :: Int          -- ^ timeout in unit of decisecond(0.1s)                      -> IO ()        -- ^ the action you want to perform, it should not block                      -> IO () registerLowResTimer_ t action = void (registerLowResTimer t action)@@ -157,7 +157,7 @@ -- | Same as 'registerLowResTimer', but allow you choose timer manager. -- registerLowResTimerOn :: LowResTimerManager   -- ^ a low resolution timer manager-                      -> Int          -- ^ timeout in unit of 0.1s+                      -> Int          -- ^ timeout in unit of decisecond(0.1s)                       -> IO ()        -- ^ the action you want to perform, it should not block                       -> IO LowResTimer registerLowResTimerOn lrtm@(LowResTimerManager queue indexLock regCounter _) t action = do@@ -205,7 +205,7 @@ -- Note timeoutLowRes is also implemented with 'Exception' underhood, which can have some surprising -- effects on some devices, e.g. use 'timeoutLowRes' with reading or writing on 'UVStream's may close -- the 'UVStream' once a reading or writing is not able to be done in time.-timeoutLowRes :: Int    -- ^ timeout in unit of 0.1s+timeoutLowRes :: Int    -- ^ timeout in unit of decisecond(0.1s)               -> IO a               -> IO (Maybe a) timeoutLowRes timeo io = do@@ -220,7 +220,7 @@ -- | Similar to 'timeoutLowRes', but raise a 'TimeOutException' to current thread -- instead of return 'Nothing' if timeout. timeoutLowResEx :: HasCallStack-                => Int    -- ^ timeout in unit of 0.1s+                => Int    -- ^ timeout in unit of decisecond(0.1s)                 -> IO a                 -> IO a timeoutLowResEx timeo io = do@@ -312,7 +312,7 @@ -- One common way to get a shared periodical updated value is to start a seperate thread and do calculation -- periodically, but doing that will stop system from being idle, which stop idle GC from running, -- and in turn disable deadlock detection, which is too bad. This function solves that.-throttle :: Int         -- ^ cache time in unit of 0.1s+throttle :: Int         -- ^ cache time in unit of decisecond(0.1s)          -> IO a        -- ^ the original IO action          -> IO (IO a)   -- ^ throttled IO action throttle t action = do@@ -334,7 +334,7 @@ -- no-ops. -- -- Note the action will run in the calling thread.-throttle_ :: Int            -- ^ cache time in unit of 0.1s+throttle_ :: Int            -- ^ cache time in unit of decisecond(0.1s)           -> IO ()          -- ^ the original IO action           -> IO (IO ())     -- ^ throttled IO action throttle_ t action = do

@@ -62,7 +62,7 @@  -- | init a IPC client 'Resource', which open a new connect when used. ---initIPCClient :: HasCallStack => IPCClientConfig -> Resource UVStream+initIPCClient :: IPCClientConfig -> Resource UVStream initIPCClient (IPCClientConfig cname tname) = do     uvm <- liftIO getUVManager     client <- initIPCStream uvm@@ -111,10 +111,10 @@         withCBytesUnsafe ipcListenName $ \ name_p -> do             throwUVIfMinus_ (uv_pipe_bind serverHandle name_p)         bracket-            (do check <- throwOOMIfNull $ hs_uv_accept_check_alloc-                throwUVIfMinus_ (hs_uv_accept_check_init check serverHandle)+            (do check <- throwOOMIfNull $ hs_uv_check_alloc+                throwUVIfMinus_ (hs_uv_check_init check serverHandle)                 return check)-            hs_uv_accept_check_close $+            hs_uv_check_close $             \ check -> do  -- The buffer passing of accept is a litte complicated here, to get maximum performance,@@ -155,8 +155,8 @@                     -- we lock uv manager here in case of next uv_run overwrite current accept buffer                     acceptBufCopy <- withUVManager' serverUVManager $ do                         _ <- tryTakeMVar m-                        acceptCountDown <- peekBufferTable serverUVManager serverSlot-                        pokeBufferTable serverUVManager serverSlot acceptBufPtr (backLog-1)+                        acceptCountDown <- peekBufferSizeTable serverUVManager serverSlot+                        pokeBufferSizeTable serverUVManager serverSlot (backLog-1)                          -- if acceptCountDown count to -1, we should resume on haskell side                         when (acceptCountDown == -1) (hs_uv_listen_resume serverHandle)

@@ -84,7 +84,6 @@ import           Z.Data.CBytes import           Z.Data.Text.ShowT          (ShowT) import           Z.IO.Exception-import           Z.IO.UV.Errno import           Z.Foreign  #include "hs_uv.h" 

@@ -123,10 +123,10 @@         withSocketAddrUnsafe tcpListenAddr $ \ addrPtr -> do             throwUVIfMinus_ (uv_tcp_bind serverHandle addrPtr 0)         bracket-            (do check <- throwOOMIfNull $ hs_uv_accept_check_alloc-                throwUVIfMinus_ (hs_uv_accept_check_init check serverHandle)+            (do check <- throwOOMIfNull $ hs_uv_check_alloc+                throwUVIfMinus_ (hs_uv_check_init check serverHandle)                 return check)-            hs_uv_accept_check_close $+            hs_uv_check_close $             \ check -> do -- The buffer passing of accept is a litte complicated here, to get maximum performance, -- we do batch accepting. i.e. recv multiple client inside libuv's event loop:@@ -168,8 +168,8 @@                     -- we lock uv manager here in case of next uv_run overwrite current accept buffer                     acceptBufCopy <- withUVManager' serverUVManager $ do                         _ <- tryTakeMVar m-                        acceptCountDown <- peekBufferTable serverUVManager serverSlot-                        pokeBufferTable serverUVManager serverSlot acceptBufPtr (backLog-1)+                        acceptCountDown <- peekBufferSizeTable serverUVManager serverSlot+                        pokeBufferSizeTable serverUVManager serverSlot (backLog-1)                          -- if acceptCountDown count to -1, we should resume on haskell side                         when (acceptCountDown == -1) (hs_uv_listen_resume serverHandle)@@ -201,7 +201,7 @@  -------------------------------------------------------------------------------- -initTCPStream :: HasCallStack => UVManager -> Resource UVStream+initTCPStream :: UVManager -> Resource UVStream initTCPStream = initUVStream (\ loop hdl -> throwUVIfMinus_ (uv_tcp_init loop hdl))  -- | Enable or disable @TCP_NODELAY@, which enable or disable Nagle’s algorithm.

@@ -73,7 +73,6 @@ import Z.Data.CBytes          as CBytes import Z.IO.Network.SocketAddr import Z.Foreign-import Z.IO.UV.Errno          (pattern UV_EMSGSIZE) import Z.IO.UV.FFI import Z.IO.UV.Manager import Z.IO.Exception@@ -116,20 +115,14 @@  -- | Initialize a UDP socket. ---initUDP :: HasCallStack-        => UDPConfig-        -> Resource UDP+initUDP :: UDPConfig -> Resource UDP initUDP (UDPConfig sbsiz maddr) = initResource     (do uvm <- getUVManager         (hdl, slot) <- withUVManager uvm $ \ loop -> do             hdl <- hs_uv_handle_alloc loop             slot <- getUVSlot uvm (peekUVHandleData hdl)-            -- clean up-            _ <- tryTakeMVar =<< getBlockMVar uvm slot-             -- init uv struct-            (do throwUVIfMinus_ (uv_udp_init loop hdl)-                ) `onException` hs_uv_handle_free hdl+            throwUVIfMinus_ (uv_udp_init loop hdl)             return (hdl, slot)          -- bind the socket if address is available@@ -373,10 +366,10 @@             -> IO () recvUDPLoop (UDPRecvConfig bufSiz bufArrSiz) udp@(UDP hdl slot uvm _ _) worker = do     bracket-        (do check <- throwOOMIfNull $ hs_uv_udp_check_alloc-            throwUVIfMinus_ (hs_uv_udp_check_init check hdl)+        (do check <- throwOOMIfNull $ hs_uv_check_alloc+            throwUVIfMinus_ (hs_uv_check_init check hdl)             return check)-        hs_uv_udp_check_close $+        hs_uv_check_close $         \ check -> do             buf@(_, rbufArr) <- newRecvBuf bufSiz bufArrSiz             withMutablePrimArrayContents rbufArr $ \ p -> do@@ -385,8 +378,7 @@                 throwUVIfMinus_ $ hs_uv_udp_check_start check             forever $ do                 msgs <- recvUDPWith udp buf bufSiz-                withMutablePrimArrayContents rbufArr $ \ p ->-                    pokeBufferTable uvm slot (castPtr p) (bufArrSiz-1)+                pokeBufferSizeTable uvm slot (bufArrSiz-1)                 forM_ msgs worker  -- | Recv messages from UDP socket, return source address if available, and a `Bool`@@ -395,10 +387,10 @@ recvUDP :: HasCallStack => UDPRecvConfig -> UDP -> IO [(Maybe SocketAddr, Bool, V.Bytes)] recvUDP (UDPRecvConfig bufSiz bufArrSiz) udp@(UDP hdl slot uvm _ _)  = do     bracket-        (do check <- throwOOMIfNull $ hs_uv_udp_check_alloc-            throwUVIfMinus_ (hs_uv_udp_check_init check hdl)+        (do check <- throwOOMIfNull $ hs_uv_check_alloc+            throwUVIfMinus_ (hs_uv_check_init check hdl)             return check)-        hs_uv_udp_check_close $+        hs_uv_check_close $         \ check -> do             buf@(_, rbufArr) <- newRecvBuf bufSiz bufArrSiz             withMutablePrimArrayContents rbufArr $ \ p -> do@@ -433,7 +425,8 @@         r <- takeMVar m `onException` (do                 -- normally we call 'uv_udp_recv_stop' in C read callback                 -- but when exception raise, here's the place to stop-                throwUVIfMinus_ $ withUVManager' uvm (uv_udp_recv_stop hdl)+                -- stop a handle twice will be a libuv error, so we don't check result+                _ <- withUVManager' uvm (uv_udp_recv_stop hdl)                 void (tryTakeMVar m))          forM [r+1..bufArrSiz-1] $ \ i -> do

@@ -257,7 +257,7 @@                     withUVManager uvm $ \ loop -> do                         slot <- throwUVIfMinus (hs_uv_spawn loop popts## pfile                                         pargs argsLen penv envLen pcwd pstdio##)-                        pid <- peekBufferTable uvm slot+                        pid <- peekBufferSizeTable uvm slot                         return (slot, pid)      exitLock <- getBlockMVar uvm slot

@@ -65,7 +65,7 @@ import System.IO.Unsafe import Z.Data.Builder as B import Z.Data.Vector as V-import Z.Data.Text.ShowT (ShowT, toBuilder)+import Z.Data.Text.ShowT (ShowT, toUTF8Builder) import Z.IO.UV.FFI import Z.IO.UV.Manager import Z.IO.UV.Errno@@ -104,7 +104,10 @@         -- since we are inside mask, this is the only place         -- async exceptions could possibly kick in, and we should stop reading         r <- takeMVar m `onException` (do-                throwUVIfMinus_ $ withUVManager' uvm (uv_read_stop hdl)+                -- normally we call 'uv_read_stop' in C read callback+                -- but when exception raise, here's the place to stop+                -- stop a handle twice will be a libuv error, so we don't check result+                _ <- withUVManager' uvm (uv_read_stop hdl)                 void (tryTakeMVar m))         if  | r > 0  -> return r             | r == fromIntegral UV_EOF -> return 0@@ -126,6 +129,12 @@             m <- getBlockMVar uvm reqSlot             _ <- tryTakeMVar m             return m+        -- we can't cancel uv_write_t with current libuv,+        -- otherwise disaster will happen if buffer got collected.+        -- so we have to turn to uninterruptibleMask_'s help.+        -- i.e. writing UVStream is an uninterruptible operation.+        -- OS will guarantee writing TTY and socket will not+        -- hang forever anyway.         throwUVIfMinus_ (uninterruptibleMask_ $ takeMVar m)     writeOutput (StdFile fd) buf len = go buf len       where@@ -155,21 +164,21 @@ stderr = unsafePerformIO (makeStdStream 2)  -- |  A global buffered stdin stream protected by 'MVar'.-stdinBuf :: MVar (BufferedInput StdStream)+stdinBuf :: MVar BufferedInput {-# NOINLINE stdinBuf #-} stdinBuf = unsafePerformIO (newBufferedInput stdin >>= newMVar)  -- |  A global buffered stdout stream protected by 'MVar'. -- -- | If you want to write logs, don't use 'stdoutBuf' directly, use 'Z.IO.Logger' instead.-stdoutBuf :: MVar (BufferedOutput StdStream)+stdoutBuf :: MVar BufferedOutput {-# NOINLINE stdoutBuf #-} stdoutBuf = unsafePerformIO (newBufferedOutput stdout >>= newMVar)  -- |  A global buffered stderr stream protected by 'MVar'. -- -- | If you want to write logs, don't use 'stderrBuf' directly, use 'Z.IO.Logger' instead.-stderrBuf :: MVar (BufferedOutput StdStream)+stderrBuf :: MVar BufferedOutput {-# NOINLINE stderrBuf #-} stderrBuf = unsafePerformIO (newBufferedOutput stderr >>= newMVar) @@ -177,14 +186,13 @@ makeStdStream fd = do     typ <- uv_guess_handle fd     if typ == UV_TTY-    then do+    then mask_ $ do         uvm <- getUVManager         withUVManager uvm $ \ loop -> do             hdl <- hs_uv_handle_alloc loop             slot <- getUVSlot uvm (peekUVHandleData hdl)             _ <- tryTakeMVar =<< getBlockMVar uvm slot   -- clear the parking spot             throwUVIfMinus_ (uv_tty_init loop hdl (fromIntegral fd))-                `onException` hs_uv_handle_free hdl             return (StdTTY hdl slot uvm)     else return (StdFile fd) @@ -210,7 +218,7 @@  -- | Print a 'ShowT' and flush to stdout. printStd :: (HasCallStack, ShowT a) => a -> IO ()-printStd s = putStd (toBuilder s)+printStd s = putStd (toUTF8Builder s)  -- | Print a 'Builder' and flush to stdout. putStd :: HasCallStack => Builder a -> IO ()@@ -220,7 +228,7 @@  -- | Print a 'ShowT' and flush to stdout, with a linefeed. printLineStd :: (HasCallStack, ShowT a) => a -> IO ()-printLineStd s = putLineStd (toBuilder s)+printLineStd s = putLineStd (toUTF8Builder s)  -- | Print a 'Builder' and flush to stdout, with a linefeed. putLineStd :: HasCallStack => Builder a -> IO ()

@@ -126,6 +126,13 @@ foreign import ccall unsafe hs_uv_cancel :: Ptr UVLoop -> UVSlot -> IO ()  --------------------------------------------------------------------------------+-- check+foreign import ccall unsafe hs_uv_check_alloc :: IO (Ptr UVHandle)+foreign import ccall unsafe hs_uv_check_init :: Ptr UVHandle    -- ^ uv_check_t+                                             -> Ptr UVHandle    -- ^ uv_handle_t+                                             -> IO CInt+foreign import ccall unsafe hs_uv_check_close :: Ptr UVHandle -> IO ()+-------------------------------------------------------------------------------- -- stream  foreign import ccall unsafe hs_uv_listen  :: Ptr UVHandle -> CInt -> IO CInt@@ -135,12 +142,7 @@ foreign import ccall unsafe uv_read_stop :: Ptr UVHandle -> IO CInt foreign import ccall unsafe hs_uv_write :: Ptr UVHandle -> Ptr Word8 -> Int -> IO UVSlotUnsafe -foreign import ccall unsafe hs_uv_accept_check_alloc :: IO (Ptr UVHandle)-foreign import ccall unsafe hs_uv_accept_check_init :: Ptr UVHandle -- ^ uv_check_t-                                                    -> Ptr UVHandle -- ^ uv_stream_t-                                                    -> IO CInt foreign import ccall unsafe hs_uv_accept_check_start :: Ptr UVHandle -> IO CInt-foreign import ccall unsafe hs_uv_accept_check_close :: Ptr UVHandle -> IO ()  -------------------------------------------------------------------------------- -- tcp & pipe@@ -210,12 +212,7 @@ foreign import ccall unsafe hs_uv_udp_recv_start :: Ptr UVHandle -> IO CInt foreign import ccall unsafe uv_udp_recv_stop :: Ptr UVHandle -> IO CInt -foreign import ccall unsafe hs_uv_udp_check_alloc :: IO (Ptr UVHandle)-foreign import ccall unsafe hs_uv_udp_check_init :: Ptr UVHandle    -- ^ uv_check_t-                                                 -> Ptr UVHandle    -- ^ udp server-                                                 -> IO CInt foreign import ccall unsafe hs_uv_udp_check_start :: Ptr UVHandle -> IO CInt-foreign import ccall unsafe hs_uv_udp_check_close :: Ptr UVHandle -> IO ()  foreign import ccall unsafe hs_uv_udp_send      :: Ptr UVHandle -> MBA## SocketAddr -> Ptr Word8 -> Int -> IO UVSlotUnsafe@@ -232,7 +229,7 @@  -- | Terminal mode. ----- When in 'UV_TTY_MODE_RAW' mode, input is always available character-by-character,+-- When in 'TTY_MODE_RAW' mode, input is always available character-by-character, -- not including modifiers. Additionally, all special processing of characters by the terminal is disabled,  -- including echoing input characters. Note that CTRL+C will no longer cause a SIGINT when in this mode. type TTYMode = CInt@@ -305,6 +302,22 @@ pattern DEFAULT_MODE :: FileMode pattern DEFAULT_MODE = 0o666 +-- | This is the file type mask.+pattern S_IFMT :: FileMode+pattern S_IFMT = #const S_IFMT++-- | This is the file type constant of a symbolic link.+pattern S_IFLNK :: FileMode+pattern S_IFLNK = #const S_IFLNK++-- | This is the file type constant of a directory file.+pattern S_IFDIR :: FileMode+pattern S_IFDIR = #const S_IFDIR++-- | This is the file type constant of a regular file.+pattern S_IFREG :: FileMode+pattern S_IFREG = #const S_IFREG+ -- non-threaded functions foreign import ccall unsafe hs_uv_fs_open    :: BA## Word8 -> FileFlag -> FileMode -> IO FD foreign import ccall unsafe hs_uv_fs_close   :: FD -> IO Int@@ -349,20 +362,20 @@  -- | If the path is not a directory, fail the open. (Not useful on regular file) ----- Note 'o_DIRECTORY' is not supported on Windows.+-- Note 'O_DIRECTORY' is not supported on Windows. pattern O_DIRECTORY :: FileFlag pattern O_DIRECTORY = #const UV_FS_O_DIRECTORY  -- |The file is opened for synchronous IO. Write operations will complete once all data and a minimum of metadata are flushed to disk. ----- Note 'o_DSYNC' is supported on Windows via @FILE_FLAG_WRITE_THROUGH@.+-- Note 'O_DSYNC' is supported on Windows via @FILE_FLAG_WRITE_THROUGH@. pattern O_DSYNC :: FileFlag pattern O_DSYNC = #const UV_FS_O_DSYNC --- | If the 'o_CREAT' flag is set and the file already exists, fail the open.+-- | If the 'O_CREAT' flag is set and the file already exists, fail the open. ----- Note In general, the behavior of 'o_EXCL' is undefined if it is used without 'o_CREAT'. There is one exception: on --- Linux 2.6 and later, 'o_EXCL' can be used without 'o_CREAT' if pathname refers to a block device. If the block +-- Note In general, the behavior of 'O_EXCL' is undefined if it is used without 'O_CREAT'. There is one exception: on +-- Linux 2.6 and later, 'O_EXCL' can be used without 'O_CREAT' if pathname refers to a block device. If the block  -- device is in use by the system (e.g., mounted), the open will fail with the error @EBUSY@. pattern O_EXCL :: FileFlag pattern O_EXCL = #const UV_FS_O_EXCL@@ -376,31 +389,31 @@  -- | Do not update the file access time when the file is read. -- --- Note 'o_NOATIME' is not supported on Windows.+-- Note 'O_NOATIME' is not supported on Windows. pattern O_NOATIME :: FileFlag pattern O_NOATIME = #const UV_FS_O_NOATIME  -- | If the path identifies a terminal device, opening the path will not cause that terminal to become the controlling terminal for the process (if the process does not already have one). (Not sure if this flag is useful) ----- Note 'o_NOCTTY' is not supported on Windows.+-- Note 'O_NOCTTY' is not supported on Windows. pattern O_NOCTTY :: FileFlag pattern O_NOCTTY = #const UV_FS_O_NOCTTY  -- | If the path is a symbolic link, fail the open. ----- Note 'o_NOFOLLOW' is not supported on Windows.+-- Note 'O_NOFOLLOW' is not supported on Windows. pattern O_NOFOLLOW :: FileFlag pattern O_NOFOLLOW = #const UV_FS_O_NOFOLLOW  -- | Open the file in nonblocking mode if possible. (Definitely not useful in Z-IO) ----- Note 'o_NONBLOCK' is not supported on Windows. (Not useful on regular file anyway)+-- Note 'O_NONBLOCK' is not supported on Windows. (Not useful on regular file anyway) pattern O_NONBLOCK :: FileFlag pattern O_NONBLOCK = #const UV_FS_O_NONBLOCK  -- | Access is intended to be random. The system can use this as a hint to optimize file caching. -- --- Note 'o_RANDOM' is only supported on Windows via @FILE_FLAG_RANDOM_ACCESS@.+-- Note 'O_RANDOM' is only supported on Windows via @FILE_FLAG_RANDOM_ACCESS@. pattern O_RANDOM :: FileFlag pattern O_RANDOM = #const UV_FS_O_RANDOM @@ -415,13 +428,13 @@  -- | Access is intended to be sequential from beginning to end. The system can use this as a hint to optimize file caching. -- --- Note 'o_SEQUENTIAL' is only supported on Windows via @FILE_FLAG_SEQUENTIAL_SCAN@.+-- Note 'O_SEQUENTIAL' is only supported on Windows via @FILE_FLAG_SEQUENTIAL_SCAN@. pattern O_SEQUENTIAL :: FileFlag pattern O_SEQUENTIAL = #const UV_FS_O_SEQUENTIAL  -- | The file is temporary and should not be flushed to disk if possible. ----- Note 'o_SHORT_LIVED' is only supported on Windows via @FILE_ATTRIBUTE_TEMPORARY@.+-- Note 'O_SHORT_LIVED' is only supported on Windows via @FILE_ATTRIBUTE_TEMPORARY@. pattern O_SHORT_LIVED :: FileFlag pattern O_SHORT_LIVED = #const UV_FS_O_SHORT_LIVED @@ -431,13 +444,13 @@  -- | The file is opened for synchronous IO. Write operations will complete once all data and all metadata are flushed to disk. ----- Note 'o_SYNC' is supported on Windows via @FILE_FLAG_WRITE_THROUGH@.+-- Note 'O_SYNC' is supported on Windows via @FILE_FLAG_WRITE_THROUGH@. pattern O_SYNC :: FileFlag pattern O_SYNC = #const UV_FS_O_SYNC  -- | The file is temporary and should not be flushed to disk if possible. ----- Note 'o_TEMPORARY' is only supported on Windows via @FILE_ATTRIBUTE_TEMPORARY@.+-- Note 'O_TEMPORARY' is only supported on Windows via @FILE_ATTRIBUTE_TEMPORARY@. pattern O_TEMPORARY :: FileFlag pattern O_TEMPORARY = #const UV_FS_O_TEMPORARY @@ -511,10 +524,10 @@  data FStat = FStat     { stDev      :: {-# UNPACK #-} !Word64-    , stMode     :: {-# UNPACK #-} !Word64+    , stMode     :: {-# UNPACK #-} !FileMode     , stNlink    :: {-# UNPACK #-} !Word64-    , stUid      :: {-# UNPACK #-} !Word64-    , stGid      :: {-# UNPACK #-} !Word64+    , stUID      :: {-# UNPACK #-} !UID+    , stGID      :: {-# UNPACK #-} !GID     , stRdev     :: {-# UNPACK #-} !Word64     , stIno      :: {-# UNPACK #-} !Word64     , stSize     :: {-# UNPACK #-} !Word64@@ -535,10 +548,10 @@ peekUVStat :: Ptr FStat -> IO FStat peekUVStat p = FStat     <$> (#{peek uv_stat_t, st_dev          } p)-    <*> (#{peek uv_stat_t, st_mode         } p)+    <*> (fromIntegral <$> (#{peek uv_stat_t, st_mode } p :: IO Word64))     <*> (#{peek uv_stat_t, st_nlink        } p)-    <*> (#{peek uv_stat_t, st_uid          } p)-    <*> (#{peek uv_stat_t, st_gid          } p)+    <*> (fromIntegral <$> (#{peek uv_stat_t, st_uid } p :: IO Word64))+    <*> (fromIntegral <$> (#{peek uv_stat_t, st_gid } p :: IO Word64))     <*> (#{peek uv_stat_t, st_rdev         } p)     <*> (#{peek uv_stat_t, st_ino          } p)     <*> (#{peek uv_stat_t, st_size         } p)@@ -692,7 +705,7 @@ #else     Word32 #endif-   deriving (Eq, Ord, Show, Generic)+   deriving (Eq, Ord, Show, Read, Generic)    deriving newtype (Storable, Prim, Unaligned, Num, EncodeJSON, ToValue, FromValue)    deriving anyclass ShowT @@ -702,7 +715,7 @@ #else     Word32 #endif-   deriving (Eq, Ord, Show, Generic)+   deriving (Eq, Ord, Show, Read, Generic)    deriving newtype (Storable, Prim, Unaligned, Num, EncodeJSON, ToValue, FromValue)    deriving anyclass ShowT @@ -774,14 +787,14 @@     , processGID :: GID -- ^ This happens only when the appropriate bits are set in the flags fields.     , processStdStreams :: (ProcessStdStream, ProcessStdStream, ProcessStdStream) -- ^ Specifying how (stdin, stdout, stderr) should be passed/created to the child, see 'ProcessStdStream'                             -    }   deriving (Eq, Ord, Show, Generic)+    }   deriving (Eq, Ord, Show, Read, Generic)         deriving anyclass (ShowT, EncodeJSON, ToValue, FromValue)  data ProcessStdStream     = ProcessIgnore     -- ^ redirect process std stream to \/dev\/null     | ProcessCreate     -- ^ create a new std stream     | ProcessInherit FD -- ^ pass an existing FD to child process as std stream-  deriving  (Eq, Ord, Show, Generic)+  deriving  (Eq, Ord, Show, Read, Generic)   deriving anyclass (ShowT, EncodeJSON, ToValue, FromValue)  processStdStreamFlag :: ProcessStdStream -> CInt@@ -921,7 +934,7 @@ foreign import ccall unsafe uv_os_setpriority :: PID -> CInt -> IO CInt  newtype PID = PID CInt -    deriving (Eq, Ord, Show, Generic)+    deriving (Eq, Ord, Show, Read, Generic)     deriving newtype (Storable, Prim, Unaligned, EncodeJSON, ToValue, FromValue)     deriving anyclass ShowT @@ -957,7 +970,7 @@     , os_release :: CBytes     , os_version :: CBytes     , os_machine :: CBytes-    }   deriving (Eq, Ord, Show, Generic)+    }   deriving (Eq, Ord, Show, Read, Generic)         deriving anyclass (ShowT, EncodeJSON, ToValue, FromValue)  getOSName :: IO OSName@@ -982,7 +995,7 @@     , passwd_gid :: GID     , passwd_shell :: CBytes     , passwd_homedir :: CBytes-    }   deriving (Eq, Ord, Show, Generic)+    }   deriving (Eq, Ord, Show, Read, Generic)         deriving anyclass (ShowT, EncodeJSON, ToValue, FromValue)  foreign import ccall unsafe uv_os_get_passwd :: MBA## PassWD -> IO CInt@@ -1024,7 +1037,7 @@     , cpu_times_sys  :: Word64  -- ^ milliseconds      , cpu_times_idle :: Word64  -- ^ milliseconds       , cpu_times_irq  :: Word64  -- ^ milliseconds-    }   deriving (Eq, Ord, Show, Generic)+    }   deriving (Eq, Ord, Show, Read, Generic)         deriving anyclass (ShowT, EncodeJSON, ToValue, FromValue)  -- | Gets information about the CPUs on the system.@@ -1058,3 +1071,20 @@     return ( indexPrimArray arr 0            , indexPrimArray arr 1            , indexPrimArray arr 2)++--------------------------------------------------------------------------------+-- fs event++foreign import ccall unsafe uv_fs_event_init :: Ptr UVLoop -> Ptr UVHandle -> IO CInt+foreign import ccall unsafe hs_uv_fs_event_start :: Ptr UVHandle -> BA## Word8 -> CUInt -> IO CInt+foreign import ccall unsafe uv_fs_event_stop :: Ptr UVHandle -> IO CInt+foreign import ccall unsafe hs_uv_fs_event_check_start :: Ptr UVHandle -> IO CInt++pattern UV_RENAME :: Word8+pattern UV_RENAME = #const UV_RENAME++pattern UV_CHANGE :: Word8+pattern UV_CHANGE = #const UV_CHANGE++pattern UV_FS_EVENT_RECURSIVE :: CUInt+pattern UV_FS_EVENT_RECURSIVE = #const UV_FS_EVENT_RECURSIVE

@@ -27,7 +27,8 @@   ( UVManager   , getUVManager   , getBlockMVar-  , peekBufferTable+  , peekBufferSizeTable+  , pokeBufferSizeTable   , pokeBufferTable   , withUVManager   , withUVManager'@@ -140,12 +141,24 @@     pokeElemOff bufTable slot buf     pokeElemOff bufSizTable slot (fromIntegral bufSiz) -peekBufferTable :: UVManager -> UVSlot -> IO Int-{-# INLINABLE peekBufferTable #-}-peekBufferTable uvm slot = do+-- | Peek buffer size table+--+-- The notes on 'pokeBufferTable' apply here too.+peekBufferSizeTable :: UVManager -> UVSlot -> IO Int+{-# INLINABLE peekBufferSizeTable #-}+peekBufferSizeTable uvm slot = do     (_, bufSizTable) <- peekUVBufferTable (uvmLoopData uvm)     fromIntegral <$> peekElemOff bufSizTable slot +-- | Poke buffer size table+--+-- The notes on 'pokeBufferTable' apply here too.+pokeBufferSizeTable :: UVManager -> UVSlot -> Int -> IO ()+{-# INLINABLE pokeBufferSizeTable #-}+pokeBufferSizeTable uvm slot bufSiz = do+    (_, bufSizTable) <- peekUVBufferTable (uvmLoopData uvm)+    pokeElemOff bufSizTable slot (fromIntegral bufSiz)+ initUVManager :: HasCallStack => Int -> Int -> Resource UVManager initUVManager siz cap = do     loop  <- initResource@@ -240,7 +253,7 @@                 -- After step finished, other threads are free to take the same slot,                 -- thus can overwrite the buffer size table, i.e. the previous result.                 ---                r <- peekBufferTable uvm slot+                r <- peekBufferSizeTable uvm slot                 tryPutMVar lock r             return c 

@@ -72,7 +72,8 @@         hdl <- hs_uv_handle_alloc loop         slot <- getUVSlot uvm (peekUVHandleData hdl)         _ <- tryTakeMVar =<< getBlockMVar uvm slot   -- clear the parking spot-        f loop hdl `onException` hs_uv_handle_free hdl+        -- this function should be run inside mask, no need to protect+        f loop hdl -- `onException` hs_uv_handle_free hdl         closed <- newIORef False         return (UVStream hdl slot uvm closed))     closeUVStream@@ -109,7 +110,8 @@         r <- takeMVar m `onException` (do                 -- normally we call 'uv_read_stop' in C read callback                 -- but when exception raise, here's the place to stop-                throwUVIfMinus_ $ withUVManager' uvm (uv_read_stop hdl)+                -- stop a handle twice will be a libuv error, so we don't check result+                _ <- withUVManager' uvm (uv_read_stop hdl)                 void (tryTakeMVar m))          if  | r > 0  -> return r

@@ -295,6 +295,26 @@ }  ////////////////////////////////////////////////////////////////////////////////+// check++// It's hard to arrange accepting notification without check handler, we can't+// do it in listen's callback, since it'll be called multiple times during uv_run.+uv_check_t* hs_uv_check_alloc(){+    uv_check_t* check = malloc(sizeof(uv_check_t));+    return check;+}++int hs_uv_check_init(uv_check_t* check, uv_handle_t* handle){+    int r = uv_check_init(handle->loop, check);+    check->data = (void*)handle;    // we link server to check's data field+    return r;+}++void hs_uv_check_close(uv_check_t* check){+    uv_close((uv_handle_t*)check, (uv_close_cb)free);+}++//////////////////////////////////////////////////////////////////////////////// // random  // non thread pool version of random

@@ -0,0 +1,72 @@+/*+ * Copyright (c) 2020 Dong Han+ *+ * All rights reserved.+ *+ * Redistribution and use in source and binary forms, with or without+ * modification, are permitted provided that the following conditions+ * are met:+ * 1. Redistributions of source code must retain the above copyright+ *    notice, this list of conditions and the following disclaimer.+ * 2. Redistributions in binary form must reproduce the above copyright+ *    notice, this list of conditions and the following disclaimer in the+ *    documentation and/or other materials provided with the distribution.+ * 3. Neither the names of the authors or the names of any contributors+ *    may be used to endorse or promote products derived from this software+ *    without specific prior written permission.+ *+ * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE+ * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS+ * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)+ * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY+ * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF+ * SUCH DAMAGE.+ */++#include <hs_uv.h>+#include <stdlib.h>++void hs_fs_event_cb(uv_fs_event_t* handle, const char* filename, int events, int status){+    if (status < 0) return;+    HsInt slot = (HsInt)handle->data;+    hs_loop_data* loop_data = handle->loop->data;++    size_t l = strlen(filename) + 2; // including the \NUL and event byte+    // we simply ignore more events if buffer can't hold it+    // libuv use a buffer size 4096, so on haskell side anything > 4096 should work+    if (loop_data->buffer_size_table[slot] >= l){+        loop_data->buffer_size_table[slot] -= l;+        char* buf = loop_data->buffer_table[slot] + loop_data->buffer_size_table[slot];+        *buf = (uint8_t)events;+        memcpy(buf+1, filename, l-1);+    }+}++int hs_uv_fs_event_start(uv_fs_event_t* handle, const char* path, unsigned int flags){+    return uv_fs_event_start(handle, hs_fs_event_cb, path, flags);+}++// Check if the fs event buffer is filled with messages, if so, unlock the fs event thread+//+void hs_fs_event_check_cb(uv_check_t* check){+    uv_fs_event_t* f =(uv_fs_event_t*)check->data;+    HsInt slot = (HsInt)f->data;+    hs_loop_data* loop_data = f->loop->data;+    size_t buffer_index = loop_data->buffer_size_table[slot];+    // This relys on GHC ByteArray# memory layout, ByteArray# length is recorded before content.+    HsInt* buffer_ptr = (HsInt*)loop_data->buffer_table[slot];+    HsInt buffer_total_len = *(buffer_ptr-1);+    if (buffer_index < buffer_total_len ) {+        loop_data->event_queue[loop_data->event_counter] = slot; // push the slot to event queue+        loop_data->event_counter += 1;+    }+}++int hs_uv_fs_event_check_start(uv_check_t* check){+    return uv_check_start(check, hs_fs_event_check_cb);+}

@@ -349,25 +349,8 @@     } } -// It's hard to arrange accepting notification without check handler, we can't-// do it in listen's callback, since it'll be called multiple times during uv_run.-uv_check_t* hs_uv_accept_check_alloc(){-    uv_check_t* check = malloc(sizeof(uv_check_t));-    return check;-}--int hs_uv_accept_check_init(uv_check_t* check, uv_stream_t* server){-    int r = uv_check_init(server->loop, check);-    check->data = (void*)server;    // we link server to check's data field-    return r;-}- int hs_uv_accept_check_start(uv_check_t* check){     return uv_check_start(check, hs_accept_check_cb);-}--void hs_uv_accept_check_close(uv_check_t* check){-    uv_close((uv_handle_t*)check, (uv_close_cb)free); }  int hs_set_socket_reuse(uv_stream_t* server) {

@@ -113,25 +113,8 @@     } } -// It's hard to arrange receiving notification without check handler, we can't-// do it in recv's callback, since it'll be called multiple times during uv_run.-uv_check_t* hs_uv_udp_check_alloc(){-    uv_check_t* check = malloc(sizeof(uv_check_t));-    return check;-}--int hs_uv_udp_check_init(uv_check_t* check, uv_udp_t* server){-    int r = uv_check_init(server->loop, check);-    check->data = (void*)server;    // we link server to check's data field-    return r;-}- int hs_uv_udp_check_start(uv_check_t* check){     return uv_check_start(check, hs_udp_check_cb);-}--void hs_uv_udp_check_close(uv_check_t* check){-    uv_close((uv_handle_t*)check, (uv_close_cb)free); }  void hs_uv_udp_send_cb(uv_udp_send_t* req, int status){

@@ -253,16 +253,20 @@ // request void hs_uv_cancel(uv_loop_t* loop, HsInt slot); + ////////////////////////////////////////////////////////////////////////////////+// request+uv_check_t* hs_uv_check_alloc();+int hs_uv_check_init(uv_check_t* check, uv_handle_t* handle);+void hs_uv_check_close(uv_check_t* check);++//////////////////////////////////////////////////////////////////////////////// // stream int hs_uv_listen(uv_stream_t* stream, int backlog); void hs_uv_listen_resume(uv_stream_t* server); int hs_uv_read_start(uv_stream_t* handle); HsInt hs_uv_write(uv_stream_t* handle, char* buf, HsInt buf_size);-uv_check_t* hs_uv_accept_check_alloc();-int hs_uv_accept_check_init(uv_check_t* check, uv_stream_t* server); int hs_uv_accept_check_start(uv_check_t* check);-void hs_uv_accept_check_close(uv_check_t* check);  //////////////////////////////////////////////////////////////////////////////// // tcp and pipe@@ -300,10 +304,7 @@ int hs_uv_udp_recv_start(uv_udp_t* handle); HsInt hs_uv_udp_send_connected(uv_udp_t* handle, char* buf, HsInt buf_siz); HsInt hs_uv_upd_send(uv_udp_t* handle, const struct sockaddr* addr, char* buf, HsInt buf_siz);-uv_check_t* hs_uv_udp_check_alloc();-int hs_uv_udp_check_init(uv_check_t* check, uv_udp_t* server); int hs_uv_udp_check_start(uv_check_t* check);-void hs_uv_udp_check_close(uv_check_t* check);  //////////////////////////////////////////////////////////////////////////////// // fs@@ -689,3 +690,8 @@ // random HsInt hs_uv_random(char* buf, size_t len, unsigned int flags); HsInt hs_uv_random_threaded(char* buf, size_t len, unsigned int flags, uv_loop_t* loop);++////////////////////////////////////////////////////////////////////////////////+// fs event+int hs_uv_fs_event_start(uv_fs_event_t* handle, const char* path, unsigned int flags);+int hs_uv_fs_event_check_start(uv_check_t* check);

@@ -1,77 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}--module Z.Compression.ZlibSpec where--import           Control.Monad-import           Data.IORef-import           Data.ByteString       as B-import           Data.ByteString.Lazy  as BL-import           Z.Data.Vector         as V-import           Z.Compression.Zlib-import qualified Codec.Compression.Zlib as TheZlib-import           Z.IO.Buffered-import           Test.QuickCheck-import           Test.QuickCheck.Function-import           Test.QuickCheck.Property-import           Test.Hspec-import           Test.Hspec.QuickCheck-import           Test.HUnit---spec :: Spec-spec = describe "zlib" $ do--    describe "decompress . compress === id" . modifyMaxSize (*50) $ do-        prop "decompress . compress === id" $ \ xs  -> do-            decompress defaultDecompressConfig-                (compress defaultCompressConfig (V.pack xs)) === V.pack xs--        prop "decompress . compress === id(with dict)" $ \ xs  -> do-            let dict = "aabbccdd"-            decompress defaultDecompressConfig{decompressDictionary = dict}-                (compress defaultCompressConfig{compressDictionary = dict} (V.pack xs))-                    === V.pack xs--        prop "compress === TheZlib.compress" $ \ xs  -> do-            V.unpack (compress defaultCompressConfig (V.pack xs)) ===-                BL.unpack (TheZlib.compressWith TheZlib.defaultCompressParams (BL.pack xs))--        prop "compress === TheZlib.compress(with dict)" $ \ xs  -> do-            V.unpack (compress defaultCompressConfig{compressDictionary = "aabbccdd"} (V.pack xs)) ===-                BL.unpack (TheZlib.compressWith-                    TheZlib.defaultCompressParams{TheZlib.compressDictionary = Just "aabbccdd" }-                        (BL.pack xs))--        prop "decompressSink . compressSource === id" $ \ xss -> do-            ref <- newIORef []-            (write, flush) <- compressSink defaultCompressConfig-                (\ x -> modifyIORef' ref (x:), return ())--            let vs = Prelude.map V.pack xss--            forM_ vs write-            flush--            source <- sourceFromList . Prelude.reverse =<< readIORef ref-            vs' <- collectSource =<< decompressSource defaultDecompressConfig source--            V.concat vs @=? V.concat vs'--        prop "decompressSink . compressSource === id(with dict)" $ \ xss -> do-            let dict = "aabbccdd"-            ref <- newIORef []-            (write, flush) <- compressSink defaultCompressConfig{compressDictionary = dict}--                (\ x -> modifyIORef' ref (x:), return ())--            let vs = Prelude.map V.pack xss--            forM_ vs write-            flush--            source <- sourceFromList . Prelude.reverse =<< readIORef ref-            vs' <- collectSource =<< decompressSource-                defaultDecompressConfig{decompressDictionary = dict} source--            V.concat vs @=? V.concat vs'-

@@ -0,0 +1,117 @@+{-# LANGUAGE OverloadedStrings #-}++module Z.IO.BIO.ConcurrentSpec where++import           Control.Concurrent+import           Control.Monad+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.Hspec+import           Test.Hspec.QuickCheck+import           Test.HUnit+import           System.IO.Unsafe+++spec :: Spec+spec = describe "BIO.Concurrent" $ do++    it "TQueueNode works as expected" $ do+        let content = [1..1000]++        (sink, src) <- newTQueueNode 2+        sumRef <- newIORef 0++        let producter = do+                src' <- sourceListWithDelay content+                runBIO (src' >|> sink)++        let consumer =  do+                (rRef, sink') <- sinkToList+                runBIO (src >|> sink')+                r <- readIORef rRef+                atomicModifyIORef' sumRef $ \ x -> (x + sum r, ())++        forkIO $ consumer+        forkIO $ consumer+        forkIO $ consumer+        forkIO $ producter+        forkIO $ producter++        threadDelay 5000000++        s <- readIORef sumRef+        s @?= (sum content * 2)++    it "TBQueueNode works as expected" $ do+        let content = [1..1000]++        (sink, src) <- newTBQueueNode 2 10+        sumRef <- newIORef 0++        let producter = do+                src' <- sourceListWithDelay content+                runBIO (src' >|> sink)++        let consumer =  do+                (rRef, sink') <- sinkToList+                runBIO (src >|> sink')+                r <- readIORef rRef+                atomicModifyIORef' sumRef $ \ x -> (x + sum r, ())++        forkIO $ consumer+        forkIO $ consumer+        forkIO $ consumer+        forkIO $ producter+        forkIO $ producter++        threadDelay 5000000++        s <- readIORef sumRef+        s @?= (sum content * 2)++    it "TChanNode works as expected" $ do+        let content = [1..1000]++        (sink, srcf) <- newBroadcastTChanNode 2+        sumRef <- newIORef 0++        let producter = do+                src' <- sourceListWithDelay content+                runBIO (src' >|> sink)++        let consumer =  do+                (rRef, sink') <- sinkToList+                src <- srcf+                runBIO (src >|> sink')+                r <- readIORef rRef+                atomicModifyIORef' sumRef $ \ x -> (x + sum r, ())++        forkIO $ consumer+        forkIO $ consumer+        forkIO $ consumer+        forkIO $ producter+        forkIO $ producter++        threadDelay 5000000++        s <- readIORef sumRef+        s @?= (sum content * 2 * 3)+++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

@@ -0,0 +1,69 @@+{-# LANGUAGE OverloadedStrings #-}++module Z.IO.BIO.ZlibSpec where++import           Control.Monad+import qualified Codec.Compression.Zlib as TheZlib+import           Data.IORef+import           Data.ByteString       as B+import           Data.ByteString.Lazy  as BL+import           Z.Data.Vector         as V+import           Z.IO.BIO.Zlib+import           Z.IO+import           Test.QuickCheck+import           Test.QuickCheck.Function+import           Test.QuickCheck.Property+import           Test.Hspec+import           Test.Hspec.QuickCheck+import           Test.HUnit+++spec :: Spec+spec = describe "zlib" $ do++    describe "decompress . compress === id" . modifyMaxSize (*10) $ do+        prop "decompress . compress === id" $ \ xs  -> do+            decompress defaultDecompressConfig+                (compress defaultCompressConfig (V.pack xs)) === V.pack xs++        prop "decompress . compress === id(with dict)" $ \ xs  -> do+            let dict = "aabbccdd"+            decompress defaultDecompressConfig{decompressDictionary = dict}+                (compress defaultCompressConfig{compressDictionary = dict} (V.pack xs))+                    === V.pack xs++        prop "compress === TheZlib.compress" $ \ xs  -> do+            V.unpack (compress defaultCompressConfig (V.pack xs)) ===+                BL.unpack (TheZlib.compressWith TheZlib.defaultCompressParams (BL.pack xs))++        prop "compress === TheZlib.compress(with dict)" $ \ xs  -> do+            V.unpack (compress defaultCompressConfig{compressDictionary = "aabbccdd"} (V.pack xs)) ===+                BL.unpack (TheZlib.compressWith+                    TheZlib.defaultCompressParams{TheZlib.compressDictionary = Just "aabbccdd" }+                        (BL.pack xs))++        prop "TheZlib.decompress . TheZlib.compress == id(performace benchmark)" $ \ xss -> do+            let vs = Prelude.map B.pack xss+                vs' = TheZlib.decompress . TheZlib.compress $ BL.fromChunks vs++            B.concat vs @=? BL.toStrict vs'++        prop "compress >|> decompress" $ \ xss -> do+            (_, c) <- newCompress defaultCompressConfig+            (_, d) <- newDecompress defaultDecompressConfig++            let vs = Prelude.map V.pack xss+            vs' <- runBlocks (c >|> d) vs++            V.concat vs @=? V.concat vs'++        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++            V.concat vs @=? V.concat vs'

@@ -0,0 +1,44 @@+{-# LANGUAGE OverloadedStrings #-}++module Z.IO.BIOSpec where++import           Control.Monad+import qualified Codec.Compression.Zlib as TheZlib+import           Data.IORef+import qualified Z.Data.Vector         as V+import           Z.IO.BIO.Zlib+import           Z.IO+import           Test.QuickCheck+import           Test.QuickCheck.Function+import           Test.QuickCheck.Property+import           Test.Hspec+import           Test.Hspec.QuickCheck+import           Test.HUnit+import           System.IO.Unsafe+++spec :: Spec+spec = describe "BIO" . modifyMaxSize (*10) $ do++    describe "decode . encode === id(Base64)" $+        prop "Base64" $ \ xs ->+            let r = unsafePerformIO $ do+                    src <- sourceFromList xs+                    (rRef, sink) <- sinkToList+                    enc <- newBase64Encoder+                    dec <- newBase64Decoder+                    runBIO $ src >|> enc >|> dec >|> sink+                    readIORef 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+                    (rRef, sink) <- sinkToList+                    let enc = hexEncoder upper+                    dec <- newHexDecoder+                    runBIO $ src >|> enc >|> dec >|> sink+                    readIORef rRef+            in V.concat r === V.concat xs+

@@ -38,12 +38,12 @@             filename = "test-file"          it "Opens and writes a file" $ do-            withResource (initFileT filename flags mode) $ \ file -> do+            withResource (initFile filename flags mode) $ \ file -> do                 o <- newBufferedOutput' 4096 file                 writeBuffer o content                 flushBuffer o -            withResource (initFileT filename flags mode) $ \ file -> do+            withResource (initFile filename flags mode) $ \ file -> do                 i <- newBufferedInput' 4096 file                 written <- readExactly size i                 written @?= content@@ -56,17 +56,17 @@             unlink filename          it "Opens and writes a file II" $ do-            withResource (initFileT filename flags mode) $ \ file -> do+            withResource (initFile filename flags mode) $ \ file -> do                 o <- newBufferedOutput' 4096 file                 writeBuffer o content2                 flushBuffer o -            withResource (initFileT filename flags mode) $ \ file -> do+            withResource (initFile filename flags mode) $ \ file -> do                 i <- newBufferedInput' 4096 file                 written <- readExactly size2 i                 written @=? content2 -            withResource (initFileT filename flags mode) $ \ file -> do+            withResource (initFile filename flags mode) $ \ file -> do                 i <- newBufferedInput' 4096 file                 Just firstLine <- readLine i                 firstLine  @=? fst (V.break (== V.c2w '\n') content2)@@ -96,7 +96,7 @@          it "link stat should be equal to target file" $ ( do -            withResource (initFileT filename flags mode) $ \ file -> return ()+            withResource (initFile filename flags mode) $ \ file -> return ()              s0 <- stat filename @@ -114,7 +114,7 @@             s0 @=? s2 {stNlink = 1, stCtim = stCtim s0}             s0 @=? s2' {stNlink = 1, stCtim = stCtim s0} -            withResource (initFileT filename flags mode) $ \ file -> do+            withResource (initFile filename flags mode) $ \ file -> do                 s4 <- fstat file                 s0 @?= s4 {stNlink = 1, stCtim = stCtim s0}             ) `finally` ( do@@ -125,7 +125,7 @@             )          it "utime result in stat change" $ do-            withResource (initFileT filename flags mode) $ \ file -> return ()+            withResource (initFile filename flags mode) $ \ file -> return ()             utime filename 1000.2000 3000.4000             s <- stat filename             print s@@ -136,7 +136,7 @@             unlink filename          it "futime result in fstat change" $ do-            withResource (initFileT filename flags mode) $ \ file -> do+            withResource (initFile filename flags mode) $ \ file -> do                 futime file 5000.6000 7000.8000                 s <- fstat file                 print s

@@ -208,7 +208,7 @@        * for modifications. Repurpose the filename for API compatibility.        * I'm not convinced this is a good thing, maybe it should go.        */-      path = e->len ? (const char*) (e + 1) : uv__basename_r(w->path);+      path = e->len ? (const char*) (e + 1) : ""; // uv__basename_r(w->path);        /* We're about to iterate over the queue and call user's callbacks.        * What can go wrong?