[v5,4/5] prserv: enable database sharing

From: Michael Opdenacker <michael.opdenacker@bootlin.com>

sqlite3 can allow multiple processes to access the database
simultaneously, but it must be opened correctly. The key change is that
the database is no longer opened in "exclusive" mode (defaulting to
shared mode). In addition, the journal is set to "WAL" mode, as this is
the most efficient for dealing with simultaneous access between
different processes. In order to keep the database performance,
synchronous mode is set to "off". The WAL journal will protect against
incomplete transactions in any given client, however the database will
not be protected against unexpected power loss from the OS (which is a
fine trade off for performance, and also the same as the previous
implementation).

The use of a database cursor enabled to remove the _execute() wrapper.
The cursor automatically makes sure that the query happens in an atomic
transaction and commits when finished.

This also removes the need for a "dirty" flag for the database and
for explicit database syncing, which simplifies the code.

Signed-off-by: Michael Opdenacker <michael.opdenacker@bootlin.com>
Signed-off-by: Joshua Watt <JPEWhacker@gmail.com>
Cc: Tim Orling <ticotimo@gmail.com>
Cc: Thomas Petazzoni <thomas.petazzoni@bootlin.com>
---
 lib/prserv/db.py   | 322 +++++++++++++++++++++------------------------
 lib/prserv/serv.py |   8 --
 2 files changed, 151 insertions(+), 179 deletions(-)

On Mon, Apr 29, 2024 at 2:14 PM <michael.opdenacker@bootlin.com> wrote:
>
> From: Michael Opdenacker <michael.opdenacker@bootlin.com>
>
> sqlite3 can allow multiple processes to access the database
> simultaneously, but it must be opened correctly. The key change is that
> the database is no longer opened in "exclusive" mode (defaulting to
> shared mode). In addition, the journal is set to "WAL" mode, as this is
> the most efficient for dealing with simultaneous access between
> different processes. In order to keep the database performance,
> synchronous mode is set to "off". The WAL journal will protect against
> incomplete transactions in any given client, however the database will
> not be protected against unexpected power loss from the OS (which is a
> fine trade off for performance, and also the same as the previous
> implementation).
>
> The use of a database cursor enabled to remove the _execute() wrapper.
> The cursor automatically makes sure that the query happens in an atomic
> transaction and commits when finished.
>
> This also removes the need for a "dirty" flag for the database and
> for explicit database syncing, which simplifies the code.
>
> Signed-off-by: Michael Opdenacker <michael.opdenacker@bootlin.com>
> Signed-off-by: Joshua Watt <JPEWhacker@gmail.com>
> Cc: Tim Orling <ticotimo@gmail.com>
> Cc: Thomas Petazzoni <thomas.petazzoni@bootlin.com>
> ---
>  lib/prserv/db.py   | 322 +++++++++++++++++++++------------------------
>  lib/prserv/serv.py |   8 --
>  2 files changed, 151 insertions(+), 179 deletions(-)
>
> diff --git a/lib/prserv/db.py b/lib/prserv/db.py
> index b2520f3158..f430586d73 100644
> --- a/lib/prserv/db.py
> +++ b/lib/prserv/db.py
> @@ -8,21 +8,13 @@ import logging
>  import os.path
>  import errno
>  import prserv
> -import time
> +import sqlite3
>
> +from contextlib import closing
>  from . import increase_revision, revision_greater, revision_smaller
>
> -try:
> -    import sqlite3
> -except ImportError:
> -    from pysqlite2 import dbapi2 as sqlite3
> -
>  logger = logging.getLogger("BitBake.PRserv")
>
> -sqlversion = sqlite3.sqlite_version_info
> -if sqlversion[0] < 3 or (sqlversion[0] == 3 and sqlversion[1] < 3):
> -    raise Exception("sqlite3 version 3.3.0 or later is required.")
> -
>  #
>  # "No History" mode - for a given query tuple (version, pkgarch, checksum),
>  # the returned value will be the largest among all the values of the same
> @@ -31,40 +23,28 @@ if sqlversion[0] < 3 or (sqlversion[0] == 3 and sqlversion[1] < 3):
>  # "History" mode - Return a new higher value for previously unseen query
>  # tuple (version, pkgarch, checksum), otherwise return historical value.
>  # Value can decrement if returning to a previous build.
> -#
>
>  class PRTable(object):
>      def __init__(self, conn, table, read_only):
>          self.conn = conn
>          self.read_only = read_only
> -        self.dirty = False
>          self.table = 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 TEXT, \
> -                        PRIMARY KEY (version, pkgarch, checksum, value));" % self.table)
> -
> -    def _execute(self, *query):
> -        """Execute a query, waiting to acquire a lock if necessary"""
> -        start = time.time()
> -        end = start + 20
> -        while True:
> -            try:
> -                return self.conn.execute(*query)
> -            except sqlite3.OperationalError as exc:
> -                if "is locked" in str(exc) and end > time.time():
> -                    continue
> -                raise exc
> +        with closing(self.conn.cursor()) as cursor:
> +            if self.read_only:
> +                table_exists = cursor.execute(
> +                            "SELECT count(*) FROM sqlite_master \
> +                            WHERE type='table' AND name='%s'" % (self.table))
> +                if not table_exists:
> +                    raise prserv.NotFoundError

