[qtcontacts-sqlite] Use a semaphore for process mutex on writes

Although SQLite supports multiple concurrent processes, the behaviour
on write contention is undesirable, resulting in starvation.

Use a semaphore to provide mutual exclusion between writing processes
so that only one has write access to the database at any given time.

src/engine/contactwriter.cpp

@@ -34,6 +34,7 @@
 #include "contactsengine.h"
 #include "contactreader.h"
 #include "contactnotifier.h"
+#include "semaphore_p.h"
 
 #include <QContactFavorite>
 #include <QContactGender>
@@ -392,9 +393,41 @@ static bool detailsEquivalent(const QContactDetail &lhs, const QContactDetail &r
     return (lhsValues == rhsValues);
 }
 
+// Adapted from the inter-process mutex in QMF
+// The first user creates the semaphore that all subsequent instances
+// attach to.  We rely on undo semantics to release locked semaphores
+// on process failure.
+class ProcessMutex
+{
+    Semaphore m_semaphore;
+
+public:
+    ProcessMutex(const QString &path)
+        : m_semaphore(path.toLatin1(), 1)
+    {
+    }
+
+    bool lock()
+    {
+        return m_semaphore.decrement();
+    }
+
+    bool unlock()
+    {
+        return m_semaphore.increment();
+    }
+
+    bool isLocked() const
+    {
+        return (m_semaphore.value() == 0);
+    }
+};
+
+
 ContactWriter::ContactWriter(const ContactsEngine &engine, const QSqlDatabase &database, ContactReader *reader)
     : m_engine(engine)
     , m_database(database)
+    , m_databaseMutex(new ProcessMutex(database.databaseName()))
     , m_findRelatedForAggregate(prepare(findRelatedForAggregate, database))
     , m_findLocalForAggregate(prepare(findLocalForAggregate, database))
     , m_findAggregateForContact(prepare(findAggregateForContact, database))
@@ -459,7 +492,18 @@ ContactWriter::~ContactWriter()
 
 bool ContactWriter::beginTransaction()
 {
-    return m_database.transaction();
+    // We use a cross-process mutex to ensure only one process can
+    // write to the DB at once.  Without locking, sqlite will back off
+    // on write contention, and the backed-off process may never get access
+    // if other processes are performing regular writes.
+    if (m_databaseMutex->lock()) {
+        if (m_database.transaction())
+            return true;
+
+        m_databaseMutex->unlock();
+    }
+
+    return false;
 }
 
 bool ContactWriter::commitTransaction()
@@ -470,6 +514,12 @@ bool ContactWriter::commitTransaction()
         return false;
     }
 
