pickle shelffile -> standard_address_space.sql

Refactoring the shelffile from a monolithic blob to sqlite3 db.
Advantages over pickle:
- Fast
- Small (25% of fill_address_space code)
- Not dependent on Python2/3 version
- Supports transactional read/(write TODO)
- Supports real-time persistence of user address-space (TODO)
- Strong typed: only INTEGER, TEXT and opc-ua to_binary() BLOB
- Drop-in replacement for memory-based AddressSpace

Built-in integrity check on generate_address_space.py dump.

Author: brubbel

opcua/common/utils.py

@@ -9,6 +9,7 @@
 import functools
 import threading
 from socket import error as SocketError
+from collections import MutableMapping
 
 try:
     import asyncio
@@ -206,4 +207,53 @@ def run_until_complete(self, coro):
         return future.result()
 
 
+class ThreadSafeDict(MutableMapping):
 
+    def __init__(self, cache=None):
+        self._lock = threading.RLock()  # FIXME: should use multiple reader, one writter pattern
+        if cache is None:
+            self._cache = {}
+        else:
+            assert(isinstance(cache, (dict, ThreadSafeDict)))
+            self._cache = cache
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self._cache = None
+
+    def __getitem__(self, key):
+        with self._lock:
+            return self._cache.__getitem__(key)
+
+    def get(self, key, value=None):
+        with self._lock:
+            return self._cache.get(key, value)
+
+    def __setitem__(self, key, value):
+        with self._lock:
+            return self._cache.__setitem__(key, value)
+
+    def __contains__(self, key):
+        with self._lock:
+            return self._cache.__contains__(key)
+
+    def __delitem__(self, key):
+        with self._lock:
+            del self._cache[key]
+
+    def __iter__(self):
+        with self._lock:
+            return iter(self._cache.keys())
+
+    def __len__(self):
+        return len(self._cache)
+
+    def keys(self):
+        with self._lock:
+            return self._cache.keys()
+
+    def empty(self):
+        with self._lock:
+            self._cache = {}

opcua/server/address_space.py

@@ -1,15 +1,9 @@
-from threading import RLock
 import logging
 from datetime import datetime
-import collections
-import shelve
-try:
-    import cPickle as pickle
-except:
-    import pickle
 
 from opcua import ua
 from opcua.server.user_manager import UserManager
+from opcua.common.utils import ThreadSafeDict
 
 
 class AttributeValue(object):
@@ -478,51 +472,30 @@ def _call(self, method):
         return res
 
 
-class AddressSpace(object):
+class AddressSpace(ThreadSafeDict):
 
     """
     The address space object stores all the nodes of the OPC-UA server
     and helper methods.
     The methods are thread safe
     """
 
-    def __init__(self):
+    def __init__(self, cache=None):
+        super(AddressSpace, self).__init__(cache)
         self.logger = logging.getLogger(__name__)
-        self._nodes = {}
-        self._lock = RLock()  # FIXME: should use multiple reader, one writter pattern
         self._datachange_callback_counter = 200
         self._handle_to_attribute_map = {}
         self._default_idx = 2
         self._nodeid_counter = {0: 20000, 1: 2000}
 
-    def __getitem__(self, nodeid):
-        with self._lock:
-            return self._nodes.__getitem__(nodeid)
-
-    def get(self, nodeid):
-        with self._lock:
-            return self._nodes.get(nodeid, None)
-
-    def __setitem__(self, nodeid, value):
-        with self._lock:
-            return self._nodes.__setitem__(nodeid, value)
-
-    def __contains__(self, nodeid):
-        with self._lock:
-            return self._nodes.__contains__(nodeid)
-
-    def __delitem__(self, nodeid):
-        with self._lock:
-            self._nodes.__delitem__(nodeid)
-
     def generate_nodeid(self, idx=None):
         if idx is None:
             idx = self._default_idx
         if idx in self._nodeid_counter:
             self._nodeid_counter[idx] += 1
         else:
             # get the biggest identifier number from the existed nodes in address space
