Rewrite ServerDeathTest to avoid fork()

I'm hoping this should resolve the flaky failures seen in
#274. At the very least,
using subprocesses instead of threads, making process lifetime explicit
(rather than relying on gtest's death test infrastructure), and ensuring
that an RPC succeeds before terminating the server should make failures
easier to debug.

Should fix #274.

test/grpc/death-server.cc

@@ -1,5 +1,12 @@
+// A dummy server used in tests.
+// Successfully responds to the 1st RPC then exits with an error on the 2nd.
+
+#include <iostream>
+
 #include "eventuals/grpc/server.h"
 #include "eventuals/head.h"
+#include "eventuals/let.h"
+#include "eventuals/loop.h"
 #include "eventuals/terminal.h"
 #include "eventuals/then.h"
 #include "examples/protos/helloworld.grpc.pb.h"
@@ -9,6 +16,8 @@ using helloworld::HelloReply;
 using helloworld::HelloRequest;
 
 using eventuals::Head;
+using eventuals::Let;
+using eventuals::Loop;
 using eventuals::Terminate;
 using eventuals::Then;
 
@@ -53,14 +62,28 @@ int main(int argc, char** argv) {
 
   auto serve = [&]() {
     return server->Accept<Greeter, HelloRequest, HelloReply>("SayHello")
-        | Head()
-        | Then([](auto&& call) {
-             // NOTE: to avoid false positives with, for example, one
-             // of the 'CHECK()'s failing above, the 'ServerDeathTest'
-             // expects the string 'accepted' to be written to stderr.
-             std::cerr << "accepted" << std::endl;
-             exit(1);
-           });
+        | Map(Let([call_count = 0](auto& call) mutable {
+             return call.Reader().Read()
+                 | Head() // Only get the first element.
+                 | Then([&](auto&& request) {
+                      ++call_count;
+                      std::cout << "Got call " << call_count << std::endl;
+                      // Respond to the first call, exit on the second.
+                      if (call_count == 1) {
+                        std::cout << "Responding to call " << call_count
+                                  << std::endl;
+                        HelloReply reply;
+                        std::string prefix("Hello ");
+                        reply.set_message(prefix + request.name());
+                        return reply;
+                      } else {
+                        std::cout << "Server terminating" << std::endl;
+                        exit(1);
+                      }
+                    })
+                 | UnaryEpilogue(call);
+           }))
+        | Loop();
   };
 
   auto [future, k] = Terminate(serve());
@@ -69,6 +92,7 @@ int main(int argc, char** argv) {
 
   // NOTE: sending this _after_ we start the eventual so that we're
   // ready to accept clients!
+  std::cout << "Server serving on port " << port << std::endl;
   SendPort(port);
 
   future.get();

test/grpc/server-death-test.cc

@@ -1,5 +1,8 @@
+#include <cstdlib>
 #include <filesystem>
+#include <iostream>
 
+#include "eventuals/catch.h"
 #include "eventuals/grpc/client.h"
 #include "eventuals/grpc/server.h"
 #include "eventuals/head.h"
@@ -16,6 +19,7 @@ using helloworld::HelloRequest;
 
 using stout::Borrowable;
 
+using eventuals::Catch;
 using eventuals::Head;
 using eventuals::Let;
 using eventuals::Terminate;
@@ -39,47 +43,20 @@ TEST_F(EventualsGrpcTest, ServerDeathTest) {
     return port;
   };
 
-  // We use a thread to fork/exec the 'death-server' so that we can
-  // simultaneously run and wait for the server to die while also
-  // running the client.
-  std::thread thread([&]() {
-    std::string path = GetRunfilePathFor("death-server").string();
-
-    std::string pipe = std::to_string(pipefds[1]);
-
-    // NOTE: doing a 'fork()' when a parent has multiple threads is
-    // wrought with potential issues because the child only gets a
-    // single one of those threads (the one that called 'fork()') so
-    // we ensure here that there is only the single extra thread.
-    ASSERT_EQ(GetThreadCount(), 2);
-
-    ASSERT_DEATH(
-        [&]() {
-          // Conventional wisdom is to do the least amount possible
-          // after a 'fork()', ideally just an 'exec*()', and that's
-          // what we're doing because when we tried to do more the
-          // tests were flaky likely do to some library that doesn't
-          // properly work after doing a 'fork()'.
-          execl(path.c_str(), path.c_str(), pipe.c_str(), nullptr);
-
-          // NOTE: if 'execve()' failed then this lamdba will return
-          // and gtest will see consider this "death" to be a failure.
-        }(),
-        // NOTE: to avoid false positives we check for a death with
-        // the string 'accepted' printed to stderr.
-        "accepted");
-  });
-
-  // NOTE: we detach the thread so that there isn't a race with the
-  // thread completing and attempting to run it's destructor which
-  // will call 'std::terminate()' if we haven't yet called
-  // 'join()'. We know it's safe to detach because the thread (which
-  // acts as the parent process for the server) can destruct itself
-  // whenever it wants because it doesn't depend on anything from the
-  // test which might have been destructed before it destructs.
-  thread.detach();
+  // Launch the server before creating the client. Run the server in a
+  // subprocess so that it can run in parallel with this test without requiring
+  // threads. Capture the server's output to a file for debugging.
+  const std::string path = GetRunfilePathFor("death-server").string();
+  const std::string pipe = std::to_string(pipefds[1]);
+  const std::string command = path + " " + pipe + " &";
 
+  std::cout << "Running server with command: " << command << std::endl;
+  CHECK_EQ(0, std::system(command.c_str()))
+      << "Failed to run command " << command;
+
+  std::cout << "Waiting for server." << std::endl;
   int port = WaitForPort();
+  std::cout << "Server active on port " << port << "." << std::endl;
 
   Borrowable<CompletionPool> pool;
 
@@ -94,13 +71,37 @@ TEST_F(EventualsGrpcTest, ServerDeathTest) {
              HelloRequest request;
              request.set_name("emily");
              return call.Writer().WriteLast(request)
+                 | call.Reader().Read()
+                 | Head()
+                 | Then([](HelloReply&& response) {
+                      std::cout << "Got reply" << response.DebugString();
+                      EXPECT_EQ("Hello emily", response.message());
+                    })
+                 // Handle the exception thrown when calling Head() on an
+                 // empty stream.
+                 // TODO(alexmc, benh): Can we handle 0 to 1 responses more
+                 // cleanly by using something like Map()?
+                 | Catch()
+                       .raised<std::runtime_error>([](auto&& error) {
+                         std::cout << "Got error: " << error.what()
+                                   << std::endl;
+                       })
                  | call.Finish();
            }));
   };
 
-  auto status = *call();
-
-  EXPECT_EQ(grpc::UNAVAILABLE, status.error_code());
+  // The first call succeeds.
+  {
+    std::cout << "Sending first call." << std::endl;
+    auto status = *call();
+    EXPECT_EQ(grpc::OK, status.error_code());
+  }
+  // The server exits on the second call.
+  {
+    std::cout << "Sending second call." << std::endl;
+    auto status = *call();
+    EXPECT_EQ(grpc::UNAVAILABLE, status.error_code());
+  }
 
   close(pipefds[0]);
   close(pipefds[1]);