From e506f896a9217324ab7a7865989f4454562aed5f Mon Sep 17 00:00:00 2001
From: Josh Durgin <josh.durgin@inktank.com>
Date: Mon, 29 Sep 2014 18:17:29 -0700
Subject: [PATCH] Objecter: check the 'initialized' atomic_t safely

shutdown() resets initialized to 0, but we can still receive messages
after this point, so fix message handlers to skip messages in this
case instead of asserting.

Also read initialized while holding Objecter::rwlock to avoid races
where e.g. handle_osd_map() checks initialized -> 1, continues,
shutdown() is called, sets initialized to 0, then handle_osd_map()
goes about its business and calls op_submit(), which would fail the
assert(initialized.read()) check. Similar races existed in other
message handlers which change Objecter state.

The Objecter is not destroyed until after its Messenger in
the MDS, OSD, and librados, so this should be safe.

Fixes: #9617
Backport: giant
Signed-off-by: Josh Durgin <josh.durgin@inktank.com>
---
 src/osdc/Objecter.cc | 40 +++++++++++++++++++++++++++++++---------
 1 file changed, 31 insertions(+), 9 deletions(-)

diff --git a/src/osdc/Objecter.cc b/src/osdc/Objecter.cc
index f3186248fd09..34858f633aa5 100644
--- a/src/osdc/Objecter.cc
+++ b/src/osdc/Objecter.cc
@@ -706,9 +706,9 @@ void Objecter::_scan_requests(OSDSession *s,
 
 void Objecter::handle_osd_map(MOSDMap *m)
 {
-  assert(initialized.read());
-
   RWLock::WLocker wl(rwlock);
+  if (!initialized.read())
+    return;
 
   assert(osdmap); 
 
@@ -1537,7 +1537,8 @@ void Objecter::tick()
   assert(rwlock.is_locked());
 
   ldout(cct, 10) << "tick" << dendl;
-  assert(initialized.read());
+  if (!initialized.read())
+    return;
 
   // we are only called by C_Tick
   assert(tick_event);
@@ -2452,7 +2453,6 @@ void Objecter::unregister_op(Op *op)
 /* This function DOES put the passed message before returning */
 void Objecter::handle_osd_op_reply(MOSDOpReply *m)
 {
-  assert(initialized.read());
   ldout(cct, 10) << "in handle_osd_op_reply" << dendl;
 
   // get pio
@@ -2461,6 +2461,11 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m)
   int osd_num = (int)m->get_source().num();
 
   RWLock::RLocker l(rwlock);
+  if (!initialized.read()) {
+    m->put();
+    return;
+  }
+
   RWLock::Context lc(rwlock, RWLock::Context::TakenForRead);
 
   map<int, OSDSession *>::iterator siter = osd_sessions.find(osd_num);
@@ -3022,9 +3027,12 @@ void Objecter::_pool_op_submit(PoolOp *op)
  */
 void Objecter::handle_pool_op_reply(MPoolOpReply *m)
 {
-  assert(initialized.read());
-
   rwlock.get_read();
+  if (!initialized.read()) {
+    rwlock.put_read();
+    m->put();
+    return;
+  }
 
   ldout(cct, 10) << "handle_pool_op_reply " << *m << dendl;
   ceph_tid_t tid = m->get_tid();
@@ -3151,11 +3159,15 @@ void Objecter::_poolstat_submit(PoolStatOp *op)
 
 void Objecter::handle_get_pool_stats_reply(MGetPoolStatsReply *m)
 {
-  assert(initialized.read());
   ldout(cct, 10) << "handle_get_pool_stats_reply " << *m << dendl;
   ceph_tid_t tid = m->get_tid();
 
   RWLock::WLocker wl(rwlock);
+  if (!initialized.read()) {
+    m->put();
+    return;
+  }
+
   map<ceph_tid_t, PoolStatOp *>::iterator iter = poolstat_ops.find(tid);
   if (iter != poolstat_ops.end()) {
     PoolStatOp *op = poolstat_ops[tid];
@@ -3254,9 +3266,11 @@ void Objecter::_fs_stats_submit(StatfsOp *op)
 
 void Objecter::handle_fs_stats_reply(MStatfsReply *m)
 {
-  assert(initialized.read());
-
   RWLock::WLocker wl(rwlock);
+  if (!initialized.read()) {
+    m->put();
+    return;
+  }
 
   ldout(cct, 10) << "handle_fs_stats_reply " << *m << dendl;
   ceph_tid_t tid = m->get_tid();
@@ -3364,6 +3378,10 @@ bool Objecter::ms_handle_reset(Connection *con)
     if (osd >= 0) {
       ldout(cct, 1) << "ms_handle_reset on osd." << osd << dendl;
       rwlock.get_write();
+      if (!initialized.read()) {
+	rwlock.put_write();
+	return false;
+      }
       map<int,OSDSession*>::iterator p = osd_sessions.find(osd);
       if (p != osd_sessions.end()) {
 	OSDSession *session = p->second;
@@ -3675,6 +3693,10 @@ void Objecter::handle_command_reply(MCommandReply *m)
   int osd_num = (int)m->get_source().num();
 
   RWLock::WLocker wl(rwlock);
+  if (!initialized.read()) {
+    m->put();
+    return;
+  }
 
   map<int, OSDSession *>::iterator siter = osd_sessions.find(osd_num);
   if (siter == osd_sessions.end()) {
-- 
2.47.3