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
|
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab ft=cpp
/*
* Ceph - scalable distributed file system
*
* Copyright (C) 2018 Red Hat, Inc.
*
* This is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License version 2.1, as published by the Free Software
* Foundation. See file COPYING.
*
*/
#pragma once
#include "include/rados/librados_fwd.hpp"
#include <memory>
#include "common/ceph_mutex.h"
#include "common/async/completion.h"
#include "common/async/yield_context.h"
#include "services/svc_rados.h"
#include "rgw_aio.h"
namespace rgw {
class Throttle {
protected:
const uint64_t window;
uint64_t pending_size = 0;
AioResultList pending;
AioResultList completed;
bool is_available() const { return pending_size <= window; }
bool has_completion() const { return !completed.empty(); }
bool is_drained() const { return pending.empty(); }
enum class Wait { None, Available, Completion, Drained };
Wait waiter = Wait::None;
bool waiter_ready() const;
public:
Throttle(uint64_t window) : window(window) {}
~Throttle() {
// must drain before destructing
ceph_assert(pending.empty());
ceph_assert(completed.empty());
}
};
// a throttle for aio operations. all public functions must be called from
// the same thread
class BlockingAioThrottle final : public Aio, private Throttle {
ceph::mutex mutex = ceph::make_mutex("AioThrottle");
ceph::condition_variable cond;
struct Pending : AioResultEntry {
BlockingAioThrottle *parent = nullptr;
uint64_t cost = 0;
librados::AioCompletion *completion = nullptr;
};
public:
BlockingAioThrottle(uint64_t window) : Throttle(window) {}
AioResultList get(const RGWSI_RADOS::Obj& obj, OpFunc&& f,
uint64_t cost, uint64_t id) override final;
void put(AioResult& r) override final;
AioResultList poll() override final;
AioResultList wait() override final;
AioResultList drain() override final;
};
// a throttle that yields the coroutine instead of blocking. all public
// functions must be called within the coroutine strand
class YieldingAioThrottle final : public Aio, private Throttle {
boost::asio::io_context& context;
yield_context yield;
struct Handler;
// completion callback associated with the waiter
using Completion = ceph::async::Completion<void(boost::system::error_code)>;
std::unique_ptr<Completion> completion;
template <typename CompletionToken>
auto async_wait(CompletionToken&& token);
struct Pending : AioResultEntry { uint64_t cost = 0; };
public:
YieldingAioThrottle(uint64_t window, boost::asio::io_context& context,
yield_context yield)
: Throttle(window), context(context), yield(yield)
{}
AioResultList get(const RGWSI_RADOS::Obj& obj, OpFunc&& f,
uint64_t cost, uint64_t id) override final;
void put(AioResult& r) override final;
AioResultList poll() override final;
AioResultList wait() override final;
AioResultList drain() override final;
};
// return a smart pointer to Aio
inline auto make_throttle(uint64_t window_size, optional_yield y)
{
std::unique_ptr<Aio> aio;
if (y) {
aio = std::make_unique<YieldingAioThrottle>(window_size,
y.get_io_context(),
y.get_yield_context());
} else {
aio = std::make_unique<BlockingAioThrottle>(window_size);
}
return aio;
}
} // namespace rgw
|
