@@ -99,6 +99,8 @@ module Streamly.Internal.FileSystem.Event.Linux
9999 , setAllEvents
100100
101101 -- ** Watch APIs
102+ -- XXX watchPaths is redundant now because we can use watchTrees with
103+ -- setRecursiveMode False. Perhaps we can use a common "watch" API.
102104 , watchPathsWith
103105 , watchPaths
104106 , watchTreesWith
@@ -107,7 +109,7 @@ module Streamly.Internal.FileSystem.Event.Linux
107109 , removeFromWatch
108110
109111 -- * Handling Events
110− , Event
112+ , Event(..)
111113 , getRoot
112114 , getRelPath
113115 , getCookie
@@ -142,19 +144,17 @@ module Streamly.Internal.FileSystem.Event.Linux
142144
143145 -- * Debugging
144146 , showEvent
145− , showEventShort
146147 )
147148where
148149
149150import Control.Monad (void, when)
150151import Control.Monad.IO.Class (MonadIO, liftIO)
151−import Data.Bifunctor (bimap)
152152import Data.Bits ((.|.), (.&.), complement)
153+import Data.Char (ord)
153154import Data.Foldable (foldlM)
154155import Data.Functor.Identity (runIdentity)
155156import Data.IntMap.Lazy (IntMap)
156−import Data.IORef (IORef, newIORef, readIORef, writeIORef)
157−import Data.Function ((&))
157+import Data.IORef (IORef, newIORef, readIORef, writeIORef, modifyIORef)
158158import Data.List.NonEmpty (NonEmpty)
159159import Data.Word (Word8, Word32)
160160import Foreign.C.Error (throwErrnoIfMinus1)
@@ -168,7 +168,6 @@ import GHC.IO.Handle.FD (mkHandleFromFD)
168168import Streamly.Prelude (SerialT)
169169import Streamly.Internal.Data.Parser (Parser)
170170import Streamly.Internal.Data.Array.Storable.Foreign.Types (Array(..))
171−import System.FilePath ((</>))
172171import System.IO (Handle, hClose, IOMode(ReadMode))
173172#if !MIN_VERSION_base(4,10,0)
174173import Control.Concurrent.MVar (readMVar)
@@ -201,7 +200,7 @@ import qualified Streamly.Internal.Unicode.Stream as U
201200--
202201data Config = Config
203202 { watchRec :: Bool
204− , createFlags :: Word32
203+ , createFlags :: Word32
205204 }
206205
207206-------------------------------------------------------------------------------
@@ -229,56 +228,19 @@ setFlag mask status cfg@Config{..} =
229228-------------------------------------------------------------------------------
230229-- Utilities
231230-------------------------------------------------------------------------------
232−splitPath :: (MonadIO m) => FilePath -> FilePath -> m [FilePath]
233−splitPath pat xs = S.toList
234− $ S.splitOnSeq (A.fromList pat) (FL.toList) (S.fromList xs)
235−
236−pathDiff :: (MonadIO m) => FilePath -> FilePath -> m FilePath
237−pathDiff pat xs = do
238− splits <- splitPath pat xs
239− let j = case length splits of
240− 0 -> ""
241− _ -> last splits
242− t = case length j of
243− 0 -> ""
244− _ -> tail j
245− return t
246−
247−dirListing :: String -> SerialT IO (Either String String)
248−dirListing s = do
249− S.filter dirFilter
250− $ S.iterateMapLeftsWith S.ahead listDir (S.yield $ (Left s))
251231
252− where
253−
254− dirFilter x =
255− case x of
256− Left _ -> True
257− Right _ -> False
258−
259− listDir dir =
260− Dir.toEither dir -- SerialT IO (Either String String)
261− & S.map (bimap prefix prefix) -- SerialT IO (Either String String)
262−
263− where
264−
265− prefix x = dir ++ "/" ++ x
266−
267−toPathList :: String -> IO [String]
268−toPathList root = do
269− S.toList $ S.map toStr $ dirListing root
270−
271− where
272−
273− toStr x =
274− case x of
275− Left d -> d
276− Right d -> d
232+-- XXX we really do not know the path encoding, all we know is that it is "/"
233+-- separated bytes. So these may fail or convert the path in an unexpected
234+-- manner. We should ultimately remove all usage of these.
277235
278236toUtf8 :: MonadIO m => String -> m (Array Word8)
279237toUtf8 = A.fromStream . U.encodeUtf8 . S.fromList
280238
281−-- | Set watch event on directory recursively.
239+utf8ToString :: Array Word8 -> String
240+utf8ToString = runIdentity . S.toList . U.decodeUtf8' . A.toStream
241+
242+-- | Watch the whole directory tree recursively instead of watching just one
243+-- level of directory.
282244--
283245-- /default: Off/
284246--
@@ -571,14 +533,23 @@ defaultConfig :: Config
571533defaultConfig =
572534 setWhenExists AddIfExists
573535 $ setAllEvents On
574− $ Config {watchRec = False, createFlags = 0}
536+ $ Config {watchRec = True, createFlags = 0}
575537
576538-------------------------------------------------------------------------------
577539-- Open an event stream
578540-------------------------------------------------------------------------------
579541
580542-- | A handle for a watch.
581−data Watch = Watch Handle (IORef (IntMap (Array Word8, Array Word8)))
543+data Watch =
544+ Watch
545+ Handle -- File handle for the watch
546+ (IORef
547+ (IntMap -- Key is the watch descriptor
548+ ( Array Word8 -- Absolute path of the watch root
549+ , Array Word8 -- Path of subdir relative to watch root
550+ )
551+ )
552+ )
582553
583554-- Instead of using the watch descriptor we can provide APIs that use the path
584555-- itself to identify the watch. That will require us to maintain a map from wd
@@ -615,15 +586,13 @@ createWatch = do
615586 ReadMode
616587 True -- use non-blocking IO
617588 Nothing -- TextEncoding (binary)
618− emptyMapRef <- newIORef Map.empty
589+ emptyMapRef <- newIORef Map.empty
619590 return $ Watch h emptyMapRef
620591
621592foreign import ccall unsafe
622593 "sys/inotify.h inotify_add_watch" c_inotify_add_watch
623594 :: CInt -> CString -> CUInt -> IO CInt
624595
625−utf8ToString :: Array Word8 -> String
626−utf8ToString = runIdentity . S.toList . U.decodeUtf8' . A.toStream
627596
628597#if !MIN_VERSION_base(4,10,0)
629598-- | Turn an existing Handle into a file descriptor. This function throws an
@@ -643,72 +612,105 @@ handleToFd h = case h of
643612 InappropriateType "handleToFd" msg Nothing Nothing
644613#endif
645614
646−-- | Given a 'Config' and a @path@ supplied as a UTF-8 encoded byte array, add
647−-- the path to the list of paths being monitored by the given watch 'Handle'.
648−-- Returns a "watch descriptor" which is used as an idenitfier in the events
649−-- generated for this particular path and also to remove the path from this
650−-- watch.
615+-- | Add a trailing "/" at the end of the path if there is none. Do not add a
616+-- "/" if the path is empty.
617+--
618+ensureTrailingSlash :: Array Word8 -> Array Word8
619+ensureTrailingSlash path =
620+ if A.length path /= 0
621+ then
622+ let mx = A.readIndex path (A.length path - 1)
623+ in case mx of
624+ Nothing -> error "ensureTrailingSlash: Bug: Invalid index"
625+ Just x ->
626+ if x /= fromIntegral (ord '/')
627+ then path <> A.fromCString# "/"#
628+ else path
629+ else path
630+
631+-- | @addToWatch cfg watch root subpath@ adds @subpath@ to the list of paths
632+-- being monitored under @root@ via the watch handle @watch@. @root@ must be
633+-- an absolute path and @subpath@ must be relative to @root@.
651634--
652635-- /Internal/
653636--
654−addToWatchExecutor ::
655− Config
656− -> Watch
657− -> Array Word8
658− -> Array Word8
659− -> IO (Config, Watch)
660−addToWatchExecutor cfg@Config{..} (Watch handle wdMap) pathArr root = do
637+addToWatch :: Config -> Watch -> Array Word8 -> Array Word8 -> IO ()
638+addToWatch cfg@Config{..} watch@(Watch handle wdMap) root0 path0 = do
639+ -- XXX do not add if the path is already added
640+ -- XXX if the watch is added by the scan and not via an event we can
641+ -- generate a create event assuming that the create may have been lost. We
642+ -- can also mark in the map that this entry was added by the scan. So if an
643+ -- actual create event later comes and tries to add this again then we can
644+ -- ignore that and drop the create event to avoid duplicate create, because
645+ -- we have already emitted it.
646+ --
647+ -- When a directory is added by the scan we should also emit create events
648+ -- for files that may have got added to the dir. However, such create
649+ -- events may get duplicated because of a race between the scan generated
650+ -- versus real events.
651+ --
652+ -- Or we may distinguish between scan generated events and real events so
653+ -- that the application can assume that other events may been lost and
654+ -- handle it. For example, if it is a dir create the application can read
655+ -- the dir to scan the files in it.
656+ --
657+ let root = ensureTrailingSlash root0
658+ path = ensureTrailingSlash path0
659+ absPath = root <> path
661660 fd <- handleToFd handle
662− wd <- A.asCString pathArr $ \pathPtr ->
663− throwErrnoIfMinus1 ("addToWatch " ++ utf8ToString pathArr) $
664− c_inotify_add_watch (fdFD fd) pathPtr (CUInt createFlags)
665− km <- readIORef wdMap
666− let k = (Map.insert (fromIntegral wd) (root, pathArr) km)
667− writeIORef wdMap k
668− return $ (cfg, Watch handle wdMap)
669661
670−addToWatch :: Config -> Watch -> Array Word8 -> IO (Config, Watch)
671−addToWatch cfg@Config{..} (Watch handle wdMap) path = do
672− fd <- handleToFd handle
673− wd <- A.asCString path $ \pathPtr ->
674− throwErrnoIfMinus1 ("addToWatch " ++ utf8ToString path) $
662+ -- XXX we need to tolerate an error where we are adding a watch for a
663+ -- non-existing file because the file may have got deleted by the time we
664+ -- added the watch. Perhaps we can have a flag in config for this and keep
665+ -- the default value to tolerate the error.
666+ --
667+ -- XXX The file may have even got deleted and then recreated which we will
668+ -- never get to know, document this.
669+ wd <- A.asCString absPath $ \pathPtr ->
670+ throwErrnoIfMinus1 ("addToWatch: " ++ utf8ToString absPath) $
675671 c_inotify_add_watch (fdFD fd) pathPtr (CUInt createFlags)
676− km <- readIORef wdMap
677− let k = (Map.insert (fromIntegral wd) (path, path) km)
678− writeIORef wdMap k
679− return $ (cfg, Watch handle wdMap)
680−
681−addToWatchRec :: Config -> Watch -> FilePath -> FilePath -> IO (Config, Watch)
682−addToWatchRec cfg watch path origRoot= do
683− subdirs <- toPathList path
684− pathArrList <- mapM toUtf8 subdirs
685− rootArr <- toUtf8 origRoot
686− foldlM (\(cfg1, w1) pathArr -> addToWatchExecutor cfg1 w1 pathArr rootArr)
687− (cfg, watch) pathArrList
672+
673+ -- We add the parent first so that we start getting events for any new
674+ -- creates and add the new subdirectories on creates while we are adding
675+ -- the children.
676+ modifyIORef wdMap (Map.insert (fromIntegral wd) (root, path))
677+
678+ -- Now add the children. If we missed any creates while we were adding the
679+ -- parent, this will make sure they are added too.
680+ --
681+ -- XXX Ensure that we generate events that we may have missed while we were
682+ -- adding the dirs.
683+ --
684+ -- XXX toDirs currently uses paths as String, we need to convert it
685+ -- to "/" separated by byte arrays.
686+ when watchRec $ do
687+ S.mapM_ (\p -> addToWatch cfg watch root (path <> p))
688+ $ S.mapM toUtf8
689+ $ Dir.toDirs $ utf8ToString absPath
688690
689691foreign import ccall unsafe
690692 "sys/inotify.h inotify_rm_watch" c_inotify_rm_watch
691693 :: CInt -> CInt -> IO CInt
692694
693−-- | Remove a path from a 'Watch', if a path was moved after adding you need to
694−-- need to provide the original path which was used to add the Watch.
695+-- | Remove an absolute root path from a 'Watch', if a path was moved after
696+-- adding you need to provide the original path which was used to add the
697+-- Watch.
695698--
696699-- /Internal/
697700--
698−removeFromWatch :: Watch -> Array Word8 -> IO Watch
701+removeFromWatch :: Watch -> Array Word8 -> IO ()
699702removeFromWatch (Watch handle wdMap) path = do
700703 fd <- handleToFd handle
701704 km <- readIORef wdMap
702705 wdMap1 <- foldlM (step fd) Map.empty (Map.toList km)
703706 writeIORef wdMap wdMap1
704− return $ Watch handle wdMap
705707
706708 where
707709
708710 step fd newMap (wd, v) = do
709711 if (fst v) == path
710712 then do
711− let err = "removeFromWatch " ++ show (utf8ToString path)
713+ let err = "removeFromWatch: " ++ show (utf8ToString path)
712714 rm = c_inotify_rm_watch (fdFD fd) (fromIntegral wd)
713715 void $ throwErrnoIfMinus1 err rm
714716 return newMap
@@ -720,32 +722,18 @@ removeFromWatch (Watch handle wdMap) path = do
720722--
721723-- /Internal/
722724--
723−openWatch ::
724− Config
725− -> NonEmpty (Array Word8)
726− -> IO (Config, Watch)
725+openWatch :: Config -> NonEmpty (Array Word8) -> IO Watch
727726openWatch cfg paths = do
728− let pathList = NonEmpty.toList paths
729− w <- createWatch
730− foldlM (\(cfg1, w1) path -> addToWatch cfg1 w1 path)
731− (cfg, w) pathList
732−
733−openWatchRec ::
734− Config
735− -> NonEmpty (Array Word8)
736− -> IO (Config, Watch)
737−openWatchRec cfg paths = do
738− let pathList = map utf8ToString $ NonEmpty.toList paths
739727 w <- createWatch
740− foldlM (\(cfg1, w1) path -> addToWatchRec cfg1 w1 path path)
741− (cfg, w) pathList
728+ mapM_ (\p -> addToWatch cfg w p (A.fromList [])) $ NonEmpty.toList paths
729+ return w
742730
743731-- | Close a 'Watch' handle.
744732--
745733-- /Internal/
746734--
747−closeWatch :: (Config, Watch) -> IO ()
748−closeWatch (_, Watch h _) = hClose h
735+closeWatch :: Watch -> IO ()
736+closeWatch (Watch h _) = hClose h
749737
750738-------------------------------------------------------------------------------
751739-- Raw events read from the watch file handle
@@ -788,8 +776,6 @@ readOneEvent cfg wt@(Watch _ wdMap) = do
788776 -- XXX need the "initial" in parsers to return a step type so that "take 0"
789777 -- can return without an input. otherwise if pathLen is 0 we will keep
790778 -- waiting to read one more char before we return this event.
791− -- PR.yieldM $ print ("Flag1 = " ++ show eflags)
792− -- PR.yieldM $ print ("pathLen = " ++ show pathLen)
793779 path <-
794780 if pathLen /= 0
795781 then do
@@ -798,30 +784,33 @@ readOneEvent cfg wt@(Watch _ wdMap) = do
798784 -- takeP
799785 pth <- PR.sliceSepByMax (== 0) pathLen (A.writeN pathLen)
800786 let remaining = pathLen - A.length pth - 1
801− -- PR.yieldM $ print ("remaining = " ++ show remaining)
802787 when (remaining /= 0) $ PR.takeEQ remaining FL.drain
803− -- PR.yieldM $ print ("takeEQ = " ++ show remaining)
804788 return pth
805789 else return $ A.fromList []
806− -- PR.yieldM $ print ("Flag2 = " ++ show eflags)
807− xm <- PR.yieldM $ readIORef wdMap
808− let base = case Map.lookup (fromIntegral ewd) xm of
809− Just path1 -> path1
810− Nothing -> (path, path)
811− let xpath = utf8ToString (snd base) </> utf8ToString path
812− _ <- if eflags .&. iN_CREATE /= 0 && eflags .&. iN_ISDIR /= 0
813− then
814− processEvent cfg wt xpath (utf8ToString $ fst base)
815− else return (cfg, wt)
816− rm2 <- PR.yieldM $ readIORef wdMap
817− pdiff <- PR.yieldM $ pathDiff (utf8ToString $ fst base) xpath
818− relPath <- PR.yieldM $ toUtf8 (pdiff)
790+ wdm <- PR.yieldM $ readIORef wdMap
791+ let (root, sub) =
792+ case Map.lookup (fromIntegral ewd) wdm of
793+ Just pair -> pair
794+ Nothing ->
795+ error $ "readOneEvent: "
796+ <> "Unknown watch descriptor: "
797+ <> show ewd
798+ let -- "sub" is guaranteed to have a trailing "/"
799+ sub1 = sub <> path
800+ -- Check for "ISDIR" first because it is less likely
801+ isDirCreate = eflags .&. iN_ISDIR /= 0 && eflags .&. iN_CREATE /= 0
802+ when (watchRec cfg && isDirCreate)
803+ $ PR.yieldM $ addToWatch cfg wt root sub1
804+ -- XXX Handle IN_DELETE, IN_DELETE_SELF, IN_MOVE_SELF, IN_MOVED_FROM,
805+ -- IN_MOVED_TO
806+ -- What if a large dir tree gets moved in to our hierarchy? Do we get a
807+ -- single event for the top level dir in this case?
819808 return $ Event
820809 { eventWd = (fromIntegral ewd)
821810 , eventFlags = eflags
822811 , eventCookie = cookie
823− , eventRelPath = relPath
824− , eventMap = rm2
812+ , eventRelPath = sub1
813+ , eventMap = wdm
825814 }
826815
827816 where
@@ -834,21 +823,8 @@ readOneEvent cfg wt@(Watch _ wdMap) = do
834823 pathLen :: Word32 <- peekByteOff ptr (len + 8)
835824 return (ewd, eflags, cookie, fromIntegral pathLen)
836825
837− processEvent ::
838− Config
839− -> Watch
840− -> FilePath
841− -> FilePath
842− -> Parser IO Word8 (Config, Watch)
843− processEvent cfg1@Config{..} wt1 xpath root = do
844− if watchRec
845− then
846− PR.yieldM $ addToWatchRec cfg1 wt1 xpath root
847− else
848− PR.yieldM $ return (cfg1, wt1)
849−
850−watchToStream :: (Config, Watch) -> SerialT IO Event
851−watchToStream (cfg, wt@(Watch handle _)) = do
826+watchToStream :: Config -> Watch -> SerialT IO Event
827+watchToStream cfg wt@(Watch handle _) = do
852828 -- Do not use too small a buffer. As per inotify man page:
853829 --
854830 -- The behavior when the buffer given to read(2) is too small to return
@@ -879,12 +855,7 @@ watchToStream (cfg, wt@(Watch handle _)) = do
879855--
880856watchPathsWith ::
881857 (Config -> Config) -> NonEmpty (Array Word8) -> SerialT IO Event
882−watchPathsWith f paths = S.bracket before after watchToStream
883−
884− where
885−
886− before = liftIO $ openWatch (f defaultConfig) paths
887− after = liftIO . closeWatch
858+watchPathsWith f = watchTreesWith (f . setRecursiveMode False)
888859
889860-- | Like 'watchPathsWith' but uses the 'defaultConfig' options.
890861--
@@ -897,22 +868,40 @@ watchPathsWith f paths = S.bracket before after watchToStream
897868watchPaths :: NonEmpty (Array Word8) -> SerialT IO Event
898869watchPaths = watchPathsWith id
899870
871+-- XXX We should not go across the mount points of network file systems or file
872+-- systems that are known to not generate any events.
873+--
900874-- | Start monitoring a list of file system paths for file system events with
901875-- the supplied configuration operation over the 'defaultConfig'. The
902876-- paths could be files or directories. When the path is a directory, the
903877-- whole directory tree under it is watched recursively. Monitoring starts from
904878-- the current time onwards.
905879--
880+-- Note that recrusive watch on a large directory tree could be expensive. When
881+-- starting a watch, the whole tree must be read and watches are started on
882+-- each directory in the tree. The initial time to start the watch as well as
883+-- the memory required is proportional to the number of directories in the
884+-- tree.
885+--
886+-- When new directories are created under the tree they are added to the watch
887+-- on receiving the directory create event. However, the creation of a dir and
888+-- adding a watch for it is not atomic. The implementation takes care of this
889+-- and makes sure that watches are added for all directories. However, In the
890+-- mean time, the directory may have received more events which may get lost.
891+-- Handling of any such lost events is yet to be implemented.
892+--
893+-- See the Linux __inotify__ man page for more details.
894+--
906895-- /Internal/
907896--
908897watchTreesWith ::
909898 (Config -> Config) -> NonEmpty (Array Word8) -> SerialT IO Event
910−watchTreesWith f paths = S.bracket before after watchToStream
899+watchTreesWith f paths = S.bracket before after (watchToStream cfg)
911900
912901 where
913902
914− before = liftIO
915− $ openWatchRec (f $ setRecursiveMode True defaultConfig) paths
903+ cfg = f defaultConfig
904+ before = liftIO $ openWatch cfg paths
916905 after = liftIO . closeWatch
917906
918907-- | Like 'watchTreesWith' but uses the 'defaultConfig' options.
@@ -928,12 +917,13 @@ watchTrees = watchTreesWith id
928917-- Examine event stream
929918-------------------------------------------------------------------------------
930919
931−-- | Get the watch descriptor corresponding to the 'Event'. A watch descriptor
932−-- identifies which of the paths being monitored has generated this event.
920+-- | Get the watch root corresponding to the 'Event'.
933921--
934922-- Note that if a path was moved after adding to the watch, this will give the
935923-- original path and not the new path after moving.
936924--
925+-- TBD: we can possibly update the watch root on a move self event.
926+--
937927-- /Internal/
938928--
939929getRoot :: Event -> Array Word8
@@ -1188,12 +1178,6 @@ isDir = getFlag iN_ISDIR
11881178-------------------------------------------------------------------------------
11891179-- Debugging
11901180-------------------------------------------------------------------------------
1191−showEventShort :: Event -> String
1192−showEventShort ev@Event{..} = (utf8ToString $ getRelPath ev)
1193− ++ "_" ++ show eventFlags
1194− ++ showev isDir "Dir"
1195−
1196− where showev f str = if f ev then "_" ++ str else ""
11971181
11981182-- | Convert an 'Event' record to a String representation.
11991183showEvent :: Event -> String
@@ -1232,4 +1216,4 @@ showEvent ev@Event{..} =
12321216#else
12331217#warning "Disabling module Streamly.Internal.FileSystem.Event.Linux. Does not support kernels older than 2.6.36."
12341218module Streamly.Internal.FileSystem.Event.Linux () where
1235−#endif
1219+#endif