Merge lp:~diegosarmentero/ubuntuone-client/darwin2-fsevents into lp:ubuntuone-client

[Diego Sarmentero]

- Refactoring windows fsevents implementation, to be usable from the future darwin one (LP: #1013323).

[Alejandro J. Cura (alecu)]

The implementation in add_watches_to_udf_ancestors is *very* specific to windows, so it needs to be moved to windows.py

---

FILESYSTEM_MONITOR_MASK gets broken in this branch because filesystem_monitor_mask seems to always be None.

---

All of this would break on darwin:
    is_valid_syncdaemon_path = None
    is_valid_os_path = None
    os_path = windowspath = None

So everything under:
   elif sys.platform == "darwin":

should come on a following branch.

---

Please rename "reference" to something less vague, and move each definition to the platform specific files.

review: Needs Fixing

[lp:~diegosarmentero/ubuntuone-client/darwin2-fsevents]

1263. By Diego Sarmentero on 2012-06-22

Updating branch

[Diego Sarmentero (diegosarmentero)]

> The implementation in add_watches_to_udf_ancestors is *very* specific to
> windows, so it needs to be moved to windows.py
>

Moved to windows, we will need to implement later for darwin something here, but not just returning True as in windows.

> ---
>
> FILESYSTEM_MONITOR_MASK gets broken in this branch because
> filesystem_monitor_mask seems to always be None.
>

This is not broken, the mask is assigned in windows.py, and for darwin i'm using it as None, so i can share more common code in common.py.

> ---
>
> All of this would break on darwin:
> is_valid_syncdaemon_path = None
> is_valid_os_path = None
> os_path = windowspath = None
>
> So everything under:
> elif sys.platform == "darwin":
>
> should come on a following branch.
>
> ---
>
> Please rename "reference" to something less vague, and move each definition to
> the platform specific files.

Done

[Alejandro J. Cura (alecu)]

> > FILESYSTEM_MONITOR_MASK gets broken in this branch because
> > filesystem_monitor_mask seems to always be None.
> >
>
> This is not broken, the mask is assigned in windows.py, and for darwin i'm
> using it as None, so i can share more common code in common.py.

You are right! Sorry about that :P

[Alejandro J. Cura (alecu)]

This comment is *Very* specific to windows:

        # On windows there is no need to add the watches to the ancestors
        # so we will always return true. The reason for this is that the
        # handles that we open stop the user from renaming the ancestors of
        # the UDF, for a user to do that he has to unsync the udf first

This too:
    For Windows:
    Also, they must not be literal paths, that is the \\?\ prefix should not be
    in the path.

Both should be in windows.py.

review: Needs Fixing

[lp:~diegosarmentero/ubuntuone-client/darwin2-fsevents]

1264. By Diego Sarmentero on 2012-06-22

merge

1265. By Diego Sarmentero on 2012-06-22

updates

1266. By Diego Sarmentero on 2012-06-22

adding not implemented error

1267. By Diego Sarmentero on 2012-06-22

moving event codes to filesystem notification init

[Diego Sarmentero (diegosarmentero)]

Branch fixed.
I added the event codes in the filesystem_notifications/__init__.py so we can create where it actually does this os dependent thing, and we can grab it from anywhere, because implementing it as a class attribute was breaking a lot of things.

Also i remove the if sys.platform == "win32": import decorators
on common, and in the future branch, when i add the darwin part, i'll make os_helper import the proper decorator based on the platform as it should be.

[Manuel de la Peña (mandel)]

review: Approve

[Alejandro J. Cura (alecu)]

This will probably break on darwin, where fs paths are utf-8 bytes, so it should go from common.py to windows.py:

473 if not isinstance(path, unicode):

----

The comment that starts with:
        # We are using this on windows and darwin.
        # For windows the logic is as follow:
        # ....

is still crap. The whole 9 lines of it, so it's not completely your fault, but it's crap.
Let's rewrite it with something simpler like this:

# We need to manually check if the path is a folder, because
# neither ReadDirectoryChangesW nor the FSEvents API tell us

Also the docstring in that function lies about "update the local subdir list".

---

review: Needs Fixing

[lp:~diegosarmentero/ubuntuone-client/darwin2-fsevents]

1268. By Diego Sarmentero on 2012-06-26

fixing comment

[Diego Sarmentero (diegosarmentero)]

> This will probably break on darwin, where fs paths are utf-8 bytes, so it
> should go from common.py to windows.py:
>
> 473 if not isinstance(path, unicode):
>
> ----
>
> The comment that starts with:
> # We are using this on windows and darwin.
> # For windows the logic is as follow:
> # ....
>
> is still crap. The whole 9 lines of it, so it's not completely your fault, but
> it's crap.
> Let's rewrite it with something simpler like this:
>
> # We need to manually check if the path is a folder, because
> # neither ReadDirectoryChangesW nor the FSEvents API tell us
>
> Also the docstring in that function lies about "update the local subdir list".
>
> ---

The comment has been fixed, and the isinstance thing is fixed in the next branch which involves actually the darwin things as we talk.

[Alejandro J. Cura (alecu)]

Code looks fine. Ran tests on Precise and windows, and they all pass.

review: Approve

=== modified file 'tests/platform/filesystem_notifications/test_windows.py'
--- tests/platform/filesystem_notifications/test_windows.py	2012-05-14 21:24:24 +0000
+++ tests/platform/filesystem_notifications/test_windows.py	2012-06-26 19:23:19 +0000
@@ -51,6 +51,7 @@
     IN_OPEN,
 )
 from ubuntuone.platform.filesystem_notifications import (
+    common,
     windows as filesystem_notifications,
 )
 from ubuntuone.platform.filesystem_notifications.windows import (
@@ -65,12 +66,11 @@
     FILE_NOTIFY_CHANGE_LAST_WRITE,
     FILE_NOTIFY_CHANGE_SECURITY,
     FILE_NOTIFY_CHANGE_LAST_ACCESS,
-    WINDOWS_ACTIONS_NAMES,
 )
 
 #create a rever mapping to use it in the tests.
 REVERSE_WINDOWS_ACTIONS = {}
