added pid/timeout/kill functionality to stats

lib/procs.cpp

@@ -25,6 +25,8 @@ std::map<pid_t, std::string> Util::Procs::plist;
 std::map<pid_t, Util::TerminationNotifier> Util::Procs::exitHandlers;
 bool Util::Procs::handler_set = false;
 
+
+
 static bool childRunning(pid_t p) {
   pid_t ret = waitpid(p, 0, WNOHANG);
   if (ret == p) {
@@ -33,10 +35,12 @@ static bool childRunning(pid_t p) {
   if (ret < 0 && errno == EINTR) {
     return childRunning(p);
   }
-  if (kill(p, 0) == 0) {
-    return true;
-  }
-  return false;
+  return !kill(p, 0);
+}
+
+/// sends sig 0 to process (pid). returns true if process is running
+bool Util::Procs::isRunnning(pid_t pid){
+  return !kill(pid, 0);
 }
 
 /// Called at exit of any program that used a Start* function.
@@ -657,9 +661,13 @@ void Util::Procs::Stop(std::string name) {
 /// Stops the process with this pid, if running.
 /// \arg name The PID of the process to stop.
 void Util::Procs::Stop(pid_t name) {
-  if (isActive(name)) {
-    kill(name, SIGTERM);
-  }
+  kill(name, SIGTERM);  
+}
+
+/// Stops the process with this pid, if running.
+/// \arg name The PID of the process to murder.
+void Util::Procs::Murder(pid_t name) {
+  kill(name, SIGKILL);  
 }
 
 /// (Attempts to) stop all running child processes.

lib/procs.h

@@ -34,10 +34,12 @@ namespace Util {
       static pid_t StartPiped2(std::string name, std::string cmd1, std::string cmd2, int * fdin, int * fdout, int * fderr1, int * fderr2);
       static void Stop(std::string name);
       static void Stop(pid_t name);
+      static void Murder(pid_t name);
       static void StopAll();
       static int Count();
       static bool isActive(std::string name);
       static bool isActive(pid_t name);
+      static bool isRunnning(pid_t pid);
       static pid_t getPid(std::string name);
       static std::string getName(pid_t name);
       static bool SetTerminationNotifier(pid_t pid, TerminationNotifier notifier);

lib/shared_memory.cpp

@@ -7,7 +7,7 @@
 #include <cstdlib>
 #include <cstdio>
 #include <unistd.h>
-
+#include <mist/procs.h>
 #include <iostream>
 #include "defines.h"
 #include "shared_memory.h"
@@ -22,6 +22,13 @@ namespace IPC {
     p[3] = val & 0xFF;
   }
 
+  /// Stores a short value of val in network order to the pointer p.
+  static void htobs(char * p, short val) {
+    p[0] = (val >> 8) & 0xFF;
+    p[1] = val & 0xFF;
+  }
+
+
   /// Stores a long long value of val in network order to the pointer p.
   static void htobll(char * p, long long val) {
     p[0] = (val >> 56) & 0xFF;
@@ -39,6 +46,11 @@ namespace IPC {
     val = ((long)p[0] << 24) | ((long)p[1] << 16) | ((long)p[2] << 8) | p[3];
   }
 
+  /// Reads a short value of p in host order to val.
+  static void btohs(char * p, unsigned short & val) {
+    val = ((short)p[0] << 8) | p[1];
+  }
+
   /// Reads a long value of p in host order to val.
   static void btohl(char * p, unsigned int & val) {
     val = ((long)p[0] << 24) | ((long)p[1] << 16) | ((long)p[2] << 8) | p[3];
@@ -581,6 +593,18 @@ namespace IPC {
     return result;
   }
 
+///\brief Sets PID field
+  void statExchange::pid(unsigned short id) {
+    htobs(data + 92, id);
+  }
+
+  ///\brief Gets PID field
+  unsigned short statExchange::pid() {
+    unsigned short result;
+    btohs(data + 92, result);
+    return result;
+  }
+
   ///\brief Creates a semaphore guard, locks the semaphore on call
   semGuard::semGuard(semaphore * thisSemaphore) : mySemaphore(thisSemaphore) {
     mySemaphore->wait();
@@ -721,14 +745,20 @@ namespace IPC {
       while (offset + payLen + (hasCounter ? 1 : 0) <= it->len) {
         if (hasCounter) {
           if (it->mapped[offset] != 0) {
-            int counter = it->mapped[offset];
+            char * counter = it->mapped+offset;
             //increase the count if needed
             if (id >= amount) {
               amount = id + 1;
               DEBUG_MSG(DLVL_VERYHIGH, "Shared memory %s is now at count %u", baseName.c_str(), amount);
             }            
+            unsigned short tmpPID = *((unsigned short *)(it->mapped+1+offset+payLen-2));
+            DEBUG_MSG(DLVL_FAIL, "get PID: %d ", tmpPID);
+            if(!Util::Procs::isRunnning(tmpPID) && !(*counter == 126 || *counter == 127 || *counter == 254 || *counter == 255)){
+              WARN_MSG("process disappeared, timing out. (pid %d)", tmpPID);    
+              *counter = 126; //if process is already dead, instant timeout.
+            }
             callback(it->mapped + offset + 1, payLen, id);
-            switch (counter) {
+            switch (*counter) {
               case 127:
                 DEBUG_MSG(DLVL_HIGH, "Client %u requested disconnect", id);
                 break;
@@ -742,9 +772,18 @@ namespace IPC {
                 DEBUG_MSG(DLVL_WARN, "Client %u disconnect timed out", id);
                 break;
               default:
+                if(*counter > 10 && *counter < 126 ){
+                  if(*counter < 30){
+                    ERROR_MSG("process unresponsive. sending sigterm to pid %d",tmpPID);
+                    Util::Procs::Stop(tmpPID); //soft kill  
+                  } else {      
+                    FAIL_MSG("process really unresponsive. sending sigkill to pid %d", tmpPID);
+                    Util::Procs::Murder(tmpPID); //improved kill      
+                  }
+                }
                 break;
             }
-            if (counter == 127 || counter == 126 || counter == 255 || counter == 254) {
+            if (*counter == 127 || *counter == 126 || *counter == 255 || *counter == 254) {
               memset(it->mapped + offset + 1, 0, payLen);
               it->mapped[offset] = 0;
             } else {
@@ -876,6 +915,8 @@ namespace IPC {
           offsetOnPage = offset;
           if (hasCounter) {
             myPage.mapped[offset] = 1;
+            *((unsigned short *)(myPage.mapped+1+offset+len-2))=getpid();
+            DEBUG_MSG(DLVL_FAIL, "set PID: %d ", *((unsigned short *)(myPage.mapped+1+offset+len-2)));
           }
           break;
         }

lib/shared_memory.h

@@ -10,7 +10,8 @@
 #include <semaphore.h>
 #endif
 
-#define STAT_EX_SIZE 92
+#define STAT_EX_SIZE 94
+#define PLAY_EX_SIZE 32
 
 namespace IPC {
 
@@ -36,6 +37,8 @@ namespace IPC {
       std::string connector();
       void crc(unsigned int sum);
       unsigned int crc();
+      void pid(unsigned short id);
+      unsigned short pid();
   private:
       ///\brief The payload for the stat exchange
       /// - 8 byte - now (timestamp of last statistics)
@@ -47,6 +50,7 @@ namespace IPC {
       /// - 20 byte - streamName (name of the stream peer is viewing)
       /// - 20 byte - connector (name of the connector the peer is using)
       /// - 4 byte - CRC32 of user agent (or zero if none)
+      /// - 2 byte - process PID
       char * data;
   };