diff options
Diffstat (limited to 'src/tools/immutable_object_cache')
19 files changed, 2399 insertions, 0 deletions
diff --git a/src/tools/immutable_object_cache/CMakeLists.txt b/src/tools/immutable_object_cache/CMakeLists.txt new file mode 100644 index 000000000..ed118ed6f --- /dev/null +++ b/src/tools/immutable_object_cache/CMakeLists.txt @@ -0,0 +1,19 @@ +set(ceph_immutable_object_cache_files + ObjectCacheStore.cc + CacheController.cc + CacheServer.cc + CacheClient.cc + CacheSession.cc + SimplePolicy.cc + Types.cc + ) +add_library(ceph_immutable_object_cache_lib STATIC ${ceph_immutable_object_cache_files}) + +add_executable(ceph-immutable-object-cache + main.cc) +target_link_libraries(ceph-immutable-object-cache + ceph_immutable_object_cache_lib + librados + StdFilesystem::filesystem + global) +install(TARGETS ceph-immutable-object-cache DESTINATION bin) diff --git a/src/tools/immutable_object_cache/CacheClient.cc b/src/tools/immutable_object_cache/CacheClient.cc new file mode 100644 index 000000000..2b837be51 --- /dev/null +++ b/src/tools/immutable_object_cache/CacheClient.cc @@ -0,0 +1,435 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include <boost/bind/bind.hpp> +#include "CacheClient.h" +#include "common/Cond.h" +#include "common/version.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_immutable_obj_cache +#undef dout_prefix +#define dout_prefix *_dout << "ceph::cache::CacheClient: " << this << " " \ + << __func__ << ": " + +namespace ceph { +namespace immutable_obj_cache { + + CacheClient::CacheClient(const std::string& file, CephContext* ceph_ctx) + : m_cct(ceph_ctx), m_io_service_work(m_io_service), + m_dm_socket(m_io_service), m_ep(stream_protocol::endpoint(file)), + m_io_thread(nullptr), m_session_work(false), m_writing(false), + m_reading(false), m_sequence_id(0) { + m_worker_thread_num = + m_cct->_conf.get_val<uint64_t>( + "immutable_object_cache_client_dedicated_thread_num"); + + if (m_worker_thread_num != 0) { + m_worker = new boost::asio::io_service(); + m_worker_io_service_work = new boost::asio::io_service::work(*m_worker); + for (uint64_t i = 0; i < m_worker_thread_num; i++) { + std::thread* thd = new std::thread([this](){m_worker->run();}); + m_worker_threads.push_back(thd); + } + } + m_bp_header = buffer::create(get_header_size()); + } + + CacheClient::~CacheClient() { + stop(); + } + + void CacheClient::run() { + m_io_thread.reset(new std::thread([this](){m_io_service.run(); })); + } + + bool CacheClient::is_session_work() { + return m_session_work.load() == true; + } + + int CacheClient::stop() { + m_session_work.store(false); + m_io_service.stop(); + + if (m_io_thread != nullptr) { + m_io_thread->join(); + } + if (m_worker_thread_num != 0) { + m_worker->stop(); + for (auto thd : m_worker_threads) { + thd->join(); + delete thd; + } + delete m_worker_io_service_work; + delete m_worker; + } + return 0; + } + + // close domain socket + void CacheClient::close() { + m_session_work.store(false); + boost::system::error_code close_ec; + m_dm_socket.close(close_ec); + if (close_ec) { + ldout(m_cct, 20) << "close: " << close_ec.message() << dendl; + } + } + + // sync connect + int CacheClient::connect() { + int ret = -1; + C_SaferCond cond; + Context* on_finish = new LambdaContext([&cond, &ret](int err) { + ret = err; + cond.complete(err); + }); + + connect(on_finish); + cond.wait(); + + return ret; + } + + // async connect + void CacheClient::connect(Context* on_finish) { + m_dm_socket.async_connect(m_ep, + boost::bind(&CacheClient::handle_connect, this, + on_finish, boost::asio::placeholders::error)); + } + + void CacheClient::handle_connect(Context* on_finish, + const boost::system::error_code& err) { + if (err) { + ldout(m_cct, 20) << "fails to connect to cache server. error : " + << err.message() << dendl; + fault(ASIO_ERROR_CONNECT, err); + on_finish->complete(-1); + return; + } + + ldout(m_cct, 20) << "successfully connected to cache server." << dendl; + on_finish->complete(0); + } + + void CacheClient::lookup_object(std::string pool_nspace, uint64_t pool_id, + uint64_t snap_id, uint64_t object_size, + std::string oid, + CacheGenContextURef&& on_finish) { + ldout(m_cct, 20) << dendl; + ObjectCacheRequest* req = new ObjectCacheReadData(RBDSC_READ, + ++m_sequence_id, 0, 0, pool_id, + snap_id, object_size, oid, pool_nspace); + req->process_msg = std::move(on_finish); + req->encode(); + + { + std::lock_guard locker{m_lock}; + m_outcoming_bl.append(req->get_payload_bufferlist()); + ceph_assert(m_seq_to_req.find(req->seq) == m_seq_to_req.end()); + m_seq_to_req[req->seq] = req; + } + + // try to send message to server. + try_send(); + + // try to receive ack from server. + try_receive(); + } + + void CacheClient::try_send() { + ldout(m_cct, 20) << dendl; + if (!m_writing.load()) { + m_writing.store(true); + send_message(); + } + } + + void CacheClient::send_message() { + ldout(m_cct, 20) << dendl; + bufferlist bl; + { + std::lock_guard locker{m_lock}; + bl.swap(m_outcoming_bl); + ceph_assert(m_outcoming_bl.length() == 0); + } + + // send bytes as many as possible. + boost::asio::async_write(m_dm_socket, + boost::asio::buffer(bl.c_str(), bl.length()), + boost::asio::transfer_exactly(bl.length()), + [this, bl](const boost::system::error_code& err, size_t cb) { + if (err || cb != bl.length()) { + fault(ASIO_ERROR_WRITE, err); + return; + } + + ceph_assert(cb == bl.length()); + + { + std::lock_guard locker{m_lock}; + if (m_outcoming_bl.length() == 0) { + m_writing.store(false); + return; + } + } + + // still have left bytes, continue to send. + send_message(); + }); + try_receive(); + } + + void CacheClient::try_receive() { + ldout(m_cct, 20) << dendl; + if (!m_reading.load()) { + m_reading.store(true); + receive_message(); + } + } + + void CacheClient::receive_message() { + ldout(m_cct, 20) << dendl; + ceph_assert(m_reading.load()); + read_reply_header(); + } + + void CacheClient::read_reply_header() { + ldout(m_cct, 20) << dendl; + /* create new head buffer for every reply */ + bufferptr bp_head(buffer::create(get_header_size())); + auto raw_ptr = bp_head.c_str(); + + boost::asio::async_read(m_dm_socket, + boost::asio::buffer(raw_ptr, get_header_size()), + boost::asio::transfer_exactly(get_header_size()), + boost::bind(&CacheClient::handle_reply_header, + this, bp_head, + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred)); + } + + void CacheClient::handle_reply_header(bufferptr bp_head, + const boost::system::error_code& ec, + size_t bytes_transferred) { + ldout(m_cct, 20) << dendl; + if (ec || bytes_transferred != get_header_size()) { + fault(ASIO_ERROR_READ, ec); + return; + } + + ceph_assert(bytes_transferred == bp_head.length()); + + uint32_t data_len = get_data_len(bp_head.c_str()); + + bufferptr bp_data(buffer::create(data_len)); + read_reply_data(std::move(bp_head), std::move(bp_data), data_len); + } + + void CacheClient::read_reply_data(bufferptr&& bp_head, + bufferptr&& bp_data, + const uint64_t data_len) { + ldout(m_cct, 20) << dendl; + auto raw_ptr = bp_data.c_str(); + boost::asio::async_read(m_dm_socket, boost::asio::buffer(raw_ptr, data_len), + boost::asio::transfer_exactly(data_len), + boost::bind(&CacheClient::handle_reply_data, + this, std::move(bp_head), std::move(bp_data), data_len, + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred)); + } + + void CacheClient::handle_reply_data(bufferptr bp_head, + bufferptr bp_data, + const uint64_t data_len, + const boost::system::error_code& ec, + size_t bytes_transferred) { + ldout(m_cct, 20) << dendl; + if (ec || bytes_transferred != data_len) { + fault(ASIO_ERROR_WRITE, ec); + return; + } + ceph_assert(bp_data.length() == data_len); + + bufferlist data_buffer; + data_buffer.append(std::move(bp_head)); + data_buffer.append(std::move(bp_data)); + + ObjectCacheRequest* reply = decode_object_cache_request(data_buffer); + data_buffer.clear(); + ceph_assert(data_buffer.length() == 0); + + process(reply, reply->seq); + + { + std::lock_guard locker{m_lock}; + if (m_seq_to_req.size() == 0 && m_outcoming_bl.length()) { + m_reading.store(false); + return; + } + } + if (is_session_work()) { + receive_message(); + } + } + + void CacheClient::process(ObjectCacheRequest* reply, uint64_t seq_id) { + ldout(m_cct, 20) << dendl; + ObjectCacheRequest* current_request = nullptr; + { + std::lock_guard locker{m_lock}; + ceph_assert(m_seq_to_req.find(seq_id) != m_seq_to_req.end()); + current_request = m_seq_to_req[seq_id]; + m_seq_to_req.erase(seq_id); + } + + ceph_assert(current_request != nullptr); + auto process_reply = new LambdaContext([current_request, reply] + (bool dedicated) { + if (dedicated) { + // dedicated thrad to execute this context. + } + current_request->process_msg.release()->complete(reply); + delete current_request; + delete reply; + }); + + if (m_worker_thread_num != 0) { + m_worker->post([process_reply]() { + process_reply->complete(true); + }); + } else { + process_reply->complete(false); + } + } + + // if there is one request fails, just execute fault, then shutdown RO. + void CacheClient::fault(const int err_type, + const boost::system::error_code& ec) { + ldout(m_cct, 20) << "fault." << ec.message() << dendl; + + if (err_type == ASIO_ERROR_CONNECT) { + ceph_assert(!m_session_work.load()); + if (ec == boost::asio::error::connection_refused) { + ldout(m_cct, 20) << "Connecting RO daenmon fails : "<< ec.message() + << ". Immutable-object-cache daemon is down ? " + << "Data will be read from ceph cluster " << dendl; + } else { + ldout(m_cct, 20) << "Connecting RO daemon fails : " + << ec.message() << dendl; + } + + if (m_dm_socket.is_open()) { + // Set to indicate what error occurred, if any. + // Note that, even if the function indicates an error, + // the underlying descriptor is closed. + boost::system::error_code close_ec; + m_dm_socket.close(close_ec); + if (close_ec) { + ldout(m_cct, 20) << "close: " << close_ec.message() << dendl; + } + } + return; + } + + if (!m_session_work.load()) { + return; + } + + /* when current session don't work, ASIO will don't receive any new request from hook. + * On the other hand, for pending request of ASIO, cancle these request, + * then call their callback. these request which are cancled by this method, + * will be re-dispatched to RADOS layer. + * make sure just have one thread to modify execute below code. */ + m_session_work.store(false); + + if (err_type == ASIO_ERROR_MSG_INCOMPLETE) { + ldout(m_cct, 20) << "ASIO In-complete message." << ec.message() << dendl; + ceph_assert(0); + } + + if (err_type == ASIO_ERROR_READ) { + ldout(m_cct, 20) << "ASIO async read fails : " << ec.message() << dendl; + } + + if (err_type == ASIO_ERROR_WRITE) { + ldout(m_cct, 20) << "ASIO asyn write fails : " << ec.message() << dendl; + // CacheClient should not occur this error. + ceph_assert(0); + } + + // currently, for any asio error, just shutdown RO. + close(); + + /* all pending request, which have entered into ASIO, + * will be re-dispatched to RADOS.*/ + { + std::lock_guard locker{m_lock}; + for (auto it : m_seq_to_req) { + it.second->type = RBDSC_READ_RADOS; + it.second->process_msg->complete(it.second); + } + m_seq_to_req.clear(); + } + + ldout(m_cct, 20) << "Because ASIO domain socket fails, just shutdown RO.\ + Later all reading will be re-dispatched RADOS layer" + << ec.message() << dendl; + } + + // TODO : re-implement this method + int CacheClient::register_client(Context* on_finish) { + ObjectCacheRequest* reg_req = new ObjectCacheRegData(RBDSC_REGISTER, + m_sequence_id++, + ceph_version_to_str()); + reg_req->encode(); + + bufferlist bl; + bl.append(reg_req->get_payload_bufferlist()); + + uint64_t ret; + boost::system::error_code ec; + + ret = boost::asio::write(m_dm_socket, + boost::asio::buffer(bl.c_str(), bl.length()), ec); + + if (ec || ret != bl.length()) { + fault(ASIO_ERROR_WRITE, ec); + return -1; + } + delete reg_req; + + ret = boost::asio::read(m_dm_socket, + boost::asio::buffer(m_bp_header.c_str(), get_header_size()), ec); + if (ec || ret != get_header_size()) { + fault(ASIO_ERROR_READ, ec); + return -1; + } + + uint64_t data_len = get_data_len(m_bp_header.c_str()); + bufferptr bp_data(buffer::create(data_len)); + + ret = boost::asio::read(m_dm_socket, boost::asio::buffer(bp_data.c_str(), + data_len), ec); + if (ec || ret != data_len) { + fault(ASIO_ERROR_READ, ec); + return -1; + } + + bufferlist data_buffer; + data_buffer.append(m_bp_header); + data_buffer.append(std::move(bp_data)); + ObjectCacheRequest* req = decode_object_cache_request(data_buffer); + if (req->type == RBDSC_REGISTER_REPLY) { + m_session_work.store(true); + on_finish->complete(0); + } else { + on_finish->complete(-1); + } + + delete req; + return 0; + } + +} // namespace immutable_obj_cache +} // namespace ceph diff --git a/src/tools/immutable_object_cache/CacheClient.h b/src/tools/immutable_object_cache/CacheClient.h new file mode 100644 index 000000000..b2f749631 --- /dev/null +++ b/src/tools/immutable_object_cache/CacheClient.h @@ -0,0 +1,84 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_CACHE_CACHE_CLIENT_H +#define CEPH_CACHE_CACHE_CLIENT_H + +#include <atomic> +#include <boost/asio.hpp> +#include <boost/asio/error.hpp> +#include <boost/algorithm/string.hpp> + +#include "include/ceph_assert.h" +#include "common/ceph_mutex.h" +#include "include/Context.h" +#include "Types.h" +#include "SocketCommon.h" + + +using boost::asio::local::stream_protocol; + +namespace ceph { +namespace immutable_obj_cache { + +class CacheClient { + public: + CacheClient(const std::string& file, CephContext* ceph_ctx); + ~CacheClient(); + void run(); + bool is_session_work(); + void close(); + int stop(); + int connect(); + void connect(Context* on_finish); + void lookup_object(std::string pool_nspace, uint64_t pool_id, + uint64_t snap_id, uint64_t object_size, std::string oid, + CacheGenContextURef&& on_finish); + int register_client(Context* on_finish); + + private: + void send_message(); + void try_send(); + void fault(const int err_type, const boost::system::error_code& err); + void handle_connect(Context* on_finish, const boost::system::error_code& err); + void try_receive(); + void receive_message(); + void process(ObjectCacheRequest* reply, uint64_t seq_id); + void read_reply_header(); + void handle_reply_header(bufferptr bp_head, + const boost::system::error_code& ec, + size_t bytes_transferred); + void read_reply_data(bufferptr&& bp_head, bufferptr&& bp_data, + const uint64_t data_len); + void handle_reply_data(bufferptr bp_head, bufferptr bp_data, + const uint64_t data_len, + const boost::system::error_code& ec, + size_t bytes_transferred); + + private: + CephContext* m_cct; + boost::asio::io_service m_io_service; + boost::asio::io_service::work m_io_service_work; + stream_protocol::socket m_dm_socket; + stream_protocol::endpoint m_ep; + std::shared_ptr<std::thread> m_io_thread; + std::atomic<bool> m_session_work; + + uint64_t m_worker_thread_num; + boost::asio::io_service* m_worker; + std::vector<std::thread*> m_worker_threads; + boost::asio::io_service::work* m_worker_io_service_work; + + std::atomic<bool> m_writing; + std::atomic<bool> m_reading; + std::atomic<uint64_t> m_sequence_id; + ceph::mutex m_lock = + ceph::make_mutex("ceph::cache::cacheclient::m_lock"); + std::map<uint64_t, ObjectCacheRequest*> m_seq_to_req; + bufferlist m_outcoming_bl; + bufferptr m_bp_header; +}; + +} // namespace immutable_obj_cache +} // namespace ceph +#endif // CEPH_CACHE_CACHE_CLIENT_H diff --git a/src/tools/immutable_object_cache/CacheController.cc b/src/tools/immutable_object_cache/CacheController.cc new file mode 100644 index 000000000..ae1636839 --- /dev/null +++ b/src/tools/immutable_object_cache/CacheController.cc @@ -0,0 +1,139 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "CacheController.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_immutable_obj_cache +#undef dout_prefix +#define dout_prefix *_dout << "ceph::cache::CacheController: " << this << " " \ + << __func__ << ": " + +namespace ceph { +namespace immutable_obj_cache { + +CacheController::CacheController(CephContext *cct, + const std::vector<const char*> &args): + m_args(args), m_cct(cct) { + ldout(m_cct, 20) << dendl; +} + +CacheController::~CacheController() { + delete m_cache_server; + delete m_object_cache_store; +} + +int CacheController::init() { + ldout(m_cct, 20) << dendl; + m_object_cache_store = new ObjectCacheStore(m_cct); + // TODO(dehao): make this configurable + int r = m_object_cache_store->init(true); + if (r < 0) { + lderr(m_cct) << "init error\n" << dendl; + return r; + } + + r = m_object_cache_store->init_cache(); + if (r < 0) { + lderr(m_cct) << "init error\n" << dendl; + } + + return r; +} + +int CacheController::shutdown() { + ldout(m_cct, 20) << dendl; + + int r; + if (m_cache_server != nullptr) { + r = m_cache_server->stop(); + if (r < 0) { + lderr(m_cct) << "stop error\n" << dendl; + return r; + } + } + + r = m_object_cache_store->shutdown(); + if (r < 0) { + lderr(m_cct) << "stop error\n" << dendl; + return r; + } + + return r; +} + +void CacheController::handle_signal(int signum) { + shutdown(); +} + +int CacheController::run() { + try { + std::string controller_path = + m_cct->_conf.get_val<std::string>("immutable_object_cache_sock"); + if (controller_path.empty()) { + lderr(m_cct) << "'immutable_object_cache_sock' path not set" << dendl; + return -EINVAL; + } + + std::remove(controller_path.c_str()); + + m_cache_server = new CacheServer(m_cct, controller_path, + std::bind(&CacheController::handle_request, this, + std::placeholders::_1, std::placeholders::_2)); + + int ret = m_cache_server->run(); + if (ret != 0) { + return ret; + } + + return 0; + } catch (std::exception& e) { + lderr(m_cct) << "Exception: " << e.what() << dendl; + return -EFAULT; + } +} + +void CacheController::handle_request(CacheSession* session, + ObjectCacheRequest* req) { + ldout(m_cct, 20) << dendl; + + switch (req->get_request_type()) { + case RBDSC_REGISTER: { + // TODO(dehao): skip register and allow clients to lookup directly + + auto req_reg_data = reinterpret_cast <ObjectCacheRegData*> (req); + session->set_client_version(req_reg_data->version); + + ObjectCacheRequest* reply = new ObjectCacheRegReplyData( + RBDSC_REGISTER_REPLY, req->seq); + session->send(reply); + break; + } + case RBDSC_READ: { + // lookup object in local cache store + std::string cache_path; + ObjectCacheReadData* req_read_data = + reinterpret_cast <ObjectCacheReadData*> (req); + bool return_dne_path = session->client_version().empty(); + int ret = m_object_cache_store->lookup_object( + req_read_data->pool_namespace, req_read_data->pool_id, + req_read_data->snap_id, req_read_data->object_size, + req_read_data->oid, return_dne_path, cache_path); + ObjectCacheRequest* reply = nullptr; + if (ret != OBJ_CACHE_PROMOTED && ret != OBJ_CACHE_DNE) { + reply = new ObjectCacheReadRadosData(RBDSC_READ_RADOS, req->seq); + } else { + reply = new ObjectCacheReadReplyData(RBDSC_READ_REPLY, + req->seq, cache_path); + } + session->send(reply); + break; + } + default: + ldout(m_cct, 5) << "can't recongize request" << dendl; + ceph_assert(0); + } +} + +} // namespace immutable_obj_cache +} // namespace ceph diff --git a/src/tools/immutable_object_cache/CacheController.h b/src/tools/immutable_object_cache/CacheController.h new file mode 100644 index 000000000..f70f6bb1c --- /dev/null +++ b/src/tools/immutable_object_cache/CacheController.h @@ -0,0 +1,40 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_CACHE_CACHE_CONTROLLER_H +#define CEPH_CACHE_CACHE_CONTROLLER_H + +#include "common/ceph_context.h" +#include "common/WorkQueue.h" +#include "CacheServer.h" +#include "ObjectCacheStore.h" + +namespace ceph { +namespace immutable_obj_cache { + +class CacheController { + public: + CacheController(CephContext *cct, const std::vector<const char*> &args); + ~CacheController(); + + int init(); + + int shutdown(); + + void handle_signal(int sinnum); + + int run(); + + void handle_request(CacheSession* session, ObjectCacheRequest* msg); + + private: + CacheServer *m_cache_server = nullptr; + std::vector<const char*> m_args; + CephContext *m_cct; + ObjectCacheStore *m_object_cache_store = nullptr; +}; + +} // namespace immutable_obj_cache +} // namespace ceph + +#endif // CEPH_CACHE_CACHE_CONTROLLER_H diff --git a/src/tools/immutable_object_cache/CacheServer.cc b/src/tools/immutable_object_cache/CacheServer.cc new file mode 100644 index 000000000..e94a47c7a --- /dev/null +++ b/src/tools/immutable_object_cache/CacheServer.cc @@ -0,0 +1,106 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include <boost/bind/bind.hpp> +#include "common/debug.h" +#include "common/ceph_context.h" +#include "CacheServer.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_immutable_obj_cache +#undef dout_prefix +#define dout_prefix *_dout << "ceph::cache::CacheServer: " << this << " " \ + << __func__ << ": " + + +namespace ceph { +namespace immutable_obj_cache { + +CacheServer::CacheServer(CephContext* cct, const std::string& file, + ProcessMsg processmsg) + : cct(cct), m_server_process_msg(processmsg), + m_local_path(file), m_acceptor(m_io_service) {} + +CacheServer::~CacheServer() { + stop(); +} + +int CacheServer::run() { + ldout(cct, 20) << dendl; + + int ret = start_accept(); + if (ret != 0) { + return ret; + } + + boost::system::error_code ec; + ret = m_io_service.run(ec); + if (ec) { + ldout(cct, 1) << "m_io_service run fails: " << ec.message() << dendl; + return -1; + } + return 0; +} + +int CacheServer::stop() { + m_io_service.stop(); + return 0; +} + +int CacheServer::start_accept() { + ldout(cct, 20) << dendl; + + boost::system::error_code ec; + m_acceptor.open(m_local_path.protocol(), ec); + if (ec) { + lderr(cct) << "failed to open domain socket: " << ec.message() << dendl; + return -ec.value(); + } + + m_acceptor.bind(m_local_path, ec); + if (ec) { + lderr(cct) << "failed to bind to domain socket '" + << m_local_path << "': " << ec.message() << dendl; + return -ec.value(); + } + + m_acceptor.listen(boost::asio::socket_base::max_connections, ec); + if (ec) { + lderr(cct) << "failed to listen on domain socket: " << ec.message() + << dendl; + return -ec.value(); + } + + accept(); + return 0; +} + +void CacheServer::accept() { + CacheSessionPtr new_session = nullptr; + + new_session.reset(new CacheSession(m_io_service, + m_server_process_msg, cct)); + + m_acceptor.async_accept(new_session->socket(), + boost::bind(&CacheServer::handle_accept, this, new_session, + boost::asio::placeholders::error)); +} + +void CacheServer::handle_accept(CacheSessionPtr new_session, + const boost::system::error_code& error) { + ldout(cct, 20) << dendl; + if (error) { + // operation_absort + lderr(cct) << "async accept fails : " << error.message() << dendl; + return; + } + + // TODO(dehao) : session setting + new_session->start(); + + // lanuch next accept + accept(); +} + +} // namespace immutable_obj_cache +} // namespace ceph diff --git a/src/tools/immutable_object_cache/CacheServer.h b/src/tools/immutable_object_cache/CacheServer.h new file mode 100644 index 000000000..31d859934 --- /dev/null +++ b/src/tools/immutable_object_cache/CacheServer.h @@ -0,0 +1,45 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_CACHE_CACHE_SERVER_H +#define CEPH_CACHE_CACHE_SERVER_H + +#include <boost/asio.hpp> +#include <boost/asio/error.hpp> + +#include "Types.h" +#include "SocketCommon.h" +#include "CacheSession.h" + + +using boost::asio::local::stream_protocol; + +namespace ceph { +namespace immutable_obj_cache { + +class CacheServer { + public: + CacheServer(CephContext* cct, const std::string& file, ProcessMsg processmsg); + ~CacheServer(); + + int run(); + int start_accept(); + int stop(); + + private: + void accept(); + void handle_accept(CacheSessionPtr new_session, + const boost::system::error_code& error); + + private: + CephContext* cct; + boost::asio::io_service m_io_service; + ProcessMsg m_server_process_msg; + stream_protocol::endpoint m_local_path; + stream_protocol::acceptor m_acceptor; +}; + +} // namespace immutable_obj_cache +} // namespace ceph + +#endif diff --git a/src/tools/immutable_object_cache/CacheSession.cc b/src/tools/immutable_object_cache/CacheSession.cc new file mode 100644 index 000000000..38c38c97d --- /dev/null +++ b/src/tools/immutable_object_cache/CacheSession.cc @@ -0,0 +1,140 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include <boost/bind/bind.hpp> +#include "common/debug.h" +#include "common/ceph_context.h" +#include "CacheSession.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_immutable_obj_cache +#undef dout_prefix +#define dout_prefix *_dout << "ceph::cache::CacheSession: " << this << " " \ + << __func__ << ": " + + +namespace ceph { +namespace immutable_obj_cache { + +CacheSession::CacheSession(io_service& io_service, + ProcessMsg processmsg, + CephContext* cct) + : m_dm_socket(io_service), + m_server_process_msg(processmsg), m_cct(cct) { + m_bp_header = buffer::create(get_header_size()); +} + +CacheSession::~CacheSession() { + close(); +} + +stream_protocol::socket& CacheSession::socket() { + return m_dm_socket; +} + +void CacheSession::set_client_version(const std::string &version) { + m_client_version = version; +} + +const std::string &CacheSession::client_version() const { + return m_client_version; +} + +void CacheSession::close() { + if (m_dm_socket.is_open()) { + boost::system::error_code close_ec; + m_dm_socket.close(close_ec); + if (close_ec) { + ldout(m_cct, 20) << "close: " << close_ec.message() << dendl; + } + } +} + +void CacheSession::start() { + read_request_header(); +} + +void CacheSession::read_request_header() { + ldout(m_cct, 20) << dendl; + boost::asio::async_read(m_dm_socket, + boost::asio::buffer(m_bp_header.c_str(), get_header_size()), + boost::asio::transfer_exactly(get_header_size()), + boost::bind(&CacheSession::handle_request_header, + shared_from_this(), boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred)); +} + +void CacheSession::handle_request_header(const boost::system::error_code& err, + size_t bytes_transferred) { + ldout(m_cct, 20) << dendl; + if (err || bytes_transferred != get_header_size()) { + fault(err); + return; + } + + read_request_data(get_data_len(m_bp_header.c_str())); +} + +void CacheSession::read_request_data(uint64_t data_len) { + ldout(m_cct, 20) << dendl; + bufferptr bp_data(buffer::create(data_len)); + boost::asio::async_read(m_dm_socket, + boost::asio::buffer(bp_data.c_str(), bp_data.length()), + boost::asio::transfer_exactly(data_len), + boost::bind(&CacheSession::handle_request_data, + shared_from_this(), bp_data, data_len, + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred)); +} + +void CacheSession::handle_request_data(bufferptr bp, uint64_t data_len, + const boost::system::error_code& err, + size_t bytes_transferred) { + ldout(m_cct, 20) << dendl; + if (err || bytes_transferred != data_len) { + fault(err); + return; + } + + bufferlist bl_data; + + bl_data.append(m_bp_header); + bl_data.append(std::move(bp)); + + ObjectCacheRequest* req = decode_object_cache_request(bl_data); + + process(req); + delete req; + read_request_header(); +} + +void CacheSession::process(ObjectCacheRequest* req) { + ldout(m_cct, 20) << dendl; + m_server_process_msg(this, req); +} + +void CacheSession::send(ObjectCacheRequest* reply) { + ldout(m_cct, 20) << dendl; + bufferlist bl; + reply->encode(); + bl.append(reply->get_payload_bufferlist()); + + boost::asio::async_write(m_dm_socket, + boost::asio::buffer(bl.c_str(), bl.length()), + boost::asio::transfer_exactly(bl.length()), + [this, bl, reply](const boost::system::error_code& err, + size_t bytes_transferred) { + delete reply; + if (err || bytes_transferred != bl.length()) { + fault(err); + return; + } + }); +} + +void CacheSession::fault(const boost::system::error_code& ec) { + ldout(m_cct, 20) << "session fault : " << ec.message() << dendl; +} + +} // namespace immutable_obj_cache +} // namespace ceph diff --git a/src/tools/immutable_object_cache/CacheSession.h b/src/tools/immutable_object_cache/CacheSession.h new file mode 100644 index 000000000..0826e8a2b --- /dev/null +++ b/src/tools/immutable_object_cache/CacheSession.h @@ -0,0 +1,56 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_CACHE_SESSION_H +#define CEPH_CACHE_SESSION_H + +#include <boost/asio.hpp> +#include <boost/asio/error.hpp> + +#include "Types.h" +#include "SocketCommon.h" + +using boost::asio::local::stream_protocol; +using boost::asio::io_service; + +namespace ceph { +namespace immutable_obj_cache { + +class CacheSession : public std::enable_shared_from_this<CacheSession> { + public: + CacheSession(io_service& io_service, ProcessMsg process_msg, + CephContext* ctx); + ~CacheSession(); + stream_protocol::socket& socket(); + void close(); + void start(); + void read_request_header(); + void handle_request_header(const boost::system::error_code& err, + size_t bytes_transferred); + void read_request_data(uint64_t data_len); + void handle_request_data(bufferptr bp, uint64_t data_len, + const boost::system::error_code& err, + size_t bytes_transferred); + void process(ObjectCacheRequest* req); + void fault(const boost::system::error_code& ec); + void send(ObjectCacheRequest* msg); + + void set_client_version(const std::string &version); + const std::string &client_version() const; + + private: + stream_protocol::socket m_dm_socket; + ProcessMsg m_server_process_msg; + CephContext* m_cct; + + std::string m_client_version; + + bufferptr m_bp_header; +}; + +typedef std::shared_ptr<CacheSession> CacheSessionPtr; + +} // namespace immutable_obj_cache +} // namespace ceph + +#endif // CEPH_CACHE_SESSION_H diff --git a/src/tools/immutable_object_cache/ObjectCacheStore.cc b/src/tools/immutable_object_cache/ObjectCacheStore.cc new file mode 100644 index 000000000..1b1eef1e3 --- /dev/null +++ b/src/tools/immutable_object_cache/ObjectCacheStore.cc @@ -0,0 +1,466 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "ObjectCacheStore.h" +#include "Utils.h" +#if __has_include(<filesystem>) +#include <filesystem> +namespace fs = std::filesystem; +#else +#include <experimental/filesystem> +namespace fs = std::experimental::filesystem; +#endif + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_immutable_obj_cache +#undef dout_prefix +#define dout_prefix *_dout << "ceph::cache::ObjectCacheStore: " << this << " " \ + << __func__ << ": " + + +namespace ceph { +namespace immutable_obj_cache { + +namespace { + +class SafeTimerSingleton : public CommonSafeTimer<ceph::mutex> { +public: + ceph::mutex lock = ceph::make_mutex + ("ceph::immutable_object_cache::SafeTimerSingleton::lock"); + + explicit SafeTimerSingleton(CephContext *cct) + : CommonSafeTimer(cct, lock, true) { + init(); + } + ~SafeTimerSingleton() { + std::lock_guard locker{lock}; + shutdown(); + } +}; + +} // anonymous namespace + +enum ThrottleTargetCode { + ROC_QOS_IOPS_THROTTLE = 1, + ROC_QOS_BPS_THROTTLE = 2 +}; + +ObjectCacheStore::ObjectCacheStore(CephContext *cct) + : m_cct(cct), m_rados(new librados::Rados()) { + + m_cache_root_dir = + m_cct->_conf.get_val<std::string>("immutable_object_cache_path"); + + if (m_cache_root_dir.back() != '/') { + m_cache_root_dir += "/"; + } + + uint64_t cache_max_size = + m_cct->_conf.get_val<Option::size_t>("immutable_object_cache_max_size"); + + double cache_watermark = + m_cct->_conf.get_val<double>("immutable_object_cache_watermark"); + + uint64_t max_inflight_ops = + m_cct->_conf.get_val<uint64_t>("immutable_object_cache_max_inflight_ops"); + + uint64_t limit = 0; + if ((limit = m_cct->_conf.get_val<uint64_t> + ("immutable_object_cache_qos_iops_limit")) != 0) { + apply_qos_tick_and_limit(ROC_QOS_IOPS_THROTTLE, + m_cct->_conf.get_val<std::chrono::milliseconds> + ("immutable_object_cache_qos_schedule_tick_min"), + limit, + m_cct->_conf.get_val<uint64_t> + ("immutable_object_cache_qos_iops_burst"), + m_cct->_conf.get_val<std::chrono::seconds> + ("immutable_object_cache_qos_iops_burst_seconds")); + } + if ((limit = m_cct->_conf.get_val<uint64_t> + ("immutable_object_cache_qos_bps_limit")) != 0) { + apply_qos_tick_and_limit(ROC_QOS_BPS_THROTTLE, + m_cct->_conf.get_val<std::chrono::milliseconds> + ("immutable_object_cache_qos_schedule_tick_min"), + limit, + m_cct->_conf.get_val<uint64_t> + ("immutable_object_cache_qos_bps_burst"), + m_cct->_conf.get_val<std::chrono::seconds> + ("immutable_object_cache_qos_bps_burst_seconds")); + } + + if ((cache_watermark <= 0) || (cache_watermark > 1)) { + lderr(m_cct) << "Invalid water mark provided, set it to default." << dendl; + cache_watermark = 0.9; + } + m_policy = new SimplePolicy(m_cct, cache_max_size, max_inflight_ops, + cache_watermark); +} + +ObjectCacheStore::~ObjectCacheStore() { + delete m_policy; + if (m_qos_enabled_flag & ROC_QOS_IOPS_THROTTLE) { + ceph_assert(m_throttles[ROC_QOS_IOPS_THROTTLE] != nullptr); + delete m_throttles[ROC_QOS_IOPS_THROTTLE]; + } + if (m_qos_enabled_flag & ROC_QOS_BPS_THROTTLE) { + ceph_assert(m_throttles[ROC_QOS_BPS_THROTTLE] != nullptr); + delete m_throttles[ROC_QOS_BPS_THROTTLE]; + } +} + +int ObjectCacheStore::init(bool reset) { + ldout(m_cct, 20) << dendl; + + int ret = m_rados->init_with_context(m_cct); + if (ret < 0) { + lderr(m_cct) << "fail to init Ceph context" << dendl; + return ret; + } + + ret = m_rados->connect(); + if (ret < 0) { + lderr(m_cct) << "fail to connect to cluster" << dendl; + return ret; + } + + // TODO(dehao): fsck and reuse existing cache objects + if (reset) { + try { + if (fs::exists(m_cache_root_dir)) { + // remove all sub folders + for (auto& p : fs::directory_iterator(m_cache_root_dir)) { + fs::remove_all(p.path()); + } + } else { + fs::create_directories(m_cache_root_dir); + } + } catch (const fs::filesystem_error& e) { + lderr(m_cct) << "failed to initialize cache store directory: " + << e.what() << dendl; + return -e.code().value(); + } + } + return 0; +} + +int ObjectCacheStore::shutdown() { + ldout(m_cct, 20) << dendl; + + m_rados->shutdown(); + return 0; +} + +int ObjectCacheStore::init_cache() { + ldout(m_cct, 20) << dendl; + std::string cache_dir = m_cache_root_dir; + + return 0; +} + +int ObjectCacheStore::do_promote(std::string pool_nspace, uint64_t pool_id, + uint64_t snap_id, std::string object_name) { + ldout(m_cct, 20) << "to promote object: " << object_name + << " from pool id: " << pool_id + << " namespace: " << pool_nspace + << " snapshot: " << snap_id << dendl; + + int ret = 0; + std::string cache_file_name = + get_cache_file_name(pool_nspace, pool_id, snap_id, object_name); + librados::IoCtx ioctx; + { + std::lock_guard _locker{m_ioctx_map_lock}; + if (m_ioctx_map.find(pool_id) == m_ioctx_map.end()) { + ret = m_rados->ioctx_create2(pool_id, ioctx); + if (ret < 0) { + lderr(m_cct) << "fail to create ioctx" << dendl; + return ret; + } + m_ioctx_map.emplace(pool_id, ioctx); + } else { + ioctx = m_ioctx_map[pool_id]; + } + } + + ioctx.set_namespace(pool_nspace); + ioctx.snap_set_read(snap_id); + + librados::bufferlist* read_buf = new librados::bufferlist(); + + auto ctx = new LambdaContext([this, read_buf, cache_file_name](int ret) { + handle_promote_callback(ret, read_buf, cache_file_name); + }); + + return promote_object(&ioctx, object_name, read_buf, ctx); +} + +int ObjectCacheStore::handle_promote_callback(int ret, bufferlist* read_buf, + std::string cache_file_name) { + ldout(m_cct, 20) << " cache_file_name: " << cache_file_name << dendl; + + // rados read error + if (ret != -ENOENT && ret < 0) { + lderr(m_cct) << "fail to read from rados" << dendl; + + m_policy->update_status(cache_file_name, OBJ_CACHE_NONE); + delete read_buf; + return ret; + } + + auto state = OBJ_CACHE_PROMOTED; + if (ret == -ENOENT) { + // object is empty + state = OBJ_CACHE_DNE; + ret = 0; + } else { + std::string cache_file_path = get_cache_file_path(cache_file_name, true); + if (cache_file_path == "") { + lderr(m_cct) << "fail to write cache file" << dendl; + m_policy->update_status(cache_file_name, OBJ_CACHE_NONE); + delete read_buf; + return -ENOSPC; + } + + ret = read_buf->write_file(cache_file_path.c_str()); + if (ret < 0) { + lderr(m_cct) << "fail to write cache file" << dendl; + + m_policy->update_status(cache_file_name, OBJ_CACHE_NONE); + delete read_buf; + return ret; + } + } + + // update metadata + ceph_assert(OBJ_CACHE_SKIP == m_policy->get_status(cache_file_name)); + m_policy->update_status(cache_file_name, state, read_buf->length()); + ceph_assert(state == m_policy->get_status(cache_file_name)); + + delete read_buf; + + evict_objects(); + + return ret; +} + +int ObjectCacheStore::lookup_object(std::string pool_nspace, uint64_t pool_id, + uint64_t snap_id, uint64_t object_size, + std::string object_name, + bool return_dne_path, + std::string& target_cache_file_path) { + ldout(m_cct, 20) << "object name = " << object_name + << " in pool ID : " << pool_id << dendl; + + int pret = -1; + std::string cache_file_name = + get_cache_file_name(pool_nspace, pool_id, snap_id, object_name); + + cache_status_t ret = m_policy->lookup_object(cache_file_name); + + switch (ret) { + case OBJ_CACHE_NONE: { + if (take_token_from_throttle(object_size, 1)) { + pret = do_promote(pool_nspace, pool_id, snap_id, object_name); + if (pret < 0) { + lderr(m_cct) << "fail to start promote" << dendl; + } + } else { + m_policy->update_status(cache_file_name, OBJ_CACHE_NONE); + } + return ret; + } + case OBJ_CACHE_PROMOTED: + target_cache_file_path = get_cache_file_path(cache_file_name); + return ret; + case OBJ_CACHE_DNE: + if (return_dne_path) { + target_cache_file_path = get_cache_file_path(cache_file_name); + } + return ret; + case OBJ_CACHE_SKIP: + return ret; + default: + lderr(m_cct) << "unrecognized object cache status" << dendl; + ceph_assert(0); + } +} + +int ObjectCacheStore::promote_object(librados::IoCtx* ioctx, + std::string object_name, + librados::bufferlist* read_buf, + Context* on_finish) { + ldout(m_cct, 20) << "object name = " << object_name << dendl; + + librados::AioCompletion* read_completion = create_rados_callback(on_finish); + // issue a zero-sized read req to get the entire obj + int ret = ioctx->aio_read(object_name, read_completion, read_buf, 0, 0); + if (ret < 0) { + lderr(m_cct) << "failed to read from rados" << dendl; + } + read_completion->release(); + + return ret; +} + +int ObjectCacheStore::evict_objects() { + ldout(m_cct, 20) << dendl; + + std::list<std::string> obj_list; + m_policy->get_evict_list(&obj_list); + for (auto& obj : obj_list) { + do_evict(obj); + } + return 0; +} + +int ObjectCacheStore::do_evict(std::string cache_file) { + ldout(m_cct, 20) << "file = " << cache_file << dendl; + + if (cache_file == "") { + return 0; + } + + std::string cache_file_path = get_cache_file_path(cache_file); + + ldout(m_cct, 20) << "evict cache: " << cache_file_path << dendl; + + // TODO(dehao): possible race on read? + int ret = std::remove(cache_file_path.c_str()); + // evict metadata + if (ret == 0) { + m_policy->update_status(cache_file, OBJ_CACHE_SKIP); + m_policy->evict_entry(cache_file); + } + + return ret; +} + +std::string ObjectCacheStore::get_cache_file_name(std::string pool_nspace, + uint64_t pool_id, + uint64_t snap_id, + std::string oid) { + return pool_nspace + ":" + std::to_string(pool_id) + ":" + + std::to_string(snap_id) + ":" + oid; +} + +std::string ObjectCacheStore::get_cache_file_path(std::string cache_file_name, + bool mkdir) { + ldout(m_cct, 20) << cache_file_name <<dendl; + + uint32_t crc = 0; + crc = ceph_crc32c(0, (unsigned char *)cache_file_name.c_str(), + cache_file_name.length()); + + std::string cache_file_dir = std::to_string(crc % 100) + "/"; + + if (mkdir) { + ldout(m_cct, 20) << "creating cache dir: " << cache_file_dir <<dendl; + std::error_code ec; + std::string new_dir = m_cache_root_dir + cache_file_dir; + if (fs::exists(new_dir, ec)) { + ldout(m_cct, 20) << "cache dir exists: " << cache_file_dir <<dendl; + return new_dir + cache_file_name; + } + + if (!fs::create_directories(new_dir, ec)) { + ldout(m_cct, 5) << "fail to create cache dir: " << new_dir + << "error: " << ec.message() << dendl; + return ""; + } + } + + return m_cache_root_dir + cache_file_dir + cache_file_name; +} + +void ObjectCacheStore::handle_throttle_ready(uint64_t tokens, uint64_t type) { + m_io_throttled = false; + std::lock_guard lock(m_throttle_lock); + if (type & ROC_QOS_IOPS_THROTTLE){ + m_iops_tokens += tokens; + } else if (type & ROC_QOS_BPS_THROTTLE){ + m_bps_tokens += tokens; + } else { + lderr(m_cct) << "unknow throttle type." << dendl; + } +} + +bool ObjectCacheStore::take_token_from_throttle(uint64_t object_size, + uint64_t object_num) { + if (m_io_throttled == true) { + return false; + } + + int flag = 0; + bool wait = false; + if (!wait && (m_qos_enabled_flag & ROC_QOS_IOPS_THROTTLE)) { + std::lock_guard lock(m_throttle_lock); + if (object_num > m_iops_tokens) { + wait = m_throttles[ROC_QOS_IOPS_THROTTLE]->get(object_num, this, + &ObjectCacheStore::handle_throttle_ready, object_num, + ROC_QOS_IOPS_THROTTLE); + } else { + m_iops_tokens -= object_num; + flag = 1; + } + } + if (!wait && (m_qos_enabled_flag & ROC_QOS_BPS_THROTTLE)) { + std::lock_guard lock(m_throttle_lock); + if (object_size > m_bps_tokens) { + wait = m_throttles[ROC_QOS_BPS_THROTTLE]->get(object_size, this, + &ObjectCacheStore::handle_throttle_ready, object_size, + ROC_QOS_BPS_THROTTLE); + } else { + m_bps_tokens -= object_size; + } + } + + if (wait) { + m_io_throttled = true; + // when passing iops throttle, but limit in bps throttle, recovery + if (flag == 1) { + std::lock_guard lock(m_throttle_lock); + m_iops_tokens += object_num; + } + } + + return !wait; +} + +static const std::map<uint64_t, std::string> THROTTLE_FLAGS = { + { ROC_QOS_IOPS_THROTTLE, "roc_qos_iops_throttle" }, + { ROC_QOS_BPS_THROTTLE, "roc_qos_bps_throttle" } +}; + +void ObjectCacheStore::apply_qos_tick_and_limit( + const uint64_t flag, + std::chrono::milliseconds min_tick, + uint64_t limit, + uint64_t burst, + std::chrono::seconds burst_seconds) { + SafeTimerSingleton* safe_timer_singleton = nullptr; + TokenBucketThrottle* throttle = nullptr; + safe_timer_singleton = + &m_cct->lookup_or_create_singleton_object<SafeTimerSingleton>( + "tools::immutable_object_cache", false, m_cct); + SafeTimer* timer = safe_timer_singleton; + ceph::mutex* timer_lock = &safe_timer_singleton->lock; + m_qos_enabled_flag |= flag; + auto throttle_flags_it = THROTTLE_FLAGS.find(flag); + ceph_assert(throttle_flags_it != THROTTLE_FLAGS.end()); + throttle = new TokenBucketThrottle(m_cct, throttle_flags_it->second, + 0, 0, timer, timer_lock); + throttle->set_schedule_tick_min(min_tick.count()); + int ret = throttle->set_limit(limit, burst, burst_seconds.count()); + if (ret < 0) { + lderr(m_cct) << throttle->get_name() << ": invalid qos parameter: " + << "burst(" << burst << ") is less than " + << "limit(" << limit << ")" << dendl; + throttle->set_limit(limit, 0, 1); + } + + ceph_assert(m_throttles.find(flag) == m_throttles.end()); + m_throttles.insert({flag, throttle}); +} + +} // namespace immutable_obj_cache +} // namespace ceph diff --git a/src/tools/immutable_object_cache/ObjectCacheStore.h b/src/tools/immutable_object_cache/ObjectCacheStore.h new file mode 100644 index 000000000..51e5a77b8 --- /dev/null +++ b/src/tools/immutable_object_cache/ObjectCacheStore.h @@ -0,0 +1,85 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_CACHE_OBJECT_CACHE_STORE_H +#define CEPH_CACHE_OBJECT_CACHE_STORE_H + +#include "common/ceph_context.h" +#include "common/ceph_mutex.h" +#include "common/Timer.h" +#include "common/Throttle.h" +#include "common/Cond.h" +#include "include/rados/librados.hpp" + +#include "SimplePolicy.h" + + +using librados::Rados; +using librados::IoCtx; +class Context; + +namespace ceph { +namespace immutable_obj_cache { + +typedef shared_ptr<librados::Rados> RadosRef; +typedef shared_ptr<librados::IoCtx> IoCtxRef; + +class ObjectCacheStore { + public: + ObjectCacheStore(CephContext *cct); + ~ObjectCacheStore(); + int init(bool reset); + int shutdown(); + int init_cache(); + int lookup_object(std::string pool_nspace, + uint64_t pool_id, uint64_t snap_id, + uint64_t object_size, + std::string object_name, + bool return_dne_path, + std::string& target_cache_file_path); + private: + enum ThrottleTypeCode { + THROTTLE_CODE_BYTE, + THROTTLE_CODE_OBJECT + }; + + std::string get_cache_file_name(std::string pool_nspace, uint64_t pool_id, + uint64_t snap_id, std::string oid); + std::string get_cache_file_path(std::string cache_file_name, + bool mkdir = false); + int evict_objects(); + int do_promote(std::string pool_nspace, uint64_t pool_id, + uint64_t snap_id, std::string object_name); + int promote_object(librados::IoCtx*, std::string object_name, + librados::bufferlist* read_buf, + Context* on_finish); + int handle_promote_callback(int, bufferlist*, std::string); + int do_evict(std::string cache_file); + + bool take_token_from_throttle(uint64_t object_size, uint64_t object_num); + void handle_throttle_ready(uint64_t tokens, uint64_t type); + void apply_qos_tick_and_limit(const uint64_t flag, + std::chrono::milliseconds min_tick, + uint64_t limit, uint64_t burst, + std::chrono::seconds burst_seconds); + + CephContext *m_cct; + RadosRef m_rados; + std::map<uint64_t, librados::IoCtx> m_ioctx_map; + ceph::mutex m_ioctx_map_lock = + ceph::make_mutex("ceph::cache::ObjectCacheStore::m_ioctx_map_lock"); + Policy* m_policy; + std::string m_cache_root_dir; + // throttle mechanism + uint64_t m_qos_enabled_flag{0}; + std::map<uint64_t, TokenBucketThrottle*> m_throttles; + bool m_io_throttled{false}; + ceph::mutex m_throttle_lock = + ceph::make_mutex("ceph::cache::ObjectCacheStore::m_throttle_lock");; + uint64_t m_iops_tokens{0}; + uint64_t m_bps_tokens{0}; +}; + +} // namespace immutable_obj_cache +} // ceph +#endif // CEPH_CACHE_OBJECT_CACHE_STORE_H diff --git a/src/tools/immutable_object_cache/Policy.h b/src/tools/immutable_object_cache/Policy.h new file mode 100644 index 000000000..7924a8919 --- /dev/null +++ b/src/tools/immutable_object_cache/Policy.h @@ -0,0 +1,34 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_CACHE_POLICY_H +#define CEPH_CACHE_POLICY_H + +#include <list> +#include <string> + +namespace ceph { +namespace immutable_obj_cache { + +typedef enum { + OBJ_CACHE_NONE = 0, + OBJ_CACHE_PROMOTED, + OBJ_CACHE_SKIP, + OBJ_CACHE_DNE, +} cache_status_t; + +class Policy { + public: + Policy() {} + virtual ~Policy() {} + virtual cache_status_t lookup_object(std::string) = 0; + virtual int evict_entry(std::string) = 0; + virtual void update_status(std::string, cache_status_t, + uint64_t size = 0) = 0; + virtual cache_status_t get_status(std::string) = 0; + virtual void get_evict_list(std::list<std::string>* obj_list) = 0; +}; + +} // namespace immutable_obj_cache +} // namespace ceph +#endif diff --git a/src/tools/immutable_object_cache/SimplePolicy.cc b/src/tools/immutable_object_cache/SimplePolicy.cc new file mode 100644 index 000000000..3a7375ba9 --- /dev/null +++ b/src/tools/immutable_object_cache/SimplePolicy.cc @@ -0,0 +1,216 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/debug.h" +#include "SimplePolicy.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_immutable_obj_cache +#undef dout_prefix +#define dout_prefix *_dout << "ceph::cache::SimplePolicy: " << this << " " \ + << __func__ << ": " + +namespace ceph { +namespace immutable_obj_cache { + +SimplePolicy::SimplePolicy(CephContext *cct, uint64_t cache_size, + uint64_t max_inflight, double watermark) + : cct(cct), m_watermark(watermark), m_max_inflight_ops(max_inflight), + m_max_cache_size(cache_size) { + + ldout(cct, 20) << "max cache size= " << m_max_cache_size + << " ,watermark= " << m_watermark + << " ,max inflight ops= " << m_max_inflight_ops << dendl; + + m_cache_size = 0; + +} + +SimplePolicy::~SimplePolicy() { + ldout(cct, 20) << dendl; + + for (auto it : m_cache_map) { + Entry* entry = (it.second); + delete entry; + } +} + +cache_status_t SimplePolicy::alloc_entry(std::string file_name) { + ldout(cct, 20) << "alloc entry for: " << file_name << dendl; + + std::unique_lock wlocker{m_cache_map_lock}; + + // cache hit when promoting + if (m_cache_map.find(file_name) != m_cache_map.end()) { + ldout(cct, 20) << "object is under promoting: " << file_name << dendl; + return OBJ_CACHE_SKIP; + } + + if ((m_cache_size < m_max_cache_size) && + (inflight_ops < m_max_inflight_ops)) { + Entry* entry = new Entry(); + ceph_assert(entry != nullptr); + m_cache_map[file_name] = entry; + wlocker.unlock(); + update_status(file_name, OBJ_CACHE_SKIP); + return OBJ_CACHE_NONE; // start promotion request + } + + // if there's no free entry, return skip to read from rados + return OBJ_CACHE_SKIP; +} + +cache_status_t SimplePolicy::lookup_object(std::string file_name) { + ldout(cct, 20) << "lookup: " << file_name << dendl; + + std::shared_lock rlocker{m_cache_map_lock}; + + auto entry_it = m_cache_map.find(file_name); + // simply promote on first lookup + if (entry_it == m_cache_map.end()) { + rlocker.unlock(); + return alloc_entry(file_name); + } + + Entry* entry = entry_it->second; + + if (entry->status == OBJ_CACHE_PROMOTED || entry->status == OBJ_CACHE_DNE) { + // bump pos in lru on hit + m_promoted_lru.lru_touch(entry); + } + + return entry->status; +} + +void SimplePolicy::update_status(std::string file_name, + cache_status_t new_status, uint64_t size) { + ldout(cct, 20) << "update status for: " << file_name + << " new status = " << new_status << dendl; + + std::unique_lock locker{m_cache_map_lock}; + + auto entry_it = m_cache_map.find(file_name); + if (entry_it == m_cache_map.end()) { + return; + } + + ceph_assert(entry_it != m_cache_map.end()); + Entry* entry = entry_it->second; + + // to promote + if (entry->status == OBJ_CACHE_NONE && new_status== OBJ_CACHE_SKIP) { + entry->status = new_status; + entry->file_name = file_name; + inflight_ops++; + return; + } + + // promoting done + if (entry->status == OBJ_CACHE_SKIP && (new_status== OBJ_CACHE_PROMOTED || + new_status== OBJ_CACHE_DNE)) { + m_promoted_lru.lru_insert_top(entry); + entry->status = new_status; + entry->size = size; + m_cache_size += entry->size; + inflight_ops--; + return; + } + + // promoting failed + if (entry->status == OBJ_CACHE_SKIP && new_status== OBJ_CACHE_NONE) { + // mark this entry as free + entry->file_name = ""; + entry->status = new_status; + + m_cache_map.erase(entry_it); + inflight_ops--; + delete entry; + return; + } + + // to evict + if ((entry->status == OBJ_CACHE_PROMOTED || entry->status == OBJ_CACHE_DNE) && + new_status== OBJ_CACHE_NONE) { + // mark this entry as free + uint64_t size = entry->size; + entry->file_name = ""; + entry->size = 0; + entry->status = new_status; + + m_promoted_lru.lru_remove(entry); + m_cache_map.erase(entry_it); + m_cache_size -= size; + delete entry; + return; + } +} + +int SimplePolicy::evict_entry(std::string file_name) { + ldout(cct, 20) << "to evict: " << file_name << dendl; + + update_status(file_name, OBJ_CACHE_NONE); + + return 0; +} + +cache_status_t SimplePolicy::get_status(std::string file_name) { + ldout(cct, 20) << file_name << dendl; + + std::shared_lock locker{m_cache_map_lock}; + auto entry_it = m_cache_map.find(file_name); + if (entry_it == m_cache_map.end()) { + return OBJ_CACHE_NONE; + } + + return entry_it->second->status; +} + +void SimplePolicy::get_evict_list(std::list<std::string>* obj_list) { + ldout(cct, 20) << dendl; + + std::unique_lock locker{m_cache_map_lock}; + // check free ratio, pop entries from LRU + if ((double)m_cache_size > m_max_cache_size * m_watermark) { + // TODO(dehao): make this configurable + int evict_num = m_cache_map.size() * 0.1; + for (int i = 0; i < evict_num; i++) { + Entry* entry = reinterpret_cast<Entry*>(m_promoted_lru.lru_expire()); + if (entry == nullptr) { + continue; + } + std::string file_name = entry->file_name; + obj_list->push_back(file_name); + } + } +} + +// for unit test +uint64_t SimplePolicy::get_free_size() { + return m_max_cache_size - m_cache_size; +} + +uint64_t SimplePolicy::get_promoting_entry_num() { + uint64_t index = 0; + std::shared_lock rlocker{m_cache_map_lock}; + for (auto it : m_cache_map) { + if (it.second->status == OBJ_CACHE_SKIP) { + index++; + } + } + return index; +} + +uint64_t SimplePolicy::get_promoted_entry_num() { + return m_promoted_lru.lru_get_size(); +} + +std::string SimplePolicy::get_evict_entry() { + Entry* entry = reinterpret_cast<Entry*>(m_promoted_lru.lru_get_next_expire()); + if (entry == nullptr) { + return ""; + } + return entry->file_name; +} + +} // namespace immutable_obj_cache +} // namespace ceph diff --git a/src/tools/immutable_object_cache/SimplePolicy.h b/src/tools/immutable_object_cache/SimplePolicy.h new file mode 100644 index 000000000..671cbd518 --- /dev/null +++ b/src/tools/immutable_object_cache/SimplePolicy.h @@ -0,0 +1,68 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_CACHE_SIMPLE_POLICY_H +#define CEPH_CACHE_SIMPLE_POLICY_H + +#include "common/ceph_context.h" +#include "common/ceph_mutex.h" +#include "include/lru.h" +#include "Policy.h" + +#include <unordered_map> +#include <string> + +namespace ceph { +namespace immutable_obj_cache { + +class SimplePolicy : public Policy { + public: + SimplePolicy(CephContext *cct, uint64_t block_num, uint64_t max_inflight, + double watermark); + ~SimplePolicy(); + + cache_status_t lookup_object(std::string file_name); + cache_status_t get_status(std::string file_name); + + void update_status(std::string file_name, + cache_status_t new_status, + uint64_t size = 0); + + int evict_entry(std::string file_name); + + void get_evict_list(std::list<std::string>* obj_list); + + uint64_t get_free_size(); + uint64_t get_promoting_entry_num(); + uint64_t get_promoted_entry_num(); + std::string get_evict_entry(); + + private: + cache_status_t alloc_entry(std::string file_name); + + class Entry : public LRUObject { + public: + cache_status_t status; + Entry() : status(OBJ_CACHE_NONE) {} + std::string file_name; + uint64_t size; + }; + + CephContext* cct; + double m_watermark; + uint64_t m_max_inflight_ops; + uint64_t m_max_cache_size; + std::atomic<uint64_t> inflight_ops = 0; + + std::unordered_map<std::string, Entry*> m_cache_map; + ceph::shared_mutex m_cache_map_lock = + ceph::make_shared_mutex("rbd::cache::SimplePolicy::m_cache_map_lock"); + + std::atomic<uint64_t> m_cache_size; + + LRU m_promoted_lru; +}; + +} // namespace immutable_obj_cache +} // namespace ceph +#endif // CEPH_CACHE_SIMPLE_POLICY_H diff --git a/src/tools/immutable_object_cache/SocketCommon.h b/src/tools/immutable_object_cache/SocketCommon.h new file mode 100644 index 000000000..99acf3609 --- /dev/null +++ b/src/tools/immutable_object_cache/SocketCommon.h @@ -0,0 +1,31 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_CACHE_SOCKET_COMMON_H +#define CEPH_CACHE_SOCKET_COMMON_H + +namespace ceph { +namespace immutable_obj_cache { + +static const int RBDSC_REGISTER = 0X11; +static const int RBDSC_READ = 0X12; +static const int RBDSC_REGISTER_REPLY = 0X13; +static const int RBDSC_READ_REPLY = 0X14; +static const int RBDSC_READ_RADOS = 0X15; + +static const int ASIO_ERROR_READ = 0X01; +static const int ASIO_ERROR_WRITE = 0X02; +static const int ASIO_ERROR_CONNECT = 0X03; +static const int ASIO_ERROR_ACCEPT = 0X04; +static const int ASIO_ERROR_MSG_INCOMPLETE = 0X05; + +class ObjectCacheRequest; +class CacheSession; + +typedef GenContextURef<ObjectCacheRequest*> CacheGenContextURef; + +typedef std::function<void(CacheSession*, ObjectCacheRequest*)> ProcessMsg; + +} // namespace immutable_obj_cache +} // namespace ceph +#endif // CEPH_CACHE_SOCKET_COMMON_H diff --git a/src/tools/immutable_object_cache/Types.cc b/src/tools/immutable_object_cache/Types.cc new file mode 100644 index 000000000..860017d6a --- /dev/null +++ b/src/tools/immutable_object_cache/Types.cc @@ -0,0 +1,184 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "Types.h" +#include "SocketCommon.h" + +#define dout_subsys ceph_subsys_immutable_obj_cache +#undef dout_prefix +#define dout_prefix *_dout << "ceph::cache::Types: " << __func__ << ": " + +namespace ceph { +namespace immutable_obj_cache { + +ObjectCacheRequest::ObjectCacheRequest() {} +ObjectCacheRequest::ObjectCacheRequest(uint16_t t, uint64_t s) + : type(t), seq(s) {} +ObjectCacheRequest::~ObjectCacheRequest() {} + +void ObjectCacheRequest::encode() { + ENCODE_START(2, 1, payload); + ceph::encode(type, payload); + ceph::encode(seq, payload); + if (!payload_empty()) { + encode_payload(); + } + ENCODE_FINISH(payload); +} + +void ObjectCacheRequest::decode(bufferlist& bl) { + auto i = bl.cbegin(); + DECODE_START(2, i); + ceph::decode(type, i); + ceph::decode(seq, i); + if (!payload_empty()) { + decode_payload(i, struct_v); + } + DECODE_FINISH(i); +} + +ObjectCacheRegData::ObjectCacheRegData() {} +ObjectCacheRegData::ObjectCacheRegData(uint16_t t, uint64_t s) + : ObjectCacheRequest(t, s) {} +ObjectCacheRegData::ObjectCacheRegData(uint16_t t, uint64_t s, + const std::string &version) + : ObjectCacheRequest(t, s), + version(version) { +} + +ObjectCacheRegData::~ObjectCacheRegData() {} + +void ObjectCacheRegData::encode_payload() { + ceph::encode(version, payload); +} + +void ObjectCacheRegData::decode_payload(bufferlist::const_iterator i, + __u8 encode_version) { + if (i.end()) { + return; + } + ceph::decode(version, i); +} + +ObjectCacheRegReplyData::ObjectCacheRegReplyData() {} +ObjectCacheRegReplyData::ObjectCacheRegReplyData(uint16_t t, uint64_t s) + : ObjectCacheRequest(t, s) {} + +ObjectCacheRegReplyData::~ObjectCacheRegReplyData() {} + +void ObjectCacheRegReplyData::encode_payload() {} + +void ObjectCacheRegReplyData::decode_payload(bufferlist::const_iterator bl, + __u8 encode_version) {} + +ObjectCacheReadData::ObjectCacheReadData(uint16_t t, uint64_t s, + uint64_t read_offset, + uint64_t read_len, + uint64_t pool_id, uint64_t snap_id, + uint64_t object_size, + std::string oid, + std::string pool_namespace) + : ObjectCacheRequest(t, s), read_offset(read_offset), + read_len(read_len), pool_id(pool_id), snap_id(snap_id), + object_size(object_size), oid(oid), pool_namespace(pool_namespace) +{} + +ObjectCacheReadData::ObjectCacheReadData(uint16_t t, uint64_t s) + : ObjectCacheRequest(t, s) {} + +ObjectCacheReadData::~ObjectCacheReadData() {} + +void ObjectCacheReadData::encode_payload() { + ceph::encode(read_offset, payload); + ceph::encode(read_len, payload); + ceph::encode(pool_id, payload); + ceph::encode(snap_id, payload); + ceph::encode(oid, payload); + ceph::encode(pool_namespace, payload); + ceph::encode(object_size, payload); +} + +void ObjectCacheReadData::decode_payload(bufferlist::const_iterator i, + __u8 encode_version) { + ceph::decode(read_offset, i); + ceph::decode(read_len, i); + ceph::decode(pool_id, i); + ceph::decode(snap_id, i); + ceph::decode(oid, i); + ceph::decode(pool_namespace, i); + if (encode_version >= 2) { + ceph::decode(object_size, i); + } +} + +ObjectCacheReadReplyData::ObjectCacheReadReplyData(uint16_t t, uint64_t s, + string cache_path) + : ObjectCacheRequest(t, s), cache_path(cache_path) {} +ObjectCacheReadReplyData::ObjectCacheReadReplyData(uint16_t t, uint64_t s) + : ObjectCacheRequest(t, s) {} + +ObjectCacheReadReplyData::~ObjectCacheReadReplyData() {} + +void ObjectCacheReadReplyData::encode_payload() { + ceph::encode(cache_path, payload); +} + +void ObjectCacheReadReplyData::decode_payload(bufferlist::const_iterator i, + __u8 encode_version) { + ceph::decode(cache_path, i); +} + +ObjectCacheReadRadosData::ObjectCacheReadRadosData() {} +ObjectCacheReadRadosData::ObjectCacheReadRadosData(uint16_t t, uint64_t s) + : ObjectCacheRequest(t, s) {} + +ObjectCacheReadRadosData::~ObjectCacheReadRadosData() {} + +void ObjectCacheReadRadosData::encode_payload() {} + +void ObjectCacheReadRadosData::decode_payload(bufferlist::const_iterator i, + __u8 encode_version) {} + +ObjectCacheRequest* decode_object_cache_request(bufferlist payload_buffer) { + ObjectCacheRequest* req = nullptr; + + uint16_t type; + uint64_t seq; + auto i = payload_buffer.cbegin(); + DECODE_START(1, i); + ceph::decode(type, i); + ceph::decode(seq, i); + DECODE_FINISH(i); + + switch (type) { + case RBDSC_REGISTER: { + req = new ObjectCacheRegData(type, seq); + break; + } + case RBDSC_READ: { + req = new ObjectCacheReadData(type, seq); + break; + } + case RBDSC_REGISTER_REPLY: { + req = new ObjectCacheRegReplyData(type, seq); + break; + } + case RBDSC_READ_REPLY: { + req = new ObjectCacheReadReplyData(type, seq); + break; + } + case RBDSC_READ_RADOS: { + req = new ObjectCacheReadRadosData(type, seq); + break; + } + default: + ceph_assert(0); + } + + req->decode(payload_buffer); + + return req; +} + +} // namespace immutable_obj_cache +} // namespace ceph diff --git a/src/tools/immutable_object_cache/Types.h b/src/tools/immutable_object_cache/Types.h new file mode 100644 index 000000000..05394d843 --- /dev/null +++ b/src/tools/immutable_object_cache/Types.h @@ -0,0 +1,136 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_CACHE_TYPES_H +#define CEPH_CACHE_TYPES_H + +#include "include/encoding.h" +#include "include/Context.h" +#include "SocketCommon.h" + +namespace ceph { +namespace immutable_obj_cache { + +namespace { +struct HeaderHelper { + uint8_t v; + uint8_t c_v; + ceph_le32 len; +}__attribute__((packed)); + +inline uint8_t get_header_size() { + return sizeof(HeaderHelper); +} + +inline uint32_t get_data_len(char* buf) { + HeaderHelper* header = reinterpret_cast<HeaderHelper*>(buf); + return header->len; +} +} // namespace + +class ObjectCacheRequest { + public: + uint16_t type; + uint64_t seq; + + bufferlist payload; + + CacheGenContextURef process_msg; + + ObjectCacheRequest(); + ObjectCacheRequest(uint16_t type, uint64_t seq); + virtual ~ObjectCacheRequest(); + + // encode consists of two steps + // step 1 : directly encode common bits using encode method of base classs. + // step 2 : according to payload_empty, determine whether addtional bits + // need to be encoded which be implements by child class. + void encode(); + void decode(bufferlist& bl); + bufferlist get_payload_bufferlist() { return payload; } + + virtual void encode_payload() = 0; + virtual void decode_payload(bufferlist::const_iterator bl_it, + __u8 encode_version) = 0; + virtual uint16_t get_request_type() = 0; + virtual bool payload_empty() = 0; +}; + +class ObjectCacheRegData : public ObjectCacheRequest { + public: + std::string version; + ObjectCacheRegData(); + ObjectCacheRegData(uint16_t t, uint64_t s, const std::string &version); + ObjectCacheRegData(uint16_t t, uint64_t s); + ~ObjectCacheRegData() override; + void encode_payload() override; + void decode_payload(bufferlist::const_iterator bl, + __u8 encode_version) override; + uint16_t get_request_type() override { return RBDSC_REGISTER; } + bool payload_empty() override { return false; } +}; + +class ObjectCacheRegReplyData : public ObjectCacheRequest { + public: + ObjectCacheRegReplyData(); + ObjectCacheRegReplyData(uint16_t t, uint64_t s); + ~ObjectCacheRegReplyData() override; + void encode_payload() override; + void decode_payload(bufferlist::const_iterator iter, + __u8 encode_version) override; + uint16_t get_request_type() override { return RBDSC_REGISTER_REPLY; } + bool payload_empty() override { return true; } +}; + +class ObjectCacheReadData : public ObjectCacheRequest { + public: + uint64_t read_offset; + uint64_t read_len; + uint64_t pool_id; + uint64_t snap_id; + uint64_t object_size = 0; + std::string oid; + std::string pool_namespace; + ObjectCacheReadData(uint16_t t, uint64_t s, uint64_t read_offset, + uint64_t read_len, uint64_t pool_id, + uint64_t snap_id, uint64_t object_size, + std::string oid, std::string pool_namespace); + ObjectCacheReadData(uint16_t t, uint64_t s); + ~ObjectCacheReadData() override; + void encode_payload() override; + void decode_payload(bufferlist::const_iterator bl, + __u8 encode_version) override; + uint16_t get_request_type() override { return RBDSC_READ; } + bool payload_empty() override { return false; } +}; + +class ObjectCacheReadReplyData : public ObjectCacheRequest { + public: + std::string cache_path; + ObjectCacheReadReplyData(uint16_t t, uint64_t s, std::string cache_path); + ObjectCacheReadReplyData(uint16_t t, uint64_t s); + ~ObjectCacheReadReplyData() override; + void encode_payload() override; + void decode_payload(bufferlist::const_iterator bl, + __u8 encode_version) override; + uint16_t get_request_type() override { return RBDSC_READ_REPLY; } + bool payload_empty() override { return false; } +}; + +class ObjectCacheReadRadosData : public ObjectCacheRequest { + public: + ObjectCacheReadRadosData(); + ObjectCacheReadRadosData(uint16_t t, uint64_t s); + ~ObjectCacheReadRadosData() override; + void encode_payload() override; + void decode_payload(bufferlist::const_iterator bl, + __u8 encode_version) override; + uint16_t get_request_type() override { return RBDSC_READ_RADOS; } + bool payload_empty() override { return true; } +}; + +ObjectCacheRequest* decode_object_cache_request(bufferlist payload_buffer); + +} // namespace immutable_obj_cache +} // namespace ceph +#endif // CEPH_CACHE_TYPES_H diff --git a/src/tools/immutable_object_cache/Utils.h b/src/tools/immutable_object_cache/Utils.h new file mode 100644 index 000000000..3c68cfa7b --- /dev/null +++ b/src/tools/immutable_object_cache/Utils.h @@ -0,0 +1,31 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_CACHE_UTILS_H +#define CEPH_CACHE_UTILS_H + +#include "include/rados/librados.hpp" +#include "include/Context.h" + +namespace ceph { +namespace immutable_obj_cache { +namespace detail { + +template <typename T, void(T::*MF)(int)> +void rados_callback(rados_completion_t c, void *arg) { + T *obj = reinterpret_cast<T*>(arg); + int r = rados_aio_get_return_value(c); + (obj->*MF)(r); +} + +} // namespace detail + +template <typename T, void(T::*MF)(int)=&T::complete> +librados::AioCompletion *create_rados_callback(T *obj) { + return librados::Rados::aio_create_completion( + obj, &detail::rados_callback<T, MF>); +} + +} // namespace immutable_obj_cache +} // namespace ceph +#endif // CEPH_CACHE_UTILS_H diff --git a/src/tools/immutable_object_cache/main.cc b/src/tools/immutable_object_cache/main.cc new file mode 100644 index 000000000..55b0d087a --- /dev/null +++ b/src/tools/immutable_object_cache/main.cc @@ -0,0 +1,84 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/ceph_argparse.h" +#include "common/config.h" +#include "common/debug.h" +#include "common/errno.h" +#include "global/global_init.h" +#include "global/signal_handler.h" +#include "CacheController.h" + +#include <vector> + +ceph::immutable_obj_cache::CacheController *cachectl = nullptr; + +void usage() { + std::cout << "usage: ceph-immutable-object-cache [options...]" << std::endl; + std::cout << "options:\n"; + std::cout << " -m monaddress[:port] connect to specified monitor\n"; + std::cout << " --keyring=<path> path to keyring for local " + << "cluster\n"; + std::cout << " --log-file=<logfile> file to log debug output\n"; + std::cout << " --debug-immutable-obj-cache=<log-level>/<memory-level> " + << "set debug level\n"; + generic_server_usage(); +} + +static void handle_signal(int signum) { + if (cachectl) + cachectl->handle_signal(signum); +} + +int main(int argc, const char **argv) { + std::vector<const char*> args; + env_to_vec(args); + argv_to_vec(argc, argv, args); + + if (ceph_argparse_need_usage(args)) { + usage(); + exit(0); + } + + auto cct = global_init(nullptr, args, CEPH_ENTITY_TYPE_CLIENT, + CODE_ENVIRONMENT_DAEMON, + CINIT_FLAG_UNPRIVILEGED_DAEMON_DEFAULTS); + + if (g_conf()->daemonize) { + global_init_daemonize(g_ceph_context); + } + + common_init_finish(g_ceph_context); + global_init_chdir(g_ceph_context); + + init_async_signal_handler(); + register_async_signal_handler(SIGHUP, sighup_handler); + register_async_signal_handler_oneshot(SIGINT, handle_signal); + register_async_signal_handler_oneshot(SIGTERM, handle_signal); + + std::vector<const char*> cmd_args; + argv_to_vec(argc, argv, cmd_args); + + cachectl = new ceph::immutable_obj_cache::CacheController(g_ceph_context, + cmd_args); + int r = cachectl->init(); + if (r < 0) { + std::cerr << "failed to initialize: " << cpp_strerror(r) << std::endl; + goto cleanup; + } + + r = cachectl->run(); + if (r < 0) { + goto cleanup; + } + + cleanup: + unregister_async_signal_handler(SIGHUP, sighup_handler); + unregister_async_signal_handler(SIGINT, handle_signal); + unregister_async_signal_handler(SIGTERM, handle_signal); + shutdown_async_signal_handler(); + + delete cachectl; + + return r < 0 ? EXIT_SUCCESS : EXIT_FAILURE; +} |