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