ceph/src/common/HeartbeatMap.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2011 Sage Weil <sage@newdream.net>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #include <signal.h>
  16
  17 #include "HeartbeatMap.h"
  18 #include "ceph_context.h"
  19 #include "common/errno.h"
  20 #include "debug.h"
  21
  22 #define dout_subsys ceph_subsys_heartbeatmap
  23 #undef dout_prefix
  24 #define dout_prefix *_dout << "heartbeat_map "
  25
  26 namespace ceph {
  27
  28 HeartbeatMap::HeartbeatMap(CephContext *cct)
  29   : m_cct(cct),
  30     m_rwlock("HeartbeatMap::m_rwlock"),
  31     m_inject_unhealthy_until(0),
  32     m_unhealthy_workers(0),
  33     m_total_workers(0)
  34 {
  35 }
  36
  37 HeartbeatMap::~HeartbeatMap()
  38 {
  39   assert(m_workers.empty());
  40 }
  41
  42 heartbeat_handle_d *HeartbeatMap::add_worker(const string& name, pthread_t thread_id)
  43 {
  44   m_rwlock.get_write();
  45   ldout(m_cct, 10) << "add_worker '" << name << "'" << dendl;
  46   heartbeat_handle_d *h = new heartbeat_handle_d(name);
  47   ANNOTATE_BENIGN_RACE_SIZED(&h->timeout, sizeof(h->timeout),
  48                              "heartbeat_handle_d timeout");
  49   ANNOTATE_BENIGN_RACE_SIZED(&h->suicide_timeout, sizeof(h->suicide_timeout),
  50                              "heartbeat_handle_d suicide_timeout");
  51   m_workers.push_front(h);
  52   h->list_item = m_workers.begin();
  53   h->thread_id = thread_id;
  54   m_rwlock.put_write();
  55   return h;
  56 }
  57
  58 void HeartbeatMap::remove_worker(const heartbeat_handle_d *h)
  59 {
  60   m_rwlock.get_write();
  61   ldout(m_cct, 10) << "remove_worker '" << h->name << "'" << dendl;
  62   m_workers.erase(h->list_item);
  63   m_rwlock.put_write();
  64   delete h;
  65 }
  66
  67 bool HeartbeatMap::_check(const heartbeat_handle_d *h, const char *who, time_t now)
  68 {
  69   bool healthy = true;
  70   time_t was;
  71
  72   was = h->timeout;
  73   if (was && was < now) {
  74     ldout(m_cct, 1) << who << " '" << h->name << "'"
  75                     << " had timed out after " << h->grace << dendl;
  76     healthy = false;
  77   }
  78   was = h->suicide_timeout;
  79   if (was && was < now) {
  80     ldout(m_cct, 1) << who << " '" << h->name << "'"
  81                     << " had suicide timed out after " << h->suicide_grace << dendl;
  82     pthread_kill(h->thread_id, SIGABRT);
  83     sleep(1);
  84     assert(0 == "hit suicide timeout");
  85   }
  86   return healthy;
  87 }
  88
  89 void HeartbeatMap::reset_timeout(heartbeat_handle_d *h, time_t grace, time_t suicide_grace)
  90 {
  91   ldout(m_cct, 20) << "reset_timeout '" << h->name << "' grace " << grace
  92                    << " suicide " << suicide_grace << dendl;
  93   time_t now = time(NULL);
  94   _check(h, "reset_timeout", now);
  95
  96   h->timeout = now + grace;
  97   h->grace = grace;
  98
  99   if (suicide_grace)
 100     h->suicide_timeout = now + suicide_grace;
 101   else
 102     h->suicide_timeout = 0;
 103   h->suicide_grace = suicide_grace;
 104 }
 105
 106 void HeartbeatMap::clear_timeout(heartbeat_handle_d *h)
 107 {
 108   ldout(m_cct, 20) << "clear_timeout '" << h->name << "'" << dendl;
 109   time_t now = time(NULL);
 110   _check(h, "clear_timeout", now);
 111   h->timeout = 0;
 112   h->suicide_timeout = 0;
 113 }
 114
 115 bool HeartbeatMap::is_healthy()
 116 {
 117   int unhealthy = 0;
 118   int total = 0;
 119   m_rwlock.get_read();
 120   time_t now = time(NULL);
 121   if (m_cct->_conf->heartbeat_inject_failure) {
 122     ldout(m_cct, 0) << "is_healthy injecting failure for next " << m_cct->_conf->heartbeat_inject_failure << " seconds" << dendl;
 123     m_inject_unhealthy_until = now + m_cct->_conf->heartbeat_inject_failure;
 124     m_cct->_conf->set_val("heartbeat_inject_failure", "0");
 125   }
 126
 127   bool healthy = true;
 128   if (now < m_inject_unhealthy_until) {
 129     ldout(m_cct, 0) << "is_healthy = false, injected failure for next " << (m_inject_unhealthy_until - now) << " seconds" << dendl;
 130     healthy = false;
 131   }
 132
 133   for (list<heartbeat_handle_d*>::iterator p = m_workers.begin();
 134        p != m_workers.end();
 135        ++p) {
 136     heartbeat_handle_d *h = *p;
 137     if (!_check(h, "is_healthy", now)) {
 138       healthy = false;
 139       unhealthy++;
 140     }
 141     total++;
 142   }
 143   m_rwlock.put_read();
 144
 145   m_unhealthy_workers = unhealthy;
 146   m_total_workers = total;
 147
 148   ldout(m_cct, 20) << "is_healthy = " << (healthy ? "healthy" : "NOT HEALTHY")
 149     << ", total workers: " << total << ", number of unhealthy: " << unhealthy << dendl;
 150   return healthy;
 151 }
 152
 153 int HeartbeatMap::get_unhealthy_workers() const
 154 {
 155   return m_unhealthy_workers;
 156 }
 157
 158 int HeartbeatMap::get_total_workers() const
 159 {
 160   return m_total_workers;
 161 }
 162
 163 void HeartbeatMap::check_touch_file()
 164 {
 165   if (is_healthy()) {
 166     string path = m_cct->_conf->heartbeat_file;
 167     if (path.length()) {
 168       int fd = ::open(path.c_str(), O_WRONLY|O_CREAT|O_CLOEXEC, 0644);
 169       if (fd >= 0) {
 170         ::utimes(path.c_str(), NULL);
 171         ::close(fd);
 172       } else {
 173         ldout(m_cct, 0) << "unable to touch " << path << ": "
 174                         << cpp_strerror(errno) << dendl;
 175       }
 176     }
 177   }
 178 }
 179
 180 }