[PATCH v5 0/5] Re-implement prserv on top of asyncrpc

[PATCH v5 0/5] Re-implement prserv on top of asyncrpc

[Scott Murray]

These changes replace the old XML-based RPC system in prserv with the
new asyncrpc implementation originally used by hashserv, and add a
read-only mode to match the hash equivalency server's support.

Changes from v4:
  * Patch 2 updated to conditionalize all_tasks/current_task usage
    from asyncio vs asyncio.Task based on Python version.  This fixes
    running against Python 3.9 where they were finally removed from
    asyncio.Task.

Changes from v3:
  * Scott Murray taking over upstreaming effort from Paul Barker.

  * Dropped patches which are currently applied to master-next, this
    series should be applied on top of the current master-next branch.

  * Patches 2-4 updated by Scott Murray to rebase on top of 3983643
    ("bitbake: asyncrpc: Catch early SIGTERM").

  * Read-only PR server support patch added to stack to get it into
    the review process.

Paul Barker (5):
  asyncrpc: Wait on writers to close with Python 3.7+
  asyncrpc: Ensure that asyncio shutdown is clean
  asyncrpc: Handle exceptions
  prserv: Replace XML RPC with modern asyncrpc implementation
  prserv: Add read-only mode

 bin/bitbake-prserv        |   4 +-
 lib/bb/asyncrpc/client.py |   3 +
 lib/bb/asyncrpc/serv.py   |  38 ++++-
 lib/prserv/db.py          |  65 +++++++--
 lib/prserv/serv.py        | 286 ++++++++++++++++++++------------------
 5 files changed, 243 insertions(+), 153 deletions(-)

-- 
2.20.1

[PATCH v5 1/5] asyncrpc: Wait on writers to close with Python 3.7+

[Scott Murray]

From: Paul Barker <pbarker@konsulko.com>

The close() method of an asyncio stream writer is not synchronous and so
may return before the writer has finished closing. In Python 3.7 the
wait_closed() method was added which can be used to ensure that a writer
has finished closing before continuing. Sadly in Python 3.6 and earlier
there is no way to guarantee that a writer has finished closing before
moving on.

Signed-off-by: Paul Barker <pbarker@konsulko.com>
---
 lib/bb/asyncrpc/client.py | 3 +++
 lib/bb/asyncrpc/serv.py   | 4 ++++
 2 files changed, 7 insertions(+)

diff --git a/lib/bb/asyncrpc/client.py b/lib/bb/asyncrpc/client.py
index 3eb4fdde..82b6068a 100644
--- a/lib/bb/asyncrpc/client.py
+++ b/lib/bb/asyncrpc/client.py
@@ -7,6 +7,7 @@ import asyncio
 import json
 import os
 import socket
+import sys
 from . import chunkify, DEFAULT_MAX_CHUNK
 
 
@@ -47,6 +48,8 @@ class AsyncClient(object):
 
         if self.writer is not None:
             self.writer.close()
+            if sys.version_info >= (3, 7):
+                await self.writer.wait_closed()
             self.writer = None
 
     async def _send_wrapper(self, proc):
diff --git a/lib/bb/asyncrpc/serv.py b/lib/bb/asyncrpc/serv.py
index 4084f300..2c219c1e 100644
--- a/lib/bb/asyncrpc/serv.py
+++ b/lib/bb/asyncrpc/serv.py
@@ -75,6 +75,8 @@ class AsyncServerConnection(object):
             self.logger.error(str(e))
         finally:
             self.writer.close()
+            if sys.version_info >= (3, 7):
+                await self.writer.wait_closed()
 
     async def dispatch_message(self, msg):
         for k in self.handlers.keys():
@@ -193,6 +195,8 @@ class AsyncServer(object):
             self.logger.error('Error from client: %s' % str(e), exc_info=True)
             traceback.print_exc()
             writer.close()
+            if sys.version_info >= (3, 7):
+                await writer.wait_closed()
         self.logger.debug('Client disconnected')
 
     def run_loop_forever(self):
-- 
2.20.1

[PATCH v5 2/5] asyncrpc: Ensure that asyncio shutdown is clean

[Scott Murray]

From: Paul Barker <pbarker@konsulko.com>

We should ensure that all async tasks are cancelled and then allowed to
finish gracefully before closing the asyncio loop.

Signed-off-by: Paul Barker <pbarker@konsulko.com>
[updated for asyncrpc SIGTERM handling changes]
Signed-off-by: Scott Murray <scott.murray@konsulko.com>
---
 lib/bb/asyncrpc/serv.py | 28 ++++++++++++++++++++++------
 1 file changed, 22 insertions(+), 6 deletions(-)

diff --git a/lib/bb/asyncrpc/serv.py b/lib/bb/asyncrpc/serv.py
index 2c219c1e..87d142c1 100644
--- a/lib/bb/asyncrpc/serv.py
+++ b/lib/bb/asyncrpc/serv.py
@@ -205,21 +205,37 @@ class AsyncServer(object):
         except KeyboardInterrupt:
             pass
 
-    def signal_handler(self):
+    def handle_signal(self):
         self.logger.debug("Got exit signal")
+        self.loop.create_task(self.shutdown())
+
+    async def shutdown(self):
+        self.logger.debug('Server shutting down')
+
+        # Stop accepting connections
+        self.server.close()
+        await self.server.wait_closed()
+
+        # Cancel all active tasks
+        if sys.version_info >= (3, 7):
+            tasks = [t for t in asyncio.all_tasks(self.loop)
+                     if t is not asyncio.current_task(self.loop)]
+        else:
+            tasks = [t for t in asyncio.Task.all_tasks(self.loop)
+                     if t is not asyncio.Task.current_task(self.loop)]
+        for task in tasks:
+            task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
         self.loop.stop()
 
     def serve_forever(self):
         asyncio.set_event_loop(self.loop)
         try:
