Z-IO 0.1.8.0 → 0.1.8.1
raw patch · 15 files changed
+104/−70 lines, 15 filesPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
API changes (from Hackage documentation)
+ Z.IO.BIO: ioBIO :: (HasCallStack => a -> IO b) -> BIO a b
+ Z.IO.BIO: pureBIO :: (a -> b) -> BIO a b
+ Z.IO.BIO.Zlib: data ZStream
- Z.IO.BIO: (>!>) :: BIO a b -> (b -> IO c) -> BIO a c
+ Z.IO.BIO: (>!>) :: BIO a b -> (HasCallStack => b -> IO c) -> BIO a c
Files
- ChangeLog.md +4/−0
- Z-IO.cabal +3/−2
- Z/IO.hs +1/−1
- Z/IO/BIO.hs +19/−7
- Z/IO/BIO/Concurrent.hs +1/−1
- Z/IO/BIO/Zlib.hsc +21/−17
- Z/IO/Buffered.hs +1/−1
- Z/IO/FileSystem.hs +1/−1
- Z/IO/FileSystem/FilePath.hsc +1/−1
- Z/IO/FileSystem/Threaded.hs +1/−1
- Z/IO/FileSystem/Watch.hs +28/−26
- Z/IO/Process.hsc +1/−1
- Z/IO/StdStream.hs +1/−1
- cbits/hs_uv_fs_event.c +19/−8
- test/Z/IO/Network/TCPSpec.hs +2/−2
ChangeLog.md view
@@ -1,5 +1,9 @@ # Revision history for Z-IO +## 0.1.8.1 -- 2020-10-24++* Export `ZStream` type from `Z.IO.BIO.Zlib`+ ## 0.1.8.0 -- 2020-10-24 * Remove type index from `BufferedInput`, `BufferedOutput`.
Z-IO.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.4 name: Z-IO-version: 0.1.8.0+version: 0.1.8.1 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!@@ -251,7 +251,7 @@ third_party/libuv/src/win/winsock.c cc-options: -Wall -DWIN32_LEAN_AND_MEAN -D_WIN32_WINNT=0x0600 -march=native- cpp-options: -Wall -DWIN32_LEAN_AND_MEAN -D_WIN32_WINNT=0x0600+ cpp-options: -DWIN32_LEAN_AND_MEAN -D_WIN32_WINNT=0x0600 include-dirs: third_party/libuv/include third_party/libuv/src -- The C runtime dependencies are imposed by libuv.@@ -383,3 +383,4 @@ ghc-options: -threaded default-language: Haskell2010+ build-tool-depends: hspec-discover:hspec-discover == 2.*
Z/IO.hs view
@@ -1,7 +1,7 @@ {-| Module : Z.IO Description : IO Umbrella module-Copyright : (c) Dong Han, 2017+Copyright : (c) Dong Han, 2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental
Z/IO/BIO.hs view
@@ -2,7 +2,7 @@ {-| Module : Z.IO.BIO Description : Buffered IO interface-Copyright : (c) Dong Han, 2017-2018+Copyright : (c) Dong Han, 2017-2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental@@ -50,6 +50,7 @@ , (>|>), (>~>), (>!>), appendSource , concatSource, zipSource , joinSink, fuseSink+ , pureBIO, ioBIO , ParseException(..) -- * Run BIO chain , runBIO@@ -130,8 +131,8 @@ -- -- 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.+-- * You shouldn't use a 'BIO' node across multiple 'BIO' chain unless the state can be reset.+-- * You shouldn't use a 'BIO' node across multiple threads unless document states otherwise. -- -- 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.@@ -193,7 +194,7 @@ (>~>) = flip fmap -- | Connect BIO to an effectful function.-(>!>) :: BIO a b -> (b -> IO c) -> BIO a c+(>!>) :: BIO a b -> (HasCallStack => b -> IO c) -> BIO a c {-# INLINE (>!>) #-} (>!>) BIO{..} f = BIO push_ pull_ where@@ -205,9 +206,18 @@ _ -> return Nothing -- | BIO node from a pure function.+--+-- BIO node made with this funtion are stateless, thus can be reused across chains. pureBIO :: (a -> b) -> BIO a b-pureBIO f = BIO (\ x -> return (Just (f x))) (return Nothing)+pureBIO f = BIO (\ x -> let !r = f x in return (Just r)) (return Nothing) +-- | BIO node from an IO function.+--+-- BIO node made with this funtion may not be stateless, it depends on if the IO function use+-- IO state.+ioBIO :: (HasCallStack => a -> IO b) -> BIO a b+ioBIO f = BIO (\ x -> Just <$!> f x) (return Nothing)+ -- | Connect two 'BIO' source, after first reach EOF, draw element from second. appendSource :: Source a -> Source a -> IO (Source a) {-# INLINE appendSource #-}@@ -685,7 +695,9 @@ re <- newReChunk 4 return (re >~> base64Decode') --- | Make a new hex encoder node.+-- | Make a hex encoder node.+--+-- Hex encoder is stateless, it can be reused across chains. hexEncoder :: Bool -- ^ uppercase? -> BIO V.Bytes V.Bytes {-# INLINABLE hexEncoder #-}@@ -721,7 +733,7 @@ return (c, BIO (push_ c) (return Nothing)) where push_ c x = do- i <- atomicAddCounter c 1+ !i <- atomicAddCounter c 1 return (Just (i, x)) -- | Make a BIO node grouping items into fixed size arrays.
Z/IO/BIO/Concurrent.hs view
@@ -4,7 +4,7 @@ {-| Module : Z.IO.BIO.Concurrent Description : Base64 codec-Copyright : (c) Dong Han, 2017-2018+Copyright : (c) Dong Han, 2017-2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental
Z/IO/BIO/Zlib.hsc view
@@ -1,33 +1,36 @@ {-| Module : Z.IO.BIO.Zlib-Description : The zlib-Copyright : (c) Dong Han, 2017-2018+Description : The zlib binding+Copyright : (c) Dong Han, 2017-2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental Portability : non-portable -This module provides <https://zlib.net zlib> bindings using 'BIO' interface.+This module provides <https://zlib.net zlib> bindings, with 'BIO' streaming interface. -} module Z.IO.BIO.Zlib( -- * Compression- CompressConfig(..)- , defaultCompressConfig- , newCompress, compressReset+ newCompress, compressReset , compress , compressBlocks- , WindowBits- , defaultWindowBits- , MemLevel- , defaultMemLevel+ , ZStream+ , CompressConfig(..)+ , defaultCompressConfig -- * Decompression- , DecompressConfig(..)- , defaultDecompressConfig , newDecompress, decompressReset , decompress , decompressBlocks+ , DecompressConfig(..)+ , defaultDecompressConfig -- * Constants+ -- ** Windows bits+ , WindowBits+ , defaultWindowBits+ -- ** Memory level+ , MemLevel+ , defaultMemLevel -- ** Strategy , Strategy , pattern Z_FILTERED@@ -114,9 +117,10 @@ CompressConfig Z_DEFAULT_COMPRESSION defaultWindowBits defaultMemLevel V.empty Z_DEFAULT_STRATEGY V.defaultChunkSize +-- | A foreign pointer to a zlib\'s @z_stream_s@ struct. data ZStream = ZStream (ForeignPtr ZStream) (IORef Bool) --- | Compress all the data written to a output.+-- | Make a new compress node. -- -- The returned 'BIO' node can be reused only if you call 'compressReset' on the 'ZStream'. newCompress :: HasCallStack@@ -187,11 +191,11 @@ throwZlibIfMinus_ (withForeignPtr fp deflateReset) writeIORef finRef False --- | Decompress some bytes.+-- | Compress some bytes. compress :: HasCallStack => CompressConfig -> V.Bytes -> V.Bytes compress conf = V.concat . unsafeRunBlock (snd <$> newCompress conf) --- | Decompress some bytes list.+-- | Compress some bytes in blocks. compressBlocks :: HasCallStack => CompressConfig -> [V.Bytes] -> [V.Bytes] compressBlocks conf = unsafeRunBlocks (snd <$> newCompress conf) @@ -205,7 +209,7 @@ defaultDecompressConfig :: DecompressConfig defaultDecompressConfig = DecompressConfig defaultWindowBits V.empty V.defaultChunkSize --- | Decompress bytes from source.+-- | Make a new decompress node. -- -- The returned 'BIO' node can be reused only if you call 'decompressReset' on the 'ZStream'. newDecompress :: DecompressConfig -> IO (ZStream, BIO V.Bytes V.Bytes)@@ -288,7 +292,7 @@ decompress :: HasCallStack => DecompressConfig -> V.Bytes -> V.Bytes decompress conf = V.concat . unsafeRunBlock (snd <$> newDecompress conf) --- | Decompress some bytes list.+-- | Decompress some bytes in blocks. decompressBlocks :: HasCallStack => DecompressConfig -> [V.Bytes] -> [V.Bytes] decompressBlocks conf = unsafeRunBlocks (snd <$> newDecompress conf)
Z/IO/Buffered.hs view
@@ -1,7 +1,7 @@ {-| Module : Z.IO.Buffered Description : Buffered IO interface-Copyright : (c) Dong Han, 2017-2018+Copyright : (c) Dong Han, 2017-2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental
Z/IO/FileSystem.hs view
@@ -1,7 +1,7 @@ {-| Module : Z.IO.FileSystem Description : Filesystem IO-Copyright : (c) Dong Han, 2017~2019+Copyright : (c) Dong Han, 2017-2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental
Z/IO/FileSystem/FilePath.hsc view
@@ -1,7 +1,7 @@ {-| Module : Z.IO.FileSystem.FilePath Description : file path toolbox-Copyright : (c) Dong Han, 2017~2019+Copyright : (c) Dong Han, 2017-2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental
Z/IO/FileSystem/Threaded.hs view
@@ -1,7 +1,7 @@ {-| Module : Z.IO.FileSystem.Threaded Description : Filesystem IO using threadpool-Copyright : (c) Dong Han, 2017~2019+Copyright : (c) Dong Han, 2017-2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental
Z/IO/FileSystem/Watch.hs view
@@ -1,7 +1,7 @@ {-| Module : Z.IO.FileSystem.Watch Description : cross-platform recursive fs watcher-Copyright : (c) Dong Han, 2017~2019+Copyright : (c) Dong Han, 2017-2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental@@ -84,14 +84,12 @@ 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+ tid <- forkIO $ watchThread mRef dir' sink modifyMVar_ mRef $ \ m -> return $! HM.insert dir' tid m) `onException` cleanUpWatcher mRef sink return (cleanUpWatcher mRef sink, srcf)@@ -103,7 +101,9 @@ forM_ m killThread void (pull sink) - watchThread mRef eRef dir sink = do+ watchThread mRef dir sink = do+ -- IORef store temp events to de-duplicated+ eRef <- newIORef Nothing uvm <- getUVManager (bracket (do withUVManager uvm $ \ loop -> do@@ -164,29 +164,31 @@ 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+ processEvent pdir mRef eRef sink = mapM_ $ \ (e, path) ->+ -- don't report event about directory itself, it will reported by its parent+ unless (CBytes.null 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)+ 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 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 (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
Z/IO/Process.hsc view
@@ -1,7 +1,7 @@ {-| Module : Z.IO.Process Description : Process utilities-Copyright : (c) Dong Han, 2018~2019+Copyright : (c) Dong Han, 2018-2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental
Z/IO/StdStream.hs view
@@ -1,7 +1,7 @@ {-| Module : Z.IO.StdStream Description : Standard Streams and TTY devices-Copyright : (c) Dong Han, 2018~2019+Copyright : (c) Dong Han, 2018-2020 License : BSD Maintainer : winterland1989@gmail.com Stability : experimental
cbits/hs_uv_fs_event.c view
@@ -36,14 +36,25 @@ 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);+ // OSX fsevent backend filename could be NULL+ if (filename == NULL){+ if (loop_data->buffer_size_table[slot] >= 2){+ loop_data->buffer_size_table[slot] -= 2;+ char* buf = loop_data->buffer_table[slot] + loop_data->buffer_size_table[slot];+ *buf = (uint8_t)events;+ // empty filename will be filtered on Haskell side+ *(buf+1) = 0; + }+ } else {+ 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);+ } } }
test/Z/IO/Network/TCPSpec.hs view
@@ -26,7 +26,7 @@ serverThread <- forkIO $ startTCPServer defaultTCPServerConfig{ tcpListenAddr = addr } echo - threadDelay 100000+ threadDelay 2000000 -- 2s replicateM_ 512 . forkIO $ withResource (initTCPClient defaultTCPClientConfig{tcpRemoteAddr = addr}) $ \ tcp -> do@@ -41,5 +41,5 @@ longMsg' <- readExactly' (V.length longMsg) i longMsg' @=? longMsg - threadDelay 5000000+ threadDelay 5000000 -- 5s killThread serverThread