heartbeatmap: introduce heartbeat_map

Each thread registered and gets a private structure it can write a timeout
value to.  The timeout is time_t and always fits in a single word, so no
locking is used to update it.

Anyone can call is_healthy() to find out if any timeouts have expired.
Eventually some background thread will do this.

Signed-off-by: Sage Weil <sage.weil@dreamhost.com>

diff --git a/src/Makefile.am b/src/Makefile.am
index 7e6f93b6c4aa34b79b67d02d9cedecf035f1aaba..e1f47c1eaf99e54de25ec85aedfed236396650cc 100644 (file)
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -726,6 +726,7 @@ libcommon_files = \
        common/simple_spin.cc \
        common/Thread.cc \
        common/Formatter.cc \
+       common/HeartbeatMap.cc \
        include/ceph_fs.cc \
        include/ceph_hash.cc \
        include/ceph_strings.cc \

diff --git a/src/common/HeartbeatMap.cc b/src/common/HeartbeatMap.cc
new file mode 100644 (file)
index 0000000..b5c20f3
--- /dev/null
+++ b/src/common/HeartbeatMap.cc
@@ -0,0 +1,80 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#include <time.h>
+
+#include "HeartbeatMap.h"
+#include "ceph_context.h"
+
+#include "debug.h"
+#define DOUT_SUBSYS heartbeatmap
+#undef dout_prefix
+#define dout_prefix *_dout << "heartbeat_map "
+
+
+HeartbeatMap::HeartbeatMap(CephContext *cct)
+  : m_cct(cct),
+    m_rwlock("HeartbeatMap::m_rwlock")
+{
+}
+
+HeartbeatMap::~HeartbeatMap()
+{
+}
+
+heartbeat_handle_d *HeartbeatMap::add_worker(pthread_t thread, string name)
+{
+  m_rwlock.get_write();
+  ldout(m_cct, 10) << "add_worker " << thread << " '" << name << "'" << dendl;
+  assert(m_workers.count(thread) == 0);
+  heartbeat_handle_d *h = new heartbeat_handle_d(thread, name);
+  m_workers[thread] = h;
+  m_rwlock.put_write();
+  return h;
+}
+
+void HeartbeatMap::remove_worker(heartbeat_handle_d *h)
+{
+  m_rwlock.get_write();
+  ldout(m_cct, 10) << "remove_worker " << h->thread << " '" << h->name << "'" << dendl;
+  map<pthread_t, heartbeat_handle_d*>::iterator p = m_workers.find(h->thread);
+  assert(p != m_workers.end());
+  m_workers.erase(p);
+  m_rwlock.put_write();
+  delete h;
+}
+
+void HeartbeatMap::touch_worker(heartbeat_handle_d *h, time_t grace)
+{
+  ldout(m_cct, 20) << "touch_worker " << h->thread << " grace " << grace << dendl;
+  h->timeout = time(NULL) + grace;
+}
+
+bool HeartbeatMap::is_healthy()
+{
+  m_rwlock.get_read();
+  time_t now = time(NULL);
+  bool healthy = true;
+  for (map<pthread_t, heartbeat_handle_d*>::iterator p = m_workers.begin();
+       p != m_workers.end();
+       ++p)
+    if (p->second->timeout && p->second->timeout < now) {
+      ldout(m_cct, 0) << "is_healthy " << p->first << " '" << p->second->name << "'"
+                     << " timed out" << dendl;
+      healthy = false;
+    }
+  m_rwlock.put_read();
+  return healthy;
+}
+

diff --git a/src/common/HeartbeatMap.h b/src/common/HeartbeatMap.h
new file mode 100644 (file)
index 0000000..6425fad
--- /dev/null
+++ b/src/common/HeartbeatMap.h
@@ -0,0 +1,55 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_HEARTBEATMAP_H
+#define CEPH_HEARTBEATMAP_H
+
+#include <pthread.h>
+
+#include <string>
+#include <map>
+
+#include "RWLock.h"
+
+class CephContext;
+
+struct heartbeat_handle_d {
+  pthread_t thread;
+  std::string name;
+  time_t timeout;
+
+  heartbeat_handle_d(pthread_t t, const std::string& n)
+    : thread(t), name(n),
+      timeout(0)
+  { }
+};
+
+class HeartbeatMap {
+ public:
+  heartbeat_handle_d *add_worker(pthread_t thread, std::string name);
+  void remove_worker(heartbeat_handle_d *h);
+  void touch_worker(heartbeat_handle_d *h, time_t grace);
+
+  bool is_healthy();
+
+  HeartbeatMap(CephContext *cct);
+  ~HeartbeatMap();
+
+ private:
+  CephContext *m_cct;
+  RWLock m_rwlock;
+  std::map<unsigned long, heartbeat_handle_d*> m_workers;
+};
+
+#endif

diff --git a/src/common/ceph_context.cc b/src/common/ceph_context.cc
index 669b9b6d93ec60d5062a803025a530d9bd4c1563..ff41516188ba0ac338f7ac9f95bc9f1b49151bff 100644 (file)
--- a/src/common/ceph_context.cc
+++ b/src/common/ceph_context.cc
@@ -19,6 +19,7 @@
 #include "common/ceph_context.h"
 #include "common/config.h"
 #include "common/debug.h"
+#include "common/HeartbeatMap.h"
 
 #include <iostream>
 #include <pthread.h>
@@ -80,13 +81,15 @@ CephContext(uint32_t module_type_)
     _module_type(module_type_),
     _service_thread(NULL),
     _admin_socket_config_obs(NULL),
-    _perf_counters_collection(NULL)
+    _perf_counters_collection(NULL),
+    _heartbeat_map(NULL)
 {
   pthread_spin_init(&_service_thread_lock, PTHREAD_PROCESS_SHARED);
   _perf_counters_collection = new PerfCountersCollection(this);
   _conf->add_observer(_doss);
   _admin_socket_config_obs = new AdminSocketConfigObs(this);
   _conf->add_observer(_admin_socket_config_obs);
+  _heartbeat_map = new HeartbeatMap(this);
 }
 
 CephContext::
@@ -94,6 +97,8 @@ CephContext::
 {
   join_service_thread();
 
+  delete _heartbeat_map;
+
   _conf->remove_observer(_admin_socket_config_obs);
   _conf->remove_observer(_doss);
 

diff --git a/src/common/ceph_context.h b/src/common/ceph_context.h
index 957ab225a3fc73d591834fbcdfe11f910081030d..ac8c39b7d0991c8965c5abea0f9e588cc7502e9b 100644 (file)
--- a/src/common/ceph_context.h
+++ b/src/common/ceph_context.h
@@ -28,6 +28,7 @@ class DoutLocker;
 class PerfCountersCollection;
 class md_config_obs_t;
 class md_config_t;
+class HeartbeatMap;
 
 /* A CephContext represents the context held by a single library user.
  * There can be multiple CephContexts in the same process.
@@ -62,6 +63,10 @@ public:
   /* Get the PerfCountersCollection of this CephContext */
   PerfCountersCollection *GetPerfCountersCollection();
 
+  HeartbeatMap *get_heartbeat_map() {
+    return _heartbeat_map;
+  }
+
 private:
   CephContext(const CephContext &rhs);
   CephContext &operator=(const CephContext &rhs);
@@ -86,6 +91,8 @@ private:
   PerfCountersCollection *_perf_counters_collection;
 
   md_config_obs_t *_perf_counters_conf_obs;
+
+  HeartbeatMap *_heartbeat_map;
 };
 
 #endif

diff --git a/src/common/config.cc b/src/common/config.cc
index c74f0b303b25916277a1248b8e08c14e98df49f2..5d5748e69b65bc92978f5e14d39b97cfca6a91ad 100644 (file)
--- a/src/common/config.cc
+++ b/src/common/config.cc
@@ -173,6 +173,7 @@ struct config_option config_optionsp[] = {
   OPTION(debug_tp, OPT_INT, 0),
   OPTION(debug_auth, OPT_INT, 1),
   OPTION(debug_finisher, OPT_INT, 1),
+  OPTION(debug_heartbeatmap, OPT_INT, 1),
   OPTION(key, OPT_STR, 0),
   OPTION(keyfile, OPT_STR, 0),
   OPTION(keyring, OPT_STR, "/etc/ceph/keyring,/etc/ceph/keyring.bin"),

diff --git a/src/common/config.h b/src/common/config.h
index a2d3ef37416410119471f9de87a9fdfcf7ffc41e..6161cc7e24e71cbe71fd4b03d41d43986a873442 100644 (file)
--- a/src/common/config.h
+++ b/src/common/config.h
@@ -187,6 +187,7 @@ public:
   int debug_tp;
   int debug_auth;
   int debug_finisher;
+  int debug_heartbeatmap;
 
   // auth
   std::string key;