Make TcpServerThreadForTesting emit error messages instead of hanging.

Previously it hanged in destructor if no client has connected.

pagespeed/kernel.gyp

@@ -464,6 +464,7 @@
       ],
       'dependencies': [
         'pagespeed_base',
+        'kernel_test_util',
         '<(DEPTH)/third_party/apr/apr.gyp:apr',
       ],
     },

pagespeed/system/tcp_server_thread_for_testing.cc

@@ -22,6 +22,7 @@
 #include "apr_network_io.h"
 #include "base/logging.h"
 #include "pagespeed/kernel/base/abstract_mutex.h"
+#include "pagespeed/kernel/base/gtest.h"
 #include "pagespeed/system/apr_thread_compatible_pool.h"
 
 namespace net_instaweb {
@@ -33,10 +34,29 @@ TcpServerThreadForTesting::TcpServerThreadForTesting(
       ready_notify_(mutex_->NewCondvar()),
       pool_(AprCreateThreadCompatiblePool(nullptr)),
       requested_listen_port_(listen_port),
-      actual_listening_port_(0) {
-}
+      actual_listening_port_(0),
+      listen_sock_(nullptr),
+      terminating_(false) {}
 
 TcpServerThreadForTesting::~TcpServerThreadForTesting() {
+  // We want to ensure that the thread is terminated and it has accepted exactly
+  // one connection. Consider several scenarios:
+  // 1. Thread was not started before destructor is called. Then the Join()
+  //    below will fail as expected.
+  // 2. Thread was started and our mutex-guarded block happened after creation
+  //    of listen_sock_. Then we will shut down the socket so accept() in the
+  //    thread will fail and the thread will close the socket and terminate.
+  // 3. Thread was started and our mutex-guarded block happened before creation
+  //    of listen_sock_. It's extremely unlikely race as it requires the
+  //    destructor to be called right after Start(). If it ever happens, CHECK()
+  //    in the thread will fail.
+  {
+    ScopedMutex lock(mutex_.get());
+    terminating_ = true;
+    if (listen_sock_) {
+      apr_socket_shutdown(listen_sock_, APR_SHUTDOWN_READWRITE);
+    }
+  }
   this->Join();
   if (pool_ != nullptr) {
     apr_pool_destroy(pool_);
@@ -59,13 +79,27 @@ void TcpServerThreadForTesting::PickListenPortOnce(
 }
 
 void TcpServerThreadForTesting::Run() {
-  apr_socket_t* listen_sock = CreateAndBindSocket();
+  // We do not want to hold mutex during accept(), hence the local copy.
+  apr_socket_t* local_listen_sock;
+  {
+    ScopedMutex lock(mutex_.get());
+    CHECK(!terminating_);
+    local_listen_sock = listen_sock_ = CreateAndBindSocket();
+  }
   apr_socket_t* accepted_socket;
-  apr_status_t status = apr_socket_accept(&accepted_socket, listen_sock, pool_);
-  CHECK_EQ(status, APR_SUCCESS)
-      << "TcpServerThreadForTesting apr_socket_accept";
-  HandleClientConnection(accepted_socket);
-  apr_socket_close(listen_sock);
+  apr_status_t status =
+      apr_socket_accept(&accepted_socket, local_listen_sock, pool_);
+  EXPECT_EQ(APR_SUCCESS, status)
+      << "TcpServerThreadForTesting: "
+         "apr_socket_accept failed (did not receive a connection?)";
+  if (status == APR_SUCCESS) {
+    HandleClientConnection(accepted_socket);
+  }
+  {
+    ScopedMutex lock(mutex_.get());
+    apr_socket_close(listen_sock_);
+    listen_sock_ = nullptr;
+  }
 }
 
 apr_port_t TcpServerThreadForTesting::GetListeningPort() {
@@ -114,11 +148,8 @@ apr_socket_t* TcpServerThreadForTesting::CreateAndBindSocket() {
   apr_port_t port = requested_listen_port_;
   CreateAndBindSocket(pool_, &sock, &port);
 
-  {
-    ScopedMutex lock(mutex_.get());
-    actual_listening_port_ = port;
-    ready_notify_->Broadcast();
-  }
+  actual_listening_port_ = port;
+  ready_notify_->Broadcast();
 
   return sock;
 }

pagespeed/system/tcp_server_thread_for_testing.h

@@ -41,7 +41,7 @@ class TcpServerThreadForTesting : public ThreadSystem::Thread {
   // The socket isn't actually created until the thread is Start()ed.
   TcpServerThreadForTesting(apr_port_t listen_port, StringPiece thread_name,
                             ThreadSystem* thread_system);
-  // Calls this->Join() which will hang if a connection has not been handled.
+  // Ensure that server has received a connection and terminate the thread.
   virtual ~TcpServerThreadForTesting();
 
   // Wait for thread to successfully start listening and then return the actual
@@ -62,7 +62,7 @@ class TcpServerThreadForTesting : public ThreadSystem::Thread {
 
   // Returns a socket bound to requested_listen_port_ if non-zero, otherwise
   // whatever the system picked. Updates actual_listening_port_.
-  apr_socket_t* CreateAndBindSocket();
+  apr_socket_t* CreateAndBindSocket() EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
   // Static alternative of CreateAndBindSocket(), used in PickListenPortOnce.
   static void CreateAndBindSocket(apr_pool_t* apr_pool, apr_socket_t** socket,
@@ -72,10 +72,12 @@ class TcpServerThreadForTesting : public ThreadSystem::Thread {
   void Run() override;
 
   scoped_ptr<ThreadSystem::CondvarCapableMutex> mutex_;
-  scoped_ptr<ThreadSystem::Condvar> ready_notify_;
+  scoped_ptr<ThreadSystem::Condvar> ready_notify_ GUARDED_BY(mutex_);
   apr_pool_t* pool_;
   const apr_port_t requested_listen_port_;
   apr_port_t actual_listening_port_ GUARDED_BY(mutex_);
+  apr_socket_t* listen_sock_ GUARDED_BY(mutex_);
+  bool terminating_ GUARDED_BY(mutex_);
 };
 
 }  // namespace net_instaweb