Added child processes handling (including kill).

castor/tape/tapeserver/threading/CMakeLists.txt

-add_library(castorTapeServerThreading Threading.cpp)
\ No newline at end of file
+add_library(castorTapeServerThreading Threading.cpp ChildProcess.cpp)
\ No newline at end of file

castor/tape/tapeserver/threading/ChildProcess.cpp

+/******************************************************************************
+ *                      ChildProcess.cpp
+ *
+ * This file is part of the Castor project.
+ * See http://castor.web.cern.ch/castor
+ *
+ * Copyright (C) 2003  CERN
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+ *
+ * 
+ *
+ * @author Castor Dev team, castor-dev@cern.ch
+ *****************************************************************************/
+
+#include "ChildProcess.hpp"
+#include <unistd.h>
+#include <stdlib.h>
+#include <sys/wait.h>
+
+void castor::tape::threading::ChildProcess::start(Cleanup & cleanup) throw (castor::tape::Exception) {
+  m_pid = fork();
+  if (!m_pid) {
+    /* We are the child process. Do our stuff and exit. */
+    cleanup();
+    exit(run());
+  } else if (-1 == m_pid) {
+    /* We are in the parent process, for failed */
+    throw castor::tape::exceptions::Errnum("Failed to fork a child process in castor::tape::threading::ChildProcess::ChildProcess()");
+  }
+  /* In parent process, child is OK. */
+  m_started = true;
+}
+
+void castor::tape::threading::ChildProcess::parseStatus(int status) {
+  if (WIFEXITED(status)) {
+     m_finished = true;
+     m_exited = true;
+     m_exitCode = WEXITSTATUS(status);
+  } else if (WIFSIGNALED(status)) {
+    m_finished = true;
+    m_wasKilled = true;
+  }
+}
+
+bool castor::tape::threading::ChildProcess::running() throw (castor::tape::Exception) {
+  /* Checking for a running process before starting gets an exception */
+  if (!m_started) throw ProcessNeverStarted();
+  /* If we are not aware of process exiting, let's check and collect exit code */
+  if (!m_finished) {
+    /* Re-check the status now. */
+    int status;
+    int ret = waitpid(m_pid, &status, WNOHANG);
+    if (-1 == ret)
+      throw castor::tape::exceptions::Errnum("Error from waitpid in castor::tape::threading::ChildProcess::running()");
+    /* Check child status*/
+    if (ret == m_pid) parseStatus(status);
+  }
+  return !m_finished;
+}
+
+void castor::tape::threading::ChildProcess::wait() throw (castor::tape::Exception) {
+  /* Checking for a running process before starting gets an exception */
+  if (!m_started) throw ProcessNeverStarted();
+  if (m_finished) return;
+  int status;
+  int ret = waitpid(m_pid, &status, 0);
+  if (-1 == ret)
+    throw castor::tape::exceptions::Errnum("Error from waitpid in castor::tape::threading::ChildProcess::wait()");
+  /* Check child status*/
+  if (ret == m_pid) parseStatus(status);
+  if(!m_finished)
+    throw castor::tape::Exception("Process did not exit after waitpid().");
+}
+
+int castor::tape::threading::ChildProcess::exitCode() throw (castor::tape::Exception) {
+  if (!m_started) throw ProcessNeverStarted();
+  if (!m_finished) {
+    int status;
+    int ret = waitpid(m_pid, &status, WNOHANG);
+    if (-1 == ret)
+      throw castor::tape::exceptions::Errnum("Error from waitpid in castor::tape::threading::ChildProcess::running()");
+    if (ret == m_pid) parseStatus(status);
+  }
+  /* Check child status*/
+  if (!m_finished) {
+    throw ProcessStillRunning();
+  }
+  if (!m_exited) {
+    throw ProcessWasKilled();
+  }
+  return m_exitCode;
+}
+
+void castor::tape::threading::ChildProcess::kill() throw (castor::tape::Exception) {
+  if (!m_started) throw ProcessNeverStarted();
+  ::kill(m_pid, SIGTERM);
+}

castor/tape/tapeserver/threading/ChildProcess.hpp

