Fix: sqlite3.OperationalError: database is locked

Multiprocessing causes locking error when write and commits
from multiple threads are interleaved. Make sure that commits
are performed under the same Lock() session.

Author: brubbel

opcua/common/sqlite3_backend.py

@@ -39,34 +39,45 @@ def execute_read(self, dbCmd = None, params = (), CB = None):
             c = self._get_conn().cursor()
             for row in c.execute(dbCmd, params):
                 CB(row)
+            c.close()
 
-    def execute_write(self, dbCmd = None, params = ()):
+    def execute_write(self, dbCmd = None, params = (), commit=True):
         with self._lock:
-            c = self._get_conn().cursor()
-            c.execute(dbCmd, params)
-
-    def commit(self):
+            conn = self._get_conn()
+            if dbCmd is not None:
+                c = conn.cursor()
+                c.execute(dbCmd, params)
+                c.close()
+            if bool(commit) is True:
+                conn.commit()
+                self._wal_throttled()
+
+    def wal_throttled_threadsafe(self):
         with self._lock:
-            self._get_conn().commit()
             self._wal_throttled()
 
-    def wal_checkpoint(self):
+    def wal_checkpoint_threadsafe(self):
         """
         Store checkpoint: forces database modifications to be persistent.
         Automatically done when sqlite cache runs over the 1000 pages threshold.
         IMPORTANT: slow operation, manual syncs are only useful for sporadic
         transactions that you really want to survive a power loss.
         """
-        self._lastCheckP = time.time()
-        c = self._get_conn().cursor()
-        c.execute('PRAGMA wal_checkpoint')
-
+        with self._lock:
+            self._wal_checkpoint()
+    
     # PRIVATE METHODS
     def _wal_throttled(self):
         # commits still require a wal_checkpoint to become persistent.
         if abs(time.time() - self._lastCheckP) < self.CHECKP_INTERVAL:
             return
-        self.wal_checkpoint()
+        self._wal_checkpoint()
+
+    def _wal_checkpoint(self):
+        self._lastCheckP = time.time()
+        c = self._get_conn().cursor()
+        c.execute('PRAGMA wal_checkpoint')
+        c.close()
 
     def _db_connect(self):
         CID = SQLite3Backend._getCID()
@@ -101,9 +112,7 @@ def _db_connect_py2(self, CID):
     def _db_disconnect(self):
         # Commit, checkpoint.
         if self.readonly is False:
-            with self._lock:
-                self._get_conn().commit()
-                self.wal_checkpoint()
+            self.wal_checkpoint_threadsafe()
         # Close all connections to database.
         for CID in self._conn:
             self._conn[CID].close()

opcua/server/address_space_sqlite.py

@@ -57,7 +57,7 @@ def __init__(self, aspace, ndata):
 
     def __setitem__(self, attrId, attr):
         def onchange_cb():
-            self.aspace._insert_attribute_threadsafe(self.nodeid, attrId, self[attrId])
+            self.aspace._insert_attribute_threadsafe(self.nodeid, attrId, self[attrId], commit=True)
         mAttr = MonitoredAttribute(attr, onchange_cb)
         dict.__setitem__(self, attrId, mAttr)
         mAttr.onchange_cb()
@@ -74,7 +74,7 @@ def __init__(self, aspace, ndata):
 
     def append(self, ref):
         list.append(self, ref)
-        self._aspace._insert_reference_threadsafe(self.nodeid, ref)
+        self._aspace._insert_reference_threadsafe(self.nodeid, ref, commit=True)
 
     def remove(self, ref):
         raise NotImplementedError
@@ -172,7 +172,6 @@ def _setitem_backend(self, nodeid, ndata):
 
         if ndata.attributes.aspace is self:
             self._write_nodedata(ndata)
-            self.backend.commit()
 
     @staticmethod
     def _nodeid_surjection(nodeid):
