Merge pull request #763 from Expensify/master

Update expensify_prod branch

.travis.yml

@@ -14,3 +14,14 @@ notifications:
 before_install:
 script: ./travis.sh
 dist: xenial
+deploy:
+    provider: releases
+    # `api_key` is an encrypted oauth token for rafecolton. See https://docs.travis-ci.com/user/deployment/releases
+    api_key:
+        secure: jW0XqsVQ7TH3qtgPoDIV6YbcH0RITL/oAnCNXhUHnlr2f4vstIN7Izpef4MTrGFf6QszlEYmSVKGtpNbpiH+KpVqc8UTW1oGjh5b3FHShGvLggwe1GtApVyN21tW9jsScTNjs3nu2kXjPSz9UfLSNi/seRhSH8+TqStG3B31ItyJQmhs4YOX+uCKQLHxCjQMeAUz7BHhvWWfmd80gmjioX9D8hXfKIPTVDgBRlC0asnSXuILicM6TSFDDhtCIvKHnH7MIF/M+0J76OYPIeSK6KkYjCDMO1Ik9E6V7eDWCFIDq5zKarOuMktOKZ9KzMEDWKK+3zxVg4VjMyHTPDJeKaDzRRcyx2Dlk1fSX3gRJfcrbTeg3lrBwdlLbYMNo3Y2sJYhIp0sn4ZTDTSmrD25qVIg3pRaXj1PSZju5PU8SZN8b1xFyPFJJPhADSuD2gQTU88hBFA12d1/2UElStDt3UnyQlTZSmm8DeJ3HtDuQ3W2e4E64fzeXDW85SxBx3ecdSL5J0gBhQIcoHg6FuxTmXOOB71nTYmJb8OVZzVlFYiWIkIm0FlPTY8QGqUXhbScp4D/mLqTWlnllLsTcVKCe8lwzRBpFMkfbLWQy4EAvqjpAAmcWxHfTH39ORrO/J5ipZX+LQAAPwF8qDoedBYsZhUf4Duj1c/U++PNbF3mDM4=
+    file:
+        - bedrock
+    skip_cleanup: true
+    prerelease: true
+    on:
+        tags: true

BedrockServer.cpp

