server: Refactor to make the event loop owned by server object

1 files changed, 203 insertions, 33 deletions

diff --git a/Source/cmServer.cxx b/Source/cmServer.cxx
index 7fc6ed7..03e0115 100644
--- a/Source/cmServer.cxx
+++ b/Source/cmServer.cxx
@@ -2,7 +2,8 @@
    file Copyright.txt or https://cmake.org/licensing for details.  */
 #include "cmServer.h"
 
-#include "cmServerConnection.h"
+#include "cmConnection.h"
+#include "cmFileMonitor.h"
 #include "cmServerDictionary.h"
 #include "cmServerProtocol.h"
 #include "cmSystemTools.h"
@@ -14,8 +15,23 @@
 #include <algorithm>
 #include <cassert>
 #include <cstdint>
+#include <memory>
 #include <utility>
 
+void on_signal(uv_signal_t* signal, int signum)
+{
+  auto conn = reinterpret_cast<cmServerBase*>(signal->data);
+  conn->OnSignal(signum);
+}
+
+static void on_walk_to_shutdown(uv_handle_t* handle, void* arg)
+{
+  (void)arg;
+  if (!uv_is_closing(handle)) {
+    uv_close(handle, &cmConnection::on_close);
+  }
+}
+
 class cmServer::DebugInfo
 {
 public:
@@ -30,11 +46,10 @@ public:
   uint64_t StartTime;
 };
 
-cmServer::cmServer(cmServerConnection* conn, bool supportExperimental)
-  : Connection(conn)
+cmServer::cmServer(cmConnection* conn, bool supportExperimental)
+  : cmServerBase(conn)
   , SupportExperimental(supportExperimental)
 {
-  this->Connection->SetServer(this);
   // Register supported protocols:
   this->RegisterProtocol(new cmServerProtocol1_0);
 }
@@ -48,23 +63,15 @@ cmServer::~cmServer()
   for (cmServerProtocol* p : this->SupportedProtocols) {
     delete p;
   }
-
-  delete this->Connection;
 }
 
-void cmServer::PopOne()
+void cmServer::ProcessRequest(cmConnection* connection,
+                              const std::string& input)
 {
-  if (this->Queue.empty()) {
-    return;
-  }
-
   Json::Reader reader;
   Json::Value value;
-  const std::string input = this->Queue.front();
-  this->Queue.erase(this->Queue.begin());
-
   if (!reader.parse(input, value)) {
-    this->WriteParseError("Failed to parse JSON input.");
+    this->WriteParseError(connection, "Failed to parse JSON input.");
     return;
   }
 
@@ -82,7 +89,7 @@ void cmServer::PopOne()
   if (request.Type == "") {
     cmServerResponse response(request);
     response.SetError("No type given in request.");
-    this->WriteResponse(response, nullptr);
+    this->WriteResponse(connection, response, nullptr);
     return;
   }
 
@@ -91,9 +98,11 @@ void cmServer::PopOne()
   if (this->Protocol) {
     this->Protocol->CMakeInstance()->SetProgressCallback(
       reportProgress, const_cast<cmServerRequest*>(&request));
-    this->WriteResponse(this->Protocol->Process(request), debug.get());
+    this->WriteResponse(connection, this->Protocol->Process(request),
+                        debug.get());
   } else {
-    this->WriteResponse(this->SetProtocolVersion(request), debug.get());
+    this->WriteResponse(connection, this->SetProtocolVersion(request),
+                        debug.get());
   }
 }
 
@@ -115,7 +124,7 @@ void cmServer::RegisterProtocol(cmServerProtocol* protocol)
   }
 }
 
-void cmServer::PrintHello() const
+void cmServer::PrintHello(cmConnection* connection) const
 {
   Json::Value hello = Json::objectValue;
   hello[kTYPE_KEY] = "hello";
@@ -134,13 +143,7 @@ void cmServer::PrintHello() const
     protocolVersions.append(tmp);
   }
 
-  this->WriteJsonObject(hello, nullptr);
-}
-
-void cmServer::QueueRequest(const std::string& request)
-{
-  this->Queue.push_back(request);
-  this->PopOne();
+  this->WriteJsonObject(connection, hello, nullptr);
 }
 
 void cmServer::reportProgress(const char* msg, float progress, void* data)
@@ -232,17 +235,26 @@ bool cmServer::Serve(std::string* errorMessage)
   }
   assert(!this->Protocol);
 
-  return Connection->ProcessEvents(errorMessage);
+  return cmServerBase::Serve(errorMessage);
 }
 
 cmFileMonitor* cmServer::FileMonitor() const
 {
-  return Connection->FileMonitor();
+  return fileMonitor.get();
 }
 
 void cmServer::WriteJsonObject(const Json::Value& jsonValue,
                                const DebugInfo* debug) const
 {
+  for (auto& connection : this->Connections) {
+    WriteJsonObject(connection.get(), jsonValue, debug);
+  }
+}
+
+void cmServer::WriteJsonObject(cmConnection* connection,
+                               const Json::Value& jsonValue,
+                               const DebugInfo* debug) const
+{
   Json::FastWriter writer;
 
   auto beforeJson = uv_hrtime();
@@ -272,7 +284,7 @@ void cmServer::WriteJsonObject(const Json::Value& jsonValue,
     }
   }
 
-  Connection->WriteData(std::string("\n") + kSTART_MAGIC + std::string("\n") +
+  connection->WriteData(std::string("\n") + kSTART_MAGIC + std::string("\n") +
                         result + kEND_MAGIC + std::string("\n"));
 }
 
@@ -334,7 +346,8 @@ void cmServer::WriteMessage(const cmServerRequest& request,
   WriteJsonObject(obj, nullptr);
 }
 
