hived: ignore the invalid connection from forked child processes

Author: yorkie

src/client/hive_server.cpp

@@ -209,6 +209,10 @@ void TrHiveServer::onCreateClient(hive_comm::TrCreateClientRequest &req, hive_co
     }
     isChild = true;
     running = false;
+
+    auto client = sender.getClient();
+    assert(client != nullptr);
+    client->disconnect(); // Disconnect the client at the child process.
   }
 }
 

src/common/hive/sender.hpp

@@ -35,7 +35,7 @@ namespace hive_comm
 
       if (TR_UNLIKELY(message == nullptr))
       {
-        DEBUG(LOG_TAG_CONTENT, "Failed to serialize HiveCommand(%d)", command.type);
+        DEBUG(LOG_TAG_ERROR, "Failed to create message for hive command: type=%d", command.type);
         return false;
       }
 
@@ -44,11 +44,23 @@ namespace hive_comm
       auto success = message->serialize(&data, &size);
       delete message;
       if (!success)
+      {
+        DEBUG(LOG_TAG_ERROR, "Failed to serialize HiveCommand(%d) from message", command.type);
         return false;
+      }
 
       auto r = sendRaw(data, size);
       free(data);
-      return r;
+
+      if (!r)
+      {
+        DEBUG(LOG_TAG_ERROR, "Failed to send HiveCommand(%d) to the peer", command.type);
+        return false;
+      }
+      else
+      {
+        return true;
+      }
     }
   };
 }

src/common/ipc.hpp

@@ -148,14 +148,29 @@ namespace ipc
     }
 
   public:
+    inline int getFd() const
+    {
+      return fd;
+    }
+
+    inline TrOneShotClient<T> *getClient() const
+    {
+      return client;
+    }
+
     bool send(T data)
     {
       return sendRaw(&data, sizeof(data));
     }
     bool sendRaw(const void *data, size_t size)
     {
-      if (fd == -1 || client == nullptr || client->invalid())
+      if (fd == -1 ||
+          client == nullptr ||
+          client->invalid())
+      {
+        DEBUG(LOG_TAG_ERROR, "Failed to send data because the client is invalid or disconnected.");
         return false;
+      }
 
       int bytesSent = 0;
       while (bytesSent < size)
@@ -168,7 +183,7 @@ namespace ipc
             continue;
           if (errno == ECONNRESET || errno == EPIPE)
             client->invalid(true);
-          DEBUG(LOG_TAG_IPC, "Failed to send data(bytes=%d): %s", remaining, strerror(errno));
+          DEBUG(LOG_TAG_ERROR, "Failed to send data(bytes=%d): %s", remaining, strerror(errno));
           return false;
         }
         else
@@ -202,7 +217,15 @@ namespace ipc
     }
 
   public:
