ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2025-02-21 18:06:46 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'pr371'

Browse Source 
Closes tahoe-lafs/tahoe-lafs#371
This commit is contained in:
Brian Warner 2016-10-25 12:39:32 -07:00
commit 233d0e29d9
5 changed files with 66 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/allmydata/client.py
View File

@ -88,6 +88,7 @@ def _valid_config_sections():
"download.umask", "download.umask",
"enabled", "enabled",
"local.directory", "local.directory",
"poll_interval",
), ),
}) })
return cfg return cfg
@ -561,18 +562,24 @@ class Client(node.Node, pollmixin.PollMixin):
if self.get_config("magic_folder", "enabled", False, boolean=True): if self.get_config("magic_folder", "enabled", False, boolean=True):
#print "magic folder enabled" #print "magic folder enabled"
upload_dircap = self.get_private_config("magic_folder_dircap")
collective_dircap = self.get_private_config("collective_dircap")
local_dir_config = self.get_config("magic_folder", "local.directory").decode("utf-8")
local_dir = abspath_expanduser_unicode(local_dir_config, base=self.basedir)
dbfile = os.path.join(self.basedir, "private", "magicfolderdb.sqlite")
dbfile = abspath_expanduser_unicode(dbfile)
from allmydata.frontends import magic_folder from allmydata.frontends import magic_folder
umask = self.get_config("magic_folder", "download.umask", 0077)
s = magic_folder.MagicFolder(self, upload_dircap, collective_dircap, local_dir, dbfile, umask) db_filename = os.path.join(self.basedir, "private", "magicfolderdb.sqlite")
local_dir_config = self.get_config("magic_folder", "local.directory").decode("utf-8")
try:
poll_interval = int(self.get_config("magic_folder", "poll_interval", 3))
except ValueError:
raise ValueError("[magic_folder] poll_interval must be an int")
s = magic_folder.MagicFolder(
client=self,
upload_dircap=self.get_private_config("magic_folder_dircap"),
collective_dircap=self.get_private_config("collective_dircap"),
local_path_u=abspath_expanduser_unicode(local_dir_config, base=self.basedir),
dbfile=abspath_expanduser_unicode(db_filename),
umask=self.get_config("magic_folder", "download.umask", 0077),
downloader_delay=poll_interval,
)
self._magic_folder = s self._magic_folder = s
s.setServiceParent(self) s.setServiceParent(self)
s.startService() s.startService()

70
src/allmydata/frontends/magic_folder.py
View File

