Merge lp:~facundo/ubuntuone-client/non-bouncing-filesystem into lp:ubuntuone-client

[Facundo Batista]

Now we avoid the filesystem event bouncing.

[Facundo Batista (facundo)]

Now we avoid the filesystem event bouncing.

Before, when SD removed a file (for example), the event FS_FILE_DELETED arrived from the filesystem. As the event from filesystem is related to a "path", it may happen that if a new file is created with that path *before* the event from filesystem arrives, the event will impact the new node, incorrectly.

So, we need to avoid these bouncing events. For this I put a MuteFilter en EQ to whom the expected events are added, and it filters them. As the one that touches disk is FSM, this is pretty clean.

Regarding tests, I added some to test MuteFilter itself, and also to test EQ filtering the events. But as FSM now has a reference to EQ, I had to change some other tests because of this.

[John Lenton (chipaca)]

I love it. Lurve, lurve, luv it.

review: Approve

[Guillermo Gonzalez (verterok)]

I *really* like this

review: Approve

=== modified file 'contrib/testing/testcase.py'
--- contrib/testing/testcase.py	2009-10-12 16:00:16 +0000
+++ contrib/testing/testcase.py	2009-10-15 15:40:24 +0000
@@ -102,6 +102,7 @@
         self.fs = fs_manager.FileSystemManager(self.data_dir,
                                                self.partials_dir, self.vm)
         self.event_q = event_queue.EventQueue(self.fs)
+        self.fs.register_eq(self.event_q)
         self.action_q = FakeActionQueue(self.event_q)
         self.state = main.SyncDaemonStateManager(self, 2, 0)
         self.event_q.subscribe(self.vm)

=== modified file 'tests/syncdaemon/test_eq_inotify.py'
--- tests/syncdaemon/test_eq_inotify.py	2009-09-30 18:02:18 +0000
+++ tests/syncdaemon/test_eq_inotify.py	2009-10-15 15:40:24 +0000
@@ -885,6 +885,185 @@
         return self._deferred
 
 