This is a race if you run the read-only server first, which seems like
a bad user experience. I'd probably create the table even if the
server is read-only (I'm pretty sure hash server works this way also)

> +            else:
> +                cursor.execute("CREATE TABLE IF NOT EXISTS %s \
> +                            (version TEXT NOT NULL, \
> +                            pkgarch TEXT NOT NULL,  \
> +                            checksum TEXT NOT NULL, \
> +                            value TEXT, \
> +                            PRIMARY KEY (version, pkgarch, checksum, value));" % self.table)
> +                self.conn.commit()
>
>      def _extremum_value(self, rows, is_max):
>          value = None
> @@ -88,49 +68,42 @@ class PRTable(object):
>      def _min_value(self, rows):
>          return self._extremum_value(rows, False)
>
> -    def sync(self):
> -        if not self.read_only:
> -            self.conn.commit()
> -            self._execute("BEGIN EXCLUSIVE TRANSACTION")
> -
> -    def sync_if_dirty(self):
> -        if self.dirty:
> -            self.sync()
> -            self.dirty = False
> -
>      def test_package(self, version, pkgarch):
>          """Returns whether the specified package version is found in the database for the specified architecture"""
>
>          # Just returns the value if found or None otherwise
> -        data=self._execute("SELECT value FROM %s WHERE version=? AND pkgarch=?;" % self.table,
> -                           (version, pkgarch))
> -        row=data.fetchone()
> -        if row is not None:
> -            return True
> -        else:
> -            return False
> +        with closing(self.conn.cursor()) as cursor:
> +            data=cursor.execute("SELECT value FROM %s WHERE version=? AND pkgarch=?;" % self.table,
> +                               (version, pkgarch))
> +            row=data.fetchone()
> +            if row is not None:
> +                return True
> +            else:
> +                return False
>
>      def test_value(self, version, pkgarch, value):
>          """Returns whether the specified value is found in the database for the specified package and architecture"""
>
>          # Just returns the value if found or None otherwise
> -        data=self._execute("SELECT value FROM %s WHERE version=? AND pkgarch=? and value=?;" % self.table,
> -                           (version, pkgarch, value))
> -        row=data.fetchone()
> -        if row is not None:
> -            return True
> -        else:
> -            return False
> +        with closing(self.conn.cursor()) as cursor:
> +            data=cursor.execute("SELECT value FROM %s WHERE version=? AND pkgarch=? and value=?;" % self.table,
> +                               (version, pkgarch, value))
> +            row=data.fetchone()
> +            if row is not None:
> +                return True
> +            else:
> +                return False
>
>
>      def find_package_max_value(self, version, pkgarch):
>          """Returns the greatest value for (version, pkgarch), or None if not found. Doesn't create a new value"""
>
> -        data = self._execute("SELECT value FROM %s where version=? AND pkgarch=?;" % (self.table),
> -                             (version, pkgarch))
> -        rows = data.fetchall()
> -        value = self._max_value(rows)
> -        return value
> +        with closing(self.conn.cursor()) as cursor:
> +            data = cursor.execute("SELECT value FROM %s where version=? AND pkgarch=?;" % (self.table),
> +                                 (version, pkgarch))
> +            rows = data.fetchall()
> +            value = self._max_value(rows)
> +            return value
>
>      def find_value(self, version, pkgarch, checksum, history=False):
>          """Returns the value for the specified checksum if found or None otherwise."""
> @@ -145,10 +118,11 @@ class PRTable(object):
>          """Returns the maximum (if is_max is True) or minimum (if is_max is False) value
>             for (version, pkgarch, checksum), or None if not found. Doesn't create a new value"""
>
> -        data = self._execute("SELECT value FROM %s where version=? AND pkgarch=? AND checksum=?;" % (self.table),
> -                             (version, pkgarch, checksum))
> -        rows = data.fetchall()
> -        return self._extremum_value(rows, is_max)
> +        with closing(self.conn.cursor()) as cursor:
> +            data = cursor.execute("SELECT value FROM %s where version=? AND pkgarch=? AND checksum=?;" % (self.table),
> +                                 (version, pkgarch, checksum))
> +            rows = data.fetchall()
> +            return self._extremum_value(rows, is_max)
>
>      def find_max_value(self, version, pkgarch, checksum):
>          return self._find_extremum_value(version, pkgarch, checksum, True)
> @@ -160,26 +134,27 @@ class PRTable(object):
>          """Take and increase the greatest "<base>.y" value for (version, pkgarch), or return "<base>.0" if not found.
>          This doesn't store a new value."""
>
> -        data = self._execute("SELECT value FROM %s where version=? AND pkgarch=? AND value LIKE '%s.%%';" % (self.table, base),
> -                             (version, pkgarch))
> -        rows = data.fetchall()
> -        value = self._max_value(rows)
> +        with closing(self.conn.cursor()) as cursor:
> +            data = cursor.execute("SELECT value FROM %s where version=? AND pkgarch=? AND value LIKE '%s.%%';" % (self.table, base),
> +                                 (version, pkgarch))
> +            rows = data.fetchall()
> +            value = self._max_value(rows)
>
> -        if value is not None:
> -            return increase_revision(value)
> -        else:
> -            return base + ".0"
> +            if value is not None:
> +                return increase_revision(value)
> +            else:
> +                return base + ".0"
>
>      def store_value(self, version, pkgarch, checksum, value):
>          """Store new value in the database"""
>
> -        try:
> -            self._execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
> -                       (version, pkgarch, checksum, value))
> -        except sqlite3.IntegrityError as exc:
> -            logger.error(str(exc))
> -
> -        self.dirty = True
> +        with closing(self.conn.cursor()) as cursor:
> +            try:
> +                cursor.execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
> +                           (version, pkgarch, checksum, value))
> +            except sqlite3.IntegrityError as exc:
> +                logger.error(str(exc))
> +            self.conn.commit()
>
>      def _get_value(self, version, pkgarch, checksum, history):
>
> @@ -215,54 +190,56 @@ class PRTable(object):
>              return None
>
>          val = None
> -        data = self._execute("SELECT value FROM %s WHERE version=? AND pkgarch=? AND checksum=?;" % self.table,
> +        with closing(self.conn.cursor()) as cursor:
> +            data = cursor.execute("SELECT value FROM %s WHERE version=? AND pkgarch=? AND checksum=?;" % self.table,
>                             (version, pkgarch, checksum))
> -        row = data.fetchone()
> -        if row is not None:
> -            val=row[0]
> -        else:
> -            #no value found, try to insert
> -            try:
> -                self._execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
> -                           (version, pkgarch, checksum, value))
> -            except sqlite3.IntegrityError as exc:
> -                logger.error(str(exc))
> +            row = data.fetchone()
> +            if row is not None:
> +                val=row[0]
> +            else:
> +                #no value found, try to insert
> +                try:
> +                    cursor.execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
> +                               (version, pkgarch, checksum, value))
> +                except sqlite3.IntegrityError as exc:
> +                    logger.error(str(exc))

