ceph/src/osd/OSD.h

   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) 2004-2006 Sage Weil <sage@newdream.net>
   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 #ifndef CEPH_OSD_H
  16 #define CEPH_OSD_H
  17
  18 #include "PG.h"
  19
  20 #include "msg/Dispatcher.h"
  21
  22 #include "common/Mutex.h"
  23 #include "common/RWLock.h"
  24 #include "common/Timer.h"
  25 #include "common/WorkQueue.h"
  26 #include "common/AsyncReserver.h"
  27 #include "common/ceph_context.h"
  28 #include "common/zipkin_trace.h"
  29
  30 #include "mgr/MgrClient.h"
  31
  32 #include "os/ObjectStore.h"
  33 #include "OSDCap.h"
  34
  35 #include "auth/KeyRing.h"
  36 #include "osd/ClassHandler.h"
  37
  38 #include "include/CompatSet.h"
  39
  40 #include "OpRequest.h"
  41 #include "Session.h"
  42
  43 #include "osd/PGQueueable.h"
  44
  45 #include <atomic>
  46 #include <map>
  47 #include <memory>
  48 #include "include/memory.h"
  49 using namespace std;
  50
  51 #include "include/unordered_map.h"
  52
  53 #include "common/shared_cache.hpp"
  54 #include "common/simple_cache.hpp"
  55 #include "common/sharedptr_registry.hpp"
  56 #include "common/WeightedPriorityQueue.h"
  57 #include "common/PrioritizedQueue.h"
  58 #include "osd/mClockOpClassQueue.h"
  59 #include "osd/mClockClientQueue.h"
  60 #include "messages/MOSDOp.h"
  61 #include "include/Spinlock.h"
  62 #include "common/EventTrace.h"
  63
  64 #define CEPH_OSD_PROTOCOL    10 /* cluster internal */
  65
  66
  67 enum {
  68   l_osd_first = 10000,
  69   l_osd_op_wip,
  70   l_osd_op,
  71   l_osd_op_inb,
  72   l_osd_op_outb,
  73   l_osd_op_lat,
  74   l_osd_op_process_lat,
  75   l_osd_op_prepare_lat,
  76   l_osd_op_r,
  77   l_osd_op_r_outb,
  78   l_osd_op_r_lat,
  79   l_osd_op_r_lat_outb_hist,
  80   l_osd_op_r_process_lat,
  81   l_osd_op_r_prepare_lat,
  82   l_osd_op_w,
  83   l_osd_op_w_inb,
  84   l_osd_op_w_lat,
  85   l_osd_op_w_lat_inb_hist,
  86   l_osd_op_w_process_lat,
  87   l_osd_op_w_prepare_lat,
  88   l_osd_op_rw,
  89   l_osd_op_rw_inb,
  90   l_osd_op_rw_outb,
  91   l_osd_op_rw_lat,
  92   l_osd_op_rw_lat_inb_hist,
  93   l_osd_op_rw_lat_outb_hist,
  94   l_osd_op_rw_process_lat,
  95   l_osd_op_rw_prepare_lat,
  96
  97   l_osd_op_before_queue_op_lat,
  98   l_osd_op_before_dequeue_op_lat,
  99
 100   l_osd_sop,
 101   l_osd_sop_inb,
 102   l_osd_sop_lat,
 103   l_osd_sop_w,
 104   l_osd_sop_w_inb,
 105   l_osd_sop_w_lat,
 106   l_osd_sop_pull,
 107   l_osd_sop_pull_lat,
 108   l_osd_sop_push,
 109   l_osd_sop_push_inb,
 110   l_osd_sop_push_lat,
 111
 112   l_osd_pull,
 113   l_osd_push,
 114   l_osd_push_outb,
 115
 116   l_osd_rop,
 117
 118   l_osd_loadavg,
 119   l_osd_buf,
 120   l_osd_history_alloc_bytes,
 121   l_osd_history_alloc_num,
 122   l_osd_cached_crc,
 123   l_osd_cached_crc_adjusted,
 124   l_osd_missed_crc,
 125
 126   l_osd_pg,
 127   l_osd_pg_primary,
 128   l_osd_pg_replica,
 129   l_osd_pg_stray,
 130   l_osd_pg_removing,
 131   l_osd_hb_to,
 132   l_osd_map,
 133   l_osd_mape,
 134   l_osd_mape_dup,
 135
 136   l_osd_waiting_for_map,
 137
 138   l_osd_map_cache_hit,
 139   l_osd_map_cache_miss,
 140   l_osd_map_cache_miss_low,
 141   l_osd_map_cache_miss_low_avg,
 142   l_osd_map_bl_cache_hit,
 143   l_osd_map_bl_cache_miss,
 144
 145   l_osd_stat_bytes,
 146   l_osd_stat_bytes_used,
 147   l_osd_stat_bytes_avail,
 148
 149   l_osd_copyfrom,
 150
 151   l_osd_tier_promote,
 152   l_osd_tier_flush,
 153   l_osd_tier_flush_fail,
 154   l_osd_tier_try_flush,
 155   l_osd_tier_try_flush_fail,
 156   l_osd_tier_evict,
 157   l_osd_tier_whiteout,
 158   l_osd_tier_dirty,
 159   l_osd_tier_clean,
 160   l_osd_tier_delay,
 161   l_osd_tier_proxy_read,
 162   l_osd_tier_proxy_write,
 163
 164   l_osd_agent_wake,
 165   l_osd_agent_skip,
 166   l_osd_agent_flush,
 167   l_osd_agent_evict,
 168
 169   l_osd_object_ctx_cache_hit,
 170   l_osd_object_ctx_cache_total,
 171
 172   l_osd_op_cache_hit,
 173   l_osd_tier_flush_lat,
 174   l_osd_tier_promote_lat,
 175   l_osd_tier_r_lat,
 176
 177   l_osd_pg_info,
 178   l_osd_pg_fastinfo,
 179   l_osd_pg_biginfo,
 180
 181   l_osd_last,
 182 };
 183
 184 // RecoveryState perf counters
 185 enum {
 186   rs_first = 20000,
 187   rs_initial_latency,
 188   rs_started_latency,
 189   rs_reset_latency,
 190   rs_start_latency,
 191   rs_primary_latency,
 192   rs_peering_latency,
 193   rs_backfilling_latency,
 194   rs_waitremotebackfillreserved_latency,
 195   rs_waitlocalbackfillreserved_latency,
 196   rs_notbackfilling_latency,
 197   rs_repnotrecovering_latency,
 198   rs_repwaitrecoveryreserved_latency,
 199   rs_repwaitbackfillreserved_latency,
 200   rs_reprecovering_latency,
 201   rs_activating_latency,
 202   rs_waitlocalrecoveryreserved_latency,
 203   rs_waitremoterecoveryreserved_latency,
 204   rs_recovering_latency,
 205   rs_recovered_latency,
 206   rs_clean_latency,
 207   rs_active_latency,
 208   rs_replicaactive_latency,
 209   rs_stray_latency,
 210   rs_getinfo_latency,
 211   rs_getlog_latency,
 212   rs_waitactingchange_latency,
 213   rs_incomplete_latency,
 214   rs_down_latency,
 215   rs_getmissing_latency,
 216   rs_waitupthru_latency,
 217   rs_notrecovering_latency,
 218   rs_last,
 219 };
 220
 221 class Messenger;
 222 class Message;
 223 class MonClient;
 224 class PerfCounters;
 225 class ObjectStore;
 226 class FuseStore;
 227 class OSDMap;
 228 class MLog;
 229 class Objecter;
 230
 231 class Watch;
 232 class PrimaryLogPG;
 233
 234 class AuthAuthorizeHandlerRegistry;
 235
 236 class TestOpsSocketHook;
 237 struct C_CompleteSplits;
 238 struct C_OpenPGs;
 239 class LogChannel;
 240 class CephContext;
 241 typedef ceph::shared_ptr<ObjectStore::Sequencer> SequencerRef;
 242 class MOSDOp;
 243
 244 class DeletingState {
 245   Mutex lock;
 246   Cond cond;
 247   enum {
 248     QUEUED,
 249     CLEARING_DIR,
 250     CLEARING_WAITING,
 251     DELETING_DIR,
 252     DELETED_DIR,
 253     CANCELED,
 254   } status;
 255   bool stop_deleting;
 256 public:
 257   const spg_t pgid;
 258   const PGRef old_pg_state;
 259   explicit DeletingState(const pair<spg_t, PGRef> &in) :
 260     lock("DeletingState::lock"), status(QUEUED), stop_deleting(false),
 261     pgid(in.first), old_pg_state(in.second) {
 262     }
 263
 264   /// transition status to CLEARING_WAITING
 265   bool pause_clearing() {
 266     Mutex::Locker l(lock);
 267     assert(status == CLEARING_DIR);
 268     if (stop_deleting) {
 269       status = CANCELED;
 270       cond.Signal();
 271       return false;
 272     }
 273     status = CLEARING_WAITING;
 274     return true;
 275   } ///< @return false if we should cancel deletion
 276
 277   /// start or resume the clearing - transition the status to CLEARING_DIR
 278   bool start_or_resume_clearing() {
 279     Mutex::Locker l(lock);
 280     assert(
 281       status == QUEUED ||
 282       status == DELETED_DIR ||
 283       status == CLEARING_WAITING);
 284     if (stop_deleting) {
 285       status = CANCELED;
 286       cond.Signal();
 287       return false;
 288     }
 289     status = CLEARING_DIR;
 290     return true;
 291   } ///< @return false if we should cancel the deletion
 292
 293   /// transition status to CLEARING_DIR
 294   bool resume_clearing() {
 295     Mutex::Locker l(lock);
 296     assert(status == CLEARING_WAITING);
 297     if (stop_deleting) {
 298       status = CANCELED;
 299       cond.Signal();
 300       return false;
 301     }
 302     status = CLEARING_DIR;
 303     return true;
 304   } ///< @return false if we should cancel deletion
 305
 306   /// transition status to deleting
 307   bool start_deleting() {
 308     Mutex::Locker l(lock);
 309     assert(status == CLEARING_DIR);
 310     if (stop_deleting) {
 311       status = CANCELED;
 312       cond.Signal();
 313       return false;
 314     }
 315     status = DELETING_DIR;
 316     return true;
 317   } ///< @return false if we should cancel deletion
 318
 319   /// signal collection removal queued
 320   void finish_deleting() {
 321     Mutex::Locker l(lock);
 322     assert(status == DELETING_DIR);
 323     status = DELETED_DIR;
 324     cond.Signal();
 325   }
 326
 327   /// try to halt the deletion
 328   bool try_stop_deletion() {
 329     Mutex::Locker l(lock);
 330     stop_deleting = true;
 331     /**
 332      * If we are in DELETING_DIR or CLEARING_DIR, there are in progress
 333      * operations we have to wait for before continuing on.  States
 334      * CLEARING_WAITING and QUEUED indicate that the remover will check
 335      * stop_deleting before queueing any further operations.  CANCELED
 336      * indicates that the remover has already halted.  DELETED_DIR
 337      * indicates that the deletion has been fully queued.
 338      */
 339     while (status == DELETING_DIR || status == CLEARING_DIR)
 340       cond.Wait(lock);
 341     return status != DELETED_DIR;
 342   } ///< @return true if we don't need to recreate the collection
 343 };
 344 typedef ceph::shared_ptr<DeletingState> DeletingStateRef;
 345
 346 class OSD;
 347
 348 class OSDService {
 349 public:
 350   OSD *osd;
 351   CephContext *cct;
 352   SharedPtrRegistry<spg_t, ObjectStore::Sequencer> osr_registry;
 353   ceph::shared_ptr<ObjectStore::Sequencer> meta_osr;
 354   SharedPtrRegistry<spg_t, DeletingState> deleting_pgs;
 355   const int whoami;
 356   ObjectStore *&store;
 357   LogClient &log_client;
 358   LogChannelRef clog;
 359   PGRecoveryStats &pg_recovery_stats;
 360 private:
 361   Messenger *&cluster_messenger;
 362   Messenger *&client_messenger;
 363 public:
 364   PerfCounters *&logger;
 365   PerfCounters *&recoverystate_perf;
 366   MonClient   *&monc;
 367   ThreadPool::BatchWorkQueue<PG> &peering_wq;
 368   GenContextWQ recovery_gen_wq;
 369   ClassHandler  *&class_handler;
 370
 371   void enqueue_back(spg_t pgid, PGQueueable qi);
 372   void enqueue_front(spg_t pgid, PGQueueable qi);
 373
 374   void maybe_inject_dispatch_delay() {
 375     if (g_conf->osd_debug_inject_dispatch_delay_probability > 0) {
 376       if (rand() % 10000 <
 377           g_conf->osd_debug_inject_dispatch_delay_probability * 10000) {
 378         utime_t t;
 379         t.set_from_double(g_conf->osd_debug_inject_dispatch_delay_duration);
 380         t.sleep();
 381       }
 382     }
 383   }
 384
 385 private:
 386   // -- map epoch lower bound --
 387   Mutex pg_epoch_lock;
 388   multiset<epoch_t> pg_epochs;
 389   map<spg_t,epoch_t> pg_epoch;
 390
 391 public:
 392   void pg_add_epoch(spg_t pgid, epoch_t epoch) {
 393     Mutex::Locker l(pg_epoch_lock);
 394     map<spg_t,epoch_t>::iterator t = pg_epoch.find(pgid);
 395     assert(t == pg_epoch.end());
 396     pg_epoch[pgid] = epoch;
 397     pg_epochs.insert(epoch);
 398   }
 399   void pg_update_epoch(spg_t pgid, epoch_t epoch) {
 400     Mutex::Locker l(pg_epoch_lock);
 401     map<spg_t,epoch_t>::iterator t = pg_epoch.find(pgid);
 402     assert(t != pg_epoch.end());
 403     pg_epochs.erase(pg_epochs.find(t->second));
 404     t->second = epoch;
 405     pg_epochs.insert(epoch);
 406   }
 407   void pg_remove_epoch(spg_t pgid) {
 408     Mutex::Locker l(pg_epoch_lock);
 409     map<spg_t,epoch_t>::iterator t = pg_epoch.find(pgid);
 410     if (t != pg_epoch.end()) {
 411       pg_epochs.erase(pg_epochs.find(t->second));
 412       pg_epoch.erase(t);
 413     }
 414   }
 415   epoch_t get_min_pg_epoch() {
 416     Mutex::Locker l(pg_epoch_lock);
 417     if (pg_epochs.empty())
 418       return 0;
 419     else
 420       return *pg_epochs.begin();
 421   }
 422
 423 private:
 424   // -- superblock --
 425   Mutex publish_lock, pre_publish_lock; // pre-publish orders before publish
 426   OSDSuperblock superblock;
 427
 428 public:
 429   OSDSuperblock get_superblock() {
 430     Mutex::Locker l(publish_lock);
 431     return superblock;
 432   }
 433   void publish_superblock(const OSDSuperblock &block) {
 434     Mutex::Locker l(publish_lock);
 435     superblock = block;
 436   }
 437
 438   int get_nodeid() const { return whoami; }
 439
 440   std::atomic<epoch_t> max_oldest_map;
 441 private:
 442   OSDMapRef osdmap;
 443
 444 public:
 445   OSDMapRef get_osdmap() {
 446     Mutex::Locker l(publish_lock);
 447     return osdmap;
 448   }
 449   epoch_t get_osdmap_epoch() {
 450     Mutex::Locker l(publish_lock);
 451     return osdmap ? osdmap->get_epoch() : 0;
 452   }
 453   void publish_map(OSDMapRef map) {
 454     Mutex::Locker l(publish_lock);
 455     osdmap = map;
 456   }
 457
 458   /*
 459    * osdmap - current published map
 460    * next_osdmap - pre_published map that is about to be published.
 461    *
 462    * We use the next_osdmap to send messages and initiate connections,
 463    * but only if the target is the same instance as the one in the map
 464    * epoch the current user is working from (i.e., the result is
 465    * equivalent to what is in next_osdmap).
 466    *
 467    * This allows the helpers to start ignoring osds that are about to
 468    * go down, and let OSD::handle_osd_map()/note_down_osd() mark them
 469    * down, without worrying about reopening connections from threads
 470    * working from old maps.
 471    */
 472 private:
 473   OSDMapRef next_osdmap;
 474   Cond pre_publish_cond;
 475
 476 public:
 477   void pre_publish_map(OSDMapRef map) {
 478     Mutex::Locker l(pre_publish_lock);
 479     next_osdmap = std::move(map);
 480   }
 481
 482   void activate_map();
 483   /// map epochs reserved below
 484   map<epoch_t, unsigned> map_reservations;
 485
 486   /// gets ref to next_osdmap and registers the epoch as reserved
 487   OSDMapRef get_nextmap_reserved() {
 488     Mutex::Locker l(pre_publish_lock);
 489     if (!next_osdmap)
 490       return OSDMapRef();
 491     epoch_t e = next_osdmap->get_epoch();
 492     map<epoch_t, unsigned>::iterator i =
 493       map_reservations.insert(make_pair(e, 0)).first;
 494     i->second++;
 495     return next_osdmap;
 496   }
 497   /// releases reservation on map
 498   void release_map(OSDMapRef osdmap) {
 499     Mutex::Locker l(pre_publish_lock);
 500     map<epoch_t, unsigned>::iterator i =
 501       map_reservations.find(osdmap->get_epoch());
 502     assert(i != map_reservations.end());
 503     assert(i->second > 0);
 504     if (--(i->second) == 0) {
 505       map_reservations.erase(i);
 506     }
 507     pre_publish_cond.Signal();
 508   }
 509   /// blocks until there are no reserved maps prior to next_osdmap
 510   void await_reserved_maps() {
 511     Mutex::Locker l(pre_publish_lock);
 512     assert(next_osdmap);
 513     while (true) {
 514       map<epoch_t, unsigned>::const_iterator i = map_reservations.cbegin();
 515       if (i == map_reservations.cend() || i->first >= next_osdmap->get_epoch()) {
 516         break;
 517       } else {
 518         pre_publish_cond.Wait(pre_publish_lock);
 519       }
 520     }
 521   }
 522
 523 private:
 524   Mutex peer_map_epoch_lock;
 525   map<int, epoch_t> peer_map_epoch;
 526 public:
 527   epoch_t get_peer_epoch(int p);
 528   epoch_t note_peer_epoch(int p, epoch_t e);
 529   void forget_peer_epoch(int p, epoch_t e);
 530
 531   void send_map(class MOSDMap *m, Connection *con);
 532   void send_incremental_map(epoch_t since, Connection *con, OSDMapRef& osdmap);
 533   MOSDMap *build_incremental_map_msg(epoch_t from, epoch_t to,
 534                                        OSDSuperblock& superblock);
 535   bool should_share_map(entity_name_t name, Connection *con, epoch_t epoch,
 536                         const OSDMapRef& osdmap, const epoch_t *sent_epoch_p);
 537   void share_map(entity_name_t name, Connection *con, epoch_t epoch,
 538                  OSDMapRef& osdmap, epoch_t *sent_epoch_p);
 539   void share_map_peer(int peer, Connection *con,
 540                       OSDMapRef map = OSDMapRef());
 541
 542   ConnectionRef get_con_osd_cluster(int peer, epoch_t from_epoch);
 543   pair<ConnectionRef,ConnectionRef> get_con_osd_hb(int peer, epoch_t from_epoch);  // (back, front)
 544   void send_message_osd_cluster(int peer, Message *m, epoch_t from_epoch);
 545   void send_message_osd_cluster(Message *m, Connection *con) {
 546     con->send_message(m);
 547   }
 548   void send_message_osd_cluster(Message *m, const ConnectionRef& con) {
 549     con->send_message(m);
 550   }
 551   void send_message_osd_client(Message *m, Connection *con) {
 552     con->send_message(m);
 553   }
 554   void send_message_osd_client(Message *m, const ConnectionRef& con) {
 555     con->send_message(m);
 556   }
 557   entity_name_t get_cluster_msgr_name() {
 558     return cluster_messenger->get_myname();
 559   }
 560
 561 private:
 562   // -- scrub scheduling --
 563   Mutex sched_scrub_lock;
 564   int scrubs_pending;
 565   int scrubs_active;
 566
 567 public:
 568   struct ScrubJob {
 569     CephContext* cct;
 570     /// pg to be scrubbed
 571     spg_t pgid;
 572     /// a time scheduled for scrub. but the scrub could be delayed if system
 573     /// load is too high or it fails to fall in the scrub hours
 574     utime_t sched_time;
 575     /// the hard upper bound of scrub time
 576     utime_t deadline;
 577     ScrubJob() : cct(nullptr) {}
 578     explicit ScrubJob(CephContext* cct, const spg_t& pg,
 579                       const utime_t& timestamp,
 580                       double pool_scrub_min_interval = 0,
 581                       double pool_scrub_max_interval = 0, bool must = true);
 582     /// order the jobs by sched_time
 583     bool operator<(const ScrubJob& rhs) const;
 584   };
 585   set<ScrubJob> sched_scrub_pg;
 586
 587   /// @returns the scrub_reg_stamp used for unregister the scrub job
 588   utime_t reg_pg_scrub(spg_t pgid, utime_t t, double pool_scrub_min_interval,
 589                        double pool_scrub_max_interval, bool must) {
 590     ScrubJob scrub(cct, pgid, t, pool_scrub_min_interval, pool_scrub_max_interval,
 591                    must);
 592     Mutex::Locker l(sched_scrub_lock);
 593     sched_scrub_pg.insert(scrub);
 594     return scrub.sched_time;
 595   }
 596   void unreg_pg_scrub(spg_t pgid, utime_t t) {
 597     Mutex::Locker l(sched_scrub_lock);
 598     size_t removed = sched_scrub_pg.erase(ScrubJob(cct, pgid, t));
 599     assert(removed);
 600   }
 601   bool first_scrub_stamp(ScrubJob *out) {
 602     Mutex::Locker l(sched_scrub_lock);
 603     if (sched_scrub_pg.empty())
 604       return false;
 605     set<ScrubJob>::iterator iter = sched_scrub_pg.begin();
 606     *out = *iter;
 607     return true;
 608   }
 609   bool next_scrub_stamp(const ScrubJob& next,
 610                         ScrubJob *out) {
 611     Mutex::Locker l(sched_scrub_lock);
 612     if (sched_scrub_pg.empty())
 613       return false;
 614     set<ScrubJob>::const_iterator iter = sched_scrub_pg.lower_bound(next);
 615     if (iter == sched_scrub_pg.cend())
 616       return false;
 617     ++iter;
 618     if (iter == sched_scrub_pg.cend())
 619       return false;
 620     *out = *iter;
 621     return true;
 622   }
 623
 624   void dumps_scrub(Formatter *f) {
 625     assert(f != nullptr);
 626     Mutex::Locker l(sched_scrub_lock);
 627
 628     f->open_array_section("scrubs");
 629     for (const auto &i: sched_scrub_pg) {
 630       f->open_object_section("scrub");
 631       f->dump_stream("pgid") << i.pgid;
 632       f->dump_stream("sched_time") << i.sched_time;
 633       f->dump_stream("deadline") << i.deadline;
 634       f->dump_bool("forced", i.sched_time == i.deadline);
 635       f->close_section();
 636     }
 637     f->close_section();
 638   }
 639
 640   bool can_inc_scrubs_pending();
 641   bool inc_scrubs_pending();
 642   void inc_scrubs_active(bool reserved);
 643   void dec_scrubs_pending();
 644   void dec_scrubs_active();
 645
 646   void reply_op_error(OpRequestRef op, int err);
 647   void reply_op_error(OpRequestRef op, int err, eversion_t v, version_t uv);
 648   void handle_misdirected_op(PG *pg, OpRequestRef op);
 649
 650
 651 private:
 652   // -- agent shared state --
 653   Mutex agent_lock;
 654   Cond agent_cond;
 655   map<uint64_t, set<PGRef> > agent_queue;
 656   set<PGRef>::iterator agent_queue_pos;
 657   bool agent_valid_iterator;
 658   int agent_ops;
 659   int flush_mode_high_count; //once have one pg with FLUSH_MODE_HIGH then flush objects with high speed
 660   set<hobject_t> agent_oids;
 661   bool agent_active;
 662   struct AgentThread : public Thread {
 663     OSDService *osd;
 664     explicit AgentThread(OSDService *o) : osd(o) {}
 665     void *entry() override {
 666       osd->agent_entry();
 667       return NULL;
 668     }
 669   } agent_thread;
 670   bool agent_stop_flag;
 671   Mutex agent_timer_lock;
 672   SafeTimer agent_timer;
 673
 674 public:
 675   void agent_entry();
 676   void agent_stop();
 677
 678   void _enqueue(PG *pg, uint64_t priority) {
 679     if (!agent_queue.empty() &&
 680         agent_queue.rbegin()->first < priority)
 681       agent_valid_iterator = false;  // inserting higher-priority queue
 682     set<PGRef>& nq = agent_queue[priority];
 683     if (nq.empty())
 684       agent_cond.Signal();
 685     nq.insert(pg);
 686   }
 687
 688   void _dequeue(PG *pg, uint64_t old_priority) {
 689     set<PGRef>& oq = agent_queue[old_priority];
 690     set<PGRef>::iterator p = oq.find(pg);
 691     assert(p != oq.end());
 692     if (p == agent_queue_pos)
 693       ++agent_queue_pos;
 694     oq.erase(p);
 695     if (oq.empty()) {
 696       if (agent_queue.rbegin()->first == old_priority)
 697         agent_valid_iterator = false;
 698       agent_queue.erase(old_priority);
 699     }
 700   }
 701
 702   /// enable agent for a pg
 703   void agent_enable_pg(PG *pg, uint64_t priority) {
 704     Mutex::Locker l(agent_lock);
 705     _enqueue(pg, priority);
 706   }
 707
 708   /// adjust priority for an enagled pg
 709   void agent_adjust_pg(PG *pg, uint64_t old_priority, uint64_t new_priority) {
 710     Mutex::Locker l(agent_lock);
 711     assert(new_priority != old_priority);
 712     _enqueue(pg, new_priority);
 713     _dequeue(pg, old_priority);
 714   }
 715
 716   /// disable agent for a pg
 717   void agent_disable_pg(PG *pg, uint64_t old_priority) {
 718     Mutex::Locker l(agent_lock);
 719     _dequeue(pg, old_priority);
 720   }
 721
 722   /// note start of an async (evict) op
 723   void agent_start_evict_op() {
 724     Mutex::Locker l(agent_lock);
 725     ++agent_ops;
 726   }
 727
 728   /// note finish or cancellation of an async (evict) op
 729   void agent_finish_evict_op() {
 730     Mutex::Locker l(agent_lock);
 731     assert(agent_ops > 0);
 732     --agent_ops;
 733     agent_cond.Signal();
 734   }
 735
 736   /// note start of an async (flush) op
 737   void agent_start_op(const hobject_t& oid) {
 738     Mutex::Locker l(agent_lock);
 739     ++agent_ops;
 740     assert(agent_oids.count(oid) == 0);
 741     agent_oids.insert(oid);
 742   }
 743
 744   /// note finish or cancellation of an async (flush) op
 745   void agent_finish_op(const hobject_t& oid) {
 746     Mutex::Locker l(agent_lock);
 747     assert(agent_ops > 0);
 748     --agent_ops;
 749     assert(agent_oids.count(oid) == 1);
 750     agent_oids.erase(oid);
 751     agent_cond.Signal();
 752   }
 753
 754   /// check if we are operating on an object
 755   bool agent_is_active_oid(const hobject_t& oid) {
 756     Mutex::Locker l(agent_lock);
 757     return agent_oids.count(oid);
 758   }
 759
 760   /// get count of active agent ops
 761   int agent_get_num_ops() {
 762     Mutex::Locker l(agent_lock);
 763     return agent_ops;
 764   }
 765
 766   void agent_inc_high_count() {
 767     Mutex::Locker l(agent_lock);
 768     flush_mode_high_count ++;
 769   }
 770
 771   void agent_dec_high_count() {
 772     Mutex::Locker l(agent_lock);
 773     flush_mode_high_count --;
 774   }
 775
 776 private:
 777   /// throttle promotion attempts
 778   std::atomic_uint promote_probability_millis{1000}; ///< probability thousands. one word.
 779   PromoteCounter promote_counter;
 780   utime_t last_recalibrate;
 781   unsigned long promote_max_objects, promote_max_bytes;
 782
 783 public:
 784   bool promote_throttle() {
 785     // NOTE: lockless!  we rely on the probability being a single word.
 786     promote_counter.attempt();
 787     if ((unsigned)rand() % 1000 > promote_probability_millis)
 788       return true;  // yes throttle (no promote)
 789     if (promote_max_objects &&
 790         promote_counter.objects > promote_max_objects)
 791       return true;  // yes throttle
 792     if (promote_max_bytes &&
 793         promote_counter.bytes > promote_max_bytes)
 794       return true;  // yes throttle
 795     return false;   //  no throttle (promote)
 796   }
 797   void promote_finish(uint64_t bytes) {
 798     promote_counter.finish(bytes);
 799   }
 800   void promote_throttle_recalibrate();
 801
 802   // -- Objecter, for tiering reads/writes from/to other OSDs --
 803   Objecter *objecter;
 804   Finisher objecter_finisher;
 805
 806   // -- Watch --
 807   Mutex watch_lock;
 808   SafeTimer watch_timer;
 809   uint64_t next_notif_id;
 810   uint64_t get_next_id(epoch_t cur_epoch) {
 811     Mutex::Locker l(watch_lock);
 812     return (((uint64_t)cur_epoch) << 32) | ((uint64_t)(next_notif_id++));
 813   }
 814
 815   // -- Recovery/Backfill Request Scheduling --
 816   Mutex recovery_request_lock;
 817   SafeTimer recovery_request_timer;
 818
 819   // For async recovery sleep
 820   bool recovery_needs_sleep = true;
 821   utime_t recovery_schedule_time = utime_t();
 822
 823   Mutex recovery_sleep_lock;
 824   SafeTimer recovery_sleep_timer;
 825
 826   // -- tids --
 827   // for ops i issue
 828   std::atomic_uint last_tid{0};
 829   ceph_tid_t get_tid() {
 830     return (ceph_tid_t)last_tid++;
 831   }
 832
 833   // -- backfill_reservation --
 834   Finisher reserver_finisher;
 835   AsyncReserver<spg_t> local_reserver;
 836   AsyncReserver<spg_t> remote_reserver;
 837
 838   // -- pg_temp --
 839 private:
 840   Mutex pg_temp_lock;
 841   struct pg_temp_t {
 842     pg_temp_t()
 843     {}
 844     pg_temp_t(vector<int> v, bool f)
 845       : acting{v}, forced{f}
 846     {}
 847     vector<int> acting;
 848     bool forced = false;
 849   };
 850   map<pg_t, pg_temp_t> pg_temp_wanted;
 851   map<pg_t, pg_temp_t> pg_temp_pending;
 852   void _sent_pg_temp();
 853   friend std::ostream& operator<<(std::ostream&, const pg_temp_t&);
 854 public:
 855   void queue_want_pg_temp(pg_t pgid, const vector<int>& want,
 856                           bool forced = false);
 857   void remove_want_pg_temp(pg_t pgid);
 858   void requeue_pg_temp();
 859   void send_pg_temp();
 860
 861   void send_pg_created(pg_t pgid);
 862
 863   void queue_for_peering(PG *pg);
 864
 865   Mutex snap_sleep_lock;
 866   SafeTimer snap_sleep_timer;
 867
 868   Mutex scrub_sleep_lock;
 869   SafeTimer scrub_sleep_timer;
 870
 871   AsyncReserver<spg_t> snap_reserver;
 872   void queue_for_snap_trim(PG *pg);
 873
 874   void queue_for_scrub(PG *pg, bool with_high_priority) {
 875     unsigned scrub_queue_priority = pg->scrubber.priority;
 876     if (with_high_priority && scrub_queue_priority < cct->_conf->osd_client_op_priority) {
 877       scrub_queue_priority = cct->_conf->osd_client_op_priority;
 878     }
 879     enqueue_back(
 880       pg->info.pgid,
 881       PGQueueable(
 882         PGScrub(pg->get_osdmap()->get_epoch()),
 883         cct->_conf->osd_scrub_cost,
 884         scrub_queue_priority,
 885         ceph_clock_now(),
 886         entity_inst_t(),
 887         pg->get_osdmap()->get_epoch()));
 888   }
 889
 890 private:
 891   // -- pg recovery and associated throttling --
 892   Mutex recovery_lock;
 893   list<pair<epoch_t, PGRef> > awaiting_throttle;
 894
 895   utime_t defer_recovery_until;
 896   uint64_t recovery_ops_active;
 897   uint64_t recovery_ops_reserved;
 898   bool recovery_paused;
 899 #ifdef DEBUG_RECOVERY_OIDS
 900   map<spg_t, set<hobject_t> > recovery_oids;
 901 #endif
 902   bool _recover_now(uint64_t *available_pushes);
 903   void _maybe_queue_recovery();
 904   void _queue_for_recovery(
 905     pair<epoch_t, PGRef> p, uint64_t reserved_pushes) {
 906     assert(recovery_lock.is_locked_by_me());
 907     enqueue_back(
 908       p.second->info.pgid,
 909       PGQueueable(
 910         PGRecovery(p.first, reserved_pushes),
 911         cct->_conf->osd_recovery_cost,
 912         cct->_conf->osd_recovery_priority,
 913         ceph_clock_now(),
 914         entity_inst_t(),
 915         p.first));
 916   }
 917 public:
 918   void start_recovery_op(PG *pg, const hobject_t& soid);
 919   void finish_recovery_op(PG *pg, const hobject_t& soid, bool dequeue);
 920   bool is_recovery_active();
 921   void release_reserved_pushes(uint64_t pushes) {
 922     Mutex::Locker l(recovery_lock);
 923     assert(recovery_ops_reserved >= pushes);
 924     recovery_ops_reserved -= pushes;
 925     _maybe_queue_recovery();
 926   }
 927   void defer_recovery(float defer_for) {
 928     defer_recovery_until = ceph_clock_now();
 929     defer_recovery_until += defer_for;
 930   }
 931   void pause_recovery() {
 932     Mutex::Locker l(recovery_lock);
 933     recovery_paused = true;
 934   }
 935   bool recovery_is_paused() {
 936     Mutex::Locker l(recovery_lock);
 937     return recovery_paused;
 938   }
 939   void unpause_recovery() {
 940     Mutex::Locker l(recovery_lock);
 941     recovery_paused = false;
 942     _maybe_queue_recovery();
 943   }
 944   void kick_recovery_queue() {
 945     Mutex::Locker l(recovery_lock);
 946     _maybe_queue_recovery();
 947   }
 948   void clear_queued_recovery(PG *pg) {
 949     Mutex::Locker l(recovery_lock);
 950     for (list<pair<epoch_t, PGRef> >::iterator i = awaiting_throttle.begin();
 951          i != awaiting_throttle.end();
 952       ) {
 953       if (i->second.get() == pg) {
 954         awaiting_throttle.erase(i);
 955         return;
 956       } else {
 957         ++i;
 958       }
 959     }
 960   }
 961   // delayed pg activation
 962   void queue_for_recovery(PG *pg) {
 963     Mutex::Locker l(recovery_lock);
 964
 965     if (pg->get_state() & (PG_STATE_FORCED_RECOVERY | PG_STATE_FORCED_BACKFILL)) {
 966       awaiting_throttle.push_front(make_pair(pg->get_osdmap()->get_epoch(), pg));
 967     } else {
 968       awaiting_throttle.push_back(make_pair(pg->get_osdmap()->get_epoch(), pg));
 969     }
 970     _maybe_queue_recovery();
 971   }
 972   void queue_recovery_after_sleep(PG *pg, epoch_t queued, uint64_t reserved_pushes) {
 973     Mutex::Locker l(recovery_lock);
 974     _queue_for_recovery(make_pair(queued, pg), reserved_pushes);
 975   }
 976
 977   void adjust_pg_priorities(const vector<PGRef>& pgs, int newflags);
 978
 979   // osd map cache (past osd maps)
 980   Mutex map_cache_lock;
 981   SharedLRU<epoch_t, const OSDMap> map_cache;
 982   SimpleLRU<epoch_t, bufferlist> map_bl_cache;
 983   SimpleLRU<epoch_t, bufferlist> map_bl_inc_cache;
 984
 985   OSDMapRef try_get_map(epoch_t e);
 986   OSDMapRef get_map(epoch_t e) {
 987     OSDMapRef ret(try_get_map(e));
 988     assert(ret);
 989     return ret;
 990   }
 991   OSDMapRef add_map(OSDMap *o) {
 992     Mutex::Locker l(map_cache_lock);
 993     return _add_map(o);
 994   }
 995   OSDMapRef _add_map(OSDMap *o);
 996
 997   void add_map_bl(epoch_t e, bufferlist& bl) {
 998     Mutex::Locker l(map_cache_lock);
 999     return _add_map_bl(e, bl);
1000   }
1001   void pin_map_bl(epoch_t e, bufferlist &bl);
1002   void _add_map_bl(epoch_t e, bufferlist& bl);
1003   bool get_map_bl(epoch_t e, bufferlist& bl) {
1004     Mutex::Locker l(map_cache_lock);
1005     return _get_map_bl(e, bl);
1006   }
1007   bool _get_map_bl(epoch_t e, bufferlist& bl);
1008
1009   void add_map_inc_bl(epoch_t e, bufferlist& bl) {
1010     Mutex::Locker l(map_cache_lock);
1011     return _add_map_inc_bl(e, bl);
1012   }
1013   void pin_map_inc_bl(epoch_t e, bufferlist &bl);
1014   void _add_map_inc_bl(epoch_t e, bufferlist& bl);
1015   bool get_inc_map_bl(epoch_t e, bufferlist& bl);
1016
1017   void clear_map_bl_cache_pins(epoch_t e);
1018
1019   void need_heartbeat_peer_update();
1020
1021   void pg_stat_queue_enqueue(PG *pg);
1022   void pg_stat_queue_dequeue(PG *pg);
1023
1024   void init();
1025   void final_init();
1026   void start_shutdown();
1027   void shutdown_reserver();
1028   void shutdown();
1029
1030 private:
1031   // split
1032   Mutex in_progress_split_lock;
1033   map<spg_t, spg_t> pending_splits; // child -> parent
1034   map<spg_t, set<spg_t> > rev_pending_splits; // parent -> [children]
1035   set<spg_t> in_progress_splits;       // child
1036
1037 public:
1038   void _start_split(spg_t parent, const set<spg_t> &children);
1039   void start_split(spg_t parent, const set<spg_t> &children) {
1040     Mutex::Locker l(in_progress_split_lock);
1041     return _start_split(parent, children);
1042   }
1043   void mark_split_in_progress(spg_t parent, const set<spg_t> &pgs);
1044   void complete_split(const set<spg_t> &pgs);
1045   void cancel_pending_splits_for_parent(spg_t parent);
1046   void _cancel_pending_splits_for_parent(spg_t parent);
1047   bool splitting(spg_t pgid);
1048   void expand_pg_num(OSDMapRef old_map,
1049                      OSDMapRef new_map);
1050   void _maybe_split_pgid(OSDMapRef old_map,
1051                          OSDMapRef new_map,
1052                          spg_t pgid);
1053   void init_splits_between(spg_t pgid, OSDMapRef frommap, OSDMapRef tomap);
1054
1055   // -- stats --
1056   Mutex stat_lock;
1057   osd_stat_t osd_stat;
1058   uint32_t seq = 0;
1059
1060   void update_osd_stat(vector<int>& hb_peers);
1061   osd_stat_t set_osd_stat(const struct store_statfs_t &stbuf,
1062                           vector<int>& hb_peers,
1063                           int num_pgs);
1064   osd_stat_t get_osd_stat() {
1065     Mutex::Locker l(stat_lock);
1066     ++seq;
1067     osd_stat.up_from = up_epoch;
1068     osd_stat.seq = ((uint64_t)osd_stat.up_from << 32) + seq;
1069     return osd_stat;
1070   }
1071   uint64_t get_osd_stat_seq() {
1072     Mutex::Locker l(stat_lock);
1073     return osd_stat.seq;
1074   }
1075
1076   // -- OSD Full Status --
1077 private:
1078   friend TestOpsSocketHook;
1079   mutable Mutex full_status_lock;
1080   enum s_names { INVALID = -1, NONE, NEARFULL, BACKFILLFULL, FULL, FAILSAFE } cur_state;  // ascending
1081   const char *get_full_state_name(s_names s) const {
1082     switch (s) {
1083     case NONE: return "none";
1084     case NEARFULL: return "nearfull";
1085     case BACKFILLFULL: return "backfillfull";
1086     case FULL: return "full";
1087     case FAILSAFE: return "failsafe";
1088     default: return "???";
1089     }
1090   }
1091   s_names get_full_state(string type) const {
1092     if (type == "none")
1093       return NONE;
1094     else if (type == "failsafe")
1095       return FAILSAFE;
1096     else if (type == "full")
1097       return FULL;
1098     else if (type == "backfillfull")
1099       return BACKFILLFULL;
1100     else if (type == "nearfull")
1101       return NEARFULL;
1102     else
1103       return INVALID;
1104   }
1105   double cur_ratio;  ///< current utilization
1106   mutable int64_t injectfull = 0;
1107   s_names injectfull_state = NONE;
1108   float get_failsafe_full_ratio();
1109   void check_full_status(float ratio);
1110   bool _check_full(s_names type, ostream &ss) const;
1111 public:
1112   bool check_failsafe_full(ostream &ss) const;
1113   bool check_full(ostream &ss) const;
1114   bool check_backfill_full(ostream &ss) const;
1115   bool check_nearfull(ostream &ss) const;
1116   bool is_failsafe_full() const;
1117   bool is_full() const;
1118   bool is_backfillfull() const;
1119   bool is_nearfull() const;
1120   bool need_fullness_update();  ///< osdmap state needs update
1121   void set_injectfull(s_names type, int64_t count);
1122   bool check_osdmap_full(const set<pg_shard_t> &missing_on);
1123
1124
1125   // -- epochs --
1126 private:
1127   mutable Mutex epoch_lock; // protects access to boot_epoch, up_epoch, bind_epoch
1128   epoch_t boot_epoch;  // _first_ epoch we were marked up (after this process started)
1129   epoch_t up_epoch;    // _most_recent_ epoch we were marked up
1130   epoch_t bind_epoch;  // epoch we last did a bind to new ip:ports
1131 public:
1132   /**
1133    * Retrieve the boot_, up_, and bind_ epochs the OSD has set. The params
1134    * can be NULL if you don't care about them.
1135    */
1136   void retrieve_epochs(epoch_t *_boot_epoch, epoch_t *_up_epoch,
1137                        epoch_t *_bind_epoch) const;
1138   /**
1139    * Set the boot, up, and bind epochs. Any NULL params will not be set.
1140    */
1141   void set_epochs(const epoch_t *_boot_epoch, const epoch_t *_up_epoch,
1142                   const epoch_t *_bind_epoch);
1143   epoch_t get_boot_epoch() const {
1144     epoch_t ret;
1145     retrieve_epochs(&ret, NULL, NULL);
1146     return ret;
1147   }
1148   epoch_t get_up_epoch() const {
1149     epoch_t ret;
1150     retrieve_epochs(NULL, &ret, NULL);
1151     return ret;
1152   }
1153   epoch_t get_bind_epoch() const {
1154     epoch_t ret;
1155     retrieve_epochs(NULL, NULL, &ret);
1156     return ret;
1157   }
1158
1159   void request_osdmap_update(epoch_t e);
1160
1161   // -- stopping --
1162   Mutex is_stopping_lock;
1163   Cond is_stopping_cond;
1164   enum {
1165     NOT_STOPPING,
1166     PREPARING_TO_STOP,
1167     STOPPING };
1168   std::atomic_int state{NOT_STOPPING};
1169   int get_state() {
1170     return state;
1171   }
1172   void set_state(int s) {
1173     state = s;
1174   }
1175   bool is_stopping() const {
1176     return state == STOPPING;
1177   }
1178   bool is_preparing_to_stop() const {
1179     return state == PREPARING_TO_STOP;
1180   }
1181   bool prepare_to_stop();
1182   void got_stop_ack();
1183
1184
1185 #ifdef PG_DEBUG_REFS
1186   Mutex pgid_lock;
1187   map<spg_t, int> pgid_tracker;
1188   map<spg_t, PG*> live_pgs;
1189   void add_pgid(spg_t pgid, PG *pg);
1190   void remove_pgid(spg_t pgid, PG *pg);
1191   void dump_live_pgids();
1192 #endif
1193
1194   explicit OSDService(OSD *osd);
1195   ~OSDService();
1196 };
1197
1198 class OSD : public Dispatcher,
1199             public md_config_obs_t {
1200   /** OSD **/
1201   Mutex osd_lock;                       // global lock
1202   SafeTimer tick_timer;    // safe timer (osd_lock)
1203
1204   // Tick timer for those stuff that do not need osd_lock
1205   Mutex tick_timer_lock;
1206   SafeTimer tick_timer_without_osd_lock;
1207 public:
1208   // config observer bits
1209   const char** get_tracked_conf_keys() const override;
1210   void handle_conf_change(const struct md_config_t *conf,
1211                           const std::set <std::string> &changed) override;
1212   void update_log_config();
1213   void check_config();
1214
1215 protected:
1216
1217   const double OSD_TICK_INTERVAL = { 1.0 };
1218   double get_tick_interval() const;
1219
1220   AuthAuthorizeHandlerRegistry *authorize_handler_cluster_registry;
1221   AuthAuthorizeHandlerRegistry *authorize_handler_service_registry;
1222
1223   Messenger   *cluster_messenger;
1224   Messenger   *client_messenger;
1225   Messenger   *objecter_messenger;
1226   MonClient   *monc; // check the "monc helpers" list before accessing directly
1227   MgrClient   mgrc;
1228   PerfCounters      *logger;
1229   PerfCounters      *recoverystate_perf;
1230   ObjectStore *store;
1231 #ifdef HAVE_LIBFUSE
1232   FuseStore *fuse_store = nullptr;
1233 #endif
1234   LogClient log_client;
1235   LogChannelRef clog;
1236
1237   int whoami;
1238   std::string dev_path, journal_path;
1239
1240   bool store_is_rotational = true;
1241   bool journal_is_rotational = true;
1242
1243   ZTracer::Endpoint trace_endpoint;
1244   void create_logger();
1245   void create_recoverystate_perf();
1246   void tick();
1247   void tick_without_osd_lock();
1248   void _dispatch(Message *m);
1249   void dispatch_op(OpRequestRef op);
1250
1251   void check_osdmap_features(ObjectStore *store);
1252
1253   // asok
1254   friend class OSDSocketHook;
1255   class OSDSocketHook *asok_hook;
1256   bool asok_command(string admin_command, cmdmap_t& cmdmap, string format, ostream& ss);
1257
1258 public:
1259   ClassHandler  *class_handler = nullptr;
1260   int get_nodeid() { return whoami; }
1261
1262   static ghobject_t get_osdmap_pobject_name(epoch_t epoch) {
1263     char foo[20];
1264     snprintf(foo, sizeof(foo), "osdmap.%d", epoch);
1265     return ghobject_t(hobject_t(sobject_t(object_t(foo), 0)));
1266   }
1267   static ghobject_t get_inc_osdmap_pobject_name(epoch_t epoch) {
1268     char foo[22];
1269     snprintf(foo, sizeof(foo), "inc_osdmap.%d", epoch);
1270     return ghobject_t(hobject_t(sobject_t(object_t(foo), 0)));
1271   }
1272
1273   static ghobject_t make_snapmapper_oid() {
1274     return ghobject_t(hobject_t(
1275       sobject_t(
1276         object_t("snapmapper"),
1277         0)));
1278   }
1279
1280   static ghobject_t make_pg_log_oid(spg_t pg) {
1281     stringstream ss;
1282     ss << "pglog_" << pg;
1283     string s;
1284     getline(ss, s);
1285     return ghobject_t(hobject_t(sobject_t(object_t(s.c_str()), 0)));
1286   }
1287
1288   static ghobject_t make_pg_biginfo_oid(spg_t pg) {
1289     stringstream ss;
1290     ss << "pginfo_" << pg;
1291     string s;
1292     getline(ss, s);
1293     return ghobject_t(hobject_t(sobject_t(object_t(s.c_str()), 0)));
1294   }
1295   static ghobject_t make_infos_oid() {
1296     hobject_t oid(sobject_t("infos", CEPH_NOSNAP));
1297     return ghobject_t(oid);
1298   }
1299   static void recursive_remove_collection(CephContext* cct,
1300                                           ObjectStore *store,
1301                                           spg_t pgid,
1302                                           coll_t tmp);
1303
1304   /**
1305    * get_osd_initial_compat_set()
1306    *
1307    * Get the initial feature set for this OSD.  Features
1308    * here are automatically upgraded.
1309    *
1310    * Return value: Initial osd CompatSet
1311    */
1312   static CompatSet get_osd_initial_compat_set();
1313
1314   /**
1315    * get_osd_compat_set()
1316    *
1317    * Get all features supported by this OSD
1318    *
1319    * Return value: CompatSet of all supported features
1320    */
1321   static CompatSet get_osd_compat_set();
1322
1323
1324 private:
1325   class C_Tick;
1326   class C_Tick_WithoutOSDLock;
1327
1328   // -- superblock --
1329   OSDSuperblock superblock;
1330
1331   void write_superblock();
1332   void write_superblock(ObjectStore::Transaction& t);
1333   int read_superblock();
1334
1335   void clear_temp_objects();
1336
1337   CompatSet osd_compat;
1338
1339   // -- state --
1340 public:
1341   typedef enum {
1342     STATE_INITIALIZING = 1,
1343     STATE_PREBOOT,
1344     STATE_BOOTING,
1345     STATE_ACTIVE,
1346     STATE_STOPPING,
1347     STATE_WAITING_FOR_HEALTHY
1348   } osd_state_t;
1349
1350   static const char *get_state_name(int s) {
1351     switch (s) {
1352     case STATE_INITIALIZING: return "initializing";
1353     case STATE_PREBOOT: return "preboot";
1354     case STATE_BOOTING: return "booting";
1355     case STATE_ACTIVE: return "active";
1356     case STATE_STOPPING: return "stopping";
1357     case STATE_WAITING_FOR_HEALTHY: return "waiting_for_healthy";
1358     default: return "???";
1359     }
1360   }
1361
1362 private:
1363   std::atomic_int state{STATE_INITIALIZING};
1364   bool waiting_for_luminous_mons = false;
1365
1366 public:
1367   int get_state() const {
1368     return state;
1369   }
1370   void set_state(int s) {
1371     state = s;
1372   }
1373   bool is_initializing() const {
1374     return state == STATE_INITIALIZING;
1375   }
1376   bool is_preboot() const {
1377     return state == STATE_PREBOOT;
1378   }
1379   bool is_booting() const {
1380     return state == STATE_BOOTING;
1381   }
1382   bool is_active() const {
1383     return state == STATE_ACTIVE;
1384   }
1385   bool is_stopping() const {
1386     return state == STATE_STOPPING;
1387   }
1388   bool is_waiting_for_healthy() const {
1389     return state == STATE_WAITING_FOR_HEALTHY;
1390   }
1391
1392 private:
1393
1394   ThreadPool peering_tp;
1395   ShardedThreadPool osd_op_tp;
1396   ThreadPool disk_tp;
1397   ThreadPool command_tp;
1398
1399   void set_disk_tp_priority();
1400   void get_latest_osdmap();
1401
1402   // -- sessions --
1403 private:
1404   void dispatch_session_waiting(Session *session, OSDMapRef osdmap);
1405   void maybe_share_map(Session *session, OpRequestRef op, OSDMapRef osdmap);
1406
1407   Mutex session_waiting_lock;
1408   set<Session*> session_waiting_for_map;
1409
1410   /// Caller assumes refs for included Sessions
1411   void get_sessions_waiting_for_map(set<Session*> *out) {
1412     Mutex::Locker l(session_waiting_lock);
1413     out->swap(session_waiting_for_map);
1414   }
1415   void register_session_waiting_on_map(Session *session) {
1416     Mutex::Locker l(session_waiting_lock);
1417     if (session_waiting_for_map.insert(session).second) {
1418       session->get();
1419     }
1420   }
1421   void clear_session_waiting_on_map(Session *session) {
1422     Mutex::Locker l(session_waiting_lock);
1423     set<Session*>::iterator i = session_waiting_for_map.find(session);
1424     if (i != session_waiting_for_map.end()) {
1425       (*i)->put();
1426       session_waiting_for_map.erase(i);
1427     }
1428   }
1429   void dispatch_sessions_waiting_on_map() {
1430     set<Session*> sessions_to_check;
1431     get_sessions_waiting_for_map(&sessions_to_check);
1432     for (set<Session*>::iterator i = sessions_to_check.begin();
1433          i != sessions_to_check.end();
1434          sessions_to_check.erase(i++)) {
1435       (*i)->session_dispatch_lock.Lock();
1436       dispatch_session_waiting(*i, osdmap);
1437       (*i)->session_dispatch_lock.Unlock();
1438       (*i)->put();
1439     }
1440   }
1441   void session_handle_reset(Session *session) {
1442     Mutex::Locker l(session->session_dispatch_lock);
1443     clear_session_waiting_on_map(session);
1444
1445     session->clear_backoffs();
1446
1447     /* Messages have connection refs, we need to clear the
1448      * connection->session->message->connection
1449      * cycles which result.
1450      * Bug #12338
1451      */
1452     session->waiting_on_map.clear_and_dispose(TrackedOp::Putter());
1453   }
1454
1455 private:
1456   /**
1457    * @defgroup monc helpers
1458    * @{
1459    * Right now we only have the one
1460    */
1461
1462   /**
1463    * Ask the Monitors for a sequence of OSDMaps.
1464    *
1465    * @param epoch The epoch to start with when replying
1466    * @param force_request True if this request forces a new subscription to
1467    * the monitors; false if an outstanding request that encompasses it is
1468    * sufficient.
1469    */
1470   void osdmap_subscribe(version_t epoch, bool force_request);
1471   /** @} monc helpers */
1472
1473   Mutex osdmap_subscribe_lock;
1474   epoch_t latest_subscribed_epoch{0};
1475
1476   // -- heartbeat --
1477   /// information about a heartbeat peer
1478   struct HeartbeatInfo {
1479     int peer;           ///< peer
1480     ConnectionRef con_front;   ///< peer connection (front)
1481     ConnectionRef con_back;    ///< peer connection (back)
1482     utime_t first_tx;   ///< time we sent our first ping request
1483     utime_t last_tx;    ///< last time we sent a ping request
1484     utime_t last_rx_front;  ///< last time we got a ping reply on the front side
1485     utime_t last_rx_back;   ///< last time we got a ping reply on the back side
1486     epoch_t epoch;      ///< most recent epoch we wanted this peer
1487
1488     bool is_unhealthy(utime_t cutoff) const {
1489       return
1490         ! ((last_rx_front > cutoff ||
1491             (last_rx_front == utime_t() && (last_tx == utime_t() ||
1492                                             first_tx > cutoff))) &&
1493            (last_rx_back > cutoff ||
1494             (last_rx_back == utime_t() && (last_tx == utime_t() ||
1495                                            first_tx > cutoff))));
1496     }
1497     bool is_healthy(utime_t cutoff) const {
1498       return last_rx_front > cutoff && last_rx_back > cutoff;
1499     }
1500
1501   };
1502   /// state attached to outgoing heartbeat connections
1503   struct HeartbeatSession : public RefCountedObject {
1504     int peer;
1505     explicit HeartbeatSession(int p) : peer(p) {}
1506   };
1507   Mutex heartbeat_lock;
1508   map<int, int> debug_heartbeat_drops_remaining;
1509   Cond heartbeat_cond;
1510   bool heartbeat_stop;
1511   std::atomic_bool heartbeat_need_update;
1512   map<int,HeartbeatInfo> heartbeat_peers;  ///< map of osd id to HeartbeatInfo
1513   utime_t last_mon_heartbeat;
1514   Messenger *hb_front_client_messenger;
1515   Messenger *hb_back_client_messenger;
1516   Messenger *hb_front_server_messenger;
1517   Messenger *hb_back_server_messenger;
1518   utime_t last_heartbeat_resample;   ///< last time we chose random peers in waiting-for-healthy state
1519   double daily_loadavg;
1520
1521   void _add_heartbeat_peer(int p);
1522   void _remove_heartbeat_peer(int p);
1523   bool heartbeat_reset(Connection *con);
1524   void maybe_update_heartbeat_peers();
1525   void reset_heartbeat_peers();
1526   bool heartbeat_peers_need_update() {
1527     return heartbeat_need_update.load();
1528   }
1529   void heartbeat_set_peers_need_update() {
1530     heartbeat_need_update.store(true);
1531   }
1532   void heartbeat_clear_peers_need_update() {
1533     heartbeat_need_update.store(false);
1534   }
1535   void heartbeat();
1536   void heartbeat_check();
1537   void heartbeat_entry();
1538   void need_heartbeat_peer_update();
1539
1540   void heartbeat_kick() {
1541     Mutex::Locker l(heartbeat_lock);
1542     heartbeat_cond.Signal();
1543   }
1544
1545   struct T_Heartbeat : public Thread {
1546     OSD *osd;
1547     explicit T_Heartbeat(OSD *o) : osd(o) {}
1548     void *entry() override {
1549       osd->heartbeat_entry();
1550       return 0;
1551     }
1552   } heartbeat_thread;
1553
1554 public:
1555   bool heartbeat_dispatch(Message *m);
1556
1557   struct HeartbeatDispatcher : public Dispatcher {
1558     OSD *osd;
1559     explicit HeartbeatDispatcher(OSD *o) : Dispatcher(o->cct), osd(o) {}
1560
1561     bool ms_can_fast_dispatch_any() const override { return true; }
1562     bool ms_can_fast_dispatch(const Message *m) const override {
1563       switch (m->get_type()) {
1564         case CEPH_MSG_PING:
1565         case MSG_OSD_PING:
1566           return true;
1567         default:
1568           return false;
1569         }
1570     }
1571     void ms_fast_dispatch(Message *m) override {
1572       osd->heartbeat_dispatch(m);
1573     }
1574     bool ms_dispatch(Message *m) override {
1575       return osd->heartbeat_dispatch(m);
1576     }
1577     bool ms_handle_reset(Connection *con) override {
1578       return osd->heartbeat_reset(con);
1579     }
1580     void ms_handle_remote_reset(Connection *con) override {}
1581     bool ms_handle_refused(Connection *con) override {
1582       return osd->ms_handle_refused(con);
1583     }
1584     bool ms_verify_authorizer(Connection *con, int peer_type,
1585                               int protocol, bufferlist& authorizer_data, bufferlist& authorizer_reply,
1586                               bool& isvalid, CryptoKey& session_key,
1587                               std::unique_ptr<AuthAuthorizerChallenge> *challenge) override {
1588       isvalid = true;
1589       return true;
1590     }
1591   } heartbeat_dispatcher;
1592
1593 private:
1594   // -- waiters --
1595   list<OpRequestRef> finished;
1596
1597   void take_waiters(list<OpRequestRef>& ls) {
1598     assert(osd_lock.is_locked());
1599     finished.splice(finished.end(), ls);
1600   }
1601   void do_waiters();
1602
1603   // -- op tracking --
1604   OpTracker op_tracker;
1605   void check_ops_in_flight();
1606   void test_ops(std::string command, std::string args, ostream& ss);
1607   friend class TestOpsSocketHook;
1608   TestOpsSocketHook *test_ops_hook;
1609   friend struct C_CompleteSplits;
1610   friend struct C_OpenPGs;
1611
1612   // -- op queue --
1613   enum class io_queue {
1614     prioritized,
1615     weightedpriority,
1616     mclock_opclass,
1617     mclock_client,
1618   };
1619   friend std::ostream& operator<<(std::ostream& out, const OSD::io_queue& q);
1620
1621   const io_queue op_queue;
1622   const unsigned int op_prio_cutoff;
1623
1624   /*
1625    * The ordered op delivery chain is:
1626    *
1627    *   fast dispatch -> pqueue back
1628    *                    pqueue front <-> to_process back
1629    *                                     to_process front  -> RunVis(item)
1630    *                                                      <- queue_front()
1631    *
1632    * The pqueue is per-shard, and to_process is per pg_slot.  Items can be
1633    * pushed back up into to_process and/or pqueue while order is preserved.
1634    *
1635    * Multiple worker threads can operate on each shard.
1636    *
1637    * Under normal circumstances, num_running == to_proces.size().  There are
1638    * two times when that is not true: (1) when waiting_for_pg == true and
1639    * to_process is accumulating requests that are waiting for the pg to be
1640    * instantiated; in that case they will all get requeued together by
1641    * wake_pg_waiters, and (2) when wake_pg_waiters just ran, waiting_for_pg
1642    * and already requeued the items.
1643    */
1644   friend class PGQueueable;
1645
1646   class ShardedOpWQ
1647     : public ShardedThreadPool::ShardedWQ<pair<spg_t,PGQueueable>>
1648   {
1649     struct ShardData {
1650       Mutex sdata_lock;
1651       Cond sdata_cond;
1652
1653       Mutex sdata_op_ordering_lock;   ///< protects all members below
1654
1655       OSDMapRef waiting_for_pg_osdmap;
1656       struct pg_slot {
1657         PGRef pg;                     ///< cached pg reference [optional]
1658         list<PGQueueable> to_process; ///< order items for this slot
1659         int num_running = 0;          ///< _process threads doing pg lookup/lock
1660
1661         /// true if pg does/did not exist. if so all new items go directly to
1662         /// to_process.  cleared by prune_pg_waiters.
1663         bool waiting_for_pg = false;
1664
1665         /// incremented by wake_pg_waiters; indicates racing _process threads
1666         /// should bail out (their op has been requeued)
1667         uint64_t requeue_seq = 0;
1668       };
1669
1670       /// map of slots for each spg_t.  maintains ordering of items dequeued
1671       /// from pqueue while _process thread drops shard lock to acquire the
1672       /// pg lock.  slots are removed only by prune_pg_waiters.
1673       unordered_map<spg_t,pg_slot> pg_slots;
1674
1675       /// priority queue
1676       std::unique_ptr<OpQueue< pair<spg_t, PGQueueable>, entity_inst_t>> pqueue;
1677
1678       void _enqueue_front(pair<spg_t, PGQueueable> item, unsigned cutoff) {
1679         unsigned priority = item.second.get_priority();
1680         unsigned cost = item.second.get_cost();
1681         if (priority >= cutoff)
1682           pqueue->enqueue_strict_front(
1683             item.second.get_owner(),
1684             priority, item);
1685         else
1686           pqueue->enqueue_front(
1687             item.second.get_owner(),
1688             priority, cost, item);
1689       }
1690
1691       ShardData(
1692         string lock_name, string ordering_lock,
1693         uint64_t max_tok_per_prio, uint64_t min_cost, CephContext *cct,
1694         io_queue opqueue)
1695         : sdata_lock(lock_name.c_str(), false, true, false, cct),
1696           sdata_op_ordering_lock(ordering_lock.c_str(), false, true,
1697                                  false, cct) {
1698         if (opqueue == io_queue::weightedpriority) {
1699           pqueue = std::unique_ptr
1700             <WeightedPriorityQueue<pair<spg_t,PGQueueable>,entity_inst_t>>(
1701               new WeightedPriorityQueue<pair<spg_t,PGQueueable>,entity_inst_t>(
1702                 max_tok_per_prio, min_cost));
1703         } else if (opqueue == io_queue::prioritized) {
1704           pqueue = std::unique_ptr
1705             <PrioritizedQueue<pair<spg_t,PGQueueable>,entity_inst_t>>(
1706               new PrioritizedQueue<pair<spg_t,PGQueueable>,entity_inst_t>(
1707                 max_tok_per_prio, min_cost));
1708         } else if (opqueue == io_queue::mclock_opclass) {
1709           pqueue = std::unique_ptr
1710             <ceph::mClockOpClassQueue>(new ceph::mClockOpClassQueue(cct));
1711         } else if (opqueue == io_queue::mclock_client) {
1712           pqueue = std::unique_ptr
1713             <ceph::mClockClientQueue>(new ceph::mClockClientQueue(cct));
1714         }
1715       }
1716     }; // struct ShardData
1717
1718     vector<ShardData*> shard_list;
1719     OSD *osd;
1720     uint32_t num_shards;
1721
1722   public:
1723     ShardedOpWQ(uint32_t pnum_shards,
1724                 OSD *o,
1725                 time_t ti,
1726                 time_t si,
1727                 ShardedThreadPool* tp)
1728       : ShardedThreadPool::ShardedWQ<pair<spg_t,PGQueueable>>(ti, si, tp),
1729         osd(o),
1730         num_shards(pnum_shards) {
1731       for (uint32_t i = 0; i < num_shards; i++) {
1732         char lock_name[32] = {0};
1733         snprintf(lock_name, sizeof(lock_name), "%s.%d", "OSD:ShardedOpWQ:", i);
1734         char order_lock[32] = {0};
1735         snprintf(order_lock, sizeof(order_lock), "%s.%d",
1736                  "OSD:ShardedOpWQ:order:", i);
1737         ShardData* one_shard = new ShardData(
1738           lock_name, order_lock,
1739           osd->cct->_conf->osd_op_pq_max_tokens_per_priority,
1740           osd->cct->_conf->osd_op_pq_min_cost, osd->cct, osd->op_queue);
1741         shard_list.push_back(one_shard);
1742       }
1743     }
1744     ~ShardedOpWQ() override {
1745       while (!shard_list.empty()) {
1746         delete shard_list.back();
1747         shard_list.pop_back();
1748       }
1749     }
1750
1751     /// wake any pg waiters after a PG is created/instantiated
1752     void wake_pg_waiters(spg_t pgid);
1753
1754     /// prune ops (and possiblye pg_slots) for pgs that shouldn't be here
1755     void prune_pg_waiters(OSDMapRef osdmap, int whoami);
1756
1757     /// clear cached PGRef on pg deletion
1758     void clear_pg_pointer(spg_t pgid);
1759
1760     /// clear pg_slots on shutdown
1761     void clear_pg_slots();
1762
1763     /// try to do some work
1764     void _process(uint32_t thread_index, heartbeat_handle_d *hb) override;
1765
1766     /// enqueue a new item
1767     void _enqueue(pair <spg_t, PGQueueable> item) override;
1768
1769     /// requeue an old item (at the front of the line)
1770     void _enqueue_front(pair <spg_t, PGQueueable> item) override;
1771
1772     void return_waiting_threads() override {
1773       for(uint32_t i = 0; i < num_shards; i++) {
1774         ShardData* sdata = shard_list[i];
1775         assert (NULL != sdata);
1776         sdata->sdata_lock.Lock();
1777         sdata->sdata_cond.Signal();
1778         sdata->sdata_lock.Unlock();
1779       }
1780     }
1781
1782     void dump(Formatter *f) {
1783       for(uint32_t i = 0; i < num_shards; i++) {
1784         ShardData* sdata = shard_list[i];
1785         char lock_name[32] = {0};
1786         snprintf(lock_name, sizeof(lock_name), "%s%d", "OSD:ShardedOpWQ:", i);
1787         assert (NULL != sdata);
1788         sdata->sdata_op_ordering_lock.Lock();
1789         f->open_object_section(lock_name);
1790         sdata->pqueue->dump(f);
1791         f->close_section();
1792         sdata->sdata_op_ordering_lock.Unlock();
1793       }
1794     }
1795
1796     /// Must be called on ops queued back to front
1797     struct Pred {
1798       spg_t pgid;
1799       list<OpRequestRef> *out_ops;
1800       uint64_t reserved_pushes_to_free;
1801       Pred(spg_t pg, list<OpRequestRef> *out_ops = 0)
1802         : pgid(pg), out_ops(out_ops), reserved_pushes_to_free(0) {}
1803       void accumulate(const PGQueueable &op) {
1804         reserved_pushes_to_free += op.get_reserved_pushes();
1805         if (out_ops) {
1806           boost::optional<OpRequestRef> mop = op.maybe_get_op();
1807           if (mop)
1808             out_ops->push_front(*mop);
1809         }
1810       }
1811       bool operator()(const pair<spg_t, PGQueueable> &op) {
1812         if (op.first == pgid) {
1813           accumulate(op.second);
1814           return true;
1815         } else {
1816           return false;
1817         }
1818       }
1819       uint64_t get_reserved_pushes_to_free() const {
1820         return reserved_pushes_to_free;
1821       }
1822     };
1823
1824     bool is_shard_empty(uint32_t thread_index) override {
1825       uint32_t shard_index = thread_index % num_shards;
1826       ShardData* sdata = shard_list[shard_index];
1827       assert(NULL != sdata);
1828       Mutex::Locker l(sdata->sdata_op_ordering_lock);
1829       return sdata->pqueue->empty();
1830     }
1831   } op_shardedwq;
1832
1833
1834   void enqueue_op(spg_t pg, OpRequestRef& op, epoch_t epoch);
1835   void dequeue_op(
1836     PGRef pg, OpRequestRef op,
1837     ThreadPool::TPHandle &handle);
1838
1839   // -- peering queue --
1840   struct PeeringWQ : public ThreadPool::BatchWorkQueue<PG> {
1841     list<PG*> peering_queue;
1842     OSD *osd;
1843     set<PG*> in_use;
1844     PeeringWQ(OSD *o, time_t ti, time_t si, ThreadPool *tp)
1845       : ThreadPool::BatchWorkQueue<PG>(
1846         "OSD::PeeringWQ", ti, si, tp), osd(o) {}
1847
1848     void _dequeue(PG *pg) override {
1849       for (list<PG*>::iterator i = peering_queue.begin();
1850            i != peering_queue.end();
1851            ) {
1852         if (*i == pg) {
1853           peering_queue.erase(i++);
1854           pg->put("PeeringWQ");
1855         } else {
1856           ++i;
1857         }
1858       }
1859     }
1860     bool _enqueue(PG *pg) override {
1861       pg->get("PeeringWQ");
1862       peering_queue.push_back(pg);
1863       return true;
1864     }
1865     bool _empty() override {
1866       return peering_queue.empty();
1867     }
1868     void _dequeue(list<PG*> *out) override;
1869     void _process(
1870       const list<PG *> &pgs,
1871       ThreadPool::TPHandle &handle) override {
1872       assert(!pgs.empty());
1873       osd->process_peering_events(pgs, handle);
1874       for (list<PG *>::const_iterator i = pgs.begin();
1875            i != pgs.end();
1876            ++i) {
1877         (*i)->put("PeeringWQ");
1878       }
1879     }
1880     void _process_finish(const list<PG *> &pgs) override {
1881       for (list<PG*>::const_iterator i = pgs.begin();
1882            i != pgs.end();
1883            ++i) {
1884         in_use.erase(*i);
1885       }
1886     }
1887     void _clear() override {
1888       assert(peering_queue.empty());
1889     }
1890   } peering_wq;
1891
1892   void process_peering_events(
1893     const list<PG*> &pg,
1894     ThreadPool::TPHandle &handle);
1895
1896   friend class PG;
1897   friend class PrimaryLogPG;
1898
1899
1900  protected:
1901
1902   // -- osd map --
1903   OSDMapRef       osdmap;
1904   OSDMapRef get_osdmap() {
1905     return osdmap;
1906   }
1907   epoch_t get_osdmap_epoch() const {
1908     return osdmap ? osdmap->get_epoch() : 0;
1909   }
1910
1911   utime_t         had_map_since;
1912   RWLock          map_lock;
1913   list<OpRequestRef>  waiting_for_osdmap;
1914   deque<utime_t> osd_markdown_log;
1915
1916   friend struct send_map_on_destruct;
1917
1918   void wait_for_new_map(OpRequestRef op);
1919   void handle_osd_map(class MOSDMap *m);
1920   void _committed_osd_maps(epoch_t first, epoch_t last, class MOSDMap *m);
1921   void trim_maps(epoch_t oldest, int nreceived, bool skip_maps);
1922   void note_down_osd(int osd);
1923   void note_up_osd(int osd);
1924   friend class C_OnMapCommit;
1925
1926   bool advance_pg(
1927     epoch_t advance_to, PG *pg,
1928     ThreadPool::TPHandle &handle,
1929     PG::RecoveryCtx *rctx,
1930     set<PGRef> *split_pgs
1931   );
1932   void consume_map();
1933   void activate_map();
1934
1935   // osd map cache (past osd maps)
1936   OSDMapRef get_map(epoch_t e) {
1937     return service.get_map(e);
1938   }
1939   OSDMapRef add_map(OSDMap *o) {
1940     return service.add_map(o);
1941   }
1942   void add_map_bl(epoch_t e, bufferlist& bl) {
1943     return service.add_map_bl(e, bl);
1944   }
1945   void pin_map_bl(epoch_t e, bufferlist &bl) {
1946     return service.pin_map_bl(e, bl);
1947   }
1948   bool get_map_bl(epoch_t e, bufferlist& bl) {
1949     return service.get_map_bl(e, bl);
1950   }
1951   void add_map_inc_bl(epoch_t e, bufferlist& bl) {
1952     return service.add_map_inc_bl(e, bl);
1953   }
1954   void pin_map_inc_bl(epoch_t e, bufferlist &bl) {
1955     return service.pin_map_inc_bl(e, bl);
1956   }
1957
1958 protected:
1959   // -- placement groups --
1960   RWLock pg_map_lock; // this lock orders *above* individual PG _locks
1961   ceph::unordered_map<spg_t, PG*> pg_map; // protected by pg_map lock
1962
1963   std::mutex pending_creates_lock;
1964   using create_from_osd_t = std::pair<pg_t, bool /* is primary*/>;
1965   std::set<create_from_osd_t> pending_creates_from_osd;
1966   unsigned pending_creates_from_mon = 0;
1967
1968   map<spg_t, list<PG::CephPeeringEvtRef> > peering_wait_for_split;
1969   PGRecoveryStats pg_recovery_stats;
1970
1971   PGPool _get_pool(int id, OSDMapRef createmap);
1972
1973   PG   *_lookup_lock_pg_with_map_lock_held(spg_t pgid);
1974   PG   *_lookup_lock_pg(spg_t pgid);
1975
1976 public:
1977   PG   *lookup_lock_pg(spg_t pgid);
1978
1979   int get_num_pgs() {
1980     RWLock::RLocker l(pg_map_lock);
1981     return pg_map.size();
1982   }
1983
1984 protected:
1985   PG   *_open_lock_pg(OSDMapRef createmap,
1986                       spg_t pg, bool no_lockdep_check=false);
1987   enum res_result {
1988     RES_PARENT,    // resurrected a parent
1989     RES_SELF,      // resurrected self
1990     RES_NONE       // nothing relevant deleting
1991   };
1992   res_result _try_resurrect_pg(
1993     OSDMapRef curmap, spg_t pgid, spg_t *resurrected, PGRef *old_pg_state);
1994
1995   PG   *_create_lock_pg(
1996     OSDMapRef createmap,
1997     spg_t pgid,
1998     bool hold_map_lock,
1999     bool backfill,
2000     int role,
2001     vector<int>& up, int up_primary,
2002     vector<int>& acting, int acting_primary,
2003     pg_history_t history,
2004     const PastIntervals& pi,
2005     ObjectStore::Transaction& t);
2006
2007   PG* _make_pg(OSDMapRef createmap, spg_t pgid);
2008   void add_newly_split_pg(PG *pg,
2009                           PG::RecoveryCtx *rctx);
2010
2011   int handle_pg_peering_evt(
2012     spg_t pgid,
2013     const pg_history_t& orig_history,
2014     const PastIntervals& pi,
2015     epoch_t epoch,
2016     PG::CephPeeringEvtRef evt);
2017   bool maybe_wait_for_max_pg(spg_t pgid, bool is_mon_create);
2018   void resume_creating_pg();
2019
2020   void load_pgs();
2021   void build_past_intervals_parallel();
2022
2023   /// build initial pg history and intervals on create
2024   void build_initial_pg_history(
2025     spg_t pgid,
2026     epoch_t created,
2027     utime_t created_stamp,
2028     pg_history_t *h,
2029     PastIntervals *pi);
2030
2031   /// project pg history from from to now
2032   bool project_pg_history(
2033     spg_t pgid, pg_history_t& h, epoch_t from,
2034     const vector<int>& lastup,
2035     int lastupprimary,
2036     const vector<int>& lastacting,
2037     int lastactingprimary
2038     ); ///< @return false if there was a map gap between from and now
2039
2040   // this must be called with pg->lock held on any pg addition to pg_map
2041   void wake_pg_waiters(PGRef pg) {
2042     assert(pg->is_locked());
2043     op_shardedwq.wake_pg_waiters(pg->info.pgid);
2044   }
2045   epoch_t last_pg_create_epoch;
2046
2047   void handle_pg_create(OpRequestRef op);
2048
2049   void split_pgs(
2050     PG *parent,
2051     const set<spg_t> &childpgids, set<PGRef> *out_pgs,
2052     OSDMapRef curmap,
2053     OSDMapRef nextmap,
2054     PG::RecoveryCtx *rctx);
2055
2056   // == monitor interaction ==
2057   Mutex mon_report_lock;
2058   utime_t last_mon_report;
2059   utime_t last_pg_stats_sent;
2060
2061   /* if our monitor dies, we want to notice it and reconnect.
2062    *  So we keep track of when it last acked our stat updates,
2063    *  and if too much time passes (and we've been sending
2064    *  more updates) then we can call it dead and reconnect
2065    *  elsewhere.
2066    */
2067   utime_t last_pg_stats_ack;
2068   float stats_ack_timeout;
2069   set<uint64_t> outstanding_pg_stats; // how many stat updates haven't been acked yet
2070
2071   // -- boot --
2072   void start_boot();
2073   void _got_mon_epochs(epoch_t oldest, epoch_t newest);
2074   void _preboot(epoch_t oldest, epoch_t newest);
2075   void _send_boot();
2076   void _collect_metadata(map<string,string> *pmeta);
2077
2078   void start_waiting_for_healthy();
2079   bool _is_healthy();
2080
2081   void send_full_update();
2082
2083   friend struct C_OSD_GetVersion;
2084
2085   // -- alive --
2086   epoch_t up_thru_wanted;
2087
2088   void queue_want_up_thru(epoch_t want);
2089   void send_alive();
2090
2091   // -- full map requests --
2092   epoch_t requested_full_first, requested_full_last;
2093
2094   void request_full_map(epoch_t first, epoch_t last);
2095   void rerequest_full_maps() {
2096     epoch_t first = requested_full_first;
2097     epoch_t last = requested_full_last;
2098     requested_full_first = 0;
2099     requested_full_last = 0;
2100     request_full_map(first, last);
2101   }
2102   void got_full_map(epoch_t e);
2103
2104   // -- failures --
2105   map<int,utime_t> failure_queue;
2106   map<int,pair<utime_t,entity_inst_t> > failure_pending;
2107
2108   void requeue_failures();
2109   void send_failures();
2110   void send_still_alive(epoch_t epoch, const entity_inst_t &i);
2111
2112   // -- pg stats --
2113   Mutex pg_stat_queue_lock;
2114   Cond pg_stat_queue_cond;
2115   xlist<PG*> pg_stat_queue;
2116   bool osd_stat_updated;
2117   uint64_t pg_stat_tid, pg_stat_tid_flushed;
2118
2119   void send_pg_stats(const utime_t &now);
2120   void handle_pg_stats_ack(class MPGStatsAck *ack);
2121   void flush_pg_stats();
2122
2123   ceph::coarse_mono_clock::time_point last_sent_beacon;
2124   Mutex min_last_epoch_clean_lock{"OSD::min_last_epoch_clean_lock"};
2125   epoch_t min_last_epoch_clean = 0;
2126   // which pgs were scanned for min_lec
2127   std::vector<pg_t> min_last_epoch_clean_pgs;
2128   void send_beacon(const ceph::coarse_mono_clock::time_point& now);
2129
2130   void pg_stat_queue_enqueue(PG *pg) {
2131     pg_stat_queue_lock.Lock();
2132     if (pg->is_primary() && !pg->stat_queue_item.is_on_list()) {
2133       pg->get("pg_stat_queue");
2134       pg_stat_queue.push_back(&pg->stat_queue_item);
2135     }
2136     osd_stat_updated = true;
2137     pg_stat_queue_lock.Unlock();
2138   }
2139   void pg_stat_queue_dequeue(PG *pg) {
2140     pg_stat_queue_lock.Lock();
2141     if (pg->stat_queue_item.remove_myself())
2142       pg->put("pg_stat_queue");
2143     pg_stat_queue_lock.Unlock();
2144   }
2145   void clear_pg_stat_queue() {
2146     pg_stat_queue_lock.Lock();
2147     while (!pg_stat_queue.empty()) {
2148       PG *pg = pg_stat_queue.front();
2149       pg_stat_queue.pop_front();
2150       pg->put("pg_stat_queue");
2151     }
2152     pg_stat_queue_lock.Unlock();
2153   }
2154   void clear_outstanding_pg_stats(){
2155     Mutex::Locker l(pg_stat_queue_lock);
2156     outstanding_pg_stats.clear();
2157   }
2158
2159   ceph_tid_t get_tid() {
2160     return service.get_tid();
2161   }
2162
2163   // -- generic pg peering --
2164   PG::RecoveryCtx create_context();
2165   void dispatch_context(PG::RecoveryCtx &ctx, PG *pg, OSDMapRef curmap,
2166                         ThreadPool::TPHandle *handle = NULL);
2167   void dispatch_context_transaction(PG::RecoveryCtx &ctx, PG *pg,
2168                                     ThreadPool::TPHandle *handle = NULL);
2169   void do_notifies(map<int,
2170                        vector<pair<pg_notify_t, PastIntervals> > >&
2171                        notify_list,
2172                    OSDMapRef map);
2173   void do_queries(map<int, map<spg_t,pg_query_t> >& query_map,
2174                   OSDMapRef map);
2175   void do_infos(map<int,
2176                     vector<pair<pg_notify_t, PastIntervals> > >& info_map,
2177                 OSDMapRef map);
2178
2179   bool require_mon_peer(const Message *m);
2180   bool require_mon_or_mgr_peer(const Message *m);
2181   bool require_osd_peer(const Message *m);
2182   /***
2183    * Verifies that we were alive in the given epoch, and that
2184    * still are.
2185    */
2186   bool require_self_aliveness(const Message *m, epoch_t alive_since);
2187   /**
2188    * Verifies that the OSD who sent the given op has the same
2189    * address as in the given map.
2190    * @pre op was sent by an OSD using the cluster messenger
2191    */
2192   bool require_same_peer_instance(const Message *m, OSDMapRef& map,
2193                                   bool is_fast_dispatch);
2194
2195   bool require_same_or_newer_map(OpRequestRef& op, epoch_t e,
2196                                  bool is_fast_dispatch);
2197
2198   void handle_pg_query(OpRequestRef op);
2199   void handle_pg_notify(OpRequestRef op);
2200   void handle_pg_log(OpRequestRef op);
2201   void handle_pg_info(OpRequestRef op);
2202   void handle_pg_trim(OpRequestRef op);
2203
2204   void handle_pg_backfill_reserve(OpRequestRef op);
2205   void handle_pg_recovery_reserve(OpRequestRef op);
2206
2207   void handle_force_recovery(Message *m);
2208
2209   void handle_pg_remove(OpRequestRef op);
2210   void _remove_pg(PG *pg);
2211
2212   // -- commands --
2213   struct Command {
2214     vector<string> cmd;
2215     ceph_tid_t tid;
2216     bufferlist indata;
2217     ConnectionRef con;
2218
2219     Command(vector<string>& c, ceph_tid_t t, bufferlist& bl, Connection *co)
2220       : cmd(c), tid(t), indata(bl), con(co) {}
2221   };
2222   list<Command*> command_queue;
2223   struct CommandWQ : public ThreadPool::WorkQueue<Command> {
2224     OSD *osd;
2225     CommandWQ(OSD *o, time_t ti, time_t si, ThreadPool *tp)
2226       : ThreadPool::WorkQueue<Command>("OSD::CommandWQ", ti, si, tp), osd(o) {}
2227
2228     bool _empty() override {
2229       return osd->command_queue.empty();
2230     }
2231     bool _enqueue(Command *c) override {
2232       osd->command_queue.push_back(c);
2233       return true;
2234     }
2235     void _dequeue(Command *pg) override {
2236       ceph_abort();
2237     }
2238     Command *_dequeue() override {
2239       if (osd->command_queue.empty())
2240         return NULL;
2241       Command *c = osd->command_queue.front();
2242       osd->command_queue.pop_front();
2243       return c;
2244     }
2245     void _process(Command *c, ThreadPool::TPHandle &) override {
2246       osd->osd_lock.Lock();
2247       if (osd->is_stopping()) {
2248         osd->osd_lock.Unlock();
2249         delete c;
2250         return;
2251       }
2252       osd->do_command(c->con.get(), c->tid, c->cmd, c->indata);
2253       osd->osd_lock.Unlock();
2254       delete c;
2255     }
2256     void _clear() override {
2257       while (!osd->command_queue.empty()) {
2258         Command *c = osd->command_queue.front();
2259         osd->command_queue.pop_front();
2260         delete c;
2261       }
2262     }
2263   } command_wq;
2264
2265   void handle_command(class MMonCommand *m);
2266   void handle_command(class MCommand *m);
2267   void do_command(Connection *con, ceph_tid_t tid, vector<string>& cmd, bufferlist& data);
2268
2269   // -- pg recovery --
2270   void do_recovery(PG *pg, epoch_t epoch_queued, uint64_t pushes_reserved,
2271                    ThreadPool::TPHandle &handle);
2272
2273
2274   // -- scrubbing --
2275   void sched_scrub();
2276   bool scrub_random_backoff();
2277   bool scrub_load_below_threshold();
2278   bool scrub_time_permit(utime_t now);
2279
2280   // -- removing --
2281   struct RemoveWQ :
2282     public ThreadPool::WorkQueueVal<pair<PGRef, DeletingStateRef> > {
2283     CephContext* cct;
2284     ObjectStore *&store;
2285     list<pair<PGRef, DeletingStateRef> > remove_queue;
2286     RemoveWQ(CephContext* cct, ObjectStore *&o, time_t ti, time_t si,
2287              ThreadPool *tp)
2288       : ThreadPool::WorkQueueVal<pair<PGRef, DeletingStateRef> >(
2289         "OSD::RemoveWQ", ti, si, tp), cct(cct), store(o) {}
2290
2291     bool _empty() override {
2292       return remove_queue.empty();
2293     }
2294     void _enqueue(pair<PGRef, DeletingStateRef> item) override {
2295       remove_queue.push_back(item);
2296     }
2297     void _enqueue_front(pair<PGRef, DeletingStateRef> item) override {
2298       remove_queue.push_front(item);
2299     }
2300     bool _dequeue(pair<PGRef, DeletingStateRef> item) {
2301       ceph_abort();
2302     }
2303     pair<PGRef, DeletingStateRef> _dequeue() override {
2304       assert(!remove_queue.empty());
2305       pair<PGRef, DeletingStateRef> item = remove_queue.front();
2306       remove_queue.pop_front();
2307       return item;
2308     }
2309     void _process(pair<PGRef, DeletingStateRef>,
2310                   ThreadPool::TPHandle &) override;
2311     void _clear() override {
2312       remove_queue.clear();
2313     }
2314     int get_remove_queue_len() {
2315       lock();
2316       int r = remove_queue.size();
2317       unlock();
2318       return r;
2319     }
2320   } remove_wq;
2321
2322   // -- status reporting --
2323   MPGStats *collect_pg_stats();
2324   std::vector<OSDHealthMetric> get_health_metrics();
2325
2326 private:
2327   bool ms_can_fast_dispatch_any() const override { return true; }
2328   bool ms_can_fast_dispatch(const Message *m) const override {
2329     switch (m->get_type()) {
2330     case CEPH_MSG_OSD_OP:
2331     case CEPH_MSG_OSD_BACKOFF:
2332     case MSG_OSD_SUBOP:
2333     case MSG_OSD_REPOP:
2334     case MSG_OSD_SUBOPREPLY:
2335     case MSG_OSD_REPOPREPLY:
2336     case MSG_OSD_PG_PUSH:
2337     case MSG_OSD_PG_PULL:
2338     case MSG_OSD_PG_PUSH_REPLY:
2339     case MSG_OSD_PG_SCAN:
2340     case MSG_OSD_PG_BACKFILL:
2341     case MSG_OSD_PG_BACKFILL_REMOVE:
2342     case MSG_OSD_EC_WRITE:
2343     case MSG_OSD_EC_WRITE_REPLY:
2344     case MSG_OSD_EC_READ:
2345     case MSG_OSD_EC_READ_REPLY:
2346     case MSG_OSD_SCRUB_RESERVE:
2347     case MSG_OSD_REP_SCRUB:
2348     case MSG_OSD_REP_SCRUBMAP:
2349     case MSG_OSD_PG_UPDATE_LOG_MISSING:
2350     case MSG_OSD_PG_UPDATE_LOG_MISSING_REPLY:
2351     case MSG_OSD_PG_RECOVERY_DELETE:
2352     case MSG_OSD_PG_RECOVERY_DELETE_REPLY:
2353       return true;
2354     default:
2355       return false;
2356     }
2357   }
2358   void ms_fast_dispatch(Message *m) override;
2359   void ms_fast_preprocess(Message *m) override;
2360   bool ms_dispatch(Message *m) override;
2361   bool ms_get_authorizer(int dest_type, AuthAuthorizer **authorizer, bool force_new) override;
2362   bool ms_verify_authorizer(Connection *con, int peer_type,
2363                             int protocol, bufferlist& authorizer, bufferlist& authorizer_reply,
2364                             bool& isvalid, CryptoKey& session_key,
2365                             std::unique_ptr<AuthAuthorizerChallenge> *challenge) override;
2366   void ms_handle_connect(Connection *con) override;
2367   void ms_handle_fast_connect(Connection *con) override;
2368   void ms_handle_fast_accept(Connection *con) override;
2369   bool ms_handle_reset(Connection *con) override;
2370   void ms_handle_remote_reset(Connection *con) override {}
2371   bool ms_handle_refused(Connection *con) override;
2372
2373   io_queue get_io_queue() const {
2374     if (cct->_conf->osd_op_queue == "debug_random") {
2375       static io_queue index_lookup[] = { io_queue::prioritized,
2376                                          io_queue::weightedpriority,
2377                                          io_queue::mclock_opclass,
2378                                          io_queue::mclock_client };
2379       srand(time(NULL));
2380       unsigned which = rand() % (sizeof(index_lookup) / sizeof(index_lookup[0]));
2381       return index_lookup[which];
2382     } else if (cct->_conf->osd_op_queue == "prioritized") {
2383       return io_queue::prioritized;
2384     } else if (cct->_conf->osd_op_queue == "mclock_opclass") {
2385       return io_queue::mclock_opclass;
2386     } else if (cct->_conf->osd_op_queue == "mclock_client") {
2387       return io_queue::mclock_client;
2388     } else {
2389       // default / catch-all is 'wpq'
2390       return io_queue::weightedpriority;
2391     }
2392   }
2393
2394   unsigned int get_io_prio_cut() const {
2395     if (cct->_conf->osd_op_queue_cut_off == "debug_random") {
2396       srand(time(NULL));
2397       return (rand() % 2 < 1) ? CEPH_MSG_PRIO_HIGH : CEPH_MSG_PRIO_LOW;
2398     } else if (cct->_conf->osd_op_queue_cut_off == "high") {
2399       return CEPH_MSG_PRIO_HIGH;
2400     } else {
2401       // default / catch-all is 'low'
2402       return CEPH_MSG_PRIO_LOW;
2403     }
2404   }
2405
2406  public:
2407   /* internal and external can point to the same messenger, they will still
2408    * be cleaned up properly*/
2409   OSD(CephContext *cct_,
2410       ObjectStore *store_,
2411       int id,
2412       Messenger *internal,
2413       Messenger *external,
2414       Messenger *hb_front_client,
2415       Messenger *hb_back_client,
2416       Messenger *hb_front_server,
2417       Messenger *hb_back_server,
2418       Messenger *osdc_messenger,
2419       MonClient *mc, const std::string &dev, const std::string &jdev);
2420   ~OSD() override;
2421
2422   // static bits
2423   static int mkfs(CephContext *cct, ObjectStore *store,
2424                   const string& dev,
2425                   uuid_d fsid, int whoami);
2426   /* remove any non-user xattrs from a map of them */
2427   void filter_xattrs(map<string, bufferptr>& attrs) {
2428     for (map<string, bufferptr>::iterator iter = attrs.begin();
2429          iter != attrs.end();
2430          ) {
2431       if (('_' != iter->first.at(0)) || (iter->first.size() == 1))
2432         attrs.erase(iter++);
2433       else ++iter;
2434     }
2435   }
2436
2437 private:
2438   int mon_cmd_maybe_osd_create(string &cmd);
2439   int update_crush_device_class();
2440   int update_crush_location();
2441
2442   static int write_meta(CephContext *cct,
2443                         ObjectStore *store,
2444                         uuid_d& cluster_fsid, uuid_d& osd_fsid, int whoami);
2445
2446   void handle_pg_scrub(struct MOSDScrub *m, PG* pg);
2447   void handle_scrub(struct MOSDScrub *m);
2448   void handle_osd_ping(class MOSDPing *m);
2449
2450   int init_op_flags(OpRequestRef& op);
2451
2452   int get_num_op_shards();
2453   int get_num_op_threads();
2454
2455   float get_osd_recovery_sleep();
2456
2457 public:
2458   static int peek_meta(ObjectStore *store, string& magic,
2459                        uuid_d& cluster_fsid, uuid_d& osd_fsid, int& whoami);
2460
2461
2462   // startup/shutdown
2463   int pre_init();
2464   int init();
2465   void final_init();
2466
2467   int enable_disable_fuse(bool stop);
2468
2469   void suicide(int exitcode);
2470   int shutdown();
2471
2472   void handle_signal(int signum);
2473
2474   /// check if we can throw out op from a disconnected client
2475   static bool op_is_discardable(const MOSDOp *m);
2476
2477 public:
2478   OSDService service;
2479   friend class OSDService;
2480 };
2481
2482
2483 std::ostream& operator<<(std::ostream& out, const OSD::io_queue& q);
2484
2485
2486 //compatibility of the executable
2487 extern const CompatSet::Feature ceph_osd_feature_compat[];
2488 extern const CompatSet::Feature ceph_osd_feature_ro_compat[];
2489 extern const CompatSet::Feature ceph_osd_feature_incompat[];
2490
2491 #endif // CEPH_OSD_H