+class MutedSignalsTests(BaseTwisted):
+    '''Test that EQ filter some signals on demand.'''
+
+    class DontHitMe(object):
+        '''we shouldn't be called'''
+        # class-closure, cannot use self, pylint: disable-msg=E0213
+        def __init__(innerself, obj):
+            innerself.obj = obj
+        def handle_default(innerself, *a):
+            '''Something here? Error!'''
+            innerself.obj.finished_error("don't hit me! received %s" % (a,))
+
+    def check_filter(self, _=None):
+        self.assertFalse(self.eq._processor._to_mute._cnt)
+        self.finished_ok()
+
+    def test_file_open(self):
+        '''Test receiving the open signal on files.'''
+        testfile = os.path.join(self.root_dir, "foo")
+        open(testfile, "w").close()
+        self.eq.add_to_mute_filter("FS_FILE_OPEN", testfile)
+        self.eq.add_to_mute_filter("FS_FILE_CLOSE_NOWRITE", testfile)
+
+        self.eq.inotify_add_watch(self.root_dir)
+        self.eq.subscribe(self.DontHitMe(self))
+
+        # generate the event
+        open(testfile)
+        reactor.callLater(.1, self.check_filter)
+        return self._deferred
+
+    def test_file_close_nowrite(self):
+        '''Test receiving the close_nowrite signal on files.'''
+        testfile = os.path.join(self.root_dir, "foo")
+        open(testfile, "w").close()
+        fh = open(testfile)
+        self.eq.add_to_mute_filter("FS_FILE_CLOSE_NOWRITE", testfile)
+
+        self.eq.inotify_add_watch(self.root_dir)
+        self.eq.subscribe(self.DontHitMe(self))
+
+        # generate the event
+        fh.close()
+        reactor.callLater(.1, self.check_filter)
+        return self._deferred
+
+    def test_file_create_close_write(self):
+        '''Test receiving the create and close_write signals on files.'''
+        testfile = os.path.join(self.root_dir, "foo")
+        self.eq.add_to_mute_filter("FS_FILE_CREATE", testfile)
+        self.eq.add_to_mute_filter("FS_FILE_OPEN", testfile)
+        self.eq.add_to_mute_filter("FS_FILE_CLOSE_WRITE", testfile)
+
+        self.eq.inotify_add_watch(self.root_dir)
+        self.eq.subscribe(self.DontHitMe(self))
+
+        # generate the event
+        open(testfile, "w").close()
+        reactor.callLater(.1, self.check_filter)
+        return self._deferred
+
+    def test_dir_create(self):
+        '''Test receiving the create signal on dirs.'''
+        testdir = os.path.join(self.root_dir, "foo")
+        self.eq.add_to_mute_filter("FS_DIR_CREATE", testdir)
+
+        self.eq.inotify_add_watch(self.root_dir)
+        self.eq.subscribe(self.DontHitMe(self))
+
+        # generate the event
+        os.mkdir(testdir)
+        reactor.callLater(.1, self.check_filter)
+        return self._deferred
+
+    def test_file_delete(self):
+        '''Test the delete signal on a file.'''
+        testfile = os.path.join(self.root_dir, "foo")
+        open(testfile, "w").close()
+        self.eq.add_to_mute_filter("FS_FILE_DELETE", testfile)
+
+        self.eq.inotify_add_watch(self.root_dir)
+        self.eq.subscribe(self.DontHitMe(self))
+
+        # generate the event
+        os.remove(testfile)
+        reactor.callLater(.1, self.check_filter)
+        return self._deferred
+
+    def test_dir_delete(self):
+        '''Test the delete signal on a dir.'''
+        testdir = os.path.join(self.root_dir, "foo")
+        os.mkdir(testdir)
+        self.eq.add_to_mute_filter("FS_DIR_DELETE", testdir)
+
+        self.eq.inotify_add_watch(self.root_dir)
+        self.eq.subscribe(self.DontHitMe(self))
+
+        # generate the event
+        os.rmdir(testdir)
+        reactor.callLater(.1, self.check_filter)
+        return self._deferred
+
+    def test_file_moved_inside(self):
+        '''Test the synthesis of the FILE_MOVE event.'''
+        fromfile = os.path.join(self.root_dir, "foo")
+        self.fs.create(fromfile, "")
+        self.fs.set_node_id(fromfile, "from_node_id")
+        tofile = os.path.join(self.root_dir, "bar")
+        self.fs.create(tofile, "")
+        self.fs.set_node_id(tofile, "to_node_id")
+        open(fromfile, "w").close()
+        self.eq.add_to_mute_filter("FS_FILE_MOVE", fromfile, tofile)
+
+        self.eq.inotify_add_watch(self.root_dir)
+        self.eq.subscribe(self.DontHitMe(self))
+
+        # generate the event
+        os.rename(fromfile, tofile)
+        reactor.callLater(.1, self.check_filter)
+        return self._deferred
+
+    def test_dir_moved_inside(self):
+        '''Test the synthesis of the DIR_MOVE event.'''
+        fromdir = os.path.join(self.root_dir, "foo")
+        self.fs.create(fromdir, "")
+        self.fs.set_node_id(fromdir, "from_node_id")
+        todir = os.path.join(self.root_dir, "bar")
+        self.fs.create(todir, "")
+        self.fs.set_node_id(todir, "to_node_id")
+        os.mkdir(fromdir)
+        self.eq.add_to_mute_filter("FS_DIR_MOVE", fromdir, todir)
+
+        self.eq.inotify_add_watch(self.root_dir)
+        self.eq.subscribe(self.DontHitMe(self))
+
+        # generate the event
+        os.rename(fromdir, todir)
+        reactor.callLater(.1, self.check_filter)
+        return self._deferred
+
+    def test_file_moved_to_conflict(self):
+        '''Test the handling of the FILE_MOVE event when dest is conflict.'''
+        fromfile = os.path.join(self.root_dir, "foo")
+        self.fs.create(fromfile, "")
+        self.fs.set_node_id(fromfile, "from_node_id")
+        tofile = os.path.join(self.root_dir, "foo.u1conflict")
+        self.fs.create(tofile, "")
+        self.fs.set_node_id(tofile, "to_node_id")
+        open(fromfile, "w").close()
+        self.eq.add_to_mute_filter("FS_FILE_MOVE", fromfile, tofile)
+
+        self.eq.inotify_add_watch(self.root_dir)
+        self.eq.subscribe(self.DontHitMe(self))
+
+        # generate the event
+        os.rename(fromfile, tofile)
+        reactor.callLater(.1, self.check_filter)
+        return self._deferred
+
+    def test_file_moved_from_partial(self):
+        '''Test the handling of the FILE_MOVE event when source is partial.'''
+        fromfile = os.path.join(self.root_dir, "mdid.u1partial.foo")
+        root_dir = os.path.join(self.root_dir, "my_files")
+        tofile = os.path.join(root_dir, "foo")
+        mypath = functools.partial(os.path.join, root_dir)
+        os.mkdir(root_dir)
+        open(fromfile, "w").close()
+        self.eq.add_to_mute_filter("FS_FILE_CREATE", tofile)
+        self.eq.add_to_mute_filter("FS_FILE_CLOSE_WRITE", tofile)
+
+        self.eq.inotify_add_watch(root_dir)
+        self.eq.subscribe(self.DontHitMe(self))
+
+        # generate the event
+        os.rename(fromfile, tofile)
+        reactor.callLater(.1, self.check_filter)
+        return self._deferred
+
+
 def test_suite():
     # pylint: disable-msg=C0111
     return unittest.TestLoader().loadTestsFromName(__name__)