-            identifier_list = sorted([nodeid.Identifier for nodeid in self._nodes.keys()
+            identifier_list = sorted([nodeid.Identifier for nodeid in self.keys()
                                       if nodeid.NamespaceIndex == idx and nodeid.NodeIdType
                                       in (ua.NodeIdType.Numeric, ua.NodeIdType.TwoByte, ua.NodeIdType.FourByte)])
             if identifier_list:
@@ -532,111 +505,19 @@ def generate_nodeid(self, idx=None):
         nodeid = ua.NodeId(self._nodeid_counter[idx], idx)
         with self._lock:  # OK since reentrant lock
             while True:
-                if nodeid in self._nodes:
+                if nodeid in self.keys():
                     nodeid = self.generate_nodeid(idx)
                 else:
                     return nodeid
 
-    def keys(self):
-        with self._lock:
-            return self._nodes.keys()
-
-    def empty(self):
-        """
-        Delete all nodes in address space
-        """
-        with self._lock:
-            self._nodes = {}
-
-    def dump(self, path):
-        """
-        Dump address space as binary to file; note that server must be stopped for this method to work
-        DO NOT DUMP AN ADDRESS SPACE WHICH IS USING A SHELF (load_aspace_shelf), ONLY CACHED NODES WILL GET DUMPED!
-        """
-        # prepare nodes in address space for being serialized
-        for nodeid, ndata in self._nodes.items():
-            # if the node has a reference to a method call, remove it so the object can be serialized
-            if ndata.call is not None:
-                self._nodes[nodeid].call = None
-
-        with open(path, 'wb') as f:
-            pickle.dump(self._nodes, f, pickle.HIGHEST_PROTOCOL)
-
-    def load(self, path):
-        """
-        Load address space from a binary file, overwriting everything in the current address space
-        """
-        with open(path, 'rb') as f:
-            self._nodes = pickle.load(f)
-
-    def make_aspace_shelf(self, path):
-        """
-        Make a shelf for containing the nodes from the standard address space; this is typically only done on first
-        start of the server. Subsequent server starts will load the shelf, nodes are then moved to a cache
-        by the LazyLoadingDict class when they are accessed. Saving data back to the shelf
-        is currently NOT supported, it is only used for the default OPC UA standard address space
-
-        Note: Intended for slow devices, such as Raspberry Pi, to greatly improve start up time
-        """
-        s = shelve.open(path, "n", protocol=pickle.HIGHEST_PROTOCOL)
-        for nodeid, ndata in self._nodes.items():
-            s[nodeid.to_string()] = ndata
-        s.close()
-
-    def load_aspace_shelf(self, path):
-        """
-        Load the standard address space nodes from a python shelve via LazyLoadingDict as needed.
-        The dump() method can no longer be used if the address space is being loaded from a shelf
-
-        Note: Intended for slow devices, such as Raspberry Pi, to greatly improve start up time
-        """
-        class LazyLoadingDict(collections.MutableMapping):
-            """
-            Special dict that only loads nodes as they are accessed. If a node is accessed it gets copied from the
-            shelve to the cache dict. All user nodes are saved in the cache ONLY. Saving data back to the shelf
-            is currently NOT supported
-            """
-            def __init__(self, source):
-                self.source = source  # python shelf
-                self.cache = {}  # internal dict
-
-            def __getitem__(self, key):
-                # try to get the item (node) from the cache, if it isn't there get it from the shelf
-                try:
-                    return self.cache[key]
-                except KeyError:
-                    node = self.cache[key] = self.source[key.to_string()]
-                    return node
-
-            def __setitem__(self, key, value):
-                # add a new item to the cache; if this item is in the shelf it is not updated
-                self.cache[key] = value
-
-            def __contains__(self, key):
-                return key in self.cache or key.to_string() in self.source
-
-            def __delitem__(self, key):
-                # only deleting items from the cache is allowed
-                del self.cache[key]
-
-            def __iter__(self):
-                # only the cache can be iterated over
-                return iter(self.cache.keys())
-
-            def __len__(self):
-                # only returns the length of items in the cache, not unaccessed items in the shelf
-                return len(self.cache)
-
-        self._nodes = LazyLoadingDict(shelve.open(path, "r"))
-
     def get_attribute_value(self, nodeid, attr):
         with self._lock:
             self.logger.debug("get attr val: %s %s", nodeid, attr)
-            if nodeid not in self._nodes:
+            if nodeid not in self.keys():
                 dv = ua.DataValue()
                 dv.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown)
                 return dv
-            node = self._nodes[nodeid]
+            node = self[nodeid]
             if attr not in node.attributes:
                 dv = ua.DataValue()
                 dv.StatusCode = ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid)
@@ -649,7 +530,7 @@ def get_attribute_value(self, nodeid, attr):
     def set_attribute_value(self, nodeid, attr, value):
         with self._lock:
             self.logger.debug("set attr val: %s %s %s", nodeid, attr, value)
-            node = self._nodes.get(nodeid, None)
+            node = self.get(nodeid, None)
             if node is None:
                 return ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown)
             attval = node.attributes.get(attr, None)
@@ -673,9 +554,9 @@ def set_attribute_value(self, nodeid, attr, value):
     def add_datachange_callback(self, nodeid, attr, callback):
         with self._lock:
             self.logger.debug("set attr callback: %s %s %s", nodeid, attr, callback)
-            if nodeid not in self._nodes:
+            if nodeid not in self.keys():
                 return ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown), 0
-            node = self._nodes[nodeid]
+            node = self[nodeid]
             if attr not in node.attributes:
                 return ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid), 0
             attval = node.attributes[attr]
@@ -689,9 +570,9 @@ def delete_datachange_callback(self, handle):
         with self._lock:
             if handle in self._handle_to_attribute_map:
                 nodeid, attr = self._handle_to_attribute_map.pop(handle)
-                self._nodes[nodeid].attributes[attr].datachange_callbacks.pop(handle)
+                self[nodeid].attributes[attr].datachange_callbacks.pop(handle)
 
     def add_method_callback(self, methodid, callback):
         with self._lock:
-            node = self._nodes[methodid]
+            node = self[methodid]
             node.call = callback