ceph/src/crimson/osd/osd_operations/replicated_request.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 "replicated_request.h"
   5
   6 #include "common/Formatter.h"
   7
   8 #include "crimson/osd/osd.h"
   9 #include "crimson/osd/osd_connection_priv.h"
  10 #include "crimson/osd/osd_operation_external_tracking.h"
  11 #include "crimson/osd/pg.h"
  12
  13 namespace {
  14   seastar::logger& logger() {
  15     return crimson::get_logger(ceph_subsys_osd);
  16   }
  17 }
  18
  19 namespace crimson::osd {
  20
  21 RepRequest::RepRequest(crimson::net::ConnectionRef&& conn,
  22                        Ref<MOSDRepOp> &&req)
  23   : conn{std::move(conn)},
  24     req{std::move(req)}
  25 {}
  26
  27 void RepRequest::print(std::ostream& os) const
  28 {
  29   os << "RepRequest("
  30      << "from=" << req->from
  31      << " req=" << *req
  32      << ")";
  33 }
  34
  35 void RepRequest::dump_detail(Formatter *f) const
  36 {
  37   f->open_object_section("RepRequest");
  38   f->dump_stream("reqid") << req->reqid;
  39   f->dump_stream("pgid") << req->get_spg();
  40   f->dump_unsigned("map_epoch", req->get_map_epoch());
  41   f->dump_unsigned("min_epoch", req->get_min_epoch());
  42   f->dump_stream("oid") << req->poid;
  43   f->dump_stream("from") << req->from;
  44   f->close_section();
  45 }
  46
  47 ConnectionPipeline &RepRequest::get_connection_pipeline()
  48 {
  49   return get_osd_priv(conn.get()).replicated_request_conn_pipeline;
  50 }
  51
  52 ClientRequest::PGPipeline &RepRequest::client_pp(PG &pg)
  53 {
  54   return pg.request_pg_pipeline;
  55 }
  56
  57 seastar::future<> RepRequest::with_pg(
  58   ShardServices &shard_services, Ref<PG> pg)
  59 {
  60   logger().debug("{}: RepRequest::with_pg", *this);
  61   IRef ref = this;
  62   return interruptor::with_interruption([this, pg] {
  63     logger().debug("{}: pg present", *this);
  64     return this->template enter_stage<interruptor>(client_pp(*pg).await_map
  65     ).then_interruptible([this, pg] {
  66       return this->template with_blocking_event<
  67         PG_OSDMapGate::OSDMapBlocker::BlockingEvent
  68       >([this, pg](auto &&trigger) {
  69         return pg->osdmap_gate.wait_for_map(
  70           std::move(trigger), req->min_epoch);
  71       });
  72     }).then_interruptible([this, pg] (auto) {
  73       return pg->handle_rep_op(req);
  74     });
  75   }, [ref](std::exception_ptr) {
  76     return seastar::now();
  77   }, pg);
  78 }
  79
  80 }