@ -63,7 +63,7 @@ class MagicFolder(service.MultiService):
name = 'magic-folder' name = 'magic-folder'
def __init__(self, client, upload_dircap, collective_dircap, local_path_u, dbfile, umask, def __init__(self, client, upload_dircap, collective_dircap, local_path_u, dbfile, umask,
pending_delay=1.0, clock=None): uploader_delay=1.0, clock=None, downloader_delay=3):
precondition_abspath(local_path_u) precondition_abspath(local_path_u)
service.MultiService.__init__(self) service.MultiService.__init__(self)
@ -80,10 +80,10 @@ class MagicFolder(service.MultiService):
upload_dirnode = self._client.create_node_from_uri(upload_dircap) upload_dirnode = self._client.create_node_from_uri(upload_dircap)
collective_dirnode = self._client.create_node_from_uri(collective_dircap) collective_dirnode = self._client.create_node_from_uri(collective_dircap)
self.uploader = Uploader(client, local_path_u, db, upload_dirnode, pending_delay, clock) self.uploader = Uploader(client, local_path_u, db, upload_dirnode, uploader_delay, clock)
self.downloader = Downloader(client, local_path_u, db, collective_dirnode, self.downloader = Downloader(client, local_path_u, db, collective_dirnode,
upload_dirnode.get_readonly_uri(), clock, self.uploader.is_pending, umask, upload_dirnode.get_readonly_uri(), clock, self.uploader.is_pending, umask,
self.set_public_status) self.set_public_status, poll_interval=downloader_delay)
self._public_status = (False, ['Magic folder has not yet started']) self._public_status = (False, ['Magic folder has not yet started'])
def enable_debug_log(self, enabled=True): def enable_debug_log(self, enabled=True):
@ -131,9 +131,8 @@ class MagicFolder(service.MultiService):
class QueueMixin(HookMixin): class QueueMixin(HookMixin):
scan_interval = 0
def __init__(self, client, local_path_u, db, name, clock, delay=0): def __init__(self, client, local_path_u, db, name, clock):
self._client = client self._client = client
self._local_path_u = local_path_u self._local_path_u = local_path_u
self._local_filepath = to_filepath(local_path_u) self._local_filepath = to_filepath(local_path_u)
@ -163,9 +162,6 @@ class QueueMixin(HookMixin):
# do we also want to bound on "maximum age"? # do we also want to bound on "maximum age"?
self._process_history = deque(maxlen=20) self._process_history = deque(maxlen=20)
self._stopped = False self._stopped = False
# XXX pass in an initial value for this; it seems like .10 broke this and it's always 0
self._turn_delay = delay
self._log('delay is %f' % self._turn_delay)
# a Deferred to wait for the _do_processing() loop to exit # a Deferred to wait for the _do_processing() loop to exit
# (gets set to the return from _do_processing() if we get that # (gets set to the return from _do_processing() if we get that
@ -208,40 +204,36 @@ class QueueMixin(HookMixin):
This is an infinite loop that processes things out of the _deque. This is an infinite loop that processes things out of the _deque.
One iteration runs self._process_deque which calls One iteration runs self._process_deque which calls
_when_queue_is_empty() and then completely drains the _deque _perform_scan() and then completely drains the _deque
(processing each item). After that we yield for _turn_deque (processing each item). After that we yield for _turn_deque
seconds. seconds.
""" """
# we subtract here so there's a scan on the very first iteration
last_scan = self._clock.seconds() - self.scan_interval
while not self._stopped: while not self._stopped:
self._log("doing iteration") self._log("doing iteration")
d = task.deferLater(self._clock, self._turn_delay, lambda: None) d = task.deferLater(self._clock, self._scan_delay(), lambda: None)
# ">=" is important here if scan scan_interval is 0
if self._clock.seconds() - last_scan >= self.scan_interval: # adds items to our deque
# XXX can't we unify the "_full_scan" vs what yield self._perform_scan()
# Downloader does...
last_scan = self._clock.seconds()
yield self._when_queue_is_empty() # (this no-op for us, only Downloader uses it...)
self._log("did scan; now %d" % last_scan)
else:
self._log("skipped scan")
# process anything in our queue # process anything in our queue
yield self._process_deque() yield self._process_deque()
self._log("one loop; call_hook iteration %r" % self)
self._call_hook(None, 'iteration')
# we want to have our callLater queued in the reactor # we want to have our callLater queued in the reactor
# *before* we trigger the 'iteration' hook, so that hook # *before* we trigger the 'iteration' hook, so that hook
# can successfully advance the Clock and bypass the delay # can successfully advance the Clock and bypass the delay
# if required (e.g. in the tests). # if required (e.g. in the tests).
self._log("one loop; call_hook iteration %r" % self)
self._call_hook(None, 'iteration')
if not self._stopped: if not self._stopped:
self._log("waiting... %r" % d) self._log("waiting... %r" % d)
yield d yield d
self._log("stopped") self._log("stopped")
def _when_queue_is_empty(self): def _scan_delay(self):
raise NotImplementedError
def _perform_scan(self):
return return
@defer.inlineCallbacks @defer.inlineCallbacks
@ -346,7 +338,7 @@ class UploadItem(QueuedItem):
class Uploader(QueueMixin): class Uploader(QueueMixin):
def __init__(self, client, local_path_u, db, upload_dirnode, pending_delay, clock): def __init__(self, client, local_path_u, db, upload_dirnode, pending_delay, clock):
QueueMixin.__init__(self, client, local_path_u, db, 'uploader', clock, delay=pending_delay) QueueMixin.__init__(self, client, local_path_u, db, 'uploader', clock)
self.is_ready = False self.is_ready = False
@ -360,8 +352,9 @@ class Uploader(QueueMixin):
self._upload_dirnode = upload_dirnode self._upload_dirnode = upload_dirnode
self._inotify = get_inotify_module() self._inotify = get_inotify_module()
self._notifier = self._inotify.INotify() self._notifier = self._inotify.INotify()
self._pending = set() # of unicode relpaths
self._pending = set() # of unicode relpaths
self._pending_delay = pending_delay
self._periodic_full_scan_duration = 10 * 60 # perform a full scan every 10 minutes self._periodic_full_scan_duration = 10 * 60 # perform a full scan every 10 minutes
self._periodic_callid = None self._periodic_callid = None
@ -423,6 +416,9 @@ class Uploader(QueueMixin):
# *really* just call this synchronously. # *really* just call this synchronously.
return self._begin_processing(None) return self._begin_processing(None)
def _scan_delay(self):
return self._pending_delay
def _full_scan(self): def _full_scan(self):
self._periodic_callid = self._clock.callLater(self._periodic_full_scan_duration, self._full_scan) self._periodic_callid = self._clock.callLater(self._periodic_full_scan_duration, self._full_scan)
self._log("FULL SCAN") self._log("FULL SCAN")
@ -753,12 +749,11 @@ class DownloadItem(QueuedItem):
class Downloader(QueueMixin, WriteFileMixin): class Downloader(QueueMixin, WriteFileMixin):
scan_interval = 3
def __init__(self, client, local_path_u, db, collective_dirnode, def __init__(self, client, local_path_u, db, collective_dirnode,
upload_readonly_dircap, clock, is_upload_pending, umask, upload_readonly_dircap, clock, is_upload_pending, umask,
status_reporter): status_reporter, poll_interval=3):
QueueMixin.__init__(self, client, local_path_u, db, 'downloader', clock, delay=self.scan_interval) QueueMixin.__init__(self, client, local_path_u, db, 'downloader', clock)
if not IDirectoryNode.providedBy(collective_dirnode): if not IDirectoryNode.providedBy(collective_dirnode):
raise AssertionError("The URI in '%s' does not refer to a directory." raise AssertionError("The URI in '%s' does not refer to a directory."
@ -772,11 +767,11 @@ class Downloader(QueueMixin, WriteFileMixin):
self._is_upload_pending = is_upload_pending self._is_upload_pending = is_upload_pending
self._umask = umask self._umask = umask
self._status_reporter = status_reporter self._status_reporter = status_reporter
self._poll_interval = poll_interval
@defer.inlineCallbacks @defer.inlineCallbacks
def start_downloading(self): def start_downloading(self):
self._log("start_downloading") self._log("start_downloading")
self._turn_delay = self.scan_interval
files = self._db.get_all_relpaths() files = self._db.get_all_relpaths()
self._log("all files %s" % files) self._log("all files %s" % files)
@ -794,7 +789,7 @@ class Downloader(QueueMixin, WriteFileMixin):
"Last tried at %s" % self.nice_current_time(), "Last tried at %s" % self.nice_current_time(),
) )
twlog.msg("Magic Folder failed initial scan: %s" % (e,)) twlog.msg("Magic Folder failed initial scan: %s" % (e,))
yield task.deferLater(self._clock, self.scan_interval, lambda: None) yield task.deferLater(self._clock, self._poll_interval, lambda: None)
def nice_current_time(self): def nice_current_time(self):
return format_time(datetime.fromtimestamp(self._clock.seconds()).timetuple()) return format_time(datetime.fromtimestamp(self._clock.seconds()).timetuple())
@ -948,13 +943,15 @@ class Downloader(QueueMixin, WriteFileMixin):
d.addCallback(_filter_batch_to_deque) d.addCallback(_filter_batch_to_deque)
return d return d
def _scan_delay(self):
return self._poll_interval
@defer.inlineCallbacks @defer.inlineCallbacks
def _when_queue_is_empty(self): def _perform_scan(self):
# XXX can we amalgamate all the "scan" stuff and just call it
# directly from QueueMixin?
x = None x = None
try: try:
x = yield self._scan(None) x = yield self._scan_remote_collective()
self._status_reporter( self._status_reporter(
True, 'Magic folder is working', True, 'Magic folder is working',
'Last scan: %s' % self.nice_current_time(), 'Last scan: %s' % self.nice_current_time(),
@ -968,9 +965,6 @@ class Downloader(QueueMixin, WriteFileMixin):
) )
defer.returnValue(x) defer.returnValue(x)
def _scan(self, ign):
return self._scan_remote_collective()
def _process(self, item): def _process(self, item):
# Downloader # Downloader
self._log("_process(%r)" % (item,)) self._log("_process(%r)" % (item,))

