--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include <condition_variable>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+#include <boost/asio.hpp>
+#include <boost/asio/spawn.hpp>
+
+#include <beast/core/placeholders.hpp>
+#include <beast/http/read.hpp>
+#include <beast/http/string_body.hpp>
+#include <beast/http/write.hpp>
+
+#include "rgw_asio_frontend.h"
+#include "rgw_asio_client.h"
+
+#define dout_subsys ceph_subsys_rgw
+
+#undef dout_prefix
+#define dout_prefix (*_dout << "asio: ")
+
+namespace {
+
+class Pauser {
+ std::mutex mutex;
+ std::condition_variable cond_ready; // signaled on ready==true
+ std::condition_variable cond_paused; // signaled on waiters==thread_count
+ bool ready{false};
+ int waiters{0};
+ public:
+ template <typename Func>
+ void pause(int thread_count, Func&& func);
+ void unpause();
+ void wait();
+};
+
+template <typename Func>
+void Pauser::pause(int thread_count, Func&& func)
+{
+ std::unique_lock<std::mutex> lock(mutex);
+ ready = false;
+ lock.unlock();
+
+ func();
+
+ // wait for all threads to pause
+ lock.lock();
+ cond_paused.wait(lock, [=] { return waiters == thread_count; });
+}
+
+void Pauser::unpause()
+{
+ std::lock_guard<std::mutex> lock(mutex);
+ ready = true;
+ cond_ready.notify_all();
+}
+
+void Pauser::wait()
+{
+ std::unique_lock<std::mutex> lock(mutex);
+ ++waiters;
+ cond_paused.notify_one(); // notify pause() that we're waiting
+ cond_ready.wait(lock, [this] { return ready; }); // wait for unpause()
+ --waiters;
+}
+
+using tcp = boost::asio::ip::tcp;
+
+// coroutine to handle a client connection to completion
+static void handle_connection(RGWProcessEnv& env, tcp::socket socket,
+ boost::asio::yield_context yield)
+{
+ auto cct = env.store->ctx();
+ boost::system::error_code ec;
+
+ beast::flat_streambuf buffer{1024};
+
+ // read messages from the socket until eof
+ for (;;) {
+ // parse the header
+ rgw::asio::parser_type parser;
+ do {
+ auto bytes = beast::http::async_read_some(socket, buffer, parser, yield[ec]);
+ buffer.consume(bytes);
+ } while (!ec && !parser.got_header());
+
+ if (ec == boost::asio::error::connection_reset ||
+ ec == boost::asio::error::eof) {
+ return;
+ }
+ if (ec) {
+ auto& message = parser.get();
+ ldout(cct, 1) << "read failed: " << ec.message() << dendl;
+ ldout(cct, 1) << "====== req done http_status=400 ======" << dendl;
+ beast::http::response<beast::http::string_body> response;
+ response.status = 400;
+ response.reason = "Bad Request";
+ response.version = message.version == 10 ? 10 : 11;
+ beast::http::prepare(response);
+ beast::http::async_write(socket, std::move(response), yield[ec]);
+ // ignore ec
+ return;
+ }
+
+ // process the request
+ RGWRequest req{env.store->get_new_req_id()};
+
+ rgw::asio::ClientIO real_client{socket, parser, buffer};
+
+ auto real_client_io = rgw::io::add_reordering(
+ rgw::io::add_buffering(cct,
+ rgw::io::add_chunking(
+ rgw::io::add_conlen_controlling(
+ &real_client))));
+ RGWRestfulIO client(cct, &real_client_io);
+ process_request(env.store, env.rest, &req, env.uri_prefix,
+ *env.auth_registry, &client, env.olog);
+
+ if (real_client.get_conn_close()) {
+ return;
+ }
+ }
+}
+
+class AsioFrontend {
+ RGWProcessEnv env;
+ boost::asio::io_service service;
+
+ tcp::acceptor acceptor;
+ tcp::socket peer_socket;
+
+ std::vector<std::thread> threads;
+ Pauser pauser;
+ std::atomic<bool> going_down{false};
+
+ CephContext* ctx() const { return env.store->ctx(); }
+
+ void accept(boost::system::error_code ec);
+
+ public:
+ AsioFrontend(const RGWProcessEnv& env)
+ : env(env), acceptor(service), peer_socket(service) {}
+
+ int init();
+ int run();
+ void stop();
+ void join();
+ void pause();
+ void unpause(RGWRados* store, rgw_auth_registry_ptr_t);
+};
+
+int AsioFrontend::init()
+{
+ auto ep = tcp::endpoint{tcp::v4(), static_cast<unsigned short>(env.port)};
+ ldout(ctx(), 4) << "frontend listening on " << ep << dendl;
+
+ boost::system::error_code ec;
+ acceptor.open(ep.protocol(), ec);
+ if (ec) {
+ lderr(ctx()) << "failed to open socket: " << ec.message() << dendl;
+ return -ec.value();
+ }
+ acceptor.set_option(tcp::acceptor::reuse_address(true));
+ acceptor.bind(ep, ec);
+ if (ec) {
+ lderr(ctx()) << "failed to bind address " << ep <<
+ ": " << ec.message() << dendl;
+ return -ec.value();
+ }
+ acceptor.listen(boost::asio::socket_base::max_connections);
+ acceptor.async_accept(peer_socket,
+ [this] (boost::system::error_code ec) {
+ return accept(ec);
+ });
+ return 0;
+}
+
+void AsioFrontend::accept(boost::system::error_code ec)
+{
+ if (!acceptor.is_open()) {
+ return;
+ } else if (ec == boost::asio::error::operation_aborted) {
+ return;
+ } else if (ec) {
+ throw ec;
+ }
+ auto socket = std::move(peer_socket);
+ // spawn a coroutine to handle the connection
+ boost::asio::spawn(service,
+ [&] (boost::asio::yield_context yield) {
+ handle_connection(env, std::move(socket), yield);
+ });
+ acceptor.async_accept(peer_socket,
+ [this] (boost::system::error_code ec) {
+ return accept(ec);
+ });
+}
+
+int AsioFrontend::run()
+{
+ auto cct = ctx();
+ const int thread_count = cct->_conf->rgw_thread_pool_size;
+ threads.reserve(thread_count);
+
+ ldout(cct, 4) << "frontend spawning " << thread_count << " threads" << dendl;
+
+ for (int i = 0; i < thread_count; i++) {
+ threads.emplace_back([=] {
+ for (;;) {
+ service.run();
+ if (going_down) {
+ break;
+ }
+ pauser.wait();
+ }
+ });
+ }
+ return 0;
+}
+
+void AsioFrontend::stop()
+{
+ ldout(ctx(), 4) << "frontend initiating shutdown..." << dendl;
+
+ going_down = true;
+
+ boost::system::error_code ec;
+ acceptor.close(ec); // unblock the run() threads
+}
+
+void AsioFrontend::join()
+{
+ if (!going_down) {
+ stop();
+ }
+ ldout(ctx(), 4) << "frontend joining threads..." << dendl;
+ for (auto& thread : threads) {
+ thread.join();
+ }
+ ldout(ctx(), 4) << "frontend done" << dendl;
+}
+
+void AsioFrontend::pause()
+{
+ ldout(ctx(), 4) << "frontend pausing threads..." << dendl;
+ pauser.pause(threads.size(), [=] {
+ // stop accepting but leave the port open
+ boost::system::error_code ec;
+ acceptor.cancel(ec);
+ });
+ ldout(ctx(), 4) << "frontend paused" << dendl;
+}
+
+void AsioFrontend::unpause(RGWRados* const store,
+ rgw_auth_registry_ptr_t auth_registry)
+{
+ env.store = store;
+ env.auth_registry = std::move(auth_registry);
+ ldout(ctx(), 4) << "frontend unpaused" << dendl;
+ service.reset();
+ acceptor.async_accept(peer_socket,
+ [this] (boost::system::error_code ec) {
+ return accept(ec);
+ });
+ pauser.unpause();
+}
+
+} // anonymous namespace
+
+class RGWAsioFrontend::Impl : public AsioFrontend {
+ public:
+ Impl(const RGWProcessEnv& env) : AsioFrontend(env) {}
+};
+
+RGWAsioFrontend::RGWAsioFrontend(const RGWProcessEnv& env)
+ : impl(new Impl(env))
+{
+}
+
+RGWAsioFrontend::~RGWAsioFrontend() = default;
+
+int RGWAsioFrontend::init()
+{
+ return impl->init();
+}
+
+int RGWAsioFrontend::run()
+{
+ return impl->run();
+}
+
+void RGWAsioFrontend::stop()
+{
+ impl->stop();
+}
+
+void RGWAsioFrontend::join()
+{
+ impl->join();
+}
+
+void RGWAsioFrontend::pause_for_new_config()
+{
+ impl->pause();
+}
+
+void RGWAsioFrontend::unpause_with_new_config(
+ RGWRados* const store,
+ rgw_auth_registry_ptr_t auth_registry
+) {
+ impl->unpause(store, std::move(auth_registry));
+}
Code Review / stor4nfv.git / blobdiff