int OSD::shutdown()
{
  if (!service.prepare_to_stop())
    return 0; // already shutting down
  osd_lock.Lock();
  if (is_stopping()) {
    osd_lock.Unlock();
    return 0;
  }
  derr << "shutdown" << dendl;

  set_state(STATE_STOPPING);

  // Debugging
  if (cct->_conf->get_val<bool>("osd_debug_shutdown")) {
    cct->_conf->set_val("debug_osd", "100");
    cct->_conf->set_val("debug_journal", "100");
    cct->_conf->set_val("debug_filestore", "100");
    cct->_conf->set_val("debug_bluestore", "100");
    cct->_conf->set_val("debug_ms", "100");
    cct->_conf->apply_changes(NULL);
  }

  // stop MgrClient earlier as it's more like an internal consumer of OSD
  mgrc.shutdown();

  service.start_shutdown();

  // stop sending work to pgs.  this just prevents any new work in _process
  // from racing with on_shutdown and potentially entering the pg after.
  op_shardedwq.drain();

  // Shutdown PGs
  {
    RWLock::RLocker l(pg_map_lock);
    for (ceph::unordered_map<spg_t, PG*>::iterator p = pg_map.begin();
        p != pg_map.end();
        ++p) {
      dout(20) << " kicking pg " << p->first << dendl;
      p->second->lock();
      p->second->on_shutdown();
      p->second->unlock();
      p->second->osr->flush();
    }
  }
  clear_pg_stat_queue();

  // drain op queue again (in case PGs requeued something)
  op_shardedwq.drain();
  {
    finished.clear(); // zap waiters (bleh, this is messy)
  }

  op_shardedwq.clear_pg_slots();

  // unregister commands
  cct->get_admin_socket()->unregister_command("status");
  cct->get_admin_socket()->unregister_command("flush_journal");
  cct->get_admin_socket()->unregister_command("dump_ops_in_flight");
  cct->get_admin_socket()->unregister_command("ops");
  cct->get_admin_socket()->unregister_command("dump_blocked_ops");
  cct->get_admin_socket()->unregister_command("dump_historic_ops");
  cct->get_admin_socket()->unregister_command("dump_historic_ops_by_duration");
  cct->get_admin_socket()->unregister_command("dump_historic_slow_ops");
  cct->get_admin_socket()->unregister_command("dump_op_pq_state");
  cct->get_admin_socket()->unregister_command("dump_blacklist");
  cct->get_admin_socket()->unregister_command("dump_watchers");
  cct->get_admin_socket()->unregister_command("dump_reservations");
  cct->get_admin_socket()->unregister_command("get_latest_osdmap");
  cct->get_admin_socket()->unregister_command("heap");
  cct->get_admin_socket()->unregister_command("set_heap_property");
  cct->get_admin_socket()->unregister_command("get_heap_property");
  cct->get_admin_socket()->unregister_command("dump_objectstore_kv_stats");
  cct->get_admin_socket()->unregister_command("dump_scrubs");
  cct->get_admin_socket()->unregister_command("calc_objectstore_db_histogram");
  cct->get_admin_socket()->unregister_command("flush_store_cache");
  cct->get_admin_socket()->unregister_command("dump_pgstate_history");
  cct->get_admin_socket()->unregister_command("compact");
  delete asok_hook;
  asok_hook = NULL;

  cct->get_admin_socket()->unregister_command("setomapval");
  cct->get_admin_socket()->unregister_command("rmomapkey");
  cct->get_admin_socket()->unregister_command("setomapheader");
  cct->get_admin_socket()->unregister_command("getomap");
  cct->get_admin_socket()->unregister_command("truncobj");
  cct->get_admin_socket()->unregister_command("injectdataerr");
  cct->get_admin_socket()->unregister_command("injectmdataerr");
  cct->get_admin_socket()->unregister_command("set_recovery_delay");
  cct->get_admin_socket()->unregister_command("trigger_scrub");
  cct->get_admin_socket()->unregister_command("injectfull");
  delete test_ops_hook;
  test_ops_hook = NULL;

  osd_lock.Unlock();

  heartbeat_lock.Lock();
  heartbeat_stop = true;
  heartbeat_cond.Signal();
  heartbeat_lock.Unlock();
  heartbeat_thread.join();

  peering_tp.drain();
  peering_wq.clear();
  peering_tp.stop();
  dout(10) << "osd tp stopped" << dendl;

  osd_op_tp.drain();
  osd_op_tp.stop();
  dout(10) << "op sharded tp stopped" << dendl;

  command_tp.drain();
  command_tp.stop();
  dout(10) << "command tp stopped" << dendl;

  disk_tp.drain();
  disk_tp.stop();
  dout(10) << "disk tp paused (new)" << dendl;

  dout(10) << "stopping agent" << dendl;
  service.agent_stop();

  osd_lock.Lock();

  reset_heartbeat_peers();

  tick_timer.shutdown();

  {
    Mutex::Locker l(tick_timer_lock);
    tick_timer_without_osd_lock.shutdown();
  }

  // note unmount epoch
  dout(10) << "noting clean unmount in epoch " << osdmap->get_epoch() << dendl;
  superblock.mounted = service.get_boot_epoch();
  superblock.clean_thru = osdmap->get_epoch();
  ObjectStore::Transaction t;
  write_superblock(t);
  int r = store->apply_transaction(service.meta_osr.get(), std::move(t));
  if (r) {
    derr << "OSD::shutdown: error writing superblock: "
	 << cpp_strerror(r) << dendl;
  }


  {
    Mutex::Locker l(pg_stat_queue_lock);
    assert(pg_stat_queue.empty());
  }

  service.shutdown_reserver();

  // Remove PGs
#ifdef PG_DEBUG_REFS
  service.dump_live_pgids();
#endif
  {
    RWLock::RLocker l(pg_map_lock);
    for (ceph::unordered_map<spg_t, PG*>::iterator p = pg_map.begin();
        p != pg_map.end();
        ++p) {
      dout(20) << " kicking pg " << p->first << dendl;
      p->second->lock();
      if (p->second->ref != 1) {
        derr << "pgid " << p->first << " has ref count of "
            << p->second->ref << dendl;
#ifdef PG_DEBUG_REFS
	p->second->dump_live_ids();
#endif
	if (cct->_conf->osd_shutdown_pgref_assert) {
	  ceph_abort();
	}
      }
      p->second->unlock();
      p->second->put("PGMap");
    }
    pg_map.clear();
  }
#ifdef PG_DEBUG_REFS
  service.dump_live_pgids();
#endif
  cct->_conf->remove_observer(this);

  dout(10) << "syncing store" << dendl;
  enable_disable_fuse(true);

  if (cct->_conf->osd_journal_flush_on_shutdown) {
    dout(10) << "flushing journal" << dendl;
    store->flush_journal();
  }

  store->umount();
  delete store;
  store = 0;
  dout(10) << "Store synced" << dendl;

  monc->shutdown();
  osd_lock.Unlock();

  osdmap = OSDMapRef();
  service.shutdown();
  op_tracker.on_shutdown();

  class_handler->shutdown();
  client_messenger->shutdown();
  cluster_messenger->shutdown();
  hb_front_client_messenger->shutdown();
  hb_back_client_messenger->shutdown();
  objecter_messenger->shutdown();
  hb_front_server_messenger->shutdown();
  hb_back_server_messenger->shutdown();

  peering_wq.clear();

  return r;
}