From: Greg Farnum <greg@inktank.com>
Date: Mon, 14 Apr 2014 17:47:16 +0000 (-0700)
Subject: OSD: protect state member with a Spinlock
X-Git-Tag: v0.81~57^2~28
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=2f97f4776fc4b07303a69595e080bd5d551dc17d;p=ceph.git

OSD: protect state member with a Spinlock

This member was previously protected by the osd_lock (although setting
SHUTDOWN was synchronized with the heartbeat lock, too), but we need
to read it for fast dispatch, so protect it under its own lock at all times.

Signed-off-by: Greg Farnum <greg@inktank.com>
---

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 2f351e5cb4b..50b077f1c11 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -915,7 +915,8 @@ OSD::OSD(CephContext *cct_, ObjectStore *store_,
   dispatch_running(false),
   asok_hook(NULL),
   osd_compat(get_osd_compat_set()),
-  state(STATE_INITIALIZING), epoch_lock(), boot_epoch(0), up_epoch(0), bind_epoch(0),
+  state_lock(), state(STATE_INITIALIZING),
+  epoch_lock(), boot_epoch(0), up_epoch(0), bind_epoch(0),
   op_tp(cct, "OSD::op_tp", cct->_conf->osd_op_threads, "osd_op_threads"),
   recovery_tp(cct, "OSD::recovery_tp", cct->_conf->osd_recovery_threads, "osd_recovery_threads"),
   disk_tp(cct, "OSD::disk_tp", cct->_conf->osd_disk_threads, "osd_disk_threads"),
@@ -1031,7 +1032,7 @@ bool OSD::asok_command(string command, cmdmap_t& cmdmap, string format,
     f->dump_stream("cluster_fsid") << superblock.cluster_fsid;
     f->dump_stream("osd_fsid") << superblock.osd_fsid;
     f->dump_unsigned("whoami", superblock.whoami);
-    f->dump_string("state", get_state_name(state));
+    f->dump_string("state", get_state_name(get_state()));
     f->dump_unsigned("oldest_map", superblock.oldest_map);
     f->dump_unsigned("newest_map", superblock.newest_map);
     {
@@ -1312,7 +1313,7 @@ int OSD::init()
   peering_wq.drain();
 
   dout(0) << "done with init, starting boot process" << dendl;
-  state = STATE_BOOTING;
+  set_state(STATE_BOOTING);
   start_boot();
 
   return 0;
@@ -1600,9 +1601,7 @@ int OSD::shutdown()
   }
   derr << "shutdown" << dendl;
 
-  heartbeat_lock.Lock();
-  state = STATE_STOPPING;
-  heartbeat_lock.Unlock();
+  set_state(STATE_STOPPING);
 
   // Debugging
   cct->_conf->set_val("debug_osd", "100");
@@ -3289,7 +3288,7 @@ void OSD::tick()
   if (is_waiting_for_healthy()) {
     if (_is_healthy()) {
       dout(1) << "healthy again, booting" << dendl;
-      state = STATE_BOOTING;
+      set_state(STATE_BOOTING);
       start_boot();
     }
   }
@@ -3689,7 +3688,7 @@ void OSD::_maybe_boot(epoch_t oldest, epoch_t newest)
 void OSD::start_waiting_for_healthy()
 {
   dout(1) << "start_waiting_for_healthy" << dendl;
-  state = STATE_WAITING_FOR_HEALTHY;
+  set_state(STATE_WAITING_FOR_HEALTHY);
   last_heartbeat_resample = utime_t();
 }
 
@@ -5585,7 +5584,7 @@ void OSD::handle_osd_map(MOSDMap *m)
 
     if (is_booting()) {
       dout(1) << "state: booting -> active" << dendl;
-      state = STATE_ACTIVE;
+      set_state(STATE_ACTIVE);
 
       // set incarnation so that osd_reqid_t's we generate for our
       // objecter requests are unique across restarts.
@@ -5596,7 +5595,7 @@ void OSD::handle_osd_map(MOSDMap *m)
   bool do_shutdown = false;
   bool do_restart = false;
   if (osdmap->get_epoch() > 0 &&
-      state == STATE_ACTIVE) {
+      is_active()) {
     if (!osdmap->exists(whoami)) {
       dout(0) << "map says i do not exist.  shutting down." << dendl;
       do_shutdown = true;   // don't call shutdown() while we have everything paused
diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index ba9f2295870..b09d8a795d9 100644
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -961,6 +961,7 @@ public:
   }
 
 private:
+  Spinlock state_lock; // protects access to state
   int state;
   Spinlock epoch_lock; // protects access to boot_epoch, up_epoch, bind_epoch
   epoch_t boot_epoch;  // _first_ epoch we were marked up (after this process started)
@@ -968,11 +969,28 @@ private:
   epoch_t bind_epoch;  // epoch we last did a bind to new ip:ports
 
 public:
-  bool is_initializing() { return state == STATE_INITIALIZING; }
-  bool is_booting() { return state == STATE_BOOTING; }
-  bool is_active() { return state == STATE_ACTIVE; }
-  bool is_stopping() { return state == STATE_STOPPING; }
-  bool is_waiting_for_healthy() { return state == STATE_WAITING_FOR_HEALTHY; }
+  int get_state() { Spinlock::Locker l(state_lock); return state; }
+  void set_state(int s) { Spinlock::Locker l(state_lock); state = s; }
+  bool is_initializing() {
+    Spinlock::Locker l(state_lock);
+    return state == STATE_INITIALIZING;
+  }
+  bool is_booting() {
+    Spinlock::Locker l(state_lock);
+    return state == STATE_BOOTING;
+  }
+  bool is_active() {
+    Spinlock::Locker l(state_lock);
+    return state == STATE_ACTIVE;
+  }
+  bool is_stopping() {
+    Spinlock::Locker l(state_lock);
+    return state == STATE_STOPPING;
+  }
+  bool is_waiting_for_healthy() {
+    Spinlock::Locker l(state_lock);
+    return state == STATE_WAITING_FOR_HEALTHY;
+  }
 
 private: