Z-IO-1.0.0.0: Z/IO/FileSystem/Watch.hs
{-|
Module : Z.IO.FileSystem.Watch
Description : cross-platform recursive fs watcher
Copyright : (c) Dong Han, 2017-2020
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, cleanup watching threads automatically when finished.
withResource (initWatchDirs ["fold_to_be_watch"] True) $ \ srcf -> do
-- dup a file event source
src <- srcf
-- print event to stdout
BIO.run_ $ src . sinkToIO printStd
@
-}
module Z.IO.FileSystem.Watch
( FileEvent(..)
, watchDirs
, initWatchDirs
) where
import Control.Concurrent
import Control.Monad
import Data.Bits
import qualified Data.HashMap.Strict as HM
import Data.IORef
#if defined(linux_HOST_OS)
import qualified Data.List as List
#endif
import Data.Primitive.PrimArray
import Data.Word
import GHC.Generics
import Z.Data.Array.Unaligned
import Z.Data.CBytes (CBytes)
import qualified Z.Data.CBytes as CBytes
import Z.Data.JSON (JSON)
import Z.Data.Text.Print (Print)
import Z.Data.Vector (defaultChunkSize)
import Z.Foreign
import Z.IO.BIO as BIO
import Z.IO.Exception
import Z.IO.FileSystem.Base
import qualified Z.IO.FileSystem.FilePath as P
import Z.IO.LowResTimer
import Z.IO.Resource
import Z.IO.UV.FFI
import Z.IO.UV.Manager
-- | File event with path info.
data FileEvent = FileAdd CBytes | FileRemove CBytes | FileModify CBytes
deriving (Show, Read, Ord, Eq, Generic)
deriving anyclass (Print, JSON)
-- | Watching a list of given directories.
watchDirs :: [CBytes] -- ^ Directories to be watched
-> Bool -- ^ recursively watch?
-> (FileEvent -> IO ()) -- ^ Callback function to handle 'FileEvent'
-> IO ()
{-# INLINABLE watchDirs #-}
watchDirs dirs rec callback = do
withResource (initWatchDirs dirs rec) $ \ srcf -> do
src <- srcf
run_ $ src . sinkToIO callback
-- | Start watching a list of given directories, stream version.
initWatchDirs :: [CBytes] -- ^ watching list
-> Bool -- ^ recursively watch?
-> Resource (IO (Source FileEvent))
{-# INLINABLE initWatchDirs #-}
initWatchDirs dirs False = do
liftIO . forM_ dirs $ \ dir -> do
b <- isDir dir
unless b (throwUVIfMinus_ (return UV_ENOTDIR))
watch_ 0 dirs
initWatchDirs dirs _ = do
#if defined(linux_HOST_OS)
-- inotify doesn't support recursive watch, so we manually maintain watch list
subDirs <- liftIO . 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] -> Resource (IO (Source FileEvent))
{-# INLINABLE watch_ #-}
watch_ flag dirs = fst <$> initResource (do
-- HashMap to store all watchers
mRef <- newMVar HM.empty
-- there's only one place to pull the sink, that is cleanUpWatcher
(sink, srcf) <- newBroadcastTChanPair 1
-- lock UVManager first
(forM_ dirs $ \ dir -> do
dir' <- P.normalize dir
tid <- forkIO $ watchThread mRef dir' sink
modifyMVar_ mRef $ \ m ->
return $! HM.insert dir' tid m) `onException` cleanUpWatcher mRef sink
return (srcf, (sink, mRef)))
(\ (_, (sink, mRef)) -> cleanUpWatcher mRef sink)
where
eventBufSiz = defaultChunkSize
cleanUpWatcher mRef sink = do
m <- takeMVar mRef
forM_ m killThread
void (sink discard EOF)
watchThread mRef dir sink = do
-- IORef store temp events to de-duplicated
eRef <- newIORef Nothing
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 = CBytes.indexBACBytes 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) ->
-- 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 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' (BIO.step_ 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' (BIO.step_ sink)