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

   1 #include "Cluster.hpp"
   2 #include "Retry.hpp"
   3
   4 #include <algorithm>
   5 #include <sys/wait.h>
   6
   7 Cluster::Cluster(Server serv, size_t cnt)
   8 : count{cnt}
   9 , proto{move(serv)}
  10 {
  11   if (!count) {
  12     count = 1;
  13   }
  14   for (size_t i = 0; i < count; ++i) {
  15     cluster.push_back(proto);
  16   }
  17 }
  18
  19 Cluster::~Cluster() {
  20   stop();
  21   wait();
  22 }
  23
  24 const vector<Server> &Cluster::getServers() const {
  25   return cluster;
  26 }
  27
  28 bool Cluster::start() {
  29   bool started = true;
  30
  31   for (auto &server : cluster) {
  32     if (!server.start()) {
  33       started = false;
  34       continue;
  35     }
  36     pids[server.getPid()] = &server;
  37   }
  38
  39   return started;
  40 }
  41
  42 void Cluster::stop(bool graceful) {
  43   for (auto &server : cluster) {
  44     server.drain();
  45     if (graceful) {
  46       server.stop();
  47     } else {
  48       // no cookies for memcached; TERM is just too slow
  49       server.signal(SIGKILL);
  50     }
  51   }
  52 }
  53
  54 bool Cluster::isStopped() {
  55   return none_of(
  56 #if HAVE_EXECUTION
  57     execution::par,
  58 #endif
  59       cluster.begin(), cluster.end(), [](Server &s) {
  60     return s.getPid() && !s.tryWait();
  61   });
  62 }
  63
  64 bool Cluster::isListening() const {
  65   return all_of(
  66 #if HAVE_EXECUTION
  67     execution::par,
  68 #endif
  69       cluster.cbegin(), cluster.cend(), [](const Server &s) {
  70     return s.isListening();
  71   });
  72 }
  73
  74 bool Cluster::ensureListening() {
  75   if (!start()) {
  76     return false;
  77   }
  78   auto listening = all_of(
  79 #if HAVE_EXECUTION
  80     execution::par,
  81 #endif
  82   cluster.begin(), cluster.end(), [](Server &s) {
  83     return s.ensureListening();
  84   });
  85
  86   pids.clear();
  87   for (auto &server : cluster) {
  88     pids[server.getPid()] = &server;
  89   }
  90
  91   return listening;
  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 }