1 files changed, 184 insertions, 68 deletions

diff --git a/bitbake/lib/prserv/serv.py b/bitbake/lib/prserv/serv.py
index 5fc8863f70..e175886308 100644
--- a/bitbake/lib/prserv/serv.py
+++ b/bitbake/lib/prserv/serv.py
@@ -12,6 +12,7 @@ import sqlite3
 import prserv
 import prserv.db
 import errno
+from . import create_async_client, revision_smaller, increase_revision
 import bb.asyncrpc
 
 logger = logging.getLogger("BitBake.PRserv")
@@ -20,16 +21,19 @@ PIDPREFIX = "/tmp/PRServer_%s_%s.pid"
 singleton = None
 
 class PRServerClient(bb.asyncrpc.AsyncServerConnection):
-    def __init__(self, socket, table, read_only):
-        super().__init__(socket, 'PRSERVICE', logger)
+    def __init__(self, socket, server):
+        super().__init__(socket, "PRSERVICE", server.logger)
+        self.server = server
+
         self.handlers.update({
-            'get-pr': self.handle_get_pr,
-            'import-one': self.handle_import_one,
-            'export': self.handle_export,
-            'is-readonly': self.handle_is_readonly,
+            "get-pr": self.handle_get_pr,
+            "test-pr": self.handle_test_pr,
+            "test-package": self.handle_test_package,
+            "max-package-pr": self.handle_max_package_pr,
+            "import-one": self.handle_import_one,
+            "export": self.handle_export,
+            "is-readonly": self.handle_is_readonly,
         })
-        self.table = table
-        self.read_only = read_only
 
     def validate_proto_version(self):
         return (self.proto_version == (1, 0))
@@ -38,104 +42,213 @@ class PRServerClient(bb.asyncrpc.AsyncServerConnection):
         try:
             return await super().dispatch_message(msg)
         except:
-            self.table.sync()
             raise
-        else:
-            self.table.sync_if_dirty()
+
+    async def handle_test_pr(self, request):
+        '''Finds the PR value corresponding to the request. If not found, returns None and doesn't insert a new value'''
+        version = request["version"]
+        pkgarch = request["pkgarch"]
+        checksum = request["checksum"]
+        history = request["history"]
+
+        value = self.server.table.find_value(version, pkgarch, checksum, history)
+        return {"value": value}
+
+    async def handle_test_package(self, request):
+        '''Tells whether there are entries for (version, pkgarch) in the db. Returns True or False'''
+        version = request["version"]
+        pkgarch = request["pkgarch"]
+
+        value = self.server.table.test_package(version, pkgarch)
+        return {"value": value}
+
+    async def handle_max_package_pr(self, request):
+        '''Finds the greatest PR value for (version, pkgarch) in the db. Returns None if no entry was found'''
+        version = request["version"]
+        pkgarch = request["pkgarch"]
+
+        value = self.server.table.find_package_max_value(version, pkgarch)
+        return {"value": value}
 
     async def handle_get_pr(self, request):
-        version = request['version']
-        pkgarch = request['pkgarch']
-        checksum = request['checksum']
+        version = request["version"]
+        pkgarch = request["pkgarch"]
+        checksum = request["checksum"]
+        history = request["history"]
 
-        response = None
-        try:
-            value = self.table.getValue(version, pkgarch, checksum)
-            response = {'value': value}
-        except prserv.NotFoundError:
-            logger.error("can not find value for (%s, %s)",version, checksum)
-        except sqlite3.Error as exc:
-            logger.error(str(exc))
+        if self.upstream_client is None:
+            value = self.server.table.get_value(version, pkgarch, checksum, history)
+            return {"value": value}
 
