Add a spin-until-initialized to messenger_test to avoid races. (#1425)

Author: lmwnshn

test/messenger/messenger_test.cpp

@@ -79,8 +79,8 @@ class MessengerTests : public TerrierTest {
     return db_main;
   }
 
-  /** A dirty hack that sleeps for a little while so that sockets can clean up. */
-  static void DirtySleep() { std::this_thread::sleep_for(std::chrono::seconds(5)); }
+  /** A dirty hack that sleeps for a little while so that sockets can wake up and clean up. */
+  static void DirtySleep() { std::this_thread::sleep_for(std::chrono::seconds(2)); }
 };
 
 // NOLINTNEXTLINE
@@ -98,14 +98,25 @@ TEST_F(MessengerTests, BasicReplicationTest) {
   // done[3] is shared memory (mmap) so that the forked processes can coordinate on when they are done.
   // This is done instead of waitpid() because I can't find a way to stop googletest from freaking out on waitpid().
   // done[0] : primary, done[1] : replica1, done[2] : replica2.
+  volatile bool *init = static_cast<volatile bool *>(
+      mmap(nullptr, 3 * sizeof(bool), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0));
+  NOISEPAGE_ASSERT(MAP_FAILED != init, "mmap() failed.");
   volatile bool *done = static_cast<volatile bool *>(
       mmap(nullptr, 3 * sizeof(bool), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0));
   NOISEPAGE_ASSERT(MAP_FAILED != done, "mmap() failed.");
 
+  init[0] = false;
+  init[1] = false;
+  init[2] = false;
   done[0] = false;
   done[1] = false;
   done[2] = false;
 
+  auto spin_until_init = [init]() {
+    while (!(init[0] && init[1] && init[2])) {
+    }
+  };
+
   auto spin_until_done = [done]() {
     while (!(done[0] && done[1] && done[2])) {
     }
@@ -115,19 +126,28 @@ TEST_F(MessengerTests, BasicReplicationTest) {
     auto primary = BuildDBMain(port_primary, port_messenger_primary, "primary");
     primary->GetNetworkLayer()->GetServer()->RunServer();
 
+    init[0] = true;
+    spin_until_init();
+    DirtySleep();
+
     while (!(done[1] && done[2])) {
     }
 
     MESSENGER_LOG_TRACE("Primary done.");
     done[0] = true;
     spin_until_done();
     MESSENGER_LOG_TRACE("Primary exit.");
+    primary->ForceShutdown();
   };
 
   VoidFn replica1_fn = [=]() {
     auto replica1 = BuildDBMain(port_replica1, port_messenger_replica1, "replica1");
     replica1->GetNetworkLayer()->GetServer()->RunServer();
 
+    init[1] = true;
+    spin_until_init();
+    DirtySleep();
+
     // Set up a connection to the primary.
     auto messenger = replica1->GetMessengerLayer()->GetMessenger();
     ConnectionDestination dest_primary = Messenger::GetEndpointIPC("primary", port_messenger_primary);
@@ -168,12 +188,17 @@ TEST_F(MessengerTests, BasicReplicationTest) {
     done[1] = true;
     spin_until_done();
     MESSENGER_LOG_TRACE("Replica 1 exit.");
+    replica1->ForceShutdown();
   };
 
   VoidFn replica2_fn = [=]() {
     auto replica2 = BuildDBMain(port_replica2, port_messenger_replica2, "replica2");
     replica2->GetNetworkLayer()->GetServer()->RunServer();
 
+    init[2] = true;
+    spin_until_init();
+    DirtySleep();
+
     // Set up a connection to the primary.
     auto messenger = replica2->GetMessengerLayer()->GetMessenger();
     ConnectionDestination dest_primary = Messenger::GetEndpointIPC("primary", port_messenger_primary);
@@ -214,6 +239,7 @@ TEST_F(MessengerTests, BasicReplicationTest) {
     done[2] = true;
     spin_until_done();
     MESSENGER_LOG_TRACE("Replica 2 exit.");
+    replica2->ForceShutdown();
   };
 
   std::vector<pid_t> pids = ForkTests({primary_fn, replica1_fn, replica2_fn});
@@ -223,9 +249,14 @@ TEST_F(MessengerTests, BasicReplicationTest) {
   }
 
   DirtySleep();
-
-  UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(const_cast<bool *>(done)), 3 * sizeof(bool));
-  NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
+  {
+    UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(const_cast<bool *>(init)), 3 * sizeof(bool));
+    NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
+  }
+  {
+    UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(const_cast<bool *>(done)), 3 * sizeof(bool));
+    NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
+  }
 }
 
 // NOLINTNEXTLINE
@@ -240,13 +271,23 @@ TEST_F(MessengerTests, BasicListenTest) {
 
   uint16_t port_listen = 8030;
 
+  volatile bool *init = static_cast<volatile bool *>(
+      mmap(nullptr, 2 * sizeof(bool), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0));
+  NOISEPAGE_ASSERT(MAP_FAILED != init, "mmap() failed.");
   volatile bool *done = static_cast<volatile bool *>(
       mmap(nullptr, 2 * sizeof(bool), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0));
   NOISEPAGE_ASSERT(MAP_FAILED != done, "mmap() failed.");
 
+  init[0] = false;
+  init[1] = false;
   done[0] = false;
   done[1] = false;
 
+  auto spin_until_init = [init]() {
+    while (!(init[0] && init[1])) {
+    }
+  };
+
   auto spin_until_done = [done]() {
     while (!(done[0] && done[1])) {
     }
@@ -269,18 +310,26 @@ TEST_F(MessengerTests, BasicListenTest) {
           }
         });
 
+    init[0] = true;
+    spin_until_init();
+    DirtySleep();
+
     while (!done[1]) {
     }
 
     MESSENGER_LOG_TRACE("Primary done.");
     done[0] = true;
     spin_until_done();
     MESSENGER_LOG_TRACE("Primary exit.");
+    primary->ForceShutdown();
   };
 
   VoidFn replica1_fn = [=]() {
     auto replica1 = BuildDBMain(port_replica1, port_messenger_replica1, "replica1");
     replica1->GetNetworkLayer()->GetServer()->RunServer();
+    init[1] = true;
+    spin_until_init();
+    DirtySleep();
 
     // Set up a connection to the primary via the listen endpoint.
     auto messenger = replica1->GetMessengerLayer()->GetMessenger();
@@ -316,6 +365,7 @@ TEST_F(MessengerTests, BasicListenTest) {
     done[1] = true;
     spin_until_done();
     MESSENGER_LOG_TRACE("Replica 1 exit.");
+    replica1->ForceShutdown();
   };
 
   std::vector<pid_t> pids = ForkTests({primary_fn, replica1_fn});
@@ -325,9 +375,14 @@ TEST_F(MessengerTests, BasicListenTest) {
   }
 
   DirtySleep();
-
-  UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(const_cast<bool *>(done)), 2 * sizeof(bool));
-  NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
+  {
+    UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(const_cast<bool *>(init)), 2 * sizeof(bool));
+    NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
+  }
+  {
+    UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(const_cast<bool *>(done)), 2 * sizeof(bool));
+    NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
+  }
 }
 
 }  // namespace noisepage::messenger