+    if (m_databaseMutex->isLocked()) {
+        m_databaseMutex->unlock();
+    } else {
+        qWarning() << "Lock error: no lock held on commit";
+    }
+
     if (!m_removedIds.isEmpty()) {
         ContactNotifier::contactsRemoved(m_removedIds);
         m_removedIds.clear();
@@ -488,6 +538,11 @@ bool ContactWriter::commitTransaction()
 void ContactWriter::rollbackTransaction()
 {
     m_database.rollback();
+    if (m_databaseMutex->isLocked()) {
+        m_databaseMutex->unlock();
+    } else {
+        qWarning() << "Lock error: no lock held on rollback";
+    }
 
     m_removedIds.clear();
     m_changedIds.clear();

src/engine/contactwriter.h

@@ -57,6 +57,7 @@
 
 QTM_USE_NAMESPACE
 
+class ProcessMutex;
 class ContactsEngine;
 class ContactReader;
 class ContactWriter
@@ -132,6 +133,7 @@ class ContactWriter
 
     const ContactsEngine &m_engine;
     QSqlDatabase m_database;
+    ProcessMutex *m_databaseMutex;
     QSqlQuery m_findRelatedForAggregate;
     QSqlQuery m_findLocalForAggregate;
     QSqlQuery m_findAggregateForContact;

src/engine/engine.pro

@@ -14,13 +14,15 @@ DEFINES += 'QTCONTACTS_SQLITE_DATABASE_DIR=\'\"/home/nemo/.local/share/data/qtco
 DEFINES += 'QTCONTACTS_SQLITE_DATABASE_NAME=\'\"contacts.db\"\''
 
 HEADERS += \
+        semaphore_p.h \
         contactsdatabase.h \
         contactsengine.h \
         contactnotifier.h \
         contactreader.h \
         contactwriter.h
 
 SOURCES += \
+        semaphore_p.cpp \
         contactsdatabase.cpp \
         contactsengine.cpp \
         contactsplugin.cpp \

src/engine/semaphore_p.cpp

@@ -0,0 +1,146 @@
+/*
+ * Copyright (C) 2013 Jolla Ltd. <matthew.vogt@jollamobile.com>
+ *
+ * You may use this file under the terms of the BSD license as follows:
+ *
+ * "Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *   * Redistributions of source code must retain the above copyright
+ *     notice, this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in
+ *     the documentation and/or other materials provided with the
+ *     distribution.
+ *   * Neither the name of Nemo Mobile nor the names of its contributors
+ *     may be used to endorse or promote products derived from this
+ *     software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE."
+ */
+
+#include "semaphore_p.h"
+
+#include <errno.h>
+#include <unistd.h>
+
+#include <sys/sem.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <sys/ipc.h>
+
+#include <QDebug>
+
+namespace {
+
+// Defined as required for ::semun
+union semun {
+    int              val;
+    struct semid_ds *buf;
+    unsigned short  *array;
+    struct seminfo  *__buf;
+};
+
+}
+
+Semaphore::Semaphore(const char *id, int initial)
+    : m_identifier(id)
+    , m_initialValue(-1)
+    , m_id(-1)
+{
+    key_t key = ::ftok(m_identifier, 0);
+
+    m_id = ::semget(key, 1, 0);
+    if (m_id == -1) {
+        if (errno != ENOENT) {
+            error("Unable to get semaphore", errno);
+        } else {
+            // The semaphore does not currently exist
+            m_id = ::semget(key, 1, IPC_CREAT | IPC_EXCL | S_IRWXU);
+            if (m_id == -1) {
+                if (errno == EEXIST) {
+                    // Someone else won the race to create the semaphore - retry get 
+                    m_id = ::semget(key, 1, 0);
+                }
+
+                if (m_id == -1) {
+                    error("Unable to create semaphore", errno);
+                }
+            } else {
+                // Set the initial value
+                union semun arg = { 0 };
+                arg.val = initial;
+
+                int status = ::semctl(m_id, 0, SETVAL, arg);
+                if (status == -1) {
+                    m_id = -1;
+                    error("Unable to initialize semaphore", errno);
+                } else {
+                    m_initialValue = initial;
+                }
+            }
+        }
+    }
+}
+
+Semaphore::~Semaphore()
+{
+}
+
+bool Semaphore::decrement()
+{
+    if (m_id == -1)
+        return false;
+
+    struct sembuf op;
+    op.sem_num = 0;
+    op.sem_op = -1;
+    op.sem_flg = SEM_UNDO;
+
+    if (::semop(m_id, &op, 1) == 0)
+        return true;
+
+    error("Unable to decrement semaphore", errno);
+    return false;
+}
+
+bool Semaphore::increment()
+{
+    if (m_id == -1)
+        return false;
+
+    struct sembuf op;
+    op.sem_num = 0;
+    op.sem_op = 1;
+    op.sem_flg = SEM_UNDO;
+
+    if (::semop(m_id, &op, 1) == 0)
+        return true;
+
+    error("Unable to increment semaphore", errno);
+    return false;
+}
+
+int Semaphore::value() const
+{
+    if (m_id == -1)
+        return -1;
+
+    return ::semctl(m_id, 0, GETVAL, 0);
+}
+
+void Semaphore::error(const char *msg, int error)
+{
+    qWarning() << QString("%1 %2: %3 (%4)").arg(msg).arg(m_identifier).arg(::strerror(error)).arg(error);
+}
+

src/engine/semaphore_p.h

@@ -0,0 +1,56 @@
+/*
+ * Copyright (C) 2013 Jolla Ltd. <matthew.vogt@jollamobile.com>
+ *
+ * You may use this file under the terms of the BSD license as follows:
+ *
+ * "Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *   * Redistributions of source code must retain the above copyright
+ *     notice, this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in
+ *     the documentation and/or other materials provided with the
+ *     distribution.
+ *   * Neither the name of Nemo Mobile nor the names of its contributors
+ *     may be used to endorse or promote products derived from this
+ *     software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE."
+ */
+
+#ifndef QTCONTACTSSQLITE_SEMAPHORE_P
+#define QTCONTACTSSQLITE_SEMAPHORE_P
+
+#include <QString>
+
+class Semaphore
+{
+public:
+    Semaphore(const char *identifier, int initial);
+    ~Semaphore();
+
+    bool decrement();
+    bool increment();
+
+    int value() const;
+
+private:
+    void error(const char *msg, int error);
+
+    const char *m_identifier;
+    int m_initialValue;
+    int m_id;
+};
+
+#endif