]>
Commit | Line | Data |
---|---|---|
7c673cae FG |
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) 2012 New Dream Network/Sage Weil <sage@newdream.net> | |
7 | * | |
8 | * This is free software; you can redistribute it and/or | |
9 | * modify it under the terms of the GNU Lesser General Public | |
10 | * License version 2.1, as published by the Free Software | |
11 | * Foundation. See file COPYING. | |
12 | */ | |
13 | ||
14 | #ifndef TRACKEDREQUEST_H_ | |
15 | #define TRACKEDREQUEST_H_ | |
7c673cae | 16 | |
31f18b77 | 17 | #include <atomic> |
9f95a23c | 18 | #include "common/ceph_mutex.h" |
7c673cae | 19 | #include "common/histogram.h" |
11fdf7f2 TL |
20 | #include "common/Thread.h" |
21 | #include "common/Clock.h" | |
11fdf7f2 TL |
22 | #include "include/spinlock.h" |
23 | #include "msg/Message.h" | |
7c673cae FG |
24 | |
25 | #define OPTRACKER_PREALLOC_EVENTS 20 | |
26 | ||
27 | class TrackedOp; | |
11fdf7f2 TL |
28 | class OpHistory; |
29 | ||
7c673cae FG |
30 | typedef boost::intrusive_ptr<TrackedOp> TrackedOpRef; |
31 | ||
11fdf7f2 TL |
32 | class OpHistoryServiceThread : public Thread |
33 | { | |
34 | private: | |
9f95a23c | 35 | std::list<std::pair<utime_t, TrackedOpRef>> _external_queue; |
11fdf7f2 TL |
36 | OpHistory* _ophistory; |
37 | mutable ceph::spinlock queue_spinlock; | |
38 | bool _break_thread; | |
39 | public: | |
40 | explicit OpHistoryServiceThread(OpHistory* parent) | |
41 | : _ophistory(parent), | |
42 | _break_thread(false) { } | |
43 | ||
44 | void break_thread(); | |
45 | void insert_op(const utime_t& now, TrackedOpRef op) { | |
46 | queue_spinlock.lock(); | |
47 | _external_queue.emplace_back(now, op); | |
48 | queue_spinlock.unlock(); | |
49 | } | |
50 | ||
51 | void *entry() override; | |
52 | }; | |
53 | ||
1e59de90 TL |
54 | enum { |
55 | l_osd_slow_op_first = 1000, | |
56 | l_osd_slow_op_count, | |
57 | l_osd_slow_op_last, | |
58 | }; | |
11fdf7f2 | 59 | |
7c673cae | 60 | class OpHistory { |
1e59de90 | 61 | CephContext* cct = nullptr; |
9f95a23c TL |
62 | std::set<std::pair<utime_t, TrackedOpRef> > arrived; |
63 | std::set<std::pair<double, TrackedOpRef> > duration; | |
64 | std::set<std::pair<utime_t, TrackedOpRef> > slow_op; | |
11fdf7f2 | 65 | ceph::mutex ops_history_lock = ceph::make_mutex("OpHistory::ops_history_lock"); |
7c673cae | 66 | void cleanup(utime_t now); |
92f5a8d4 TL |
67 | std::atomic_size_t history_size{0}; |
68 | std::atomic_uint32_t history_duration{0}; | |
69 | std::atomic_size_t history_slow_op_size{0}; | |
70 | std::atomic_uint32_t history_slow_op_threshold{0}; | |
71 | std::atomic_bool shutdown{false}; | |
11fdf7f2 TL |
72 | OpHistoryServiceThread opsvc; |
73 | friend class OpHistoryServiceThread; | |
1e59de90 | 74 | std::unique_ptr<PerfCounters> logger; |
7c673cae FG |
75 | |
76 | public: | |
1e59de90 TL |
77 | OpHistory(CephContext *c) : cct(c), opsvc(this) { |
78 | PerfCountersBuilder b(cct, "osd-slow-ops", | |
79 | l_osd_slow_op_first, l_osd_slow_op_last); | |
80 | b.add_u64_counter(l_osd_slow_op_count, "slow_ops_count", | |
81 | "Number of operations taking over ten second"); | |
82 | ||
83 | logger.reset(b.create_perf_counters()); | |
84 | cct->get_perfcounters_collection()->add(logger.get()); | |
85 | ||
11fdf7f2 TL |
86 | opsvc.create("OpHistorySvc"); |
87 | } | |
7c673cae | 88 | ~OpHistory() { |
11fdf7f2 TL |
89 | ceph_assert(arrived.empty()); |
90 | ceph_assert(duration.empty()); | |
91 | ceph_assert(slow_op.empty()); | |
1e59de90 TL |
92 | if(logger) { |
93 | cct->get_perfcounters_collection()->remove(logger.get()); | |
94 | logger.reset(); | |
95 | } | |
11fdf7f2 TL |
96 | } |
97 | void insert(const utime_t& now, TrackedOpRef op) | |
98 | { | |
99 | if (shutdown) | |
100 | return; | |
101 | ||
102 | opsvc.insert_op(now, op); | |
7c673cae | 103 | } |
11fdf7f2 TL |
104 | |
105 | void _insert_delayed(const utime_t& now, TrackedOpRef op); | |
9f95a23c TL |
106 | void dump_ops(utime_t now, ceph::Formatter *f, std::set<std::string> filters = {""}, bool by_duration=false); |
107 | void dump_slow_ops(utime_t now, ceph::Formatter *f, std::set<std::string> filters = {""}); | |
7c673cae | 108 | void on_shutdown(); |
92f5a8d4 | 109 | void set_size_and_duration(size_t new_size, uint32_t new_duration) { |
7c673cae FG |
110 | history_size = new_size; |
111 | history_duration = new_duration; | |
112 | } | |
92f5a8d4 | 113 | void set_slow_op_size_and_threshold(size_t new_size, uint32_t new_threshold) { |
7c673cae FG |
114 | history_slow_op_size = new_size; |
115 | history_slow_op_threshold = new_threshold; | |
116 | } | |
117 | }; | |
118 | ||
119 | struct ShardedTrackingData; | |
120 | class OpTracker { | |
121 | friend class OpHistory; | |
31f18b77 | 122 | std::atomic<int64_t> seq = { 0 }; |
9f95a23c | 123 | std::vector<ShardedTrackingData*> sharded_in_flight_list; |
7c673cae | 124 | OpHistory history; |
11fdf7f2 | 125 | uint32_t num_optracker_shards; |
7c673cae FG |
126 | float complaint_time; |
127 | int log_threshold; | |
11fdf7f2 | 128 | std::atomic<bool> tracking_enabled; |
9f95a23c | 129 | ceph::shared_mutex lock = ceph::make_shared_mutex("OpTracker::lock"); |
7c673cae FG |
130 | |
131 | public: | |
132 | CephContext *cct; | |
133 | OpTracker(CephContext *cct_, bool tracking, uint32_t num_shards); | |
134 | ||
135 | void set_complaint_and_threshold(float time, int threshold) { | |
136 | complaint_time = time; | |
137 | log_threshold = threshold; | |
138 | } | |
139 | void set_history_size_and_duration(uint32_t new_size, uint32_t new_duration) { | |
140 | history.set_size_and_duration(new_size, new_duration); | |
141 | } | |
142 | void set_history_slow_op_size_and_threshold(uint32_t new_size, uint32_t new_threshold) { | |
143 | history.set_slow_op_size_and_threshold(new_size, new_threshold); | |
144 | } | |
11fdf7f2 TL |
145 | bool is_tracking() const { |
146 | return tracking_enabled; | |
147 | } | |
7c673cae | 148 | void set_tracking(bool enable) { |
7c673cae FG |
149 | tracking_enabled = enable; |
150 | } | |
1e59de90 | 151 | bool dump_ops_in_flight(ceph::Formatter *f, bool print_only_blocked = false, std::set<std::string> filters = {""}, bool count_only = false); |
9f95a23c TL |
152 | bool dump_historic_ops(ceph::Formatter *f, bool by_duration = false, std::set<std::string> filters = {""}); |
153 | bool dump_historic_slow_ops(ceph::Formatter *f, std::set<std::string> filters = {""}); | |
7c673cae FG |
154 | bool register_inflight_op(TrackedOp *i); |
155 | void unregister_inflight_op(TrackedOp *i); | |
11fdf7f2 | 156 | void record_history_op(TrackedOpRef&& i); |
7c673cae FG |
157 | |
158 | void get_age_ms_histogram(pow2_hist_t *h); | |
159 | ||
11fdf7f2 TL |
160 | /** |
161 | * walk through ops in flight | |
162 | * | |
163 | * @param oldest_sec the amount of time since the oldest op was initiated | |
164 | * @param check a function consuming tracked ops, the function returns | |
165 | * false if it don't want to be fed with more ops | |
166 | * @return True if there are any Ops to warn on, false otherwise | |
167 | */ | |
168 | bool visit_ops_in_flight(utime_t* oldest_secs, | |
169 | std::function<bool(TrackedOp&)>&& visit); | |
170 | /** | |
171 | * walk through slow ops in flight | |
172 | * | |
173 | * @param[out] oldest_sec the amount of time since the oldest op was initiated | |
174 | * @param[out] num_slow_ops total number of slow ops | |
175 | * @param[out] num_warned_ops total number of warned ops | |
176 | * @param on_warn a function consuming tracked ops, the function returns | |
177 | * false if it don't want to be fed with more ops | |
178 | * @return True if there are any Ops to warn on, false otherwise | |
179 | */ | |
180 | bool with_slow_ops_in_flight(utime_t* oldest_secs, | |
181 | int* num_slow_ops, | |
182 | int* num_warned_ops, | |
183 | std::function<void(TrackedOp&)>&& on_warn); | |
7c673cae FG |
184 | /** |
185 | * Look for Ops which are too old, and insert warning | |
186 | * strings for each Op that is too old. | |
187 | * | |
9f95a23c TL |
188 | * @param summary[out] a std::string summarizing slow Ops. |
189 | * @param warning_strings[out] A std::vector<std::string> reference which is filled | |
190 | * with a warning std::string for each old Op. | |
11fdf7f2 | 191 | * @param slow[out] total number of slow ops |
7c673cae FG |
192 | * @return True if there are any Ops to warn on, false otherwise. |
193 | */ | |
11fdf7f2 | 194 | bool check_ops_in_flight(std::string* summary, |
9f95a23c | 195 | std::vector<std::string> &warning_strings, |
11fdf7f2 | 196 | int* slow = nullptr); |
7c673cae FG |
197 | |
198 | void on_shutdown() { | |
199 | history.on_shutdown(); | |
200 | } | |
201 | ~OpTracker(); | |
202 | ||
203 | template <typename T, typename U> | |
204 | typename T::Ref create_request(U params) | |
205 | { | |
206 | typename T::Ref retval(new T(params, this)); | |
207 | retval->tracking_start(); | |
11fdf7f2 | 208 | if (is_tracking()) { |
11fdf7f2 | 209 | retval->mark_event("throttled", params->get_throttle_stamp()); |
9f95a23c | 210 | retval->mark_event("header_read", params->get_recv_stamp()); |
11fdf7f2 TL |
211 | retval->mark_event("all_read", params->get_recv_complete_stamp()); |
212 | retval->mark_event("dispatched", params->get_dispatch_stamp()); | |
213 | } | |
214 | ||
7c673cae FG |
215 | return retval; |
216 | } | |
217 | }; | |
218 | ||
7c673cae FG |
219 | class TrackedOp : public boost::intrusive::list_base_hook<> { |
220 | private: | |
221 | friend class OpHistory; | |
222 | friend class OpTracker; | |
223 | ||
224 | boost::intrusive::list_member_hook<> tracker_item; | |
225 | ||
226 | public: | |
227 | typedef boost::intrusive::list< | |
228 | TrackedOp, | |
229 | boost::intrusive::member_hook< | |
230 | TrackedOp, | |
231 | boost::intrusive::list_member_hook<>, | |
232 | &TrackedOp::tracker_item> > tracked_op_list_t; | |
233 | ||
234 | // for use when clearing lists. e.g., | |
235 | // ls.clear_and_dispose(TrackedOp::Putter()); | |
236 | struct Putter { | |
237 | void operator()(TrackedOp *op) { | |
238 | op->put(); | |
239 | } | |
240 | }; | |
241 | ||
242 | protected: | |
243 | OpTracker *tracker; ///< the tracker we are associated with | |
244 | std::atomic_int nref = {0}; ///< ref count | |
245 | ||
246 | utime_t initiated_at; | |
247 | ||
248 | struct Event { | |
249 | utime_t stamp; | |
11fdf7f2 | 250 | std::string str; |
7c673cae | 251 | |
11fdf7f2 | 252 | Event(utime_t t, std::string_view s) : stamp(t), str(s) {} |
7c673cae FG |
253 | |
254 | int compare(const char *s) const { | |
11fdf7f2 | 255 | return str.compare(s); |
7c673cae FG |
256 | } |
257 | ||
258 | const char *c_str() const { | |
11fdf7f2 | 259 | return str.c_str(); |
7c673cae FG |
260 | } |
261 | ||
9f95a23c | 262 | void dump(ceph::Formatter *f) const { |
7c673cae | 263 | f->dump_stream("time") << stamp; |
11fdf7f2 | 264 | f->dump_string("event", str); |
7c673cae FG |
265 | } |
266 | }; | |
267 | ||
9f95a23c | 268 | std::vector<Event> events; ///< std::list of events and their times |
11fdf7f2 | 269 | mutable ceph::mutex lock = ceph::make_mutex("TrackedOp::lock"); ///< to protect the events list |
9f95a23c | 270 | uint64_t seq = 0; ///< a unique value std::set by the OpTracker |
7c673cae FG |
271 | |
272 | uint32_t warn_interval_multiplier = 1; //< limits output of a given op warning | |
273 | ||
274 | enum { | |
275 | STATE_UNTRACKED = 0, | |
276 | STATE_LIVE, | |
277 | STATE_HISTORY | |
278 | }; | |
9f95a23c | 279 | std::atomic<int> state = {STATE_UNTRACKED}; |
7c673cae | 280 | |
9f95a23c | 281 | mutable std::string desc_str; ///< protected by lock |
7c673cae | 282 | mutable const char *desc = nullptr; ///< readable without lock |
9f95a23c | 283 | mutable std::atomic<bool> want_new_desc = {false}; |
7c673cae FG |
284 | |
285 | TrackedOp(OpTracker *_tracker, const utime_t& initiated) : | |
286 | tracker(_tracker), | |
287 | initiated_at(initiated) | |
288 | { | |
289 | events.reserve(OPTRACKER_PREALLOC_EVENTS); | |
290 | } | |
291 | ||
292 | /// output any type-specific data you want to get when dump() is called | |
9f95a23c | 293 | virtual void _dump(ceph::Formatter *f) const {} |
7c673cae FG |
294 | /// if you want something else to happen when events are marked, implement |
295 | virtual void _event_marked() {} | |
296 | /// return a unique descriptor of the Op; eg the message it's attached to | |
9f95a23c | 297 | virtual void _dump_op_descriptor_unlocked(std::ostream& stream) const = 0; |
7c673cae | 298 | /// called when the last non-OpTracker reference is dropped |
c07f9fc5 FG |
299 | virtual void _unregistered() {} |
300 | ||
9f95a23c | 301 | virtual bool filter_out(const std::set<std::string>& filters) { return true; } |
7c673cae FG |
302 | |
303 | public: | |
304 | ZTracer::Trace osd_trace; | |
305 | ZTracer::Trace pg_trace; | |
306 | ZTracer::Trace store_trace; | |
307 | ZTracer::Trace journal_trace; | |
308 | ||
309 | virtual ~TrackedOp() {} | |
310 | ||
311 | void get() { | |
312 | ++nref; | |
313 | } | |
314 | void put() { | |
11fdf7f2 TL |
315 | again: |
316 | auto nref_snap = nref.load(); | |
317 | if (nref_snap == 1) { | |
7c673cae FG |
318 | switch (state.load()) { |
319 | case STATE_UNTRACKED: | |
320 | _unregistered(); | |
321 | delete this; | |
322 | break; | |
323 | ||
324 | case STATE_LIVE: | |
325 | mark_event("done"); | |
326 | tracker->unregister_inflight_op(this); | |
11fdf7f2 TL |
327 | _unregistered(); |
328 | if (!tracker->is_tracking()) { | |
329 | delete this; | |
330 | } else { | |
331 | state = TrackedOp::STATE_HISTORY; | |
332 | tracker->record_history_op( | |
333 | TrackedOpRef(this, /* add_ref = */ false)); | |
334 | } | |
7c673cae FG |
335 | break; |
336 | ||
337 | case STATE_HISTORY: | |
338 | delete this; | |
339 | break; | |
340 | ||
341 | default: | |
342 | ceph_abort(); | |
343 | } | |
11fdf7f2 TL |
344 | } else if (!nref.compare_exchange_weak(nref_snap, nref_snap - 1)) { |
345 | goto again; | |
7c673cae FG |
346 | } |
347 | } | |
348 | ||
349 | const char *get_desc() const { | |
350 | if (!desc || want_new_desc.load()) { | |
11fdf7f2 | 351 | std::lock_guard l(lock); |
7c673cae FG |
352 | _gen_desc(); |
353 | } | |
354 | return desc; | |
355 | } | |
356 | private: | |
357 | void _gen_desc() const { | |
9f95a23c | 358 | std::ostringstream ss; |
7c673cae FG |
359 | _dump_op_descriptor_unlocked(ss); |
360 | desc_str = ss.str(); | |
361 | desc = desc_str.c_str(); | |
362 | want_new_desc = false; | |
363 | } | |
364 | public: | |
365 | void reset_desc() { | |
366 | want_new_desc = true; | |
367 | } | |
368 | ||
369 | const utime_t& get_initiated() const { | |
370 | return initiated_at; | |
371 | } | |
372 | ||
373 | double get_duration() const { | |
11fdf7f2 | 374 | std::lock_guard l(lock); |
7c673cae FG |
375 | if (!events.empty() && events.rbegin()->compare("done") == 0) |
376 | return events.rbegin()->stamp - get_initiated(); | |
377 | else | |
378 | return ceph_clock_now() - get_initiated(); | |
379 | } | |
380 | ||
11fdf7f2 | 381 | void mark_event(std::string_view event, utime_t stamp=ceph_clock_now()); |
7c673cae | 382 | |
1adf2230 AA |
383 | void mark_nowarn() { |
384 | warn_interval_multiplier = 0; | |
385 | } | |
386 | ||
11fdf7f2 TL |
387 | virtual std::string_view state_string() const { |
388 | std::lock_guard l(lock); | |
389 | return events.empty() ? std::string_view() : std::string_view(events.rbegin()->str); | |
7c673cae FG |
390 | } |
391 | ||
9f95a23c | 392 | void dump(utime_t now, ceph::Formatter *f) const; |
7c673cae FG |
393 | |
394 | void tracking_start() { | |
395 | if (tracker->register_inflight_op(this)) { | |
11fdf7f2 | 396 | events.emplace_back(initiated_at, "initiated"); |
7c673cae FG |
397 | state = STATE_LIVE; |
398 | } | |
399 | } | |
400 | ||
401 | // ref counting via intrusive_ptr, with special behavior on final | |
402 | // put for historical op tracking | |
403 | friend void intrusive_ptr_add_ref(TrackedOp *o) { | |
404 | o->get(); | |
405 | } | |
406 | friend void intrusive_ptr_release(TrackedOp *o) { | |
407 | o->put(); | |
408 | } | |
409 | }; | |
410 | ||
411 | ||
412 | #endif |