+/******************************************************************************
+ *                      ChildProcess.hpp
+ *
+ * This file is part of the Castor project.
+ * See http://castor.web.cern.ch/castor
+ *
+ * Copyright (C) 2003  CERN
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+ *
+ * 
+ *
+ * @author Castor Dev team, castor-dev@cern.ch
+ *****************************************************************************/
+#pragma once
+
+#include "../exception/Exception.hpp"
+#include <unistd.h>
+
+
+namespace castor {
+namespace tape {
+namespace threading {
+  /**
+   * A class allowing forking of a child process, and subsequent follow up
+   * of the child process. Status check, killing, return code collection.
+   */
+  class ChildProcess {
+  public:
+    /**
+     * Helper functor for child to clean up unneeded parent resources 
+     * after forking.
+     */
+    class Cleanup {
+    public:
+      virtual void operator() () = 0;
+    };
+    /**
+     * Exceptions for wrong usage.
+     */
+    class ProcessStillRunning: public castor::tape::Exception {
+    public:
+      ProcessStillRunning(const std::string & what = "Process still running"):
+      castor::tape::Exception::Exception(what) {}
+    };
+    
+    class ProcessNeverStarted: public castor::tape::Exception {
+    public:
+      ProcessNeverStarted(const std::string & what = "Process never started"):
+      castor::tape::Exception::Exception(what) {}
+    };
+    
+    class ProcessWasKilled: public castor::tape::Exception {
+    public:
+      ProcessWasKilled(const std::string & what = "Process was killed"):
+      castor::tape::Exception::Exception(what) {}
+    };
+    
+    ChildProcess(): m_started(false), m_finished(false), m_exited(false),
+      m_wasKilled(false) {}
+    /** start function, taking as an argument a callback for parent's
+     * resources cleanup. A child process can only be fired once. */
+    void start(Cleanup & cleanup) throw (castor::tape::Exception);
+    /** Check running status */
+    bool running() throw (castor::tape::Exception);
+    /** Wait for completion */
+    void wait() throw (castor::tape::Exception);
+    /** collect exit code */
+    int exitCode() throw (castor::tape::Exception);
+    /** kill */
+    void kill() throw (castor::tape::Exception);
+  private:
+    pid_t m_pid;
+    /** Was the process started? */
+    bool m_started;
+    /** As the process finished? */
+    bool m_finished;
+    /** Did the process exit cleanly? */
+    bool m_exited;
+    /** Was the process killed? */
+    bool m_wasKilled;
+    int m_exitCode;
+    /** The function actually being run in the child process. The value returned
+     * by run() will be the exit code of the process (if we get that far) */
+    virtual int run() = 0;
+    void parseStatus(int status);
+  };
+} // namespace threading
+} // namespace tape
+} // namespace castor

castor/tape/tapeserver/threading/ThreadingMTTests.cpp

@@ -24,40 +24,44 @@
 
 #include <gtest/gtest.h>
 #include "Threading.hpp"