-void cmServer::WriteParseError(const std::string& message) const
+void cmServer::WriteParseError(cmConnection* connection,
+                               const std::string& message) const
 {
   Json::Value obj = Json::objectValue;
   obj[kTYPE_KEY] = kERROR_TYPE;
@@ -342,7 +355,7 @@ void cmServer::WriteParseError(const std::string& message) const
   obj[kREPLY_TO_KEY] = "";
   obj[kCOOKIE_KEY] = "";
 
-  this->WriteJsonObject(obj, nullptr);
+  this->WriteJsonObject(connection, obj, nullptr);
 }
 
 void cmServer::WriteSignal(const std::string& name,
@@ -358,7 +371,8 @@ void cmServer::WriteSignal(const std::string& name,
   WriteJsonObject(obj, nullptr);
 }
 
-void cmServer::WriteResponse(const cmServerResponse& response,
+void cmServer::WriteResponse(cmConnection* connection,
+                             const cmServerResponse& response,
                              const DebugInfo* debug) const
 {
   assert(response.IsComplete());
@@ -371,5 +385,161 @@ void cmServer::WriteResponse(const cmServerResponse& response,
     obj[kERROR_MESSAGE_KEY] = response.ErrorMessage();
   }
 
-  this->WriteJsonObject(obj, debug);
+  this->WriteJsonObject(connection, obj, debug);
+}
+
+void cmServer::OnConnected(cmConnection* connection)
+{
+  PrintHello(connection);
+}
+
+void cmServer::OnServeStart()
+{
+  cmServerBase::OnServeStart();
+  fileMonitor = std::make_shared<cmFileMonitor>(GetLoop());
+}
+
+void cmServer::StartShutDown()
+{
+  if (fileMonitor) {
+    fileMonitor->StopMonitoring();
+    fileMonitor.reset();
+  }
+  cmServerBase::StartShutDown();
+}
+
+static void __start_thread(void* arg)
+{
+  auto server = reinterpret_cast<cmServerBase*>(arg);
+  std::string error;
+  server->Serve(&error);
+}
+
+bool cmServerBase::StartServeThread()
+{
+  ServeThreadRunning = true;
+  uv_thread_create(&ServeThread, __start_thread, this);
+  return true;
+}
+
+bool cmServerBase::Serve(std::string* errorMessage)
+{
+  errorMessage->clear();
+
+  uv_signal_init(&Loop, &this->SIGINTHandler);
+  uv_signal_init(&Loop, &this->SIGHUPHandler);
+
+  this->SIGINTHandler.data = this;
+  this->SIGHUPHandler.data = this;
+
+  uv_signal_start(&this->SIGINTHandler, &on_signal, SIGINT);
+  uv_signal_start(&this->SIGHUPHandler, &on_signal, SIGHUP);
+
+  OnServeStart();
+
+  for (auto& connection : Connections) {
+    if (!connection->OnServeStart(errorMessage)) {
+      return false;
+    }
+  }
+
+  if (uv_run(&Loop, UV_RUN_DEFAULT) != 0) {
+    *errorMessage = "Internal Error: Event loop stopped in unclean state.";
+    StartShutDown();
+    return false;
+  }
+
+  ServeThreadRunning = false;
+  return true;
+}
+
+void cmServerBase::OnConnected(cmConnection*)
+{
+}
+
+void cmServerBase::OnDisconnect()
+{
+}
+
+void cmServerBase::OnServeStart()
+{
+  uv_signal_start(&this->SIGINTHandler, &on_signal, SIGINT);
+  uv_signal_start(&this->SIGHUPHandler, &on_signal, SIGHUP);
+}
+
+void cmServerBase::StartShutDown()
+{
+  if (!uv_is_closing((const uv_handle_t*)&this->SIGINTHandler)) {
+    uv_signal_stop(&this->SIGINTHandler);
+  }
+
+  if (!uv_is_closing((const uv_handle_t*)&this->SIGHUPHandler)) {
+    uv_signal_stop(&this->SIGHUPHandler);
+  }
+
+  for (auto& connection : Connections) {
+    connection->OnServerShuttingDown();
+  }
+  Connections.clear();
+
+  uv_stop(&Loop);
+
+  uv_walk(&Loop, on_walk_to_shutdown, CM_NULLPTR);
+
+  uv_run(&Loop, UV_RUN_DEFAULT);
+}
+
+bool cmServerBase::OnSignal(int signum)
+{
+  (void)signum;
+  StartShutDown();
+  return true;
+}
+
+cmServerBase::cmServerBase(cmConnection* connection)
+{
+  uv_loop_init(&Loop);
+
+  uv_signal_init(&Loop, &this->SIGINTHandler);
+  uv_signal_init(&Loop, &this->SIGHUPHandler);
+
+  this->SIGINTHandler.data = this;
+  this->SIGHUPHandler.data = this;
+
+  AddNewConnection(connection);
+}
+
+cmServerBase::~cmServerBase()
+{
+
+  if (ServeThreadRunning) {
+    StartShutDown();
+    uv_thread_join(&ServeThread);
+  }
+
+  uv_loop_close(&Loop);
+}
+
+void cmServerBase::AddNewConnection(cmConnection* ownedConnection)
+{
+  Connections.emplace_back(ownedConnection);
+  ownedConnection->SetServer(this);
+}
+
+uv_loop_t* cmServerBase::GetLoop()
+{
+  return &Loop;
+}
+
+void cmServerBase::OnDisconnect(cmConnection* pConnection)
+{
+  auto pred = [pConnection](const std::unique_ptr<cmConnection>& m) {
+    return m.get() == pConnection;
+  };
+  Connections.erase(
+    std::remove_if(Connections.begin(), Connections.end(), pred),
+    Connections.end());
+  if (Connections.empty()) {
+    StartShutDown();
+  }
 }