-            self.loop.add_signal_handler(signal.SIGTERM, self.signal_handler)
+            self.loop.add_signal_handler(signal.SIGTERM, self.handle_signal)
             signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGTERM])
+            self.loop.add_signal_handler(signal.SIGINT, self.handle_signal)
 
             self.run_loop_forever()
-            self.server.close()
-
-            self.loop.run_until_complete(self.server.wait_closed())
-            self.logger.debug('Server shutting down')
         finally:
             if self.close_loop:
                 if sys.version_info >= (3, 6):
-- 
2.20.1

[PATCH v5 3/5] asyncrpc: Handle exceptions

[Scott Murray]

From: Paul Barker <pbarker@konsulko.com>

If an async task raises an exception it may cause the asyncio loop to
hang or may otherwise leave our server in a bad state. To avoid these
issues we can add an exception handler to the asyncio loop which
attempts to gracefully shut down the server if an unhandled exception
occurs.

Signed-off-by: Paul Barker <pbarker@konsulko.com>
[updated for asyncrpc SIGTERM handling changes]
Signed-off-by: Scott Murray <scott.murray@konsulko.com>
---
 lib/bb/asyncrpc/serv.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/lib/bb/asyncrpc/serv.py b/lib/bb/asyncrpc/serv.py
index 87d142c1..0e60997d 100644
--- a/lib/bb/asyncrpc/serv.py
+++ b/lib/bb/asyncrpc/serv.py
@@ -209,6 +209,11 @@ class AsyncServer(object):
         self.logger.debug("Got exit signal")
         self.loop.create_task(self.shutdown())
 
+    def handle_exception(self, loop, context):
+        msg = context.get("exception", context["message"])
+        self.logger.error("Caught exception: %s" % (msg))
+        loop.create_task(self.shutdown())
+
     async def shutdown(self):
         self.logger.debug('Server shutting down')
 
@@ -231,6 +236,7 @@ class AsyncServer(object):
     def serve_forever(self):
         asyncio.set_event_loop(self.loop)
         try:
+            self.loop.set_exception_handler(self.handle_exception)
             self.loop.add_signal_handler(signal.SIGTERM, self.handle_signal)
             signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGTERM])
             self.loop.add_signal_handler(signal.SIGINT, self.handle_signal)
-- 
2.20.1

[PATCH v5 4/5] prserv: Replace XML RPC with modern asyncrpc implementation

[Scott Murray]

From: Paul Barker <pbarker@konsulko.com>

Update the prserv client and server classes to use the modern json and
asyncio based RPC system implemented by the asyncrpc module.

Signed-off-by: Paul Barker <pbarker@konsulko.com>
[updated for asyncrpc SIGTERM handling changes]
Signed-off-by: Scott Murray <scott.murray@konsulko.com>
---
 lib/prserv/serv.py | 265 +++++++++++++++++++++++----------------------
 1 file changed, 136 insertions(+), 129 deletions(-)

diff --git a/lib/prserv/serv.py b/lib/prserv/serv.py
index 5e322bf8..2c572e0f 100644
--- a/lib/prserv/serv.py
+++ b/lib/prserv/serv.py
@@ -4,157 +4,158 @@
 
 import os,sys,logging
 import signal, time
-from xmlrpc.server import SimpleXMLRPCServer, SimpleXMLRPCRequestHandler
 import socket
 import io
 import sqlite3
-import bb.server.xmlrpcclient
 import prserv
 import prserv.db
 import errno
-import multiprocessing
+import bb.asyncrpc
 
 logger = logging.getLogger("BitBake.PRserv")
 
-class Handler(SimpleXMLRPCRequestHandler):
-    def _dispatch(self,method,params):
-        try:
-            value=self.server.funcs[method](*params)
-        except:
-            import traceback
-            traceback.print_exc()
-            raise
-        return value
-
 PIDPREFIX = "/tmp/PRServer_%s_%s.pid"
 singleton = None
 
+class PRServerClient(bb.asyncrpc.AsyncServerConnection):
+    def __init__(self, reader, writer, table):
+        super().__init__(reader, writer, 'PRSERVICE', logger)
+        self.handlers.update({
+            'get-pr': self.handle_get_pr,
+            'import-one': self.handle_import_one,
+            'export': self.handle_export,
+        })
+        self.table = table
 
-class PRServer(SimpleXMLRPCServer):
-    def __init__(self, dbfile, logfile, interface):
-        ''' constructor '''
-        try:
-            SimpleXMLRPCServer.__init__(self, interface,
-                                        logRequests=False, allow_none=True)
-        except socket.error:
-            ip=socket.gethostbyname(interface[0])
-            port=interface[1]
-            msg="PR Server unable to bind to %s:%s\n" % (ip, port)
-            sys.stderr.write(msg)
-            raise PRServiceConfigError
-
-        self.dbfile=dbfile
-        self.logfile=logfile
-        self.host, self.port = self.socket.getsockname()
-
-        self.register_function(self.getPR, "getPR")
-        self.register_function(self.ping, "ping")
-        self.register_function(self.export, "export")
-        self.register_function(self.importone, "importone")
-        self.register_introspection_functions()
-
-        self.iter_count = 0
-        # 60 iterations between syncs or sync if dirty every ~30 seconds
-        self.iterations_between_sync = 60
-
-    def sigint_handler(self, signum, stack):
-        if self.table:
-            self.table.sync()
-
-    def sigterm_handler(self, signum, stack):
-        if self.table:
-            self.table.sync()
-        raise(SystemExit)
+    def validate_proto_version(self):
+        return (self.proto_version == (1, 0))
 
