+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-
-#ifndef RGW_PROCESS_H
-#define RGW_PROCESS_H
-
-#include "rgw_common.h"
-#include "rgw_rados.h"
-#include "rgw_acl.h"
-#include "rgw_auth_registry.h"
-#include "rgw_user.h"
-#include "rgw_op.h"
-#include "rgw_rest.h"
-
-#include "include/assert.h"
-
-#include "common/WorkQueue.h"
-#include "common/Throttle.h"
-
-#include <atomic>
-
-#if !defined(dout_subsys)
-#define dout_subsys ceph_subsys_rgw
-#define def_dout_subsys
-#endif
-
-#define dout_context g_ceph_context
-
-extern void signal_shutdown();
-
-struct RGWProcessEnv {
- RGWRados *store;
- RGWREST *rest;
- OpsLogSocket *olog;
- int port;
- std::string uri_prefix;
- std::shared_ptr<rgw::auth::StrategyRegistry> auth_registry;
-};
-
-class RGWFrontendConfig;
-
-class RGWProcess {
- deque<RGWRequest*> m_req_queue;
-protected:
- CephContext *cct;
- RGWRados* store;
- rgw_auth_registry_ptr_t auth_registry;
- OpsLogSocket* olog;
- ThreadPool m_tp;
- Throttle req_throttle;
- RGWREST* rest;
- RGWFrontendConfig* conf;
- int sock_fd;
- std::string uri_prefix;
-
- struct RGWWQ : public ThreadPool::WorkQueue<RGWRequest> {
- RGWProcess* process;
- RGWWQ(RGWProcess* p, time_t timeout, time_t suicide_timeout, ThreadPool* tp)
- : ThreadPool::WorkQueue<RGWRequest>("RGWWQ", timeout, suicide_timeout,
- tp), process(p) {}
-
- bool _enqueue(RGWRequest* req) override {
- process->m_req_queue.push_back(req);
- perfcounter->inc(l_rgw_qlen);
- dout(20) << "enqueued request req=" << hex << req << dec << dendl;
- _dump_queue();
- return true;
- }
-
- void _dequeue(RGWRequest* req) override {
- ceph_abort();
- }
-
- bool _empty() override {
- return process->m_req_queue.empty();
- }
-
- RGWRequest* _dequeue() override {
- if (process->m_req_queue.empty())
- return NULL;
- RGWRequest *req = process->m_req_queue.front();
- process->m_req_queue.pop_front();
- dout(20) << "dequeued request req=" << hex << req << dec << dendl;
- _dump_queue();
- perfcounter->inc(l_rgw_qlen, -1);
- return req;
- }
-
- using ThreadPool::WorkQueue<RGWRequest>::_process;
-
- void _process(RGWRequest *req, ThreadPool::TPHandle &) override {
- perfcounter->inc(l_rgw_qactive);
- process->handle_request(req);
- process->req_throttle.put(1);
- perfcounter->inc(l_rgw_qactive, -1);
- }
-
- void _dump_queue();
-
- void _clear() override {
- assert(process->m_req_queue.empty());
- }
- } req_wq;
-
-public:
- RGWProcess(CephContext* const cct,
- RGWProcessEnv* const pe,
- const int num_threads,
- RGWFrontendConfig* const conf)
- : cct(cct),
- store(pe->store),
- auth_registry(pe->auth_registry),
- olog(pe->olog),
- m_tp(cct, "RGWProcess::m_tp", "tp_rgw_process", num_threads),
- req_throttle(cct, "rgw_ops", num_threads * 2),
- rest(pe->rest),
- conf(conf),
- sock_fd(-1),
- uri_prefix(pe->uri_prefix),
- req_wq(this, g_conf->rgw_op_thread_timeout,
- g_conf->rgw_op_thread_suicide_timeout, &m_tp) {
- }
-
- virtual ~RGWProcess() = default;
-
- virtual void run() = 0;
- virtual void handle_request(RGWRequest *req) = 0;
-
- void pause() {
- m_tp.pause();
- }
-
- void unpause_with_new_config(RGWRados* const store,
- rgw_auth_registry_ptr_t auth_registry) {
- this->store = store;
- this->auth_registry = std::move(auth_registry);
- m_tp.unpause();
- }
-
- void close_fd() {
- if (sock_fd >= 0) {
- ::close(sock_fd);
- sock_fd = -1;
- }
- }
-}; /* RGWProcess */
-
-class RGWFCGXProcess : public RGWProcess {
- int max_connections;
-public:
-
- /* have a bit more connections than threads so that requests are
- * still accepted even if we're still processing older requests */
- RGWFCGXProcess(CephContext* const cct,
- RGWProcessEnv* const pe,
- const int num_threads,
- RGWFrontendConfig* const conf)
- : RGWProcess(cct, pe, num_threads, conf),
- max_connections(num_threads + (num_threads >> 3)) {
- }
-
- void run() override;
- void handle_request(RGWRequest* req) override;
-};
-
-class RGWProcessControlThread : public Thread {
- RGWProcess *pprocess;
-public:
- RGWProcessControlThread(RGWProcess *_pprocess) : pprocess(_pprocess) {}
-
- void *entry() override {
- pprocess->run();
- return NULL;
- }
-};
-
-class RGWLoadGenProcess : public RGWProcess {
- RGWAccessKey access_key;
-public:
- RGWLoadGenProcess(CephContext* cct, RGWProcessEnv* pe, int num_threads,
- RGWFrontendConfig* _conf) :
- RGWProcess(cct, pe, num_threads, _conf) {}
- void run() override;
- void checkpoint();
- void handle_request(RGWRequest* req) override;
- void gen_request(const string& method, const string& resource,
- int content_length, std::atomic<bool>* fail_flag);
-
- void set_access_key(RGWAccessKey& key) { access_key = key; }
-};
-
-/* process stream request */
-extern int process_request(RGWRados* store,
- RGWREST* rest,
- RGWRequest* req,
- const std::string& frontend_prefix,
- const rgw_auth_registry_t& auth_registry,
- RGWRestfulIO* client_io,
- OpsLogSocket* olog);
-
-extern int rgw_process_authenticated(RGWHandler_REST* handler,
- RGWOp*& op,
- RGWRequest* req,
- req_state* s,
- bool skip_retarget = false);
-
-#if defined(def_dout_subsys)
-#undef def_dout_subsys
-#undef dout_subsys
-#endif
-#undef dout_context
-
-#endif /* RGW_PROCESS_H */
Code Review / stor4nfv.git / blobdiff