=== modified file 'tests/syncdaemon/test_eventqueue.py'
--- tests/syncdaemon/test_eventqueue.py	2009-09-01 20:35:36 +0000
+++ tests/syncdaemon/test_eventqueue.py	2009-10-15 15:40:24 +0000
@@ -47,6 +47,7 @@
         self.fs.set_by_path(path=self.root_dir,
                               local_hash=None, server_hash=None)
         self.eq = event_queue.EventQueue(self.fs)
+        self.fs.register_eq(self.eq)
 
     def tearDown(self):
         '''Clean up the tests.'''
@@ -301,6 +302,71 @@
         return d
 
 
+class MuteFilterTests(unittest.TestCase):
+    '''Tests the MuteFilter class.'''
+
+    def setUp(self):
+        self.mf = event_queue.MuteFilter()
+
+    def test_empty(self):
+        '''Nothing there.'''
+        self.assertFalse(self.mf._cnt)
+
+    def test_add_one(self):
+        '''Adds one element.'''
+        self.mf.add("foo")
+        self.assertEqual(self.mf._cnt, dict(foo=1))
+
+    def test_add_two_different(self):
+        '''Adds two different elements.'''
+        self.mf.add("foo")
+        self.mf.add("bar")
+        self.assertEqual(self.mf._cnt, dict(foo=1, bar=1))
+
+    def test_add_two_equal(self):
+        '''Adds one element twice.'''
+        self.mf.add("foo")
+        self.mf.add("foo")
+        self.assertEqual(self.mf._cnt, dict(foo=2))
+
+    def test_add_two_equal_and_third(self):
+        '''Adds one element.'''
+        self.mf.add("foo")
+        self.mf.add("bar")
+        self.mf.add("bar")
+        self.assertEqual(self.mf._cnt, dict(foo=1, bar=2))
+
+    def test_pop_simple(self):
+        '''Pops one element.'''
+        self.mf.add("foo")
+        self.assertFalse(self.mf.pop("bar"))
+        self.assertEqual(self.mf._cnt, dict(foo=1))
+        self.assertTrue(self.mf.pop("foo"))
+        self.assertFalse(self.mf._cnt)
+
+    def test_pop_complex(self):
+        '''Pops several elements.'''
+        # add several
+        self.mf.add("foo")
+        self.mf.add("bar")
+        self.mf.add("bar")
+        self.assertEqual(self.mf._cnt, dict(foo=1, bar=2))
+
+        # clean bar
+        self.assertTrue(self.mf.pop("bar"))
+        self.assertEqual(self.mf._cnt, dict(foo=1, bar=1))
+        self.assertTrue(self.mf.pop("bar"))
+        self.assertEqual(self.mf._cnt, dict(foo=1))
+        self.assertFalse(self.mf.pop("bar"))
+        self.assertEqual(self.mf._cnt, dict(foo=1))
+
+        # clean foo
+        self.assertTrue(self.mf.pop("foo"))
+        self.assertFalse(self.mf._cnt)
+        self.assertFalse(self.mf.pop("foo"))
+        self.assertFalse(self.mf._cnt)
+
+
 def test_suite():
     # pylint: disable-msg=C0111
     return unittest.TestLoader().loadTestsFromName(__name__)

=== modified file 'tests/syncdaemon/test_fsm.py'
--- tests/syncdaemon/test_fsm.py	2009-10-09 17:52:07 +0000
+++ tests/syncdaemon/test_fsm.py	2009-10-15 15:40:24 +0000
@@ -36,6 +36,7 @@
     METADATA_VERSION,
 )
 from ubuntuone.syncdaemon.volume_manager import Share, allow_writes
+from ubuntuone.syncdaemon.event_queue import EventQueue
 
 TESTS_DIR = os.path.join(os.getcwd(), "tmp")
 
@@ -62,12 +63,15 @@
         self.partials_dir = os.path.join(TESTS_DIR, "partials")
         self.fsm = FileSystemManager(self.fsmdir, self.partials_dir,
                                      FakeVolumeManager(self.root_dir))
+        self.eq = EventQueue(self.fsm)
+        self.fsm.register_eq(self.eq)
         self.share = self.create_share('share', 'share_name',
                                             self.fsm, self.shares_dir)
         self.share_path = self.share.path
 
     def tearDown(self):
         """ Clean up the tests. """
+        self.eq.shutdown()
         self.rmtree(TESTS_DIR)
 
     @staticmethod
@@ -2090,7 +2094,7 @@
                 os.chmod(os.path.join(dirpath, dir), 0777)
             for file in files:
                 os.chmod(os.path.join(dirpath, file), 0666)
-        shutil.rmtree(TESTS_DIR)
+        FSMTestCase.tearDown(self)
 
     def test_file_ro_share_fail(self):
         """ Test that manual creation of a file, fails on a ro-share. """
@@ -2241,21 +2245,25 @@
         FSMTestCase.setUp(self)
         # create a ro share
         self.share_ro = self.create_share('share_ro', 'share_ro_name',
-                                       self.fsm, self.shares_dir, access_level='View')
+                                          self.fsm, self.shares_dir,
+                                          access_level='View')
         self.share_ro_path = self.share_ro.path
 
     def test_write_in_ro_share(self):
         """test the EnableShareWrite context manager in a ro share"""
         path = os.path.join(self.share_ro_path, 'foo', 'a_file_in_a_ro_share')
         data = 'yes I can write!'
-        can_write_parent = os.access(os.path.dirname(self.share_ro_path), os.W_OK)
+        can_write_parent = os.access(os.path.dirname(self.share_ro_path),
+                                     os.W_OK)
         with EnableShareWrite(self.share_ro, path):
             with open(path, 'w') as f:
                 f.write(data)
         self.assertEquals(data, open(path, 'r').read())
         self.assertFalse(os.access(self.share_ro_path, os.W_OK))
         # check that the parent permissions are ok
-        self.assertEquals(can_write_parent, os.access(os.path.dirname(self.share_ro_path), os.W_OK))
+        self.assertEquals(can_write_parent,
+                          os.access(os.path.dirname(self.share_ro_path),
+                                    os.W_OK))
         # fail to write directly in the share
         self.assertRaises(IOError, open, path, 'w')
 
@@ -2291,7 +2299,8 @@
         os.makedirs(self.root_dir)
         self.data_dir = os.path.join(TESTS_DIR, "data")
         self.partials_dir = os.path.join(TESTS_DIR, "partials")
-        self.main = FakeMain(self.root_dir, self.shares_dir, self.data_dir, self.partials_dir)
+        self.main = FakeMain(self.root_dir, self.shares_dir,
+                             self.data_dir, self.partials_dir)
         self.fsm = self.main.fs
         self.share = self.create_share('share', 'share_name',
                                            self.fsm, self.shares_dir)
@@ -2300,7 +2309,7 @@
     def tearDown(self):
         """ Clean up the tests. """
         self.main.shutdown()
-        FSMTestCase.tearDown(self)
+        self.rmtree(TESTS_DIR)
 
 
     @staticmethod

=== modified file 'tests/syncdaemon/test_localrescan.py'
--- tests/syncdaemon/test_localrescan.py	2009-09-22 15:19:25 +0000
+++ tests/syncdaemon/test_localrescan.py	2009-10-15 15:40:24 +0000
@@ -46,9 +46,10 @@
         '''Store stuff as pushed.'''
         self.pushed.append((event, path))
 