You probably want to make this whole SELECT / INSERT block a loop, and
instead of repeating the SELECT below, retry the loop on an integrity
error. The only reason you should be getting an integrity error is if
you have a race between multiple servers accessing the database at
once. In that case, looping back and trying to fall out in the
previous SELECT is better.

This is especially important if you delete entries in the database, as
in pathological cases you might have to loop several times before you
fall out of the loop (although typically, you would only need to retry
once).

I'd recommend this paradigm any other places you do SELECT / INSERT
where the INSERT depends on the SELECT.

>
> -            self.dirty = True
> +                self.conn.commit()
>
> -            data = self._execute("SELECT value FROM %s WHERE version=? AND pkgarch=? AND checksum=?;" % self.table,
> +                data = cursor.execute("SELECT value FROM %s WHERE version=? AND pkgarch=? AND checksum=?;" % self.table,
>                             (version, pkgarch, checksum))
> -            row = data.fetchone()
> -            if row is not None:
> -                val = row[0]
> +                row = data.fetchone()
> +                if row is not None:
> +                    val = row[0]
>          return val
>
>      def _import_no_hist(self, version, pkgarch, checksum, value):
>          if self.read_only:
>              return None
>
> -        try:
> -            #try to insert
> -            self._execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
> -                           (version, pkgarch, checksum, value))
> -        except sqlite3.IntegrityError as exc:
> -            #already have the record, try to update
> +        with closing(self.conn.cursor()) as cursor:
>              try:
> -                self._execute("UPDATE %s SET value=? WHERE version=? AND pkgarch=? AND checksum=? AND value<?"
> -                              % (self.table),
> -                               (value, version, pkgarch, checksum, value))
> +                #try to insert
> +                cursor.execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
> +                               (version, pkgarch, checksum, value))
>              except sqlite3.IntegrityError as exc:
> -                logger.error(str(exc))
> +                #already have the record, try to update
> +                try:
> +                    cursor.execute("UPDATE %s SET value=? WHERE version=? AND pkgarch=? AND checksum=? AND value<?"
> +                                  % (self.table),
> +                                   (value, version, pkgarch, checksum, value))
> +                except sqlite3.IntegrityError as exc:
> +                    logger.error(str(exc))

