diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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`. diff --git a/Z-IO.cabal b/Z-IO.cabal --- a/Z-IO.cabal +++ b/Z-IO.cabal @@ -1,6 +1,6 @@ cabal-version: 2.4 name: Z-IO -version: 0.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.* diff --git a/Z/IO.hs b/Z/IO.hs --- a/Z/IO.hs +++ b/Z/IO.hs @@ -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 diff --git a/Z/IO/BIO.hs b/Z/IO/BIO.hs --- a/Z/IO/BIO.hs +++ b/Z/IO/BIO.hs @@ -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. diff --git a/Z/IO/BIO/Concurrent.hs b/Z/IO/BIO/Concurrent.hs --- a/Z/IO/BIO/Concurrent.hs +++ b/Z/IO/BIO/Concurrent.hs @@ -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 diff --git a/Z/IO/BIO/Zlib.hsc b/Z/IO/BIO/Zlib.hsc --- a/Z/IO/BIO/Zlib.hsc +++ b/Z/IO/BIO/Zlib.hsc @@ -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 bindings using 'BIO' interface. +This module provides 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) diff --git a/Z/IO/Buffered.hs b/Z/IO/Buffered.hs --- a/Z/IO/Buffered.hs +++ b/Z/IO/Buffered.hs @@ -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 diff --git a/Z/IO/FileSystem.hs b/Z/IO/FileSystem.hs --- a/Z/IO/FileSystem.hs +++ b/Z/IO/FileSystem.hs @@ -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 diff --git a/Z/IO/FileSystem/FilePath.hsc b/Z/IO/FileSystem/FilePath.hsc --- a/Z/IO/FileSystem/FilePath.hsc +++ b/Z/IO/FileSystem/FilePath.hsc @@ -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 diff --git a/Z/IO/FileSystem/Threaded.hs b/Z/IO/FileSystem/Threaded.hs --- a/Z/IO/FileSystem/Threaded.hs +++ b/Z/IO/FileSystem/Threaded.hs @@ -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 diff --git a/Z/IO/FileSystem/Watch.hs b/Z/IO/FileSystem/Watch.hs --- a/Z/IO/FileSystem/Watch.hs +++ b/Z/IO/FileSystem/Watch.hs @@ -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 diff --git a/Z/IO/Process.hsc b/Z/IO/Process.hsc --- a/Z/IO/Process.hsc +++ b/Z/IO/Process.hsc @@ -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 diff --git a/Z/IO/StdStream.hs b/Z/IO/StdStream.hs --- a/Z/IO/StdStream.hs +++ b/Z/IO/StdStream.hs @@ -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 diff --git a/cbits/hs_uv_fs_event.c b/cbits/hs_uv_fs_event.c --- a/cbits/hs_uv_fs_event.c +++ b/cbits/hs_uv_fs_event.c @@ -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); + } } } diff --git a/test/Z/IO/Network/TCPSpec.hs b/test/Z/IO/Network/TCPSpec.hs --- a/test/Z/IO/Network/TCPSpec.hs +++ b/test/Z/IO/Network/TCPSpec.hs @@ -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