Merge branch 'master' into tyler-cleanup-libstuff

Author: tylerkaraszewski

.github/PULL_REQUEST_TEMPLATE.md

@@ -0,0 +1,9 @@
+### Details
+
+### Fixed Issues
+Fixes GH_LINK
+
+### Tests
+
+_________
+**Internal Testing Reminder:** when changing bedrock, please compile auth against your new changes

Makefile

@@ -60,7 +60,7 @@ clean:
 mbedtls/library/libmbedcrypto.a mbedtls/library/libmbedtls.a mbedtls/library/libmbedx509.a:
 	git submodule init
 	git submodule update
-	cd mbedtls && git checkout -q 04a049bda1ceca48060b57bc4bcf5203ce591421
+	cd mbedtls && git checkout -q v2.26.0
 	cd mbedtls && $(MAKE) no_test
 
 # We select all of the cpp files (and manually add sqlite3.c) that will be in libstuff.

libstuff/SHTTPSManager.cpp

@@ -225,6 +225,7 @@ SStandaloneHTTPSManager::Transaction* SStandaloneHTTPSManager::_httpsSend(const
     Socket* s = openSocket(host, x509);
     if (!s) {
         delete transaction;
+        delete x509;
         return _createErrorTransaction();
     }
 

libstuff/SRandom.cpp

@@ -1,6 +1,12 @@
 #include "SRandom.h"
 
+#ifdef VALGRIND
+// random_device breaks valgrind.
+mt19937_64 SRandom::_generator = mt19937_64();
+#else
 mt19937_64 SRandom::_generator = mt19937_64(random_device()());
+#endif
+
 uniform_int_distribution<uint64_t> SRandom::_distribution64 = uniform_int_distribution<uint64_t>();
 
 uint64_t SRandom::limitedRand64(uint64_t minNum, uint64_t maxNum) {

libstuff/SSignal.cpp

@@ -21,6 +21,9 @@ void _SSignal_StackTrace(int signum, siginfo_t *info, void *ucontext);
 // A boolean indicating whether or not we've initialized our signal thread.
 atomic_flag _SSignal_threadInitialized = ATOMIC_FLAG_INIT;
 
+// Set to true to stop the signal thread.
+atomic<bool> _SSignal_threadStopFlag(false);
+
 // The signals we've received since the last time this was cleared.
 atomic<uint64_t> _SSignal_pendingSignalBitMask(0);
 
@@ -85,7 +88,7 @@ void SInitializeSignals() {
     sigprocmask(SIG_BLOCK, &signals, 0);
 
     // This is the signal action structure we'll use to specify what to listen for.
-    struct sigaction newAction;
+    struct sigaction newAction = {0};
 
     // The old style handler is explicitly null
     newAction.sa_handler = nullptr;
@@ -109,7 +112,6 @@ void SInitializeSignals() {
     bool threadAlreadyStarted = _SSignal_threadInitialized.test_and_set();
     if (!threadAlreadyStarted) {
         _SSignal_signalThread = thread(_SSignal_signalHandlerThreadFunc);
-        _SSignal_signalThread.detach();
     }
 }
 
@@ -124,10 +126,24 @@ void _SSignal_signalHandlerThreadFunc() {
 
     // Now we wait for any signal to occur.
     while (true) {
+
         // Wait for a signal to appear.
-        int signum = 0;
-        int result = sigwait(&signals, &signum);
-        if (!result) {
+        siginfo_t siginfo = {0};
+        struct timespec timeout;
+        timeout.tv_sec = 1;
+        timeout.tv_nsec = 0;
+        int result = -1;
+        while (result == -1) {
+            result = sigtimedwait(&signals, &siginfo, &timeout);
+            if (_SSignal_threadStopFlag) {
+                // Done.
+                SINFO("Stopping signal handler thread.");
+                return;
+            }
+        }
+        int signum = siginfo.si_signo;
+
+        if (result > 0) {
             // Do the same handling for these functions here as any other thread.
             if (signum == SIGSEGV || signum == SIGABRT || signum == SIGFPE || signum == SIGILL || signum == SIGBUS) {
                 _SSignal_StackTrace(signum, nullptr, nullptr);
@@ -140,6 +156,16 @@ void _SSignal_signalHandlerThreadFunc() {
     }
 }
 
+void SStopSignalThread() {
+    _SSignal_threadStopFlag = true;
+    if (_SSignal_threadInitialized.test_and_set()) {
+        // Send ourselves a singnal to interrupt our thread.
+        SINFO("Joining signal thread.");
+        _SSignal_signalThread.join();
+        _SSignal_threadInitialized.clear();
+    }
+}
+
 void _SSignal_StackTrace(int signum, siginfo_t *info, void *ucontext) {
     if (signum == SIGSEGV || signum == SIGABRT || signum == SIGFPE || signum == SIGILL || signum == SIGBUS) {
         // If we haven't already saved a signal number, we'll do it now. Any signal we catch here will generate a

libstuff/STCPNode.cpp

@@ -49,21 +49,16 @@ const string& STCPNode::stateName(STCPNode::State state) {
 }
 
 STCPNode::State STCPNode::stateFromName(const string& name) {
-    string normalizedName = SToUpper(name);
-
-    // Accept both old and new state names, but map them all to the new states.
-    // We can delete the old names once the entire cluster is using the new names.
+    const string normalizedName = SToUpper(name);
     static map<string, State> lookup = {
         {"SEARCHING", SEARCHING},
         {"SYNCHRONIZING", SYNCHRONIZING},
         {"WAITING", WAITING},
         {"STANDINGUP", STANDINGUP},
         {"LEADING", LEADING},
-        {"MASTERING", LEADING},
         {"STANDINGDOWN", STANDINGDOWN},
         {"SUBSCRIBING", SUBSCRIBING},
         {"FOLLOWING", FOLLOWING},
-        {"SLAVING", FOLLOWING},
     };
     auto it = lookup.find(normalizedName);
     if (it == lookup.end()) {

libstuff/libstuff.h

@@ -206,6 +206,8 @@ string SGetSignalDescription();
 // Clear all outstanding signals.
 void SClearSignals();
 
+void SStopSignalThread();
+
 // And also exception stuff.
 string SGetCurrentExceptionName();
 void STerminateHandler(void);

main.cpp

@@ -197,11 +197,6 @@ int main(int argc, char* argv[]) {
         }
     }
 
-    // Start libstuff. Generally, we want to initialize libstuff immediately on any new thread, but we wait until after
-    // the `fork` above has completed, as we can get strange behaviors from signal handlers across forked processes.
-    SInitialize("main", (args.isSet("-overrideProcessName") ? args["-overrideProcessName"].c_str() : 0));
-    SLogLevel(LOG_INFO);
-
     if (args.isSet("-version")) {
         // Just output the version
         cout << VERSION << endl;
@@ -277,6 +272,12 @@ int main(int argc, char* argv[]) {
         cout << endl;
         return 1;
     }
+
+    // Start libstuff. Generally, we want to initialize libstuff immediately on any new thread, but we wait until after
+    // the `fork` above has completed, as we can get strange behaviors from signal handlers across forked processes.
+    SInitialize("main", (args.isSet("-overrideProcessName") ? args["-overrideProcessName"].c_str() : 0));
+    SLogLevel(LOG_INFO);
+
     if (args.isSet("-v")) {
         // Verbose logging
         SINFO("Enabling verbose logging");
@@ -395,6 +396,9 @@ int main(int argc, char* argv[]) {
     delete _server;
     SINFO("BedrockServer deleted");
 
+    // Finished with our signal handler.
+    SStopSignalThread();
+
     // All done
     SINFO("Graceful process shutdown complete");
     return 0;

mbedtls

@@ -1,3 +1,4 @@
+#pragma once
 #include <libstuff/libstuff.h>
 #include "../BedrockPlugin.h"
 

plugins/Cache.h

@@ -1,3 +1,4 @@
+#pragma once
 #include <libstuff/libstuff.h>
 #include "../BedrockPlugin.h"
 

plugins/Jobs.h

@@ -35,12 +35,21 @@ string SQLite::initializeFilename(const string& filename) {
 }
 
 SQLite::SharedData& SQLite::initializeSharedData(sqlite3* db, const string& filename, const vector<string>& journalNames) {
-    static map<string, SharedData*> sharedDataLookupMap;
+    static struct SharedDataLookupMapType {
+        map<string, SharedData*> m;
+        ~SharedDataLookupMapType() {
+            for (auto& p : m) {
+                delete(p.second);
+            }
+            m.clear();
+        }
+    } sharedDataLookupMap;
+
     static mutex instantiationMutex;
     lock_guard<mutex> lock(instantiationMutex);
-    auto sharedDataIterator = sharedDataLookupMap.find(filename);
-    if (sharedDataIterator == sharedDataLookupMap.end()) {
-        SharedData* sharedData = new SharedData();
+    auto sharedDataIterator = sharedDataLookupMap.m.find(filename);
+    if (sharedDataIterator == sharedDataLookupMap.m.end()) {
+        SharedData* sharedData = new SharedData(); // This is never deleted.
 
         // Read the highest commit count from the database, and store it in commitCount.
         string query = "SELECT MAX(maxIDs) FROM (" + _getJournalQuery(journalNames, {"SELECT MAX(id) as maxIDs FROM"}, true) + ")";
@@ -60,7 +69,7 @@ SQLite::SharedData& SQLite::initializeSharedData(sqlite3* db, const string& file
         }
 
         // Insert our SharedData object into the global map.
-        sharedDataLookupMap.emplace(filename, sharedData);
+        sharedDataLookupMap.m.emplace(filename, sharedData);
         return *sharedData;
     } else {
         // Otherwise, use the existing one.
@@ -263,7 +272,16 @@ int SQLite::_sqliteWALCallback(void* data, sqlite3* db, const char* dbName, int
             SINFO("[checkpoint] Ready for complete checkpoint but skipping because less than 100 commits since last complete checkpoint.");
             return SQLITE_OK;
         }
-        // If we get here, then full checkpoints are enabled, and we have enough pages in the WAL file to perform one.
+
+        int dbInUse = 0;
+        int useCheckResult = sqlite3_file_control(db, "main", SQLITE_FCNTL_EXTERNAL_READER, (void*)&dbInUse);
+        if (useCheckResult == SQLITE_OK && dbInUse) {
+            SINFO("Skipping complete checkpoint because external transaction in progress.");
+            return SQLITE_OK;
+        }
+
+        // If we get here, then full checkpoints are enabled, we have enough pages in the WAL file to perform one, and
+        // nothing else is stopping us from running one.
         SINFO("[checkpoint] " << pageCount << " pages behind, beginning complete checkpoint.");
 
         // This thread will run independently. We capture the variables we need here and pass them by value.

sqlitecluster/SQLite.cpp

@@ -68,6 +68,7 @@ const string SQLiteNode::consistencyLevelNames[] = {"ASYNC",
 atomic<int64_t> SQLiteNode::_currentCommandThreadID(0);
 
 const vector<STCPNode::Peer*> SQLiteNode::initPeers(const string& peerListString) {
+    _state = UNKNOWN;
     vector<Peer*> peerList;
     list<string> parsedPeerList = SParseList(peerListString);
     for (const string& peerString : parsedPeerList) {
@@ -98,6 +99,7 @@ SQLiteNode::SQLiteNode(SQLiteServer& server, SQLitePool& dbPool, const string& n
     : STCPNode(name, host, initPeers(peerList), max(SQL_NODE_DEFAULT_RECV_TIMEOUT, SQL_NODE_SYNCHRONIZING_RECV_TIMEOUT)),
       _dbPool(dbPool),
       _db(_dbPool.getBase()),
+      _state(UNKNOWN),
       _commitState(CommitState::UNINITIALIZED),
       _server(server),
       _stateChangeCount(0),

sqlitecluster/SQLiteNode.cpp

@@ -1,3 +1,4 @@
+#pragma once
 #include <libstuff/libstuff.h>
 #include <BedrockPlugin.h>
 #include <BedrockServer.h>

test/clustertest/testplugin/TestPlugin.h

@@ -65,7 +65,7 @@ struct LeadingTest : tpunit::TestFixture {
             SData cmd("Status");
             string response = newMaster.executeWaitVerifyContent(cmd);
             STable json = SParseJSONObject(response);
-            if (json["state"] == "LEADING" || json["state"] == "MASTERING") {
+            if (json["state"] == "LEADING") {
                 success = true;
                 break;
             }
@@ -118,9 +118,9 @@ struct LeadingTest : tpunit::TestFixture {
             STable json1 = SParseJSONObject(responses[1]);
             STable json2 = SParseJSONObject(responses[2]);
 
-            if ((json0["state"] == "LEADING" || json0["state"] == "MASTERING") && 
-                (json1["state"] == "FOLLOWING" || json1["state"] == "SLAVING") && 
-                (json2["state"] == "FOLLOWING" || json2["state"] == "SLAVING")) {
+            if (json0["state"] == "LEADING" &&
+                json1["state"] == "FOLLOWING" &&
+                json2["state"] == "FOLLOWING") {
 
                 break;
             }
@@ -171,7 +171,7 @@ struct LeadingTest : tpunit::TestFixture {
                     continue;
                 }
             }
-            if(json["state"] == "FOLLOWING" || json["state"] == "SLAVING") {
+            if(json["state"] == "FOLLOWING") {
                 // Make sure it was following before it was synchronizing.
                 ASSERT_TRUE(wasSynchronizing);
                 wasFollowing = true;

test/clustertest/tests/LeadingTest.cpp

@@ -146,16 +146,7 @@ string BedrockTester::startServer(bool wait) {
             }
         }
 
-        // Convert our c++ strings to old-school C strings for exec.
-        char* cargs[args.size() + 1];
-        int count = 0;
-        for(string arg : args) {
-            char* newstr = (char*)malloc(arg.size() + 1);
-            strcpy(newstr, arg.c_str());
-            cargs[count] = newstr;
-            count++;
-        }
-        cargs[count] = 0;
+
 
         // Make sure the ports we need are free.
         int portsFree = 0;
@@ -168,6 +159,35 @@ string BedrockTester::startServer(bool wait) {
                  << _controlPort << ") to start server, things will probably fail." << endl;
         }
 
+#ifdef VALGRIND
+    #define xstr(a) str(a)
+    #define str(a) #a
+
+    list<string> valgrind = SParseList(xstr(VALGRIND), ' ');
+    if (valgrind.size()) {
+        serverName = valgrind.front();
+        cout << "Starting bedrock server in '" << serverName << "' with args: " << endl;
+        auto it = valgrind.rbegin();
+        while (it != valgrind.rend()) {
+            args.push_front(*it);
+            it++;
+        }
+        cout << SComposeList(args, " ") << endl;
+        cout << "==========================" << endl;
+    }
+#endif
+
+        // Convert our c++ strings to old-school C strings for exec.
+        char* cargs[args.size() + 1];
+        int count = 0;
+        for(string arg : args) {
+            char* newstr = (char*)malloc(arg.size() + 1);
+            strcpy(newstr, arg.c_str());
+            cargs[count] = newstr;
+            count++;
+        }
+        cargs[count] = 0;
+
         // And then start the new server!
         execvp(serverName.c_str(), cargs);