@@ -211,7 +210,6 @@ def dump(self, namespaceidx=AddressSpace.DEFAULT_USER_NAMESPACE_INDEX):
         """
         with self._lock:
             self._dump(namespaceidx)
-            self.backend.commit()
         print("Export to {:s} completed".format(str(self.backend)))
 
     def _dump(self, namespaceidx=AddressSpace.DEFAULT_USER_NAMESPACE_INDEX):
@@ -224,16 +222,19 @@ def _dump(self, namespaceidx=AddressSpace.DEFAULT_USER_NAMESPACE_INDEX):
             assert(nodeid == ndata.nodeid)
             assert(isinstance(ndata, NodeData))
             if nodeid.NamespaceIndex == namespaceidx:
-                self._write_nodedata(ndata)
+                self._write_nodedata(ndata, commit=False)
                 continue
             # inter-namespace references.
             for ref in ndata.references:
                 if ref.NodeId.NamespaceIndex != namespaceidx:
                     continue
                 mapNodeId = AddressSpaceSQLite._nodeid_surjection(ndata.nodeid)
-                self._insert_reference(mapNodeId, ref)
+                self._insert_reference(mapNodeId, ref, commit=False)
 
-        # 3. Integrity checks.
+        # 3. commit
+        self.backend.execute_write(dbCmd=None, commit=True)
+
+        # 4. Integrity checks.
         for nodeid, ndata in self._cache.items():
             if nodeid.NamespaceIndex != namespaceidx:
                 continue
@@ -242,22 +243,22 @@ def _dump(self, namespaceidx=AddressSpace.DEFAULT_USER_NAMESPACE_INDEX):
             AddressSpaceSQLite._cmp_nodedata(ndata, ndata2)
 
     # Write NodeData to database
-    def _write_nodedata(self, ndata):
+    def _write_nodedata(self, ndata, commit=True):
         mapNodeId = AddressSpaceSQLite._nodeid_surjection(ndata.nodeid)
-        self._write_attributes(mapNodeId, ndata)
-        self._write_references(mapNodeId, ndata)
+        self._write_attributes(mapNodeId, ndata, commit=commit)
+        self._write_references(mapNodeId, ndata, commit=commit)
 
-    def _write_attributes(self, nodeid, ndata):
+    def _write_attributes(self, nodeid, ndata, commit=True):
         assert(nodeid.NodeIdType == NodeIdType.Numeric)
         assert(isinstance(ndata.attributes, dict))
         for attrId, attr in ndata.attributes.items():
-            AddressSpaceSQLite._insert_attribute(self.backend, nodeid, attrId, attr)
+            AddressSpaceSQLite._insert_attribute(self.backend, nodeid, attrId, attr, commit=commit)
 
-    def _write_references(self, nodeid, ndata):
+    def _write_references(self, nodeid, ndata, commit=True):
         assert(nodeid.NodeIdType == NodeIdType.Numeric)
         assert(isinstance(ndata.references, list))
         for ref in ndata.references:
-            AddressSpaceSQLite._insert_reference(self.backend, nodeid, ref)
+            AddressSpaceSQLite._insert_reference(self.backend, nodeid, ref, commit=commit)
 
     # Read NodeData from database
     @staticmethod
@@ -309,22 +310,20 @@ def _create_attr_table(backend, table=ATTR_TABLE_NAME, drop=False):
         ]
         if drop is True:
             dropCmd = 'DROP TABLE IF EXISTS "{tn}"'.format(tn=table)
-            backend.execute_write(dropCmd)
+            backend.execute_write(dropCmd, commit=True)
         cmd = 'CREATE TABLE IF NOT EXISTS "{tn}" ({c})'.format(tn=table, c=', '.join(ATTR_TABLE_COLS))
-        backend.execute_write(cmd)
+        backend.execute_write(cmd, commit=True)
 
-    def _insert_attribute_threadsafe(self, nodeid, attrId, attr, table=ATTR_TABLE_NAME):
+    def _insert_attribute_threadsafe(self, nodeid, attrId, attr, table=ATTR_TABLE_NAME, commit=True):
         with self._lock:
             if nodeid == AddressSpaceSQLite.CUR_TIME_NODEID:
-                pass # Prevents SD-card wear: don't write the time.
+                # Prevent sd-card wear: don't write the time. Use as trigger for WAL checkpoints.
+                self.backend.wal_throttled_threadsafe()
             else:
-                AddressSpaceSQLite._insert_attribute(self.backend, nodeid, attrId, attr, table)
-            # CurrentTime-node updates result in commits at COMMIT_INTERVAL sec.
-            # Commits without previous actual transactions don't touch the file.
-            self.backend.commit()
+                AddressSpaceSQLite._insert_attribute(self.backend, nodeid, attrId, attr, table, commit=commit)
 
     @staticmethod
-    def _insert_attribute(backend, nodeid, attrId, attr, table=ATTR_TABLE_NAME):
+    def _insert_attribute(backend, nodeid, attrId, attr, table=ATTR_TABLE_NAME, commit=True):
         assert(nodeid.NodeIdType == NodeIdType.Numeric)
         assert(isinstance(attrId, ua.AttributeIds))
         assert(isinstance(attr, AttributeValue))
@@ -363,7 +362,7 @@ def _insert_attribute(backend, nodeid, attrId, attr, table=ATTR_TABLE_NAME):
           sqlite3.Binary(ua.ua_binary.variant_to_binary(attr.value.Value)),
           str(nodeid)
         )
-        backend.execute_write(cmd, params=params)
+        backend.execute_write(cmd, params=params, commit=commit)
 
     @staticmethod
     def _read_attribute_row(row):
@@ -402,17 +401,16 @@ def _create_refs_table(backend, table=REFS_TABLE_NAME, drop=False):
         ]
         if drop is True:
             dropCmd = 'DROP TABLE IF EXISTS "{tn}"'.format(tn=table)
-            backend.execute_write(dropCmd)
+            backend.execute_write(dropCmd, commit=True)
         cmd = 'CREATE TABLE IF NOT EXISTS "{tn}" ({c})'.format(tn=table, c=', '.join(REFS_TABLE_COLS))
-        backend.execute_write(cmd)
+        backend.execute_write(cmd, commit=True)
 
-    def _insert_reference_threadsafe(self, nodeid, ref, table=REFS_TABLE_NAME):
+    def _insert_reference_threadsafe(self, nodeid, ref, table=REFS_TABLE_NAME, commit=True):
         with self._lock:
-            AddressSpaceSQLite._insert_reference(self.backend, nodeid, ref, table)
-            self.backend.commit()
+            AddressSpaceSQLite._insert_reference(self.backend, nodeid, ref, table, commit=commit)
 
     @staticmethod
-    def _insert_reference(backend, nodeid, ref, table=REFS_TABLE_NAME):
+    def _insert_reference(backend, nodeid, ref, table=REFS_TABLE_NAME, commit=True):
         # NumericNodeId is required for searching.
         assert(nodeid.NodeIdType == NodeIdType.Numeric)
         assert(isinstance(ref, ua.uaprotocol_auto.ReferenceDescription))
@@ -452,7 +450,7 @@ def _insert_reference(backend, nodeid, ref, table=REFS_TABLE_NAME):
           sqlite3.Binary(ua.ua_binary.nodeid_to_binary(ref.TypeDefinition)),
           str(nodeid)
         )
-        backend.execute_write(cmd, params=params)
+        backend.execute_write(cmd, params=params, commit=commit)
 
     @staticmethod
     def _read_reference_row(row):