git.m6w6.name Git - awesomized/libmemcached/blob - test/lib/Cluster.cpp

   1 #include "Cluster.hpp"
   2 #include "Retry.hpp"
   3
   4 #include <sys/wait.h>
   5
   6 Cluster::Cluster(Server serv, uint16_t cnt)
   7 : count{cnt}
   8 , proto{move(serv)}
   9 {
  10   if (!count) {
  11     count = 1;
  12   }
  13   for (int i = 0; i < count; ++i) {
  14     cluster.push_back(proto);
  15   }
  16 }
  17
  18 Cluster::~Cluster() {
  19   stop();
  20   wait();
  21 }
  22
  23 const vector<Server> &Cluster::getServers() const {
  24   return cluster;
  25 }
  26
  27 bool Cluster::start() {
  28   bool started = true;
  29
  30   for (auto &server : cluster) {
  31     if (!startServer(server)) {
  32       started = false;
  33     }
  34   }
  35
  36   return started;
  37 }
  38
  39 void Cluster::stop(bool graceful) {
  40   for (auto &server : cluster) {
  41     server.drain();
  42     if (graceful) {
  43       server.stop();
  44     } else {
  45       // no cookies for memcached; TERM is just too slow
  46       server.signal(SIGKILL);
  47     }
  48   }
  49 }
  50
  51 bool Cluster::isStopped() {
  52   for (auto &server : cluster) {
  53     if (server.getPid() && !server.tryWait()) {
  54       return false;
  55     }
  56   }
  57   return true;
  58 }
  59
  60 bool Cluster::isListening() {
  61   for (auto &server : cluster) {
  62     Retry server_is_listening{[&] {
  63       if (!server.isListening()) {
  64         // zombie?
  65         auto old_pid = server.getPid();
  66         if (server.tryWait()) {
  67           cerr << "Collected zombie " << server << "(old pid=" << old_pid << ")\n";
  68           pids.erase(old_pid);
  69           // restart
  70           startServer(server);
  71         }
  72         if (!server.isListening()) {
  73           return false;
  74         }
  75       }
  76       return true;
  77     }};
  78     if (!server_is_listening()) {
  79       return false;
  80     }
  81   }
  82
  83   return true;
  84 }
  85
  86 bool Cluster::startServer(Server &server) {
  87   if (server.start().has_value()) {
  88     pids[server.getPid()] = &server;
  89     return true;
  90   }
  91   return false;
  92 }
  93
  94 void Cluster::wait() {
  95   siginfo_t inf;
  96
  97   while (!isStopped()) {
  98     if (waitid(P_ALL, 0, &inf, WEXITED | WNOWAIT)) {
  99       perror("Cluster::wait waitid()");
 100       return;
 101     }
 102
 103     auto server = pids.find(inf.si_pid);
 104     if (server != pids.end()) {
 105       server->second->wait();
 106     }
 107   }
 108 }
 109
 110 bool Cluster::restart() {
 111   stop();
 112   wait();
 113   return start();
 114 }