8
src/allmydata/test/cli/test_magic_folder.py
View File

@ -122,8 +122,8 @@ class MagicFolderCLITestMixin(CLITestMixin, GridTestMixin, NonASCIIPathMixin):
d = defer.succeed(None) d = defer.succeed(None)
def _clean(ign): def _clean(ign):
d = self.magicfolder.finish() d = self.magicfolder.finish()
self.magicfolder.uploader._clock.advance(self.magicfolder.uploader.scan_interval + 1) self.magicfolder.uploader._clock.advance(self.magicfolder.uploader._pending_delay + 1)
self.magicfolder.downloader._clock.advance(self.magicfolder.downloader.scan_interval + 1) self.magicfolder.downloader._clock.advance(self.magicfolder.downloader._poll_interval + 1)
return d return d
d.addCallback(_clean) d.addCallback(_clean)
@ -139,10 +139,10 @@ class MagicFolderCLITestMixin(CLITestMixin, GridTestMixin, NonASCIIPathMixin):
local_path_u=local_magic_dir, local_path_u=local_magic_dir,
dbfile=dbfile, dbfile=dbfile,
umask=0o077, umask=0o077,
pending_delay=0.2,
clock=clock, clock=clock,
uploader_delay=0.2,
downloader_delay=0,
) )
magicfolder.downloader._turn_delay = 0
magicfolder.setServiceParent(self.get_client(client_num)) magicfolder.setServiceParent(self.get_client(client_num))
magicfolder.ready() magicfolder.ready()