-    def freeze_begin(self, *a):
+    def _fake(self, *a):
         '''fake'''
-    inotify_add_watch = inotify_has_watch = freeze_rollback = is_frozen = freeze_begin
+    inotify_add_watch = inotify_has_watch = freeze_rollback = is_frozen = _fake
+    freeze_begin = add_to_mute_filter = _fake
 
     def freeze_commit(self, events):
         '''just store events'''
@@ -88,6 +89,7 @@
                                                         self.vm)
         self.fsm.create(usrdir, "")
         self.eq = FakeEQ()
+        self.fsm.register_eq(self.eq)
         self.aq = FakeAQ()
 
     def tearDown(self):

=== modified file 'tests/syncdaemon/test_sync.py'
--- tests/syncdaemon/test_sync.py	2009-09-01 20:35:36 +0000
+++ tests/syncdaemon/test_sync.py	2009-10-15 15:40:24 +0000
@@ -38,6 +38,7 @@
 from ubuntuone.syncdaemon.main import Main
 from ubuntuone.syncdaemon.sync import FSKey
 from ubuntuone.syncdaemon.volume_manager import Share
+from ubuntuone.syncdaemon.event_queue import EventQueue
 
 DBusInterface.test = True
 
@@ -57,12 +58,15 @@
         os.makedirs(self.partials_dir)
         self.fsm = FileSystemManager(self.fsmdir, self.partials_dir,
                                      FakeVolumeManager(self.root_dir))
+        self.eq = EventQueue(self.fsm)
+        self.fsm.register_eq(self.eq)
         self.share = self.create_share('share', 'share_name',
                                             self.fsm, self.shares_dir)
         self.share_path = self.share.path
 
     def tearDown(self):
         """ Clean up the test dir"""
+        self.eq.shutdown()
         shutil.rmtree(self.test_dir)
 
     @staticmethod

=== modified file 'ubuntuone/syncdaemon/event_queue.py'
--- ubuntuone/syncdaemon/event_queue.py	2009-09-30 18:02:18 +0000
+++ ubuntuone/syncdaemon/event_queue.py	2009-10-15 15:40:24 +0000
@@ -149,6 +149,35 @@
 
 DEFAULT_HANDLER = "handle_default" # receives (event_name, *args, **kwargs)
 
+
+class MuteFilter(object):
+    '''Stores what needs to be muted.'''
+    def __init__(self):
+        self._cnt = {}
+        self.log = logging.getLogger('ubuntuone.SyncDaemon.MuteFilter')
+
+    def add(self, element):
+        '''Adds and element to the filter.'''
+        self.log.debug("Adding: %s", element)
+        self._cnt[element] = self._cnt.get(element, 0) + 1
+
+    def pop(self, element):
+        '''Pops an element from the filter, if there, and returns if it was.'''
+        if element not in self._cnt:
+            return False
+
+        self._cnt[element] = self._cnt.get(element, 0) - 1
+        if not self._cnt[element]:
+            # reached zero
+            del self._cnt[element]
+
+        # log what happened and how many items we have left
+        q = sum(self._cnt.itervalues())
+        self.log.debug("Blocking %s (%d left)", element, q)
+
+        return True
+
+
 class _INotifyProcessor(pyinotify.ProcessEvent):
     '''Helper class that is called from inpotify when an event happens.
 
@@ -162,6 +191,24 @@
         self.timer = None
         self.frozen_path = None
         self.frozen_evts = False
+        self._to_mute = MuteFilter()
+
+    def add_to_mute_filter(self, event, *paths):
+        '''Add an event and path(s) to the mute filter.'''
+        # all events have one path except the MOVEs
+        if event in ("FS_FILE_MOVE", "FS_DIR_MOVE"):
+            f_path, t_path = paths
+            is_from_forreal = not self.is_ignored(f_path)
+            is_to_forreal = not self.is_ignored(t_path)
+            if is_from_forreal and is_to_forreal:
+                self._to_mute.add((event, f_path, t_path))
+            elif is_to_forreal:
+                self._to_mute.add(('FS_FILE_CREATE', t_path))
+                self._to_mute.add(('FS_FILE_CLOSE_WRITE', t_path))
+        else:
+            path = paths[0]
+            if not self.is_ignored(path):
+                self._to_mute.add((event, path))
 
     def on_timeout(self):
         '''Called on timeout.'''
@@ -174,7 +221,7 @@
             try:
                 self.timer.cancel()
             except error.AlreadyCalled:
-                # self.timeout() was *just* called, do noting here
+                # self.timeout() was *just* called, do nothing here
                 return
         self.push_event(self.held_event)
         self.held_event = None
@@ -234,7 +281,7 @@
                 try:
                     self.timer.cancel()
                 except error.AlreadyCalled:
-                    # self.timeout() was *just* called, do noting here
+                    # self.timeout() was *just* called, do nothing here
                     pass
                 else:
                     f_path = os.path.join(self.held_event.path,
@@ -255,17 +302,17 @@
                             evtname = "FS_FILE_"
                         if f_share_id != t_share_id:
                             # if the share_id are != push a delete/create
-                            self.eq.push(evtname+"DELETE", f_path)
-                            self.eq.push(evtname+"CREATE", t_path)
+                            self.eq_push(evtname+"DELETE", f_path)
+                            self.eq_push(evtname+"CREATE", t_path)
                             if not this_is_a_dir:
-                                self.eq.push('FS_FILE_CLOSE_WRITE', t_path)
+                                self.eq_push('FS_FILE_CLOSE_WRITE', t_path)
                         else:
-                            self.eq.push(evtname+"MOVE", f_path, t_path)
+                            self.eq_push(evtname+"MOVE", f_path, t_path)
                     elif is_to_forreal:
                         # this is the case of a MOVE from something ignored
                         # to a valid filename
-                        self.eq.push('FS_FILE_CREATE', t_path)
-                        self.eq.push('FS_FILE_CLOSE_WRITE', t_path)
+                        self.eq_push('FS_FILE_CREATE', t_path)
+                        self.eq_push('FS_FILE_CLOSE_WRITE', t_path)
 
                     self.held_event = None
                 return
@@ -279,7 +326,12 @@
             self.push_event(event)
             t_path = os.path.join(event.path, event.name)
             if not os.path.isdir(t_path):
-                self.eq.push('FS_FILE_CLOSE_WRITE', t_path)
+                self.eq_push('FS_FILE_CLOSE_WRITE', t_path)
+
+    def eq_push(self, *event_data):
+        '''Sends to EQ the event data, maybe filtering it.'''
+        if not self._to_mute.pop(event_data):
+            self.eq.push(*event_data)
 
     def process_default(self, event):
         '''Push the event into the EventQueue.'''
@@ -313,7 +365,7 @@
         if not self.is_ignored(fullpath):
             if evt_name == 'FS_DIR_DELETE':
                 self.handle_dir_delete(fullpath)
-            self.eq.push(evt_name, fullpath)
+            self.eq_push(evt_name, fullpath)
 
     def freeze_begin(self, path):
         """Puts in hold all the events for this path."""
@@ -346,7 +398,7 @@
         # push the received events
         for evt_name, path in events:
             if not self.is_ignored(path):
-                self.eq.push(evt_name, path)
+                self.eq_push(evt_name, path)
 
         self.frozen_path = None
         self.frozen_evts = False
@@ -360,9 +412,9 @@
             if path == fullpath:
                 continue
             if is_dir:
-                self.eq.push('FS_DIR_DELETE', path)
+                self.eq_push('FS_DIR_DELETE', path)
             else:
-                self.eq.push('FS_FILE_DELETE', path)
+                self.eq_push('FS_FILE_DELETE', path)
 
 
 class EventQueue(object):
@@ -384,6 +436,10 @@
         self.dispatch_queue = Queue()
         self.empty_event_queue_callbacks = set()
 
+    def add_to_mute_filter(self, *info):
+        '''Adds info to mute filter in the processor.'''
+        self._processor.add_to_mute_filter(*info)
+
     def add_empty_event_queue_callback(self, callback):
         """add a callback for when the even queue has no more events."""
         self.empty_event_queue_callbacks.add(callback)

=== modified file 'ubuntuone/syncdaemon/filesystem_manager.py'
--- ubuntuone/syncdaemon/filesystem_manager.py	2009-10-13 16:03:54 +0000
+++ ubuntuone/syncdaemon/filesystem_manager.py	2009-10-15 15:40:24 +0000
@@ -221,6 +221,7 @@
                                     cache_compact_threshold=4)
         self.shares = {}
         self.vm = vm
+        self.eq = None  # this will be registered later
 
         # create the indexes
         self._idx_path = {}
@@ -244,9 +245,14 @@
         logger("initialized: idx_path: %s, idx_node_id: %s, shares: %s" %
                (len(self._idx_path), len(self._idx_node_id), len(self.shares)))
 
+    def register_eq(self, eq):
+        '''Registers an EventQueue here.'''
+        self.eq = eq
+
     def _safe_fs_iteritems(self):
-        """returns a 'safe' iterator over the items of the FileShelf.
-        it's 'safe' because broked metadata objects are deleted and not
+        """Returns a 'safe' iterator over the items of the FileShelf.
+
+        It's 'safe' because broken metadata objects are deleted and not
         returned.
         """
         def safeget_mdobj(mdid):
@@ -566,6 +572,11 @@
         # pylint: disable-msg=W0704
         try:
             with contextlib.nested(from_context, to_context):
+                if mdobj["is_dir"]:
+                    expected_event = "FS_DIR_MOVE"
+                else:
+                    expected_event = "FS_FILE_MOVE"
+                self.eq.add_to_mute_filter(expected_event, path_from, path_to)
                 shutil.move(path_from, path_to)
         except IOError, e:
             # file was not yet created
@@ -652,8 +663,10 @@
         try:
             with self._enable_share_write(mdobj['share_id'], path):
                 if self.is_dir(path=path):
+                    self.eq.add_to_mute_filter("FS_DIR_DELETE", path)
                     os.rmdir(path)
                 else:
+                    self.eq.add_to_mute_filter("FS_FILE_DELETE", path)
                     os.remove(path)
         except OSError, e:
             if e.errno == errno.ENOTEMPTY:
@@ -674,6 +687,11 @@
             to_path = base_to_path + "." + str(ind)
         with self._enable_share_write(mdobj['share_id'], path):
             try:
+                if mdobj["is_dir"]:
+                    expected_event = "FS_DIR_MOVE"
+                else:
+                    expected_event = "FS_FILE_MOVE"
+                self.eq.add_to_mute_filter(expected_event, path, to_path)
                 os.rename(path, to_path)
             except OSError, e:
                 if e.errno == errno.ENOENT:
@@ -739,7 +757,10 @@
         with self._enable_share_write(share_id, os.path.dirname(path)):
             # if we don't have the dir yet, create it
             if is_dir and not os.path.exists(path):
+                self.eq.add_to_mute_filter("FS_DIR_CREATE", path)
                 os.mkdir(path)
+
+        # don't alert EQ, partials are in other directory, not watched
         open(partial_path, "w").close()
         mdobj["info"]["last_partial_created"] = time.time()
         mdobj["info"]["is_partial"] = True
@@ -786,6 +807,8 @@
 
         partial_path = self._get_partial_path(mdobj)
         with self._enable_share_write(share_id, path):
+            self.eq.add_to_mute_filter("FS_FILE_CREATE", path)
+            self.eq.add_to_mute_filter("FS_FILE_CLOSE_WRITE", path)
             shutil.move(partial_path, path)
         mdobj["local_hash"] = local_hash
         mdobj["info"]["last_downloaded"] = time.time()
@@ -805,6 +828,7 @@
         partial_path = self._get_partial_path(mdobj)
         #pylint: disable-msg=W0704
         try:
+            # don't alert EQ, partials are in other directory, not watched
             os.remove(partial_path)
         except OSError, e:
             # we only remove it if its there.

=== modified file 'ubuntuone/syncdaemon/main.py'
--- ubuntuone/syncdaemon/main.py	2009-10-09 08:19:05 +0000
+++ ubuntuone/syncdaemon/main.py	2009-10-15 15:40:24 +0000
@@ -109,6 +109,7 @@
                                                        self.partials_dir,
                                                        self.vm)
         self.event_q = event_queue.EventQueue(self.fs)
+        self.fs.register_eq(self.event_q)
         self.oauth_client = OAuthClient(self.realm)
         self.state = SyncDaemonStateManager(self, handshake_timeout,
                                             max_handshake_timeouts)