-    int getFd() { return fd; }
+    inline int getFd() const
+    {
+      return fd;
+    }
+
+    inline TrOneShotClient<T> *getClient() const
+    {
+      return client;
+    }
 
     /**
      * Try to receive a new instance T with a timeout.
@@ -342,7 +365,7 @@ namespace ipc
       disconnect();
     }
 
-  private:
+  public:
     bool connect(int port, bool blocking)
     {
       if (connected == true)
@@ -377,6 +400,8 @@ namespace ipc
       }
       connected = false;
     }
+
+  private:
     bool sendHandshake(int customId)
     {
       if (fd == -1 || connected == false)

src/common/ipc_message.hpp

@@ -173,7 +173,10 @@ namespace ipc
 
       char *buffer = (char *)malloc(bufferSize);
       if (buffer == nullptr)
+      {
+        DEBUG(LOG_TAG_ERROR, "[OOM] Failed to allocate memory(%zu bytes) for IPC message", bufferSize);
         return false; // out of memory
+      }
 
       size_t offset = 0;
       // Write header

src/runtime/constellation.cpp

@@ -121,7 +121,7 @@ uint32_t TrConstellation::open(string url, optional<TrDocumentRequestInit> init)
   auto content = contentManager->makeContent();
   if (content == nullptr)
   {
-    DEBUG(LOG_TAG_ERROR, "Failed to create a new content: %s", url.c_str());
+    DEBUG(LOG_TAG_ERROR, "Failed to select or create a new content process: %s", url.c_str());
     return 0;
   }
 

src/runtime/content.cpp

@@ -33,9 +33,22 @@ void TrContentRuntime::preStart()
   // Send the create process request to the hive daemon.
   TrDocumentRequestInit init;
   init.id = id;
-  contentManager->hived->createClient(init, [this](pid_t pid)
-                                      { this->pid = pid; });
-  reportDocumentEvent(TrDocumentEventType::SpawnProcess);
+
+  auto setupContentOnCreated = [this](pid_t pid)
+  {
+    DEBUG(LOG_TAG_CONTENT, "Content(%d) is created with pid: %d", id, pid);
+    assert(pid > 0 && "The process ID should be valid.");
+    this->pid = pid;
+  };
+  if (contentManager->hived->createClient(init, setupContentOnCreated))
+  {
+    reportDocumentEvent(TrDocumentEventType::SpawnProcess);
+  }
+  else
+  {
+    DEBUG(LOG_TAG_ERROR, "Failed to pre-start the content(%d) process.", id);
+    available = false; // If the process creation fails, set available to false.
+  }
 }
 
 void TrContentRuntime::start(TrDocumentRequestInit &init)
@@ -63,16 +76,15 @@ void TrContentRuntime::resume()
 
 void TrContentRuntime::dispose(bool waitsForExit)
 {
-  assert(pid != 0);
-  if (pid == INVALID_PID) // The process is not started or exited.
-    return;
-
   available = false;
   disableRendering = true;
   contentManager->constellation->mediaManager->removeSoundSourcesByContent(id); // Remove the sound sources.
   contentManager->hived->terminateClient(id);
+
   if (waitsForExit)
   {
+    assert(pid != INVALID_PID);
+
     DEBUG(LOG_TAG_CONTENT, "Waiting for the content(%d) to exit...", id);
     unique_lock<mutex> lock(exitingMutex);
     exitedCv.wait(lock, [this]

src/runtime/content.hpp

@@ -232,6 +232,14 @@ class TrContentRuntime final : public std::enable_shared_from_this<TrContentRunt
   bool removeXRSession(xr::TrXRSession *session);
 
 private:
+  inline bool isUsed() const
+  {
+    return used.load();
+  }
+  inline void setUsed()
+  {
+    used = true;
+  }
   void recvCommandBuffers(WorkerThread &worker, uint32_t timeout);
   void recvEvent();
   void recvMediaRequest();
@@ -297,7 +305,7 @@ class TrContentRuntime final : public std::enable_shared_from_this<TrContentRunt
    */
   std::atomic<bool> started = false;
   /**
-   * The flag `available` is to indicate the content is available for the client process, it's set to true when the client process is
+   * The flag `available` is to indicate the content is available for the client process, it's set to `true` when the client process is
    * started or pre-started.
    */
   std::atomic<bool> available = false;

src/runtime/content_manager.cpp

@@ -105,21 +105,33 @@ shared_ptr<TrContentRuntime> TrContentManager::makeContent()
     unique_lock<shared_mutex> lock(contentsMutex);
     for (auto content : contents)
     {
-      if (!content->used)
+      // Check if the content is available and not destroyed.
+      if (!content->available ||
+          content->pid == INVALID_PID ||
+          content->shouldDestroy)
+        continue;
+
+      if (!content->isUsed())
       {
-        content->used = true;
+        content->setUsed();
         contentToUse = content;
+        DEBUG(LOG_TAG_CONTENT,
+              "Selected an pre-content runtime #%d(%d) for use",
+              content->id,
+              content->pid.load());
         break;
       }
     }
   }
+
   if (contentToUse == nullptr)
   {
     // Create a new content runtime when there is no available content.
     contentToUse = TrContentRuntime::Make(this);
+    DEBUG(LOG_TAG_CONTENT, "Failed to find an available content, creating a new one #%d", contentToUse->id);
     {
       unique_lock<shared_mutex> lock(contentsMutex);
-      contentToUse->used = true;
+      contentToUse->setUsed();
       contents.push_back(contentToUse);
     }
   }

src/runtime/hive_daemon.cpp

