Working correctly now, including correct Server thread termination handling

Author: clauspruefer

src/ResultProcessor.cpp

@@ -31,15 +31,15 @@ void ResultProcessor::terminate(int _ignored)
 
 int ResultProcessor::_getFDFromParent(uint16_t fd)
 {
-    // Send the requested FD number to the parent
+    // send the requested FD number to the parent
     if (write(_FDPassingSocketFD, &fd, sizeof(fd)) != sizeof(fd)) {
         ERR("Failed to send FD request to parent");
         return -1;
     }
-    
+
     // Receive the FD from the parent
     int received_fd = Syscall::recvFD(_FDPassingSocketFD);
-    
+
     if (received_fd < 0) {
         ERR("Failed to receive FD from parent");
         return -1;
@@ -79,12 +79,12 @@ pid_t ResultProcessor::forkProcessResultProcessor(ResultProcessorSHMPointer_t SH
         //- connect to parent's FD passing server
         const char* socket_path = "/tmp/falcon-fd-passing.sock";
         _FDPassingSocketFD = Syscall::connectFDPassingClient(socket_path);
-        
+
         if (_FDPassingSocketFD < 0) {
             ERR("ResultProcessor: Failed to connect to FD passing server");
             exit(1);
         }
-        
+
         DBG(120, "ResultProcessor: Connected to FD passing server");
 
         //- overwrite parent termination handler
@@ -93,6 +93,9 @@ pid_t ResultProcessor::forkProcessResultProcessor(ResultProcessorSHMPointer_t SH
         //- spectre userspace protection
         prctl(PR_SET_SPECULATION_CTRL, PR_SPEC_INDIRECT_BRANCH, PR_SPEC_ENABLE, 0, 0);
 
+        //- drop privileges
+        Permission::dropPrivileges(RUNAS_USER_DEFAULT, RUNAS_GROUP_DEFAULT);
+
         DBG(120, "Child ResultProcessor Process PID:" << getpid() << " FDPassingSocketFD:" << _FDPassingSocketFD);
         DBG(120, "Child ResultProcessor SharedMemAddress:" << SHMAdresses.StaticFSPtr);
         DBG(120, "Child ResultProcessor Atomic Address:" << StaticFSLock << " Value:" << *(StaticFSLock));
@@ -148,7 +151,11 @@ pid_t ResultProcessor::forkProcessResultProcessor(ResultProcessorSHMPointer_t SH
             }
         }
 
-        DBG(-1, "Exit Parent ResultProcessor Process.");
+        DBG(-1, "Parent ResultProcessor closing Unix Domain Socket.");
+
+        close(_FDPassingSocketFD);
+
+        DBG(-1, "Exit parent ResultProcessor process.");
         exit(0);
     }
 }
@@ -198,7 +205,6 @@ uint16_t ResultProcessor::_processPythonASResults()
 {
     uint16_t processed = 0;
 
-    //for (const auto& Namespace: _Namespaces) {
     for (const auto& Namespace: ConfigRef.Namespaces) {
         for (const auto &Index: _VHostOffsetsPrecalc.at(Namespace.first)) {
             atomic_uint16_t* CanReadAddr = static_cast<atomic_uint16_t*>(getMetaAddress(Index, 0));

src/Server.cpp

@@ -3,6 +3,8 @@
 using namespace std;
 
 static bool RunServer = true;
+static std::thread _FDPassingThread;
+
 Configuration ConfigRef = Configuration();
 
 std::vector<pid_t> Server::ChildPIDs;
@@ -41,7 +43,7 @@ void Server::init()
     //- set client handler namespaces
     setClientHandlerConfig();
 
-    //- TODO: just set if exists in config, else default
+    //- TODO: only set if exists in config, else default
     SocketListenAddress = ConfigRef.ServerAddress;
     SocketListenPort = ConfigRef.ServerPort;
 
@@ -67,7 +69,7 @@ void Server::init()
 
     //- fork result processor process
     ResultProcessor::setVHostOffsets(ASRequestHandlerRef.getOffsetsPrecalc());
-    pid_t resultProcessorPID = ResultProcessor::forkProcessResultProcessor( { _SHMStaticFS, _SHMPythonASMeta, _SHMPythonASRequests, _SHMPythonASResults } );
+    const pid_t resultProcessorPID = ResultProcessor::forkProcessResultProcessor( { _SHMStaticFS, _SHMPythonASMeta, _SHMPythonASRequests, _SHMPythonASResults } );
     if (resultProcessorPID > 0) {
         addChildPID(resultProcessorPID);
     }
@@ -77,7 +79,7 @@ void Server::init()
     forkProcessASHandler( { _SHMPythonASMeta, _SHMPythonASRequests, _SHMPythonASResults } );
 
     //- check interpreter count
-    uint ASInterpreterCount = getASInterpreterCount();
+    const uint ASInterpreterCount = getASInterpreterCount();
     DBG(50, "Sum AS Interpreters:" << ASInterpreterCount);
 
     //- apply cpu bound processing
@@ -116,8 +118,23 @@ void Server::terminateChildren()
 void Server::terminate(int _ignored)
 {
     DBG(-1, "SIGTERM Main Server received, shutting down");
-    terminateChildren();
     RunServer = false;
+    terminateChildren();
+
+    if (_FDPassingThread.joinable()) {
+        _FDPassingThread.join();
+    }
+
+    std::this_thread::sleep_for(chrono::milliseconds(100));
+
+    if (_FDPassingThread.joinable()) {
+        _FDPassingThread.join();
+    }
+
+    std::this_thread::sleep_for(chrono::milliseconds(100));
+    if (_FDPassingThread.joinable()) {
+        _FDPassingThread.join();
+    }
 }
 
 void Server::setupSocket()
@@ -266,7 +283,7 @@ void Server::setupFDPassingServer()
 
     DBG(120, "FD Passing Server socket created at:" << socket_path);
 
-    // Start thread to handle FD passing requests
+    // start thread to handle FD passing requests
     _FDPassingThread = std::thread(&Server::handleFDPassingRequests, this);
 }
 
@@ -278,35 +295,36 @@ void Server::handleFDPassingRequests()
     int flags = fcntl(_FDPassingServerFD, F_GETFL, 0);
     fcntl(_FDPassingServerFD, F_SETFL, flags | O_NONBLOCK);
 
-    std::vector<int> client_fds;
+    std::vector<int> ClientFDs;
 
     while(RunServer) {
 
         struct sockaddr_un client_addr;
         socklen_t client_len = sizeof(client_addr);
+        int NewClientFD = accept(_FDPassingServerFD, (struct sockaddr*)&client_addr, &client_len);
+
+        if (NewClientFD >= 0) {
+            DBG(120, "FD passing new client connected, fd:" << NewClientFD);
 
-        int new_client_fd = accept(_FDPassingServerFD, (struct sockaddr*)&client_addr, &client_len);
+            // set client_fd non-blocking
+            int flags = fcntl(NewClientFD, F_GETFL, 0);
+            fcntl(NewClientFD, F_SETFL, flags | O_NONBLOCK);
 
-        if (new_client_fd >= 0) {
-            DBG(120, "FD passing new client connected, fd:" << new_client_fd);
-            client_fds.push_back(new_client_fd);
+            ClientFDs.push_back(NewClientFD);
         }
 
-        auto it = client_fds.begin();
-        while (it != client_fds.end()) {
+        auto it = ClientFDs.begin();
+        while (it != ClientFDs.end()) {
             int client_fd = *it;
             uint16_t requested_fd;
-
             ssize_t n = read(client_fd, &requested_fd, sizeof(requested_fd));
-
             if (n == sizeof(requested_fd)) {
                 DBG(120, "FD passing request for FD:" << requested_fd);
-
                 // send the requested FD to the client
                 if (Syscall::sendFD(client_fd, requested_fd) < 0) {
                     ERR("Failed to send FD:" << requested_fd);
                     close(client_fd);
-                    it = client_fds.erase(it);
+                    it = ClientFDs.erase(it);
                 } else {
                     DBG(120, "Successfully sent FD:" << requested_fd);
                     ++it;
@@ -315,25 +333,38 @@ void Server::handleFDPassingRequests()
                 // connection closed
                 DBG(120, "FD passing client disconnected, fd:" << client_fd);
                 close(client_fd);
-                it = client_fds.erase(it);
+                it = ClientFDs.erase(it);
             } else if (n < 0) {
                 if (errno != EAGAIN && errno != EWOULDBLOCK) {
                     ERR("Failed to read requested FD number:" << strerror(errno));
                     close(client_fd);
-                    it = client_fds.erase(it);
+                    it = ClientFDs.erase(it);
                 } else {
                     ++it;
                 }
             }
         }
-
         std::this_thread::sleep_for(chrono::microseconds(IDLE_SLEEP_MICROSECONDS));
     }
 
+    DBG(120, "FD Passing handler thread exiting");
+
+    std::this_thread::sleep_for(chrono::milliseconds(50));
+
+    DBG(120, "FD Passing handler thread closing client connections");
+
     // clean up
-    for (int fd : client_fds) {
+    //close(client_fd);
+    for (const int& fd : ClientFDs) {
         close(fd);
     }
 
-    DBG(120, "FD Passing handler thread exiting");
+    std::this_thread::sleep_for(chrono::milliseconds(50));
+
+    DBG(120, "FD Passing handler thread closing serve unix domain socket");
+
+    close(_FDPassingServerFD);
+    unlink("/tmp/falcon-fd-passing.sock");
+
+    DBG(120, "FD Passing handler thread finally exiting");
 }

src/Server.hpp

@@ -62,7 +62,6 @@ class Server : private ResultProcessor, private ASProcessHandler, private Client
     void* _SHMPythonASResults;
 
     int _FDPassingServerFD;
-    std::thread _FDPassingThread;
 
     static std::vector<pid_t> ChildPIDs;