10
src/allmydata/test/test_client.py
View File

@ -322,14 +322,14 @@ class Basic(testutil.ReallyEqualMixin, testutil.NonASCIIPathMixin, unittest.Test
class MockMagicFolder(service.MultiService): class MockMagicFolder(service.MultiService):
name = 'magic-folder' name = 'magic-folder'
def __init__(self, client, upload_dircap, collective_dircap, local_dir, dbfile, umask, inotify=None, def __init__(self, client, upload_dircap, collective_dircap, local_path_u, dbfile, umask, inotify=None,
pending_delay=1.0): uploader_delay=1.0, clock=None, downloader_delay=3):
service.MultiService.__init__(self) service.MultiService.__init__(self)
self.client = client self.client = client
self._umask = umask self._umask = umask
self.upload_dircap = upload_dircap self.upload_dircap = upload_dircap
self.collective_dircap = collective_dircap self.collective_dircap = collective_dircap
self.local_dir = local_dir self.local_dir = local_path_u
self.dbfile = dbfile self.dbfile = dbfile
self.inotify = inotify self.inotify = inotify
@ -376,8 +376,8 @@ class Basic(testutil.ReallyEqualMixin, testutil.NonASCIIPathMixin, unittest.Test
class Boom(Exception): class Boom(Exception):
pass pass
def BoomMagicFolder(client, upload_dircap, collective_dircap, local_dir, dbfile, def BoomMagicFolder(client, upload_dircap, collective_dircap, local_path_u, dbfile,
inotify=None, pending_delay=1.0): umask, inotify=None, uploader_delay=1.0, clock=None, downloader_delay=3):
raise Boom() raise Boom()
self.patch(allmydata.frontends.magic_folder, 'MagicFolder', BoomMagicFolder) self.patch(allmydata.frontends.magic_folder, 'MagicFolder', BoomMagicFolder)

14
src/allmydata/test/test_magic_folder.py
View File

@ -111,13 +111,13 @@ def iterate_downloader(magic):
# can do either of these: # can do either of these:
#d = magic.downloader._process_deque() #d = magic.downloader._process_deque()
d = magic.downloader.set_hook('iteration') d = magic.downloader.set_hook('iteration')
magic.downloader._clock.advance(magic.downloader.scan_interval + 1) magic.downloader._clock.advance(magic.downloader._poll_interval + 1)
return d return d
def iterate_uploader(magic): def iterate_uploader(magic):
d = magic.uploader.set_hook('iteration') d = magic.uploader.set_hook('iteration')
magic.uploader._clock.advance(magic.uploader.scan_interval + 1) magic.uploader._clock.advance(magic.uploader._pending_delay + 1)
return d return d
@defer.inlineCallbacks @defer.inlineCallbacks
@ -308,7 +308,7 @@ class MagicFolderAliceBobTestMixin(MagicFolderCLITestMixin, ShouldFailMixin, Rea
self.alice_fileops = FileOperationsHelper(self.alice_magicfolder.uploader, self.inject_inotify) self.alice_fileops = FileOperationsHelper(self.alice_magicfolder.uploader, self.inject_inotify)
d0 = self.alice_magicfolder.uploader.set_hook('iteration') d0 = self.alice_magicfolder.uploader.set_hook('iteration')
d1 = self.alice_magicfolder.downloader.set_hook('iteration') d1 = self.alice_magicfolder.downloader.set_hook('iteration')
self.alice_clock.advance(self.alice_magicfolder.uploader.scan_interval + 1) self.alice_clock.advance(self.alice_magicfolder.uploader._pending_delay + 1)
d0.addCallback(lambda ign: d1) d0.addCallback(lambda ign: d1)
d0.addCallback(lambda ign: result) d0.addCallback(lambda ign: result)
return d0 return d0
@ -332,7 +332,7 @@ class MagicFolderAliceBobTestMixin(MagicFolderCLITestMixin, ShouldFailMixin, Rea
self.bob_fileops = FileOperationsHelper(self.bob_magicfolder.uploader, self.inject_inotify) self.bob_fileops = FileOperationsHelper(self.bob_magicfolder.uploader, self.inject_inotify)
d0 = self.bob_magicfolder.uploader.set_hook('iteration') d0 = self.bob_magicfolder.uploader.set_hook('iteration')
d1 = self.bob_magicfolder.downloader.set_hook('iteration') d1 = self.bob_magicfolder.downloader.set_hook('iteration')
self.bob_clock.advance(self.alice_magicfolder.uploader.scan_interval + 1) self.bob_clock.advance(self.alice_magicfolder.uploader._pending_delay + 1)
d0.addCallback(lambda ign: d1) d0.addCallback(lambda ign: d1)
d0.addCallback(lambda ign: result) d0.addCallback(lambda ign: result)
return d0 return d0
@ -346,8 +346,8 @@ class MagicFolderAliceBobTestMixin(MagicFolderCLITestMixin, ShouldFailMixin, Rea
d1 = self.bob_magicfolder.finish() d1 = self.bob_magicfolder.finish()
for mf in [self.alice_magicfolder, self.bob_magicfolder]: for mf in [self.alice_magicfolder, self.bob_magicfolder]:
for loader in [mf.uploader, mf.downloader]: mf.uploader._clock.advance(mf.uploader._pending_delay + 1)
loader._clock.advance(loader.scan_interval + 1) mf.downloader._clock.advance(mf.downloader._poll_interval + 1)
yield d0 yield d0
yield d1 yield d1
@ -541,7 +541,7 @@ class MagicFolderAliceBobTestMixin(MagicFolderCLITestMixin, ShouldFailMixin, Rea
# now, we ONLY want to do the scan, not a full iteration of # now, we ONLY want to do the scan, not a full iteration of
# the process loop. So we do just the scan part "by hand" in # the process loop. So we do just the scan part "by hand" in
# Bob's downloader # Bob's downloader
yield self.bob_magicfolder.downloader._when_queue_is_empty() yield self.bob_magicfolder.downloader._perform_scan()
# while we're delving into internals, I guess we might as well # while we're delving into internals, I guess we might as well
# confirm that we did queue up an item to download # confirm that we did queue up an item to download
self.assertEqual(1, len(self.bob_magicfolder.downloader._deque)) self.assertEqual(1, len(self.bob_magicfolder.downloader._deque))