[gentoo-portage-dev] [PATCH v2] Log changes between vdb_metadata.pickle updates

[Zac Medico]

This adds add support to generate a vdb_metadata_delta.json file
which tracks package merges / unmerges that occur between updates to
vdb_metadata.pickle. IndexedVardb can use the delta together with
vdb_metadata.pickle to reconstruct a complete view of /var/db/pkg,
so that it can avoid expensive listdir calls in /var/db/pkg/*.
Note that vdb_metadata.pickle is only updated periodically, in
order to avoid excessive re-writes of a large file.

In order to test the performance gains from this patch, you need to
generate /var/cache/edb/vdb_metadata_delta.json first, which will
happen automatically if you run 'emerge -p anything' with root
privileges.
---
This updated patch splits out a VdbMetadataDelta class into a new file, reducing
vartree.py bloat.

 pym/portage/dbapi/IndexedVardb.py     |  22 ++++-
 pym/portage/dbapi/VdbMetadataDelta.py | 156 ++++++++++++++++++++++++++++++++++
 pym/portage/dbapi/vartree.py          |  42 ++++++---
 3 files changed, 209 insertions(+), 11 deletions(-)
 create mode 100644 pym/portage/dbapi/VdbMetadataDelta.py

diff --git a/pym/portage/dbapi/IndexedVardb.py b/pym/portage/dbapi/IndexedVardb.py
index 424defc..38bfeed 100644
--- a/pym/portage/dbapi/IndexedVardb.py
+++ b/pym/portage/dbapi/IndexedVardb.py
@@ -3,6 +3,7 @@
 
 import portage
 from portage.dep import Atom
+from portage.exception import InvalidData
 from portage.versions import _pkg_str
 
 class IndexedVardb(object):
@@ -42,7 +43,26 @@ class IndexedVardb(object):
 		if self._cp_map is not None:
 			return iter(sorted(self._cp_map))
 
-		return self._iter_cp_all()
+		delta_data = self._vardb._cache_delta.loadRace()
+		if delta_data is None:
+			return self._iter_cp_all()
+
+		self._vardb._cache_delta.applyDelta(delta_data)
+
+		self._cp_map = cp_map = {}
+		for cpv in self._vardb._aux_cache["packages"]:
+			try:
+				cpv = _pkg_str(cpv)
+			except InvalidData:
+				continue
+
+			cp_list = cp_map.get(cpv.cp)
+			if cp_list is None:
+				cp_list = []
+				cp_map[cpv.cp] = cp_list
+			cp_list.append(cpv)
+
+		return iter(sorted(self._cp_map))
 
 	def _iter_cp_all(self):
 		self._cp_map = cp_map = {}
diff --git a/pym/portage/dbapi/VdbMetadataDelta.py b/pym/portage/dbapi/VdbMetadataDelta.py
new file mode 100644
index 0000000..e6a5c47
--- /dev/null
+++ b/pym/portage/dbapi/VdbMetadataDelta.py
@@ -0,0 +1,156 @@
+# Copyright 2014 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+
+import errno
+import io
+import json
+import os
+
+from portage import _encodings
+from portage.util import atomic_ofstream
+
+class VdbMetadataDelta(object):
+
+	_format_version  = "1"
+
+	def __init__(self, vardb):
+		self._vardb = vardb
+
+	def initialize(self, timestamp):
+		f = atomic_ofstream(self._vardb._cache_delta_filename, 'w',
+			encoding=_encodings['repo.content'], errors='strict')
+		json.dump({
+			"version": self._format_version,
+			"timestamp": timestamp
+			}, f, ensure_ascii=False)
+		f.close()
+
+	def load(self):
+
+		if not os.path.exists(self._vardb._aux_cache_filename):
+			# If the primary cache doesn't exist yet, then
+			# we can't record a delta against it.
+			return None
+
+		try:
+			with io.open(self._vardb._cache_delta_filename, 'r',
+				encoding=_encodings['repo.content'],
+				errors='strict') as f:
+				cache_obj = json.load(f)
+		except EnvironmentError as e:
+			if e.errno not in (errno.ENOENT, errno.ESTALE):
+				raise
+		except (SystemExit, KeyboardInterrupt):
+			raise
+		except Exception:
+			# Corrupt, or not json format.
+			pass
+		else:
+			try:
+				version = cache_obj["version"]
+			except KeyError:
+				pass
+			else:
+				# If the timestamp recorded in the deltas file
+				# doesn't match aux_cache_timestamp, then the
+				# deltas are not valid. This means that deltas
+				# cannot be recorded until after the next
+				# vdb_metadata.pickle update, in order to
+				# guarantee consistency.
+				if version == self._format_version:
+					try:
+						deltas = cache_obj["deltas"]
+					except KeyError:
+						cache_obj["deltas"] = deltas = []
+
+					if isinstance(deltas, list):
+						return cache_obj
+
+		return None
+
+	def loadRace(self):
+		"""
+		This calls self.load() and validates the timestamp
+		against the currently loaded self._vardb._aux_cache. If a
+		concurrent update causes the timestamps to be inconsistent,
+		then it reloads the caches and tries one more time before
+		it aborts. In practice, the race is very unlikely, so
+		this will usually succeed on the first try.
+		"""
+
+		tries = 2
+		while tries:
+			tries -= 1
+			cache_delta = self.load()
+			if cache_delta is not None and \
+				cache_delta.get("timestamp") != \
+				self._vardb._aux_cache.get("timestamp", False):
+				self._vardb._aux_cache_obj = None
+			else:
+				return cache_delta
+
+		return None
+
+	def recordEvent(self, event, cpv, slot, counter):
+
+		self._vardb.lock()
+		try:
+			deltas_obj = self.load()
+
+			if deltas_obj is None:
+				# We can't record meaningful deltas without
+				# a pre-existing state.
+				return
+
+			delta_node = {
+				"event": event,
+				"package": cpv.cp,
+				"version": cpv.version,
+				"slot": slot,
+				"counter": "%s" % counter
+			}
+
+			deltas_obj["deltas"].append(delta_node)
+
+			# Eliminate earlier nodes cancelled out by later nodes
+			# that have identical package and slot attributes.
+			filtered_list = []
+			slot_keys = set()
+			version_keys = set()
+			for delta_node in reversed(deltas_obj["deltas"]):
+				slot_key = (delta_node["package"],
+					delta_node["slot"])
+				version_key = (delta_node["package"],
+					delta_node["version"])
+				if not (slot_key in slot_keys or \
+					version_key in version_keys):
+					filtered_list.append(delta_node)
+					slot_keys.add(slot_key)
+					version_keys.add(version_key)
+
+			filtered_list.reverse()
+			deltas_obj["deltas"] = filtered_list
+
+			f = atomic_ofstream(self._vardb._cache_delta_filename,
+				mode='w', encoding=_encodings['repo.content'])
+			json.dump(deltas_obj, f, ensure_ascii=False)
+			f.close()
+
+		finally:
+			self._vardb.unlock()
+
+	def applyDelta(self, data):
+		packages = self._vardb._aux_cache["packages"]
+		for delta in data["deltas"]:
+			cpv = delta["package"] + "-" + delta["version"]
+			event = delta["event"]
+			if event == "add":
+				# Use aux_get to populate the cache
+				# for this cpv.
+				if cpv not in packages:
+					try:
+						self._vardb.aux_get(cpv, ["DESCRIPTION"])
+					except KeyError:
+						pass
+			elif event == "remove":
+				packages.pop(cpv, None)
diff --git a/pym/portage/dbapi/vartree.py b/pym/portage/dbapi/vartree.py
index 6ab4b92..e0cd5f1 100644
--- a/pym/portage/dbapi/vartree.py
+++ b/pym/portage/dbapi/vartree.py
@@ -63,6 +63,7 @@ from portage import _os_merge
 from portage import _selinux_merge
 from portage import _unicode_decode
 from portage import _unicode_encode
+from .VdbMetadataDelta import VdbMetadataDelta
 
 from _emerge.EbuildBuildDir import EbuildBuildDir
 from _emerge.EbuildPhase import EbuildPhase
@@ -177,6 +178,9 @@ class vardbapi(dbapi):
 		self._aux_cache_obj = None
 		self._aux_cache_filename = os.path.join(self._eroot,
 			CACHE_PATH, "vdb_metadata.pickle")
+		self._cache_delta_filename = os.path.join(self._eroot,
+			CACHE_PATH, "vdb_metadata_delta.json")
+		self._cache_delta = VdbMetadataDelta(self)
 		self._counter_path = os.path.join(self._eroot,
 			CACHE_PATH, "counter")
 
@@ -556,22 +560,31 @@ class vardbapi(dbapi):
 		long as at least part of the cache is still valid)."""
 		if self._flush_cache_enabled and \
 			self._aux_cache is not None and \
-			len(self._aux_cache["modified"]) >= self._aux_cache_threshold and \
-			secpass >= 2:
+			secpass >= 2 and \
+			(len(self._aux_cache["modified"]) >= self._aux_cache_threshold or
+			not os.path.exists(self._cache_delta_filename)):
+
+			ensure_dirs(os.path.dirname(self._aux_cache_filename))
+
 			self._owners.populate() # index any unindexed contents
 			valid_nodes = set(self.cpv_all())
 			for cpv in list(self._aux_cache["packages"]):
 				if cpv not in valid_nodes:
 					del self._aux_cache["packages"][cpv]
 			del self._aux_cache["modified"]
-			try:
-				f = atomic_ofstream(self._aux_cache_filename, 'wb')
-				pickle.dump(self._aux_cache, f, protocol=2)
-				f.close()
-				apply_secpass_permissions(
-					self._aux_cache_filename, gid=portage_gid, mode=0o644)
-			except (IOError, OSError) as e:
-				pass
+			timestamp = time.time()
+			self._aux_cache["timestamp"] = timestamp
+
+			f = atomic_ofstream(self._aux_cache_filename, 'wb')
+			pickle.dump(self._aux_cache, f, protocol=2)
+			f.close()
+			apply_secpass_permissions(
+				self._aux_cache_filename, mode=0o644)
+
+			self._cache_delta.initialize(timestamp)
+			apply_secpass_permissions(
+				self._cache_delta_filename, mode=0o644)
+
 			self._aux_cache["modified"] = set()
 
 	@property
@@ -1590,6 +1603,13 @@ class dblink(object):
 				self.dbdir, noiselevel=-1)
 			return
 
+		if self.dbdir is self.dbpkgdir:
+			counter, = self.vartree.dbapi.aux_get(
+				self.mycpv, ["COUNTER"])
+			self.vartree.dbapi._cache_delta.recordEvent(
+				"remove", self.mycpv,
+				self.settings["SLOT"].split("/")[0], counter)
+
 		shutil.rmtree(self.dbdir)
 		# If empty, remove parent category directory.
 		try:
@@ -4196,6 +4216,8 @@ class dblink(object):
 			self.delete()
 			_movefile(self.dbtmpdir, self.dbpkgdir, mysettings=self.settings)
 			self._merged_path(self.dbpkgdir, os.lstat(self.dbpkgdir))
+			self.vartree.dbapi._cache_delta.recordEvent(
+				"add", self.mycpv, slot, counter)
 		finally:
 			self.unlockdb()
 
-- 
2.0.4