@@ -163,7 +163,7 @@ void BedrockServer::sync(const SData& args,
 
     // Initialize the DB.
     int64_t mmapSizeGB = args.isSet("-mmapSizeGB") ? stoll(args["-mmapSizeGB"]) : 0;
-    SQLite db(args["-db"], args.calc("-cacheSize"), true, args.calc("-maxJournalSize"), -1, workerThreads - 1, args["-synchronous"], mmapSizeGB);
+    SQLite db(args["-db"], args.calc("-cacheSize"), true, args.calc("-maxJournalSize"), -1, workerThreads - 1, args["-synchronous"], mmapSizeGB, args.test("-pageLogging"));
 
     // And the command processor.
     BedrockCore core(db, server);
@@ -703,7 +703,7 @@ void BedrockServer::worker(const SData& args,
     // Worker 0 is the "blockingCommit" thread.
     SInitialize(threadId ? "worker" + to_string(threadId) : "blockingCommit");
     int64_t mmapSizeGB = args.isSet("-mmapSizeGB") ? stoll(args["-mmapSizeGB"]) : 0;
-    SQLite db(args["-db"], args.calc("-cacheSize"), false, args.calc("-maxJournalSize"), threadId, threadCount - 1, args["-synchronous"], mmapSizeGB);
+    SQLite db(args["-db"], args.calc("-cacheSize"), false, args.calc("-maxJournalSize"), threadId, threadCount - 1, args["-synchronous"], mmapSizeGB, args.test("-pageLogging"));
     BedrockCore core(db, server);
 
     // Command to work on. This default command is replaced when we find work to do.

sqlitecluster/SQLite.cpp

@@ -6,6 +6,9 @@
 // Globally shared mutex for locking around commits and creating/destroying instances.
 recursive_mutex SQLite::_commitLock;
 
+atomic<int64_t> SQLite::_transactionAttemptCount(0);
+mutex SQLite::_pageLogMutex;
+
 // Global map for looking up shared data by file when creating new instances.
 map<string, SQLite::SharedData*> SQLite::_sharedDataLookupMap;
 
@@ -19,7 +22,7 @@ atomic<int> SQLite::fullCheckpointPageMin(25000); // Approx 100mb (pages are ass
 atomic<bool> SQLite::enableTrace(false);
 
 SQLite::SQLite(const string& filename, int cacheSize, bool enableFullCheckpoints, int maxJournalSize, int journalTable,
-               int maxRequiredJournalTableID, const string& synchronous, int64_t mmapSizeGB) :
+               int maxRequiredJournalTableID, const string& synchronous, int64_t mmapSizeGB, bool pageLoggingEnabled) :
     whitelist(nullptr),
     _maxJournalSize(maxJournalSize),
     _insideTransaction(false),
@@ -39,7 +42,9 @@ SQLite::SQLite(const string& filename, int cacheSize, bool enableFullCheckpoints
     _queryCount(0),
     _cacheHits(0),
     _useCache(false),
-    _isDeterministicQuery(false)
+    _isDeterministicQuery(false),
+    _pageLoggingEnabled(pageLoggingEnabled),
+    _currentTransactionAttemptCount(-1)
 {
     // Perform sanity checks.
     SASSERT(!filename.empty());
@@ -114,6 +119,11 @@ SQLite::SQLite(const string& filename, int cacheSize, bool enableFullCheckpoints
     const int DB_WRITE_OPEN_FLAGS = SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_NOMUTEX;
     SASSERT(!sqlite3_open_v2(filename.c_str(), &_db, DB_WRITE_OPEN_FLAGS, NULL));
 
+    // Turn on page logging if specified.
+    if (_pageLoggingEnabled) {
+        sqlite3_begin_concurrent_report_enable(_db, 1);
+    }
+
     // WAL is what allows simultaneous read/writing.
     SASSERT(!SQuery(_db, "enabling write ahead logging", "PRAGMA journal_mode = WAL;"));
 
@@ -446,6 +456,7 @@ bool SQLite::beginConcurrentTransaction(bool useCache, const string& transaction
 
     SDEBUG("[concurrent] Beginning transaction");
     uint64_t before = STimeNow();
+    _currentTransactionAttemptCount = -1;
     _insideTransaction = !SQuery(_db, "starting db transaction", "BEGIN CONCURRENT");
     _queryCache.clear();
     _transactionName = transactionName;
@@ -750,7 +761,15 @@ int SQLite::commit() {
 
     uint64_t before = STimeNow();
     uint64_t beforeCommit = STimeNow();
-    result = SQuery(_db, "committing db transaction", "COMMIT");
+    if (_pageLoggingEnabled) {
+        {
+            lock_guard<mutex> lock(_pageLogMutex);
+            _currentTransactionAttemptCount = _transactionAttemptCount.fetch_add(1);
+            result = SQuery(_db, "committing db transaction", "COMMIT");
+        }
+    } else {
+        result = SQuery(_db, "committing db transaction", "COMMIT");
+    }
     SINFO("SQuery 'COMMIT' took " << ((STimeNow() - beforeCommit)/1000) << "ms.");
 
     // And record pages after the commit.
@@ -769,6 +788,12 @@ int SQLite::commit() {
     // If there were conflicting commits, will return SQLITE_BUSY_SNAPSHOT
     SASSERT(result == SQLITE_OK || result == SQLITE_BUSY_SNAPSHOT);
     if (result == SQLITE_OK) {
+        if (_currentTransactionAttemptCount != -1) {
+            string logLine = SWHEREAMI + "[row-level-locking] transaction attempt:" +
+                             to_string(_currentTransactionAttemptCount) + " committed. report: " +
+                             sqlite3_begin_concurrent_report(_db);
+            syslog(LOG_DEBUG, "%s", logLine.c_str());
+        }
         _commitElapsed += STimeNow() - before;
         _journalSize = newJournalSize;
         _sharedData->_commitCount++;
@@ -793,7 +818,12 @@ int SQLite::commit() {
         _queryCount = 0;
         _cacheHits = 0;
     } else {
-        SINFO("Commit failed, waiting for rollback.");
+        if (_currentTransactionAttemptCount != -1) {
+            string logLine = SWHEREAMI  + "[row-level-locking] transaction attempt:" +
+                             to_string(_currentTransactionAttemptCount) + " conflict, will roll back.";
+            syslog(LOG_DEBUG, "%s", logLine.c_str());
+            SINFO("Commit failed, waiting for rollback.");
+        }
     }
 
     // if we got SQLITE_BUSY_SNAPSHOT, then we're *still* holding commitLock, and it will need to be unlocked by
@@ -841,6 +871,13 @@ void SQLite::rollback() {
             _rollbackElapsed += STimeNow() - before;
         }
 
+        if (_currentTransactionAttemptCount != -1) {
+            string logLine = SWHEREAMI + "[row-level-locking] transaction attempt:" +
+                             to_string(_currentTransactionAttemptCount) + " rolled back. report: " +
+                             sqlite3_begin_concurrent_report(_db);
+            syslog(LOG_DEBUG, "%s", logLine.c_str());
+        }
+
         // Finally done with this.
         _insideTransaction = false;
         _uncommittedHash.clear();

sqlitecluster/SQLite.h

@@ -45,7 +45,7 @@ class SQLite {
     //
     // mmapSizeGB: address space to use for memory-mapped IO, in GB.
     SQLite(const string& filename, int cacheSize, bool enableFullCheckpoints, int maxJournalSize, int journalTable,
-           int maxRequiredJournalTableID, const string& synchronous = "", int64_t mmapSizeGB = 0);
+           int maxRequiredJournalTableID, const string& synchronous = "", int64_t mmapSizeGB = 0, bool pageLoggingEnabled = false);
     ~SQLite();
 
     // Returns the canonicalized filename for this database
@@ -435,4 +435,9 @@ class SQLite {
 
     // Will be set to false while running a non-deterministic query to prevent it's result being cached.
     bool _isDeterministicQuery;
+
+    bool _pageLoggingEnabled;
+    static atomic<int64_t> _transactionAttemptCount;
+    static mutex _pageLogMutex;
+    int64_t _currentTransactionAttemptCount;
 };