SharedMemory/src/WorkerMonitorThread.cc

0001 // -*- C++ -*-
0002 //
0003 // Package:     FWCore/SharedMemory
0004 // Class  :     WorkerMonitorThread
0005 //
0006 // Implementation:
0007 //     [Notes on implementation]
0008 //
0009 // Original Author:  Chris Jones
0010 //         Created:  21/01/2020
0011 //
0012
0013 // system include files
0014 #include <cerrno>
0015 #include <csignal>
0016 #include <iostream>
0017 #include <unistd.h>
0018
0019 // user include files
0020 #include "FWCore/SharedMemory/interface/WorkerMonitorThread.h"
0021
0022 //
0023 // constants, enums and typedefs
0024 //
0025 using namespace edm::shared_memory;
0026
0027 //
0028 // static data member definitions
0029 //
0030 std::atomic<bool> WorkerMonitorThread::s_helperThreadDone = false;
0031 std::atomic<int> WorkerMonitorThread::s_pipeReadEnd = 0;
0032 std::atomic<int> WorkerMonitorThread::s_pipeWriteEnd = 0;
0033
0034 //
0035 // constructors and destructor
0036 //
0037
0038 //
0039 // member functions
0040 //
0041 void WorkerMonitorThread::run() {
0042   //std::cerr << "Started cleanup thread\n";
0043   sigset_t ensemble;
0044
0045   sigemptyset(&ensemble);
0046   sigaddset(&ensemble, SIGABRT);
0047   sigaddset(&ensemble, SIGILL);
0048   sigaddset(&ensemble, SIGBUS);
0049   sigaddset(&ensemble, SIGSEGV);
0050   sigaddset(&ensemble, SIGTERM);
0051   pthread_sigmask(SIG_BLOCK, &ensemble, nullptr);
0052
0053   //std::cerr << "Start loop\n";
0054   helperReady_ = true;
0055   while (true) {
0056     int signal = -1;
0057     auto res = read(s_pipeReadEnd.load(), &signal, sizeof(signal) / sizeof(char));
0058     if (res == -1) {
0059       if (errno == EINTR) {
0060         continue;
0061       }
0062       abort();
0063     }
0064     if (signal != 0) {
0065       if (actionSet_) {
0066         action_();
0067       }
0068       std::cerr << "Worker: SIGNAL CAUGHT " << signal << "\n";
0069       s_helperThreadDone = true;
0070       break;
0071     } /* else {
0072       std::cerr << "SIGNAL woke\n";
0073     } */
0074   }
0075   //std::cerr << "Ending cleanup thread\n";
0076 }
0077
0078 void WorkerMonitorThread::startThread() {
0079   {
0080     //Setup watchdog thread for crashing signals
0081
0082     int pipeEnds[2] = {0, 0};
0083     auto ret = pipe(pipeEnds);
0084     if (ret != 0) {
0085       abort();
0086     }
0087     s_pipeReadEnd.store(pipeEnds[0]);
0088     s_pipeWriteEnd.store(pipeEnds[1]);
0089     //Need to use signal handler since signals generated
0090     // from within a program are thread specific which can
0091     // only be handed by a signal handler
0092     setupSignalHandling();
0093
0094     std::thread t(&WorkerMonitorThread::run, this);
0095     t.detach();
0096     helperThread_ = std::move(t);
0097   }
0098   while (helperReady_.load() == false) {
0099   }
0100 }
0101
0102 void WorkerMonitorThread::setupSignalHandling() {
0103   struct sigaction act;
0104   act.sa_sigaction = sig_handler;
0105   act.sa_flags = SA_SIGINFO;
0106   sigemptyset(&act.sa_mask);
0107   sigaction(SIGABRT, &act, nullptr);
0108   sigaction(SIGILL, &act, nullptr);
0109   sigaction(SIGBUS, &act, nullptr);
0110   sigaction(SIGSEGV, &act, nullptr);
0111   sigaction(SIGTERM, &act, nullptr);
0112 }
0113
0114 void WorkerMonitorThread::stop() {
0115   stopRequested_ = true;
0116   int sig = 0;
0117   write(s_pipeWriteEnd.load(), &sig, sizeof(int) / sizeof(char));
0118 }
0119
0120 //
0121 // const member functions
0122 //
0123
0124 //
0125 // static member functions
0126 //
0127 void WorkerMonitorThread::sig_handler(int sig, siginfo_t*, void*) {
0128   write(s_pipeWriteEnd.load(), &sig, sizeof(int) / sizeof(char));
0129   while (not s_helperThreadDone) {
0130   };
0131   signal(sig, SIG_DFL);
0132   raise(sig);
0133 }