int OSD::init()
{
  CompatSet initial, diff;
  Mutex::Locker lock(osd_lock);
  if (is_stopping())
    return 0;

  tick_timer.init();
  tick_timer_without_osd_lock.init();
  service.recovery_request_timer.init();
  service.recovery_sleep_timer.init();

  // mount.
  dout(2) << "init " << dev_path
	  << " (looks like " << (store_is_rotational ? "hdd" : "ssd") << ")"
	  << dendl;
  dout(2) << "journal " << journal_path << dendl;
  assert(store);  // call pre_init() first!

  store->set_cache_shards(get_num_op_shards());

  int r = store->mount();
  if (r < 0) {
    derr << "OSD:init: unable to mount object store" << dendl;
    return r;
  }
  journal_is_rotational = store->is_journal_rotational();
  dout(2) << "journal looks like " << (journal_is_rotational ? "hdd" : "ssd")
          << dendl;

  enable_disable_fuse(false);

  dout(2) << "boot" << dendl;

  // initialize the daily loadavg with current 15min loadavg
  double loadavgs[3];
  if (getloadavg(loadavgs, 3) == 3) {
    daily_loadavg = loadavgs[2];
  } else {
    derr << "OSD::init() : couldn't read loadavgs\n" << dendl;
    daily_loadavg = 1.0;
  }

  int rotating_auth_attempts = 0;

  // sanity check long object name handling
  {
    hobject_t l;
    l.oid.name = string(cct->_conf->osd_max_object_name_len, 'n');
    l.set_key(string(cct->_conf->osd_max_object_name_len, 'k'));
    l.nspace = string(cct->_conf->osd_max_object_namespace_len, 's');
    r = store->validate_hobject_key(l);
    if (r < 0) {
      derr << "backend (" << store->get_type() << ") is unable to support max "
	   << "object name[space] len" << dendl;
      derr << "   osd max object name len = "
	   << cct->_conf->osd_max_object_name_len << dendl;
      derr << "   osd max object namespace len = "
	   << cct->_conf->osd_max_object_namespace_len << dendl;
      derr << cpp_strerror(r) << dendl;
      if (cct->_conf->osd_check_max_object_name_len_on_startup) {
	goto out;
      }
      derr << "osd_check_max_object_name_len_on_startup = false, starting anyway"
	   << dendl;
    } else {
      dout(20) << "configured osd_max_object_name[space]_len looks ok" << dendl;
    }
  }

  // read superblock
  r = read_superblock();
  if (r < 0) {
    derr << "OSD::init() : unable to read osd superblock" << dendl;
    r = -EINVAL;
    goto out;
  }

  if (osd_compat.compare(superblock.compat_features) < 0) {
    derr << "The disk uses features unsupported by the executable." << dendl;
    derr << " ondisk features " << superblock.compat_features << dendl;
    derr << " daemon features " << osd_compat << dendl;

    if (osd_compat.writeable(superblock.compat_features)) {
      CompatSet diff = osd_compat.unsupported(superblock.compat_features);
      derr << "it is still writeable, though. Missing features: " << diff << dendl;
      r = -EOPNOTSUPP;
      goto out;
    }
    else {
      CompatSet diff = osd_compat.unsupported(superblock.compat_features);
      derr << "Cannot write to disk! Missing features: " << diff << dendl;
      r = -EOPNOTSUPP;
      goto out;
    }
  }

  assert_warn(whoami == superblock.whoami);
  if (whoami != superblock.whoami) {
    derr << "OSD::init: superblock says osd"
	 << superblock.whoami << " but I am osd." << whoami << dendl;
    r = -EINVAL;
    goto out;
  }

  initial = get_osd_initial_compat_set();
  diff = superblock.compat_features.unsupported(initial);
  if (superblock.compat_features.merge(initial)) {
    // We need to persist the new compat_set before we
    // do anything else
    dout(5) << "Upgrading superblock adding: " << diff << dendl;
    ObjectStore::Transaction t;
    write_superblock(t);
    r = store->apply_transaction(service.meta_osr.get(), std::move(t));
    if (r < 0)
      goto out;
  }

  // make sure snap mapper object exists
  if (!store->exists(coll_t::meta(), OSD::make_snapmapper_oid())) {
    dout(10) << "init creating/touching snapmapper object" << dendl;
    ObjectStore::Transaction t;
    t.touch(coll_t::meta(), OSD::make_snapmapper_oid());
    r = store->apply_transaction(service.meta_osr.get(), std::move(t));
    if (r < 0)
      goto out;
  }

  class_handler = new ClassHandler(cct);
  cls_initialize(class_handler);

  if (cct->_conf->osd_open_classes_on_start) {
    int r = class_handler->open_all_classes();
    if (r)
      dout(1) << "warning: got an error loading one or more classes: " << cpp_strerror(r) << dendl;
  }

  // load up "current" osdmap
  assert_warn(!osdmap);
  if (osdmap) {
    derr << "OSD::init: unable to read current osdmap" << dendl;
    r = -EINVAL;
    goto out;
  }
  osdmap = get_map(superblock.current_epoch);
  check_osdmap_features(store);

  create_recoverystate_perf();

  {
    epoch_t bind_epoch = osdmap->get_epoch();
    service.set_epochs(NULL, NULL, &bind_epoch);
  }

  clear_temp_objects();

  // initialize osdmap references in sharded wq
  op_shardedwq.prune_pg_waiters(osdmap, whoami);

  // load up pgs (as they previously existed)
  load_pgs();

  dout(2) << "superblock: I am osd." << superblock.whoami << dendl;
  dout(0) << "using " << op_queue << " op queue with priority op cut off at " <<
    op_prio_cutoff << "." << dendl;

  create_logger();

  // i'm ready!
  client_messenger->add_dispatcher_head(this);
  cluster_messenger->add_dispatcher_head(this);

  hb_front_client_messenger->add_dispatcher_head(&heartbeat_dispatcher);
  hb_back_client_messenger->add_dispatcher_head(&heartbeat_dispatcher);
  hb_front_server_messenger->add_dispatcher_head(&heartbeat_dispatcher);
  hb_back_server_messenger->add_dispatcher_head(&heartbeat_dispatcher);

  objecter_messenger->add_dispatcher_head(service.objecter);

  monc->set_want_keys(CEPH_ENTITY_TYPE_MON | CEPH_ENTITY_TYPE_OSD
                      | CEPH_ENTITY_TYPE_MGR);
  r = monc->init();
  if (r < 0)
    goto out;

  /**
   * FIXME: this is a placeholder implementation that unconditionally
   * sends every is_primary PG's stats every time we're called, unlike
   * the existing mon PGStats mechanism that uses pg_stat_queue and acks.
   * This has equivalent cost to the existing worst case where all
   * PGs are busy and their stats are always enqueued for sending.
   */
  mgrc.set_pgstats_cb([this](){
      RWLock::RLocker l(map_lock);
      
      utime_t had_for = ceph_clock_now() - had_map_since;
      osd_stat_t cur_stat = service.get_osd_stat();
      cur_stat.os_perf_stat = store->get_cur_stats();

      MPGStats *m = new MPGStats(monc->get_fsid(), osdmap->get_epoch(), had_for);
      m->osd_stat = cur_stat;

      Mutex::Locker lec{min_last_epoch_clean_lock};
      min_last_epoch_clean = osdmap->get_epoch();
      min_last_epoch_clean_pgs.clear();
      RWLock::RLocker lpg(pg_map_lock);
      for (const auto &i : pg_map) {
        PG *pg = i.second;
        if (!pg->is_primary()) {
          continue;
        }

        pg->pg_stats_publish_lock.Lock();
        if (pg->pg_stats_publish_valid) {
          m->pg_stat[pg->info.pgid.pgid] = pg->pg_stats_publish;
	  const auto lec = pg->pg_stats_publish.get_effective_last_epoch_clean();
	  min_last_epoch_clean = min(min_last_epoch_clean, lec);
	  min_last_epoch_clean_pgs.push_back(pg->info.pgid.pgid);
        }
        pg->pg_stats_publish_lock.Unlock();
      }

      return m;
  });

  mgrc.init();
  client_messenger->add_dispatcher_head(&mgrc);

  // tell monc about log_client so it will know about mon session resets
  monc->set_log_client(&log_client);
  update_log_config();

  peering_tp.start();
  
  service.init();
  service.publish_map(osdmap);
  service.publish_superblock(superblock);
  service.max_oldest_map = superblock.oldest_map;

  osd_op_tp.start();
  disk_tp.start();
  command_tp.start();

  set_disk_tp_priority();

  // start the heartbeat
  heartbeat_thread.create("osd_srv_heartbt");

  // tick
  tick_timer.add_event_after(cct->_conf->osd_heartbeat_interval, new C_Tick(this));
  {
    Mutex::Locker l(tick_timer_lock);
    tick_timer_without_osd_lock.add_event_after(cct->_conf->osd_heartbeat_interval, new C_Tick_WithoutOSDLock(this));
  }

  osd_lock.Unlock();

  r = monc->authenticate();
  if (r < 0) {
    derr << __func__ << " authentication failed: " << cpp_strerror(r)
         << dendl;
    osd_lock.Lock(); // locker is going to unlock this on function exit
    if (is_stopping())
      r = 0;
    goto monout;
  }

  while (monc->wait_auth_rotating(30.0) < 0) {
    derr << "unable to obtain rotating service keys; retrying" << dendl;
    ++rotating_auth_attempts;
    if (rotating_auth_attempts > g_conf->max_rotating_auth_attempts) {
        derr << __func__ << " wait_auth_rotating timed out" << dendl;
        osd_lock.Lock(); // make locker happy
        if (!is_stopping()) {
            r = -ETIMEDOUT;
        }
        goto monout;
    }
  }

  r = update_crush_device_class();
  if (r < 0) {
    derr << __func__ << " unable to update_crush_device_class: "
	 << cpp_strerror(r) << dendl;
    osd_lock.Lock();
    goto monout;
  }

  r = update_crush_location();
  if (r < 0) {
    derr << __func__ << " unable to update_crush_location: "
         << cpp_strerror(r) << dendl;
    osd_lock.Lock();
    goto monout;
  }

  osd_lock.Lock();
  if (is_stopping())
    return 0;

  // start objecter *after* we have authenticated, so that we don't ignore
  // the OSDMaps it requests.
  service.final_init();

  check_config();

  dout(10) << "ensuring pgs have consumed prior maps" << dendl;
  consume_map();
  peering_wq.drain();

  dout(0) << "done with init, starting boot process" << dendl;

  // subscribe to any pg creations
  monc->sub_want("osd_pg_creates", last_pg_create_epoch, 0);

  // MgrClient needs this (it doesn't have MonClient reference itself)
  monc->sub_want("mgrmap", 0, 0);

  // we don't need to ask for an osdmap here; objecter will
  //monc->sub_want("osdmap", osdmap->get_epoch(), CEPH_SUBSCRIBE_ONETIME);

  monc->renew_subs();

  start_boot();

  return 0;
monout:
  exit(1);

out:
  enable_disable_fuse(true);
  store->umount();
  delete store;
  store = NULL;
  return r;
}