1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
#include "crimson/common/exception.h"
#include "crimson/osd/osdmap_gate.h"
#include "crimson/osd/shard_services.h"
#include "common/Formatter.h"
namespace {
seastar::logger& logger() {
return crimson::get_logger(ceph_subsys_osd);
}
}
namespace crimson::osd {
template <OSDMapGateType OSDMapGateTypeV>
void OSDMapGate<OSDMapGateTypeV>::OSDMapBlocker::dump_detail(Formatter *f) const
{
f->open_object_section("OSDMapGate");
f->dump_int("epoch", epoch);
f->close_section();
}
template <OSDMapGateType OSDMapGateTypeV>
seastar::future<epoch_t> OSDMapGate<OSDMapGateTypeV>::wait_for_map(
typename OSDMapBlocker::BlockingEvent::TriggerI&& trigger,
epoch_t epoch,
ShardServices *shard_services)
{
if (__builtin_expect(stopping, false)) {
return seastar::make_exception_future<epoch_t>(
crimson::common::system_shutdown_exception());
}
if (current >= epoch) {
return seastar::make_ready_future<epoch_t>(current);
} else {
logger().info("evt epoch is {}, i have {}, will wait", epoch, current);
auto &blocker = waiting_peering.emplace(
epoch, std::make_pair(blocker_type, epoch)).first->second;
auto fut = blocker.promise.get_shared_future();
if (shard_services) {
return trigger.maybe_record_blocking(
shard_services->osdmap_subscribe(current, true).then(
[fut=std::move(fut)]() mutable {
return std::move(fut);
}),
blocker);
} else {
return trigger.maybe_record_blocking(std::move(fut), blocker);
}
}
}
template <OSDMapGateType OSDMapGateTypeV>
void OSDMapGate<OSDMapGateTypeV>::got_map(epoch_t epoch) {
if (epoch == 0) {
return;
}
ceph_assert(epoch > current);
current = epoch;
auto first = waiting_peering.begin();
auto last = waiting_peering.upper_bound(epoch);
std::for_each(first, last, [epoch](auto& blocked_requests) {
blocked_requests.second.promise.set_value(epoch);
});
waiting_peering.erase(first, last);
}
template <OSDMapGateType OSDMapGateTypeV>
seastar::future<> OSDMapGate<OSDMapGateTypeV>::stop() {
logger().info("osdmap::stop");
stopping = true;
auto first = waiting_peering.begin();
auto last = waiting_peering.end();
std::for_each(first, last, [](auto& blocked_requests) {
blocked_requests.second.promise.set_exception(
crimson::common::system_shutdown_exception());
});
return seastar::now();
}
template class OSDMapGate<OSDMapGateType::PG>;
template class OSDMapGate<OSDMapGateType::OSD>;
} // namespace crimson::osd
|