-    def process_request(self, request, client_address):
-        if request is None:
-            return
+    async def dispatch_message(self, msg):
         try:
-            self.finish_request(request, client_address)
-            self.shutdown_request(request)
-            self.iter_count = (self.iter_count + 1) % self.iterations_between_sync
-            if self.iter_count == 0:
-                self.table.sync_if_dirty()
+            await super().dispatch_message(msg)
         except:
-            self.handle_error(request, client_address)
-            self.shutdown_request(request)
             self.table.sync()
-        self.table.sync_if_dirty()
+            raise
 
-    def serve_forever(self, poll_interval=0.5):
-        signal.signal(signal.SIGINT, self.sigint_handler)
-        signal.signal(signal.SIGTERM, self.sigterm_handler)
+        self.table.sync_if_dirty()
 
-        self.db = prserv.db.PRData(self.dbfile)
-        self.table = self.db["PRMAIN"]
-        return super().serve_forever(poll_interval)
+    async def handle_get_pr(self, request):
+        version = request['version']
+        pkgarch = request['pkgarch']
+        checksum = request['checksum']
 
-    def export(self, version=None, pkgarch=None, checksum=None, colinfo=True):
+        response = None
         try:
-            return self.table.export(version, pkgarch, checksum, colinfo)
+            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))
-            return None
 
-    def importone(self, version, pkgarch, checksum, value):
-        return self.table.importone(version, pkgarch, checksum, value)
+        self.write_message(response)
 
-    def ping(self):
-        return True
+    async def handle_import_one(self, request):
+        version = request['version']
+        pkgarch = request['pkgarch']
+        checksum = request['checksum']
+        value = request['value']
+
+        value = self.table.importone(version, pkgarch, checksum, value)
+        if value is not None:
+            response = {'value': value}
+        else:
+            response = None
+        self.write_message(response)
 
-    def getinfo(self):
-        return (self.host, self.port)
+    async def handle_export(self, request):
+        version = request['version']
+        pkgarch = request['pkgarch']
+        checksum = request['checksum']
+        colinfo = request['colinfo']
 
-    def getPR(self, version, pkgarch, checksum):
         try:
-            return self.table.getValue(version, pkgarch, checksum)
-        except prserv.NotFoundError:
-            logger.error("can not find value for (%s, %s)",version, checksum)
-            return None
+            (metainfo, datainfo) = self.table.export(version, pkgarch, checksum, colinfo)
         except sqlite3.Error as exc:
             logger.error(str(exc))
-            return None
+            metainfo = datainfo = None
 
-class PRServSingleton(object):
-    def __init__(self, dbfile, logfile, interface):
-        self.dbfile = dbfile
-        self.logfile = logfile
-        self.interface = interface
-        self.host = None
-        self.port = None
+        response = {'metainfo': metainfo, 'datainfo': datainfo}
+        self.write_message(response)
 
-    def start(self):
-        self.prserv = PRServer(self.dbfile, self.logfile, self.interface)
-        self.process = multiprocessing.Process(target=self.prserv.serve_forever)
-        self.process.start()
+class PRServer(bb.asyncrpc.AsyncServer):
+    def __init__(self, dbfile, loop=None):
+        super().__init__(logger, loop)
+        self.dbfile = dbfile
+        self.table = None
 
-        self.host, self.port = self.prserv.getinfo()
+    def accept_client(self, reader, writer):
+        return PRServerClient(reader, writer, self.table)
 
-    def getinfo(self):
-        return (self.host, self.port)
+    def serve_forever(self):
+        self.db = prserv.db.PRData(self.dbfile)
+        self.table = self.db["PRMAIN"]
 
-class PRServerConnection(object):
-    def __init__(self, host, port):
-        if is_local_special(host, port):
-            host, port = singleton.getinfo()
-        self.host = host
-        self.port = port
-        self.connection, self.transport = bb.server.xmlrpcclient._create_server(self.host, self.port)
+        logger.debug("Started PRServer with DBfile: %s, Address: %s, PID: %s" %
+                     (self.dbfile, self.address, str(os.getpid())))
 
-    def getPR(self, version, pkgarch, checksum):
-        return self.connection.getPR(version, pkgarch, checksum)
+        super().serve_forever()
 
-    def ping(self):
-        return self.connection.ping()
+        self.table.sync_if_dirty()
+        self.db.disconnect()
 
-    def export(self,version=None, pkgarch=None, checksum=None, colinfo=True):
-        return self.connection.export(version, pkgarch, checksum, colinfo)
+    def signal_handler(self):
+        super().signal_handler()
+        if self.table:
+            self.table.sync()
 
-    def importone(self, version, pkgarch, checksum, value):
-        return self.connection.importone(version, pkgarch, checksum, value)
+class PRServSingleton(object):
+    def __init__(self, dbfile, logfile, host, port):
+        self.dbfile = dbfile
+        self.logfile = logfile
+        self.host = host
+        self.port = port
 