-        return response
+        # We have an upstream server.
+        # Check whether the local server already knows the requested configuration.
+        # If the configuration is a new one, the generated value we will add will
+        # depend on what's on the upstream server. That's why we're calling find_value()
+        # instead of get_value() directly.
+
+        value = self.server.table.find_value(version, pkgarch, checksum, history)
+        upstream_max = await self.upstream_client.max_package_pr(version, pkgarch)
+
+        if value is not None:
+
+            # The configuration is already known locally.
+
+            if history:
+                value = self.server.table.get_value(version, pkgarch, checksum, history)
+            else:
+                existing_value = value
+                # In "no history", we need to make sure the value doesn't decrease
+                # and is at least greater than the maximum upstream value
+                # and the maximum local value
+
+                local_max = self.server.table.find_package_max_value(version, pkgarch)
+                if revision_smaller(value, local_max):
+                    value = increase_revision(local_max)
+
+                if revision_smaller(value, upstream_max):
+                    # Ask upstream whether it knows the checksum
+                    upstream_value = await self.upstream_client.test_pr(version, pkgarch, checksum)
+                    if upstream_value is None:
+                        # Upstream doesn't have our checksum, let create a new one
+                        value = upstream_max + ".0"
+                    else:
+                        # Fine to take the same value as upstream
+                        value = upstream_max
+
+                if not value == existing_value and not self.server.read_only:
+                    self.server.table.store_value(version, pkgarch, checksum, value)
+
+            return {"value": value}
+
+        # The configuration is a new one for the local server
+        # Let's ask the upstream server whether it knows it
+
+        known_upstream = await self.upstream_client.test_package(version, pkgarch)
+
+        if not known_upstream:
+
+            # The package is not known upstream, must be a local-only package
+            # Let's compute the PR number using the local-only method
+
+            value = self.server.table.get_value(version, pkgarch, checksum, history)
+            return {"value": value}
+
+        # The package is known upstream, let's ask the upstream server
+        # whether it knows our new output hash
+
+        value = await self.upstream_client.test_pr(version, pkgarch, checksum)
+
+        if value is not None:
+
+            # Upstream knows this output hash, let's store it and use it too.
+
+            if not self.server.read_only:
+                self.server.table.store_value(version, pkgarch, checksum, value)
+            # If the local server is read only, won't be able to store the new
+            # value in the database and will have to keep asking the upstream server
+            return {"value": value}
+
+        # The output hash doesn't exist upstream, get the most recent number from upstream (x)
+        # Then, we want to have a new PR value for the local server: x.y
+
+        upstream_max = await self.upstream_client.max_package_pr(version, pkgarch)
+        # Here we know that the package is known upstream, so upstream_max can't be None
+        subvalue = self.server.table.find_new_subvalue(version, pkgarch, upstream_max)
+
+        if not self.server.read_only:
+            self.server.table.store_value(version, pkgarch, checksum, subvalue)
+
+        return {"value": subvalue}
+
+    async def process_requests(self):
+        if self.server.upstream is not None:
+            self.upstream_client = await create_async_client(self.server.upstream)
+        else:
+            self.upstream_client = None
+
+        try:
+            await super().process_requests()
+        finally:
+            if self.upstream_client is not None:
+                await self.upstream_client.close()
 
     async def handle_import_one(self, request):
         response = None
-        if not self.read_only:
-            version = request['version']
-            pkgarch = request['pkgarch']
-            checksum = request['checksum']
-            value = request['value']
+        if not self.server.read_only:
+            version = request["version"]
+            pkgarch = request["pkgarch"]
+            checksum = request["checksum"]
+            value = request["value"]
 
-            value = self.table.importone(version, pkgarch, checksum, value)
+            value = self.server.table.importone(version, pkgarch, checksum, value)
             if value is not None:
-                response = {'value': value}
+                response = {"value": value}
 
         return response
 
     async def handle_export(self, request):
-        version = request['version']
-        pkgarch = request['pkgarch']
-        checksum = request['checksum']
-        colinfo = request['colinfo']
+        version = request["version"]
+        pkgarch = request["pkgarch"]
+        checksum = request["checksum"]
+        colinfo = request["colinfo"]
+        history = request["history"]
 
         try:
-            (metainfo, datainfo) = self.table.export(version, pkgarch, checksum, colinfo)
+            (metainfo, datainfo) = self.server.table.export(version, pkgarch, checksum, colinfo, history)
         except sqlite3.Error as exc:
-            logger.error(str(exc))
+            self.logger.error(str(exc))
             metainfo = datainfo = None
 
-        return {'metainfo': metainfo, 'datainfo': datainfo}
+        return {"metainfo": metainfo, "datainfo": datainfo}
 
     async def handle_is_readonly(self, request):
-        return {'readonly': self.read_only}
+        return {"readonly": self.server.read_only}
 
 class PRServer(bb.asyncrpc.AsyncServer):
-    def __init__(self, dbfile, read_only=False):
+    def __init__(self, dbfile, read_only=False, upstream=None):
         super().__init__(logger)
         self.dbfile = dbfile
         self.table = None
         self.read_only = read_only
+        self.upstream = upstream
 
     def accept_client(self, socket):
-        return PRServerClient(socket, self.table, self.read_only)
+        return PRServerClient(socket, self)
 
     def start(self):
         tasks = super().start()
         self.db = prserv.db.PRData(self.dbfile, read_only=self.read_only)
         self.table = self.db["PRMAIN"]
 
-        logger.info("Started PRServer with DBfile: %s, Address: %s, PID: %s" %
+        self.logger.info("Started PRServer with DBfile: %s, Address: %s, PID: %s" %
                      (self.dbfile, self.address, str(os.getpid())))
 
+        if self.upstream is not None:
+            self.logger.info("And upstream PRServer: %s " % (self.upstream))
+
         return tasks
 
     async def stop(self):
-        self.table.sync_if_dirty()
         self.db.disconnect()
         await super().stop()
 
-    def signal_handler(self):
-        super().signal_handler()
-        if self.table:
-            self.table.sync()
-
 class PRServSingleton(object):
-    def __init__(self, dbfile, logfile, host, port):
+    def __init__(self, dbfile, logfile, host, port, upstream):
         self.dbfile = dbfile
         self.logfile = logfile
         self.host = host
         self.port = port
+        self.upstream = upstream
 
     def start(self):
-        self.prserv = PRServer(self.dbfile)
+        self.prserv = PRServer(self.dbfile, upstream=self.upstream)
         self.prserv.start_tcp_server(socket.gethostbyname(self.host), self.port)
         self.process = self.prserv.serve_as_process(log_level=logging.WARNING)
 
         if not self.prserv.address:
             raise PRServiceConfigError
         if not self.port:
-            self.port = int(self.prserv.address.rsplit(':', 1)[1])
+            self.port = int(self.prserv.address.rsplit(":", 1)[1])
 
 def run_as_daemon(func, pidfile, logfile):
     """
@@ -171,18 +284,18 @@ def run_as_daemon(func, pidfile, logfile):
     # stdout/stderr or it could be 'real' unix fd forking where we need
     # to physically close the fds to prevent the program launching us from
     # potentially hanging on a pipe. Handle both cases.
-    si = open('/dev/null', 'r')
+    si = open("/dev/null", "r")
     try:
-        os.dup2(si.fileno(),sys.stdin.fileno())
+        os.dup2(si.fileno(), sys.stdin.fileno())
     except (AttributeError, io.UnsupportedOperation):
         sys.stdin = si
-    so = open(logfile, 'a+')
+    so = open(logfile, "a+")
     try:
-        os.dup2(so.fileno(),sys.stdout.fileno())
+        os.dup2(so.fileno(), sys.stdout.fileno())
     except (AttributeError, io.UnsupportedOperation):
         sys.stdout = so
     try:
-        os.dup2(so.fileno(),sys.stderr.fileno())
+        os.dup2(so.fileno(), sys.stderr.fileno())
     except (AttributeError, io.UnsupportedOperation):
         sys.stderr = so
 
@@ -200,14 +313,14 @@ def run_as_daemon(func, pidfile, logfile):
 
     # write pidfile
     pid = str(os.getpid())
-    with open(pidfile, 'w') as pf:
+    with open(pidfile, "w") as pf:
         pf.write("%s\n" % pid)
 
     func()
     os.remove(pidfile)
     os._exit(0)
 
-def start_daemon(dbfile, host, port, logfile, read_only=False):
+def start_daemon(dbfile, host, port, logfile, read_only=False, upstream=None):
     ip = socket.gethostbyname(host)
     pidfile = PIDPREFIX % (ip, port)
     try:
@@ -223,7 +336,7 @@ def start_daemon(dbfile, host, port, logfile, read_only=False):
 
     dbfile = os.path.abspath(dbfile)
     def daemon_main():
-        server = PRServer(dbfile, read_only=read_only)
+        server = PRServer(dbfile, read_only=read_only, upstream=upstream)
         server.start_tcp_server(ip, port)
         server.serve_forever()
 
@@ -245,15 +358,15 @@ def stop_daemon(host, port):
         # so at least advise the user which ports the corresponding server is listening
         ports = []
         portstr = ""
-        for pf in glob.glob(PIDPREFIX % (ip,'*')):
+        for pf in glob.glob(PIDPREFIX % (ip, "*")):
             bn = os.path.basename(pf)
             root, _ = os.path.splitext(bn)
-            ports.append(root.split('_')[-1])
+            ports.append(root.split("_")[-1])
         if len(ports):
-            portstr = "Wrong port? Other ports listening at %s: %s" % (host, ' '.join(ports))
+            portstr = "Wrong port? Other ports listening at %s: %s" % (host, " ".join(ports))
 
         sys.stderr.write("pidfile %s does not exist. Daemon not running? %s\n"
-                         % (pidfile,portstr))
+                         % (pidfile, portstr))
         return 1
 
     try:
@@ -284,7 +397,7 @@ def is_running(pid):
     return True
 
 def is_local_special(host, port):
-    if (host == 'localhost' or host == '127.0.0.1') and not port:
+    if (host == "localhost" or host == "127.0.0.1") and not port:
         return True
     else:
         return False
@@ -295,7 +408,7 @@ class PRServiceConfigError(Exception):
 def auto_start(d):
     global singleton
 
-    host_params = list(filter(None, (d.getVar('PRSERV_HOST') or '').split(':')))
+    host_params = list(filter(None, (d.getVar("PRSERV_HOST") or "").split(":")))
     if not host_params:
         # Shutdown any existing PR Server
         auto_shutdown()
@@ -304,12 +417,15 @@ def auto_start(d):
     if len(host_params) != 2:
         # Shutdown any existing PR Server
         auto_shutdown()
-        logger.critical('\n'.join(['PRSERV_HOST: incorrect format',
+        logger.critical("\n".join(["PRSERV_HOST: incorrect format",
                 'Usage: PRSERV_HOST = "<hostname>:<port>"']))
         raise PRServiceConfigError
 
     host = host_params[0].strip().lower()
     port = int(host_params[1])
+
+    upstream = d.getVar("PRSERV_UPSTREAM") or None
+
     if is_local_special(host, port):
         import bb.utils
         cachedir = (d.getVar("PERSISTENT_DIR") or d.getVar("CACHE"))
@@ -324,7 +440,7 @@ def auto_start(d):
                auto_shutdown()
         if not singleton:
             bb.utils.mkdirhier(cachedir)
-            singleton = PRServSingleton(os.path.abspath(dbfile), os.path.abspath(logfile), host, port)
+            singleton = PRServSingleton(os.path.abspath(dbfile), os.path.abspath(logfile), host, port, upstream)
             singleton.start()
     if singleton:
         host = singleton.host
@@ -357,8 +473,8 @@ def connect(host, port):
 
     global singleton
 
-    if host.strip().lower() == 'localhost' and not port:
-        host = 'localhost'
+    if host.strip().lower() == "localhost" and not port:
+        host = "localhost"
         port = singleton.port
 
     conn = client.PRClient()