Minor fixes (#192)

* Fix flaky point-to-point test

* Remove unused piece of code

* Re-order fields to remove struct padding

* Fix missing initializers for primitive members

* fix possible fd leak

* Pass by const reference where it's more efficient

* More unused variables

* copy->move

* find ":" -> ':'

* Use defaulted instead of empty constructors

* Executor con/destructor fixup

* PTP broker raciness fixes

* DummyExecutor short sleep to make it much less likely to fail under TSan

include/faabric/scheduler/MpiWorld.h

@@ -230,7 +230,7 @@ class MpiWorld
 
     // Track ranks that are local to this world, and local/remote leaders
     // MPITOPTP - this information exists in the broker
-    int localLeader;
+    int localLeader = -1;
     std::vector<int> localRanks;
     std::vector<int> remoteLeaders;
     void initLocalRemoteLeaders();

include/faabric/scheduler/Scheduler.h

@@ -35,9 +35,9 @@ class ExecutorTask
                  bool needsSnapshotPushIn,
                  bool skipResetIn);
 
-    int messageIndex = 0;
     std::shared_ptr<faabric::BatchExecuteRequest> req;
     std::shared_ptr<std::atomic<int>> batchCounter;
+    int messageIndex = 0;
     bool needsSnapshotPush = false;
     bool skipReset = false;
 };
@@ -49,7 +49,7 @@ class Executor
 
     explicit Executor(faabric::Message& msg);
 
-    virtual ~Executor();
+    virtual ~Executor() = default;
 
     void executeTasks(std::vector<int> msgIdxs,
                       std::shared_ptr<faabric::BatchExecuteRequest> req);
@@ -222,7 +222,7 @@ class Scheduler
 
     // ---- Host resources and hosts ----
     faabric::HostResources thisHostResources;
-    std::atomic<int32_t> thisHostUsedSlots;
+    std::atomic<int32_t> thisHostUsedSlots = 0;
 
     void updateHostResources();
 

include/faabric/snapshot/SnapshotRegistry.h

@@ -12,7 +12,7 @@ namespace faabric::snapshot {
 class SnapshotRegistry
 {
   public:
-    SnapshotRegistry();
+    SnapshotRegistry() = default;
 
     faabric::util::SnapshotData& getSnapshot(const std::string& key);
 

include/faabric/snapshot/SnapshotServer.h

@@ -40,6 +40,6 @@ class SnapshotServer final : public faabric::transport::MessageEndpointServer
 
   private:
     faabric::transport::PointToPointBroker& broker;
-    std::atomic_size_t diffsAppliedCounter;
+    std::atomic_size_t diffsAppliedCounter = 0;
 };
 }

include/faabric/state/InMemoryStateRegistry.h

