ceph/src/rgw/rgw_client_io.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 #ifndef CEPH_RGW_CLIENT_IO_H
   5 #define CEPH_RGW_CLIENT_IO_H
   6
   7 #include <exception>
   8 #include <string>
   9 #include <string_view>
  10 #include <streambuf>
  11 #include <istream>
  12 #include <stdlib.h>
  13 #include <system_error>
  14
  15 #include "include/types.h"
  16 #include "rgw_common.h"
  17
  18
  19 class RGWRestfulIO;
  20
  21 namespace rgw {
  22 namespace io {
  23
  24 using Exception = std::system_error;
  25
  26 /* The minimal and simplest subset of methods that a client of RadosGW can be
  27  * interacted with. */
  28 class BasicClient {
  29 protected:
  30   virtual int init_env(CephContext *cct) = 0;
  31
  32 public:
  33   virtual ~BasicClient() = default;
  34
  35   /* Initialize the BasicClient and inject CephContext. */
  36   int init(CephContext *cct);
  37
  38   /* Return the RGWEnv describing the environment that a given request lives in.
  39    * The method does not throw exceptions. */
  40   virtual RGWEnv& get_env() noexcept = 0;
  41
  42   /* Complete request.
  43    * On success returns number of bytes generated for a direct client of RadosGW.
  44    * On failure throws rgw::io::Exception containing errno. */
  45   virtual size_t complete_request() = 0;
  46 }; /* rgw::io::Client */
  47
  48
  49 class Accounter {
  50 public:
  51   virtual ~Accounter() = default;
  52
  53   /* Enable or disable the accounting of both sent and received data. Changing
  54    * the state does not affect the counters. */
  55   virtual void set_account(bool enabled) = 0;
  56
  57   /* Return number of bytes sent to a direct client of RadosGW (direct means
  58    * eg. a web server instance in the case of using FastCGI front-end) when
  59    * the accounting was enabled. */
  60   virtual uint64_t get_bytes_sent() const = 0;
  61
  62   /* Return number of bytes received from a direct client of RadosGW (direct
  63    * means eg. a web server instance in the case of using FastCGI front-end)
  64    * when the accounting was enabled. */
  65   virtual uint64_t get_bytes_received() const = 0;
  66 }; /* rgw::io::Accounter */
  67
  68
  69 /* Interface abstracting restful interactions with clients, usually through
  70  * the HTTP protocol. The methods participating in the response generation
  71  * process should be called in the specific order:
  72  *   1. send_100_continue() - at most once,
  73  *   2. send_status() - exactly once,
  74  *   3. Any of:
  75  *      a. send_header(),
  76  *      b. send_content_length() XOR send_chunked_transfer_encoding()
  77  *         Please note that only one of those two methods must be called
  78            at most once.
  79  *   4. complete_header() - exactly once,
  80  *   5. send_body()
  81  *   6. complete_request() - exactly once.
  82  * There are no restrictions on flush() - it may be called in any moment.
  83  *
  84  * Receiving data from a client isn't a subject to any further call order
  85  * restrictions besides those imposed by BasicClient. That is, get_env()
  86  * and recv_body can be mixed. */
  87 class RestfulClient : public BasicClient {
  88   template<typename T> friend class DecoratedRestfulClient;
  89
  90 public:
  91   /* Generate the 100 Continue message.
  92    * On success returns number of bytes generated for a direct client of RadosGW.
  93    * On failure throws rgw::io::Exception containing errno. */
  94   virtual size_t send_100_continue() = 0;
  95
  96   /* Generate the response's status part taking the HTTP status code as @status
  97    * and its name pointed in @status_name.
  98    * On success returns number of bytes generated for a direct client of RadosGW.
  99    * On failure throws rgw::io::Exception containing errno. */
 100   virtual size_t send_status(int status, const char *status_name) = 0;
 101
 102   /* Generate header. On success returns number of bytes generated for a direct
 103    * client of RadosGW. On failure throws rgw::io::Exception containing errno.
 104    *
 105    * std::string_view is being used because of length it internally carries. */
 106   virtual size_t send_header(const std::string_view& name,
 107                              const std::string_view& value) = 0;
 108
 109   /* Inform a client about a content length. Takes number of bytes as @len.
 110    * On success returns number of bytes generated for a direct client of RadosGW.
 111    * On failure throws rgw::io::Exception containing errno.
 112    *
 113    * CALL LIMITATIONS:
 114    *  - The method must be called EXACTLY ONCE.
 115    *  - The method is interchangeable with send_chunked_transfer_encoding(). */
 116   virtual size_t send_content_length(uint64_t len) = 0;
 117
 118   /* Inform a client that the chunked transfer encoding will be used.
 119    * On success returns number of bytes generated for a direct client of RadosGW.
 120    * On failure throws rgw::io::Exception containing errno.
 121    *
 122    * CALL LIMITATIONS:
 123    *  - The method must be called EXACTLY ONCE.
 124    *  - The method is interchangeable with send_content_length(). */
 125   virtual size_t send_chunked_transfer_encoding() {
 126     /* This is a null implementation. We don't send anything here, even the HTTP
 127      * header. The intended behaviour should be provided through a decorator or
 128      * directly by a given front-end. */
 129     return 0;
 130   }
 131
 132   /* Generate completion (the CRLF sequence separating headers and body in
 133    * the case of HTTP) of headers. On success returns number of generated bytes
 134    * for a direct client of RadosGW. On failure throws rgw::io::Exception with
 135    * errno. */
 136   virtual size_t complete_header() = 0;
 137
 138   /* Receive no more than @max bytes from a request's body and store it in
 139    * buffer pointed by @buf. On success returns number of bytes received from
 140    * a direct client of RadosGW that has been stored in @buf. On failure throws
 141    * rgw::io::Exception containing errno. */
 142   virtual size_t recv_body(char* buf, size_t max) = 0;
 143
 144   /* Generate a part of response's body by taking exactly @len bytes from
 145    * the buffer pointed by @buf. On success returns number of generated bytes
 146    * of response's body. On failure throws rgw::io::Exception. */
 147   virtual size_t send_body(const char* buf, size_t len) = 0;
 148
 149   /* Flushes all already generated data to a direct client of RadosGW.
 150    * On failure throws rgw::io::Exception containing errno. */
 151   virtual void flush() = 0;
 152 } /* rgw::io::RestfulClient */;
 153
 154
 155 /* Abstract decorator over any implementation of rgw::io::RestfulClient
 156  * which could be provided both as a pointer-to-object or the object itself. */
 157 template <typename DecorateeT>
 158 class DecoratedRestfulClient : public RestfulClient {
 159   template<typename T> friend class DecoratedRestfulClient;
 160   friend RGWRestfulIO;
 161
 162   typedef typename std::remove_pointer<DecorateeT>::type DerefedDecorateeT;
 163
 164   static_assert(std::is_base_of<RestfulClient, DerefedDecorateeT>::value,
 165                 "DecorateeT must be a subclass of rgw::io::RestfulClient");
 166
 167   DecorateeT decoratee;
 168
 169   /* There is an indirection layer over accessing decoratee to share the same
 170    * code base between dynamic and static decorators. The difference is about
 171    * what we store internally: pointer to a decorated object versus the whole
 172    * object itself. */
 173   template <typename T = void,
 174             typename std::enable_if<
 175     ! std::is_pointer<DecorateeT>::value, T>::type* = nullptr>
 176   DerefedDecorateeT& get_decoratee() {
 177     return decoratee;
 178   }
 179
 180 protected:
 181   template <typename T = void,
 182             typename std::enable_if<
 183     std::is_pointer<DecorateeT>::value, T>::type* = nullptr>
 184   DerefedDecorateeT& get_decoratee() {
 185     return *decoratee;
 186   }
 187
 188   /* Dynamic decorators (those storing a pointer instead of the decorated
 189    * object itself) can be reconfigured on-the-fly. HOWEVER: there are no
 190    * facilities for orchestrating such changes. Callers must take care of
 191    * atomicity and thread-safety. */
 192   template <typename T = void,
 193             typename std::enable_if<
 194     std::is_pointer<DecorateeT>::value, T>::type* = nullptr>
 195   void set_decoratee(DerefedDecorateeT& new_dec) {
 196     decoratee = &new_dec;
 197   }
 198
 199   int init_env(CephContext *cct) override {
 200     return get_decoratee().init_env(cct);
 201   }
 202
 203 public:
 204   explicit DecoratedRestfulClient(DecorateeT&& decoratee)
 205     : decoratee(std::forward<DecorateeT>(decoratee)) {
 206   }
 207
 208   size_t send_status(const int status,
 209                      const char* const status_name) override {
 210     return get_decoratee().send_status(status, status_name);
 211   }
 212
 213   size_t send_100_continue() override {
 214     return get_decoratee().send_100_continue();
 215   }
 216
 217   size_t send_header(const std::string_view& name,
 218                      const std::string_view& value) override {
 219     return get_decoratee().send_header(name, value);
 220   }
 221
 222   size_t send_content_length(const uint64_t len) override {
 223     return get_decoratee().send_content_length(len);
 224   }
 225
 226   size_t send_chunked_transfer_encoding() override {
 227     return get_decoratee().send_chunked_transfer_encoding();
 228   }
 229
 230   size_t complete_header() override {
 231     return get_decoratee().complete_header();
 232   }
 233
 234   size_t recv_body(char* const buf, const size_t max) override {
 235     return get_decoratee().recv_body(buf, max);
 236   }
 237
 238   size_t send_body(const char* const buf,
 239                    const size_t len) override {
 240     return get_decoratee().send_body(buf, len);
 241   }
 242
 243   void flush() override {
 244     return get_decoratee().flush();
 245   }
 246
 247   RGWEnv& get_env() noexcept override {
 248     return get_decoratee().get_env();
 249   }
 250
 251   size_t complete_request() override {
 252     return get_decoratee().complete_request();
 253   }
 254 } /* rgw::io::DecoratedRestfulClient */;
 255
 256
 257 /* Interface that should be provided by a front-end class wanting to use
 258  * the low-level buffering offered by i.e. StaticOutputBufferer. */
 259 class BuffererSink {
 260 public:
 261   virtual ~BuffererSink() = default;
 262
 263   /* Send exactly @len bytes from the memory location pointed by @buf.
 264    * On success returns @len. On failure throws rgw::io::Exception. */
 265   virtual size_t write_data(const char *buf, size_t len) = 0;
 266 };
 267
 268 /* Utility class providing RestfulClient's implementations with facilities
 269  * for low-level buffering without relying on dynamic memory allocations.
 270  * The buffer is carried entirely on stack. This narrows down applicability
 271  * to these situations where buffers are relatively small. This perfectly
 272  * fits the needs of composing an HTTP header. Without that a front-end
 273  * might need to issue a lot of small IO operations leading to increased
 274  * overhead on syscalls and fragmentation of a message if the Nagle's
 275  * algorithm won't be able to form a single TCP segment (usually when
 276  * running on extremely fast network interfaces like the loopback). */
 277 template <size_t BufferSizeV = 4096>
 278 class StaticOutputBufferer : public std::streambuf {
 279   static_assert(BufferSizeV >= sizeof(std::streambuf::char_type),
 280                 "Buffer size must be bigger than a single char_type.");
 281
 282   using std::streambuf::int_type;
 283
 284   int_type overflow(const int_type c) override {
 285     *pptr() = c;
 286     pbump(sizeof(std::streambuf::char_type));
 287
 288     if (! sync()) {
 289       /* No error, the buffer has been successfully synchronized. */
 290       return c;
 291     } else {
 292       return std::streambuf::traits_type::eof();
 293     }
 294   }
 295
 296   int sync() override {
 297     const auto len = static_cast<size_t>(std::streambuf::pptr() -
 298                                          std::streambuf::pbase());
 299     std::streambuf::pbump(-len);
 300     sink.write_data(std::streambuf::pbase(), len);
 301     /* Always return success here. In case of failure write_data() will throw
 302      * rgw::io::Exception. */
 303     return 0;
 304   }
 305
 306   BuffererSink& sink;
 307   std::streambuf::char_type buffer[BufferSizeV];
 308
 309 public:
 310   explicit StaticOutputBufferer(BuffererSink& sink)
 311     : sink(sink) {
 312     constexpr size_t len = sizeof(buffer) - sizeof(std::streambuf::char_type);
 313     std::streambuf::setp(buffer, buffer + len);
 314   }
 315 };
 316
 317 } /* namespace io */
 318 } /* namespace rgw */
 319
 320
 321 /* We're doing this nasty thing only because of extensive usage of templates
 322  * to implement the static decorator pattern. C++ templates de facto enforce
 323  * mixing interfaces with implementation. Additionally, those classes derive
 324  * from RGWRestfulIO defined here. I believe that including in the middle of
 325  * file is still better than polluting it directly. */
 326 #include "rgw_client_io_filters.h"
 327
 328
 329 /* RGWRestfulIO: high level interface to interact with RESTful clients. What
 330  * differentiates it from rgw::io::RestfulClient is providing more specific APIs
 331  * like rgw::io::Accounter or the AWS Auth v4 stuff implemented by filters
 332  * while hiding the pipelined architecture from clients.
 333  *
 334  * rgw::io::Accounter came in as a part of rgw::io::AccountingFilter. */
 335 class RGWRestfulIO : public rgw::io::AccountingFilter<rgw::io::RestfulClient*> {
 336   std::vector<std::shared_ptr<DecoratedRestfulClient>> filters;
 337
 338 public:
 339   ~RGWRestfulIO() override = default;
 340
 341   RGWRestfulIO(CephContext *_cx, rgw::io::RestfulClient* engine)
 342     : AccountingFilter<rgw::io::RestfulClient*>(_cx, std::move(engine)) {
 343   }
 344
 345   void add_filter(std::shared_ptr<DecoratedRestfulClient> new_filter) {
 346     new_filter->set_decoratee(this->get_decoratee());
 347     this->set_decoratee(*new_filter);
 348     filters.emplace_back(std::move(new_filter));
 349   }
 350 }; /* RGWRestfulIO */
 351
 352
 353 /* Type conversions to work around lack of req_state type hierarchy matching
 354  * (e.g.) REST backends (may be replaced w/dynamic typed req_state). */
 355 static inline rgw::io::RestfulClient* RESTFUL_IO(struct req_state* s) {
 356   ceph_assert(dynamic_cast<rgw::io::RestfulClient*>(s->cio) != nullptr);
 357
 358   return static_cast<rgw::io::RestfulClient*>(s->cio);
 359 }
 360
 361 static inline rgw::io::Accounter* ACCOUNTING_IO(struct req_state* s) {
 362   auto ptr = dynamic_cast<rgw::io::Accounter*>(s->cio);
 363   ceph_assert(ptr != nullptr);
 364
 365   return ptr;
 366 }
 367
 368 static inline RGWRestfulIO* AWS_AUTHv4_IO(const req_state* const s) {
 369   ceph_assert(dynamic_cast<RGWRestfulIO*>(s->cio) != nullptr);
 370
 371   return static_cast<RGWRestfulIO*>(s->cio);
 372 }
 373
 374
 375 class RGWClientIOStreamBuf : public std::streambuf {
 376 protected:
 377   RGWRestfulIO &rio;
 378   size_t const window_size;
 379   size_t const putback_size;
 380   std::vector<char> buffer;
 381
 382 public:
 383   RGWClientIOStreamBuf(RGWRestfulIO &rio, size_t ws, size_t ps = 1)
 384     : rio(rio),
 385       window_size(ws),
 386       putback_size(ps),
 387       buffer(ws + ps)
 388   {
 389     setg(nullptr, nullptr, nullptr);
 390   }
 391
 392   std::streambuf::int_type underflow() override {
 393     if (gptr() < egptr()) {
 394       return traits_type::to_int_type(*gptr());
 395     }
 396
 397     char * const base = buffer.data();
 398     char * start;
 399
 400     if (nullptr != eback()) {
 401       /* We need to skip moving bytes on first underflow. In such case
 402        * there is simply no previous data we should preserve for unget()
 403        * or something similar. */
 404       std::memmove(base, egptr() - putback_size, putback_size);
 405       start = base + putback_size;
 406     } else {
 407       start = base;
 408     }
 409
 410     size_t read_len = 0;
 411     try {
 412       read_len = rio.recv_body(base, window_size);
 413     } catch (rgw::io::Exception&) {
 414       return traits_type::eof();
 415     }
 416     if (0 == read_len) {
 417       return traits_type::eof();
 418     }
 419
 420     setg(base, start, start + read_len);
 421
 422     return traits_type::to_int_type(*gptr());
 423   }
 424 };
 425
 426 class RGWClientIOStream : private RGWClientIOStreamBuf, public std::istream {
 427 /* Inheritance from RGWClientIOStreamBuf is a kind of shadow, undirect
 428  * form of composition here. We cannot do that explicitly because istream
 429  * ctor is being called prior to construction of any member of this class. */
 430
 431 public:
 432   explicit RGWClientIOStream(RGWRestfulIO &s)
 433     : RGWClientIOStreamBuf(s, 1, 2),
 434       std::istream(static_cast<RGWClientIOStreamBuf *>(this)) {
 435   }
 436 };
 437
 438 #endif /* CEPH_RGW_CLIENT_IO_H */