Add full support for protocol.LAST_COMMIT_INFO (#45)

Previously the LAST_COMMIT_INFO feature was only supported at the
protocol level: the driver always reported there was no information
available about the last commit.

Last commit ensures "read your own writes" capability for committed
transactions, even when connections are using different TEs.

Implement an internal structure to store last commit information:
- The information is saved as a process-wide set of data
- Allows the same process to connect to different databases, and to
  connect to the same database multiple times (to the same or to
  different TEs).

In order to support Python clients that may use threading with different
connections, use a Lock() to access the last commit details.  Python
threading doesn't provide reader/writer locks; rather than write one
we'll use a simple mutex and assume this won't be a performance
bottleneck.

Author: madscientist

pynuodb/encodedsession.py

@@ -15,10 +15,11 @@
 import struct
 import decimal
 import sys
+import threading
 
 try:
     from typing import Any, Collection, Dict, List  # pylint: disable=unused-import
-    from typing import Mapping, Optional  # pylint: disable=unused-import
+    from typing import Mapping, Optional, Tuple  # pylint: disable=unused-import
 except ImportError:
     pass
 
@@ -103,6 +104,23 @@ class EncodedSession(session.Session):  # pylint: disable=too-many-public-method
     __lastNodeId = 0
     __lastCommitSeq = 0
 
+    # Manage the last commit info
+
+    class DbInfo(object):
+        """Last commit information for a given database."""
+        size = None  # type: int
+        info = None  # type: Dict[int, Tuple[int, int]]
+
+        def __init__(self, size):
+            # type: (int) -> None
+            self.size = size
+            self.info = {}
+
+    # If we decide this lock is causing performance issues we can implement
+    # a reader/writer lock.
+    __dblock = threading.Lock()
+    __databases = None  # type: Dict[str, DbInfo]
+
     @property
     def db_uuid(self):
         # type: () -> Optional[uuid.UUID]
@@ -143,7 +161,7 @@ def __init__(self, host, service='SQL2', options=None, **kwargs):
         super(EncodedSession, self).__init__(host, service=service,
                                              options=options, **kwargs)
 
-    def open_database(self, db_name, password, parameters):  # pylint: disable=too-many-branches
+    def open_database(self, db_name, password, parameters):  # pylint: disable=too-many-branches,too-many-statements
         # type: (str, str, Dict[str, str]) -> None
         """Perform a handshake as a SQL client with a NuoDB TE.
 
@@ -210,6 +228,13 @@ def open_database(self, db_name, password, parameters):  # pylint: disable=too-m
             self.__connectedNodeID = self.getInt()
             self.__maxNodes = self.getInt()
 
+            dbid = str(self.db_uuid)
+            with EncodedSession.__dblock:
+                if EncodedSession.__databases is None:
+                    EncodedSession.__databases = {}
+                if dbid not in EncodedSession.__databases:
+                    EncodedSession.__databases[dbid] = EncodedSession.DbInfo(self.__maxNodes)
+
         self.__sessionVersion = protocolVersion
 
         if not self.tls_encrypted:
@@ -257,6 +282,14 @@ def send_close(self):
         self._exchangeMessages()
         self.close()
 
+    def __set_dbinfo(self, sid, txid, seqid):
+        # type: (int, int, int) -> None
+        with EncodedSession.__dblock:
+            info = EncodedSession.__databases[str(self.db_uuid)].info
+            lci = info.get(sid, (-1, -1))
+            if seqid > lci[1]:
+                info[sid] = (txid, seqid)
+
     def send_commit(self):
         # type: () -> int
         """Commit an open transaction on this connection.
@@ -268,6 +301,7 @@ def send_commit(self):
         self.__lastTxnId = self.getInt()
         self.__lastNodeId = self.getInt()
         self.__lastCommitSeq = self.getInt()
+        self.__set_dbinfo(self.__lastNodeId, self.__lastTxnId, self.__lastCommitSeq)
         return self.__lastTxnId
 
     def send_rollback(self):
@@ -309,6 +343,13 @@ def create_statement(self):
         self._exchangeMessages()
         return statement.Statement(self.getInt())
 
+    def __execute_postfix(self):
+        # type: () -> None
+        txid = self.getInt()
+        sid = self.getInt()
+        seqid = self.getInt()
+        self.__set_dbinfo(sid, txid, seqid)
+
     def execute_statement(self, stmt, query):
         # type: (statement.Statement, str) -> statement.ExecutionResult
         """Execute a query using the given statement.
@@ -322,6 +363,7 @@ def execute_statement(self, stmt, query):
 
         result = self.getInt()
         rowcount = self.getInt()
+        self.__execute_postfix()
 
         return statement.ExecutionResult(stmt, result, rowcount)
 
@@ -370,6 +412,7 @@ def execute_prepared_statement(
 
         result = self.getInt()
         rowcount = self.getInt()
+        self.__execute_postfix()
 
         return statement.ExecutionResult(prepared_statement, result, rowcount)
 
@@ -408,6 +451,8 @@ def execute_batch_prepared_statement(self, prepared_statement, param_lists):
         if error_string is not None:
             raise BatchError(error_string, results)
 
+        self.__execute_postfix()
+
         return results
 
     def fetch_result_set(self, stmt):
@@ -1127,7 +1172,13 @@ def _setup_statement(self, handle, msgId):
         """
         self._putMessageId(msgId)
         if self.__sessionVersion >= protocol.LAST_COMMIT_INFO:
-            self.putInt(self.getCommitInfo(self.__connectedNodeID))
+            with EncodedSession.__dblock:
+                dbinfo = EncodedSession.__databases[str(self.db_uuid)]
+                self.putInt(len(dbinfo.info))
+                for sid, tup in dbinfo.info.items():
+                    self.putInt(sid)
+                    self.putInt(tup[0])
+                    self.putInt(tup[1])
         self.putInt(handle)
 
         return self
@@ -1164,8 +1215,3 @@ def _takeBytes(self, length):
             return self.__input[self.__inpos:self.__inpos + length]
         finally:
             self.__inpos += length
-
-    def getCommitInfo(self, nodeID):  # pylint: disable=no-self-use
-        # type: (int) -> int
-        """Return the last commit info.  Does not support last commit."""
-        return 0 * nodeID