ceph/src/crimson/osd/osd_operations/pg_advance_map.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include <seastar/core/future.hh>
   5
   6 #include "include/types.h"
   7 #include "common/Formatter.h"
   8 #include "crimson/osd/pg.h"
   9 #include "crimson/osd/osdmap_service.h"
  10 #include "crimson/osd/shard_services.h"
  11 #include "crimson/osd/osd_operations/pg_advance_map.h"
  12 #include "crimson/osd/osd_operation_external_tracking.h"
  13 #include "osd/PeeringState.h"
  14
  15 namespace {
  16   seastar::logger& logger() {
  17     return crimson::get_logger(ceph_subsys_osd);
  18   }
  19 }
  20
  21 namespace crimson::osd {
  22
  23 PGAdvanceMap::PGAdvanceMap(
  24   ShardServices &shard_services, Ref<PG> pg, epoch_t to,
  25   PeeringCtx &&rctx, bool do_init)
  26   : shard_services(shard_services), pg(pg), to(to),
  27     rctx(std::move(rctx)), do_init(do_init) {}
  28
  29 PGAdvanceMap::~PGAdvanceMap() {}
  30
  31 void PGAdvanceMap::print(std::ostream &lhs) const
  32 {
  33   lhs << "PGAdvanceMap("
  34       << "pg=" << pg->get_pgid()
  35       << " from=" << (from ? *from : -1)
  36       << " to=" << to;
  37   if (do_init) {
  38     lhs << " do_init";
  39   }
  40   lhs << ")";
  41 }
  42
  43 void PGAdvanceMap::dump_detail(Formatter *f) const
  44 {
  45   f->open_object_section("PGAdvanceMap");
  46   f->dump_stream("pgid") << pg->get_pgid();
  47   if (from) {
  48     f->dump_int("from", *from);
  49   }
  50   f->dump_int("to", to);
  51   f->dump_bool("do_init", do_init);
  52   f->close_section();
  53 }
  54
  55 seastar::future<> PGAdvanceMap::start()
  56 {
  57   using cached_map_t = OSDMapService::cached_map_t;
  58
  59   logger().debug("{}: start", *this);
  60
  61   IRef ref = this;
  62   return enter_stage<>(
  63     pg->peering_request_pg_pipeline.process
  64   ).then([this] {
  65     from = pg->get_osdmap_epoch();
  66     auto fut = seastar::now();
  67     if (do_init) {
  68       fut = pg->handle_initialize(rctx
  69       ).then([this] {
  70         return pg->handle_activate_map(rctx);
  71       });
  72     }
  73     return fut.then([this] {
  74       return seastar::do_for_each(
  75         boost::make_counting_iterator(*from + 1),
  76         boost::make_counting_iterator(to + 1),
  77         [this](epoch_t next_epoch) {
  78           return shard_services.get_map(next_epoch).then(
  79             [this] (cached_map_t&& next_map) {
  80               logger().debug("{}: advancing map to {}",
  81                              *this, next_map->get_epoch());
  82               return pg->handle_advance_map(next_map, rctx);
  83             });
  84         }).then([this] {
  85           return pg->handle_activate_map(rctx).then([this] {
  86             logger().debug("{}: map activated", *this);
  87             if (do_init) {
  88               shard_services.pg_created(pg->get_pgid(), pg);
  89               logger().info("PGAdvanceMap::start new pg {}", *pg);
  90             }
  91             return seastar::when_all_succeed(
  92               pg->get_need_up_thru()
  93               ? shard_services.send_alive(
  94                 pg->get_same_interval_since())
  95               : seastar::now(),
  96               shard_services.dispatch_context(
  97                 pg->get_collection_ref(),
  98                 std::move(rctx)));
  99           });
 100         }).then_unpack([this] {
 101           logger().debug("{}: sending pg temp", *this);
 102           return shard_services.send_pg_temp();
 103         });
 104     });
 105   }).then([this, ref=std::move(ref)] {
 106     logger().debug("{}: complete", *this);
 107   });
 108 }
 109
 110 }