]>
Commit | Line | Data |
---|---|---|
f67539c2 TL |
1 | // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- |
2 | // vim: ts=8 sw=2 smarttab | |
3 | /* | |
4 | * Ceph - scalable distributed file system | |
5 | * | |
6 | * Copyright (C) 2020 Red Hat <contact@redhat.com> | |
7 | * Author: Adam C. Emerson | |
8 | * | |
9 | * This is free software; you can redistribute it and/or | |
10 | * modify it under the terms of the GNU Lesser General Public | |
11 | * License version 2.1, as published by the Free Software | |
12 | * Foundation. See file COPYING. | |
13 | * | |
14 | */ | |
15 | ||
16 | #ifndef CEPH_RGW_CLS_FIFO_LEGACY_H | |
17 | #define CEPH_RGW_CLS_FIFO_LEGACY_H | |
18 | ||
19 | #include <cstdint> | |
20 | #include <deque> | |
21 | #include <map> | |
22 | #include <memory> | |
23 | #include <mutex> | |
24 | #include <optional> | |
25 | #include <string_view> | |
26 | #include <vector> | |
27 | ||
28 | #undef FMT_HEADER_ONLY | |
29 | #define FMT_HEADER_ONLY 1 | |
30 | #include <fmt/format.h> | |
31 | ||
32 | #include "include/rados/librados.hpp" | |
33 | #include "include/buffer.h" | |
34 | #include "include/function2.hpp" | |
35 | ||
36 | #include "common/async/yield_context.h" | |
37 | ||
38 | #include "cls/fifo/cls_fifo_types.h" | |
39 | #include "cls/fifo/cls_fifo_ops.h" | |
40 | ||
41 | #include "librados/AioCompletionImpl.h" | |
42 | ||
43 | #include "rgw_tools.h" | |
44 | ||
45 | namespace rgw::cls::fifo { | |
46 | namespace cb = ceph::buffer; | |
47 | namespace fifo = rados::cls::fifo; | |
48 | namespace lr = librados; | |
49 | ||
50 | inline constexpr std::uint64_t default_max_part_size = 4 * 1024 * 1024; | |
51 | inline constexpr std::uint64_t default_max_entry_size = 32 * 1024; | |
52 | ||
53 | void create_meta(lr::ObjectWriteOperation* op, std::string_view id, | |
54 | std::optional<fifo::objv> objv, | |
55 | std::optional<std::string_view> oid_prefix, | |
56 | bool exclusive = false, | |
57 | std::uint64_t max_part_size = default_max_part_size, | |
58 | std::uint64_t max_entry_size = default_max_entry_size); | |
b3b6e05e | 59 | int get_meta(const DoutPrefixProvider *dpp, lr::IoCtx& ioctx, const std::string& oid, |
f67539c2 TL |
60 | std::optional<fifo::objv> objv, fifo::info* info, |
61 | std::uint32_t* part_header_size, | |
62 | std::uint32_t* part_entry_overhead, | |
63 | std::uint64_t tid, optional_yield y, | |
64 | bool probe = false); | |
f67539c2 TL |
65 | struct marker { |
66 | std::int64_t num = 0; | |
67 | std::uint64_t ofs = 0; | |
68 | ||
69 | marker() = default; | |
70 | marker(std::int64_t num, std::uint64_t ofs) : num(num), ofs(ofs) {} | |
71 | static marker max() { | |
72 | return { std::numeric_limits<decltype(num)>::max(), | |
73 | std::numeric_limits<decltype(ofs)>::max() }; | |
74 | } | |
75 | ||
76 | std::string to_string() { | |
77 | return fmt::format("{:0>20}:{:0>20}", num, ofs); | |
78 | } | |
79 | }; | |
80 | ||
81 | struct list_entry { | |
82 | cb::list data; | |
83 | std::string marker; | |
84 | ceph::real_time mtime; | |
85 | }; | |
86 | ||
87 | using part_info = fifo::part_header; | |
88 | ||
89 | /// This is an implementation of FIFO using librados to facilitate | |
90 | /// backports. Please see /src/neorados/cls/fifo.h for full | |
91 | /// information. | |
92 | /// | |
93 | /// This library uses optional_yield. Please see | |
94 | /// /src/common/async/yield_context.h. In summary, optional_yield | |
95 | /// contains either a spawn::yield_context (in which case the current | |
96 | /// coroutine is suspended until completion) or null_yield (in which | |
97 | /// case the current thread is blocked until completion.) | |
98 | /// | |
99 | /// Please see the librados documentation for information on | |
100 | /// AioCompletion and IoCtx. | |
101 | ||
102 | class FIFO { | |
103 | friend struct Reader; | |
104 | friend struct Updater; | |
105 | friend struct Trimmer; | |
106 | friend struct InfoGetter; | |
107 | friend struct Pusher; | |
108 | friend struct NewPartPreparer; | |
109 | friend struct NewHeadPreparer; | |
110 | friend struct JournalProcessor; | |
111 | friend struct Lister; | |
112 | ||
113 | mutable lr::IoCtx ioctx; | |
114 | CephContext* cct = static_cast<CephContext*>(ioctx.cct()); | |
115 | const std::string oid; | |
116 | std::mutex m; | |
117 | std::uint64_t next_tid = 0; | |
118 | ||
119 | fifo::info info; | |
120 | ||
121 | std::uint32_t part_header_size = 0xdeadbeef; | |
122 | std::uint32_t part_entry_overhead = 0xdeadbeef; | |
123 | ||
124 | std::optional<marker> to_marker(std::string_view s); | |
125 | ||
126 | FIFO(lr::IoCtx&& ioc, | |
127 | std::string oid) | |
128 | : ioctx(std::move(ioc)), oid(oid) {} | |
129 | ||
130 | std::string generate_tag() const; | |
131 | ||
20effc67 TL |
132 | int apply_update(const DoutPrefixProvider *dpp, |
133 | fifo::info* info, | |
f67539c2 TL |
134 | const fifo::objv& objv, |
135 | const fifo::update& update, | |
136 | std::uint64_t tid); | |
b3b6e05e | 137 | int _update_meta(const DoutPrefixProvider *dpp, const fifo::update& update, |
f67539c2 TL |
138 | fifo::objv version, bool* pcanceled, |
139 | std::uint64_t tid, optional_yield y); | |
b3b6e05e | 140 | void _update_meta(const DoutPrefixProvider *dpp, const fifo::update& update, |
f67539c2 TL |
141 | fifo::objv version, bool* pcanceled, |
142 | std::uint64_t tid, lr::AioCompletion* c); | |
b3b6e05e | 143 | int create_part(const DoutPrefixProvider *dpp, int64_t part_num, std::string_view tag, std::uint64_t tid, |
f67539c2 | 144 | optional_yield y); |
b3b6e05e | 145 | int remove_part(const DoutPrefixProvider *dpp, int64_t part_num, std::string_view tag, std::uint64_t tid, |
f67539c2 | 146 | optional_yield y); |
b3b6e05e TL |
147 | int process_journal(const DoutPrefixProvider *dpp, std::uint64_t tid, optional_yield y); |
148 | void process_journal(const DoutPrefixProvider *dpp, std::uint64_t tid, lr::AioCompletion* c); | |
149 | int _prepare_new_part(const DoutPrefixProvider *dpp, bool is_head, std::uint64_t tid, optional_yield y); | |
150 | void _prepare_new_part(const DoutPrefixProvider *dpp, bool is_head, std::uint64_t tid, lr::AioCompletion* c); | |
151 | int _prepare_new_head(const DoutPrefixProvider *dpp, std::uint64_t tid, optional_yield y); | |
152 | void _prepare_new_head(const DoutPrefixProvider *dpp, std::uint64_t tid, lr::AioCompletion* c); | |
153 | int push_entries(const DoutPrefixProvider *dpp, const std::deque<cb::list>& data_bufs, | |
f67539c2 TL |
154 | std::uint64_t tid, optional_yield y); |
155 | void push_entries(const std::deque<cb::list>& data_bufs, | |
156 | std::uint64_t tid, lr::AioCompletion* c); | |
b3b6e05e | 157 | int trim_part(const DoutPrefixProvider *dpp, int64_t part_num, uint64_t ofs, |
f67539c2 TL |
158 | std::optional<std::string_view> tag, bool exclusive, |
159 | std::uint64_t tid, optional_yield y); | |
20effc67 | 160 | void trim_part(const DoutPrefixProvider *dpp, int64_t part_num, uint64_t ofs, |
f67539c2 TL |
161 | std::optional<std::string_view> tag, bool exclusive, |
162 | std::uint64_t tid, lr::AioCompletion* c); | |
163 | ||
164 | /// Force refresh of metadata, yielding/blocking style | |
b3b6e05e | 165 | int read_meta(const DoutPrefixProvider *dpp, std::uint64_t tid, optional_yield y); |
f67539c2 | 166 | /// Force refresh of metadata, with a librados Completion |
b3b6e05e | 167 | void read_meta(const DoutPrefixProvider *dpp, std::uint64_t tid, lr::AioCompletion* c); |
f67539c2 TL |
168 | |
169 | public: | |
170 | ||
171 | FIFO(const FIFO&) = delete; | |
172 | FIFO& operator =(const FIFO&) = delete; | |
173 | FIFO(FIFO&&) = delete; | |
174 | FIFO& operator =(FIFO&&) = delete; | |
175 | ||
176 | /// Open an existing FIFO. | |
b3b6e05e | 177 | static int open(const DoutPrefixProvider *dpp, lr::IoCtx ioctx, //< IO Context |
f67539c2 TL |
178 | std::string oid, //< OID for metadata object |
179 | std::unique_ptr<FIFO>* fifo, //< OUT: Pointer to FIFO object | |
180 | optional_yield y, //< Optional yield context | |
181 | /// Operation will fail if FIFO is not at this version | |
182 | std::optional<fifo::objv> objv = std::nullopt, | |
183 | /// Probing for existence, don't print errors if we | |
184 | /// can't find it. | |
185 | bool probe = false); | |
186 | /// Create a new or open an existing FIFO. | |
b3b6e05e | 187 | static int create(const DoutPrefixProvider *dpp, lr::IoCtx ioctx, //< IO Context |
f67539c2 TL |
188 | std::string oid, //< OID for metadata object |
189 | std::unique_ptr<FIFO>* fifo, //< OUT: Pointer to FIFO object | |
190 | optional_yield y, //< Optional yield context | |
191 | /// Operation will fail if the FIFO exists and is | |
192 | /// not of this version. | |
193 | std::optional<fifo::objv> objv = std::nullopt, | |
194 | /// Prefix for all objects | |
195 | std::optional<std::string_view> oid_prefix = std::nullopt, | |
196 | /// Fail if the FIFO already exists | |
197 | bool exclusive = false, | |
198 | /// Maximum allowed size of parts | |
199 | std::uint64_t max_part_size = default_max_part_size, | |
200 | /// Maximum allowed size of entries | |
201 | std::uint64_t max_entry_size = default_max_entry_size); | |
202 | ||
203 | /// Force refresh of metadata, yielding/blocking style | |
b3b6e05e | 204 | int read_meta(const DoutPrefixProvider *dpp, optional_yield y); |
f67539c2 TL |
205 | /// Get currently known metadata |
206 | const fifo::info& meta() const; | |
207 | /// Get partition header and entry overhead size | |
208 | std::pair<std::uint32_t, std::uint32_t> get_part_layout_info() const; | |
209 | /// Push an entry to the FIFO | |
b3b6e05e TL |
210 | int push(const DoutPrefixProvider *dpp, |
211 | const cb::list& bl, //< Entry to push | |
f67539c2 TL |
212 | optional_yield y //< Optional yield |
213 | ); | |
214 | /// Push an entry to the FIFO | |
b3b6e05e | 215 | void push(const DoutPrefixProvider *dpp, const cb::list& bl, //< Entry to push |
f67539c2 TL |
216 | lr::AioCompletion* c //< Async Completion |
217 | ); | |
218 | /// Push entries to the FIFO | |
20effc67 TL |
219 | int push(const DoutPrefixProvider *dpp, |
220 | const std::vector<cb::list>& data_bufs, //< Entries to push | |
f67539c2 TL |
221 | optional_yield y //< Optional yield |
222 | ); | |
223 | /// Push entries to the FIFO | |
b3b6e05e | 224 | void push(const DoutPrefixProvider *dpp, const std::vector<cb::list>& data_bufs, //< Entries to push |
f67539c2 TL |
225 | lr::AioCompletion* c //< Async Completion |
226 | ); | |
227 | /// List entries | |
b3b6e05e TL |
228 | int list(const DoutPrefixProvider *dpp, |
229 | int max_entries, //< Maximum entries to list | |
f67539c2 TL |
230 | /// Point after which to begin listing. Start at tail if null |
231 | std::optional<std::string_view> markstr, | |
232 | std::vector<list_entry>* out, //< OUT: entries | |
233 | /// OUT: True if more entries in FIFO beyond the last returned | |
234 | bool* more, | |
235 | optional_yield y //< Optional yield | |
236 | ); | |
b3b6e05e TL |
237 | void list(const DoutPrefixProvider *dpp, |
238 | int max_entries, //< Maximum entries to list | |
f67539c2 TL |
239 | /// Point after which to begin listing. Start at tail if null |
240 | std::optional<std::string_view> markstr, | |
241 | std::vector<list_entry>* out, //< OUT: entries | |
242 | /// OUT: True if more entries in FIFO beyond the last returned | |
243 | bool* more, | |
244 | lr::AioCompletion* c //< Async Completion | |
245 | ); | |
246 | /// Trim entries, coroutine/block style | |
b3b6e05e TL |
247 | int trim(const DoutPrefixProvider *dpp, |
248 | std::string_view markstr, //< Position to which to trim, inclusive | |
f67539c2 TL |
249 | bool exclusive, //< If true, do not trim the target entry |
250 | //< itself, just all those before it. | |
251 | optional_yield y //< Optional yield | |
252 | ); | |
253 | /// Trim entries, librados AioCompletion style | |
b3b6e05e TL |
254 | void trim(const DoutPrefixProvider *dpp, |
255 | std::string_view markstr, //< Position to which to trim, inclusive | |
f67539c2 TL |
256 | bool exclusive, //< If true, do not trim the target entry |
257 | //< itself, just all those before it. | |
258 | lr::AioCompletion* c //< librados AIO Completion | |
259 | ); | |
260 | /// Get part info | |
b3b6e05e | 261 | int get_part_info(const DoutPrefixProvider *dpp, int64_t part_num, /// Part number |
f67539c2 TL |
262 | fifo::part_header* header, //< OUT: Information |
263 | optional_yield y //< Optional yield | |
264 | ); | |
265 | /// Get part info | |
266 | void get_part_info(int64_t part_num, //< Part number | |
267 | fifo::part_header* header, //< OUT: Information | |
268 | lr::AioCompletion* c //< AIO Completion | |
269 | ); | |
270 | /// A convenience method to fetch the part information for the FIFO | |
271 | /// head, using librados::AioCompletion, since | |
272 | /// libradio::AioCompletions compose lousily. | |
b3b6e05e | 273 | void get_head_info(const DoutPrefixProvider *dpp, fu2::unique_function< //< Function to receive info |
f67539c2 TL |
274 | void(int r, fifo::part_header&&)>, |
275 | lr::AioCompletion* c //< AIO Completion | |
276 | ); | |
277 | }; | |
278 | ||
279 | template<typename T> | |
280 | struct Completion { | |
281 | private: | |
b3b6e05e | 282 | const DoutPrefixProvider *_dpp; |
f67539c2 TL |
283 | lr::AioCompletion* _cur = nullptr; |
284 | lr::AioCompletion* _super; | |
285 | public: | |
286 | ||
287 | using Ptr = std::unique_ptr<T>; | |
288 | ||
289 | lr::AioCompletion* cur() const { | |
290 | return _cur; | |
291 | } | |
292 | lr::AioCompletion* super() const { | |
293 | return _super; | |
294 | } | |
295 | ||
b3b6e05e | 296 | Completion(const DoutPrefixProvider *dpp, lr::AioCompletion* super) : _dpp(dpp), _super(super) { |
f67539c2 TL |
297 | super->pc->get(); |
298 | } | |
299 | ||
300 | ~Completion() { | |
301 | if (_super) { | |
302 | _super->pc->put(); | |
303 | } | |
304 | if (_cur) | |
305 | _cur->release(); | |
306 | _super = nullptr; | |
307 | _cur = nullptr; | |
308 | } | |
309 | ||
310 | // The only times that aio_operate can return an error are: | |
311 | // 1. The completion contains a null pointer. This should just | |
312 | // crash, and in our case it does. | |
313 | // 2. An attempt is made to write to a snapshot. RGW doesn't use | |
314 | // snapshots, so we don't care. | |
315 | // | |
316 | // So we will just assert that initiating an Aio operation succeeds | |
317 | // and not worry about recovering. | |
318 | static lr::AioCompletion* call(Ptr&& p) { | |
319 | p->_cur = lr::Rados::aio_create_completion(static_cast<void*>(p.get()), | |
320 | &cb); | |
321 | auto c = p->_cur; | |
322 | p.release(); | |
323 | return c; | |
324 | } | |
325 | static void complete(Ptr&& p, int r) { | |
326 | auto c = p->_super; | |
327 | p->_super = nullptr; | |
328 | rgw_complete_aio_completion(c, r); | |
329 | } | |
330 | ||
331 | static void cb(lr::completion_t, void* arg) { | |
332 | auto t = static_cast<T*>(arg); | |
333 | auto r = t->_cur->get_return_value(); | |
334 | t->_cur->release(); | |
335 | t->_cur = nullptr; | |
b3b6e05e | 336 | t->handle(t->_dpp, Ptr(t), r); |
f67539c2 TL |
337 | } |
338 | }; | |
339 | ||
340 | } | |
341 | ||
342 | #endif // CEPH_RGW_CLS_FIFO_LEGACY_H |