watch_win32: use weakrefs to avoid immortal reference cycles

ChangeLog

@@ -45,6 +45,12 @@
     * Fixed operation of OSFS on win32 when it points to the root of a drive.
     * Made SubFS a subclass of WrapFS, and moved it into its own module at
       fs.wrapfs.subfs.
+    * OSFSWatchMixin improvements:
+        * ensure that immortal reference cycles aren't created.
+        * watch_inotify:  allow more than one watcher on a single path.
+        * watch_win32:  report errors if the filesystem does't support
+                        ReadDirectoryChangesW.
+        * watch_win32:  don't create immortal reference cycles.
     * Fix OSFS.add_watcher on linux platforms; previous version would cancel
       any watchers when a new one was added.
     * MountFS: added support for mounting at the root directory, and for

fs/osfs/watch_win32.py

@@ -16,6 +16,7 @@ import struct
 import ctypes
 import ctypes.wintypes
 import traceback
+import weakref
 
 try:
     LPVOID = ctypes.wintypes.LPVOID
@@ -192,6 +193,7 @@ class WatchedDirectory(object):
         self.callback = callback
         self.recursive = recursive
         self.handle = None
+        self.error = None
         self.handle = CreateFileW(path,
                           FILE_LIST_DIRECTORY,
                           FILE_SHARE_READ | FILE_SHARE_WRITE,
@@ -219,9 +221,14 @@ class WatchedDirectory(object):
         overlapped.OffsetHigh = 0
         overlapped.Pointer = 0
         overlapped.hEvent = 0
-        ReadDirectoryChangesW(self.handle,
-                              ctypes.byref(self.result),len(self.result),
-                              self.recursive,self.flags,None,overlapped,None)
+        try:
+            ReadDirectoryChangesW(self.handle,
+                                  ctypes.byref(self.result),len(self.result),
+                                  self.recursive,self.flags,None,
+                                  overlapped,None)
+        except WindowsError, e:
+            self.error = e
+            self.close()
 
     def complete(self,nbytes):
         if nbytes == 0:
@@ -370,8 +377,11 @@ class WatchThread(threading.Thread):
                         try:
                             while True:
                                 w = self._new_watches.get_nowait()
-                                CreateIoCompletionPort(w.handle,self._iocp,hash(w),0)
-                                w.post()
+                                if w.handle is not None:
+                                    CreateIoCompletionPort(w.handle,
+                                                           self._iocp,
+                                                           hash(w),0)
+                                    w.post()
                                 w.ready.set()
                         except Queue.Empty:
                             pass
@@ -405,17 +415,26 @@ class OSFSWatchMixin(WatchableFSMixin):
         w = super(OSFSWatchMixin,self).add_watcher(callback,path,events,recursive)
         syspath = self.getsyspath(path)
         wt = self.__get_watch_thread()
+        #  Careful not to create a reference cycle here.
+        weak_self = weakref.ref(self)
         def handle_event(event_class,path,*args,**kwds):
+            selfref = weak_self()
+            if selfref is None:
+                return
             try:
-                path = self.unsyspath(path)
+                path = selfref.unsyspath(path)
             except ValueError:
                 pass
             else:
                 if event_class in (MOVED_SRC,MOVED_DST) and args and args[0]:
-                    args = (self.unsyspath(args[0]),) + args[1:]
-                event = event_class(self,path,*args,**kwds)
+                    args = (selfref.unsyspath(args[0]),) + args[1:]
+                event = event_class(selfref,path,*args,**kwds)
                 w.handle_event(event)
         w._watch_objs = wt.add_watcher(handle_event,syspath,w.events,w.recursive)
+        for wd in w._watch_objs:
+            if wd.error is not None:
+                self.del_watcher(w)
+                raise wd.error
         return w
 
     @convert_os_errors

fs/tests/test_watch.py

@@ -7,6 +7,8 @@
 import os
 import sys
 import time
+import gc
+import pickle
 import unittest
 
 from fs.path import *
@@ -176,6 +178,7 @@ class WatcherTestCases:
         changes.close()
 
 
+
 from fs import tempfs, osfs
 class TestWatchers_TempFS(unittest.TestCase,FSTestCases,WatcherTestCases):
 

fs/watch.py

@@ -21,6 +21,12 @@ An FS object that wants to be "watchable" must provide the following methods:
       Remove the given watcher object, or any watchers associated with
       the given callback.
 
+
+If you would prefer to read changes from a filesystem in a blocking fashion
+rather than using callbacks, you can use the function 'iter_changes' to obtain
+an iterator over the change events.
+
+
 """
 
 import weakref
@@ -554,15 +560,9 @@ def ensure_watchable(fs,wrapper_class=PollingWatchableFS,*args,**kwds):
     if isinstance(fs,wrapper_class):
         return fs
     try:
-        #  Try to add a watcher to a path that's unlikely to exist.
-        #  It's OK if the path does exist, since we remove the watcher.
-        #  It just might be slightly slower as the fs will actually have
-        #  to establish the watcher.
-        w = fs.add_watcher(lambda e: None,"/somepaththatsnotlikelytoexist")
-    except (AttributeError,UnsupportedError):
+        w = fs.add_watcher(lambda e: None,"/",recursive=False)
+    except (AttributeError,FSError):
         return wrapper_class(fs,*args,**kwds)
-    except FSError:
-        pass
     else:
         fs.del_watcher(w)
     return fs