-for key, value in filesystem_notifications.WINDOWS_ACTIONS.iteritems():
+for key, value in common.COMMON_ACTIONS.iteritems():
     REVERSE_WINDOWS_ACTIONS[value] = key
 
 
@@ -147,7 +147,7 @@
             # group all events in a single lists which is not what the COM API
             # does.
             str_events = [
-                (WINDOWS_ACTIONS_NAMES[action], path) for action, path in
+                (common.COMMON_ACTIONS_NAMES[action], path) for action, path in
                  events]
             self.raw_events.append(str_events)
             return events
 
=== modified file 'ubuntuone/platform/filesystem_notifications/__init__.py'
--- ubuntuone/platform/filesystem_notifications/__init__.py	2012-05-16 16:24:23 +0000
+++ ubuntuone/platform/filesystem_notifications/__init__.py	2012-06-26 19:23:19 +0000
@@ -32,6 +32,7 @@
 
 
 if sys.platform == "win32":
+    EVENT_CODES = [1, 2, 3, 4, 5]
     from ubuntuone.platform.filesystem_notifications import windows
     FilesystemMonitor = windows.FilesystemMonitor
     _GeneralINotifyProcessor = windows.NotifyProcessor
 
=== renamed file 'ubuntuone/platform/filesystem_notifications/windows.py' => 'ubuntuone/platform/filesystem_notifications/common.py'
--- ubuntuone/platform/filesystem_notifications/windows.py	2012-05-14 21:24:24 +0000
+++ ubuntuone/platform/filesystem_notifications/common.py	2012-06-26 19:23:19 +0000
@@ -26,41 +26,14 @@
 # do not wish to do so, delete this exception statement from your
 # version.  If you delete this exception statement from all source
 # files in the program, then also delete it here.
-"""File notifications on windows."""
+"""Generic File notifications."""
 
 import logging
 import os
 
-from uuid import uuid4
+from twisted.internet import defer
 
-from twisted.internet import defer, reactor
-from twisted.python.failure import Failure
-from pywintypes import OVERLAPPED
-from win32api import CloseHandle
-from win32con import (
-    FILE_SHARE_READ,
-    FILE_SHARE_WRITE,
-    FILE_FLAG_BACKUP_SEMANTICS,
-    FILE_NOTIFY_CHANGE_FILE_NAME,
-    FILE_NOTIFY_CHANGE_DIR_NAME,
-    FILE_NOTIFY_CHANGE_ATTRIBUTES,
-    FILE_NOTIFY_CHANGE_SIZE,
-    FILE_NOTIFY_CHANGE_LAST_WRITE,
-    FILE_NOTIFY_CHANGE_SECURITY,
-    OPEN_EXISTING)
-from win32file import (
-    AllocateReadBuffer,
-    CreateFileW,
-    GetOverlappedResult,
-    ReadDirectoryChangesW,
-    FILE_FLAG_OVERLAPPED,
-    FILE_NOTIFY_INFORMATION)
-from win32event import (
-    CreateEvent,
-    INFINITE,
-    SetEvent,
-    WaitForMultipleObjects,
-    WAIT_OBJECT_0)
+from ubuntuone.platform.filesystem_notifications import EVENT_CODES
 from ubuntuone.platform.filesystem_notifications.pyinotify_agnostic import (
     Event,
     WatchManagerError,
@@ -75,40 +48,32 @@
     IN_MOVED_FROM,
     IN_MOVED_TO,
     IN_MODIFY)
+
+from ubuntuone import logger
+
 from ubuntuone.platform.os_helper.windows import (
     is_valid_syncdaemon_path,
-    is_valid_windows_path,
-    get_syncdaemon_valid_path,
-    windowspath,
+    is_valid_windows_path as is_valid_os_path,
+    windowspath as os_path,
 )
-from ubuntuone import logger
-
-# our logging level
-TRACE = logger.TRACE
-
-# constant found in the msdn documentation:
-# http://msdn.microsoft.com/en-us/library/ff538834(v=vs.85).aspx
-FILE_LIST_DIRECTORY = 0x0001
-FILE_NOTIFY_CHANGE_LAST_ACCESS = 0x00000020
-FILE_NOTIFY_CHANGE_CREATION = 0x00000040
-
-# a map between the few events that we have on windows and those
+
+# a map between the few events that we have on common platforms and those
 # found in pyinotify
