Slave channel implementation

nxcomp/Loop.cpp

@@ -359,7 +359,7 @@ static void RestoreSignal(int signal);
 
 static int HandleChildren();
 
-static int HandleChild(int child);
+int HandleChild(int child);
 static int CheckChild(int pid, int status);
 static int WaitChild(int child, const char *label, int force);
 
@@ -6250,6 +6250,8 @@ int HandleChildren()
     return 1;
   }
 
+  proxy->checkSlaves();
+
   //
   // This can actually happen either because we
   // reset the pid of the child process as soon

nxcomp/Proxy.cpp

@@ -29,6 +29,8 @@
 #include <string.h>
 #include <sys/types.h>
 #include <sys/stat.h>
+#include <signal.h>
+
 #ifdef ANDROID
 #include <netinet/in.h>
 #include <netinet/ip.h>
@@ -92,6 +94,8 @@ struct sockaddr_un
 
 extern void CleanupListeners();
 
+extern int HandleChild(int);
+
 //
 // Default size of string buffers.
 //
@@ -165,6 +169,7 @@ Proxy::Proxy(int fd)
 
     fdMap_[channelId]      = nothing;
     channelMap_[channelId] = nothing;
+    slavePidMap_[channelId] = nothing;
   }
 
   inputChannel_  = nothing;
@@ -296,6 +301,66 @@ Proxy::~Proxy()
     }
   }
 
+  //
+  // Kill all active slave channel children, and
+  // give them 5 seconds to exit nicely.
+
+  #ifdef DEBUG
+  *logofs << "Proxy: Killing active slaves" << endl;
+  #endif
+
+  int slave_count = 999;
+  int loop_count = 0;
+
+  while(slave_count > 0 && loop_count < 50)
+  {
+    slave_count = 0;
+
+    for (int channelId = 0; channelId<CONNECTIONS_LIMIT; channelId++)
+    {
+      int pid = slavePidMap_[channelId];
+
+      if (pid > 1) {
+         slave_count++;
+
+         #ifdef DEBUG
+         *logofs << "Proxy: Active slave with pid " << pid << logofs_flush;
+         #endif
+
+         if ( loop_count == 0 )
+         {
+             #ifdef DEBUG
+             *logofs << "Proxy: Sending SIGTERM to " << pid << logofs_flush;
+             #endif
+             kill(pid, SIGTERM);
+         }
+         else if ( loop_count == 25 )
+         {
+             #ifdef DEBUG
+             *logofs << "Proxy: Sending SIGKILL to " << pid << logofs_flush;
+             #endif
+             kill(pid, SIGKILL);
+         }
+
+         if (HandleChild(pid))
+         {
+             #ifdef DEBUG
+             *logofs << "Proxy: Slave " << pid << " terminated" << logofs_flush;
+             #endif
+             slavePidMap_[channelId] = nothing;
+         }
+      }
+    }
+
+    if ( slave_count > 0 )
+    {
+      cerr << "Proxy: Error: Failed to kill all slave channel processes. " << slave_count << " processes still remaining." << endl;
+    }
+
+    usleep(200000);
+    loop_count++;
+  }
+
   delete transport_;
   delete compressor_;
 
@@ -6308,42 +6373,94 @@ int Proxy::handleNewGenericConnectionFromProxyUnix(int channelId, T_channel_type
 
 int Proxy::handleNewSlaveConnectionFromProxy(int channelId)
 {
-  //
-  // Implementation is incomplete. Opening a
-  // slave channel should let the proxy fork
-  // a new client and pass to it the channel
-  // descriptors. For now we make the channel
-  // fail immediately.
-  //
-  // #include <fcntl.h>
-  // #include <sys/types.h>
-  // #include <sys/stat.h>
-  //
-  // char *slaveServer = "/dev/null";
-  //
-  // #ifdef TEST
-  // *logofs << "Proxy: Opening file '" << slaveServer
-  //         << "'.\n" << logofs_flush;
-  // #endif
-  //
-  // int serverFd = open(slaveServer, O_RDWR);
-  //
-  // if (handlePostConnectionFromProxy(channelId, serverFd, channel_slave, "slave") < 0)
-  // {
-  //   return -1;
-  // }
-  //
 
-  #ifdef WARNING
-  *logofs << "Proxy: Refusing new slave connection for "
-          << "channel ID#" << channelId << "\n"
-          << logofs_flush;
-  #endif
+  cerr << "Info" << ": New slave connection on "
+       << "channel ID#" << channelId << "\n";
+
+  char *nx_slave_cmd = getenv("NX_SLAVE_CMD");
+  if (nx_slave_cmd == NULL) {
+    return -1;
+  }
 
-  cerr << "Warning" << ": Refusing new slave connection for "
-          << "channel ID#" << channelId << "\n";
+  int spair[2];
+  if (socketpair(AF_UNIX, SOCK_STREAM, 0, spair) == -1) {
+    perror("socketpair");
+    return -1;
+  }
 
-  return -1;
+  int serverFd = spair[0];
+  int clientFd = spair[1];
+
+  if (handlePostConnectionFromProxy(channelId, serverFd, channel_slave, "slave") < 0)
+  {
+    close(serverFd);
+    close(clientFd);
+    return -1;
+  }
+
+
+  int pid = fork();
+  if (pid == 0)
+  {
+
+    if (dup2(clientFd, 0) == -1)
+    {
+      perror("dup2");
+      exit(1);
+    }
+
+    if (dup2(clientFd, 1) == -1)
+    {
+      perror("dup2");
+      exit(1);
+    }
+
+    close(serverFd);
+    close(clientFd);
+
+    /* Close FDs used by NX, QVD #1208 */
+    for (int fd = 3; fd < 256; fd++) {
+        close(fd);
+    }
+
+    char *const argv[2] = {nx_slave_cmd, NULL};
+
+    if (execv(nx_slave_cmd, argv) == -1)
+    {
+      perror("execv");
+    }
+    exit(1);
+
+  }
+  else if (pid == -1)
+  {
+    // TODO Test this!
+    perror("fork");
+    close(serverFd);
+    close(clientFd);
+    return -1;
+  }
+
+  close(clientFd);
+  slavePidMap_[channelId] = pid;
+
+  cerr << "Info" << ": slave channel ID#" << channelId << " handler has PID " << pid << endl;
+
+  return 1;
+}
+
+void Proxy::checkSlaves()
+{
+  for (int channelId = 0; channelId<CONNECTIONS_LIMIT; channelId++)
+  {
+    int pid = slavePidMap_[channelId];
+
+    if (pid > 1 && HandleChild(pid))
+    {
+      slavePidMap_[channelId] = nothing;
+      cerr << "Info:" << " Handled death of slave with pid " << pid << endl;
+    }
+  }
 }
 
 int Proxy::handlePostConnectionFromProxy(int channelId, int serverFd,

nxcomp/Proxy.h

@@ -302,6 +302,8 @@ class Proxy
 
   int handleNewSlaveConnectionFromProxy(int channelId);
 
+  void checkSlaves();
+
   //
   // Force closure of channels.
   //
@@ -1268,6 +1270,7 @@ class Proxy
 
   int channelMap_[CONNECTIONS_LIMIT];
   int fdMap_[CONNECTIONS_LIMIT];
+  int slavePidMap_[CONNECTIONS_LIMIT];
 };
 
 #endif /* Proxy_H */