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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
|
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
#pragma once
#include <boost/intrusive_ptr.hpp>
#include <seastar/core/future.hh>
#include "include/common_fwd.h"
#include "osd_operation.h"
#include "msg/MessageRef.h"
#include "crimson/common/exception.h"
#include "crimson/os/futurized_collection.h"
#include "osd/PeeringState.h"
#include "crimson/osd/osdmap_service.h"
#include "crimson/osd/object_context.h"
#include "common/AsyncReserver.h"
namespace crimson::net {
class Messenger;
}
namespace crimson::mgr {
class Client;
}
namespace crimson::mon {
class Client;
}
namespace crimson::os {
class FuturizedStore;
}
class OSDMap;
class PeeringCtx;
class BufferedRecoveryMessages;
namespace crimson::osd {
/**
* Represents services available to each PG
*/
class ShardServices : public md_config_obs_t {
using cached_map_t = boost::local_shared_ptr<const OSDMap>;
OSDMapService &osdmap_service;
const int whoami;
crimson::net::Messenger &cluster_msgr;
crimson::net::Messenger &public_msgr;
crimson::mon::Client &monc;
crimson::mgr::Client &mgrc;
crimson::os::FuturizedStore &store;
crimson::common::CephContext cct;
PerfCounters *perf = nullptr;
PerfCounters *recoverystate_perf = nullptr;
const char** get_tracked_conf_keys() const final;
void handle_conf_change(const ConfigProxy& conf,
const std::set <std::string> &changed) final;
public:
ShardServices(
OSDMapService &osdmap_service,
const int whoami,
crimson::net::Messenger &cluster_msgr,
crimson::net::Messenger &public_msgr,
crimson::mon::Client &monc,
crimson::mgr::Client &mgrc,
crimson::os::FuturizedStore &store);
seastar::future<> send_to_osd(
int peer,
MessageRef m,
epoch_t from_epoch);
crimson::os::FuturizedStore &get_store() {
return store;
}
crimson::common::CephContext *get_cct() {
return &cct;
}
// OSDMapService
const OSDMapService &get_osdmap_service() const {
return osdmap_service;
}
// Op Management
OperationRegistry registry;
OperationThrottler throttler;
template <typename T, typename... Args>
auto start_operation(Args&&... args) {
if (__builtin_expect(stopping, false)) {
throw crimson::common::system_shutdown_exception();
}
auto op = registry.create_operation<T>(std::forward<Args>(args)...);
return std::make_pair(op, op->start());
}
seastar::future<> stop() {
stopping = true;
return registry.stop();
}
// Loggers
PerfCounters &get_recoverystate_perf_logger() {
return *recoverystate_perf;
}
PerfCounters &get_perf_logger() {
return *perf;
}
/// Dispatch and reset ctx transaction
seastar::future<> dispatch_context_transaction(
crimson::os::CollectionRef col, PeeringCtx &ctx);
/// Dispatch and reset ctx messages
seastar::future<> dispatch_context_messages(
BufferedRecoveryMessages &&ctx);
/// Dispatch ctx and dispose of context
seastar::future<> dispatch_context(
crimson::os::CollectionRef col,
PeeringCtx &&ctx);
/// Dispatch ctx and dispose of ctx, transaction must be empty
seastar::future<> dispatch_context(
PeeringCtx &&ctx) {
return dispatch_context({}, std::move(ctx));
}
// PG Temp State
private:
// TODO: hook into map processing and some kind of heartbeat/peering
// message processing
struct pg_temp_t {
std::vector<int> acting;
bool forced = false;
};
map<pg_t, pg_temp_t> pg_temp_wanted;
map<pg_t, pg_temp_t> pg_temp_pending;
friend std::ostream& operator<<(std::ostream&, const pg_temp_t&);
public:
void queue_want_pg_temp(pg_t pgid, const vector<int>& want,
bool forced = false);
void remove_want_pg_temp(pg_t pgid);
void requeue_pg_temp();
seastar::future<> send_pg_temp();
// Shard-local OSDMap
private:
cached_map_t osdmap;
public:
void update_map(cached_map_t new_osdmap);
cached_map_t &get_osdmap();
// PG Created State
private:
set<pg_t> pg_created;
public:
seastar::future<> send_pg_created(pg_t pgid);
seastar::future<> send_pg_created();
void prune_pg_created();
unsigned get_pg_num() const {
return num_pgs;
}
void inc_pg_num() {
++num_pgs;
}
void dec_pg_num() {
--num_pgs;
}
seastar::future<> osdmap_subscribe(version_t epoch, bool force_request);
// Time state
ceph::mono_time startup_time = ceph::mono_clock::now();
ceph::signedspan get_mnow() const {
return ceph::mono_clock::now() - startup_time;
}
HeartbeatStampsRef get_hb_stamps(int peer);
std::map<int, HeartbeatStampsRef> heartbeat_stamps;
crimson::osd::ObjectContextRegistry obc_registry;
// Async Reservers
private:
unsigned num_pgs = 0;
struct DirectFinisher {
void queue(Context *c) {
c->complete(0);
}
} finisher;
// prevent creating new osd operations when system is shutting down,
// this is necessary because there are chances that a new operation
// is created, after the interruption of all ongoing operations, and
// creats and waits on a new and may-never-resolve future, in which
// case the shutdown may never succeed.
bool stopping = false;
public:
AsyncReserver<spg_t, DirectFinisher> local_reserver;
AsyncReserver<spg_t, DirectFinisher> remote_reserver;
private:
epoch_t up_thru_wanted = 0;
public:
seastar::future<> send_alive(epoch_t want);
};
}
|