-WINDOWS_ACTIONS = {
-  1: IN_CREATE,
-  2: IN_DELETE,
-  3: IN_MODIFY,
-  4: IN_MOVED_FROM,
-  5: IN_MOVED_TO,
+COMMON_ACTIONS = {
+  EVENT_CODES[0]: IN_CREATE,
+  EVENT_CODES[1]: IN_DELETE,
+  EVENT_CODES[2]: IN_MODIFY,
+  EVENT_CODES[3]: IN_MOVED_FROM,
+  EVENT_CODES[4]: IN_MOVED_TO,
 }
 
 # a map of the actions to names so that we have better logs.
-WINDOWS_ACTIONS_NAMES = {
-  1: 'IN_CREATE',
-  2: 'IN_DELETE',
-  3: 'IN_MODIFY',
-  4: 'IN_MOVED_FROM',
-  5: 'IN_MOVED_TO',
+COMMON_ACTIONS_NAMES = {
+  EVENT_CODES[0]: 'IN_CREATE',
+  EVENT_CODES[1]: 'IN_DELETE',
+  EVENT_CODES[2]: 'IN_MODIFY',
+  EVENT_CODES[3]: 'IN_MOVED_FROM',
+  EVENT_CODES[4]: 'IN_MOVED_TO',
 }
 
 # translates quickly the event and it's is_dir state to our standard events
@@ -125,17 +90,8 @@
     IN_MOVED_TO: 'FS_FILE_CREATE',
     IN_MOVED_TO | IN_ISDIR: 'FS_DIR_CREATE'}
 
-# the default mask to be used in the watches added by the FilesystemMonitor
-# class
-FILESYSTEM_MONITOR_MASK = FILE_NOTIFY_CHANGE_FILE_NAME | \
-    FILE_NOTIFY_CHANGE_DIR_NAME | \
-    FILE_NOTIFY_CHANGE_ATTRIBUTES | \
-    FILE_NOTIFY_CHANGE_SIZE | \
-    FILE_NOTIFY_CHANGE_LAST_WRITE | \
-    FILE_NOTIFY_CHANGE_SECURITY | \
-    FILE_NOTIFY_CHANGE_LAST_ACCESS
-
-THREADPOOL_MAX = 20
+# our logging level
+TRACE = logger.TRACE
 
 
 # The implementation of the code that is provided as the pyinotify substitute
@@ -144,15 +100,11 @@
 
     def __init__(self, watch_descriptor, path, mask, auto_add, processor,
         buf_size=8192):
-        super(Watch, self).__init__()
-        self.log = logging.getLogger('ubuntuone.SyncDaemon.platform.windows.' +
+        self.log = logging.getLogger('ubuntuone.SyncDaemon.platform.common.' +
             'filesystem_notifications.Watch')
         self.log.setLevel(TRACE)
         self._processor = processor
         self._buf_size = buf_size
-        self._wait_stop = CreateEvent(None, 0, 0, None)
-        self._overlapped = OVERLAPPED()
-        self._overlapped.hEvent = CreateEvent(None, 0, 0, None)
         self._watching = False
         self._descriptor = watch_descriptor
         self._auto_add = auto_add
@@ -170,36 +122,25 @@
             path += os.path.sep
         self._path = os.path.abspath(path)
         self._mask = mask
-        # this deferred is fired when the watch has started monitoring
-        # a directory from a thread
-        self._watch_started_deferred = defer.Deferred()
-        # and this one is fired when the watch has stopped
-        self._watch_stopped_deferred = defer.Deferred()
 
-    @is_valid_windows_path(path_indexes=[1])
+    @is_valid_os_path(path_indexes=[1])
     def _update_subdirs(self, path, event):
         """Adds the path to the internal subdirs.
 
         The given path is considered to be a path and therefore this
         will not be checked.
         """
-        if WINDOWS_ACTIONS[event] == IN_CREATE:
+        if COMMON_ACTIONS[event] == IN_CREATE:
             self._subdirs.add(path)
-        elif WINDOWS_ACTIONS[event] == IN_DELETE and\
-                path in self._subdirs:
+        elif COMMON_ACTIONS[event] == IN_DELETE and path in self._subdirs:
             self._subdirs.remove(path)
 
-    @is_valid_windows_path(path_indexes=[1])
+    @is_valid_os_path(path_indexes=[1])
     def _path_is_dir(self, path):
-        """Check if the path is a dir and update the local subdir list."""
+        """Check if the path is a dir."""
 
-        # The logic of this function is the following:
-        # 1. ReadDirectoryChangesW changes does not send if a path
-        # is a new dir or not.
-        # 2. We keep track of subdirs that in self._subdir.
-        # 3. We check if a path is a dir by:
-        #    * Asking the os if the path exists.
-        #    * Finding the path in self._subdirs
+        # We need to manually check if the path is a folder, because
+        # neither ReadDirectoryChangesW nor the FSEvents API tell us
 
         is_dir = False
         if os.path.exists(path):
@@ -210,131 +151,7 @@
         self.log.debug('Is path %r a dir? %s', path, is_dir)
         return is_dir
 
-    def _process_events(self, events):
-        """Process the events from the queue."""
-        # do not do it if we stop watching and the events are empty
-        if not self._watching:
-            return
-
-        # we transform the events to be the same as the one in pyinotify
-        # and then use the proc_fun
-        for action, file_name in events:
-            if any([file_name.startswith(path)
-                        for path in self._ignore_paths]):
-                continue
-            # map the windows events to the pyinotify ones, tis is dirty but
-            # makes the multiplatform better, linux was first :P
-            syncdaemon_path = get_syncdaemon_valid_path(
-                                        os.path.join(self._path, file_name))
-            full_dir_path = os.path.join(self._path, file_name)
-            is_dir = self._path_is_dir(full_dir_path)
-            if is_dir:
-                # we need to update the list of subdirs that we have
-                self._update_subdirs(full_dir_path, action)
-            mask = WINDOWS_ACTIONS[action]
-            head, tail = os.path.split(file_name)
-            if is_dir:
-                mask |= IN_ISDIR
-            event_raw_data = {
-                'wd': self._descriptor,
-                'dir': is_dir,
-                'mask': mask,
-                'name': tail,
-                'path': '.'}
-            # by the way in which the win api fires the events we know for
-            # sure that no move events will be added in the wrong order, this
-            # is kind of hacky, I dont like it too much
-            if WINDOWS_ACTIONS[action] == IN_MOVED_FROM:
-                self._cookie = str(uuid4())
-                self._source_pathname = tail
-                event_raw_data['cookie'] = self._cookie
-            if WINDOWS_ACTIONS[action] == IN_MOVED_TO:
-                event_raw_data['src_pathname'] = self._source_pathname
-                event_raw_data['cookie'] = self._cookie
-            event = Event(event_raw_data)
-            # FIXME: event deduces the pathname wrong and we need to manually
-            # set it
-            event.pathname = syncdaemon_path
-            # add the event only if we do not have an exclude filter or
-            # the exclude filter returns False, that is, the event will not
-            # be excluded
-            self.log.debug('Pushing event %r to processor.', event)
-            self._processor(event)
-
-    def _call_deferred(self, f, *args):
-        """Executes the deferred call avoiding possible race conditions."""
-        if not self._watch_started_deferred.called:
-            f(*args)
-
-    def _watch_wrapper(self):
-        """Wrap _watch, and errback on any unhandled error."""
-        try:
-            self._watch()
-        except Exception:
-            reactor.callFromThread(self._call_deferred,
-                self._watch_started_deferred.errback, Failure())
-
-    def _watch(self):
-        """Watch a path that is a directory."""
-        self.log.debug('Adding watch for %r (exists? %r is dir? %r).',
-                       self._path,
-                       os.path.exists(self._path), os.path.isdir(self._path))
-        # we are going to be using the ReadDirectoryChangesW whihc requires
-        # a directory handle and the mask to be used.
-        self._watch_handle = CreateFileW(
-            self._path,
-            FILE_LIST_DIRECTORY,
-            FILE_SHARE_READ | FILE_SHARE_WRITE,
-            None,
-            OPEN_EXISTING,
-            FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OVERLAPPED,
-            None)
-
-        try:
-            self._watch_loop(self._watch_handle)
-        finally:
-            CloseHandle(self._watch_handle)
-            self._watch_handle = None
-            reactor.callFromThread(self.stopped.callback, True)
-
-    def _watch_loop(self, handle):
-        """The loop where we watch the directory."""
-        while True:
-            # important information to know about the parameters:
-            # param 1: the handle to the dir
-            # param 2: the size to be used in the kernel to store events
-            # that might be lost while the call is being performed. This
-            # is complicated to fine tune since if you make lots of watcher
-            # you migh used too much memory and make your OS to BSOD
-            buf = AllocateReadBuffer(self._buf_size)
-            ReadDirectoryChangesW(
-                handle,
-                buf,
-                self._auto_add,
-                self._mask,
-                self._overlapped,
-            )
-            if not self._watch_started_deferred.called:
-                reactor.callFromThread(self._call_deferred,
-                    self._watch_started_deferred.callback, True)
-            # wait for an event and ensure that we either stop or read the
-            # data
-            rc = WaitForMultipleObjects((self._wait_stop,
-                                         self._overlapped.hEvent),
-                                         0, INFINITE)
-            if rc == WAIT_OBJECT_0:
-                # Stop event
-                break
-            # if we continue, it means that we got some data, lets read it
-            data = GetOverlappedResult(handle, self._overlapped, True)
-            # lets ead the data and store it in the results
-            events = FILE_NOTIFY_INFORMATION(buf, data)
-            self.log.debug('Got from ReadDirectoryChangesW %r.',
-                    [(WINDOWS_ACTIONS_NAMES[action], path) for action, path in
-                        events])
-            reactor.callFromThread(self._process_events, events)
-
-    @is_valid_windows_path(path_indexes=[1])
+    @is_valid_os_path(path_indexes=[1])
     def ignore_path(self, path):
         """Add the path of the events to ignore."""
         if not path.endswith(os.path.sep):
@@ -343,7 +160,7 @@
             path = path[len(self._path):]
             self._ignore_paths.append(path)
 
-    @is_valid_windows_path(path_indexes=[1])
+    @is_valid_os_path(path_indexes=[1])
     def remove_ignored_path(self, path):
         """Reaccept path."""
         if not path.endswith(os.path.sep):
@@ -363,16 +180,12 @@
         # process the events.
         self.log.debug('Start watching path.')
         self._watching = True
-        reactor.callInThread(self._watch_wrapper)
-        return self._watch_started_deferred
 
     def stop_watching(self):
         """Tell the watch to stop processing events."""
         self.log.info('Stop watching %s', self._path)
-        SetEvent(self._wait_stop)
         self._watching = False
         self._subdirs = set()
-        return self.stopped
 
     def update(self, mask, auto_add=False):
         """Update the info used by the watcher."""
@@ -389,29 +202,18 @@
     def auto_add(self):
         return self._auto_add
 
-    @property
-    def started(self):
-        """A deferred that will be called when the watch is running."""
-        return self._watch_started_deferred
-
-    @property
-    def stopped(self):
-        """A deferred fired when the watch thread has finished."""
-        return self._watch_stopped_deferred
-
 
 class WatchManager(object):
     """Implement the same functions as pyinotify.WatchManager.
 
-    All paths passed to methods in this class should be windows paths.
+    All paths passed to methods in this class should be proper os paths.
 
     """
 
     def __init__(self, processor):
         """Init the manager to keep trak of the different watches."""
-        super(WatchManager, self).__init__()
         self._processor = processor
-        self.log = logging.getLogger('ubuntuone.SyncDaemon.platform.windows.'
+        self.log = logging.getLogger('ubuntuone.SyncDaemon.platform.common.'
             + 'filesystem_notifications.WatchManager')
         self.log.setLevel(TRACE)
         self._wdm = {}
@@ -420,12 +222,8 @@
 
     def stop(self):
         """Close the manager and stop all watches."""
-        self.log.debug('Stopping watches.')
-        wait_list = []
-        for current_wd in self._wdm:
-            wait_list.append(self._wdm[current_wd].stop_watching())
-            self.log.debug('Stopping Watch on %r.', self._wdm[current_wd].path)
-        return defer.DeferredList(wait_list)
+        # Should be implemented for each platform
+        raise NotImplementedError("Not implemented on this platform.")
 
     def get_watch(self, wd):
         """Return the watch with the given descriptor."""
@@ -442,8 +240,7 @@
         except KeyError, e:
             logging.error(str(e))
 
-    def _add_single_watch(self, path, mask, auto_add=False,
-        quiet=True):
+    def _add_single_watch(self, path, mask, auto_add=False, quiet=True):
         if path in self._ignored_paths:
             # simply removed it from the filter
             self._ignored_paths.remove(path)
@@ -452,23 +249,21 @@
         self.log.debug('add_single_watch(%s, %s, %s, %s)', path, mask,
             auto_add, quiet)
 
-        # adjust the number of threads based on the UDFs watched
-        reactor.suggestThreadPoolSize(THREADPOOL_MAX + self._wd_count + 1)
-        self._wdm[self._wd_count] = Watch(self._wd_count, path,
-                                          mask, auto_add, self._processor)
-        d = self._wdm[self._wd_count].start_watching()
-        self._wd_count += 1
-        return d
-
-    @is_valid_windows_path(path_indexes=[1])
-    def add_watch(self, path, mask, auto_add=False,
-        quiet=True):
+        return self._adding_watch(path, mask, auto_add)
+
+    def _adding_watch(self, path, mask, auto_add):
+        """Add the watch to the dict and start it."""
+        # This should be implemented for each OS
+        raise NotImplementedError("Not implemented on this platform.")
+
+    @is_valid_os_path(path_indexes=[1])
+    def add_watch(self, path, mask, auto_add=False, quiet=True):
         """Add a new path tp be watch.
 
         The method will ensure that the path is not already present.
         """
         if not isinstance(path, unicode):
-            e = NotImplementedError("No implementation on windows.")
+            e = NotImplementedError("No implementation on this platform.")
             return defer.fail(e)
         wd = self.get_wd(path)
         if wd is None:
@@ -480,9 +275,9 @@
 
     def update_watch(self, wd, mask=None, rec=False,
                      auto_add=False, quiet=True):
-        raise NotImplementedError("Not implemented on windows.")
+        raise NotImplementedError("Not implemented on this platform.")
 
-    @is_valid_windows_path(path_indexes=[1])
+    @is_valid_os_path(path_indexes=[1])
     def get_wd(self, path):
         """Return the watcher that is used to watch the given path."""
         if not path.endswith(os.path.sep):
@@ -512,7 +307,7 @@
             if not quiet:
                 raise WatchManagerError('Watch %s was not found' % wd, {})
 
-    @is_valid_windows_path(path_indexes=[1])
+    @is_valid_os_path(path_indexes=[1])
     def rm_path(self, path):
         """Remove a watch to the given path."""
         wd = self.get_wd(path)
@@ -527,9 +322,6 @@
     This interface will be exposed to syncdaemon, ergo all passed
     and returned paths must be a sequence of BYTES encoded with utf8.
 
-    Also, they must not be literal paths, that is the \\?\ prefix should not be
-    in the path.
-
     """
 
     def __init__(self, monitor, ignore_config=None):
@@ -549,13 +341,10 @@
         """Add an event and path(s) to the mute filter."""
         self.general_processor.add_to_mute_filter(event, paths)
 
-    @is_valid_syncdaemon_path(path_indexes=[1])
     def platform_is_ignored(self, path):
         """Should we ignore this path in the current platform.?"""
-        # don't support links yet
-        if path.endswith('.lnk'):
-            return True
-        return False
+        # This should be implemented in each platform
+        raise NotImplementedError("Not implemented on this platform.")
 
     @is_valid_syncdaemon_path(path_indexes=[1])
     def is_ignored(self, path):
@@ -572,7 +361,7 @@
         # lets ignore dir changes
         if event.dir:
             return
-        # on windows we just get IN_MODIFY, lets always fake
+        # on someplatforms we just get IN_MODIFY, lets always fake
         # an OPEN & CLOSE_WRITE couple
         raw_open = raw_close = {
            'wd': event.wd,
@@ -665,7 +454,7 @@
                 self.release_held_event()
                 self.general_processor.push_event(event)
         else:
-            # We should never get here on windows, I really do not know how we
+            # We should never get here, I really do not know how we
             # got here
             self.general_processor.log.warn(
                             'Cookie does not match the previoues held event!')
@@ -739,11 +528,12 @@
 
     def __init__(self, eq, fs, ignore_config=None, timeout=1):
         self.log = logging.getLogger('ubuntuone.SyncDaemon.FSMonitor')
+        self.filesystem_monitor_mask = None
         self.log.setLevel(TRACE)
         self.fs = fs
         self.eq = eq
-        self._processor = NotifyProcessor(self, ignore_config)
-        self._watch_manager = WatchManager(self._processor)
+        # You will need to create the NotifyProcessor and WatchManager
+        # in each OS-specific implementation
 
     def add_to_mute_filter(self, event, **info):
         """Add info to mute filter in the processor."""
@@ -757,27 +547,24 @@
         """Prepares the EQ to be closed."""
         return self._watch_manager.stop()
 
-    @windowspath(path_indexes=[1])
+    @os_path(path_indexes=[1])
     def rm_watch(self, dirpath):
         """Remove watch from a dir."""
         # trust the implementation of the manager
         self._watch_manager.rm_path(dirpath)
 
-    @windowspath(path_indexes=[1])
+    @os_path(path_indexes=[1])
     def add_watch(self, dirpath):
         """Add watch to a dir."""
         # the logic to check if the watch is already set
         # is all in WatchManager.add_watch
         return self._watch_manager.add_watch(dirpath,
-                             FILESYSTEM_MONITOR_MASK, auto_add=True)
+                             self.filesystem_monitor_mask, auto_add=True)
 
     def add_watches_to_udf_ancestors(self, volume):
         """Add a inotify watch to volume's ancestors if it's an UDF."""
-        # On windows there is no need to add the watches to the ancestors
-        # so we will always return true. The reason for this is that the
-        # handles that we open stop the user from renaming the ancestors of
-        # the UDF, for a user to do that he has to unsync the udf first
-        return defer.succeed(True)
+
+        # Should be implemented in each OS if necessary
 
     def is_frozen(self):
         """Checks if there's something frozen."""
 
=== added file 'ubuntuone/platform/filesystem_notifications/windows.py'
--- ubuntuone/platform/filesystem_notifications/windows.py	1970-01-01 00:00:00 +0000
+++ ubuntuone/platform/filesystem_notifications/windows.py	2012-06-26 19:23:19 +0000
@@ -0,0 +1,308 @@
+# -*- coding: utf-8 *-*
+#
+# Copyright 2011-2012 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License version 3, as published
+# by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# In addition, as a special exception, the copyright holders give
+# permission to link the code of portions of this program with the
+# OpenSSL library under certain conditions as described in each
+# individual source file, and distribute linked combinations
+# including the two.
+# You must obey the GNU General Public License in all respects
+# for all of the code used other than OpenSSL.  If you modify
+# file(s) with this exception, you may extend this exception to your
+# version of the file(s), but you are not obligated to do so.  If you
+# do not wish to do so, delete this exception statement from your
+# version.  If you delete this exception statement from all source
+# files in the program, then also delete it here.
+"""File notifications on windows."""
+
+import os
+from uuid import uuid4
+
+from twisted.internet import defer, reactor
+from twisted.python.failure import Failure
+
+from pywintypes import OVERLAPPED
+from win32api import CloseHandle
+from win32con import (
+    FILE_SHARE_READ,
+    FILE_SHARE_WRITE,
+    FILE_FLAG_BACKUP_SEMANTICS,
+    FILE_NOTIFY_CHANGE_FILE_NAME,
+    FILE_NOTIFY_CHANGE_DIR_NAME,
+    FILE_NOTIFY_CHANGE_ATTRIBUTES,
+    FILE_NOTIFY_CHANGE_SIZE,
+    FILE_NOTIFY_CHANGE_LAST_WRITE,
+    FILE_NOTIFY_CHANGE_SECURITY,
+    OPEN_EXISTING)
+from win32file import (
+    AllocateReadBuffer,
+    CreateFileW,
+    GetOverlappedResult,
+    ReadDirectoryChangesW,
+    FILE_FLAG_OVERLAPPED,
+    FILE_NOTIFY_INFORMATION)
+from win32event import (
+    CreateEvent,
+    INFINITE,
+    SetEvent,
+    WaitForMultipleObjects,
+    WAIT_OBJECT_0)
+
+from ubuntuone.platform.os_helper.windows import (
+    is_valid_syncdaemon_path,
+    get_syncdaemon_valid_path,
+)
+
+from ubuntuone.platform.filesystem_notifications import common
+
+
+# constant found in the msdn documentation:
+# http://msdn.microsoft.com/en-us/library/ff538834(v=vs.85).aspx
+FILE_LIST_DIRECTORY = 0x0001
+FILE_NOTIFY_CHANGE_LAST_ACCESS = 0x00000020
+FILE_NOTIFY_CHANGE_CREATION = 0x00000040
+
+THREADPOOL_MAX = 20
+
+
+class Watch(common.Watch):
+    """Implement the same functions as pyinotify.Watch."""
+
+    def __init__(self, watch_descriptor, path, mask, auto_add, processor,
+        buf_size=8192):
+        super(Watch, self).__init__(watch_descriptor, path, mask, auto_add,
+                                    processor, buf_size)
+        self._wait_stop = CreateEvent(None, 0, 0, None)
+        self._overlapped = OVERLAPPED()
+        self._overlapped.hEvent = CreateEvent(None, 0, 0, None)
+        # remember the subdirs we have so that when we have a delete we can
+        # check if it was a remove
+        self._subdirs = set()
+        # this deferred is fired when the watch has started monitoring
+        # a directory from a thread
+        self._watch_started_deferred = defer.Deferred()
+        # and this one is fired when the watch has stopped
+        self._watch_stopped_deferred = defer.Deferred()
+
+    def _process_events(self, events):
+        """Process the events from the queue."""
+        # do not do it if we stop watching and the events are empty
+        if not self._watching:
+            return
+
+        # we transform the events to be the same as the one in pyinotify
+        # and then use the proc_fun
+        for action, file_name in events:
+            if any([file_name.startswith(path)
+                        for path in self._ignore_paths]):
+                continue
+            # map the windows events to the pyinotify ones, tis is dirty but
+            # makes the multiplatform better, linux was first :P
+            syncdaemon_path = get_syncdaemon_valid_path(
+                                        os.path.join(self._path, file_name))
+            full_dir_path = os.path.join(self._path, file_name)
+            is_dir = self._path_is_dir(full_dir_path)
+            if is_dir:
+                # we need to update the list of subdirs that we have
+                self._update_subdirs(full_dir_path, action)
+            mask = common.COMMON_ACTIONS[action]
+            head, tail = os.path.split(file_name)
+            if is_dir:
+                mask |= common.IN_ISDIR
+            event_raw_data = {
+                'wd': self._descriptor,
+                'dir': is_dir,
+                'mask': mask,
+                'name': tail,
+                'path': '.'}
+            # by the way in which the win api fires the events we know for
+            # sure that no move events will be added in the wrong order, this
+            # is kind of hacky, I dont like it too much
+            if common.COMMON_ACTIONS[action] == common.IN_MOVED_FROM:
+                self._cookie = str(uuid4())
+                self._source_pathname = tail
+                event_raw_data['cookie'] = self._cookie
+            if common.COMMON_ACTIONS[action] == common.IN_MOVED_TO:
+                event_raw_data['src_pathname'] = self._source_pathname
+                event_raw_data['cookie'] = self._cookie
+            event = common.Event(event_raw_data)
+            # FIXME: event deduces the pathname wrong and we need to manually
+            # set it
+            event.pathname = syncdaemon_path
+            # add the event only if we do not have an exclude filter or
+            # the exclude filter returns False, that is, the event will not
+            # be excluded
+            self.log.debug('Pushing event %r to processor.', event)
+            self._processor(event)
+
+    def _call_deferred(self, f, *args):
+        """Executes the deferred call avoiding possible race conditions."""
+        if not self._watch_started_deferred.called:
+            f(*args)
+
+    def _watch_wrapper(self):
+        """Wrap _watch, and errback on any unhandled error."""
+        try:
+            self._watch()
+        except Exception:
+            reactor.callFromThread(self._call_deferred,
+                self._watch_started_deferred.errback, Failure())
+
+    def _watch(self):
+        """Watch a path that is a directory."""
+        self.log.debug('Adding watch for %r (exists? %r is dir? %r).',
+                       self._path,
+                       os.path.exists(self._path), os.path.isdir(self._path))
+        # we are going to be using the ReadDirectoryChangesW whihc requires
+        # a directory handle and the mask to be used.
+        self._watch_handle = CreateFileW(
+            self._path,
+            FILE_LIST_DIRECTORY,
+            FILE_SHARE_READ | FILE_SHARE_WRITE,
+            None,
+            OPEN_EXISTING,
+            FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OVERLAPPED,
+            None)
+
+        try:
+            self._watch_loop(self._watch_handle)
+        finally:
+            CloseHandle(self._watch_handle)
+            self._watch_handle = None
+            reactor.callFromThread(self.stopped.callback, True)
+
+    def _watch_loop(self, handle):
+        """The loop where we watch the directory."""
+        while True:
+            # important information to know about the parameters:
+            # param 1: the handle to the dir
+            # param 2: the size to be used in the kernel to store events
+            # that might be lost while the call is being performed. This
+            # is complicated to fine tune since if you make lots of watcher
+            # you migh used too much memory and make your OS to BSOD
+            buf = AllocateReadBuffer(self._buf_size)
+            ReadDirectoryChangesW(
+                handle,
+                buf,
+                self._auto_add,
+                self._mask,
+                self._overlapped,
+            )
+            if not self._watch_started_deferred.called:
+                reactor.callFromThread(self._call_deferred,
+                    self._watch_started_deferred.callback, True)
+            # wait for an event and ensure that we either stop or read the
+            # data
+            rc = WaitForMultipleObjects((self._wait_stop,
+                                         self._overlapped.hEvent),
+                                         0, INFINITE)
+            if rc == WAIT_OBJECT_0:
+                # Stop event
+                break
+            # if we continue, it means that we got some data, lets read it
+            data = GetOverlappedResult(handle, self._overlapped, True)
+            # lets ead the data and store it in the results
+            events = FILE_NOTIFY_INFORMATION(buf, data)
+            self.log.debug('Got from ReadDirectoryChangesW %r.',
+                    [(common.COMMON_ACTIONS_NAMES[action], path) \
+                    for action, path in events])
+            reactor.callFromThread(self._process_events, events)
+
+    def start_watching(self):
+        """Tell the watch to start processing events."""
+        super(Watch, self).start_watching()
+        reactor.callInThread(self._watch_wrapper)
+        return self._watch_started_deferred
+
+    def stop_watching(self):
+        """Tell the watch to stop processing events."""
+        super(Watch, self).stop_watching()
+        SetEvent(self._wait_stop)
+        return self.stopped
+
+    @property
+    def started(self):
+        """A deferred that will be called when the watch is running."""
+        return self._watch_started_deferred
+
+    @property
+    def stopped(self):
+        """A deferred fired when the watch thread has finished."""
+        return self._watch_stopped_deferred
+
+
+class WatchManager(common.WatchManager):
+
+    def stop(self):
+        """Close the manager and stop all watches."""
+        self.log.debug('Stopping watches.')
+        wait_list = []
+        for current_wd in self._wdm:
+            wait_list.append(self._wdm[current_wd].stop_watching())
+            self.log.debug('Stopping Watch on %r.', self._wdm[current_wd].path)
+        return defer.DeferredList(wait_list)
+
+    def _adding_watch(self, path, mask, auto_add):
+        # adjust the number of threads based on the UDFs watched
+        watch = Watch(self._wd_count, path, mask, auto_add, self._processor)
+        reactor.suggestThreadPoolSize(THREADPOOL_MAX + self._wd_count + 1)
+        self._wdm[self._wd_count] = watch
+        d = self._wdm[self._wd_count].start_watching()
+        self._wd_count += 1
+        return d
+
+
+class NotifyProcessor(common.NotifyProcessor):
+    """
+    Processor that takes care of dealing with the events.
+
+    Also, they must not be literal paths, that is the \\?\ prefix should not be
+    in the path.
+    """
+
+    @is_valid_syncdaemon_path(path_indexes=[1])
+    def platform_is_ignored(self, path):
+        """Should we ignore this path in the current platform.?"""
+        # don't support links yet
+        if path.endswith('.lnk'):
+            return True
+        return False
+
+
+class FilesystemMonitor(common.FilesystemMonitor):
+    """Manages the signals from filesystem."""
+
+    def __init__(self, eq, fs, ignore_config=None, timeout=1):
+        super(FilesystemMonitor, self).__init__(eq, fs, ignore_config, timeout)
+        self._processor = NotifyProcessor(self, ignore_config)
+        self._watch_manager = WatchManager(self._processor)
+        # the default mask to be used in the watches
+        # added by the FilesystemMonitor class
+        self.filesystem_monitor_mask = FILE_NOTIFY_CHANGE_FILE_NAME | \
+            FILE_NOTIFY_CHANGE_DIR_NAME | \
+            FILE_NOTIFY_CHANGE_ATTRIBUTES | \
+            FILE_NOTIFY_CHANGE_SIZE | \
+            FILE_NOTIFY_CHANGE_LAST_WRITE | \
+            FILE_NOTIFY_CHANGE_SECURITY | \
+            FILE_NOTIFY_CHANGE_LAST_ACCESS
+
+    def add_watches_to_udf_ancestors(self, volume):
+        """Add a inotify watch to volume's ancestors if it's an UDF."""
+        # There is no need to add the watches to the ancestors
+        # so we will always return true. The reason for this is that the
+        # handles that we open stop the user from renaming the ancestors of
+        # the UDF, for a user to do that he has to unsync the udf first
+        return defer.succeed(True)