Skip to content

P
pyfs
Commit 71d37e0c by rfkelly0
Options

OSFSWatchMixin implementation using ReadDirectoryChangesW on win32

parent 13ced551
Showing with 234 additions and 189 deletions
fs/osfs/__init__.py
...@@ -89,6 +89,17 @@ class OSFS(OSFSXAttrMixin,OSFSWatchMixin,FS): ...@@ -89,6 +89,17 @@ class OSFS(OSFSXAttrMixin,OSFSWatchMixin,FS):
path = path.decode(self.encoding) path = path.decode(self.encoding)
return path return path
def unsyspath(self,path):
"""Convert a system-level path into an FS-level path.
This basically the reverse of getsyspath(). If the path does not
refer to a location within this filesystem, ValueError is raised.
"""
path = os.path.normpath(os.path.abspath(path))
if not path.startswith(self.root_path + os.path.sep):
raise ValueError("path not within this FS: %s" % (path,))
return path[len(self.root_path):]
@convert_os_errors @convert_os_errors
def open(self, path, mode="r", **kwargs): def open(self, path, mode="r", **kwargs):
mode = filter(lambda c: c in "rwabt+",mode) mode = filter(lambda c: c in "rwabt+",mode)
......
fs/osfs/watch.py
...@@ -15,182 +15,27 @@ from fs.errors import * ...@@ -15,182 +15,27 @@ from fs.errors import *
from fs.path import * from fs.path import *
from fs.watch import * from fs.watch import *
try: OSFSWatchMixin = None
import pyinotify
except ImportError: # Try using native implementation on win32
pyinotify = None if sys.platform == "win32":
try:
from fs.osfs.watch_win32 import OSFSWatchMixin
if pyinotify is not None: except ImportError:
pass
class OSFSWatchMixin(WatchableFSMixin):
"""Mixin providing change-watcher support via pyinotify.""" # Try using pyinotify if available
if OSFSWatchMixin is None:
__watch_lock = threading.Lock() try:
__watch_manager = None from fs.osfs.watch_inotify import OSFSWatchMixin
__watch_notifier = None except ImportError:
pass
def close(self):
super(OSFSWatchMixin,self).close() # Fall back to raising UnsupportedError
self.__shutdown_watch_manager(force=True) if OSFSWatchMixin is None:
self.notify_watchers(CLOSED)
def add_watcher(self,callback,path="/",events=None,recursive=True):
w = super(OSFSWatchMixin,self).add_watcher(callback,path,events,recursive)
syspath = self.getsyspath(path)
if isinstance(syspath,unicode):
syspath = syspath.encode(sys.getfilesystemencoding())
wm = self.__get_watch_manager()
evtmask = self.__get_event_mask(events)
def process_events(event):
self.__route_event(w,event)
kwds = dict(rec=recursive,auto_add=recursive,quiet=False)
try:
wids = wm.add_watch(syspath,evtmask,process_events,**kwds)
except pyinotify.WatchManagerError, e:
raise OperationFailedError("add_watcher",details=e)
w._pyinotify_id = wids[syspath]
return w
def del_watcher(self,watcher_or_callback):
wm = self.__get_watch_manager()
if isinstance(watcher_or_callback,Watcher):
watchers = [watcher_or_callback]
else:
watchers = self._find_watchers(watcher_or_callback)
for watcher in watchers:
wm.rm_watch(watcher._pyinotify_id,rec=watcher.recursive)
super(OSFSWatchMixin,self).del_watcher(watcher)
if not wm._wmd:
self.__shutdown_watch_manager()
def __get_event_mask(self,events):
"""Convert the given set of events into a pyinotify event mask."""
if events is None:
events = (EVENT,)
mask = 0
for evt in events:
if issubclass(ACCESSED,evt):
mask |= pyinotify.IN_ACCESS
if issubclass(CREATED,evt):
mask |= pyinotify.IN_CREATE
if issubclass(REMOVED,evt):
mask |= pyinotify.IN_DELETE
mask |= pyinotify.IN_DELETE_SELF
if issubclass(MODIFIED,evt):
mask |= pyinotify.IN_ATTRIB
mask |= pyinotify.IN_MODIFY
mask |= pyinotify.IN_CLOSE_WRITE
if issubclass(MOVED_SRC,evt):
mask |= pyinotify.IN_MOVED_FROM
mask |= pyinotify.IN_MOVED_TO
if issubclass(MOVED_DST,evt):
mask |= pyinotify.IN_MOVED_FROM
mask |= pyinotify.IN_MOVED_TO
if issubclass(OVERFLOW,evt):
mask |= pyinotify.IN_Q_OVERFLOW
if issubclass(CLOSED,evt):
mask |= pyinotify.IN_UNMOUNT
return mask
def unsyspath(self,path):
"""Convert a system-level path into an FS-level path."""
path = normpath(path)
if not isprefix(self.root_path,path):
raise ValueError("path not within this FS: %s" % (path,))
return path[len(self.root_path):]
def __route_event(self,watcher,inevt):
"""Convert pyinotify event into fs.watch event, then handle it."""
try:
path = self.unsyspath(inevt.pathname)
except ValueError:
return
try:
src_path = inevt.src_pathname
if src_path is not None:
src_path = self.unsyspath(src_path)
except (AttributeError,ValueError):
src_path = None
if inevt.mask & pyinotify.IN_ACCESS:
watcher.handle_event(ACCESSED(self,path))
if inevt.mask & pyinotify.IN_CREATE:
watcher.handle_event(CREATED(self,path))
# Recursive watching of directories in pyinotify requires
# the creation of a new watch for each subdir, resulting in
# a race condition whereby events in the subdir are missed.
# We'd prefer to duplicate events than to miss them.
if inevt.mask & pyinotify.IN_ISDIR:
try:
# pyinotify does this for dirs itself, we only.
# need to worry about newly-created files.
for child in self.listdir(path,files_only=True):
cpath = pathjoin(path,child)
self.notify_watchers(CREATED,cpath)
self.notify_watchers(MODIFIED,cpath,True,True)
except FSError:
pass
if inevt.mask & pyinotify.IN_DELETE:
watcher.handle_event(REMOVED(self,path))
if inevt.mask & pyinotify.IN_DELETE_SELF:
watcher.handle_event(REMOVED(self,path))
if inevt.mask & pyinotify.IN_ATTRIB:
watcher.handle_event(MODIFIED(self,path,True,False))
if inevt.mask & pyinotify.IN_MODIFY:
watcher.handle_event(MODIFIED(self,path,True,True))
if inevt.mask & pyinotify.IN_CLOSE_WRITE:
watcher.handle_event(MODIFIED(self,path,True,True))
if inevt.mask & pyinotify.IN_MOVED_FROM:
# Sorry folks, I'm not up for decoding the destination path.
watcher.handle_event(MOVED_SRC(self,path,None))
if inevt.mask & pyinotify.IN_MOVED_TO:
if getattr(inevt,"src_pathname",None):
watcher.handle_event(MOVED_SRC(self,src_path,path))
watcher.handle_event(MOVED_DST(self,path,src_path))
else:
watcher.handle_event(MOVED_DST(self,path,None))
if inevt.mask & pyinotify.IN_Q_OVERFLOW:
watcher.handle_event(OVERFLOW(self))
if inevt.mask & pyinotify.IN_UNMOUNT:
watcher.handle_event(CLOSE(self))
def __get_watch_manager(self):
"""Get the shared watch manager, initializing if necessary."""
if OSFSWatchMixin.__watch_notifier is None:
self.__watch_lock.acquire()
try:
if self.__watch_notifier is None:
wm = pyinotify.WatchManager()
n = pyinotify.ThreadedNotifier(wm)
n.start()
OSFSWatchMixin.__watch_manager = wm
OSFSWatchMixin.__watch_notifier = n
finally:
self.__watch_lock.release()
return OSFSWatchMixin.__watch_manager
def __shutdown_watch_manager(self,force=False):
"""Stop the shared watch manager, if there are no watches left."""
self.__watch_lock.acquire()
try:
if OSFSWatchMixin.__watch_manager is None:
return
if not force and OSFSWatchMixin.__watch_manager._wmd:
return
OSFSWatchMixin.__watch_notifier.stop()
OSFSWatchMixin.__watch_notifier = None
OSFSWatchMixin.__watch_manager = None
finally:
self.__watch_lock.release()
else:
class OSFSWatchMixin(object): class OSFSWatchMixin(object):
"""Mixin disabling change-watcher support."""
def add_watcher(self,*args,**kwds): def add_watcher(self,*args,**kwds):
raise UnsupportedError raise UnsupportedError
def del_watcher(self,watcher_or_callback): def del_watcher(self,watcher_or_callback):
raise UnsupportedError raise UnsupportedError
......
fs/osfs/watch_inotify.py 0 → 100644
"""
fs.osfs.watch_inotify
=============
Change watcher support for OSFS, backed by pyinotify.
"""
import os
import sys
import errno
import threading
from fs.errors import *
from fs.path import *
from fs.watch import *
import pyinotify
class OSFSWatchMixin(WatchableFSMixin):
"""Mixin providing change-watcher support via pyinotify."""
__watch_lock = threading.Lock()
__watch_manager = None
__watch_notifier = None
def close(self):
super(OSFSWatchMixin,self).close()
self.__shutdown_watch_manager(force=True)
self.notify_watchers(CLOSED)
def add_watcher(self,callback,path="/",events=None,recursive=True):
w = super(OSFSWatchMixin,self).add_watcher(callback,path,events,recursive)
syspath = self.getsyspath(path)
if isinstance(syspath,unicode):
syspath = syspath.encode(sys.getfilesystemencoding())
wm = self.__get_watch_manager()
evtmask = self.__get_event_mask(events)
def process_events(event):
self.__route_event(w,event)
kwds = dict(rec=recursive,auto_add=recursive,quiet=False)
try:
wids = wm.add_watch(syspath,evtmask,process_events,**kwds)
except pyinotify.WatchManagerError, e:
raise OperationFailedError("add_watcher",details=e)
w._pyinotify_id = wids[syspath]
return w
def del_watcher(self,watcher_or_callback):
wm = self.__get_watch_manager()
if isinstance(watcher_or_callback,Watcher):
watchers = [watcher_or_callback]
else:
watchers = self._find_watchers(watcher_or_callback)
for watcher in watchers:
wm.rm_watch(watcher._pyinotify_id,rec=watcher.recursive)
super(OSFSWatchMixin,self).del_watcher(watcher)
if not wm._wmd:
self.__shutdown_watch_manager()
def __get_event_mask(self,events):
"""Convert the given set of events into a pyinotify event mask."""
if events is None:
events = (EVENT,)
mask = 0
for evt in events:
if issubclass(ACCESSED,evt):
mask |= pyinotify.IN_ACCESS
if issubclass(CREATED,evt):
mask |= pyinotify.IN_CREATE
if issubclass(REMOVED,evt):
mask |= pyinotify.IN_DELETE
mask |= pyinotify.IN_DELETE_SELF
if issubclass(MODIFIED,evt):
mask |= pyinotify.IN_ATTRIB
mask |= pyinotify.IN_MODIFY
mask |= pyinotify.IN_CLOSE_WRITE
if issubclass(MOVED_SRC,evt):
mask |= pyinotify.IN_MOVED_FROM
mask |= pyinotify.IN_MOVED_TO
if issubclass(MOVED_DST,evt):
mask |= pyinotify.IN_MOVED_FROM
mask |= pyinotify.IN_MOVED_TO
if issubclass(OVERFLOW,evt):
mask |= pyinotify.IN_Q_OVERFLOW
if issubclass(CLOSED,evt):
mask |= pyinotify.IN_UNMOUNT
return mask
def __route_event(self,watcher,inevt):
"""Convert pyinotify event into fs.watch event, then handle it."""
try:
path = self.unsyspath(inevt.pathname)
except ValueError:
return
try:
src_path = inevt.src_pathname
if src_path is not None:
src_path = self.unsyspath(src_path)
except (AttributeError,ValueError):
src_path = None
if inevt.mask & pyinotify.IN_ACCESS:
watcher.handle_event(ACCESSED(self,path))
if inevt.mask & pyinotify.IN_CREATE:
watcher.handle_event(CREATED(self,path))
# Recursive watching of directories in pyinotify requires
# the creation of a new watch for each subdir, resulting in
# a race condition whereby events in the subdir are missed.
# We'd prefer to duplicate events than to miss them.
if inevt.mask & pyinotify.IN_ISDIR:
try:
# pyinotify does this for dirs itself, we only.
# need to worry about newly-created files.
for child in self.listdir(path,files_only=True):
cpath = pathjoin(path,child)
self.notify_watchers(CREATED,cpath)
self.notify_watchers(MODIFIED,cpath,True)
except FSError:
pass
if inevt.mask & pyinotify.IN_DELETE:
watcher.handle_event(REMOVED(self,path))
if inevt.mask & pyinotify.IN_DELETE_SELF:
watcher.handle_event(REMOVED(self,path))
if inevt.mask & pyinotify.IN_ATTRIB:
watcher.handle_event(MODIFIED(self,path,False))
if inevt.mask & pyinotify.IN_MODIFY:
watcher.handle_event(MODIFIED(self,path,True))
if inevt.mask & pyinotify.IN_CLOSE_WRITE:
watcher.handle_event(MODIFIED(self,path,True))
if inevt.mask & pyinotify.IN_MOVED_FROM:
# Sorry folks, I'm not up for decoding the destination path.
watcher.handle_event(MOVED_SRC(self,path,None))
if inevt.mask & pyinotify.IN_MOVED_TO:
if getattr(inevt,"src_pathname",None):
watcher.handle_event(MOVED_SRC(self,src_path,path))
watcher.handle_event(MOVED_DST(self,path,src_path))
else:
watcher.handle_event(MOVED_DST(self,path,None))
if inevt.mask & pyinotify.IN_Q_OVERFLOW:
watcher.handle_event(OVERFLOW(self))
if inevt.mask & pyinotify.IN_UNMOUNT:
watcher.handle_event(CLOSE(self))
def __get_watch_manager(self):
"""Get the shared watch manager, initializing if necessary."""
if OSFSWatchMixin.__watch_notifier is None:
self.__watch_lock.acquire()
try:
if self.__watch_notifier is None:
wm = pyinotify.WatchManager()
n = pyinotify.ThreadedNotifier(wm)
n.start()
OSFSWatchMixin.__watch_manager = wm
OSFSWatchMixin.__watch_notifier = n
finally:
self.__watch_lock.release()
return OSFSWatchMixin.__watch_manager
def __shutdown_watch_manager(self,force=False):
"""Stop the shared watch manager, if there are no watches left."""
self.__watch_lock.acquire()
try:
if OSFSWatchMixin.__watch_manager is None:
return
if not force and OSFSWatchMixin.__watch_manager._wmd:
return
OSFSWatchMixin.__watch_notifier.stop()
OSFSWatchMixin.__watch_notifier = None
OSFSWatchMixin.__watch_manager = None
finally:
self.__watch_lock.release()
fs/tests/test_watch.py
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
""" """
import os import os
import sys
import time import time
import unittest import unittest
...@@ -14,9 +15,17 @@ from fs.watch import * ...@@ -14,9 +15,17 @@ from fs.watch import *
from fs.tests import FSTestCases from fs.tests import FSTestCases
try: try:
import pyinotify from fs.osfs import watch_inotify
except ImportError: except ImportError:
pyinotify = None watch_inotify = None
if sys.platform == "win32":
try:
from fs.osfs import watch_win32
except ImportError:
watch_win32 = None
else:
watch_win32 = None
class WatcherTestCases: class WatcherTestCases:
...@@ -40,7 +49,7 @@ class WatcherTestCases: ...@@ -40,7 +49,7 @@ class WatcherTestCases:
self.watchfs._poll_cond.wait() self.watchfs._poll_cond.wait()
self.watchfs._poll_cond.release() self.watchfs._poll_cond.release()
else: else:
time.sleep(0.5) time.sleep(2)#0.5)
def assertEventOccurred(self,cls,path=None,**attrs): def assertEventOccurred(self,cls,path=None,**attrs):
if not self.checkEventOccurred(cls,path,**attrs): if not self.checkEventOccurred(cls,path,**attrs):
...@@ -73,6 +82,13 @@ class WatcherTestCases: ...@@ -73,6 +82,13 @@ class WatcherTestCases:
old_atime = self.fs.getinfo("hello").get("accessed_time") old_atime = self.fs.getinfo("hello").get("accessed_time")
self.assertEquals(self.fs.getcontents("hello"),"hello world") self.assertEquals(self.fs.getcontents("hello"),"hello world")
if not isinstance(self.watchfs,PollingWatchableFS): if not isinstance(self.watchfs,PollingWatchableFS):
# Help it along by updting the atime.
# TODO: why is this necessary?
if self.fs.hassyspath("hello"):
syspath = self.fs.getsyspath("hello")
mtime = os.stat(syspath).st_mtime
atime = int(time.time())
os.utime(self.fs.getsyspath("hello"),(atime,mtime))
self.assertEventOccurred(ACCESSED,"/hello") self.assertEventOccurred(ACCESSED,"/hello")
elif old_atime is not None: elif old_atime is not None:
# Some filesystems don't update atime synchronously, or only # Some filesystems don't update atime synchronously, or only
...@@ -159,7 +175,9 @@ class TestWatchers_TempFS(unittest.TestCase,FSTestCases,WatcherTestCases): ...@@ -159,7 +175,9 @@ class TestWatchers_TempFS(unittest.TestCase,FSTestCases,WatcherTestCases):
self.fs = tempfs.TempFS() self.fs = tempfs.TempFS()
watchfs = osfs.OSFS(self.fs.root_path) watchfs = osfs.OSFS(self.fs.root_path)
self.watchfs = ensure_watchable(watchfs,poll_interval=0.1) self.watchfs = ensure_watchable(watchfs,poll_interval=0.1)
if pyinotify is not None: if watch_inotify is not None:
self.assertEquals(watchfs,self.watchfs)
if watch_win32 is not None:
self.assertEquals(watchfs,self.watchfs) self.assertEquals(watchfs,self.watchfs)
def tearDown(self): def tearDown(self):
......
fs/watch.py
...@@ -61,10 +61,9 @@ class REMOVED(EVENT): ...@@ -61,10 +61,9 @@ class REMOVED(EVENT):
class MODIFIED(EVENT): class MODIFIED(EVENT):
"""Event fired when a file or directory is modified.""" """Event fired when a file or directory is modified."""
def __init__(self,fs,path,meta=False,data=False): def __init__(self,fs,path,data_changed=False):
super(MODIFIED,self).__init__(fs,path) super(MODIFIED,self).__init__(fs,path)
self.meta = meta self.data_changed = data_changed
self.data = data
class MOVED_DST(EVENT): class MOVED_DST(EVENT):
"""Event fired when a file or directory is the target of a move.""" """Event fired when a file or directory is the target of a move."""
...@@ -217,11 +216,11 @@ class WatchedFile(object): ...@@ -217,11 +216,11 @@ class WatchedFile(object):
def flush(self): def flush(self):
self.file.flush() self.file.flush()
self.fs.notify_watchers(MODIFIED,self.path,True,True) self.fs.notify_watchers(MODIFIED,self.path,True)
def close(self): def close(self):
self.file.close() self.file.close()
self.fs.notify_watchers(MODIFIED,self.path,True,True) self.fs.notify_watchers(MODIFIED,self.path,True)
class WatchableFS(WrapFS,WatchableFSMixin): class WatchableFS(WrapFS,WatchableFSMixin):
...@@ -322,7 +321,7 @@ class WatchableFS(WrapFS,WatchableFSMixin): ...@@ -322,7 +321,7 @@ class WatchableFS(WrapFS,WatchableFSMixin):
for src_path,isdir in src_paths.iteritems(): for src_path,isdir in src_paths.iteritems():
path = pathjoin(dst,src_path) path = pathjoin(dst,src_path)
if src_path in dst_paths: if src_path in dst_paths:
self.notify_watchers(MODIFIED,path,True,not isdir) self.notify_watchers(MODIFIED,path,not isdir)
else: else:
self.notify_watchers(CREATED,path) self.notify_watchers(CREATED,path)
for dst_path,isdir in dst_paths.iteritems(): for dst_path,isdir in dst_paths.iteritems():
...@@ -332,11 +331,11 @@ class WatchableFS(WrapFS,WatchableFSMixin): ...@@ -332,11 +331,11 @@ class WatchableFS(WrapFS,WatchableFSMixin):
def setxattr(self,path,name,value): def setxattr(self,path,name,value):
super(WatchableFS,self).setxattr(path,name,value) super(WatchableFS,self).setxattr(path,name,value)
self.notify_watchers(MODIFIED,path,True,False) self.notify_watchers(MODIFIED,path,False)
def delxattr(self,path,name): def delxattr(self,path,name):
super(WatchableFS,self).delxattr(path,name,value) super(WatchableFS,self).delxattr(path,name,value)
self.notify_watchers(MODIFIED,path,True,False) self.notify_watchers(MODIFIED,path,False)
...@@ -377,7 +376,6 @@ class PollingWatchableFS(WatchableFS): ...@@ -377,7 +376,6 @@ class PollingWatchableFS(WatchableFS):
pass pass
def _on_path_delete(self,event): def _on_path_delete(self,event):
print "DELETE", event.path
self._path_info.clear(event.path) self._path_info.clear(event.path)
def _poll_for_changes(self): def _poll_for_changes(self):
...@@ -421,7 +419,7 @@ class PollingWatchableFS(WatchableFS): ...@@ -421,7 +419,7 @@ class PollingWatchableFS(WatchableFS):
self.notify_watchers(CREATED,dirnm) self.notify_watchers(CREATED,dirnm)
else: else:
if new_info != old_info: if new_info != old_info:
self.notify_watchers(MODIFIED,dirnm,True,False) self.notify_watchers(MODIFIED,dirnm,False)
# Check the metadata for each file in the directory. # Check the metadata for each file in the directory.
# We assume that if the file's data changes, something in its # We assume that if the file's data changes, something in its
# metadata will also change; don't want to read through each file! # metadata will also change; don't want to read through each file!
...@@ -454,7 +452,7 @@ class PollingWatchableFS(WatchableFS): ...@@ -454,7 +452,7 @@ class PollingWatchableFS(WatchableFS):
was_modified = True was_modified = True
break break
if was_modified: if was_modified:
self.notify_watchers(MODIFIED,fpath,True,True) self.notify_watchers(MODIFIED,fpath,True)
elif was_accessed: elif was_accessed:
self.notify_watchers(ACCESSED,fpath) self.notify_watchers(ACCESSED,fpath)
# Check for deletion of cached child entries. # Check for deletion of cached child entries.
...@@ -463,7 +461,6 @@ class PollingWatchableFS(WatchableFS): ...@@ -463,7 +461,6 @@ class PollingWatchableFS(WatchableFS):
return return
cpath = pathjoin(dirnm,childnm) cpath = pathjoin(dirnm,childnm)
if not self.wrapped_fs.exists(cpath): if not self.wrapped_fs.exists(cpath):
print "REMOVED", cpath
self.notify_watchers(REMOVED,cpath) self.notify_watchers(REMOVED,cpath)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment