ceph/src/mds/MDSRank.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2015 Red Hat
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #include "common/debug.h"
  16 #include "common/errno.h"
  17
  18 #include "messages/MClientRequestForward.h"
  19 #include "messages/MMDSLoadTargets.h"
  20 #include "messages/MMDSMap.h"
  21 #include "messages/MMDSTableRequest.h"
  22 #include "messages/MCommand.h"
  23 #include "messages/MCommandReply.h"
  24
  25 #include "MDSDaemon.h"
  26 #include "MDSMap.h"
  27 #include "SnapClient.h"
  28 #include "SnapServer.h"
  29 #include "MDBalancer.h"
  30 #include "Locker.h"
  31 #include "Server.h"
  32 #include "InoTable.h"
  33 #include "mon/MonClient.h"
  34 #include "common/HeartbeatMap.h"
  35 #include "ScrubStack.h"
  36
  37
  38 #include "MDSRank.h"
  39
  40 #define dout_context g_ceph_context
  41 #define dout_subsys ceph_subsys_mds
  42 #undef dout_prefix
  43 #define dout_prefix *_dout << "mds." << whoami << '.' << incarnation << ' '
  44
  45 MDSRank::MDSRank(
  46     mds_rank_t whoami_,
  47     Mutex &mds_lock_,
  48     LogChannelRef &clog_,
  49     SafeTimer &timer_,
  50     Beacon &beacon_,
  51     MDSMap *& mdsmap_,
  52     Messenger *msgr,
  53     MonClient *monc_,
  54     Context *respawn_hook_,
  55     Context *suicide_hook_)
  56   :
  57     whoami(whoami_), incarnation(0),
  58     mds_lock(mds_lock_), clog(clog_), timer(timer_),
  59     mdsmap(mdsmap_),
  60     objecter(new Objecter(g_ceph_context, msgr, monc_, nullptr, 0, 0)),
  61     server(NULL), mdcache(NULL), locker(NULL), mdlog(NULL),
  62     balancer(NULL), scrubstack(NULL),
  63     damage_table(whoami_),
  64     inotable(NULL), snapserver(NULL), snapclient(NULL),
  65     sessionmap(this), logger(NULL), mlogger(NULL),
  66     op_tracker(g_ceph_context, g_conf->mds_enable_op_tracker,
  67                g_conf->osd_num_op_tracker_shard),
  68     last_state(MDSMap::STATE_BOOT),
  69     state(MDSMap::STATE_BOOT),
  70     cluster_degraded(false), stopping(false),
  71     purge_queue(g_ceph_context, whoami_,
  72       mdsmap_->get_metadata_pool(), objecter,
  73       new FunctionContext(
  74           [this](int r){
  75           // Purge Queue operates inside mds_lock when we're calling into
  76           // it, and outside when in background, so must handle both cases.
  77           if (mds_lock.is_locked_by_me()) {
  78             damaged();
  79           } else {
  80             damaged_unlocked();
  81           }
  82         }
  83       )
  84     ),
  85     progress_thread(this), dispatch_depth(0),
  86     hb(NULL), last_tid(0), osd_epoch_barrier(0), beacon(beacon_),
  87     mds_slow_req_count(0),
  88     last_client_mdsmap_bcast(0),
  89     messenger(msgr), monc(monc_),
  90     respawn_hook(respawn_hook_),
  91     suicide_hook(suicide_hook_),
  92     standby_replaying(false)
  93 {
  94   hb = g_ceph_context->get_heartbeat_map()->add_worker("MDSRank", pthread_self());
  95
  96   purge_queue.update_op_limit(*mdsmap);
  97
  98   objecter->unset_honor_osdmap_full();
  99
 100   finisher = new Finisher(msgr->cct);
 101
 102   mdcache = new MDCache(this, purge_queue);
 103   mdlog = new MDLog(this);
 104   balancer = new MDBalancer(this, messenger, monc);
 105
 106   scrubstack = new ScrubStack(mdcache, finisher);
 107
 108   inotable = new InoTable(this);
 109   snapserver = new SnapServer(this, monc);
 110   snapclient = new SnapClient(this);
 111
 112   server = new Server(this);
 113   locker = new Locker(this, mdcache);
 114
 115   op_tracker.set_complaint_and_threshold(msgr->cct->_conf->mds_op_complaint_time,
 116                                          msgr->cct->_conf->mds_op_log_threshold);
 117   op_tracker.set_history_size_and_duration(msgr->cct->_conf->mds_op_history_size,
 118                                            msgr->cct->_conf->mds_op_history_duration);
 119 }
 120
 121 MDSRank::~MDSRank()
 122 {
 123   if (hb) {
 124     g_ceph_context->get_heartbeat_map()->remove_worker(hb);
 125   }
 126
 127   if (scrubstack) { delete scrubstack; scrubstack = NULL; }
 128   if (mdcache) { delete mdcache; mdcache = NULL; }
 129   if (mdlog) { delete mdlog; mdlog = NULL; }
 130   if (balancer) { delete balancer; balancer = NULL; }
 131   if (inotable) { delete inotable; inotable = NULL; }
 132   if (snapserver) { delete snapserver; snapserver = NULL; }
 133   if (snapclient) { delete snapclient; snapclient = NULL; }
 134   if (mdsmap) { delete mdsmap; mdsmap = 0; }
 135
 136   if (server) { delete server; server = 0; }
 137   if (locker) { delete locker; locker = 0; }
 138
 139   if (logger) {
 140     g_ceph_context->get_perfcounters_collection()->remove(logger);
 141     delete logger;
 142     logger = 0;
 143   }
 144   if (mlogger) {
 145     g_ceph_context->get_perfcounters_collection()->remove(mlogger);
 146     delete mlogger;
 147     mlogger = 0;
 148   }
 149
 150   delete finisher;
 151   finisher = NULL;
 152
 153   delete suicide_hook;
 154   suicide_hook = NULL;
 155
 156   delete respawn_hook;
 157   respawn_hook = NULL;
 158
 159   delete objecter;
 160   objecter = nullptr;
 161 }
 162
 163 void MDSRankDispatcher::init()
 164 {
 165   objecter->init();
 166   messenger->add_dispatcher_head(objecter);
 167
 168   objecter->start();
 169
 170   update_log_config();
 171   create_logger();
 172
 173   // Expose the OSDMap (already populated during MDS::init) to anyone
 174   // who is interested in it.
 175   handle_osd_map();
 176
 177   progress_thread.create("mds_rank_progr");
 178
 179   purge_queue.init();
 180
 181   finisher->start();
 182 }
 183
 184 void MDSRank::update_targets(utime_t now)
 185 {
 186   // get MonMap's idea of my export_targets
 187   const set<mds_rank_t>& map_targets = mdsmap->get_mds_info(get_nodeid()).export_targets;
 188
 189   dout(20) << "updating export targets, currently " << map_targets.size() << " ranks are targets" << dendl;
 190
 191   bool send = false;
 192   set<mds_rank_t> new_map_targets;
 193
 194   auto it = export_targets.begin();
 195   while (it != export_targets.end()) {
 196     mds_rank_t rank = it->first;
 197     double val = it->second.get(now);
 198     dout(20) << "export target mds." << rank << " value is " << val << " @ " << now << dendl;
 199
 200     if (val <= 0.01) {
 201       dout(15) << "export target mds." << rank << " is no longer an export target" << dendl;
 202       export_targets.erase(it++);
 203       send = true;
 204       continue;
 205     }
 206     if (!map_targets.count(rank)) {
 207       dout(15) << "export target mds." << rank << " not in map's export_targets" << dendl;
 208       send = true;
 209     }
 210     new_map_targets.insert(rank);
 211     it++;
 212   }
 213   if (new_map_targets.size() < map_targets.size()) {
 214     dout(15) << "export target map holds stale targets, sending update" << dendl;
 215     send = true;
 216   }
 217
 218   if (send) {
 219     dout(15) << "updating export_targets, now " << new_map_targets.size() << " ranks are targets" << dendl;
 220     MMDSLoadTargets* m = new MMDSLoadTargets(mds_gid_t(monc->get_global_id()), new_map_targets);
 221     monc->send_mon_message(m);
 222   }
 223 }
 224
 225 void MDSRank::hit_export_target(utime_t now, mds_rank_t rank, double amount)
 226 {
 227   double rate = g_conf->mds_bal_target_decay;
 228   if (amount < 0.0) {
 229     amount = 100.0/g_conf->mds_bal_target_decay; /* a good default for "i am trying to keep this export_target active" */
 230   }
 231   auto em = export_targets.emplace(std::piecewise_construct, std::forward_as_tuple(rank), std::forward_as_tuple(now, DecayRate(rate)));
 232   if (em.second) {
 233     dout(15) << "hit export target (new) " << amount << " @ " << now << dendl;
 234   } else {
 235     dout(15) << "hit export target " << amount << " @ " << now << dendl;
 236   }
 237   em.first->second.hit(now, amount);
 238 }
 239
 240 void MDSRankDispatcher::tick()
 241 {
 242   heartbeat_reset();
 243
 244   if (beacon.is_laggy()) {
 245     dout(5) << "tick bailing out since we seem laggy" << dendl;
 246     return;
 247   }
 248
 249   check_ops_in_flight();
 250
 251   // Wake up thread in case we use to be laggy and have waiting_for_nolaggy
 252   // messages to progress.
 253   progress_thread.signal();
 254
 255   // make sure mds log flushes, trims periodically
 256   mdlog->flush();
 257
 258   if (is_active() || is_stopping()) {
 259     mdcache->trim();
 260     mdcache->trim_client_leases();
 261     mdcache->check_memory_usage();
 262     mdlog->trim();  // NOT during recovery!
 263   }
 264
 265   // log
 266   mds_load_t load = balancer->get_load(ceph_clock_now());
 267
 268   if (logger) {
 269     logger->set(l_mds_load_cent, 100 * load.mds_load());
 270     logger->set(l_mds_dispatch_queue_len, messenger->get_dispatch_queue_len());
 271     logger->set(l_mds_subtrees, mdcache->num_subtrees());
 272
 273     mdcache->log_stat();
 274   }
 275
 276   // ...
 277   if (is_clientreplay() || is_active() || is_stopping()) {
 278     server->find_idle_sessions();
 279     locker->tick();
 280   }
 281
 282   if (is_reconnect())
 283     server->reconnect_tick();
 284
 285   if (is_active()) {
 286     balancer->tick();
 287     mdcache->find_stale_fragment_freeze();
 288     mdcache->migrator->find_stale_export_freeze();
 289     if (snapserver)
 290       snapserver->check_osd_map(false);
 291   }
 292
 293   if (is_active() || is_stopping()) {
 294     update_targets(ceph_clock_now());
 295   }
 296
 297   // shut down?
 298   if (is_stopping()) {
 299     mdlog->trim();
 300     if (mdcache->shutdown_pass()) {
 301       uint64_t pq_progress = 0 ;
 302       uint64_t pq_total = 0;
 303       size_t pq_in_flight = 0;
 304       if (!purge_queue.drain(&pq_progress, &pq_total, &pq_in_flight)) {
 305         dout(7) << "shutdown_pass=true, but still waiting for purge queue"
 306                 << dendl;
 307         // This takes unbounded time, so we must indicate progress
 308         // to the administrator: we do it in a slightly imperfect way
 309         // by sending periodic (tick frequency) clog messages while
 310         // in this state.
 311         clog->info() << "MDS rank " << whoami << " waiting for purge queue ("
 312           << std::dec << pq_progress << "/" << pq_total << " " << pq_in_flight
 313           << " files purging" << ")";
 314       } else {
 315         dout(7) << "shutdown_pass=true, finished w/ shutdown, moving to "
 316                    "down:stopped" << dendl;
 317         stopping_done();
 318       }
 319     }
 320     else {
 321       dout(7) << "shutdown_pass=false" << dendl;
 322     }
 323   }
 324
 325   // Expose ourselves to Beacon to update health indicators
 326   beacon.notify_health(this);
 327 }
 328
 329 void MDSRankDispatcher::shutdown()
 330 {
 331   // It should never be possible for shutdown to get called twice, because
 332   // anyone picking up mds_lock checks if stopping is true and drops
 333   // out if it is.
 334   assert(stopping == false);
 335   stopping = true;
 336
 337   dout(1) << __func__ << ": shutting down rank " << whoami << dendl;
 338
 339   timer.shutdown();
 340
 341   // MDLog has to shut down before the finisher, because some of its
 342   // threads block on IOs that require finisher to complete.
 343   mdlog->shutdown();
 344
 345   // shut down cache
 346   mdcache->shutdown();
 347
 348   purge_queue.shutdown();
 349
 350   mds_lock.Unlock();
 351   finisher->stop(); // no flushing
 352   mds_lock.Lock();
 353
 354   if (objecter->initialized)
 355     objecter->shutdown();
 356
 357   monc->shutdown();
 358
 359   op_tracker.on_shutdown();
 360
 361   progress_thread.shutdown();
 362
 363   // release mds_lock for finisher/messenger threads (e.g.
 364   // MDSDaemon::ms_handle_reset called from Messenger).
 365   mds_lock.Unlock();
 366
 367   // shut down messenger
 368   messenger->shutdown();
 369
 370   mds_lock.Lock();
 371
 372   // Workaround unclean shutdown: HeartbeatMap will assert if
 373   // worker is not removed (as we do in ~MDS), but ~MDS is not
 374   // always called after suicide.
 375   if (hb) {
 376     g_ceph_context->get_heartbeat_map()->remove_worker(hb);
 377     hb = NULL;
 378   }
 379 }
 380
 381 /**
 382  * Helper for simple callbacks that call a void fn with no args.
 383  */
 384 class C_MDS_VoidFn : public MDSInternalContext
 385 {
 386   typedef void (MDSRank::*fn_ptr)();
 387   protected:
 388    fn_ptr fn;
 389   public:
 390   C_MDS_VoidFn(MDSRank *mds_, fn_ptr fn_)
 391     : MDSInternalContext(mds_), fn(fn_)
 392   {
 393     assert(mds_);
 394     assert(fn_);
 395   }
 396
 397   void finish(int r) override
 398   {
 399     (mds->*fn)();
 400   }
 401 };
 402
 403 int64_t MDSRank::get_metadata_pool()
 404 {
 405     return mdsmap->get_metadata_pool();
 406 }
 407
 408 MDSTableClient *MDSRank::get_table_client(int t)
 409 {
 410   switch (t) {
 411   case TABLE_ANCHOR: return NULL;
 412   case TABLE_SNAP: return snapclient;
 413   default: ceph_abort();
 414   }
 415 }
 416
 417 MDSTableServer *MDSRank::get_table_server(int t)
 418 {
 419   switch (t) {
 420   case TABLE_ANCHOR: return NULL;
 421   case TABLE_SNAP: return snapserver;
 422   default: ceph_abort();
 423   }
 424 }
 425
 426 void MDSRank::suicide()
 427 {
 428   if (suicide_hook) {
 429     suicide_hook->complete(0);
 430     suicide_hook = NULL;
 431   }
 432 }
 433
 434 void MDSRank::respawn()
 435 {
 436   if (respawn_hook) {
 437     respawn_hook->complete(0);
 438     respawn_hook = NULL;
 439   }
 440 }
 441
 442 void MDSRank::damaged()
 443 {
 444   assert(whoami != MDS_RANK_NONE);
 445   assert(mds_lock.is_locked_by_me());
 446
 447   beacon.set_want_state(mdsmap, MDSMap::STATE_DAMAGED);
 448   monc->flush_log();  // Flush any clog error from before we were called
 449   beacon.notify_health(this);  // Include latest status in our swan song
 450   beacon.send_and_wait(g_conf->mds_mon_shutdown_timeout);
 451
 452   // It's okay if we timed out and the mon didn't get our beacon, because
 453   // another daemon (or ourselves after respawn) will eventually take the
 454   // rank and report DAMAGED again when it hits same problem we did.
 455
 456   respawn();  // Respawn into standby in case mon has other work for us
 457 }
 458
 459 void MDSRank::damaged_unlocked()
 460 {
 461   Mutex::Locker l(mds_lock);
 462   damaged();
 463 }
 464
 465 void MDSRank::handle_write_error(int err)
 466 {
 467   if (err == -EBLACKLISTED) {
 468     derr << "we have been blacklisted (fenced), respawning..." << dendl;
 469     respawn();
 470     return;
 471   }
 472
 473   if (g_conf->mds_action_on_write_error >= 2) {
 474     derr << "unhandled write error " << cpp_strerror(err) << ", suicide..." << dendl;
 475     respawn();
 476   } else if (g_conf->mds_action_on_write_error == 1) {
 477     derr << "unhandled write error " << cpp_strerror(err) << ", force readonly..." << dendl;
 478     mdcache->force_readonly();
 479   } else {
 480     // ignore;
 481     derr << "unhandled write error " << cpp_strerror(err) << ", ignore..." << dendl;
 482   }
 483 }
 484
 485 void *MDSRank::ProgressThread::entry()
 486 {
 487   Mutex::Locker l(mds->mds_lock);
 488   while (true) {
 489     while (!mds->stopping &&
 490            mds->finished_queue.empty() &&
 491            (mds->waiting_for_nolaggy.empty() || mds->beacon.is_laggy())) {
 492       cond.Wait(mds->mds_lock);
 493     }
 494
 495     if (mds->stopping) {
 496       break;
 497     }
 498
 499     mds->_advance_queues();
 500   }
 501
 502   return NULL;
 503 }
 504
 505
 506 void MDSRank::ProgressThread::shutdown()
 507 {
 508   assert(mds->mds_lock.is_locked_by_me());
 509   assert(mds->stopping);
 510
 511   if (am_self()) {
 512     // Stopping is set, we will fall out of our main loop naturally
 513   } else {
 514     // Kick the thread to notice mds->stopping, and join it
 515     cond.Signal();
 516     mds->mds_lock.Unlock();
 517     if (is_started())
 518       join();
 519     mds->mds_lock.Lock();
 520   }
 521 }
 522
 523 bool MDSRankDispatcher::ms_dispatch(Message *m)
 524 {
 525   bool ret;
 526   inc_dispatch_depth();
 527   ret = _dispatch(m, true);
 528   dec_dispatch_depth();
 529   return ret;
 530 }
 531
 532 /* If this function returns true, it recognizes the message and has taken the
 533  * reference. If it returns false, it has done neither. */
 534 bool MDSRank::_dispatch(Message *m, bool new_msg)
 535 {
 536   if (is_stale_message(m)) {
 537     m->put();
 538     return true;
 539   }
 540
 541   if (beacon.is_laggy()) {
 542     dout(10) << " laggy, deferring " << *m << dendl;
 543     waiting_for_nolaggy.push_back(m);
 544   } else if (new_msg && !waiting_for_nolaggy.empty()) {
 545     dout(10) << " there are deferred messages, deferring " << *m << dendl;
 546     waiting_for_nolaggy.push_back(m);
 547   } else {
 548     if (!handle_deferrable_message(m)) {
 549       dout(0) << "unrecognized message " << *m << dendl;
 550       return false;
 551     }
 552
 553     heartbeat_reset();
 554   }
 555
 556   if (dispatch_depth > 1)
 557     return true;
 558
 559   // finish any triggered contexts
 560   _advance_queues();
 561
 562   if (beacon.is_laggy()) {
 563     // We've gone laggy during dispatch, don't do any
 564     // more housekeeping
 565     return true;
 566   }
 567
 568   // done with all client replayed requests?
 569   if (is_clientreplay() &&
 570       mdcache->is_open() &&
 571       replay_queue.empty() &&
 572       beacon.get_want_state() == MDSMap::STATE_CLIENTREPLAY) {
 573     int num_requests = mdcache->get_num_client_requests();
 574     dout(10) << " still have " << num_requests << " active replay requests" << dendl;
 575     if (num_requests == 0)
 576       clientreplay_done();
 577   }
 578
 579   // hack: thrash exports
 580   static utime_t start;
 581   utime_t now = ceph_clock_now();
 582   if (start == utime_t())
 583     start = now;
 584   /*double el = now - start;
 585   if (el > 30.0 &&
 586     el < 60.0)*/
 587   for (int i=0; i<g_conf->mds_thrash_exports; i++) {
 588     set<mds_rank_t> s;
 589     if (!is_active()) break;
 590     mdsmap->get_mds_set(s, MDSMap::STATE_ACTIVE);
 591     if (s.size() < 2 || CInode::count() < 10)
 592       break;  // need peers for this to work.
 593     if (mdcache->migrator->get_num_exporting() > g_conf->mds_thrash_exports * 5 ||
 594         mdcache->migrator->get_export_queue_size() > g_conf->mds_thrash_exports * 10)
 595       break;
 596
 597     dout(7) << "mds thrashing exports pass " << (i+1) << "/" << g_conf->mds_thrash_exports << dendl;
 598
 599     // pick a random dir inode
 600     CInode *in = mdcache->hack_pick_random_inode();
 601
 602     list<CDir*> ls;
 603     in->get_dirfrags(ls);
 604     if (!ls.empty()) {  // must be an open dir.
 605       list<CDir*>::iterator p = ls.begin();
 606       int n = rand() % ls.size();
 607       while (n--)
 608         ++p;
 609       CDir *dir = *p;
 610       if (!dir->get_parent_dir()) continue;    // must be linked.
 611       if (!dir->is_auth()) continue;           // must be auth.
 612
 613       mds_rank_t dest;
 614       do {
 615         int k = rand() % s.size();
 616         set<mds_rank_t>::iterator p = s.begin();
 617         while (k--) ++p;
 618         dest = *p;
 619       } while (dest == whoami);
 620       mdcache->migrator->export_dir_nicely(dir,dest);
 621     }
 622   }
 623   // hack: thrash fragments
 624   for (int i=0; i<g_conf->mds_thrash_fragments; i++) {
 625     if (!is_active()) break;
 626     if (mdcache->get_num_fragmenting_dirs() > 5 * g_conf->mds_thrash_fragments) break;
 627     dout(7) << "mds thrashing fragments pass " << (i+1) << "/" << g_conf->mds_thrash_fragments << dendl;
 628
 629     // pick a random dir inode
 630     CInode *in = mdcache->hack_pick_random_inode();
 631
 632     list<CDir*> ls;
 633     in->get_dirfrags(ls);
 634     if (ls.empty()) continue;                // must be an open dir.
 635     CDir *dir = ls.front();
 636     if (!dir->get_parent_dir()) continue;    // must be linked.
 637     if (!dir->is_auth()) continue;           // must be auth.
 638     frag_t fg = dir->get_frag();
 639     if (mdsmap->allows_dirfrags()) {
 640       if ((fg == frag_t() || (rand() % (1 << fg.bits()) == 0))) {
 641         mdcache->split_dir(dir, 1);
 642       } else {
 643         balancer->queue_merge(dir);
 644       }
 645     }
 646   }
 647
 648   // hack: force hash root?
 649   /*
 650   if (false &&
 651       mdcache->get_root() &&
 652       mdcache->get_root()->dir &&
 653       !(mdcache->get_root()->dir->is_hashed() ||
 654         mdcache->get_root()->dir->is_hashing())) {
 655     dout(0) << "hashing root" << dendl;
 656     mdcache->migrator->hash_dir(mdcache->get_root()->dir);
 657   }
 658   */
 659
 660   update_mlogger();
 661   return true;
 662 }
 663
 664 void MDSRank::update_mlogger()
 665 {
 666   if (mlogger) {
 667     mlogger->set(l_mdm_ino, CInode::count());
 668     mlogger->set(l_mdm_dir, CDir::count());
 669     mlogger->set(l_mdm_dn, CDentry::count());
 670     mlogger->set(l_mdm_cap, Capability::count());
 671     mlogger->set(l_mdm_inoa, CInode::increments());
 672     mlogger->set(l_mdm_inos, CInode::decrements());
 673     mlogger->set(l_mdm_dira, CDir::increments());
 674     mlogger->set(l_mdm_dirs, CDir::decrements());
 675     mlogger->set(l_mdm_dna, CDentry::increments());
 676     mlogger->set(l_mdm_dns, CDentry::decrements());
 677     mlogger->set(l_mdm_capa, Capability::increments());
 678     mlogger->set(l_mdm_caps, Capability::decrements());
 679     mlogger->set(l_mdm_buf, buffer::get_total_alloc());
 680   }
 681 }
 682
 683 /*
 684  * lower priority messages we defer if we seem laggy
 685  */
 686 bool MDSRank::handle_deferrable_message(Message *m)
 687 {
 688   int port = m->get_type() & 0xff00;
 689
 690   switch (port) {
 691   case MDS_PORT_CACHE:
 692     ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS);
 693     mdcache->dispatch(m);
 694     break;
 695
 696   case MDS_PORT_MIGRATOR:
 697     ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS);
 698     mdcache->migrator->dispatch(m);
 699     break;
 700
 701   default:
 702     switch (m->get_type()) {
 703       // SERVER
 704     case CEPH_MSG_CLIENT_SESSION:
 705     case CEPH_MSG_CLIENT_RECONNECT:
 706       ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_CLIENT);
 707       // fall-thru
 708     case CEPH_MSG_CLIENT_REQUEST:
 709       server->dispatch(m);
 710       break;
 711     case MSG_MDS_SLAVE_REQUEST:
 712       ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS);
 713       server->dispatch(m);
 714       break;
 715
 716     case MSG_MDS_HEARTBEAT:
 717       ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS);
 718       balancer->proc_message(m);
 719       break;
 720
 721     case MSG_MDS_TABLE_REQUEST:
 722       ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS);
 723       {
 724         MMDSTableRequest *req = static_cast<MMDSTableRequest*>(m);
 725         if (req->op < 0) {
 726           MDSTableClient *client = get_table_client(req->table);
 727               client->handle_request(req);
 728         } else {
 729           MDSTableServer *server = get_table_server(req->table);
 730           server->handle_request(req);
 731         }
 732       }
 733       break;
 734
 735     case MSG_MDS_LOCK:
 736     case MSG_MDS_INODEFILECAPS:
 737       ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS);
 738       locker->dispatch(m);
 739       break;
 740
 741     case CEPH_MSG_CLIENT_CAPS:
 742     case CEPH_MSG_CLIENT_CAPRELEASE:
 743     case CEPH_MSG_CLIENT_LEASE:
 744       ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_CLIENT);
 745       locker->dispatch(m);
 746       break;
 747
 748     default:
 749       return false;
 750     }
 751   }
 752
 753   return true;
 754 }
 755
 756 /**
 757  * Advance finished_queue and waiting_for_nolaggy.
 758  *
 759  * Usually drain both queues, but may not drain waiting_for_nolaggy
 760  * if beacon is currently laggy.
 761  */
 762 void MDSRank::_advance_queues()
 763 {
 764   assert(mds_lock.is_locked_by_me());
 765
 766   while (!finished_queue.empty()) {
 767     dout(7) << "mds has " << finished_queue.size() << " queued contexts" << dendl;
 768     dout(10) << finished_queue << dendl;
 769     list<MDSInternalContextBase*> ls;
 770     ls.swap(finished_queue);
 771     while (!ls.empty()) {
 772       dout(10) << " finish " << ls.front() << dendl;
 773       ls.front()->complete(0);
 774       ls.pop_front();
 775
 776       heartbeat_reset();
 777     }
 778   }
 779
 780   while (!waiting_for_nolaggy.empty()) {
 781     // stop if we're laggy now!
 782     if (beacon.is_laggy())
 783       break;
 784
 785     Message *old = waiting_for_nolaggy.front();
 786     waiting_for_nolaggy.pop_front();
 787
 788     if (is_stale_message(old)) {
 789       old->put();
 790     } else {
 791       dout(7) << " processing laggy deferred " << *old << dendl;
 792       if (!handle_deferrable_message(old)) {
 793         dout(0) << "unrecognized message " << *old << dendl;
 794         old->put();
 795       }
 796     }
 797
 798     heartbeat_reset();
 799   }
 800 }
 801
 802 /**
 803  * Call this when you take mds_lock, or periodically if you're going to
 804  * hold the lock for a long time (e.g. iterating over clients/inodes)
 805  */
 806 void MDSRank::heartbeat_reset()
 807 {
 808   // Any thread might jump into mds_lock and call us immediately
 809   // after a call to suicide() completes, in which case MDSRank::hb
 810   // has been freed and we are a no-op.
 811   if (!hb) {
 812       assert(stopping);
 813       return;
 814   }
 815
 816   // NB not enabling suicide grace, because the mon takes care of killing us
 817   // (by blacklisting us) when we fail to send beacons, and it's simpler to
 818   // only have one way of dying.
 819   g_ceph_context->get_heartbeat_map()->reset_timeout(hb, g_conf->mds_beacon_grace, 0);
 820 }
 821
 822 bool MDSRank::is_stale_message(Message *m) const
 823 {
 824   // from bad mds?
 825   if (m->get_source().is_mds()) {
 826     mds_rank_t from = mds_rank_t(m->get_source().num());
 827     if (!mdsmap->have_inst(from) ||
 828         mdsmap->get_inst(from) != m->get_source_inst() ||
 829         mdsmap->is_down(from)) {
 830       // bogus mds?
 831       if (m->get_type() == CEPH_MSG_MDS_MAP) {
 832         dout(5) << "got " << *m << " from old/bad/imposter mds " << m->get_source()
 833                 << ", but it's an mdsmap, looking at it" << dendl;
 834       } else if (m->get_type() == MSG_MDS_CACHEEXPIRE &&
 835                  mdsmap->get_inst(from) == m->get_source_inst()) {
 836         dout(5) << "got " << *m << " from down mds " << m->get_source()
 837                 << ", but it's a cache_expire, looking at it" << dendl;
 838       } else {
 839         dout(5) << "got " << *m << " from down/old/bad/imposter mds " << m->get_source()
 840                 << ", dropping" << dendl;
 841         return true;
 842       }
 843     }
 844   }
 845   return false;
 846 }
 847
 848
 849 void MDSRank::send_message(Message *m, Connection *c)
 850 {
 851   assert(c);
 852   c->send_message(m);
 853 }
 854
 855
 856 void MDSRank::send_message_mds(Message *m, mds_rank_t mds)
 857 {
 858   if (!mdsmap->is_up(mds)) {
 859     dout(10) << "send_message_mds mds." << mds << " not up, dropping " << *m << dendl;
 860     m->put();
 861     return;
 862   }
 863
 864   // send mdsmap first?
 865   if (mds != whoami && peer_mdsmap_epoch[mds] < mdsmap->get_epoch()) {
 866     messenger->send_message(new MMDSMap(monc->get_fsid(), mdsmap),
 867                             mdsmap->get_inst(mds));
 868     peer_mdsmap_epoch[mds] = mdsmap->get_epoch();
 869   }
 870
 871   // send message
 872   messenger->send_message(m, mdsmap->get_inst(mds));
 873 }
 874
 875 void MDSRank::forward_message_mds(Message *m, mds_rank_t mds)
 876 {
 877   assert(mds != whoami);
 878
 879   // client request?
 880   if (m->get_type() == CEPH_MSG_CLIENT_REQUEST &&
 881       (static_cast<MClientRequest*>(m))->get_source().is_client()) {
 882     MClientRequest *creq = static_cast<MClientRequest*>(m);
 883     creq->inc_num_fwd();    // inc forward counter
 884
 885     /*
 886      * don't actually forward if non-idempotent!
 887      * client has to do it.  although the MDS will ignore duplicate requests,
 888      * the affected metadata may migrate, in which case the new authority
 889      * won't have the metareq_id in the completed request map.
 890      */
 891     // NEW: always make the client resend!
 892     bool client_must_resend = true;  //!creq->can_forward();
 893
 894     // tell the client where it should go
 895     messenger->send_message(new MClientRequestForward(creq->get_tid(), mds, creq->get_num_fwd(),
 896                                                       client_must_resend),
 897                             creq->get_source_inst());
 898
 899     if (client_must_resend) {
 900       m->put();
 901       return;
 902     }
 903   }
 904
 905   // these are the only types of messages we should be 'forwarding'; they
 906   // explicitly encode their source mds, which gets clobbered when we resend
 907   // them here.
 908   assert(m->get_type() == MSG_MDS_DIRUPDATE ||
 909          m->get_type() == MSG_MDS_EXPORTDIRDISCOVER);
 910
 911   // send mdsmap first?
 912   if (peer_mdsmap_epoch[mds] < mdsmap->get_epoch()) {
 913     messenger->send_message(new MMDSMap(monc->get_fsid(), mdsmap),
 914                             mdsmap->get_inst(mds));
 915     peer_mdsmap_epoch[mds] = mdsmap->get_epoch();
 916   }
 917
 918   messenger->send_message(m, mdsmap->get_inst(mds));
 919 }
 920
 921
 922
 923 void MDSRank::send_message_client_counted(Message *m, client_t client)
 924 {
 925   Session *session =  sessionmap.get_session(entity_name_t::CLIENT(client.v));
 926   if (session) {
 927     send_message_client_counted(m, session);
 928   } else {
 929     dout(10) << "send_message_client_counted no session for client." << client << " " << *m << dendl;
 930   }
 931 }
 932
 933 void MDSRank::send_message_client_counted(Message *m, Connection *connection)
 934 {
 935   Session *session = static_cast<Session *>(connection->get_priv());
 936   if (session) {
 937     session->put();  // do not carry ref
 938     send_message_client_counted(m, session);
 939   } else {
 940     dout(10) << "send_message_client_counted has no session for " << m->get_source_inst() << dendl;
 941     // another Connection took over the Session
 942   }
 943 }
 944
 945 void MDSRank::send_message_client_counted(Message *m, Session *session)
 946 {
 947   version_t seq = session->inc_push_seq();
 948   dout(10) << "send_message_client_counted " << session->info.inst.name << " seq "
 949            << seq << " " << *m << dendl;
 950   if (session->connection) {
 951     session->connection->send_message(m);
 952   } else {
 953     session->preopen_out_queue.push_back(m);
 954   }
 955 }
 956
 957 void MDSRank::send_message_client(Message *m, Session *session)
 958 {
 959   dout(10) << "send_message_client " << session->info.inst << " " << *m << dendl;
 960   if (session->connection) {
 961     session->connection->send_message(m);
 962   } else {
 963     session->preopen_out_queue.push_back(m);
 964   }
 965 }
 966
 967 /**
 968  * This is used whenever a RADOS operation has been cancelled
 969  * or a RADOS client has been blacklisted, to cause the MDS and
 970  * any clients to wait for this OSD epoch before using any new caps.
 971  *
 972  * See doc/cephfs/eviction
 973  */
 974 void MDSRank::set_osd_epoch_barrier(epoch_t e)
 975 {
 976   dout(4) << __func__ << ": epoch=" << e << dendl;
 977   osd_epoch_barrier = e;
 978 }
 979
 980 void MDSRank::retry_dispatch(Message *m)
 981 {
 982   inc_dispatch_depth();
 983   _dispatch(m, false);
 984   dec_dispatch_depth();
 985 }
 986
 987 utime_t MDSRank::get_laggy_until() const
 988 {
 989   return beacon.get_laggy_until();
 990 }
 991
 992 bool MDSRank::is_daemon_stopping() const
 993 {
 994   return stopping;
 995 }
 996
 997 void MDSRank::request_state(MDSMap::DaemonState s)
 998 {
 999   dout(3) << "request_state " << ceph_mds_state_name(s) << dendl;
1000   beacon.set_want_state(mdsmap, s);
1001   beacon.send();
1002 }
1003
1004
1005 class C_MDS_BootStart : public MDSInternalContext {
1006   MDSRank::BootStep nextstep;
1007 public:
1008   C_MDS_BootStart(MDSRank *m, MDSRank::BootStep n)
1009     : MDSInternalContext(m), nextstep(n) {}
1010   void finish(int r) override {
1011     mds->boot_start(nextstep, r);
1012   }
1013 };
1014
1015
1016 void MDSRank::boot_start(BootStep step, int r)
1017 {
1018   // Handle errors from previous step
1019   if (r < 0) {
1020     if (is_standby_replay() && (r == -EAGAIN)) {
1021       dout(0) << "boot_start encountered an error EAGAIN"
1022               << ", respawning since we fell behind journal" << dendl;
1023       respawn();
1024     } else if (r == -EINVAL || r == -ENOENT) {
1025       // Invalid or absent data, indicates damaged on-disk structures
1026       clog->error() << "Error loading MDS rank " << whoami << ": "
1027         << cpp_strerror(r);
1028       damaged();
1029       assert(r == 0);  // Unreachable, damaged() calls respawn()
1030     } else {
1031       // Completely unexpected error, give up and die
1032       dout(0) << "boot_start encountered an error, failing" << dendl;
1033       suicide();
1034       return;
1035     }
1036   }
1037
1038   assert(is_starting() || is_any_replay());
1039
1040   switch(step) {
1041     case MDS_BOOT_INITIAL:
1042       {
1043         mdcache->init_layouts();
1044
1045         MDSGatherBuilder gather(g_ceph_context,
1046             new C_MDS_BootStart(this, MDS_BOOT_OPEN_ROOT));
1047         dout(2) << "boot_start " << step << ": opening inotable" << dendl;
1048         inotable->set_rank(whoami);
1049         inotable->load(gather.new_sub());
1050
1051         dout(2) << "boot_start " << step << ": opening sessionmap" << dendl;
1052         sessionmap.set_rank(whoami);
1053         sessionmap.load(gather.new_sub());
1054
1055         dout(2) << "boot_start " << step << ": opening mds log" << dendl;
1056         mdlog->open(gather.new_sub());
1057
1058         if (mdsmap->get_tableserver() == whoami) {
1059           dout(2) << "boot_start " << step << ": opening snap table" << dendl;
1060           snapserver->set_rank(whoami);
1061           snapserver->load(gather.new_sub());
1062         }
1063
1064         gather.activate();
1065       }
1066       break;
1067     case MDS_BOOT_OPEN_ROOT:
1068       {
1069         dout(2) << "boot_start " << step << ": loading/discovering base inodes" << dendl;
1070
1071         MDSGatherBuilder gather(g_ceph_context,
1072             new C_MDS_BootStart(this, MDS_BOOT_PREPARE_LOG));
1073
1074         mdcache->open_mydir_inode(gather.new_sub());
1075
1076         purge_queue.open(new C_IO_Wrapper(this, gather.new_sub()));
1077
1078         if (is_starting() ||
1079             whoami == mdsmap->get_root()) {  // load root inode off disk if we are auth
1080           mdcache->open_root_inode(gather.new_sub());
1081         } else {
1082           // replay.  make up fake root inode to start with
1083           (void)mdcache->create_root_inode();
1084         }
1085         gather.activate();
1086       }
1087       break;
1088     case MDS_BOOT_PREPARE_LOG:
1089       if (is_any_replay()) {
1090         dout(2) << "boot_start " << step << ": replaying mds log" << dendl;
1091         mdlog->replay(new C_MDS_BootStart(this, MDS_BOOT_REPLAY_DONE));
1092       } else {
1093         dout(2) << "boot_start " << step << ": positioning at end of old mds log" << dendl;
1094         mdlog->append();
1095         starting_done();
1096       }
1097       break;
1098     case MDS_BOOT_REPLAY_DONE:
1099       assert(is_any_replay());
1100
1101       // Sessiontable and inotable should be in sync after replay, validate
1102       // that they are consistent.
1103       validate_sessions();
1104
1105       replay_done();
1106       break;
1107   }
1108 }
1109
1110 void MDSRank::validate_sessions()
1111 {
1112   assert(mds_lock.is_locked_by_me());
1113   std::vector<Session*> victims;
1114
1115   // Identify any sessions which have state inconsistent with other,
1116   // after they have been loaded from rados during startup.
1117   // Mitigate bugs like: http://tracker.ceph.com/issues/16842
1118   const auto &sessions = sessionmap.get_sessions();
1119   for (const auto &i : sessions) {
1120     Session *session = i.second;
1121     interval_set<inodeno_t> badones;
1122     if (inotable->intersects_free(session->info.prealloc_inos, &badones)) {
1123       clog->error() << "Client session loaded with invalid preallocated "
1124                           "inodes, evicting session " << *session;
1125
1126       // Make the session consistent with inotable so that it can
1127       // be cleanly torn down
1128       session->info.prealloc_inos.subtract(badones);
1129
1130       victims.push_back(session);
1131     }
1132   }
1133
1134   for (const auto &session: victims) {
1135     server->kill_session(session, nullptr);
1136   }
1137 }
1138
1139 void MDSRank::starting_done()
1140 {
1141   dout(3) << "starting_done" << dendl;
1142   assert(is_starting());
1143   request_state(MDSMap::STATE_ACTIVE);
1144
1145   mdcache->open_root();
1146
1147   // start new segment
1148   mdlog->start_new_segment();
1149 }
1150
1151
1152 void MDSRank::calc_recovery_set()
1153 {
1154   // initialize gather sets
1155   set<mds_rank_t> rs;
1156   mdsmap->get_recovery_mds_set(rs);
1157   rs.erase(whoami);
1158   mdcache->set_recovery_set(rs);
1159
1160   dout(1) << " recovery set is " << rs << dendl;
1161 }
1162
1163
1164 void MDSRank::replay_start()
1165 {
1166   dout(1) << "replay_start" << dendl;
1167
1168   if (is_standby_replay())
1169     standby_replaying = true;
1170
1171   calc_recovery_set();
1172
1173   // Check if we need to wait for a newer OSD map before starting
1174   Context *fin = new C_IO_Wrapper(this, new C_MDS_BootStart(this, MDS_BOOT_INITIAL));
1175   bool const ready = objecter->wait_for_map(
1176       mdsmap->get_last_failure_osd_epoch(),
1177       fin);
1178
1179   if (ready) {
1180     delete fin;
1181     boot_start();
1182   } else {
1183     dout(1) << " waiting for osdmap " << mdsmap->get_last_failure_osd_epoch()
1184             << " (which blacklists prior instance)" << dendl;
1185   }
1186 }
1187
1188
1189 class MDSRank::C_MDS_StandbyReplayRestartFinish : public MDSIOContext {
1190   uint64_t old_read_pos;
1191 public:
1192   C_MDS_StandbyReplayRestartFinish(MDSRank *mds_, uint64_t old_read_pos_) :
1193     MDSIOContext(mds_), old_read_pos(old_read_pos_) {}
1194   void finish(int r) override {
1195     mds->_standby_replay_restart_finish(r, old_read_pos);
1196   }
1197 };
1198
1199 void MDSRank::_standby_replay_restart_finish(int r, uint64_t old_read_pos)
1200 {
1201   if (old_read_pos < mdlog->get_journaler()->get_trimmed_pos()) {
1202     dout(0) << "standby MDS fell behind active MDS journal's expire_pos, restarting" << dendl;
1203     respawn(); /* we're too far back, and this is easier than
1204                   trying to reset everything in the cache, etc */
1205   } else {
1206     mdlog->standby_trim_segments();
1207     boot_start(MDS_BOOT_PREPARE_LOG, r);
1208   }
1209 }
1210
1211 inline void MDSRank::standby_replay_restart()
1212 {
1213   if (standby_replaying) {
1214     /* Go around for another pass of replaying in standby */
1215     dout(4) << "standby_replay_restart (as standby)" << dendl;
1216     mdlog->get_journaler()->reread_head_and_probe(
1217       new C_MDS_StandbyReplayRestartFinish(
1218         this,
1219         mdlog->get_journaler()->get_read_pos()));
1220   } else {
1221     /* We are transitioning out of standby: wait for OSD map update
1222        before making final pass */
1223     dout(1) << "standby_replay_restart (final takeover pass)" << dendl;
1224     Context *fin = new C_IO_Wrapper(this, new C_MDS_BootStart(this, MDS_BOOT_PREPARE_LOG));
1225     bool const ready =
1226       objecter->wait_for_map(mdsmap->get_last_failure_osd_epoch(), fin);
1227     if (ready) {
1228       delete fin;
1229       mdlog->get_journaler()->reread_head_and_probe(
1230         new C_MDS_StandbyReplayRestartFinish(
1231           this,
1232           mdlog->get_journaler()->get_read_pos()));
1233     } else {
1234       dout(1) << " waiting for osdmap " << mdsmap->get_last_failure_osd_epoch()
1235               << " (which blacklists prior instance)" << dendl;
1236     }
1237   }
1238 }
1239
1240 class MDSRank::C_MDS_StandbyReplayRestart : public MDSInternalContext {
1241 public:
1242   explicit C_MDS_StandbyReplayRestart(MDSRank *m) : MDSInternalContext(m) {}
1243   void finish(int r) override {
1244     assert(!r);
1245     mds->standby_replay_restart();
1246   }
1247 };
1248
1249 void MDSRank::replay_done()
1250 {
1251   dout(1) << "replay_done" << (standby_replaying ? " (as standby)" : "") << dendl;
1252
1253   if (is_standby_replay()) {
1254     // The replay was done in standby state, and we are still in that state
1255     assert(standby_replaying);
1256     dout(10) << "setting replay timer" << dendl;
1257     timer.add_event_after(g_conf->mds_replay_interval,
1258                           new C_MDS_StandbyReplayRestart(this));
1259     return;
1260   } else if (standby_replaying) {
1261     // The replay was done in standby state, we have now _left_ that state
1262     dout(10) << " last replay pass was as a standby; making final pass" << dendl;
1263     standby_replaying = false;
1264     standby_replay_restart();
1265     return;
1266   } else {
1267     // Replay is complete, journal read should be up to date
1268     assert(mdlog->get_journaler()->get_read_pos() == mdlog->get_journaler()->get_write_pos());
1269     assert(!is_standby_replay());
1270
1271     // Reformat and come back here
1272     if (mdlog->get_journaler()->get_stream_format() < g_conf->mds_journal_format) {
1273         dout(4) << "reformatting journal on standbyreplay->replay transition" << dendl;
1274         mdlog->reopen(new C_MDS_BootStart(this, MDS_BOOT_REPLAY_DONE));
1275         return;
1276     }
1277   }
1278
1279   dout(1) << "making mds journal writeable" << dendl;
1280   mdlog->get_journaler()->set_writeable();
1281   mdlog->get_journaler()->trim_tail();
1282
1283   if (g_conf->mds_wipe_sessions) {
1284     dout(1) << "wiping out client sessions" << dendl;
1285     sessionmap.wipe();
1286     sessionmap.save(new C_MDSInternalNoop);
1287   }
1288   if (g_conf->mds_wipe_ino_prealloc) {
1289     dout(1) << "wiping out ino prealloc from sessions" << dendl;
1290     sessionmap.wipe_ino_prealloc();
1291     sessionmap.save(new C_MDSInternalNoop);
1292   }
1293   if (g_conf->mds_skip_ino) {
1294     inodeno_t i = g_conf->mds_skip_ino;
1295     dout(1) << "skipping " << i << " inodes" << dendl;
1296     inotable->skip_inos(i);
1297     inotable->save(new C_MDSInternalNoop);
1298   }
1299
1300   if (mdsmap->get_num_in_mds() == 1 &&
1301       mdsmap->get_num_failed_mds() == 0) { // just me!
1302     dout(2) << "i am alone, moving to state reconnect" << dendl;
1303     request_state(MDSMap::STATE_RECONNECT);
1304   } else {
1305     dout(2) << "i am not alone, moving to state resolve" << dendl;
1306     request_state(MDSMap::STATE_RESOLVE);
1307   }
1308 }
1309
1310 void MDSRank::reopen_log()
1311 {
1312   dout(1) << "reopen_log" << dendl;
1313   mdcache->rollback_uncommitted_fragments();
1314 }
1315
1316
1317 void MDSRank::resolve_start()
1318 {
1319   dout(1) << "resolve_start" << dendl;
1320
1321   reopen_log();
1322
1323   mdcache->resolve_start(new C_MDS_VoidFn(this, &MDSRank::resolve_done));
1324   finish_contexts(g_ceph_context, waiting_for_resolve);
1325 }
1326 void MDSRank::resolve_done()
1327 {
1328   dout(1) << "resolve_done" << dendl;
1329   request_state(MDSMap::STATE_RECONNECT);
1330 }
1331
1332 void MDSRank::reconnect_start()
1333 {
1334   dout(1) << "reconnect_start" << dendl;
1335
1336   if (last_state == MDSMap::STATE_REPLAY) {
1337     reopen_log();
1338   }
1339
1340   // Drop any blacklisted clients from the SessionMap before going
1341   // into reconnect, so that we don't wait for them.
1342   objecter->enable_blacklist_events();
1343   std::set<entity_addr_t> blacklist;
1344   epoch_t epoch = 0;
1345   objecter->with_osdmap([this, &blacklist, &epoch](const OSDMap& o) {
1346       o.get_blacklist(&blacklist);
1347       epoch = o.get_epoch();
1348   });
1349   auto killed = server->apply_blacklist(blacklist);
1350   dout(4) << "reconnect_start: killed " << killed << " blacklisted sessions ("
1351           << blacklist.size() << " blacklist entries, "
1352           << sessionmap.get_sessions().size() << ")" << dendl;
1353   if (killed) {
1354     set_osd_epoch_barrier(epoch);
1355   }
1356
1357   server->reconnect_clients(new C_MDS_VoidFn(this, &MDSRank::reconnect_done));
1358   finish_contexts(g_ceph_context, waiting_for_reconnect);
1359 }
1360 void MDSRank::reconnect_done()
1361 {
1362   dout(1) << "reconnect_done" << dendl;
1363   request_state(MDSMap::STATE_REJOIN);    // move to rejoin state
1364 }
1365
1366 void MDSRank::rejoin_joint_start()
1367 {
1368   dout(1) << "rejoin_joint_start" << dendl;
1369   mdcache->rejoin_send_rejoins();
1370 }
1371 void MDSRank::rejoin_start()
1372 {
1373   dout(1) << "rejoin_start" << dendl;
1374   mdcache->rejoin_start(new C_MDS_VoidFn(this, &MDSRank::rejoin_done));
1375 }
1376 void MDSRank::rejoin_done()
1377 {
1378   dout(1) << "rejoin_done" << dendl;
1379   mdcache->show_subtrees();
1380   mdcache->show_cache();
1381
1382   // funny case: is our cache empty?  no subtrees?
1383   if (!mdcache->is_subtrees()) {
1384     if (whoami == 0) {
1385       // The root should always have a subtree!
1386       clog->error() << "No subtrees found for root MDS rank!";
1387       damaged();
1388       assert(mdcache->is_subtrees());
1389     } else {
1390       dout(1) << " empty cache, no subtrees, leaving cluster" << dendl;
1391       request_state(MDSMap::STATE_STOPPED);
1392     }
1393     return;
1394   }
1395
1396   if (replay_queue.empty())
1397     request_state(MDSMap::STATE_ACTIVE);
1398   else
1399     request_state(MDSMap::STATE_CLIENTREPLAY);
1400 }
1401
1402 void MDSRank::clientreplay_start()
1403 {
1404   dout(1) << "clientreplay_start" << dendl;
1405   finish_contexts(g_ceph_context, waiting_for_replay);  // kick waiters
1406   mdcache->start_files_to_recover();
1407   queue_one_replay();
1408 }
1409
1410 bool MDSRank::queue_one_replay()
1411 {
1412   if (replay_queue.empty()) {
1413     mdlog->wait_for_safe(new C_MDS_VoidFn(this, &MDSRank::clientreplay_done));
1414     return false;
1415   }
1416   queue_waiter(replay_queue.front());
1417   replay_queue.pop_front();
1418   return true;
1419 }
1420
1421 void MDSRank::clientreplay_done()
1422 {
1423   dout(1) << "clientreplay_done" << dendl;
1424   request_state(MDSMap::STATE_ACTIVE);
1425 }
1426
1427 void MDSRank::active_start()
1428 {
1429   dout(1) << "active_start" << dendl;
1430
1431   if (last_state == MDSMap::STATE_CREATING) {
1432     mdcache->open_root();
1433   }
1434
1435   mdcache->clean_open_file_lists();
1436   mdcache->export_remaining_imported_caps();
1437   finish_contexts(g_ceph_context, waiting_for_replay);  // kick waiters
1438   mdcache->start_files_to_recover();
1439
1440   mdcache->reissue_all_caps();
1441   mdcache->activate_stray_manager();
1442
1443   finish_contexts(g_ceph_context, waiting_for_active);  // kick waiters
1444 }
1445
1446 void MDSRank::recovery_done(int oldstate)
1447 {
1448   dout(1) << "recovery_done -- successful recovery!" << dendl;
1449   assert(is_clientreplay() || is_active());
1450
1451   // kick snaptable (resent AGREEs)
1452   if (mdsmap->get_tableserver() == whoami) {
1453     set<mds_rank_t> active;
1454     mdsmap->get_clientreplay_or_active_or_stopping_mds_set(active);
1455     snapserver->finish_recovery(active);
1456   }
1457
1458   if (oldstate == MDSMap::STATE_CREATING)
1459     return;
1460
1461   mdcache->start_recovered_truncates();
1462   mdcache->do_file_recover();
1463
1464   // tell connected clients
1465   //bcast_mds_map();     // not anymore, they get this from the monitor
1466
1467   mdcache->populate_mydir();
1468 }
1469
1470 void MDSRank::creating_done()
1471 {
1472   dout(1)<< "creating_done" << dendl;
1473   request_state(MDSMap::STATE_ACTIVE);
1474 }
1475
1476 void MDSRank::boot_create()
1477 {
1478   dout(3) << "boot_create" << dendl;
1479
1480   MDSGatherBuilder fin(g_ceph_context, new C_MDS_VoidFn(this, &MDSRank::creating_done));
1481
1482   mdcache->init_layouts();
1483
1484   snapserver->set_rank(whoami);
1485   inotable->set_rank(whoami);
1486   sessionmap.set_rank(whoami);
1487
1488   // start with a fresh journal
1489   dout(10) << "boot_create creating fresh journal" << dendl;
1490   mdlog->create(fin.new_sub());
1491
1492   // open new journal segment, but do not journal subtree map (yet)
1493   mdlog->prepare_new_segment();
1494
1495   if (whoami == mdsmap->get_root()) {
1496     dout(3) << "boot_create creating fresh hierarchy" << dendl;
1497     mdcache->create_empty_hierarchy(fin.get());
1498   }
1499
1500   dout(3) << "boot_create creating mydir hierarchy" << dendl;
1501   mdcache->create_mydir_hierarchy(fin.get());
1502
1503   // fixme: fake out inotable (reset, pretend loaded)
1504   dout(10) << "boot_create creating fresh inotable table" << dendl;
1505   inotable->reset();
1506   inotable->save(fin.new_sub());
1507
1508   // write empty sessionmap
1509   sessionmap.save(fin.new_sub());
1510
1511   // Create empty purge queue
1512   purge_queue.create(new C_IO_Wrapper(this, fin.new_sub()));
1513
1514   // initialize tables
1515   if (mdsmap->get_tableserver() == whoami) {
1516     dout(10) << "boot_create creating fresh snaptable" << dendl;
1517     snapserver->reset();
1518     snapserver->save(fin.new_sub());
1519   }
1520
1521   assert(g_conf->mds_kill_create_at != 1);
1522
1523   // ok now journal it
1524   mdlog->journal_segment_subtree_map(fin.new_sub());
1525   mdlog->flush();
1526
1527   // Usually we do this during reconnect, but creation skips that.
1528   objecter->enable_blacklist_events();
1529
1530   fin.activate();
1531 }
1532
1533 void MDSRank::stopping_start()
1534 {
1535   dout(2) << "stopping_start" << dendl;
1536
1537   if (mdsmap->get_num_in_mds() == 1 && !sessionmap.empty()) {
1538     // we're the only mds up!
1539     dout(0) << "we are the last MDS, and have mounted clients: we cannot flush our journal.  suicide!" << dendl;
1540     suicide();
1541   }
1542
1543   mdcache->shutdown_start();
1544 }
1545
1546 void MDSRank::stopping_done()
1547 {
1548   dout(2) << "stopping_done" << dendl;
1549
1550   // tell monitor we shut down cleanly.
1551   request_state(MDSMap::STATE_STOPPED);
1552 }
1553
1554 void MDSRankDispatcher::handle_mds_map(
1555     MMDSMap *m,
1556     MDSMap *oldmap)
1557 {
1558   // I am only to be passed MDSMaps in which I hold a rank
1559   assert(whoami != MDS_RANK_NONE);
1560
1561   MDSMap::DaemonState oldstate = state;
1562   mds_gid_t mds_gid = mds_gid_t(monc->get_global_id());
1563   state = mdsmap->get_state_gid(mds_gid);
1564   if (state != oldstate) {
1565     last_state = oldstate;
1566     incarnation = mdsmap->get_inc_gid(mds_gid);
1567   }
1568
1569   version_t epoch = m->get_epoch();
1570
1571   // note source's map version
1572   if (m->get_source().is_mds() &&
1573       peer_mdsmap_epoch[mds_rank_t(m->get_source().num())] < epoch) {
1574     dout(15) << " peer " << m->get_source()
1575              << " has mdsmap epoch >= " << epoch
1576              << dendl;
1577     peer_mdsmap_epoch[mds_rank_t(m->get_source().num())] = epoch;
1578   }
1579
1580   // Validate state transitions while I hold a rank
1581   if (!MDSMap::state_transition_valid(oldstate, state)) {
1582     derr << "Invalid state transition " << ceph_mds_state_name(oldstate)
1583       << "->" << ceph_mds_state_name(state) << dendl;
1584     respawn();
1585   }
1586
1587   if (oldstate != state) {
1588     // update messenger.
1589     if (state == MDSMap::STATE_STANDBY_REPLAY) {
1590       dout(1) << "handle_mds_map i am now mds." << mds_gid << "." << incarnation
1591               << " replaying mds." << whoami << "." << incarnation << dendl;
1592       messenger->set_myname(entity_name_t::MDS(mds_gid));
1593     } else {
1594       dout(1) << "handle_mds_map i am now mds." << whoami << "." << incarnation << dendl;
1595       messenger->set_myname(entity_name_t::MDS(whoami));
1596     }
1597   }
1598
1599   // tell objecter my incarnation
1600   if (objecter->get_client_incarnation() != incarnation)
1601     objecter->set_client_incarnation(incarnation);
1602
1603   // for debug
1604   if (g_conf->mds_dump_cache_on_map)
1605     mdcache->dump_cache();
1606
1607   // did it change?
1608   if (oldstate != state) {
1609     dout(1) << "handle_mds_map state change "
1610             << ceph_mds_state_name(oldstate) << " --> "
1611             << ceph_mds_state_name(state) << dendl;
1612     beacon.set_want_state(mdsmap, state);
1613
1614     if (oldstate == MDSMap::STATE_STANDBY_REPLAY) {
1615         dout(10) << "Monitor activated us! Deactivating replay loop" << dendl;
1616         assert (state == MDSMap::STATE_REPLAY);
1617     } else {
1618       // did i just recover?
1619       if ((is_active() || is_clientreplay()) &&
1620           (oldstate == MDSMap::STATE_CREATING ||
1621            oldstate == MDSMap::STATE_REJOIN ||
1622            oldstate == MDSMap::STATE_RECONNECT))
1623         recovery_done(oldstate);
1624
1625       if (is_active()) {
1626         active_start();
1627       } else if (is_any_replay()) {
1628         replay_start();
1629       } else if (is_resolve()) {
1630         resolve_start();
1631       } else if (is_reconnect()) {
1632         reconnect_start();
1633       } else if (is_rejoin()) {
1634         rejoin_start();
1635       } else if (is_clientreplay()) {
1636         clientreplay_start();
1637       } else if (is_creating()) {
1638         boot_create();
1639       } else if (is_starting()) {
1640         boot_start();
1641       } else if (is_stopping()) {
1642         assert(oldstate == MDSMap::STATE_ACTIVE);
1643         stopping_start();
1644       }
1645     }
1646   }
1647
1648   // RESOLVE
1649   // is someone else newly resolving?
1650   if (is_resolve() || is_reconnect() || is_rejoin() ||
1651       is_clientreplay() || is_active() || is_stopping()) {
1652     if (!oldmap->is_resolving() && mdsmap->is_resolving()) {
1653       set<mds_rank_t> resolve;
1654       mdsmap->get_mds_set(resolve, MDSMap::STATE_RESOLVE);
1655       dout(10) << " resolve set is " << resolve << dendl;
1656       calc_recovery_set();
1657       mdcache->send_resolves();
1658     }
1659   }
1660
1661   // REJOIN
1662   // is everybody finally rejoining?
1663   if (is_rejoin() || is_clientreplay() || is_active() || is_stopping()) {
1664     // did we start?
1665     if (!oldmap->is_rejoining() && mdsmap->is_rejoining())
1666       rejoin_joint_start();
1667
1668     // did we finish?
1669     if (g_conf->mds_dump_cache_after_rejoin &&
1670         oldmap->is_rejoining() && !mdsmap->is_rejoining())
1671       mdcache->dump_cache();      // for DEBUG only
1672
1673     if (oldstate >= MDSMap::STATE_REJOIN) {
1674       // ACTIVE|CLIENTREPLAY|REJOIN => we can discover from them.
1675       set<mds_rank_t> olddis, dis;
1676       oldmap->get_mds_set(olddis, MDSMap::STATE_ACTIVE);
1677       oldmap->get_mds_set(olddis, MDSMap::STATE_CLIENTREPLAY);
1678       oldmap->get_mds_set(olddis, MDSMap::STATE_REJOIN);
1679       mdsmap->get_mds_set(dis, MDSMap::STATE_ACTIVE);
1680       mdsmap->get_mds_set(dis, MDSMap::STATE_CLIENTREPLAY);
1681       mdsmap->get_mds_set(dis, MDSMap::STATE_REJOIN);
1682       for (set<mds_rank_t>::iterator p = dis.begin(); p != dis.end(); ++p)
1683         if (*p != whoami &&            // not me
1684             olddis.count(*p) == 0) {  // newly so?
1685           mdcache->kick_discovers(*p);
1686           mdcache->kick_open_ino_peers(*p);
1687         }
1688     }
1689   }
1690
1691   cluster_degraded = mdsmap->is_degraded();
1692   if (oldmap->is_degraded() && !cluster_degraded && state >= MDSMap::STATE_ACTIVE) {
1693     dout(1) << "cluster recovered." << dendl;
1694     auto it = waiting_for_active_peer.find(MDS_RANK_NONE);
1695     if (it != waiting_for_active_peer.end()) {
1696       queue_waiters(it->second);
1697       waiting_for_active_peer.erase(it);
1698     }
1699   }
1700
1701   // did someone go active?
1702   if (oldstate >= MDSMap::STATE_CLIENTREPLAY &&
1703       (is_clientreplay() || is_active() || is_stopping())) {
1704     set<mds_rank_t> oldactive, active;
1705     oldmap->get_mds_set(oldactive, MDSMap::STATE_ACTIVE);
1706     oldmap->get_mds_set(oldactive, MDSMap::STATE_CLIENTREPLAY);
1707     mdsmap->get_mds_set(active, MDSMap::STATE_ACTIVE);
1708     mdsmap->get_mds_set(active, MDSMap::STATE_CLIENTREPLAY);
1709     for (set<mds_rank_t>::iterator p = active.begin(); p != active.end(); ++p)
1710       if (*p != whoami &&            // not me
1711           oldactive.count(*p) == 0)  // newly so?
1712         handle_mds_recovery(*p);
1713   }
1714
1715   // did someone fail?
1716   //   new down?
1717   {
1718     set<mds_rank_t> olddown, down;
1719     oldmap->get_down_mds_set(&olddown);
1720     mdsmap->get_down_mds_set(&down);
1721     for (set<mds_rank_t>::iterator p = down.begin(); p != down.end(); ++p) {
1722       if (oldmap->have_inst(*p) && olddown.count(*p) == 0) {
1723         messenger->mark_down(oldmap->get_inst(*p).addr);
1724         handle_mds_failure(*p);
1725       }
1726     }
1727   }
1728
1729   // did someone fail?
1730   //   did their addr/inst change?
1731   {
1732     set<mds_rank_t> up;
1733     mdsmap->get_up_mds_set(up);
1734     for (set<mds_rank_t>::iterator p = up.begin(); p != up.end(); ++p) {
1735       if (oldmap->have_inst(*p) &&
1736          oldmap->get_inst(*p) != mdsmap->get_inst(*p)) {
1737         messenger->mark_down(oldmap->get_inst(*p).addr);
1738         handle_mds_failure(*p);
1739       }
1740     }
1741   }
1742
1743   if (is_clientreplay() || is_active() || is_stopping()) {
1744     // did anyone stop?
1745     set<mds_rank_t> oldstopped, stopped;
1746     oldmap->get_stopped_mds_set(oldstopped);
1747     mdsmap->get_stopped_mds_set(stopped);
1748     for (set<mds_rank_t>::iterator p = stopped.begin(); p != stopped.end(); ++p)
1749       if (oldstopped.count(*p) == 0)      // newly so?
1750         mdcache->migrator->handle_mds_failure_or_stop(*p);
1751   }
1752
1753   {
1754     map<epoch_t,list<MDSInternalContextBase*> >::iterator p = waiting_for_mdsmap.begin();
1755     while (p != waiting_for_mdsmap.end() && p->first <= mdsmap->get_epoch()) {
1756       list<MDSInternalContextBase*> ls;
1757       ls.swap(p->second);
1758       waiting_for_mdsmap.erase(p++);
1759       finish_contexts(g_ceph_context, ls);
1760     }
1761   }
1762
1763   if (is_active()) {
1764     // Before going active, set OSD epoch barrier to latest (so that
1765     // we don't risk handing out caps to clients with old OSD maps that
1766     // might not include barriers from the previous incarnation of this MDS)
1767     set_osd_epoch_barrier(objecter->with_osdmap(
1768                             std::mem_fn(&OSDMap::get_epoch)));
1769   }
1770
1771   if (is_active()) {
1772     bool found = false;
1773     MDSMap::mds_info_t info = mdsmap->get_info(whoami);
1774
1775     for (map<mds_gid_t,MDSMap::mds_info_t>::const_iterator p = mdsmap->get_mds_info().begin();
1776        p != mdsmap->get_mds_info().end();
1777        ++p) {
1778       if (p->second.state == MDSMap::STATE_STANDBY_REPLAY &&
1779           (p->second.standby_for_rank == whoami ||(info.name.length() && p->second.standby_for_name == info.name))) {
1780         found = true;
1781         break;
1782       }
1783       if (found)
1784         mdlog->set_write_iohint(0);
1785       else
1786         mdlog->set_write_iohint(CEPH_OSD_OP_FLAG_FADVISE_DONTNEED);
1787     }
1788   }
1789
1790   if (oldmap->get_max_mds() != mdsmap->get_max_mds()) {
1791     purge_queue.update_op_limit(*mdsmap);
1792   }
1793 }
1794
1795 void MDSRank::handle_mds_recovery(mds_rank_t who)
1796 {
1797   dout(5) << "handle_mds_recovery mds." << who << dendl;
1798
1799   mdcache->handle_mds_recovery(who);
1800
1801   if (mdsmap->get_tableserver() == whoami) {
1802     snapserver->handle_mds_recovery(who);
1803   }
1804
1805   queue_waiters(waiting_for_active_peer[who]);
1806   waiting_for_active_peer.erase(who);
1807 }
1808
1809 void MDSRank::handle_mds_failure(mds_rank_t who)
1810 {
1811   if (who == whoami) {
1812     dout(5) << "handle_mds_failure for myself; not doing anything" << dendl;
1813     return;
1814   }
1815   dout(5) << "handle_mds_failure mds." << who << dendl;
1816
1817   mdcache->handle_mds_failure(who);
1818
1819   snapclient->handle_mds_failure(who);
1820 }
1821
1822 bool MDSRankDispatcher::handle_asok_command(
1823     std::string command, cmdmap_t& cmdmap, Formatter *f,
1824                     std::ostream& ss)
1825 {
1826   if (command == "dump_ops_in_flight" ||
1827              command == "ops") {
1828     if (!op_tracker.dump_ops_in_flight(f)) {
1829       ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
1830           please enable \"osd_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
1831     }
1832   } else if (command == "dump_blocked_ops") {
1833     if (!op_tracker.dump_ops_in_flight(f, true)) {
1834       ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
1835         Please enable \"osd_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
1836     }
1837   } else if (command == "dump_historic_ops") {
1838     if (!op_tracker.dump_historic_ops(f)) {
1839       ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
1840           please enable \"osd_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
1841     }
1842   } else if (command == "dump_historic_ops_by_duration") {
1843     if (!op_tracker.dump_historic_ops(f, true)) {
1844       ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
1845           please enable \"osd_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
1846     }
1847   } else if (command == "osdmap barrier") {
1848     int64_t target_epoch = 0;
1849     bool got_val = cmd_getval(g_ceph_context, cmdmap, "target_epoch", target_epoch);
1850
1851     if (!got_val) {
1852       ss << "no target epoch given";
1853       return true;
1854     }
1855
1856     mds_lock.Lock();
1857     set_osd_epoch_barrier(target_epoch);
1858     mds_lock.Unlock();
1859
1860     C_SaferCond cond;
1861     bool already_got = objecter->wait_for_map(target_epoch, &cond);
1862     if (!already_got) {
1863       dout(4) << __func__ << ": waiting for OSD epoch " << target_epoch << dendl;
1864       cond.wait();
1865     }
1866   } else if (command == "session ls") {
1867     Mutex::Locker l(mds_lock);
1868
1869     heartbeat_reset();
1870
1871     dump_sessions(SessionFilter(), f);
1872   } else if (command == "session evict") {
1873     std::string client_id;
1874     const bool got_arg = cmd_getval(g_ceph_context, cmdmap, "client_id", client_id);
1875     if(!got_arg) {
1876       ss << "Invalid client_id specified";
1877       return true;
1878     }
1879
1880     mds_lock.Lock();
1881     std::stringstream dss;
1882     bool evicted = evict_client(strtol(client_id.c_str(), 0, 10), true,
1883         g_conf->mds_session_blacklist_on_evict, dss);
1884     if (!evicted) {
1885       dout(15) << dss.str() << dendl;
1886       ss << dss.str();
1887     }
1888     mds_lock.Unlock();
1889   } else if (command == "scrub_path") {
1890     string path;
1891     vector<string> scrubop_vec;
1892     cmd_getval(g_ceph_context, cmdmap, "scrubops", scrubop_vec);
1893     cmd_getval(g_ceph_context, cmdmap, "path", path);
1894     command_scrub_path(f, path, scrubop_vec);
1895   } else if (command == "tag path") {
1896     string path;
1897     cmd_getval(g_ceph_context, cmdmap, "path", path);
1898     string tag;
1899     cmd_getval(g_ceph_context, cmdmap, "tag", tag);
1900     command_tag_path(f, path, tag);
1901   } else if (command == "flush_path") {
1902     string path;
1903     cmd_getval(g_ceph_context, cmdmap, "path", path);
1904     command_flush_path(f, path);
1905   } else if (command == "flush journal") {
1906     command_flush_journal(f);
1907   } else if (command == "get subtrees") {
1908     command_get_subtrees(f);
1909   } else if (command == "export dir") {
1910     string path;
1911     if(!cmd_getval(g_ceph_context, cmdmap, "path", path)) {
1912       ss << "malformed path";
1913       return true;
1914     }
1915     int64_t rank;
1916     if(!cmd_getval(g_ceph_context, cmdmap, "rank", rank)) {
1917       ss << "malformed rank";
1918       return true;
1919     }
1920     command_export_dir(f, path, (mds_rank_t)rank);
1921   } else if (command == "dump cache") {
1922     Mutex::Locker l(mds_lock);
1923     string path;
1924     int r;
1925     if(!cmd_getval(g_ceph_context, cmdmap, "path", path)) {
1926       r = mdcache->dump_cache(f);
1927     } else {
1928       r = mdcache->dump_cache(path);
1929     }
1930
1931     if (r != 0) {
1932       ss << "Failed to dump cache: " << cpp_strerror(r);
1933     }
1934   } else if (command == "dump tree") {
1935     string root;
1936     int64_t depth;
1937     cmd_getval(g_ceph_context, cmdmap, "root", root);
1938     if (!cmd_getval(g_ceph_context, cmdmap, "depth", depth))
1939       depth = -1;
1940     {
1941       Mutex::Locker l(mds_lock);
1942       int r = mdcache->dump_cache(root, depth, f);
1943       if (r != 0) {
1944         ss << "Failed to dump tree: " << cpp_strerror(r);
1945       }
1946     }
1947   } else if (command == "force_readonly") {
1948     Mutex::Locker l(mds_lock);
1949     mdcache->force_readonly();
1950   } else if (command == "dirfrag split") {
1951     command_dirfrag_split(cmdmap, ss);
1952   } else if (command == "dirfrag merge") {
1953     command_dirfrag_merge(cmdmap, ss);
1954   } else if (command == "dirfrag ls") {
1955     command_dirfrag_ls(cmdmap, ss, f);
1956   } else {
1957     return false;
1958   }
1959
1960   return true;
1961 }
1962
1963 class C_MDS_Send_Command_Reply : public MDSInternalContext
1964 {
1965 protected:
1966   MCommand *m;
1967 public:
1968   C_MDS_Send_Command_Reply(MDSRank *_mds, MCommand *_m) :
1969     MDSInternalContext(_mds), m(_m) { m->get(); }
1970   void send (int r, const std::string& out_str) {
1971     bufferlist bl;
1972     MDSDaemon::send_command_reply(m, mds, r, bl, out_str);
1973     m->put();
1974   }
1975   void finish (int r) override {
1976     send(r, "");
1977   }
1978 };
1979
1980 /**
1981  * This function drops the mds_lock, so don't do anything with
1982  * MDSRank after calling it (we could have gone into shutdown): just
1983  * send your result back to the calling client and finish.
1984  */
1985 void MDSRankDispatcher::evict_clients(const SessionFilter &filter, MCommand *m)
1986 {
1987   C_MDS_Send_Command_Reply *reply = new C_MDS_Send_Command_Reply(this, m);
1988
1989   if (is_any_replay()) {
1990     reply->send(-EAGAIN, "MDS is replaying log");
1991     delete reply;
1992     return;
1993   }
1994
1995   std::list<Session*> victims;
1996   const auto sessions = sessionmap.get_sessions();
1997   for (const auto p : sessions)  {
1998     if (!p.first.is_client()) {
1999       continue;
2000     }
2001
2002     Session *s = p.second;
2003
2004     if (filter.match(*s, std::bind(&Server::waiting_for_reconnect, server, std::placeholders::_1))) {
2005       victims.push_back(s);
2006     }
2007   }
2008
2009   dout(20) << __func__ << " matched " << victims.size() << " sessions" << dendl;
2010
2011   if (victims.empty()) {
2012     reply->send(0, "");
2013     delete reply;
2014     return;
2015   }
2016
2017   C_GatherBuilder gather(g_ceph_context, reply);
2018   for (const auto s : victims) {
2019     std::stringstream ss;
2020     evict_client(s->info.inst.name.num(), false,
2021                  g_conf->mds_session_blacklist_on_evict, ss, gather.new_sub());
2022   }
2023   gather.activate();
2024 }
2025
2026 void MDSRankDispatcher::dump_sessions(const SessionFilter &filter, Formatter *f) const
2027 {
2028   // Dump sessions, decorated with recovery/replay status
2029   f->open_array_section("sessions");
2030   const ceph::unordered_map<entity_name_t, Session*> session_map = sessionmap.get_sessions();
2031   for (ceph::unordered_map<entity_name_t,Session*>::const_iterator p = session_map.begin();
2032        p != session_map.end();
2033        ++p)  {
2034     if (!p->first.is_client()) {
2035       continue;
2036     }
2037
2038     Session *s = p->second;
2039
2040     if (!filter.match(*s, std::bind(&Server::waiting_for_reconnect, server, std::placeholders::_1))) {
2041       continue;
2042     }
2043
2044     f->open_object_section("session");
2045     f->dump_int("id", p->first.num());
2046
2047     f->dump_int("num_leases", s->leases.size());
2048     f->dump_int("num_caps", s->caps.size());
2049
2050     f->dump_string("state", s->get_state_name());
2051     f->dump_int("replay_requests", is_clientreplay() ? s->get_request_count() : 0);
2052     f->dump_unsigned("completed_requests", s->get_num_completed_requests());
2053     f->dump_bool("reconnecting", server->waiting_for_reconnect(p->first.num()));
2054     f->dump_stream("inst") << s->info.inst;
2055     f->open_object_section("client_metadata");
2056     for (map<string, string>::const_iterator i = s->info.client_metadata.begin();
2057          i != s->info.client_metadata.end(); ++i) {
2058       f->dump_string(i->first.c_str(), i->second);
2059     }
2060     f->close_section(); // client_metadata
2061     f->close_section(); //session
2062   }
2063   f->close_section(); //sessions
2064 }
2065
2066 void MDSRank::command_scrub_path(Formatter *f, const string& path, vector<string>& scrubop_vec)
2067 {
2068   bool force = false;
2069   bool recursive = false;
2070   bool repair = false;
2071   for (vector<string>::iterator i = scrubop_vec.begin() ; i != scrubop_vec.end(); ++i) {
2072     if (*i == "force")
2073       force = true;
2074     else if (*i == "recursive")
2075       recursive = true;
2076     else if (*i == "repair")
2077       repair = true;
2078   }
2079   C_SaferCond scond;
2080   {
2081     Mutex::Locker l(mds_lock);
2082     mdcache->enqueue_scrub(path, "", force, recursive, repair, f, &scond);
2083   }
2084   scond.wait();
2085   // scrub_dentry() finishers will dump the data for us; we're done!
2086 }
2087
2088 void MDSRank::command_tag_path(Formatter *f,
2089     const string& path, const std::string &tag)
2090 {
2091   C_SaferCond scond;
2092   {
2093     Mutex::Locker l(mds_lock);
2094     mdcache->enqueue_scrub(path, tag, true, true, false, f, &scond);
2095   }
2096   scond.wait();
2097 }
2098
2099 void MDSRank::command_flush_path(Formatter *f, const string& path)
2100 {
2101   C_SaferCond scond;
2102   {
2103     Mutex::Locker l(mds_lock);
2104     mdcache->flush_dentry(path, &scond);
2105   }
2106   int r = scond.wait();
2107   f->open_object_section("results");
2108   f->dump_int("return_code", r);
2109   f->close_section(); // results
2110 }
2111
2112 /**
2113  * Wrapper around _command_flush_journal that
2114  * handles serialization of result
2115  */
2116 void MDSRank::command_flush_journal(Formatter *f)
2117 {
2118   assert(f != NULL);
2119
2120   std::stringstream ss;
2121   const int r = _command_flush_journal(&ss);
2122   f->open_object_section("result");
2123   f->dump_string("message", ss.str());
2124   f->dump_int("return_code", r);
2125   f->close_section();
2126 }
2127
2128 /**
2129  * Implementation of "flush journal" asok command.
2130  *
2131  * @param ss
2132  * Optionally populate with a human readable string describing the
2133  * reason for any unexpected return status.
2134  */
2135 int MDSRank::_command_flush_journal(std::stringstream *ss)
2136 {
2137   assert(ss != NULL);
2138
2139   Mutex::Locker l(mds_lock);
2140
2141   if (mdcache->is_readonly()) {
2142     dout(5) << __func__ << ": read-only FS" << dendl;
2143     return -EROFS;
2144   }
2145
2146   if (!is_active()) {
2147     dout(5) << __func__ << ": MDS not active, no-op" << dendl;
2148     return 0;
2149   }
2150
2151   // I need to seal off the current segment, and then mark all previous segments
2152   // for expiry
2153   mdlog->start_new_segment();
2154   int r = 0;
2155
2156   // Flush initially so that all the segments older than our new one
2157   // will be elegible for expiry
2158   {
2159     C_SaferCond mdlog_flushed;
2160     mdlog->flush();
2161     mdlog->wait_for_safe(new MDSInternalContextWrapper(this, &mdlog_flushed));
2162     mds_lock.Unlock();
2163     r = mdlog_flushed.wait();
2164     mds_lock.Lock();
2165     if (r != 0) {
2166       *ss << "Error " << r << " (" << cpp_strerror(r) << ") while flushing journal";
2167       return r;
2168     }
2169   }
2170
2171   // Because we may not be the last wait_for_safe context on MDLog, and
2172   // subsequent contexts might wake up in the middle of our later trim_all
2173   // and interfere with expiry (by e.g. marking dirs/dentries dirty
2174   // on previous log segments), we run a second wait_for_safe here.
2175   // See #10368
2176   {
2177     C_SaferCond mdlog_cleared;
2178     mdlog->wait_for_safe(new MDSInternalContextWrapper(this, &mdlog_cleared));
2179     mds_lock.Unlock();
2180     r = mdlog_cleared.wait();
2181     mds_lock.Lock();
2182     if (r != 0) {
2183       *ss << "Error " << r << " (" << cpp_strerror(r) << ") while flushing journal";
2184       return r;
2185     }
2186   }
2187
2188   // Put all the old log segments into expiring or expired state
2189   dout(5) << __func__ << ": beginning segment expiry" << dendl;
2190   r = mdlog->trim_all();
2191   if (r != 0) {
2192     *ss << "Error " << r << " (" << cpp_strerror(r) << ") while trimming log";
2193     return r;
2194   }
2195
2196   // Attach contexts to wait for all expiring segments to expire
2197   MDSGatherBuilder expiry_gather(g_ceph_context);
2198
2199   const std::set<LogSegment*> &expiring_segments = mdlog->get_expiring_segments();
2200   for (std::set<LogSegment*>::const_iterator i = expiring_segments.begin();
2201        i != expiring_segments.end(); ++i) {
2202     (*i)->wait_for_expiry(expiry_gather.new_sub());
2203   }
2204   dout(5) << __func__ << ": waiting for " << expiry_gather.num_subs_created()
2205           << " segments to expire" << dendl;
2206
2207   if (expiry_gather.has_subs()) {
2208     C_SaferCond cond;
2209     expiry_gather.set_finisher(new MDSInternalContextWrapper(this, &cond));
2210     expiry_gather.activate();
2211
2212     // Drop mds_lock to allow progress until expiry is complete
2213     mds_lock.Unlock();
2214     int r = cond.wait();
2215     mds_lock.Lock();
2216
2217     assert(r == 0);  // MDLog is not allowed to raise errors via wait_for_expiry
2218   }
2219
2220   dout(5) << __func__ << ": expiry complete, expire_pos/trim_pos is now " << std::hex <<
2221     mdlog->get_journaler()->get_expire_pos() << "/" <<
2222     mdlog->get_journaler()->get_trimmed_pos() << dendl;
2223
2224   // Now everyone I'm interested in is expired
2225   mdlog->trim_expired_segments();
2226
2227   dout(5) << __func__ << ": trim complete, expire_pos/trim_pos is now " << std::hex <<
2228     mdlog->get_journaler()->get_expire_pos() << "/" <<
2229     mdlog->get_journaler()->get_trimmed_pos() << dendl;
2230
2231   // Flush the journal header so that readers will start from after the flushed region
2232   C_SaferCond wrote_head;
2233   mdlog->get_journaler()->write_head(&wrote_head);
2234   mds_lock.Unlock();  // Drop lock to allow messenger dispatch progress
2235   r = wrote_head.wait();
2236   mds_lock.Lock();
2237   if (r != 0) {
2238       *ss << "Error " << r << " (" << cpp_strerror(r) << ") while writing header";
2239       return r;
2240   }
2241
2242   dout(5) << __func__ << ": write_head complete, all done!" << dendl;
2243
2244   return 0;
2245 }
2246
2247
2248 void MDSRank::command_get_subtrees(Formatter *f)
2249 {
2250   assert(f != NULL);
2251   Mutex::Locker l(mds_lock);
2252
2253   std::list<CDir*> subtrees;
2254   mdcache->list_subtrees(subtrees);
2255
2256   f->open_array_section("subtrees");
2257   for (std::list<CDir*>::iterator i = subtrees.begin(); i != subtrees.end(); ++i) {
2258     const CDir *dir = *i;
2259
2260     f->open_object_section("subtree");
2261     {
2262       f->dump_bool("is_auth", dir->is_auth());
2263       f->dump_int("auth_first", dir->get_dir_auth().first);
2264       f->dump_int("auth_second", dir->get_dir_auth().second);
2265       f->open_object_section("dir");
2266       dir->dump(f);
2267       f->close_section();
2268     }
2269     f->close_section();
2270   }
2271   f->close_section();
2272 }
2273
2274
2275 void MDSRank::command_export_dir(Formatter *f,
2276     const std::string &path,
2277     mds_rank_t target)
2278 {
2279   int r = _command_export_dir(path, target);
2280   f->open_object_section("results");
2281   f->dump_int("return_code", r);
2282   f->close_section(); // results
2283 }
2284
2285 int MDSRank::_command_export_dir(
2286     const std::string &path,
2287     mds_rank_t target)
2288 {
2289   Mutex::Locker l(mds_lock);
2290   filepath fp(path.c_str());
2291
2292   if (target == whoami || !mdsmap->is_up(target) || !mdsmap->is_in(target)) {
2293     derr << "bad MDS target " << target << dendl;
2294     return -ENOENT;
2295   }
2296
2297   CInode *in = mdcache->cache_traverse(fp);
2298   if (!in) {
2299     derr << "Bath path '" << path << "'" << dendl;
2300     return -ENOENT;
2301   }
2302   CDir *dir = in->get_dirfrag(frag_t());
2303   if (!dir || !(dir->is_auth())) {
2304     derr << "bad export_dir path dirfrag frag_t() or dir not auth" << dendl;
2305     return -EINVAL;
2306   }
2307
2308   mdcache->migrator->export_dir(dir, target);
2309   return 0;
2310 }
2311
2312 CDir *MDSRank::_command_dirfrag_get(
2313     const cmdmap_t &cmdmap,
2314     std::ostream &ss)
2315 {
2316   std::string path;
2317   bool got = cmd_getval(g_ceph_context, cmdmap, "path", path);
2318   if (!got) {
2319     ss << "missing path argument";
2320     return NULL;
2321   }
2322
2323   std::string frag_str;
2324   if (!cmd_getval(g_ceph_context, cmdmap, "frag", frag_str)) {
2325     ss << "missing frag argument";
2326     return NULL;
2327   }
2328
2329   CInode *in = mdcache->cache_traverse(filepath(path.c_str()));
2330   if (!in) {
2331     // TODO really we should load something in if it's not in cache,
2332     // but the infrastructure is harder, and we might still be unable
2333     // to act on it if someone else is auth.
2334     ss << "directory '" << path << "' inode not in cache";
2335     return NULL;
2336   }
2337
2338   frag_t fg;
2339
2340   if (!fg.parse(frag_str.c_str())) {
2341     ss << "frag " << frag_str << " failed to parse";
2342     return NULL;
2343   }
2344
2345   CDir *dir = in->get_dirfrag(fg);
2346   if (!dir) {
2347     ss << "frag 0x" << std::hex << in->ino() << "/" << fg << " not in cache ("
2348           "use `dirfrag ls` to see if it should exist)";
2349     return NULL;
2350   }
2351
2352   if (!dir->is_auth()) {
2353     ss << "frag " << dir->dirfrag() << " not auth (auth = "
2354        << dir->authority() << ")";
2355     return NULL;
2356   }
2357
2358   return dir;
2359 }
2360
2361 bool MDSRank::command_dirfrag_split(
2362     cmdmap_t cmdmap,
2363     std::ostream &ss)
2364 {
2365   Mutex::Locker l(mds_lock);
2366   if (!mdsmap->allows_dirfrags()) {
2367     ss << "dirfrags are disallowed by the mds map!";
2368     return false;
2369   }
2370
2371   int64_t by = 0;
2372   if (!cmd_getval(g_ceph_context, cmdmap, "bits", by)) {
2373     ss << "missing bits argument";
2374     return false;
2375   }
2376
2377   if (by <= 0) {
2378     ss << "must split by >0 bits";
2379     return false;
2380   }
2381
2382   CDir *dir = _command_dirfrag_get(cmdmap, ss);
2383   if (!dir) {
2384     return false;
2385   }
2386
2387   mdcache->split_dir(dir, by);
2388
2389   return true;
2390 }
2391
2392 bool MDSRank::command_dirfrag_merge(
2393     cmdmap_t cmdmap,
2394     std::ostream &ss)
2395 {
2396   Mutex::Locker l(mds_lock);
2397   std::string path;
2398   bool got = cmd_getval(g_ceph_context, cmdmap, "path", path);
2399   if (!got) {
2400     ss << "missing path argument";
2401     return false;
2402   }
2403
2404   std::string frag_str;
2405   if (!cmd_getval(g_ceph_context, cmdmap, "frag", frag_str)) {
2406     ss << "missing frag argument";
2407     return false;
2408   }
2409
2410   CInode *in = mdcache->cache_traverse(filepath(path.c_str()));
2411   if (!in) {
2412     ss << "directory '" << path << "' inode not in cache";
2413     return false;
2414   }
2415
2416   frag_t fg;
2417   if (!fg.parse(frag_str.c_str())) {
2418     ss << "frag " << frag_str << " failed to parse";
2419     return false;
2420   }
2421
2422   mdcache->merge_dir(in, fg);
2423
2424   return true;
2425 }
2426
2427 bool MDSRank::command_dirfrag_ls(
2428     cmdmap_t cmdmap,
2429     std::ostream &ss,
2430     Formatter *f)
2431 {
2432   Mutex::Locker l(mds_lock);
2433   std::string path;
2434   bool got = cmd_getval(g_ceph_context, cmdmap, "path", path);
2435   if (!got) {
2436     ss << "missing path argument";
2437     return false;
2438   }
2439
2440   CInode *in = mdcache->cache_traverse(filepath(path.c_str()));
2441   if (!in) {
2442     ss << "directory inode not in cache";
2443     return false;
2444   }
2445
2446   f->open_array_section("frags");
2447   std::list<frag_t> frags;
2448   // NB using get_leaves_under instead of get_dirfrags to give
2449   // you the list of what dirfrags may exist, not which are in cache
2450   in->dirfragtree.get_leaves_under(frag_t(), frags);
2451   for (std::list<frag_t>::iterator i = frags.begin();
2452        i != frags.end(); ++i) {
2453     f->open_object_section("frag");
2454     f->dump_int("value", i->value());
2455     f->dump_int("bits", i->bits());
2456     std::ostringstream frag_str;
2457     frag_str << std::hex << i->value() << "/" << std::dec << i->bits();
2458     f->dump_string("str", frag_str.str());
2459     f->close_section();
2460   }
2461   f->close_section();
2462
2463   return true;
2464 }
2465
2466 void MDSRank::dump_status(Formatter *f) const
2467 {
2468   if (state == MDSMap::STATE_REPLAY ||
2469       state == MDSMap::STATE_STANDBY_REPLAY) {
2470     mdlog->dump_replay_status(f);
2471   } else if (state == MDSMap::STATE_RESOLVE) {
2472     mdcache->dump_resolve_status(f);
2473   } else if (state == MDSMap::STATE_RECONNECT) {
2474     server->dump_reconnect_status(f);
2475   } else if (state == MDSMap::STATE_REJOIN) {
2476     mdcache->dump_rejoin_status(f);
2477   } else if (state == MDSMap::STATE_CLIENTREPLAY) {
2478     dump_clientreplay_status(f);
2479   }
2480 }
2481
2482 void MDSRank::dump_clientreplay_status(Formatter *f) const
2483 {
2484   f->open_object_section("clientreplay_status");
2485   f->dump_unsigned("clientreplay_queue", replay_queue.size());
2486   f->dump_unsigned("active_replay", mdcache->get_num_client_requests());
2487   f->close_section();
2488 }
2489
2490 void MDSRankDispatcher::update_log_config()
2491 {
2492   map<string,string> log_to_monitors;
2493   map<string,string> log_to_syslog;
2494   map<string,string> log_channel;
2495   map<string,string> log_prio;
2496   map<string,string> log_to_graylog;
2497   map<string,string> log_to_graylog_host;
2498   map<string,string> log_to_graylog_port;
2499   uuid_d fsid;
2500   string host;
2501
2502   if (parse_log_client_options(g_ceph_context, log_to_monitors, log_to_syslog,
2503                                log_channel, log_prio, log_to_graylog,
2504                                log_to_graylog_host, log_to_graylog_port,
2505                                fsid, host) == 0)
2506     clog->update_config(log_to_monitors, log_to_syslog,
2507                         log_channel, log_prio, log_to_graylog,
2508                         log_to_graylog_host, log_to_graylog_port,
2509                         fsid, host);
2510   dout(10) << __func__ << " log_to_monitors " << log_to_monitors << dendl;
2511 }
2512
2513 void MDSRank::create_logger()
2514 {
2515   dout(10) << "create_logger" << dendl;
2516   {
2517     PerfCountersBuilder mds_plb(g_ceph_context, "mds", l_mds_first, l_mds_last);
2518
2519     mds_plb.add_u64_counter(
2520       l_mds_request, "request", "Requests", "req",
2521       PerfCountersBuilder::PRIO_CRITICAL);
2522     mds_plb.add_u64_counter(l_mds_reply, "reply", "Replies");
2523     mds_plb.add_time_avg(
2524       l_mds_reply_latency, "reply_latency", "Reply latency", "rlat",
2525       PerfCountersBuilder::PRIO_CRITICAL);
2526     mds_plb.add_u64_counter(
2527       l_mds_forward, "forward", "Forwarding request", "fwd",
2528       PerfCountersBuilder::PRIO_INTERESTING);
2529     mds_plb.add_u64_counter(l_mds_dir_fetch, "dir_fetch", "Directory fetch");
2530     mds_plb.add_u64_counter(l_mds_dir_commit, "dir_commit", "Directory commit");
2531     mds_plb.add_u64_counter(l_mds_dir_split, "dir_split", "Directory split");
2532     mds_plb.add_u64_counter(l_mds_dir_merge, "dir_merge", "Directory merge");
2533
2534     mds_plb.add_u64(l_mds_inode_max, "inode_max", "Max inodes, cache size");
2535     mds_plb.add_u64(l_mds_inodes, "inodes", "Inodes", "inos",
2536                     PerfCountersBuilder::PRIO_CRITICAL);
2537     mds_plb.add_u64(l_mds_inodes_top, "inodes_top", "Inodes on top");
2538     mds_plb.add_u64(l_mds_inodes_bottom, "inodes_bottom", "Inodes on bottom");
2539     mds_plb.add_u64(
2540       l_mds_inodes_pin_tail, "inodes_pin_tail", "Inodes on pin tail");
2541     mds_plb.add_u64(l_mds_inodes_pinned, "inodes_pinned", "Inodes pinned");
2542     mds_plb.add_u64(l_mds_inodes_expired, "inodes_expired", "Inodes expired");
2543     mds_plb.add_u64(
2544       l_mds_inodes_with_caps, "inodes_with_caps", "Inodes with capabilities");
2545     mds_plb.add_u64(l_mds_caps, "caps", "Capabilities", "caps",
2546                     PerfCountersBuilder::PRIO_INTERESTING);
2547     mds_plb.add_u64(l_mds_subtrees, "subtrees", "Subtrees");
2548
2549     mds_plb.add_u64_counter(l_mds_traverse, "traverse", "Traverses");
2550     mds_plb.add_u64_counter(l_mds_traverse_hit, "traverse_hit", "Traverse hits");
2551     mds_plb.add_u64_counter(l_mds_traverse_forward, "traverse_forward",
2552                             "Traverse forwards");
2553     mds_plb.add_u64_counter(l_mds_traverse_discover, "traverse_discover",
2554                             "Traverse directory discovers");
2555     mds_plb.add_u64_counter(l_mds_traverse_dir_fetch, "traverse_dir_fetch",
2556                             "Traverse incomplete directory content fetchings");
2557     mds_plb.add_u64_counter(l_mds_traverse_remote_ino, "traverse_remote_ino",
2558                             "Traverse remote dentries");
2559     mds_plb.add_u64_counter(l_mds_traverse_lock, "traverse_lock",
2560                             "Traverse locks");
2561
2562     mds_plb.add_u64(l_mds_load_cent, "load_cent", "Load per cent");
2563     mds_plb.add_u64(l_mds_dispatch_queue_len, "q", "Dispatch queue length");
2564
2565     mds_plb.add_u64_counter(l_mds_exported, "exported", "Exports");
2566     mds_plb.add_u64_counter(
2567       l_mds_exported_inodes, "exported_inodes", "Exported inodes", "exi",
2568       PerfCountersBuilder::PRIO_INTERESTING);
2569     mds_plb.add_u64_counter(l_mds_imported, "imported", "Imports");
2570     mds_plb.add_u64_counter(
2571       l_mds_imported_inodes, "imported_inodes", "Imported inodes", "imi",
2572       PerfCountersBuilder::PRIO_INTERESTING);
2573     logger = mds_plb.create_perf_counters();
2574     g_ceph_context->get_perfcounters_collection()->add(logger);
2575   }
2576
2577   {
2578     PerfCountersBuilder mdm_plb(g_ceph_context, "mds_mem", l_mdm_first, l_mdm_last);
2579     mdm_plb.add_u64(l_mdm_ino, "ino", "Inodes");
2580     mdm_plb.add_u64_counter(l_mdm_inoa, "ino+", "Inodes opened");
2581     mdm_plb.add_u64_counter(l_mdm_inos, "ino-", "Inodes closed");
2582     mdm_plb.add_u64(l_mdm_dir, "dir", "Directories");
2583     mdm_plb.add_u64_counter(l_mdm_dira, "dir+", "Directories opened");
2584     mdm_plb.add_u64_counter(l_mdm_dirs, "dir-", "Directories closed");
2585     mdm_plb.add_u64(l_mdm_dn, "dn", "Dentries");
2586     mdm_plb.add_u64_counter(l_mdm_dna, "dn+", "Dentries opened");
2587     mdm_plb.add_u64_counter(l_mdm_dns, "dn-", "Dentries closed");
2588     mdm_plb.add_u64(l_mdm_cap, "cap", "Capabilities");
2589     mdm_plb.add_u64_counter(l_mdm_capa, "cap+", "Capabilities added");
2590     mdm_plb.add_u64_counter(l_mdm_caps, "cap-", "Capabilities removed");
2591     mdm_plb.add_u64(l_mdm_rss, "rss", "RSS");
2592     mdm_plb.add_u64(l_mdm_heap, "heap", "Heap size");
2593     mdm_plb.add_u64(l_mdm_buf, "buf", "Buffer size");
2594     mlogger = mdm_plb.create_perf_counters();
2595     g_ceph_context->get_perfcounters_collection()->add(mlogger);
2596   }
2597
2598   mdlog->create_logger();
2599   server->create_logger();
2600   purge_queue.create_logger();
2601   sessionmap.register_perfcounters();
2602   mdcache->register_perfcounters();
2603 }
2604
2605 void MDSRank::check_ops_in_flight()
2606 {
2607   vector<string> warnings;
2608   int slow = 0;
2609   if (op_tracker.check_ops_in_flight(warnings, &slow)) {
2610     for (vector<string>::iterator i = warnings.begin();
2611         i != warnings.end();
2612         ++i) {
2613       clog->warn() << *i;
2614     }
2615   }
2616
2617   // set mds slow request count
2618   mds_slow_req_count = slow;
2619   return;
2620 }
2621
2622 void MDSRankDispatcher::handle_osd_map()
2623 {
2624   if (is_active() && snapserver) {
2625     snapserver->check_osd_map(true);
2626   }
2627
2628   server->handle_osd_map();
2629
2630   purge_queue.update_op_limit(*mdsmap);
2631
2632   std::set<entity_addr_t> newly_blacklisted;
2633   objecter->consume_blacklist_events(&newly_blacklisted);
2634   auto epoch = objecter->with_osdmap([](const OSDMap &o){return o.get_epoch();});
2635   dout(4) << "handle_osd_map epoch " << epoch << ", "
2636           << newly_blacklisted.size() << " new blacklist entries" << dendl;
2637   auto victims = server->apply_blacklist(newly_blacklisted);
2638   if (victims) {
2639     set_osd_epoch_barrier(epoch);
2640   }
2641
2642
2643   // By default the objecter only requests OSDMap updates on use,
2644   // we would like to always receive the latest maps in order to
2645   // apply policy based on the FULL flag.
2646   objecter->maybe_request_map();
2647 }
2648
2649 bool MDSRank::evict_client(int64_t session_id,
2650     bool wait, bool blacklist, std::stringstream& err_ss,
2651     Context *on_killed)
2652 {
2653   assert(mds_lock.is_locked_by_me());
2654
2655   // Mutually exclusive args
2656   assert(!(wait && on_killed != nullptr));
2657
2658   if (is_any_replay()) {
2659     err_ss << "MDS is replaying log";
2660     return false;
2661   }
2662
2663   Session *session = sessionmap.get_session(
2664       entity_name_t(CEPH_ENTITY_TYPE_CLIENT, session_id));
2665   if (!session) {
2666     err_ss << "session " << session_id << " not in sessionmap!";
2667     return false;
2668   }
2669
2670   dout(4) << "Preparing blacklist command... (wait=" << wait << ")" << dendl;
2671   stringstream ss;
2672   ss << "{\"prefix\":\"osd blacklist\", \"blacklistop\":\"add\",";
2673   ss << "\"addr\":\"";
2674   ss << session->info.inst.addr;
2675   ss << "\"}";
2676   std::string tmp = ss.str();
2677   std::vector<std::string> cmd = {tmp};
2678
2679   auto kill_mds_session = [this, session_id, on_killed](){
2680     assert(mds_lock.is_locked_by_me());
2681     Session *session = sessionmap.get_session(
2682         entity_name_t(CEPH_ENTITY_TYPE_CLIENT, session_id));
2683     if (session) {
2684       if (on_killed) {
2685         server->kill_session(session, on_killed);
2686       } else {
2687         C_SaferCond on_safe;
2688         server->kill_session(session, &on_safe);
2689
2690         mds_lock.Unlock();
2691         on_safe.wait();
2692         mds_lock.Lock();
2693       }
2694     } else {
2695       dout(1) << "session " << session_id << " was removed while we waited "
2696       "for blacklist" << dendl;
2697
2698       // Even though it wasn't us that removed it, kick our completion
2699       // as the session has been removed.
2700       if (on_killed) {
2701         on_killed->complete(0);
2702       }
2703     }
2704   };
2705
2706   auto background_blacklist = [this, session_id, cmd](std::function<void ()> fn){
2707     assert(mds_lock.is_locked_by_me());
2708
2709     Context *on_blacklist_done = new FunctionContext([this, session_id, fn](int r) {
2710       objecter->wait_for_latest_osdmap(
2711        new C_OnFinisher(
2712          new FunctionContext([this, session_id, fn](int r) {
2713               Mutex::Locker l(mds_lock);
2714               auto epoch = objecter->with_osdmap([](const OSDMap &o){
2715                   return o.get_epoch();
2716               });
2717
2718               set_osd_epoch_barrier(epoch);
2719
2720               fn();
2721             }), finisher)
2722        );
2723     });
2724
2725     dout(4) << "Sending mon blacklist command: " << cmd[0] << dendl;
2726     monc->start_mon_command(cmd, {}, nullptr, nullptr, on_blacklist_done);
2727   };
2728
2729   auto blocking_blacklist = [this, cmd, &err_ss, background_blacklist](){
2730     C_SaferCond inline_ctx;
2731     background_blacklist([&inline_ctx](){inline_ctx.complete(0);});
2732     mds_lock.Unlock();
2733     inline_ctx.wait();
2734     mds_lock.Lock();
2735   };
2736
2737   if (wait) {
2738     if (blacklist) {
2739       blocking_blacklist();
2740     }
2741
2742     // We dropped mds_lock, so check that session still exists
2743     session = sessionmap.get_session(entity_name_t(CEPH_ENTITY_TYPE_CLIENT,
2744           session_id));
2745     if (!session) {
2746       dout(1) << "session " << session_id << " was removed while we waited "
2747                  "for blacklist" << dendl;
2748       return true;
2749     }
2750     kill_mds_session();
2751   } else {
2752     if (blacklist) {
2753       background_blacklist(kill_mds_session);
2754     } else {
2755       kill_mds_session();
2756     }
2757   }
2758
2759   return true;
2760 }
2761
2762 void MDSRank::bcast_mds_map()
2763 {
2764   dout(7) << "bcast_mds_map " << mdsmap->get_epoch() << dendl;
2765
2766   // share the map with mounted clients
2767   set<Session*> clients;
2768   sessionmap.get_client_session_set(clients);
2769   for (set<Session*>::const_iterator p = clients.begin();
2770        p != clients.end();
2771        ++p)
2772     (*p)->connection->send_message(new MMDSMap(monc->get_fsid(), mdsmap));
2773   last_client_mdsmap_bcast = mdsmap->get_epoch();
2774 }
2775
2776 MDSRankDispatcher::MDSRankDispatcher(
2777     mds_rank_t whoami_,
2778     Mutex &mds_lock_,
2779     LogChannelRef &clog_,
2780     SafeTimer &timer_,
2781     Beacon &beacon_,
2782     MDSMap *& mdsmap_,
2783     Messenger *msgr,
2784     MonClient *monc_,
2785     Context *respawn_hook_,
2786     Context *suicide_hook_)
2787   : MDSRank(whoami_, mds_lock_, clog_, timer_, beacon_, mdsmap_,
2788       msgr, monc_, respawn_hook_, suicide_hook_)
2789 {}
2790
2791 bool MDSRankDispatcher::handle_command(
2792   const cmdmap_t &cmdmap,
2793   MCommand *m,
2794   int *r,
2795   std::stringstream *ds,
2796   std::stringstream *ss,
2797   bool *need_reply)
2798 {
2799   assert(r != nullptr);
2800   assert(ds != nullptr);
2801   assert(ss != nullptr);
2802
2803   *need_reply = true;
2804
2805   std::string prefix;
2806   cmd_getval(g_ceph_context, cmdmap, "prefix", prefix);
2807
2808   if (prefix == "session ls" || prefix == "client ls") {
2809     std::vector<std::string> filter_args;
2810     cmd_getval(g_ceph_context, cmdmap, "filters", filter_args);
2811
2812     SessionFilter filter;
2813     *r = filter.parse(filter_args, ss);
2814     if (*r != 0) {
2815       return true;
2816     }
2817
2818     Formatter *f = new JSONFormatter(true);
2819     dump_sessions(filter, f);
2820     f->flush(*ds);
2821     delete f;
2822     return true;
2823   } else if (prefix == "session evict" || prefix == "client evict") {
2824     std::vector<std::string> filter_args;
2825     cmd_getval(g_ceph_context, cmdmap, "filters", filter_args);
2826
2827     SessionFilter filter;
2828     *r = filter.parse(filter_args, ss);
2829     if (*r != 0) {
2830       return true;
2831     }
2832
2833     evict_clients(filter, m);
2834
2835     *need_reply = false;
2836     return true;
2837   } else if (prefix == "damage ls") {
2838     Formatter *f = new JSONFormatter(true);
2839     damage_table.dump(f);
2840     f->flush(*ds);
2841     delete f;
2842     return true;
2843   } else if (prefix == "damage rm") {
2844     damage_entry_id_t id = 0;
2845     bool got = cmd_getval(g_ceph_context, cmdmap, "damage_id", (int64_t&)id);
2846     if (!got) {
2847       *r = -EINVAL;
2848       return true;
2849     }
2850
2851     damage_table.erase(id);
2852     return true;
2853   } else {
2854     return false;
2855   }
2856 }
2857
2858 epoch_t MDSRank::get_osd_epoch() const
2859 {
2860   return objecter->with_osdmap(std::mem_fn(&OSDMap::get_epoch));
2861 }
2862