ceph/src/crimson/osd/osdmap_gate.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 "crimson/common/exception.h"
   5 #include "crimson/osd/osdmap_gate.h"
   6 #include "crimson/osd/shard_services.h"
   7 #include "common/Formatter.h"
   8
   9 namespace {
  10   seastar::logger& logger() {
  11     return crimson::get_logger(ceph_subsys_osd);
  12   }
  13 }
  14
  15 namespace crimson::osd {
  16
  17 void OSDMapGate::OSDMapBlocker::dump_detail(Formatter *f) const
  18 {
  19   f->open_object_section("OSDMapGate");
  20   f->dump_int("epoch", epoch);
  21   f->close_section();
  22 }
  23
  24 blocking_future<epoch_t> OSDMapGate::wait_for_map(epoch_t epoch)
  25 {
  26   if (__builtin_expect(stopping, false)) {
  27     return make_exception_blocking_future<epoch_t>(
  28         crimson::common::system_shutdown_exception());
  29   }
  30   if (current >= epoch) {
  31     return make_ready_blocking_future<epoch_t>(current);
  32   } else {
  33     logger().info("evt epoch is {}, i have {}, will wait", epoch, current);
  34     auto &blocker = waiting_peering.emplace(
  35       epoch, make_pair(blocker_type, epoch)).first->second;
  36     auto fut = blocker.promise.get_shared_future();
  37     if (shard_services) {
  38       return blocker.make_blocking_future(
  39         (*shard_services).get().osdmap_subscribe(current, true).then(
  40           [fut=std::move(fut)]() mutable {
  41             return std::move(fut);
  42           }));
  43     } else {
  44       return blocker.make_blocking_future(std::move(fut));
  45     }
  46   }
  47 }
  48
  49 void OSDMapGate::got_map(epoch_t epoch) {
  50   current = epoch;
  51   auto first = waiting_peering.begin();
  52   auto last = waiting_peering.upper_bound(epoch);
  53   std::for_each(first, last, [epoch](auto& blocked_requests) {
  54     blocked_requests.second.promise.set_value(epoch);
  55   });
  56   waiting_peering.erase(first, last);
  57 }
  58
  59 seastar::future<> OSDMapGate::stop() {
  60   logger().info("osdmap::stop");
  61   stopping = true;
  62   auto first = waiting_peering.begin();
  63   auto last = waiting_peering.end();
  64   std::for_each(first, last, [](auto& blocked_requests) {
  65     blocked_requests.second.promise.set_exception(
  66         crimson::common::system_shutdown_exception());
  67   });
  68   return seastar::now();
  69 }
  70
  71 }