Looping is suggested here also.

>
> -        self.dirty = True
> +                self.conn.commit()
>
> -        data = self._execute("SELECT value FROM %s WHERE version=? AND pkgarch=? AND checksum=? AND value>=?;" % self.table,
> +            data = cursor.execute("SELECT value FROM %s WHERE version=? AND pkgarch=? AND checksum=? AND value>=?;" % self.table,
>                              (version, pkgarch, checksum, value))

You already have all the data for everything you inserted.... I'd
probably just return based on what you already know instead of another
SELECT, otherwise this could race (with a deletion) and fail.
Alternatively, I think sqlite can return the data for the
just-inserted-row, but I forget how off the top of my head.

> -        row=data.fetchone()
> -        if row is not None:
> -            return row[0]
> -        else:
> -            return None
> +            row=data.fetchone()
> +            if row is not None:
> +                return row[0]
> +            else:
> +                return None
>
>      def importone(self, version, pkgarch, checksum, value, history=False):
>          if history:
> @@ -272,56 +249,57 @@ class PRTable(object):
>
>      def export(self, version, pkgarch, checksum, colinfo, history=False):
>          metainfo = {}
> -        #column info
> -        if colinfo:
> -            metainfo["tbl_name"] = self.table
> -            metainfo["core_ver"] = prserv.__version__
> -            metainfo["col_info"] = []
> -            data = self._execute("PRAGMA table_info(%s);" % self.table)
> +        with closing(self.conn.cursor()) as cursor:
> +            #column info
> +            if colinfo:
> +                metainfo["tbl_name"] = self.table
> +                metainfo["core_ver"] = prserv.__version__
> +                metainfo["col_info"] = []
> +                data = cursor.execute("PRAGMA table_info(%s);" % self.table)
> +                for row in data:
> +                    col = {}
> +                    col["name"] = row["name"]
> +                    col["type"] = row["type"]
> +                    col["notnull"] = row["notnull"]
> +                    col["dflt_value"] = row["dflt_value"]
> +                    col["pk"] = row["pk"]
> +                    metainfo["col_info"].append(col)
> +
> +            #data info
> +            datainfo = []
> +
> +            if history:
> +                sqlstmt = "SELECT * FROM %s as T1 WHERE 1=1 " % self.table
> +            else:
> +                sqlstmt = "SELECT T1.version, T1.pkgarch, T1.checksum, T1.value FROM %s as T1, \
> +                        (SELECT version, pkgarch, max(value) as maxvalue FROM %s GROUP BY version, pkgarch) as T2 \
> +                        WHERE T1.version=T2.version AND T1.pkgarch=T2.pkgarch AND T1.value=T2.maxvalue " % (self.table, self.table)
> +            sqlarg = []
> +            where = ""
> +            if version:
> +                where += "AND T1.version=? "
> +                sqlarg.append(str(version))
> +            if pkgarch:
> +                where += "AND T1.pkgarch=? "
> +                sqlarg.append(str(pkgarch))
> +            if checksum:
> +                where += "AND T1.checksum=? "
> +                sqlarg.append(str(checksum))
> +
> +            sqlstmt += where + ";"
> +
> +            if len(sqlarg):
> +                data = cursor.execute(sqlstmt, tuple(sqlarg))
> +            else:
> +                data = cursor.execute(sqlstmt)
>              for row in data:
> -                col = {}
> -                col["name"] = row["name"]
> -                col["type"] = row["type"]
> -                col["notnull"] = row["notnull"]
> -                col["dflt_value"] = row["dflt_value"]
> -                col["pk"] = row["pk"]
> -                metainfo["col_info"].append(col)
> -
> -        #data info
> -        datainfo = []
> -
> -        if history:
> -            sqlstmt = "SELECT * FROM %s as T1 WHERE 1=1 " % self.table
> -        else:
> -            sqlstmt = "SELECT T1.version, T1.pkgarch, T1.checksum, T1.value FROM %s as T1, \
> -                    (SELECT version, pkgarch, max(value) as maxvalue FROM %s GROUP BY version, pkgarch) as T2 \
> -                    WHERE T1.version=T2.version AND T1.pkgarch=T2.pkgarch AND T1.value=T2.maxvalue " % (self.table, self.table)
> -        sqlarg = []
> -        where = ""
> -        if version:
> -            where += "AND T1.version=? "
> -            sqlarg.append(str(version))
> -        if pkgarch:
> -            where += "AND T1.pkgarch=? "
> -            sqlarg.append(str(pkgarch))
> -        if checksum:
> -            where += "AND T1.checksum=? "
> -            sqlarg.append(str(checksum))
> -
> -        sqlstmt += where + ";"
> -
> -        if len(sqlarg):
> -            data = self._execute(sqlstmt, tuple(sqlarg))
> -        else:
> -            data = self._execute(sqlstmt)
> -        for row in data:
> -            if row["version"]:
> -                col = {}
> -                col["version"] = row["version"]
> -                col["pkgarch"] = row["pkgarch"]
> -                col["checksum"] = row["checksum"]
> -                col["value"] = row["value"]
> -                datainfo.append(col)
> +                if row["version"]:
> +                    col = {}
> +                    col["version"] = row["version"]
> +                    col["pkgarch"] = row["pkgarch"]
> +                    col["checksum"] = row["checksum"]
> +                    col["value"] = row["value"]
> +                    datainfo.append(col)
>          return (metainfo, datainfo)
>
>      def dump_db(self, fd):
> @@ -345,14 +323,15 @@ class PRData(object):
>                  raise e
>          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=sqlite3.connect(uri, uri=True)
>          self.connection.row_factory=sqlite3.Row
> -        if not self.read_only:
> -            self.connection.execute("pragma synchronous = off;")
> -            self.connection.execute("PRAGMA journal_mode = MEMORY;")
> +        self.connection.execute("PRAGMA synchronous = OFF;")
> +        self.connection.execute("PRAGMA journal_mode = WAL;")
> +        self.connection.commit()
>          self._tables={}
>
>      def disconnect(self):
> +        self.connection.commit()
>          self.connection.close()
>
>      def __getitem__(self, tblname):
> @@ -370,3 +349,4 @@ class PRData(object):
>              del self._tables[tblname]
>          logger.info("drop table %s" % (tblname))
>          self.connection.execute("DROP TABLE IF EXISTS %s;" % tblname)
> +        self.connection.commit()
> diff --git a/lib/prserv/serv.py b/lib/prserv/serv.py
> index 05573d06cc..fd673b1851 100644
> --- a/lib/prserv/serv.py
> +++ b/lib/prserv/serv.py
> @@ -44,8 +44,6 @@ class PRServerClient(bb.asyncrpc.AsyncServerConnection):
>          except:
>              self.server.table.sync()
>              raise
> -        else:
> -            self.server.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'''
> @@ -233,15 +231,9 @@ class PRServer(bb.asyncrpc.AsyncServer):
>          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, upstream):
>          self.dbfile = dbfile
> --
> 2.34.1
>

Hi Joshua

Many thanks for the review and guidance. See my comments and questions 
below.

On 4/29/24 at 22:34, Joshua Watt wrote:
>> +        with closing(self.conn.cursor()) as cursor:
>> +            if self.read_only:
>> +                table_exists = cursor.execute(
>> +                            "SELECT count(*) FROM sqlite_master \
>> +                            WHERE type='table' AND name='%s'" % (self.table))
>> +                if not table_exists:
>> +                    raise prserv.NotFoundError
> This is a race if you run the read-only server first, which seems like
> a bad user experience. I'd probably create the table even if the
> server is read-only (I'm pretty sure hash server works this way also)


Fixed. Thanks.

>
>> +            row = data.fetchone()
>> +            if row is not None:
>> +                val=row[0]
>> +            else:
>> +                #no value found, try to insert
>> +                try:
>> +                    cursor.execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
>> +                               (version, pkgarch, checksum, value))
>> +                except sqlite3.IntegrityError as exc:
>> +                    logger.error(str(exc))
> You probably want to make this whole SELECT / INSERT block a loop, and
> instead of repeating the SELECT below, retry the loop on an integrity
> error. The only reason you should be getting an integrity error is if
> you have a race between multiple servers accessing the database at
> once. In that case, looping back and trying to fall out in the
> previous SELECT is better.
>
> This is especially important if you delete entries in the database, as
> in pathological cases you might have to loop several times before you
> fall out of the loop (although typically, you would only need to retry
> once).
>
> I'd recommend this paradigm any other places you do SELECT / INSERT
> where the INSERT depends on the SELECT.

This is actually coming from the prserv legacy code, which I haven't 
really touched yet.

Is this kind of event really expected to happen with sqlite3? I see that 
hashserve only checks for "sqlite3.IntegrityError" once, so maybe it's 
too pessimistic to anticipate errors here... Do we also want to support 
multiple builders accessing the same PR database?

Anyway, if I really have to do it, I guess I'd also have to do it in the 
store_value() function which currently just logs an error without 
retrying. But if we loop, doesn't it add the risk to create an endless 
loop?

Cheers,
Michael.

On Tue, Apr 30, 2024 at 6:58 AM Michael Opdenacker
<michael.opdenacker@bootlin.com> wrote:
>
> Hi Joshua
>
> Many thanks for the review and guidance. See my comments and questions
> below.
>
> On 4/29/24 at 22:34, Joshua Watt wrote:
> >> +        with closing(self.conn.cursor()) as cursor:
> >> +            if self.read_only:
> >> +                table_exists = cursor.execute(
> >> +                            "SELECT count(*) FROM sqlite_master \
> >> +                            WHERE type='table' AND name='%s'" % (self.table))
> >> +                if not table_exists:
> >> +                    raise prserv.NotFoundError
> > This is a race if you run the read-only server first, which seems like
> > a bad user experience. I'd probably create the table even if the
> > server is read-only (I'm pretty sure hash server works this way also)
>
>
> Fixed. Thanks.
>
> >
> >> +            row = data.fetchone()
> >> +            if row is not None:
> >> +                val=row[0]
> >> +            else:
> >> +                #no value found, try to insert
> >> +                try:
> >> +                    cursor.execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
> >> +                               (version, pkgarch, checksum, value))
> >> +                except sqlite3.IntegrityError as exc:
> >> +                    logger.error(str(exc))
> > You probably want to make this whole SELECT / INSERT block a loop, and
> > instead of repeating the SELECT below, retry the loop on an integrity
> > error. The only reason you should be getting an integrity error is if
> > you have a race between multiple servers accessing the database at
> > once. In that case, looping back and trying to fall out in the
> > previous SELECT is better.
> >
> > This is especially important if you delete entries in the database, as
> > in pathological cases you might have to loop several times before you
> > fall out of the loop (although typically, you would only need to retry
> > once).
> >
> > I'd recommend this paradigm any other places you do SELECT / INSERT
> > where the INSERT depends on the SELECT.
>
> This is actually coming from the prserv legacy code, which I haven't
> really touched yet.
>
> Is this kind of event really expected to happen with sqlite3? I see that
> hashserve only checks for "sqlite3.IntegrityError" once, so maybe it's
> too pessimistic to anticipate errors here... Do we also want to support
> multiple builders accessing the same PR database?

The hash server API and databases are designed to avoid a SELECT /
INSERT dependency. If you look, the APIs are either just reading with
SELECT, or doing an INSERT that ignores if the entry already exists.
After the INSERT, it sometimes does a SELECT to try and read back the
inserted data, but this is fine as long as it correctly handles the
case where it might not get back the data that was just inserted (if
it's racing with a delete), which it does correctly. (e.g. INSERT ->
SELECT is fine, SELECT -> INSERT requires a retry).

>
> Anyway, if I really have to do it, I guess I'd also have to do it in the
> store_value() function which currently just logs an error without
> retrying. But if we loop, doesn't it add the risk to create an endless
> loop?

Sans bugs, no. In order to loop again, the database state must change
between the SELECT and INSERT statements (remember, a cursor is not a
read-lock on the database). As such, some other server sharing the
database *must* have made progress and successfully done an INSERT, so
holistically, the entire system (all servers) have made progress and
it's not an endless loop.


>
> Cheers,
> Michael.
>
> --
>
> Michael Opdenacker, Bootlin
> Embedded Linux and Kernel engineering
> https://bootlin.com
>

On Tue, Apr 30, 2024 at 7:23 AM Joshua Watt <jpewhacker@gmail.com> wrote:
>
> On Tue, Apr 30, 2024 at 6:58 AM Michael Opdenacker
> <michael.opdenacker@bootlin.com> wrote:
> >
> > Hi Joshua
> >
> > Many thanks for the review and guidance. See my comments and questions
> > below.
> >
> > On 4/29/24 at 22:34, Joshua Watt wrote:
> > >> +        with closing(self.conn.cursor()) as cursor:
> > >> +            if self.read_only:
> > >> +                table_exists = cursor.execute(
> > >> +                            "SELECT count(*) FROM sqlite_master \
> > >> +                            WHERE type='table' AND name='%s'" % (self.table))
> > >> +                if not table_exists:
> > >> +                    raise prserv.NotFoundError
> > > This is a race if you run the read-only server first, which seems like
> > > a bad user experience. I'd probably create the table even if the
> > > server is read-only (I'm pretty sure hash server works this way also)
> >
> >
> > Fixed. Thanks.
> >
> > >
> > >> +            row = data.fetchone()
> > >> +            if row is not None:
> > >> +                val=row[0]
> > >> +            else:
> > >> +                #no value found, try to insert
> > >> +                try:
> > >> +                    cursor.execute("INSERT INTO %s VALUES (?, ?, ?, ?);"  % (self.table),
> > >> +                               (version, pkgarch, checksum, value))
> > >> +                except sqlite3.IntegrityError as exc:
> > >> +                    logger.error(str(exc))
> > > You probably want to make this whole SELECT / INSERT block a loop, and
> > > instead of repeating the SELECT below, retry the loop on an integrity
> > > error. The only reason you should be getting an integrity error is if
> > > you have a race between multiple servers accessing the database at
> > > once. In that case, looping back and trying to fall out in the
> > > previous SELECT is better.
> > >
> > > This is especially important if you delete entries in the database, as
> > > in pathological cases you might have to loop several times before you
> > > fall out of the loop (although typically, you would only need to retry
> > > once).
> > >
> > > I'd recommend this paradigm any other places you do SELECT / INSERT
> > > where the INSERT depends on the SELECT.
> >
> > This is actually coming from the prserv legacy code, which I haven't
> > really touched yet.
> >
> > Is this kind of event really expected to happen with sqlite3? I see that
> > hashserve only checks for "sqlite3.IntegrityError" once, so maybe it's
> > too pessimistic to anticipate errors here... Do we also want to support
> > multiple builders accessing the same PR database?
>
> The hash server API and databases are designed to avoid a SELECT /
> INSERT dependency. If you look, the APIs are either just reading with
> SELECT, or doing an INSERT that ignores if the entry already exists.
> After the INSERT, it sometimes does a SELECT to try and read back the
> inserted data, but this is fine as long as it correctly handles the
> case where it might not get back the data that was just inserted (if
> it's racing with a delete), which it does correctly. (e.g. INSERT ->
> SELECT is fine, SELECT -> INSERT requires a retry).

Sorry, also forgot to mention: if your use case is _just_ a read-write
server and a read-only server, then you are correct, you do not need
to worry about retries. I think that's your only use case for this
work, so the above statements are just for posterity, as I suspect we
will be doing more work on the PR server after you finish the current
work :)

>
> >
> > Anyway, if I really have to do it, I guess I'd also have to do it in the
> > store_value() function which currently just logs an error without
> > retrying. But if we loop, doesn't it add the risk to create an endless
> > loop?
>
> Sans bugs, no. In order to loop again, the database state must change
> between the SELECT and INSERT statements (remember, a cursor is not a
> read-lock on the database). As such, some other server sharing the
> database *must* have made progress and successfully done an INSERT, so
> holistically, the entire system (all servers) have made progress and
> it's not an endless loop.
>
>
> >
> > Cheers,
> > Michael.
> >
> > --
> >
> > Michael Opdenacker, Bootlin
> > Embedded Linux and Kernel engineering
> > https://bootlin.com
> >

Hi Joshua,

On 4/30/24 at 15:55, Joshua Watt wrote:
>
> Sorry, also forgot to mention: if your use case is _just_ a read-write
> server and a read-only server, then you are correct, you do not need
> to worry about retries. I think that's your only use case for this
> work, so the above statements are just for posterity, as I suspect we
> will be doing more work on the PR server after you finish the current
> work :)


Great, good to know thanks!
This will make things simpler indeed.
Cheers
Michael.