Make sync database multiprocessing-safe.

4 files changed, 143 insertions, 43 deletions

diff --git a/client/src/leap/soledad/client/crypto.py b/client/src/leap/soledad/client/crypto.py
index 5ae5937f..eb5a4f64 100644
--- a/client/src/leap/soledad/client/crypto.py
+++ b/client/src/leap/soledad/client/crypto.py
@@ -224,7 +224,7 @@ class SoledadCrypto(object):
 
         The password is derived using HMAC having sha256 as underlying hash
         function. The key used for HMAC are the first
-        C{soledad.REMOTE_STORAGE_SECRET_KENGTH} bytes of Soledad's storage
+        C{soledad.REMOTE_STORAGE_SECRET_LENGTH} bytes of Soledad's storage
         secret stripped from the first MAC_KEY_LENGTH characters. The HMAC
         message is C{doc_id}.
 
@@ -623,9 +623,8 @@ class SyncEncrypterPool(SyncEncryptDecryptPool):
 
         con = self._sync_db
         with self._sync_db_write_lock:
-            with con:
-                con.execute(sql_del, (doc_id, ))
-                con.execute(sql_ins, (doc_id, doc_rev, content))
+            con.execute(sql_del, (doc_id, ))
+            con.execute(sql_ins, (doc_id, doc_rev, content))
 
 
 def decrypt_doc_task(doc_id, doc_rev, content, gen, trans_id, key, secret):
@@ -726,11 +725,10 @@ class SyncDecrypterPool(SyncEncryptDecryptPool):
 
         con = self._sync_db
         with self._sync_db_write_lock:
-            with con:
-                con.execute(sql_del, (doc_id, ))
-                con.execute(
-                    sql_ins,
-                    (doc_id, doc_rev, docstr, gen, trans_id, 1))
+            con.execute(sql_del, (doc_id, ))
+            con.execute(
+                sql_ins,
+                (doc_id, doc_rev, docstr, gen, trans_id, 1))
 
     def insert_received_doc(self, doc_id, doc_rev, content, gen, trans_id):
         """
@@ -757,11 +755,10 @@ class SyncDecrypterPool(SyncEncryptDecryptPool):
             self.TABLE_NAME,)
         con = self._sync_db
         with self._sync_db_write_lock:
-            with con:
-                con.execute(sql_del, (doc_id,))
-                con.execute(
-                    sql_ins,
-                    (doc_id, doc_rev, content, gen, trans_id, 0))
+            con.execute(sql_del, (doc_id,))
+            con.execute(
+                sql_ins,
+                (doc_id, doc_rev, content, gen, trans_id, 0))
 
     def delete_received_doc(self, doc_id, doc_rev):
         """
@@ -776,8 +773,7 @@ class SyncDecrypterPool(SyncEncryptDecryptPool):
             self.TABLE_NAME,)
         con = self._sync_db
         with self._sync_db_write_lock:
-            with con:
-                con.execute(sql_del, (doc_id, doc_rev))
+            con.execute(sql_del, (doc_id, doc_rev))
 
     def decrypt_doc(self, doc_id, rev, content, gen, trans_id,
                     source_replica_uid, workers=True):
@@ -878,12 +874,7 @@ class SyncDecrypterPool(SyncEncryptDecryptPool):
         if encrypted is not None:
             sql += " WHERE encrypted = %d" % int(encrypted)
         sql += " ORDER BY gen ASC"
-        c = self._sync_db.cursor()
-        c.execute(sql)
-        # TODO: due to unknown reasons, the fetchall() method may return empty
-        # values, so we filter them out here. We have to perform some tests to
-        # understand why and when this happens.
-        docs = filter(lambda entry: len(entry) > 0, c.fetchall())
+        docs = self._sync_db.select(sql)
         return docs
 
     def get_insertable_docs_by_gen(self):
@@ -894,7 +885,7 @@ class SyncDecrypterPool(SyncEncryptDecryptPool):
         decrypted_docs = self.get_docs_by_generation(encrypted=False)
         insertable = []
         for doc_id, rev, content, gen, trans_id, encrypted in all_docs:
-            next_decrypted = decrypted_docs.pop(0)
+            next_decrypted = decrypted_docs.next()
             if doc_id == next_decrypted[0]:
                 insertable.append((doc_id, rev, content, gen, trans_id))
             else:
@@ -915,14 +906,13 @@ class SyncDecrypterPool(SyncEncryptDecryptPool):
         if self._sync_db is None:
             logger.warning("cannot return count with null sync_db")
             return
-        c = self._sync_db.cursor()
         sql = "SELECT COUNT(*) FROM %s" % (self.TABLE_NAME,)
         if encrypted is not None:
             sql += " WHERE encrypted = %d" % int(encrypted)
-        c.execute(sql)
-        res = c.fetchone()
+        res = self._sync_db.select(sql)
         if res is not None:
-            return res[0]
+            val = res.next()
+            return val[0]
         else:
             return 0
 
@@ -932,8 +922,6 @@ class SyncDecrypterPool(SyncEncryptDecryptPool):
         decrypt worker to decrypt each one of them.
         """
         docs_by_generation = self.get_docs_by_generation(encrypted=True)
-        logger.debug("Sync decrypter pool: There are %d documents to " \
-                     "decrypt." % len(docs_by_generation))
         for doc_id, rev, content, gen, trans_id, _ \
                 in filter(None, docs_by_generation):
             self.decrypt_doc(
diff --git a/client/src/leap/soledad/client/mp_safe_db.py b/client/src/leap/soledad/client/mp_safe_db.py
new file mode 100644
index 00000000..a9ab5649
--- /dev/null
+++ b/client/src/leap/soledad/client/mp_safe_db.py
@@ -0,0 +1,101 @@
+# -*- coding: utf-8 -*-
+# crypto.py
+# Copyright (C) 2014 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+
+"""
+Multiprocessing-safe SQLite database.
+"""
+
+
+from threading import Thread
+from Queue import Queue
+from sqlite3 import connect as sqlite3_connect
+
+
+# Thanks to http://code.activestate.com/recipes/526618/
+
+class MPSafeSQLiteDB(Thread):
+    """
+    A multiprocessing-safe SQLite database accessor.
+    """
+
+    CLOSE = "--close--"
+    NO_MORE = "--no more--"
+
+    def __init__(self, db_path):
+        """
+        Initialize the process
+        """
+        Thread.__init__(self)
+        self._db_path = db_path
+        self._requests = Queue()
+        self.start()
+
+    def run(self):
+        """
+        Run the multiprocessing-safe database accessor.
+        """
+        conn = sqlite3_connect(self._db_path)
+        while True:
+            req, arg, res = self._requests.get()
+            if req == self.CLOSE:
+                break
+            with conn:
+                cursor = conn.cursor()
+                cursor.execute(req, arg)
+                if res:
+                    for rec in cursor.fetchall():
+                        res.put(rec)
+                    res.put(self.NO_MORE)
+        conn.close()
+
+    def execute(self, req, arg=None, res=None):
+        """
+        Execute a request on the database.
+
+        :param req: The request to be executed.
+        :type req: str
+        :param arg: The arguments for the request.
+        :type arg: tuple
+        :param res: A queue to write request results.
+        :type res: multiprocessing.Queue
+        """
+        self._requests.put((req, arg or tuple(), res))
+
+    def select(self, req, arg=None):
+        """
+        Run a select query on the database and yield results.
+
+        :param req: The request to be executed.
+        :type req: str
+        :param arg: The arguments for the request.
+        :type arg: tuple
+        """
+        res = Queue()
+        self.execute(req, arg, res)
+        while True:
+            rec=res.get()
+            if rec == self.NO_MORE:
+                break
+            yield rec
+
+    def close(self):
+        """
+        Close the database connection.
+        """
+        self.execute(self.CLOSE)
+        self.join()
diff --git a/client/src/leap/soledad/client/sqlcipher.py b/client/src/leap/soledad/client/sqlcipher.py
index 5a30b125..85b0391b 100644
--- a/client/src/leap/soledad/client/sqlcipher.py
+++ b/client/src/leap/soledad/client/sqlcipher.py
@@ -44,7 +44,6 @@ handled by Soledad should be created by SQLCipher >= 2.0.
 import logging
 import multiprocessing
 import os
-import sqlite3
 import string
 import threading
 import time
@@ -63,6 +62,7 @@ from leap.soledad.client.crypto import SyncEncrypterPool, SyncDecrypterPool
 from leap.soledad.client.target import SoledadSyncTarget
 from leap.soledad.client.target import PendingReceivedDocsSyncError
 from leap.soledad.client.sync import SoledadSynchronizer
+from leap.soledad.client.mp_safe_db import MPSafeSQLiteDB
 from leap.soledad.common.document import SoledadDocument
 
 
@@ -549,8 +549,7 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
             sync_db_path = "%s-sync" % sqlcipher_file
         else:
             sync_db_path = ":memory:"
-        self._sync_db = sqlite3.connect(sync_db_path,
-                                        check_same_thread=False)
+        self._sync_db = MPSafeSQLiteDB(sync_db_path)
         self._sync_db_write_lock = threading.Lock()
         self._create_sync_db_tables()
         self.sync_queue = multiprocessing.Queue()
@@ -567,9 +566,8 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
             decr.TABLE_NAME, decr.FIELD_NAMES))
 
         with self._sync_db_write_lock:
-            with self._sync_db:
-                self._sync_db.execute(sql_encr)
-                self._sync_db.execute(sql_decr)
+            self._sync_db.execute(sql_encr)
+            self._sync_db.execute(sql_decr)
 
     #
     # Symmetric encryption of syncing docs
@@ -1076,16 +1074,28 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
         Close db_handle and close syncer.
         """
         logger.debug("Sqlcipher backend: closing")
+        # stop the sync watcher for deferred encryption
         if self._sync_watcher is not None:
             self._sync_watcher.stop()
             self._sync_watcher.shutdown()
+        # close all open syncers
         for url in self._syncers:
             _, syncer = self._syncers[url]
             syncer.close()
+        # stop the encryption pool
         if self._sync_enc_pool is not None:
             self._sync_enc_pool.close()
+        # close the actual database
         if self._db_handle is not None:
             self._db_handle.close()
+        # close the sync database
+        if self._sync_db is not None:
+            self._sync_db.close()
+        # close the sync queue
+        if self.sync_queue is not None:
+            self.sync_queue.close()
+            del self.sync_queue
+            self.sync_queue = None
 
     @property
     def replica_uid(self):
diff --git a/client/src/leap/soledad/client/target.py b/client/src/leap/soledad/client/target.py
index 5fe55216..01e1231a 100644
--- a/client/src/leap/soledad/client/target.py
+++ b/client/src/leap/soledad/client/target.py
@@ -1346,13 +1346,16 @@ class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
         :type doc_rev: str
         """
         encr = SyncEncrypterPool
-        c = self._sync_db.cursor()
         sql = ("SELECT content FROM %s WHERE doc_id=? and rev=?" % (
             encr.TABLE_NAME,))
-        c.execute(sql, (doc_id, doc_rev))
-        res = c.fetchall()
-        if len(res) != 0:
-            return res[0][0]
+        res = self._sync_db.select(sql, (doc_id, doc_rev))
+        try:
+            val = res.next()
+            return val[0]
+        except StopIteration:
+            # no doc found
+            return None
+
 
     def delete_encrypted_docs_from_db(self, docs_ids):
         """
@@ -1365,12 +1368,10 @@ class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
         """
         if docs_ids:
             encr = SyncEncrypterPool
-            c = self._sync_db.cursor()
             for doc_id, doc_rev in docs_ids:
                 sql = ("DELETE FROM %s WHERE doc_id=? and rev=?" % (
                     encr.TABLE_NAME,))
-                c.execute(sql, (doc_id, doc_rev))
-            self._sync_db.commit()
+                self._sync_db.execute(sql, (doc_id, doc_rev))
 
     def _save_encrypted_received_doc(self, doc, gen, trans_id, idx, total):
         """