+#include "ChildProcess.hpp"
 
 /* This is a collection of multi threaded unit tests, which can (and should
  be passed through helgrind). */
 
 namespace ThreadedUnitTests {
-  class Thread_and_basic_locking: public castor::tape::threading::Thread {
+
+  class Thread_and_basic_locking : public castor::tape::threading::Thread {
   public:
     int counter;
     castor::tape::threading::Mutex mutex;
   private:
+
     void run() {
-      for (int i=0; i<100; i++) {
+      for (int i = 0; i < 100; i++) {
         castor::tape::threading::MutexLocker ml(&mutex);
         counter++;
       }
     }
   };
-  
+
   TEST(castor_tape_threading, Thread_and_basic_locking) {
     /* If we have race conditions here, helgrind will trigger. */
     Thread_and_basic_locking mt;
     mt.counter = 0;
     mt.start();
-    for(int i=0; i<100; i++) {
+    for (int i = 0; i < 100; i++) {
       castor::tape::threading::MutexLocker ml(&mt.mutex);
       mt.counter--;
     }
     mt.wait();
     ASSERT_EQ(0, mt.counter);
   }
-  
+
   template <class S>
-  class Semaphore_ping_pong: public castor::tape::threading::Thread {
+  class Semaphore_ping_pong : public castor::tape::threading::Thread {
   public:
+
     void thread0() {
       int i = 100;
       while (i > 0) {
@@ -68,6 +72,7 @@ namespace ThreadedUnitTests {
     }
   private:
     S m_sem0, m_sem1;
+
     void run() {
       int i = 100;
       while (i > 0) {
@@ -77,27 +82,29 @@ namespace ThreadedUnitTests {
       }
     }
   };
-  
+
   TEST(castor_tape_threading, PosixSemaphore_ping_pong) {
     Semaphore_ping_pong<castor::tape::threading::PosixSemaphore> spp;
     spp.start();
     spp.thread0();
     spp.wait();
   }
-  
+
   TEST(castor_tape_threading, CondVarSemaphore_ping_pong) {
     Semaphore_ping_pong<castor::tape::threading::CondVarSemaphore> spp;
     spp.start();
     spp.thread0();
     spp.wait();
   }
-  
-  class Thread_exception_throwing: public castor::tape::threading::Thread {
+
+  class Thread_exception_throwing : public castor::tape::threading::Thread {
   private:
+
     void run() {
       throw castor::tape::Exception("Exception in child thread");
     }
   };
+
   TEST(castor_tape_threading, Thread_exception_throwing) {
     Thread_exception_throwing t, t2;
     t.start();
@@ -110,5 +117,66 @@ namespace ThreadedUnitTests {
       ASSERT_NE(std::string::npos, w.find("Exception in child thread"));
     }
   }
+
+  class emptyCleanup : public castor::tape::threading::ChildProcess::Cleanup {
+  public:
+
+    virtual void operator ()() { };
+  };
+
+  class myOtherProcess : public castor::tape::threading::ChildProcess {
+  private:
+
+    int run() {
+      /* Just sleep a bit so the parent process gets a chance to see us running */
+      struct timespec ts;
+      ts.tv_sec = 0;
+      ts.tv_nsec = 50000;
+      nanosleep(&ts, NULL);
+      return 123;
+    }
+  };
+
+  TEST(castor_tape_threading, ChildProcess_return_value) {
+    myOtherProcess cp;
+    emptyCleanup cleanup;
+    EXPECT_THROW(cp.exitCode(), castor::tape::threading::ChildProcess::ProcessNeverStarted);
+    EXPECT_NO_THROW(cp.start(cleanup));
+    EXPECT_THROW(cp.exitCode(), castor::tape::threading::ChildProcess::ProcessStillRunning);
+    EXPECT_NO_THROW(cp.wait());
+    ASSERT_EQ(123, cp.exitCode());
+  }
+
+  class myInfiniteSpinner : public castor::tape::threading::ChildProcess {
+  private:
+
+    int run() {
+      /* Loop forever (politely) */
+      while (true) {
+        struct timespec ts;
+        ts.tv_sec = 0;
+        ts.tv_nsec = 10*1000*1000;
+        nanosleep(&ts, NULL);
+      }
+      return 321;
+    }
+  };
+
+  TEST(castor_tape_threading, ChildProcess_killing) {
+    myInfiniteSpinner cp;
+    emptyCleanup cleanup;
+    EXPECT_THROW(cp.kill(), castor::tape::threading::ChildProcess::ProcessNeverStarted);
+    EXPECT_NO_THROW(cp.start(cleanup));
+    EXPECT_THROW(cp.exitCode(), castor::tape::threading::ChildProcess::ProcessStillRunning);
+    ASSERT_EQ(true, cp.running());
+    EXPECT_NO_THROW(cp.kill());
+    /* The effect is not immediate, wait a bit. */
+    struct timespec ts;
+    ts.tv_sec = 0;
+    ts.tv_nsec = 10*1000*1000;
+    nanosleep(&ts, NULL);
+    ASSERT_EQ(false, cp.running());
+    EXPECT_THROW(cp.exitCode(), castor::tape::threading::ChildProcess::ProcessWasKilled);
+  }
 } // namespace ThreadedUnitTests
 

castor/tape/tapeserver/threading/ThreadingTests.cpp

@@ -24,6 +24,8 @@
 
 #include <gtest/gtest.h>
 #include "Threading.hpp"
+#include "ChildProcess.hpp"
+#include <time.h>
 
 /* Note: those tests create multi threading errors on purpose and should not
  * be run in helgrind */