@@ -93,8 +93,12 @@ void TrHiveDaemon::shutdown()
 bool TrHiveDaemon::createClient(TrDocumentRequestInit &requestInit, function<void(pid_t)> callback)
 {
   if (commandSender == nullptr)
+  {
+    DEBUG(LOG_TAG_ERROR, "Skipped creating a client because the command sender is not initialized.");
     return false;
+  }
 
+  unique_lock<shared_mutex> lock(mutexForCommand);
   hive_comm::TrCreateClientRequest req(requestInit);
   {
     unique_lock<shared_mutex> lock(mutexForCreateProcessCallbacks);
@@ -106,7 +110,7 @@ bool TrHiveDaemon::createClient(TrDocumentRequestInit &requestInit, function<voi
 bool TrHiveDaemon::terminateClient(uint32_t id)
 {
   assert(commandSender != nullptr);
-
+  unique_lock<shared_mutex> lock(mutexForCommand);
   hive_comm::TrTerminateClientRequest req(id);
   return commandSender->sendCommand(req);
 }
@@ -115,7 +119,12 @@ void TrHiveDaemon::tick()
 {
   recvOutput();
   if (!checkDaemonAlive())
+  {
+    DEBUG(LOG_TAG_ERROR, "The hive daemon is not alive, cleaning up and restarting...");
+    cleanupDaemonResources();
+    start();
     return;
+  }
 
   // Check for new client
   acceptChanClient(0);
@@ -188,13 +197,17 @@ void TrHiveDaemon::onDeamonProcess()
 
 void TrHiveDaemon::onNewChanClient(TrOneShotClient<hive_comm::TrHiveCommandMessage> &client)
 {
-  if (commandReceiver != nullptr || commandSender != nullptr)
+  if (commandReceiver == nullptr && commandSender == nullptr)
   {
-    commandReceiver.reset();
-    commandSender.reset();
+    commandReceiver = make_unique<hive_comm::TrHiveCommandReceiver>(&client);
+    commandSender = make_unique<hive_comm::TrHiveCommandSender>(&client);
+  }
+  else
+  {
+    commandChanServer->removeClient(&client);
+    DEBUG(LOG_TAG_ERROR,
+          "Received a new hived command channel client, but the command receiver or sender is already initialized.");
   }
-  commandReceiver = make_unique<hive_comm::TrHiveCommandReceiver>(&client);
-  commandSender = make_unique<hive_comm::TrHiveCommandSender>(&client);
 }
 
 void TrHiveDaemon::acceptChanClient(int timeout)
@@ -238,6 +251,45 @@ bool TrHiveDaemon::checkDaemonAlive()
   return true;
 }
 
+void TrHiveDaemon::cleanupDaemonResources()
+{
+  // Reset daemon state
+  daemonPid = -1;
+  daemonReady = false;
+
+  // Close child pipes if they're open
+  if (childPipes[0] != -1)
+  {
+    close(childPipes[0]);
+    childPipes[0] = -1;
+  }
+  if (childPipes[1] != -1)
+  {
+    close(childPipes[1]);
+    childPipes[1] = -1;
+  }
+
+  // Stop and reset command receiver worker
+  if (recvWorker != nullptr)
+  {
+    recvWorker->stop();
+    recvWorker.reset();
+  }
+
+  // Reset command channel components
+  if (commandReceiver != nullptr)
+    commandReceiver.reset();
+  if (commandSender != nullptr)
+    commandSender.reset();
+  if (commandChanServer != nullptr)
+    commandChanServer.reset();
+
+  // Clear pending callbacks and notify them of failure
+  pendingCreateProcessCallbacks.clear();
+
+  DEBUG(LOG_TAG_CONTENT, "Hive daemon resources cleaned up.");
+}
+
 bool TrHiveDaemon::recvOutput()
 {
   if (!started())
@@ -300,22 +352,29 @@ void TrHiveDaemon::recvCommand()
     case hive_comm::TrHiveCommandType::CreateClientResponse:
     {
       auto res = hive_comm::TrHiveCommandBase::FromMessage<hive_comm::TrCreateClientResponse>(commandMessage);
+      DEBUG(LOG_TAG_CONTENT, "<< CreateClientResponse(%d) with pid(%d)", res.documentId, res.pid);
+
       function<void(pid_t)> callback;
       {
         shared_lock<shared_mutex> lock(mutexForCreateProcessCallbacks);
         auto it = pendingCreateProcessCallbacks.find(res.documentId);
         if (it != pendingCreateProcessCallbacks.end())
         {
           callback = it->second;
+          pendingCreateProcessCallbacks.erase(it);
         }
       }
-      if (callback)
+
+      if (TR_LIKELY(callback))
         callback(res.pid);
+      else
+        DEBUG(LOG_TAG_ERROR, "No callback found for CreateClientResponse(%d)", res.documentId);
       break;
     }
     case hive_comm::TrHiveCommandType::TerminateClientResponse:
     {
       auto res = hive_comm::TrHiveCommandBase::FromMessage<hive_comm::TrTerminateClientResponse>(commandMessage);
+      DEBUG(LOG_TAG_CONTENT, "<< TerminateClientResponse(%d)", res.documentId);
       // TODO: handle the response?
       break;
     }
@@ -327,7 +386,8 @@ void TrHiveDaemon::recvCommand()
     case hive_comm::TrHiveCommandType::OnExitEvent:
     {
       auto exitEvent = hive_comm::TrHiveCommandBase::FromMessage<hive_comm::TrOnExitEvent>(commandMessage);
-      DEBUG(LOG_TAG_ERROR, "Received exit event from client(%d): %d", exitEvent.documentId, exitEvent.code);
+      DEBUG(LOG_TAG_CONTENT, "<< OnExitEvent(%d) with code(%d)", exitEvent.documentId, exitEvent.code);
+
       auto contentToExit = constellation->contentManager->getContent(exitEvent.documentId, true);
       if (contentToExit != nullptr)
         contentToExit->onClientProcessExited(exitEvent.code);