From 774b646d0b5615054ec25dbf5efa85fedaf2b312 Mon Sep 17 00:00:00 2001 From: Fabio Erculiani Date: Fri, 12 Mar 2010 17:24:07 +0100 Subject: [PATCH] [entropy.services.repository] rewrite EAPI3 repository updates handling --- .../entropy/services/repository/commands.py | 53 ++--- .../entropy/services/repository/interfaces.py | 205 +++++++++--------- 2 files changed, 123 insertions(+), 135 deletions(-) diff --git a/libraries/entropy/services/repository/commands.py b/libraries/entropy/services/repository/commands.py index 8db434058..fe5c1d848 100644 --- a/libraries/entropy/services/repository/commands.py +++ b/libraries/entropy/services/repository/commands.py @@ -103,10 +103,10 @@ class Repository(SocketCommands): dbpath = self.get_database_path(repository, arch, product, branch) mtime = self.get_database_mtime(repository, arch, product, branch) - rev_mtime = self.get_revision_mtime(repository, arch, product, branch) + rev_id = self.get_database_revision(repository, arch, product, branch) cached = self.HostInterface.get_dcache( - x + ('docmd_dbdiff', mtime, rev_mtime,), repository) + x + ('docmd_dbdiff', mtime, rev_id,), repository) if cached is not None: std_checksum, secure_checksum, myids = cached else: @@ -118,7 +118,7 @@ class Repository(SocketCommands): myids = dbconn.listAllIdpackages() cached = std_checksum, secure_checksum, myids self.HostInterface.set_dcache( - x + ('docmd_dbdiff', mtime, rev_mtime,), cached, repository) + x + ('docmd_dbdiff', mtime, rev_id,), cached, repository) dbconn.closeDB() foreign_idpackages = set(foreign_idpackages) @@ -155,10 +155,10 @@ class Repository(SocketCommands): return valid mtime = self.get_database_mtime(repository, arch, product, branch) - rev_mtime = self.get_revision_mtime(repository, arch, product, branch) + rev_id = self.get_database_revision(repository, arch, product, branch) cached = self.HostInterface.get_dcache( (repository, arch, product, branch, 'docmd_repository_metadata', - mtime, rev_mtime), repository) + mtime, rev_id), repository) if cached is not None: return cached @@ -175,7 +175,7 @@ class Repository(SocketCommands): self.HostInterface.set_dcache( (repository, arch, product, branch, 'docmd_repository_metadata', - mtime, rev_mtime,), metadata, repository) + mtime, rev_id,), metadata, repository) dbconn.closeDB() return metadata @@ -201,10 +201,10 @@ class Repository(SocketCommands): return valid mtime = self.get_database_mtime(repository, arch, product, branch) - rev_mtime = self.get_revision_mtime(repository, arch, product, branch) + rev_id = self.get_database_revision(repository, arch, product, branch) cached = self.HostInterface.get_dcache( (repository, arch, product, branch, 'docmd_package_sets', - mtime, rev_mtime,), repository) + mtime, rev_id,), repository) if cached is not None: return cached @@ -216,7 +216,7 @@ class Repository(SocketCommands): self.HostInterface.set_dcache( (repository, arch, product, branch, 'docmd_package_sets', - mtime, rev_mtime,), data, repository) + mtime, rev_id,), data, repository) dbconn.closeDB() return data @@ -242,10 +242,10 @@ class Repository(SocketCommands): return valid mtime = self.get_database_mtime(repository, arch, product, branch) - rev_mtime = self.get_revision_mtime(repository, arch, product, branch) + rev_id = self.get_database_revision(repository, arch, product, branch) cached = self.HostInterface.get_dcache( (repository, arch, product, branch, 'docmd_treeupdates', mtime, - rev_mtime), repository) + rev_id), repository) if cached is not None: return cached @@ -259,7 +259,7 @@ class Repository(SocketCommands): self.HostInterface.set_dcache( (repository, arch, product, branch, 'docmd_treeupdates', mtime, - rev_mtime,), data, repository) + rev_id,), data, repository) dbconn.closeDB() return data @@ -296,10 +296,10 @@ class Repository(SocketCommands): return valid mtime = self.get_database_mtime(repository, arch, product, branch) - rev_mtime = self.get_revision_mtime(repository, arch, product, branch) + rev_id = self.get_database_revision(repository, arch, product, branch) cached = self.HostInterface.get_dcache( (repository, arch, product, branch, idpackages, - 'docmd_pkginfo_strict', mtime, rev_mtime), + 'docmd_pkginfo_strict', mtime, rev_id), repository) if cached is not None: return cached @@ -325,7 +325,7 @@ class Repository(SocketCommands): self.HostInterface.set_dcache( (repository, arch, product, branch, idpackages, - 'docmd_pkginfo_strict', mtime, rev_mtime,), + 'docmd_pkginfo_strict', mtime, rev_id,), result, repository) dbconn.closeDB() return result @@ -344,25 +344,10 @@ class Repository(SocketCommands): mtime = 0.0 return mtime - def get_revision_mtime(self, repository, arch, product, branch): - dbpath = self.get_database_revision(repository, arch, product, branch) - try: - mtime = os.path.getmtime(dbpath) - except (OSError, IOError,): - mtime = 0.0 - return mtime - def get_database_revision(self, repository, arch, product, branch): - repoitems = (repository, arch, product, branch,) - mydbroot = self.HostInterface.repositories[repoitems]['dbpath'] - dbrev_path = os.path.join(mydbroot, etpConst['etpdatabaserevisionfile']) + x = (repository, arch, product, branch,) + l_dbrev = self.HostInterface.repositories[x]['live_db_rev'] dbrev = -1 - if os.path.isfile(dbrev_path) and os.access(dbrev_path, os.R_OK): - db_f = open(dbrev_path, "r") - f_dbrev = db_f.readline().strip() - db_f.close() - try: - dbrev = int(f_dbrev) - except ValueError: - pass + if l_dbrev: + dbrev = l_dbrev return dbrev diff --git a/libraries/entropy/services/repository/interfaces.py b/libraries/entropy/services/repository/interfaces.py index 56eb8ea1e..44cc57773 100644 --- a/libraries/entropy/services/repository/interfaces.py +++ b/libraries/entropy/services/repository/interfaces.py @@ -78,7 +78,7 @@ class Server(SocketHost): self.LockScanner.kill() def start_repository_lock_scanner(self): - self.LockScanner = TimeScheduled(0.5, self.lock_scan) + self.LockScanner = TimeScheduled(5, self.lock_scan) self.LockScanner.start() def set_repository_db_availability(self, repo_tuple): @@ -102,15 +102,19 @@ class Server(SocketHost): return True def lock_scan(self): + do_clear = set() for repository, arch, product, branch in self.repositories: + x = (repository, arch, product, branch,) self.set_repository_db_availability(x) + if not self.repositories[x]['enabled']: if x in self.syscache['dbs_not_available']: continue self.syscache['dbs_not_available'].add(x) - mytxt = blue("%s.") % (_("database does not exist. Locking services for it"),) + mytxt = blue("%s.") % ( + _("database does not exist. Locking services for it"),) self.output( "[%s] %s" % ( brown(str(x)), @@ -121,57 +125,50 @@ class Server(SocketHost): ) do_clear.add(repository) continue - if os.path.isfile(self.repositories[x]['download_lock']) and \ - not self.repositories[x]['locked']: - self.repositories[x]['locked'] = True - mydbpath = os.path.join(self.repositories[x]['dbpath'], etpConst['etpdatabasefile']) - self.close_db(mydbpath) - self.eapi3_lock_repo(*x) - do_clear.add(repository) - mytxt = blue("%s.") % (_("database got locked. Locking services for it"),) - self.output( - "[%s] %s" % ( - brown(str(x)), - mytxt, - ), - importance = 1, - type = "info" - ) - elif not os.path.isfile(self.repositories[x]['download_lock']) and \ - self.repositories[x]['locked']: - # setup variables - dbpath = self.repositories[x]['dbpath'] - cmethod = self.repositories[x]['cmethod'] - cmethod_data = etpConst['etpdatabasecompressclasses'].get(cmethod) - unpack_method = cmethod_data[1] - compressed_dbfile = etpConst[cmethod_data[2]] - compressed_dbpath = os.path.join(dbpath, compressed_dbfile) + saved_rev = self.repositories[x]['live_db_rev'] + dbrev_path = os.path.join(self.repositories[x]['dbpath'], + etpConst['etpdatabaserevisionfile']) + db_path = os.path.join(self.repositories[x]['dbpath'], + etpConst['etpdatabasefile']) - if not (os.access(compressed_dbpath, os.R_OK | os.W_OK) and \ - os.path.isfile(compressed_dbpath)): - mytxt = darkred("%s: %s !!") % ( - _("cannot unlock database, compressed file not found"), - compressed_dbpath, - ) - self.output( - "[%s] %s" % ( - brown(str(x)), - mytxt, - ), - importance = 1, - type = "warning" - ) - self.syscache['dbs_not_available'].add(x) - do_clear.add(repository) - continue + cur_rev = None + if os.path.isfile(dbrev_path): + cur_f = open(dbrev_path, "r") + cur_rev = cur_f.readline().strip() + cur_f.close() - # make sure this db is closed - mydbpath = os.path.join(dbpath, etpConst['etpdatabasefile']) - self.close_db(mydbpath) + if cur_rev == saved_rev: + continue - mytxt = blue("%s: %s") % ( - _("unpacking compressed database"), + self.repositories[x]['locked'] = True + # trash old databases + self.close_db(db_path) + do_clear.add(repository) + mytxt = blue("%s.") % ( + _("repository changed. Updating metadata"),) + self.output( + "[%s] %s" % ( + brown(str(x)), + mytxt, + ), + importance = 1, + type = "info" + ) + + # now unpack and unlock + dbpath = self.repositories[x]['dbpath'] + cmethod = self.repositories[x]['cmethod'] + cmethod_data = etpConst['etpdatabasecompressclasses'].get( + cmethod) + unpack_method = cmethod_data[1] + compressed_dbfile = etpConst[cmethod_data[2]] + compressed_dbpath = os.path.join(dbpath, compressed_dbfile) + + if not (os.access(compressed_dbpath, os.R_OK | os.W_OK) and \ + os.path.isfile(compressed_dbpath)): + mytxt = darkred("%s: %s !!") % ( + _("cannot unlock database, compressed file not found"), compressed_dbpath, ) self.output( @@ -180,62 +177,71 @@ class Server(SocketHost): mytxt, ), importance = 1, - type = "info" + type = "warning" ) + self.syscache['dbs_not_available'].add(x) + do_clear.add(repository) + continue - # now unpack compressed db in place - unpack_func = getattr(entropy.tools, unpack_method) - generated_outpath = unpack_func(compressed_dbpath) - if mydbpath != generated_outpath: - try: - os.rename(generated_outpath, mydbpath) - except OSError: - shutil.move(generated_outpath, mydbpath) + # make sure this db is closed + self.close_db(db_path) - mytxt = blue("%s. %s:") % ( - _("unlocking and indexing database"), - _("hash"), - ) - self.output( - "[%s] %s" % ( - brown(str(x)), - mytxt, - ), - importance = 1, - type = "info" - ) - # woohoo, got unlocked eventually - mydb = self.open_db(mydbpath, docache = False) - mydb.createAllIndexes() - self.output( - darkgreen(str(mydb.checksum(do_order = True, strict = False, strings = True))), - importance = 1, - type = "info" - ) - mydb.closeDB() - self.__cacher.discard() - EntropyCacher.clear_cache_item( - Server.CACHE_ID+"/"+repository+"/") - self.repositories[x]['locked'] = False - self.eapi3_unlock_repo(*x) + mytxt = blue("%s: %s") % ( + _("unpacking compressed database"), + compressed_dbpath, + ) + self.output( + "[%s] %s" % ( + brown(str(x)), + mytxt, + ), + importance = 1, + type = "info" + ) + + # now unpack compressed db in place + unpack_func = getattr(entropy.tools, unpack_method) + generated_outpath = unpack_func(compressed_dbpath) + if db_path != generated_outpath: + try: + os.rename(generated_outpath, db_path) + except OSError: + shutil.move(generated_outpath, db_path) + + mytxt = blue("%s. %s:") % ( + _("unlocking and indexing database"), + _("hash"), + ) + self.output( + "[%s] %s" % ( + brown(str(x)), + mytxt, + ), + importance = 1, + type = "info" + ) + # woohoo, got unlocked eventually + mydb = self.open_db(db_path, docache = False) + mydb.createAllIndexes() + db_ck = mydb.checksum(do_order = True, strict = False, + strings = True) + self.output( + darkgreen(str(db_ck)), + importance = 1, + type = "info" + ) + mydb.closeDB() + self.__cacher.discard() + EntropyCacher.clear_cache_item( + Server.CACHE_ID+"/"+repository+"/") + + self.repositories[x]['live_db_rev'] = cur_rev + self.repositories[x]['locked'] = False self.__cacher.discard() for repo in do_clear: EntropyCacher.clear_cache_item(Server.CACHE_ID+"/"+repo+"/") - def eapi3_lock_repo(self, repository, arch, product, branch): - lock_file = os.path.join(self.repositories[(repository, arch, product, branch,)]['dbpath'], etpConst['etpdatabaseeapi3lockfile']) - if not os.path.lexists(lock_file): - f = open(lock_file, "w") - f.write("this repository is EAPI3 locked") - f.flush() - f.close() - - def eapi3_unlock_repo(self, repository, arch, product, branch): - lock_file = os.path.join(self.repositories[(repository, arch, product, branch,)]['dbpath'], etpConst['etpdatabaseeapi3lockfile']) - if os.path.isfile(lock_file): - os.remove(lock_file) - def get_dcache(self, item, repo = '_norepo_'): return entropy.dump.loadobj(Server.CACHE_ID+"/"+repo+"/"+str(hash(item))) @@ -286,10 +292,7 @@ class Server(SocketHost): continue break self.repositories[x]['dbrevision'] = myrev - self.repositories[x]['download_lock'] = os.path.join( - mydbpath, - etpConst['etpdatabasedownloadlockfile'] - ) + self.repositories[x]['live_db_rev'] = None if 'cmethod' not in self.repositories[x]: raise AttributeError("cmethod not specified for: %s" % (x,)) if self.repositories[x]['cmethod'] not in etpConst['etpdatabasesupportedcformats']: