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
|
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
/*
* Ceph - scalable distributed file system
*
* Copyright (C) 2017 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.
*/
#ifndef LIBRADOS_ASIO_H
#define LIBRADOS_ASIO_H
#include "include/rados/librados.hpp"
#include "common/async/completion.h"
/// Defines asynchronous librados operations that satisfy all of the
/// "Requirements on asynchronous operations" imposed by the C++ Networking TS
/// in section 13.2.7. Many of the type and variable names below are taken
/// directly from those requirements.
///
/// The current draft of the Networking TS (as of 2017-11-27) is available here:
/// http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2017/n4711.pdf
///
/// The boost::asio documentation duplicates these requirements here:
/// http://www.boost.org/doc/libs/1_66_0/doc/html/boost_asio/reference/asynchronous_operations.html
namespace librados {
namespace detail {
/// unique_ptr with custom deleter for AioCompletion
struct AioCompletionDeleter {
void operator()(AioCompletion *c) { c->release(); }
};
using unique_aio_completion_ptr =
std::unique_ptr<AioCompletion, AioCompletionDeleter>;
/// Invokes the given completion handler. When the type of Result is not void,
/// storage is provided for it and that result is passed as an additional
/// argument to the handler.
template <typename Result>
struct Invoker {
using Signature = void(boost::system::error_code, Result);
Result result;
template <typename Completion>
void dispatch(Completion&& completion, boost::system::error_code ec) {
ceph::async::dispatch(std::move(completion), ec, std::move(result));
}
};
// specialization for Result=void
template <>
struct Invoker<void> {
using Signature = void(boost::system::error_code);
template <typename Completion>
void dispatch(Completion&& completion, boost::system::error_code ec) {
ceph::async::dispatch(std::move(completion), ec);
}
};
template <typename Result>
struct AsyncOp : Invoker<Result> {
unique_aio_completion_ptr aio_completion;
using Signature = typename Invoker<Result>::Signature;
using Completion = ceph::async::Completion<Signature, AsyncOp<Result>>;
static void aio_dispatch(completion_t cb, void *arg) {
// reclaim ownership of the completion
auto p = std::unique_ptr<Completion>{static_cast<Completion*>(arg)};
// move result out of Completion memory being freed
auto op = std::move(p->user_data);
const int ret = op.aio_completion->get_return_value();
boost::system::error_code ec;
if (ret < 0) {
ec.assign(-ret, boost::system::system_category());
}
op.dispatch(std::move(p), ec);
}
template <typename Executor1, typename CompletionHandler>
static auto create(const Executor1& ex1, CompletionHandler&& handler) {
auto p = Completion::create(ex1, std::move(handler));
p->user_data.aio_completion.reset(
Rados::aio_create_completion(p.get(), aio_dispatch));
return p;
}
};
} // namespace detail
/// Calls IoCtx::aio_read() and arranges for the AioCompletion to call a
/// given handler with signature (boost::system::error_code, bufferlist).
template <typename ExecutionContext, typename CompletionToken>
auto async_read(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
size_t len, uint64_t off, CompletionToken&& token)
{
using Op = detail::AsyncOp<bufferlist>;
using Signature = typename Op::Signature;
boost::asio::async_completion<CompletionToken, Signature> init(token);
auto p = Op::create(ctx.get_executor(), init.completion_handler);
auto& op = p->user_data;
int ret = io.aio_read(oid, op.aio_completion.get(), &op.result, len, off);
if (ret < 0) {
auto ec = boost::system::error_code{-ret, boost::system::system_category()};
ceph::async::post(std::move(p), ec, bufferlist{});
} else {
p.release(); // release ownership until completion
}
return init.result.get();
}
/// Calls IoCtx::aio_write() and arranges for the AioCompletion to call a
/// given handler with signature (boost::system::error_code).
template <typename ExecutionContext, typename CompletionToken>
auto async_write(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
bufferlist &bl, size_t len, uint64_t off,
CompletionToken&& token)
{
using Op = detail::AsyncOp<void>;
using Signature = typename Op::Signature;
boost::asio::async_completion<CompletionToken, Signature> init(token);
auto p = Op::create(ctx.get_executor(), init.completion_handler);
auto& op = p->user_data;
int ret = io.aio_write(oid, op.aio_completion.get(), bl, len, off);
if (ret < 0) {
auto ec = boost::system::error_code{-ret, boost::system::system_category()};
ceph::async::post(std::move(p), ec);
} else {
p.release(); // release ownership until completion
}
return init.result.get();
}
/// Calls IoCtx::aio_operate() and arranges for the AioCompletion to call a
/// given handler with signature (boost::system::error_code, bufferlist).
template <typename ExecutionContext, typename CompletionToken>
auto async_operate(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
ObjectReadOperation *read_op, int flags,
CompletionToken&& token)
{
using Op = detail::AsyncOp<bufferlist>;
using Signature = typename Op::Signature;
boost::asio::async_completion<CompletionToken, Signature> init(token);
auto p = Op::create(ctx.get_executor(), init.completion_handler);
auto& op = p->user_data;
int ret = io.aio_operate(oid, op.aio_completion.get(), read_op,
flags, &op.result);
if (ret < 0) {
auto ec = boost::system::error_code{-ret, boost::system::system_category()};
ceph::async::post(std::move(p), ec, bufferlist{});
} else {
p.release(); // release ownership until completion
}
return init.result.get();
}
/// Calls IoCtx::aio_operate() and arranges for the AioCompletion to call a
/// given handler with signature (boost::system::error_code).
template <typename ExecutionContext, typename CompletionToken>
auto async_operate(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
ObjectWriteOperation *write_op, int flags,
CompletionToken &&token)
{
using Op = detail::AsyncOp<void>;
using Signature = typename Op::Signature;
boost::asio::async_completion<CompletionToken, Signature> init(token);
auto p = Op::create(ctx.get_executor(), init.completion_handler);
auto& op = p->user_data;
int ret = io.aio_operate(oid, op.aio_completion.get(), write_op, flags);
if (ret < 0) {
auto ec = boost::system::error_code{-ret, boost::system::system_category()};
ceph::async::post(std::move(p), ec);
} else {
p.release(); // release ownership until completion
}
return init.result.get();
}
/// Calls IoCtx::aio_notify() and arranges for the AioCompletion to call a
/// given handler with signature (boost::system::error_code, bufferlist).
template <typename ExecutionContext, typename CompletionToken>
auto async_notify(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
bufferlist& bl, uint64_t timeout_ms, CompletionToken &&token)
{
using Op = detail::AsyncOp<bufferlist>;
using Signature = typename Op::Signature;
boost::asio::async_completion<CompletionToken, Signature> init(token);
auto p = Op::create(ctx.get_executor(), init.completion_handler);
auto& op = p->user_data;
int ret = io.aio_notify(oid, op.aio_completion.get(),
bl, timeout_ms, &op.result);
if (ret < 0) {
auto ec = boost::system::error_code{-ret, boost::system::system_category()};
ceph::async::post(std::move(p), ec, bufferlist{});
} else {
p.release(); // release ownership until completion
}
return init.result.get();
}
} // namespace librados
#endif // LIBRADOS_ASIO_H
|
