objecter: separate locked and unlocked init/shutdown

We don't want to hold the lock while we register the admin socket commands
or else we create a lock cycle when we try to process them later.

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

src/librados/RadosClient.cc

@@ -181,12 +181,13 @@ int librados::RadosClient::connect()
   }
   messenger->set_myname(entity_name_t::CLIENT(monclient.get_global_id()));
 
+  objecter->init_unlocked();
   lock.Lock();
 
   timer.init();
 
   objecter->set_client_incarnation(0);
-  objecter->init();
+  objecter->init_locked();
   monclient.renew_subs();
 
   while (osdmap.get_epoch() == 0) {
@@ -220,11 +221,16 @@ void librados::RadosClient::shutdown()
     finisher.stop();
   }
   monclient.shutdown();
-  if (objecter && state == CONNECTED)
-    objecter->shutdown();
+  bool need_objecter = false;
+  if (objecter && state == CONNECTED) {
+    need_objecter = true;
+    objecter->shutdown_locked();
+  }
   state = DISCONNECTED;
   timer.shutdown();   // will drop+retake lock
   lock.Unlock();
+  if (need_objecter)
+    objecter->shutdown_unlocked();
   if (messenger) {
     messenger->shutdown();
     messenger->wait();

src/mds/Dumper.cc

@@ -66,8 +66,9 @@ void Dumper::init(int rank)
   client_t whoami = monc->get_global_id();
   messenger->set_myname(entity_name_t::CLIENT(whoami.v));
 
+  objecter->init_unlocked();
   lock.Lock();
-  objecter->init();
+  objecter->init_locked();
   objecter->wait_for_osd_map();
   timer.init();
   lock.Unlock();
@@ -77,7 +78,9 @@ void Dumper::shutdown()
 {
   lock.Lock();
   timer.shutdown();
+  objecter->shutdown_locked();
   lock.Unlock();
+  objecter->shutdown_unlocked();
 }
 
 void Dumper::dump(const char *dump_file)

src/mds/MDS.cc

@@ -462,6 +462,8 @@ int MDS::init(int wanted_state)
   monc->authenticate();
   monc->wait_auth_rotating(30.0);
 
+  objecter->init_unlocked();
+
   mds_lock.Lock();
   if (want_state == CEPH_MDS_STATE_DNE) {
     mds_lock.Unlock();
@@ -504,8 +506,8 @@ int MDS::init(int wanted_state)
   beacon_start();
   whoami = -1;
   messenger->set_myname(entity_name_t::MDS(whoami));
-
-  objecter->init();
+  
+  objecter->init_locked();
 
   monc->sub_want("mdsmap", 0, 0);
   monc->renew_subs();
@@ -1560,7 +1562,7 @@ void MDS::suicide()
   // shut down cache
   mdcache->shutdown();
 
-  objecter->shutdown();
+  objecter->shutdown_locked();
   
   // shut down messenger
   messenger->shutdown();

src/mds/Resetter.cc

@@ -76,8 +76,9 @@ void Resetter::init(int rank)
   client_t whoami = monc->get_global_id();
   messenger->set_myname(entity_name_t::CLIENT(whoami.v));
 
+  objecter->init_unlocked();
   lock.Lock();
-  objecter->init();
+  objecter->init_locked();
   objecter->wait_for_osd_map();
   timer.init();
   lock.Unlock();
@@ -87,7 +88,9 @@ void Resetter::shutdown()
 {
   lock.Lock();
   timer.shutdown();
+  objecter->shutdown_locked();
   lock.Unlock();
+  objecter->shutdown_unlocked();
   messenger->shutdown();
   messenger->wait();
 }

src/osdc/Objecter.cc

@@ -121,9 +121,8 @@ enum {
 
 // messages ------------------------------
 
-void Objecter::init()
+void Objecter::init_unlocked()
 {
-  assert(client_lock.is_locked());
   assert(!initialized);
 
   if (!logger) {
@@ -209,6 +208,12 @@ void Objecter::init()
     lderr(cct) << "error registering admin socket command: "
 	       << cpp_strerror(-ret) << dendl;
   }
+}
+
+void Objecter::init_locked()
+{
+  assert(client_lock.is_locked());
+  assert(!initialized);
 
   schedule_tick();
   maybe_request_map();
@@ -216,7 +221,7 @@ void Objecter::init()
   initialized = true;
 }
 
-void Objecter::shutdown() 
+void Objecter::shutdown_locked() 
 {
   assert(client_lock.is_locked());
   assert(initialized);
@@ -232,7 +237,10 @@ void Objecter::shutdown()
     timer.cancel_event(tick_event);
     tick_event = NULL;
   }
+}
 
+void Objecter::shutdown_unlocked()
+{
   if (m_request_state_hook) {
     AdminSocket* admin_socket = cct->get_admin_socket();
     admin_socket->unregister_command("objecter_requests");

src/osdc/Objecter.h

@@ -969,8 +969,10 @@ public:
     assert(!logger);
   }
 
-  void init();
-  void shutdown();
+  void init_unlocked();
+  void init_locked();
+  void shutdown_locked();
+  void shutdown_unlocked();
 
   /**
    * Tell the objecter to throttle outgoing ops according to its