-    def getinfo(self):
-        return self.host, self.port
+    def start(self):
+        self.prserv = PRServer(self.dbfile)
+        self.prserv.start_tcp_server(self.host, self.port)
+        self.process = self.prserv.serve_as_process()
+
+        if not self.port:
+            self.port = int(self.prserv.address.rsplit(':', 1)[1])
+
+class PRAsyncClient(bb.asyncrpc.AsyncClient):
+    def __init__(self):
+        super().__init__('PRSERVICE', '1.0', logger)
+
+    async def getPR(self, version, pkgarch, checksum):
+        response = await self.send_message(
+            {'get-pr': {'version': version, 'pkgarch': pkgarch, 'checksum': checksum}}
+        )
+        if response:
+            return response['value']
+
+    async def importone(self, version, pkgarch, checksum, value):
+        response = await self.send_message(
+            {'import-one': {'version': version, 'pkgarch': pkgarch, 'checksum': checksum, 'value': value}}
+        )
+        if response:
+            return response['value']
+
+    async def export(self, version, pkgarch, checksum, colinfo):
+        response = await self.send_message(
+            {'export': {'version': version, 'pkgarch': pkgarch, 'checksum': checksum, 'colinfo': colinfo}}
+        )
+        if response:
+            return (response['metainfo'], response['datainfo'])
+
+class PRClient(bb.asyncrpc.Client):
+    def __init__(self):
+        super().__init__()
+        self._add_methods('getPR', 'importone', 'export')
+
+    def _get_async_client(self):
+        return PRAsyncClient()
 
 def run_as_daemon(func, pidfile, logfile):
     """
@@ -240,15 +241,13 @@ def start_daemon(dbfile, host, port, logfile):
                             % pidfile)
         return 1
 
-    server = PRServer(os.path.abspath(dbfile), os.path.abspath(logfile), (ip,port))
-    run_as_daemon(server.serve_forever, pidfile, os.path.abspath(logfile))
+    dbfile = os.path.abspath(dbfile)
+    def daemon_main():
+        server = PRServer(dbfile)
+        server.start_tcp_server(host, port)
+        server.serve_forever()
 
-    # Sometimes, the port (i.e. localhost:0) indicated by the user does not match with
-    # the one the server actually is listening, so at least warn the user about it
-    _,rport = server.getinfo()
-    if port != rport:
-        sys.stdout.write("Server is listening at port %s instead of %s\n"
-                         % (rport,port))
+    run_as_daemon(daemon_main, pidfile, os.path.abspath(logfile))
     return 0
 
 def stop_daemon(host, port):
@@ -302,7 +301,7 @@ def is_running(pid):
     return True
 
 def is_local_special(host, port):
-    if host.strip().upper() == 'localhost'.upper() and (not port):
+    if host.strip().lower() == 'localhost' and not port:
         return True
     else:
         return False
@@ -340,20 +339,19 @@ def auto_start(d):
                auto_shutdown()
         if not singleton:
             bb.utils.mkdirhier(cachedir)
-            singleton = PRServSingleton(os.path.abspath(dbfile), os.path.abspath(logfile), ("localhost",0))
+            singleton = PRServSingleton(os.path.abspath(dbfile), os.path.abspath(logfile), "localhost", 0)
             singleton.start()
     if singleton:
-        host, port = singleton.getinfo()
+        host = singleton.host
+        port = singleton.port
     else:
         host = host_params[0]
         port = int(host_params[1])
 
     try:
-        connection = PRServerConnection(host,port)
-        connection.ping()
-        realhost, realport = connection.getinfo()
-        return str(realhost) + ":" + str(realport)
-        
+        ping(host, port)
+        return str(host) + ":" + str(port)
+
     except Exception:
         logger.critical("PRservice %s:%d not available" % (host, port))
         raise PRServiceConfigError
@@ -366,8 +364,17 @@ def auto_shutdown():
         singleton = None
 
 def ping(host, port):
-    conn=PRServerConnection(host, port)
+    conn=PRClient()
+    conn.connect_tcp(host, port)
     return conn.ping()
 
 def connect(host, port):
-    return PRServerConnection(host, port)
+    global singleton
+
+    if host.strip().lower() == 'localhost' and not port:
+        host = 'localhost'
+        port = singleton.port
+
+    conn = PRClient()
+    conn.connect_tcp(host, port)
+    return conn
-- 
2.20.1

[PATCH v5 5/5] prserv: Add read-only mode

[Scott Murray]

From: Paul Barker <pbarker@konsulko.com>

Signed-off-by: Paul Barker <pbarker@konsulko.com>
---
 bin/bitbake-prserv |  4 ++-
 lib/prserv/db.py   | 65 ++++++++++++++++++++++++++++++++++++----------
 lib/prserv/serv.py | 49 ++++++++++++++++++++++------------
 3 files changed, 86 insertions(+), 32 deletions(-)

diff --git a/bin/bitbake-prserv b/bin/bitbake-prserv
index 1e9b6cbc..bef5ef68 100755
--- a/bin/bitbake-prserv
+++ b/bin/bitbake-prserv
@@ -36,12 +36,14 @@ def main():
                       dest="host", type="string", default=PRHOST_DEFAULT)
     parser.add_option("--port", help="port number(default: 8585)", action="store",
                       dest="port", type="int", default=PRPORT_DEFAULT)
+    parser.add_option("-r", "--read-only", help="open database in read-only mode",
+                      action="store_true")
 
     options, args = parser.parse_args(sys.argv)
     prserv.init_logger(os.path.abspath(options.logfile),options.loglevel)
 
     if options.start:
-        ret=prserv.serv.start_daemon(options.dbfile, options.host, options.port,os.path.abspath(options.logfile))
+        ret=prserv.serv.start_daemon(options.dbfile, options.host, options.port,os.path.abspath(options.logfile), options.read_only)
     elif options.stop:
         ret=prserv.serv.stop_daemon(options.host, options.port)
     else:
diff --git a/lib/prserv/db.py b/lib/prserv/db.py
index cb2a2461..2710d4a2 100644
--- a/lib/prserv/db.py
+++ b/lib/prserv/db.py
@@ -30,21 +30,29 @@ if sqlversion[0] < 3 or (sqlversion[0] == 3 and sqlversion[1] < 3):
 #
 
 class PRTable(object):
-    def __init__(self, conn, table, nohist):
+    def __init__(self, conn, table, nohist, read_only):
         self.conn = conn
         self.nohist = nohist
+        self.read_only = read_only
         self.dirty = False
         if nohist:
             self.table = "%s_nohist" % table 
         else:
             self.table = "%s_hist" % table 
 
-        self._execute("CREATE TABLE IF NOT EXISTS %s \
-                    (version TEXT NOT NULL, \
-                    pkgarch TEXT NOT NULL,  \
-                    checksum TEXT NOT NULL, \
-                    value INTEGER, \
-                    PRIMARY KEY (version, pkgarch, checksum));" % self.table)
+        if self.read_only:
+            table_exists = self._execute(
+                        "SELECT count(*) FROM sqlite_master \
+                        WHERE type='table' AND name='%s'" % (self.table))
+            if not table_exists:
+                raise prserv.NotFoundError
+        else:
+            self._execute("CREATE TABLE IF NOT EXISTS %s \
+                        (version TEXT NOT NULL, \
+                        pkgarch TEXT NOT NULL,  \
+                        checksum TEXT NOT NULL, \
+                        value INTEGER, \
+                        PRIMARY KEY (version, pkgarch, checksum));" % self.table)
 
     def _execute(self, *query):
         """Execute a query, waiting to acquire a lock if necessary"""
@@ -59,8 +67,9 @@ class PRTable(object):
                 raise exc
 
     def sync(self):
-        self.conn.commit()
-        self._execute("BEGIN EXCLUSIVE TRANSACTION")
+        if not self.read_only:
+            self.conn.commit()
+            self._execute("BEGIN EXCLUSIVE TRANSACTION")
 
     def sync_if_dirty(self):
         if self.dirty:
@@ -75,6 +84,15 @@ class PRTable(object):
             return row[0]
         else:
             #no value found, try to insert
+            if self.read_only:
+                data = self._execute("SELECT ifnull(max(value)+1,0) FROM %s where version=? AND pkgarch=?;" % (self.table),
+                                   (version, pkgarch))
+                row = data.fetchone()
+                if row is not None:
+                    return row[0]
+                else:
+                    return 0
+
             try:
                 self._execute("INSERT INTO %s VALUES (?, ?, ?, (select ifnull(max(value)+1,0) from %s where version=? AND pkgarch=?));"
                            % (self.table,self.table),
@@ -103,6 +121,15 @@ class PRTable(object):
             return row[0]
         else:
             #no value found, try to insert
+            if self.read_only:
+                data = self._execute("SELECT ifnull(max(value)+1,0) FROM %s where version=? AND pkgarch=?;" % (self.table),
+                                   (version, pkgarch))
+                row = data.fetchone()
+                if row is not None:
+                    return row[0]
+                else:
+                    return 0
+
             try:
                 self._execute("INSERT OR REPLACE INTO %s VALUES (?, ?, ?, (select ifnull(max(value)+1,0) from %s where version=? AND pkgarch=?));"
                                % (self.table,self.table),
@@ -128,6 +155,9 @@ class PRTable(object):
             return self._getValueHist(version, pkgarch, checksum)
 
     def _importHist(self, version, pkgarch, checksum, value):
+        if self.read_only:
+            return None
+
         val = None 
         data = self._execute("SELECT value FROM %s WHERE version=? AND pkgarch=? AND checksum=?;" % self.table,
                            (version, pkgarch, checksum))
@@ -152,6 +182,9 @@ class PRTable(object):
         return val
 
     def _importNohist(self, version, pkgarch, checksum, value):
+        if self.read_only:
+            return None
+
         try:
             #try to insert
             self._execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
@@ -245,19 +278,23 @@ class PRTable(object):
 
 class PRData(object):
     """Object representing the PR database"""
-    def __init__(self, filename, nohist=True):
+    def __init__(self, filename, nohist=True, read_only=False):
         self.filename=os.path.abspath(filename)
         self.nohist=nohist
+        self.read_only = read_only
         #build directory hierarchy
         try:
             os.makedirs(os.path.dirname(self.filename))
         except OSError as e:
             if e.errno != errno.EEXIST:
                 raise e
-        self.connection=sqlite3.connect(self.filename, isolation_level="EXCLUSIVE", check_same_thread = False)
+        uri = "file:%s%s" % (self.filename, "?mode=ro" if self.read_only else "")
+        logger.debug("Opening PRServ database '%s'" % (uri))
+        self.connection=sqlite3.connect(uri, uri=True, isolation_level="EXCLUSIVE", check_same_thread = False)
         self.connection.row_factory=sqlite3.Row
-        self.connection.execute("pragma synchronous = off;")
-        self.connection.execute("PRAGMA journal_mode = MEMORY;")
+        if not self.read_only:
+            self.connection.execute("pragma synchronous = off;")
+            self.connection.execute("PRAGMA journal_mode = MEMORY;")
         self._tables={}
 
     def disconnect(self):
@@ -270,7 +307,7 @@ class PRData(object):
         if tblname in self._tables:
             return self._tables[tblname]
         else:
-            tableobj = self._tables[tblname] = PRTable(self.connection, tblname, self.nohist)
+            tableobj = self._tables[tblname] = PRTable(self.connection, tblname, self.nohist, self.read_only)
             return tableobj
 
     def __delitem__(self, tblname):
diff --git a/lib/prserv/serv.py b/lib/prserv/serv.py
index 2c572e0f..618bd452 100644
--- a/lib/prserv/serv.py
+++ b/lib/prserv/serv.py
@@ -18,14 +18,16 @@ PIDPREFIX = "/tmp/PRServer_%s_%s.pid"
 singleton = None
 
 class PRServerClient(bb.asyncrpc.AsyncServerConnection):
-    def __init__(self, reader, writer, table):
+    def __init__(self, reader, writer, table, read_only):
         super().__init__(reader, writer, 'PRSERVICE', logger)
         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,
         })
         self.table = table
+        self.read_only = read_only
 
     def validate_proto_version(self):
         return (self.proto_version == (1, 0))
@@ -56,16 +58,17 @@ class PRServerClient(bb.asyncrpc.AsyncServerConnection):
         self.write_message(response)
 
     async def handle_import_one(self, request):
-        version = request['version']
-        pkgarch = request['pkgarch']
-        checksum = request['checksum']
-        value = request['value']
+        response = None
+        if not self.read_only:
+            version = request['version']
+            pkgarch = request['pkgarch']
+            checksum = request['checksum']
+            value = request['value']
+
+            value = self.table.importone(version, pkgarch, checksum, value)
+            if value is not None:
+                response = {'value': value}
 
-        value = self.table.importone(version, pkgarch, checksum, value)
-        if value is not None:
-            response = {'value': value}
-        else:
-            response = None
         self.write_message(response)
 
     async def handle_export(self, request):
@@ -83,20 +86,25 @@ class PRServerClient(bb.asyncrpc.AsyncServerConnection):
         response = {'metainfo': metainfo, 'datainfo': datainfo}
         self.write_message(response)
 
+    async def handle_is_readonly(self, request):
+        response = {'readonly': self.read_only}
+        self.write_message(response)
+
 class PRServer(bb.asyncrpc.AsyncServer):
-    def __init__(self, dbfile, loop=None):
+    def __init__(self, dbfile, loop=None, read_only=False):
         super().__init__(logger, loop)
         self.dbfile = dbfile
         self.table = None
+        self.read_only = read_only
 
     def accept_client(self, reader, writer):
-        return PRServerClient(reader, writer, self.table)
+        return PRServerClient(reader, writer, self.table, self.read_only)
 
     def serve_forever(self):
-        self.db = prserv.db.PRData(self.dbfile)
+        self.db = prserv.db.PRData(self.dbfile, read_only=self.read_only)
         self.table = self.db["PRMAIN"]
 
-        logger.debug("Started PRServer with DBfile: %s, Address: %s, PID: %s" %
+        logger.info("Started PRServer with DBfile: %s, Address: %s, PID: %s" %
                      (self.dbfile, self.address, str(os.getpid())))
 
         super().serve_forever()
@@ -149,10 +157,17 @@ class PRAsyncClient(bb.asyncrpc.AsyncClient):
         if response:
             return (response['metainfo'], response['datainfo'])
 
+    async def is_readonly(self):
+        response = await self.send_message(
+            {'is-readonly': {}}
+        )
+        if response:
+            return response['readonly']
+
 class PRClient(bb.asyncrpc.Client):
     def __init__(self):
         super().__init__()
-        self._add_methods('getPR', 'importone', 'export')
+        self._add_methods('getPR', 'importone', 'export', 'is_readonly')
 
     def _get_async_client(self):
         return PRAsyncClient()
@@ -227,7 +242,7 @@ def run_as_daemon(func, pidfile, logfile):
     os.remove(pidfile)
     os._exit(0)
 
-def start_daemon(dbfile, host, port, logfile):
+def start_daemon(dbfile, host, port, logfile, read_only=False):
     ip = socket.gethostbyname(host)
     pidfile = PIDPREFIX % (ip, port)
     try:
@@ -243,7 +258,7 @@ def start_daemon(dbfile, host, port, logfile):
 
     dbfile = os.path.abspath(dbfile)
     def daemon_main():
-        server = PRServer(dbfile)
+        server = PRServer(dbfile, read_only=read_only)
         server.start_tcp_server(host, port)
         server.serve_forever()
 
-- 
2.20.1

Re: [bitbake-devel] [PATCH v5 5/5] prserv: Add read-only mode

[Armin Kuster]

Paul or Scott,

On 7/27/21 10:21 AM, Scott Murray wrote:
> From: Paul Barker <pbarker@konsulko.com>
>
> Signed-off-by: Paul Barker <pbarker@konsulko.com>

Does this address  this open defect?
https://bugzilla.yoctoproject.org/show_bug.cgi?id=13659

- Armin
> ---
>  bin/bitbake-prserv |  4 ++-
>  lib/prserv/db.py   | 65 ++++++++++++++++++++++++++++++++++++----------
>  lib/prserv/serv.py | 49 ++++++++++++++++++++++------------
>  3 files changed, 86 insertions(+), 32 deletions(-)
>
> diff --git a/bin/bitbake-prserv b/bin/bitbake-prserv
> index 1e9b6cbc..bef5ef68 100755
> --- a/bin/bitbake-prserv
> +++ b/bin/bitbake-prserv
> @@ -36,12 +36,14 @@ def main():
>                        dest="host", type="string", default=PRHOST_DEFAULT)
>      parser.add_option("--port", help="port number(default: 8585)", action="store",
>                        dest="port", type="int", default=PRPORT_DEFAULT)
> +    parser.add_option("-r", "--read-only", help="open database in read-only mode",
> +                      action="store_true")
>  
>      options, args = parser.parse_args(sys.argv)
>      prserv.init_logger(os.path.abspath(options.logfile),options.loglevel)
>  
>      if options.start:
> -        ret=prserv.serv.start_daemon(options.dbfile, options.host, options.port,os.path.abspath(options.logfile))
> +        ret=prserv.serv.start_daemon(options.dbfile, options.host, options.port,os.path.abspath(options.logfile), options.read_only)
>      elif options.stop:
>          ret=prserv.serv.stop_daemon(options.host, options.port)
>      else:
> diff --git a/lib/prserv/db.py b/lib/prserv/db.py
> index cb2a2461..2710d4a2 100644
> --- a/lib/prserv/db.py
> +++ b/lib/prserv/db.py
> @@ -30,21 +30,29 @@ if sqlversion[0] < 3 or (sqlversion[0] == 3 and sqlversion[1] < 3):
>  #
>  
>  class PRTable(object):
> -    def __init__(self, conn, table, nohist):
> +    def __init__(self, conn, table, nohist, read_only):
>          self.conn = conn
>          self.nohist = nohist
> +        self.read_only = read_only
>          self.dirty = False
>          if nohist:
>              self.table = "%s_nohist" % table 
>          else:
>              self.table = "%s_hist" % table 
>  
> -        self._execute("CREATE TABLE IF NOT EXISTS %s \
> -                    (version TEXT NOT NULL, \
> -                    pkgarch TEXT NOT NULL,  \
> -                    checksum TEXT NOT NULL, \
> -                    value INTEGER, \
> -                    PRIMARY KEY (version, pkgarch, checksum));" % self.table)
> +        if self.read_only:
> +            table_exists = self._execute(
> +                        "SELECT count(*) FROM sqlite_master \
> +                        WHERE type='table' AND name='%s'" % (self.table))
> +            if not table_exists:
> +                raise prserv.NotFoundError
> +        else:
> +            self._execute("CREATE TABLE IF NOT EXISTS %s \
> +                        (version TEXT NOT NULL, \
> +                        pkgarch TEXT NOT NULL,  \
> +                        checksum TEXT NOT NULL, \
> +                        value INTEGER, \
> +                        PRIMARY KEY (version, pkgarch, checksum));" % self.table)
>  
>      def _execute(self, *query):
>          """Execute a query, waiting to acquire a lock if necessary"""
> @@ -59,8 +67,9 @@ class PRTable(object):
>                  raise exc
>  
>      def sync(self):
> -        self.conn.commit()
> -        self._execute("BEGIN EXCLUSIVE TRANSACTION")
> +        if not self.read_only:
> +            self.conn.commit()
> +            self._execute("BEGIN EXCLUSIVE TRANSACTION")
>  
>      def sync_if_dirty(self):
>          if self.dirty:
> @@ -75,6 +84,15 @@ class PRTable(object):
>              return row[0]
>          else:
>              #no value found, try to insert
> +            if self.read_only:
> +                data = self._execute("SELECT ifnull(max(value)+1,0) FROM %s where version=? AND pkgarch=?;" % (self.table),
> +                                   (version, pkgarch))
> +                row = data.fetchone()
> +                if row is not None:
> +                    return row[0]
> +                else:
> +                    return 0
> +
>              try:
>                  self._execute("INSERT INTO %s VALUES (?, ?, ?, (select ifnull(max(value)+1,0) from %s where version=? AND pkgarch=?));"
>                             % (self.table,self.table),
> @@ -103,6 +121,15 @@ class PRTable(object):
>              return row[0]
>          else:
>              #no value found, try to insert
> +            if self.read_only:
> +                data = self._execute("SELECT ifnull(max(value)+1,0) FROM %s where version=? AND pkgarch=?;" % (self.table),
> +                                   (version, pkgarch))
> +                row = data.fetchone()
> +                if row is not None:
> +                    return row[0]
> +                else:
> +                    return 0
> +
>              try:
>                  self._execute("INSERT OR REPLACE INTO %s VALUES (?, ?, ?, (select ifnull(max(value)+1,0) from %s where version=? AND pkgarch=?));"
>                                 % (self.table,self.table),
> @@ -128,6 +155,9 @@ class PRTable(object):
>              return self._getValueHist(version, pkgarch, checksum)
>  
>      def _importHist(self, version, pkgarch, checksum, value):
> +        if self.read_only:
> +            return None
> +
>          val = None 
>          data = self._execute("SELECT value FROM %s WHERE version=? AND pkgarch=? AND checksum=?;" % self.table,
>                             (version, pkgarch, checksum))
> @@ -152,6 +182,9 @@ class PRTable(object):
>          return val
>  
>      def _importNohist(self, version, pkgarch, checksum, value):
> +        if self.read_only:
> +            return None
> +
>          try:
>              #try to insert
>              self._execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
> @@ -245,19 +278,23 @@ class PRTable(object):
>  
>  class PRData(object):
>      """Object representing the PR database"""
> -    def __init__(self, filename, nohist=True):
> +    def __init__(self, filename, nohist=True, read_only=False):
>          self.filename=os.path.abspath(filename)
>          self.nohist=nohist
> +        self.read_only = read_only
>          #build directory hierarchy
>          try:
>              os.makedirs(os.path.dirname(self.filename))
>          except OSError as e:
>              if e.errno != errno.EEXIST:
>                  raise e
> -        self.connection=sqlite3.connect(self.filename, isolation_level="EXCLUSIVE", check_same_thread = False)
> +        uri = "file:%s%s" % (self.filename, "?mode=ro" if self.read_only else "")
> +        logger.debug("Opening PRServ database '%s'" % (uri))
> +        self.connection=sqlite3.connect(uri, uri=True, isolation_level="EXCLUSIVE", check_same_thread = False)
>          self.connection.row_factory=sqlite3.Row
> -        self.connection.execute("pragma synchronous = off;")
> -        self.connection.execute("PRAGMA journal_mode = MEMORY;")
> +        if not self.read_only:
> +            self.connection.execute("pragma synchronous = off;")
> +            self.connection.execute("PRAGMA journal_mode = MEMORY;")
>          self._tables={}
>  
>      def disconnect(self):
> @@ -270,7 +307,7 @@ class PRData(object):
>          if tblname in self._tables:
>              return self._tables[tblname]
>          else:
> -            tableobj = self._tables[tblname] = PRTable(self.connection, tblname, self.nohist)
> +            tableobj = self._tables[tblname] = PRTable(self.connection, tblname, self.nohist, self.read_only)
>              return tableobj
>  
>      def __delitem__(self, tblname):
> diff --git a/lib/prserv/serv.py b/lib/prserv/serv.py
> index 2c572e0f..618bd452 100644
> --- a/lib/prserv/serv.py
> +++ b/lib/prserv/serv.py
> @@ -18,14 +18,16 @@ PIDPREFIX = "/tmp/PRServer_%s_%s.pid"
>  singleton = None
>  
>  class PRServerClient(bb.asyncrpc.AsyncServerConnection):
> -    def __init__(self, reader, writer, table):
> +    def __init__(self, reader, writer, table, read_only):
>          super().__init__(reader, writer, 'PRSERVICE', logger)
>          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,
>          })
>          self.table = table
> +        self.read_only = read_only
>  
>      def validate_proto_version(self):
>          return (self.proto_version == (1, 0))
> @@ -56,16 +58,17 @@ class PRServerClient(bb.asyncrpc.AsyncServerConnection):
>          self.write_message(response)
>  
>      async def handle_import_one(self, request):
> -        version = request['version']
> -        pkgarch = request['pkgarch']
> -        checksum = request['checksum']
> -        value = request['value']
> +        response = None
> +        if not self.read_only:
> +            version = request['version']
> +            pkgarch = request['pkgarch']
> +            checksum = request['checksum']
> +            value = request['value']
> +
> +            value = self.table.importone(version, pkgarch, checksum, value)
> +            if value is not None:
> +                response = {'value': value}
>  
> -        value = self.table.importone(version, pkgarch, checksum, value)
> -        if value is not None:
> -            response = {'value': value}
> -        else:
> -            response = None
>          self.write_message(response)
>  
>      async def handle_export(self, request):
> @@ -83,20 +86,25 @@ class PRServerClient(bb.asyncrpc.AsyncServerConnection):
>          response = {'metainfo': metainfo, 'datainfo': datainfo}
>          self.write_message(response)
>  
> +    async def handle_is_readonly(self, request):
> +        response = {'readonly': self.read_only}
> +        self.write_message(response)
> +
>  class PRServer(bb.asyncrpc.AsyncServer):
> -    def __init__(self, dbfile, loop=None):
> +    def __init__(self, dbfile, loop=None, read_only=False):
>          super().__init__(logger, loop)
>          self.dbfile = dbfile
>          self.table = None
> +        self.read_only = read_only
>  
>      def accept_client(self, reader, writer):
> -        return PRServerClient(reader, writer, self.table)
> +        return PRServerClient(reader, writer, self.table, self.read_only)
>  
>      def serve_forever(self):
> -        self.db = prserv.db.PRData(self.dbfile)
> +        self.db = prserv.db.PRData(self.dbfile, read_only=self.read_only)
>          self.table = self.db["PRMAIN"]
>  
> -        logger.debug("Started PRServer with DBfile: %s, Address: %s, PID: %s" %
> +        logger.info("Started PRServer with DBfile: %s, Address: %s, PID: %s" %
>                       (self.dbfile, self.address, str(os.getpid())))
>  
>          super().serve_forever()
> @@ -149,10 +157,17 @@ class PRAsyncClient(bb.asyncrpc.AsyncClient):
>          if response:
>              return (response['metainfo'], response['datainfo'])
>  
> +    async def is_readonly(self):
> +        response = await self.send_message(
> +            {'is-readonly': {}}
> +        )
> +        if response:
> +            return response['readonly']
> +
>  class PRClient(bb.asyncrpc.Client):
>      def __init__(self):
>          super().__init__()
> -        self._add_methods('getPR', 'importone', 'export')
> +        self._add_methods('getPR', 'importone', 'export', 'is_readonly')
>  
>      def _get_async_client(self):
>          return PRAsyncClient()
> @@ -227,7 +242,7 @@ def run_as_daemon(func, pidfile, logfile):
>      os.remove(pidfile)
>      os._exit(0)
>  
> -def start_daemon(dbfile, host, port, logfile):
> +def start_daemon(dbfile, host, port, logfile, read_only=False):
>      ip = socket.gethostbyname(host)
>      pidfile = PIDPREFIX % (ip, port)
>      try:
> @@ -243,7 +258,7 @@ def start_daemon(dbfile, host, port, logfile):
>  
>      dbfile = os.path.abspath(dbfile)
>      def daemon_main():
> -        server = PRServer(dbfile)
> +        server = PRServer(dbfile, read_only=read_only)
>          server.start_tcp_server(host, port)
>          server.serve_forever()
>  
>
> 
>

Re: [PATCH v5 5/5] prserv: Add read-only mode

[Jan-Simon Möller]

[-- Attachment #1: Type: text/plain, Size: 71 bytes --]

Hi Armin!

Yes, it is. Its driven by AGL and Paul/Scott work on it.

[-- Attachment #2: Type: text/html, Size: 79 bytes --]