@@ -8,7 +8,7 @@ namespace faabric::state {
 class InMemoryStateRegistry
 {
   public:
-    InMemoryStateRegistry();
+    InMemoryStateRegistry() = default;
 
     std::string getMasterIP(const std::string& user,
                             const std::string& key,

include/faabric/transport/Message.h

@@ -12,11 +12,12 @@ namespace faabric::transport {
 class Message
 {
   public:
-    Message(const zmq::message_t& msgIn);
+    explicit Message(const zmq::message_t& msgIn);
 
-    Message(int sizeIn);
+    explicit Message(int sizeIn);
 
-    Message();
+    // Empty message signals shutdown
+    Message() = default;
 
     char* data();
 
@@ -31,6 +32,6 @@ class Message
   private:
     std::vector<uint8_t> bytes;
 
-    bool _more;
+    bool _more = false;
 };
 }

include/faabric/transport/PointToPointBroker.h

@@ -5,6 +5,7 @@
 #include <faabric/util/locks.h>
 #include <faabric/util/scheduling.h>
 
+#include <atomic>
 #include <condition_variable>
 #include <queue>
 #include <set>
@@ -69,7 +70,7 @@ class PointToPointGroup
     int groupId = 0;
     int groupSize = 0;
 
-    std::mutex mx;
+    std::shared_mutex mx;
 
     // Transport
     faabric::transport::PointToPointBroker& ptpBroker;
@@ -80,7 +81,7 @@ class PointToPointGroup
 
     // Distributed lock
     std::stack<int> recursiveLockOwners;
-    int lockOwnerIdx = -1;
+    std::atomic<int> lockOwnerIdx = -1;
     std::queue<int> lockWaiters;
 
     void notifyLocked(int groupIdx);

include/faabric/util/func.h

@@ -10,7 +10,7 @@ namespace faabric::util {
 std::string funcToString(const faabric::Message& msg, bool includeId);
 
 std::string funcToString(
-  const std::shared_ptr<faabric::BatchExecuteRequest> req);
+  const std::shared_ptr<faabric::BatchExecuteRequest>& req);
 
 unsigned int setMessageId(faabric::Message& msg);
 

include/faabric/util/snapshot.h

@@ -30,11 +30,11 @@ enum SnapshotMergeOperation
 class SnapshotDiff
 {
   public:
+    const uint8_t* data = nullptr;
+    size_t size = 0;
     SnapshotDataType dataType = SnapshotDataType::Raw;
     SnapshotMergeOperation operation = SnapshotMergeOperation::Overwrite;
     uint32_t offset = 0;
-    size_t size = 0;
-    const uint8_t* data = nullptr;
 
     bool noChange = false;
 

src/scheduler/Executor.cpp

@@ -24,9 +24,9 @@ ExecutorTask::ExecutorTask(int messageIndexIn,
                            std::shared_ptr<std::atomic<int>> batchCounterIn,
                            bool needsSnapshotPushIn,
                            bool skipResetIn)
-  : messageIndex(messageIndexIn)
-  , req(reqIn)
-  , batchCounter(batchCounterIn)
+  : req(std::move(reqIn))
+  , batchCounter(std::move(batchCounterIn))
+  , messageIndex(messageIndexIn)
   , needsSnapshotPush(needsSnapshotPushIn)
   , skipReset(skipResetIn)
 {}
@@ -53,8 +53,6 @@ Executor::Executor(faabric::Message& msg)
     }
 }
 
-Executor::~Executor() {}
-
 void Executor::finish()
 {
     SPDLOG_DEBUG("Executor {} shutting down", id);

src/scheduler/MpiMessageBuffer.cpp

@@ -16,7 +16,7 @@ int MpiMessageBuffer::size()
 
 void MpiMessageBuffer::addMessage(PendingAsyncMpiMessage msg)
 {
-    pendingMsgs.push_back(msg);
+    pendingMsgs.emplace_back(std::move(msg));
 }
 
 void MpiMessageBuffer::deleteMessage(const MpiMessageIterator& msgIt)

src/snapshot/SnapshotRegistry.cpp

@@ -7,8 +7,6 @@
 #include <sys/mman.h>
 
 namespace faabric::snapshot {
-SnapshotRegistry::SnapshotRegistry() {}
-
 faabric::util::SnapshotData& SnapshotRegistry::getSnapshot(
   const std::string& key)
 {

src/state/InMemoryStateRegistry.cpp

@@ -17,8 +17,6 @@ InMemoryStateRegistry& getInMemoryStateRegistry()
     return reg;
 }
 
-InMemoryStateRegistry::InMemoryStateRegistry() {}
-
 static std::string getMasterKey(const std::string& user, const std::string& key)
 {
     std::string masterKey = MASTER_KEY_PREFIX + user + "_" + key;

src/transport/Message.cpp

@@ -16,9 +16,6 @@ Message::Message(int sizeIn)
   , _more(false)
 {}
 
-// Empty message signals shutdown
-Message::Message() {}
-
 char* Message::data()
 {
     return reinterpret_cast<char*>(bytes.data());

src/transport/PointToPointBroker.cpp

@@ -150,16 +150,18 @@ void PointToPointGroup::lock(int groupIdx, bool recursive)
     if (masterIsLocal) {
         bool acquiredLock = false;
         {
-            faabric::util::UniqueLock lock(mx);
+            faabric::util::FullLock lock(mx);
 
             if (recursive && (recursiveLockOwners.empty() ||
                               recursiveLockOwners.top() == groupIdx)) {
                 // Recursive and either free, or already locked by this idx
                 recursiveLockOwners.push(groupIdx);
                 acquiredLock = true;
-            } else if (!recursive && (lockOwnerIdx == NO_LOCK_OWNER_IDX)) {
+            } else if (!recursive &&
+                       (lockOwnerIdx.load(std::memory_order_acquire) ==
+                        NO_LOCK_OWNER_IDX)) {
                 // Non-recursive and free
-                lockOwnerIdx = groupIdx;
+                lockOwnerIdx.store(groupIdx, std::memory_order_release);
                 acquiredLock = true;
             }
         }
@@ -183,7 +185,7 @@ void PointToPointGroup::lock(int groupIdx, bool recursive)
             notifyLocked(groupIdx);
         } else {
             {
-                faabric::util::UniqueLock lock(mx);
+                faabric::util::FullLock lock(mx);
                 // Need to wait to get the lock
                 lockWaiters.push(groupIdx);
             }
@@ -249,7 +251,7 @@ void PointToPointGroup::unlock(int groupIdx, bool recursive)
       ptpBroker.getHostForReceiver(groupId, POINT_TO_POINT_MASTER_IDX);
 
     if (host == conf.endpointHost) {
-        faabric::util::UniqueLock lock(mx);
+        faabric::util::FullLock lock(mx);
 
         SPDLOG_TRACE("Group idx {} unlocking {} ({} waiters, recursive {})",
                      groupIdx,
@@ -270,12 +272,14 @@ void PointToPointGroup::unlock(int groupIdx, bool recursive)
                 lockWaiters.pop();
             }
         } else {
-            lockOwnerIdx = NO_LOCK_OWNER_IDX;
-
             if (!lockWaiters.empty()) {
-                lockOwnerIdx = lockWaiters.front();
+                lockOwnerIdx.store(lockWaiters.front(),
+                                   std::memory_order_release);
                 notifyLocked(lockWaiters.front());
                 lockWaiters.pop();
+            } else {
+                lockOwnerIdx.store(NO_LOCK_OWNER_IDX,
+                                   std::memory_order_release);
             }
         }
     } else {
@@ -362,14 +366,15 @@ void PointToPointGroup::notify(int groupIdx)
 int PointToPointGroup::getLockOwner(bool recursive)
 {
     if (recursive) {
+        faabric::util::SharedLock lock(mx);
         if (!recursiveLockOwners.empty()) {
             return recursiveLockOwners.top();
         }
 
         return NO_LOCK_OWNER_IDX;
     }
 
-    return lockOwnerIdx;
+    return lockOwnerIdx.load(std::memory_order_acquire);
 }
 
 PointToPointBroker::PointToPointBroker()

src/util/func.cpp

@@ -36,7 +36,7 @@ std::string funcToString(const faabric::Message& msg, bool includeId)
 }
 
 std::string funcToString(
-  const std::shared_ptr<faabric::BatchExecuteRequest> req)
+  const std::shared_ptr<faabric::BatchExecuteRequest>& req)
 {
     return funcToString(req->messages(0), false);
 }

src/util/json.cpp

@@ -327,7 +327,7 @@ std::map<std::string, std::string> getStringStringMapFromJson(
       std::string(valuePtr, valuePtr + it->value.GetStringLength()));
     std::string keyVal;
     while (std::getline(ss, keyVal, ',')) {
-        auto pos = keyVal.find(":");
+        auto pos = keyVal.find(':');
         std::string key = keyVal.substr(0, pos);
         map[key] = keyVal.erase(0, pos + sizeof(char));
     }
@@ -350,7 +350,7 @@ std::map<std::string, int> getStringIntMapFromJson(Document& doc,
       std::string(valuePtr, valuePtr + it->value.GetStringLength()));
     std::string keyVal;
     while (std::getline(ss, keyVal, ',')) {
-        auto pos = keyVal.find(":");
+        auto pos = keyVal.find(':');
         std::string key = keyVal.substr(0, pos);
         int val = std::stoi(keyVal.erase(0, pos + sizeof(char)));
         map[key] = val;

src/util/memory.cpp

@@ -96,6 +96,7 @@ void resetDirtyTracking()
     size_t nWritten = fwrite(value, sizeof(char), 1, fd);
     if (nWritten != 1) {
         SPDLOG_ERROR("Failed to write to clear_refs ({})", nWritten);
+        fclose(fd);
         throw std::runtime_error("Failed to write to clear_refs");
     }
 

tests/dist/fixtures.h

@@ -28,7 +28,7 @@ class DistTestsFixture
         faabric::scheduler::setExecutorFactory(fac);
     }
 
-    ~DistTestsFixture() {}
+    ~DistTestsFixture() = default;
 
     std::string getWorkerIP()
     {

tests/test/endpoint/test_endpoint_api.cpp

@@ -34,8 +34,6 @@ class EndpointApiTestExecutor final : public Executor
     {
         faabric::Message& msg = reqOrig->mutable_messages()->at(msgIdx);
 
-        std::string funcStr = faabric::util::funcToString(msg, true);
-
         int returnVal = 0;
         if (msg.function() == "valid") {
             msg.set_outputdata(
@@ -76,7 +74,7 @@ class EndpointApiTestFixture : public SchedulerTestFixture
         setExecutorFactory(executorFactory);
     }
 
-    ~EndpointApiTestFixture() {}
+    ~EndpointApiTestFixture() = default;
 
   protected:
     std::shared_ptr<EndpointApiTestExecutorFactory> executorFactory;

tests/test/redis/test_redis.cpp

@@ -76,7 +76,6 @@ TEST_CASE("Basic Redis operations", "[redis]")
     SECTION("Test enqueue/ dequeue bytes")
     {
         // Enqueue some values
-        std::vector<uint8_t> bytesA = faabric::util::stringToBytes(VALUE_A);
         redisQueue.enqueueBytes(QUEUE_NAME, BYTES_A);
         redisQueue.enqueueBytes(QUEUE_NAME, BYTES_B);
         redisQueue.enqueueBytes(QUEUE_NAME, BYTES_C);
@@ -662,7 +661,7 @@ TEST_CASE("Test range set pipeline")
 
 void checkDequeueBytes(Redis& redis,
                        const std::string& queueName,
-                       const std::vector<uint8_t> expected)
+                       const std::vector<uint8_t>& expected)
 {
     unsigned long bufferLen = expected.size();
     auto buffer = std::vector<uint8_t>(bufferLen, 0);

tests/test/state/test_state.cpp

@@ -114,12 +114,6 @@ class StateServerTestFixture
         std::vector<uint8_t> actual(values.size(), 0);
         setDummyData(values);
 
-        // Get, with optional pull
-        int nMessages = 1;
-        if (doPull) {
-            nMessages = 2;
-        }
-
         // Initial pull
         const std::shared_ptr<state::StateKeyValue>& localKv = getLocalKv();
         localKv->pull();

tests/test/transport/test_point_to_point.cpp

@@ -281,9 +281,9 @@ TEST_CASE_METHOD(PointToPointClientServerFixture,
     // shared integer
     std::thread t([this, &decision, &sharedInt] {
         SLEEP_MS(1000);
-        broker.setUpLocalMappingsFromSchedulingDecision(decision);
 
         sharedInt.fetch_add(100);
+        broker.setUpLocalMappingsFromSchedulingDecision(decision);
     });
 
     broker.waitForMappingsOnThisHost(groupId);

tests/test/util/test_snapshot.cpp

@@ -15,12 +15,11 @@ namespace tests {
 class SnapshotMergeTestFixture : public SnapshotTestFixture
 {
   public:
-    SnapshotMergeTestFixture() {}
-    ~SnapshotMergeTestFixture() {}
+    SnapshotMergeTestFixture() = default;
+    ~SnapshotMergeTestFixture() = default;
 
   protected:
     std::string snapKey;
-    int snapPages;
     faabric::util::SnapshotData snap;
 
     uint8_t* setUpSnapshot(int snapPages, int sharedMemPages)