ceph/src/rgw/rgw_file.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 RGW_FILE_H
   5 #define RGW_FILE_H
   6
   7 #include "include/rados/rgw_file.h"
   8
   9 /* internal header */
  10 #include <string.h>
  11 #include <string_view>
  12 #include <sys/stat.h>
  13 #include <stdint.h>
  14
  15 #include <atomic>
  16 #include <chrono>
  17 #include <thread>
  18 #include <mutex>
  19 #include <vector>
  20 #include <deque>
  21 #include <algorithm>
  22 #include <functional>
  23 #include <boost/intrusive_ptr.hpp>
  24 #include <boost/range/adaptor/reversed.hpp>
  25 #include <boost/container/flat_map.hpp>
  26 #include <boost/variant.hpp>
  27 #include <boost/optional.hpp>
  28 #include "xxhash.h"
  29 #include "include/buffer.h"
  30 #include "common/cohort_lru.h"
  31 #include "common/ceph_timer.h"
  32 #include "rgw_common.h"
  33 #include "rgw_user.h"
  34 #include "rgw_lib.h"
  35 #include "rgw_ldap.h"
  36 #include "rgw_token.h"
  37 #include "rgw_putobj_processor.h"
  38 #include "rgw_aio_throttle.h"
  39 #include "rgw_compression.h"
  40
  41
  42 /* XXX
  43  * ASSERT_H somehow not defined after all the above (which bring
  44  * in common/debug.h [e.g., dout])
  45  */
  46 #include "include/ceph_assert.h"
  47
  48
  49 #define RGW_RWXMODE  (S_IRWXU | S_IRWXG | S_IRWXO)
  50
  51 #define RGW_RWMODE (RGW_RWXMODE &                       \
  52                       ~(S_IXUSR | S_IXGRP | S_IXOTH))
  53
  54
  55 namespace rgw {
  56
  57   template <typename T>
  58   static inline void ignore(T &&) {}
  59
  60
  61   namespace bi = boost::intrusive;
  62
  63   class RGWLibFS;
  64   class RGWFileHandle;
  65   class RGWWriteRequest;
  66
  67   inline bool operator <(const struct timespec& lhs,
  68                          const struct timespec& rhs) {
  69     if (lhs.tv_sec == rhs.tv_sec)
  70       return lhs.tv_nsec < rhs.tv_nsec;
  71     else
  72       return lhs.tv_sec < rhs.tv_sec;
  73   }
  74
  75   inline bool operator ==(const struct timespec& lhs,
  76                           const struct timespec& rhs) {
  77     return ((lhs.tv_sec == rhs.tv_sec) &&
  78             (lhs.tv_nsec == rhs.tv_nsec));
  79   }
  80
  81   /*
  82    * XXX
  83    * The current 64-bit, non-cryptographic hash used here is intended
  84    * for prototyping only.
  85    *
  86    * However, the invariant being prototyped is that objects be
  87    * identifiable by their hash components alone.  We believe this can
  88    * be legitimately implemented using 128-hash values for bucket and
  89    * object components, together with a cluster-resident cryptographic
  90    * key.  Since an MD5 or SHA-1 key is 128 bits and the (fast),
  91    * non-cryptographic CityHash128 hash algorithm takes a 128-bit seed,
  92    * speculatively we could use that for the final hash computations.
  93    */
  94   struct fh_key
  95   {
  96     rgw_fh_hk fh_hk {};
  97     uint32_t version;
  98
  99     static constexpr uint64_t seed = 8675309;
 100
 101     fh_key() : version(0) {}
 102
 103     fh_key(const rgw_fh_hk& _hk)
 104       : fh_hk(_hk), version(0) {
 105       // nothing
 106     }
 107
 108     fh_key(const uint64_t bk, const uint64_t ok)
 109       : version(0) {
 110       fh_hk.bucket = bk;
 111       fh_hk.object = ok;
 112     }
 113
 114     fh_key(const uint64_t bk, const char *_o, const std::string& _t)
 115       : version(0) {
 116       fh_hk.bucket = bk;
 117       std::string to = _t + ":" + _o;
 118       fh_hk.object = XXH64(to.c_str(), to.length(), seed);
 119     }
 120
 121     fh_key(const std::string& _b, const std::string& _o,
 122            const std::string& _t /* tenant */)
 123       : version(0) {
 124       std::string tb = _t + ":" + _b;
 125       std::string to = _t + ":" + _o;
 126       fh_hk.bucket = XXH64(tb.c_str(), tb.length(), seed);
 127       fh_hk.object = XXH64(to.c_str(), to.length(), seed);
 128     }
 129
 130     void encode(buffer::list& bl) const {
 131       ENCODE_START(2, 1, bl);
 132       encode(fh_hk.bucket, bl);
 133       encode(fh_hk.object, bl);
 134       encode((uint32_t)2, bl);
 135       ENCODE_FINISH(bl);
 136     }
 137
 138     void decode(bufferlist::const_iterator& bl) {
 139       DECODE_START(2, bl);
 140       decode(fh_hk.bucket, bl);
 141       decode(fh_hk.object, bl);
 142       if (struct_v >= 2) {
 143         decode(version, bl);
 144       }
 145       DECODE_FINISH(bl);
 146     }
 147
 148     friend std::ostream& operator<<(std::ostream &os, fh_key const &fhk);
 149
 150   }; /* fh_key */
 151
 152   WRITE_CLASS_ENCODER(fh_key);
 153
 154   inline bool operator<(const fh_key& lhs, const fh_key& rhs)
 155   {
 156     return ((lhs.fh_hk.bucket < rhs.fh_hk.bucket) ||
 157             ((lhs.fh_hk.bucket == rhs.fh_hk.bucket) &&
 158               (lhs.fh_hk.object < rhs.fh_hk.object)));
 159   }
 160
 161   inline bool operator>(const fh_key& lhs, const fh_key& rhs)
 162   {
 163     return (rhs < lhs);
 164   }
 165
 166   inline bool operator==(const fh_key& lhs, const fh_key& rhs)
 167   {
 168     return ((lhs.fh_hk.bucket == rhs.fh_hk.bucket) &&
 169             (lhs.fh_hk.object == rhs.fh_hk.object));
 170   }
 171
 172   inline bool operator!=(const fh_key& lhs, const fh_key& rhs)
 173   {
 174     return !(lhs == rhs);
 175   }
 176
 177   inline bool operator<=(const fh_key& lhs, const fh_key& rhs)
 178   {
 179     return (lhs < rhs) || (lhs == rhs);
 180   }
 181
 182   using boost::variant;
 183   using boost::container::flat_map;
 184
 185   typedef std::tuple<bool, bool> DecodeAttrsResult;
 186
 187   class RGWFileHandle : public cohort::lru::Object
 188   {
 189     struct rgw_file_handle fh;
 190     std::mutex mtx;
 191
 192     RGWLibFS* fs;
 193     RGWFileHandle* bucket;
 194     RGWFileHandle* parent;
 195     /* const */ std::string name; /* XXX file or bucket name */
 196     /* const */ fh_key fhk;
 197
 198     using lock_guard = std::lock_guard<std::mutex>;
 199     using unique_lock = std::unique_lock<std::mutex>;
 200
 201     /* TODO: keeping just the last marker is sufficient for
 202      * nfs-ganesha 2.4.5; in the near future, nfs-ganesha will
 203      * be able to hint the name of the next dirent required,
 204      * from which we can directly synthesize a RADOS marker.
 205      * using marker_cache_t = flat_map<uint64_t, rgw_obj_key>;
 206      */
 207
 208     struct State {
 209       uint64_t dev;
 210       uint64_t size;
 211       uint64_t nlink;
 212       uint32_t owner_uid; /* XXX need Unix attr */
 213       uint32_t owner_gid; /* XXX need Unix attr */
 214       mode_t unix_mode;
 215       struct timespec ctime;
 216       struct timespec mtime;
 217       struct timespec atime;
 218       uint32_t version;
 219       State() : dev(0), size(0), nlink(1), owner_uid(0), owner_gid(0), unix_mode(0),
 220                 ctime{0,0}, mtime{0,0}, atime{0,0}, version(0) {}
 221     } state;
 222
 223     struct file {
 224       RGWWriteRequest* write_req;
 225       file() : write_req(nullptr) {}
 226       ~file();
 227     };
 228
 229     struct directory {
 230
 231       static constexpr uint32_t FLAG_NONE =     0x0000;
 232
 233       uint32_t flags;
 234       rgw_obj_key last_marker;
 235       struct timespec last_readdir;
 236
 237       directory() : flags(FLAG_NONE), last_readdir{0,0} {}
 238     };
 239
 240     void clear_state();
 241     void advance_mtime(uint32_t flags = FLAG_NONE);
 242
 243     boost::variant<file, directory> variant_type;
 244
 245     uint16_t depth;
 246     uint32_t flags;
 247
 248     ceph::buffer::list etag;
 249     ceph::buffer::list acls;
 250
 251   public:
 252     const static std::string root_name;
 253
 254     static constexpr uint16_t MAX_DEPTH = 256;
 255
 256     static constexpr uint32_t FLAG_NONE =    0x0000;
 257     static constexpr uint32_t FLAG_OPEN =    0x0001;
 258     static constexpr uint32_t FLAG_ROOT =    0x0002;
 259     static constexpr uint32_t FLAG_CREATE =  0x0004;
 260     static constexpr uint32_t FLAG_CREATING =  0x0008;
 261     static constexpr uint32_t FLAG_SYMBOLIC_LINK = 0x0009;
 262     static constexpr uint32_t FLAG_DIRECTORY = 0x0010;
 263     static constexpr uint32_t FLAG_BUCKET = 0x0020;
 264     static constexpr uint32_t FLAG_LOCK =   0x0040;
 265     static constexpr uint32_t FLAG_DELETED = 0x0080;
 266     static constexpr uint32_t FLAG_UNLINK_THIS = 0x0100;
 267     static constexpr uint32_t FLAG_LOCKED = 0x0200;
 268     static constexpr uint32_t FLAG_STATELESS_OPEN = 0x0400;
 269     static constexpr uint32_t FLAG_EXACT_MATCH = 0x0800;
 270     static constexpr uint32_t FLAG_MOUNT = 0x1000;
 271     static constexpr uint32_t FLAG_IN_CB = 0x2000;
 272
 273 #define CREATE_FLAGS(x) \
 274     ((x) & ~(RGWFileHandle::FLAG_CREATE|RGWFileHandle::FLAG_LOCK))
 275
 276     static constexpr uint32_t RCB_MASK = \
 277       RGW_SETATTR_MTIME|RGW_SETATTR_CTIME|RGW_SETATTR_ATIME|RGW_SETATTR_SIZE;
 278
 279     friend class RGWLibFS;
 280
 281   private:
 282     explicit RGWFileHandle(RGWLibFS* _fs)
 283       : fs(_fs), bucket(nullptr), parent(nullptr), variant_type{directory()},
 284         depth(0), flags(FLAG_NONE)
 285       {
 286         fh.fh_hk.bucket = 0;
 287         fh.fh_hk.object = 0;
 288         /* root */
 289         fh.fh_type = RGW_FS_TYPE_DIRECTORY;
 290         variant_type = directory();
 291         /* stat */
 292         state.unix_mode = RGW_RWXMODE|S_IFDIR;
 293         /* pointer to self */
 294         fh.fh_private = this;
 295       }
 296
 297     uint64_t init_fsid(std::string& uid) {
 298       return XXH64(uid.c_str(), uid.length(), fh_key::seed);
 299     }
 300
 301     void init_rootfs(std::string& fsid, const std::string& object_name,
 302                      bool is_bucket) {
 303       /* fh_key */
 304       fh.fh_hk.bucket = XXH64(fsid.c_str(), fsid.length(), fh_key::seed);
 305       fh.fh_hk.object = XXH64(object_name.c_str(), object_name.length(),
 306                               fh_key::seed);
 307       fhk = fh.fh_hk;
 308       name = object_name;
 309
 310       state.dev = init_fsid(fsid);
 311
 312       if (is_bucket) {
 313         flags |= RGWFileHandle::FLAG_BUCKET | RGWFileHandle::FLAG_MOUNT;
 314         bucket = this;
 315         depth = 1;
 316       } else {
 317         flags |= RGWFileHandle::FLAG_ROOT | RGWFileHandle::FLAG_MOUNT;
 318       }
 319     }
 320
 321   public:
 322     RGWFileHandle(RGWLibFS* _fs, RGWFileHandle* _parent,
 323                   const fh_key& _fhk, std::string& _name, uint32_t _flags)
 324       : fs(_fs), bucket(nullptr), parent(_parent), name(std::move(_name)),
 325         fhk(_fhk), flags(_flags) {
 326
 327       if (parent->is_root()) {
 328         fh.fh_type = RGW_FS_TYPE_DIRECTORY;
 329         variant_type = directory();
 330         flags |= FLAG_BUCKET;
 331       } else {
 332         bucket = parent->is_bucket() ? parent
 333           : parent->bucket;
 334         if (flags & FLAG_DIRECTORY) {
 335           fh.fh_type = RGW_FS_TYPE_DIRECTORY;
 336           variant_type = directory();
 337         } else if(flags & FLAG_SYMBOLIC_LINK) {
 338           fh.fh_type = RGW_FS_TYPE_SYMBOLIC_LINK;
 339           variant_type = file();
 340         } else {
 341           fh.fh_type = RGW_FS_TYPE_FILE;
 342           variant_type = file();
 343         }
 344       }
 345
 346       depth = parent->depth + 1;
 347
 348       /* save constant fhk */
 349       fh.fh_hk = fhk.fh_hk; /* XXX redundant in fh_hk */
 350
 351       /* inherits parent's fsid */
 352       state.dev = parent->state.dev;
 353
 354       switch (fh.fh_type) {
 355       case RGW_FS_TYPE_DIRECTORY:
 356         state.unix_mode = RGW_RWXMODE|S_IFDIR;
 357         /* virtual directories are always invalid */
 358         advance_mtime();
 359         break;
 360       case RGW_FS_TYPE_FILE:
 361         state.unix_mode = RGW_RWMODE|S_IFREG;
 362         break;
 363       case RGW_FS_TYPE_SYMBOLIC_LINK:
 364         state.unix_mode = RGW_RWMODE|S_IFLNK;
 365         break;
 366       default:
 367         break;
 368       }
 369
 370       /* pointer to self */
 371       fh.fh_private = this;
 372     }
 373
 374     const std::string& get_name() const {
 375       return name;
 376     }
 377
 378     const fh_key& get_key() const {
 379       return fhk;
 380     }
 381
 382     directory* get_directory() {
 383       return boost::get<directory>(&variant_type);
 384     }
 385
 386     size_t get_size() const { return state.size; }
 387
 388     const char* stype() {
 389       return is_dir() ? "DIR" : "FILE";
 390     }
 391
 392     uint16_t get_depth() const { return depth; }
 393
 394     struct rgw_file_handle* get_fh() { return &fh; }
 395
 396     RGWLibFS* get_fs() { return fs; }
 397
 398     RGWFileHandle* get_parent() { return parent; }
 399
 400     uint32_t get_owner_uid() const { return state.owner_uid; }
 401     uint32_t get_owner_gid() const { return state.owner_gid; }
 402
 403     struct timespec get_ctime() const { return state.ctime; }
 404     struct timespec get_mtime() const { return state.mtime; }
 405
 406     const ceph::buffer::list& get_etag() const { return etag; }
 407     const ceph::buffer::list& get_acls() const { return acls; }
 408
 409     void create_stat(struct stat* st, uint32_t mask) {
 410       if (mask & RGW_SETATTR_UID)
 411         state.owner_uid = st->st_uid;
 412
 413       if (mask & RGW_SETATTR_GID)
 414         state.owner_gid = st->st_gid;
 415
 416       if (mask & RGW_SETATTR_MODE)  {
 417         switch (fh.fh_type) {
 418         case RGW_FS_TYPE_DIRECTORY:
 419           state.unix_mode = st->st_mode|S_IFDIR;
 420           break;
 421         case RGW_FS_TYPE_FILE:
 422           state.unix_mode = st->st_mode|S_IFREG;
 423           break;
 424         case RGW_FS_TYPE_SYMBOLIC_LINK:
 425           state.unix_mode = st->st_mode|S_IFLNK;
 426           break;
 427       default:
 428         break;
 429         }
 430       }
 431
 432       if (mask & RGW_SETATTR_ATIME)
 433         state.atime = st->st_atim;
 434
 435       if (mask & RGW_SETATTR_MTIME) {
 436         if (fh.fh_type != RGW_FS_TYPE_DIRECTORY)
 437           state.mtime = st->st_mtim;
 438       }
 439
 440       if (mask & RGW_SETATTR_CTIME)
 441         state.ctime = st->st_ctim;
 442     }
 443
 444     int stat(struct stat* st, uint32_t flags = FLAG_NONE) {
 445       /* partial Unix attrs */
 446       /* FIPS zeroization audit 20191115: this memset is not security
 447        * related. */
 448       memset(st, 0, sizeof(struct stat));
 449       st->st_dev = state.dev;
 450       st->st_ino = fh.fh_hk.object; // XXX
 451
 452       st->st_uid = state.owner_uid;
 453       st->st_gid = state.owner_gid;
 454
 455       st->st_mode = state.unix_mode;
 456
 457       switch (fh.fh_type) {
 458       case RGW_FS_TYPE_DIRECTORY:
 459         /* virtual directories are always invalid */
 460         advance_mtime(flags);
 461         st->st_nlink = state.nlink;
 462         break;
 463       case RGW_FS_TYPE_FILE:
 464         st->st_nlink = 1;
 465         st->st_blksize = 4096;
 466         st->st_size = state.size;
 467         st->st_blocks = (state.size) / 512;
 468         break;
 469       case RGW_FS_TYPE_SYMBOLIC_LINK:
 470         st->st_nlink = 1;
 471         st->st_blksize = 4096;
 472         st->st_size = state.size;
 473         st->st_blocks = (state.size) / 512;
 474         break;
 475       default:
 476         break;
 477       }
 478
 479 #ifdef HAVE_STAT_ST_MTIMESPEC_TV_NSEC
 480       st->st_atimespec = state.atime;
 481       st->st_mtimespec = state.mtime;
 482       st->st_ctimespec = state.ctime;
 483 #else
 484       st->st_atim = state.atime;
 485       st->st_mtim = state.mtime;
 486       st->st_ctim = state.ctime;
 487 #endif
 488
 489       return 0;
 490     }
 491
 492     const std::string& bucket_name() const {
 493       if (is_root())
 494         return root_name;
 495       if (is_bucket())
 496         return name;
 497       return bucket->object_name();
 498     }
 499
 500     const std::string& object_name() const { return name; }
 501
 502     std::string full_object_name(bool omit_bucket = false) const {
 503       std::string path;
 504       std::vector<const std::string*> segments;
 505       int reserve = 0;
 506       const RGWFileHandle* tfh = this;
 507       while (tfh && !tfh->is_root() && !(tfh->is_bucket() && omit_bucket)) {
 508         segments.push_back(&tfh->object_name());
 509         reserve += (1 + tfh->object_name().length());
 510         tfh = tfh->parent;
 511       }
 512       int pos = 1;
 513       path.reserve(reserve);
 514       for (auto& s : boost::adaptors::reverse(segments)) {
 515         if (pos > 1) {
 516           path += "/";
 517         } else {
 518           if (!omit_bucket &&
 519               ((path.length() == 0) || (path.front() != '/')))
 520             path += "/";
 521         }
 522         path += *s;
 523         ++pos;
 524       }
 525       return path;
 526     }
 527
 528     inline std::string relative_object_name() const {
 529       return full_object_name(true /* omit_bucket */);
 530     }
 531
 532     inline std::string relative_object_name2() {
 533       std::string rname = full_object_name(true /* omit_bucket */);
 534       if (is_dir()) {
 535         rname += "/";
 536       }
 537       return rname;
 538     }
 539
 540     inline std::string format_child_name(const std::string& cbasename,
 541                                          bool is_dir) const {
 542       std::string child_name{relative_object_name()};
 543       if ((child_name.size() > 0) &&
 544           (child_name.back() != '/'))
 545         child_name += "/";
 546       child_name += cbasename;
 547       if (is_dir)
 548         child_name += "/";
 549       return child_name;
 550     }
 551
 552     inline std::string make_key_name(const char *name) const {
 553       std::string key_name{full_object_name()};
 554       if (key_name.length() > 0)
 555         key_name += "/";
 556       key_name += name;
 557       return key_name;
 558     }
 559
 560     fh_key make_fhk(const std::string& name);
 561
 562     void add_marker(uint64_t off, const rgw_obj_key& marker,
 563                     uint8_t obj_type) {
 564       using std::get;
 565       directory* d = get<directory>(&variant_type);
 566       if (d) {
 567         unique_lock guard(mtx);
 568         d->last_marker = marker;
 569       }
 570     }
 571
 572     const rgw_obj_key* find_marker(uint64_t off) const {
 573       using std::get;
 574       if (off > 0) {
 575         const directory* d = get<directory>(&variant_type);
 576         if (d ) {
 577           return &d->last_marker;
 578         }
 579       }
 580       return nullptr;
 581     }
 582
 583     int offset_of(const std::string& name, int64_t *offset, uint32_t flags) {
 584       if (unlikely(! is_dir())) {
 585         return -EINVAL;
 586       }
 587       *offset = XXH64(name.c_str(), name.length(), fh_key::seed);
 588       return 0;
 589     }
 590
 591     bool is_open() const { return flags & FLAG_OPEN; }
 592     bool is_root() const { return flags & FLAG_ROOT; }
 593     bool is_mount() const { return flags & FLAG_MOUNT; }
 594     bool is_bucket() const { return flags & FLAG_BUCKET; }
 595     bool is_object() const { return !is_bucket(); }
 596     bool is_file() const { return (fh.fh_type == RGW_FS_TYPE_FILE); }
 597     bool is_dir() const { return (fh.fh_type == RGW_FS_TYPE_DIRECTORY); }
 598     bool is_link() const { return (fh.fh_type == RGW_FS_TYPE_SYMBOLIC_LINK); }
 599     bool creating() const { return flags & FLAG_CREATING; }
 600     bool deleted() const { return flags & FLAG_DELETED; }
 601     bool stateless_open() const { return flags & FLAG_STATELESS_OPEN; }
 602     bool has_children() const;
 603
 604     int open(uint32_t gsh_flags) {
 605       lock_guard guard(mtx);
 606       if (! is_open()) {
 607         if (gsh_flags & RGW_OPEN_FLAG_V3) {
 608           flags |= FLAG_STATELESS_OPEN;
 609         }
 610         flags |= FLAG_OPEN;
 611         return 0;
 612       }
 613       return -EPERM;
 614     }
 615
 616     typedef boost::variant<uint64_t*, const char*> readdir_offset;
 617
 618     int readdir(rgw_readdir_cb rcb, void *cb_arg, readdir_offset offset,
 619                 bool *eof, uint32_t flags);
 620
 621     int write(uint64_t off, size_t len, size_t *nbytes, void *buffer);
 622
 623     int commit(uint64_t offset, uint64_t length, uint32_t flags) {
 624       /* NFS3 and NFSv4 COMMIT implementation
 625        * the current atomic update strategy doesn't actually permit
 626        * clients to read-stable until either CLOSE (NFSv4+) or the
 627        * expiration of the active write timer (NFS3).  In the
 628        * interim, the client may send an arbitrary number of COMMIT
 629        * operations which must return a success result */
 630       return 0;
 631     }
 632
 633     int write_finish(uint32_t flags = FLAG_NONE);
 634     int close();
 635
 636     void open_for_create() {
 637       lock_guard guard(mtx);
 638       flags |= FLAG_CREATING;
 639     }
 640
 641     void clear_creating() {
 642       lock_guard guard(mtx);
 643       flags &= ~FLAG_CREATING;
 644     }
 645
 646     void inc_nlink(const uint64_t n) {
 647       state.nlink += n;
 648     }
 649
 650     void set_nlink(const uint64_t n) {
 651       state.nlink = n;
 652     }
 653
 654     void set_size(const size_t size) {
 655       state.size = size;
 656     }
 657
 658     void set_times(const struct timespec &ts) {
 659       state.ctime = ts;
 660       state.mtime = state.ctime;
 661       state.atime = state.ctime;
 662     }
 663
 664     void set_times(real_time t) {
 665       set_times(real_clock::to_timespec(t));
 666     }
 667
 668     void set_ctime(const struct timespec &ts) {
 669       state.ctime = ts;
 670     }
 671
 672     void set_mtime(const struct timespec &ts) {
 673       state.mtime = ts;
 674     }
 675
 676     void set_atime(const struct timespec &ts) {
 677       state.atime = ts;
 678     }
 679
 680     void set_etag(const ceph::buffer::list& _etag ) {
 681       etag = _etag;
 682     }
 683
 684     void set_acls(const ceph::buffer::list& _acls ) {
 685       acls = _acls;
 686     }
 687
 688     void encode(buffer::list& bl) const {
 689       ENCODE_START(2, 1, bl);
 690       encode(uint32_t(fh.fh_type), bl);
 691       encode(state.dev, bl);
 692       encode(state.size, bl);
 693       encode(state.nlink, bl);
 694       encode(state.owner_uid, bl);
 695       encode(state.owner_gid, bl);
 696       encode(state.unix_mode, bl);
 697       for (const auto& t : { state.ctime, state.mtime, state.atime }) {
 698         encode(real_clock::from_timespec(t), bl);
 699       }
 700       encode((uint32_t)2, bl);
 701       ENCODE_FINISH(bl);
 702     }
 703
 704     void decode(bufferlist::const_iterator& bl) {
 705       DECODE_START(2, bl);
 706       uint32_t fh_type;
 707       decode(fh_type, bl);
 708       if ((fh.fh_type != fh_type) &&
 709          (fh_type == RGW_FS_TYPE_SYMBOLIC_LINK))
 710         fh.fh_type = RGW_FS_TYPE_SYMBOLIC_LINK;
 711       ceph_assert(fh.fh_type == fh_type);
 712       decode(state.dev, bl);
 713       decode(state.size, bl);
 714       decode(state.nlink, bl);
 715       decode(state.owner_uid, bl);
 716       decode(state.owner_gid, bl);
 717       decode(state.unix_mode, bl);
 718       ceph::real_time enc_time;
 719       for (auto t : { &(state.ctime), &(state.mtime), &(state.atime) }) {
 720         decode(enc_time, bl);
 721         *t = real_clock::to_timespec(enc_time);
 722       }
 723       if (struct_v >= 2) {
 724         decode(state.version, bl);
 725       }
 726       DECODE_FINISH(bl);
 727     }
 728
 729     void encode_attrs(ceph::buffer::list& ux_key1,
 730                       ceph::buffer::list& ux_attrs1);
 731
 732     DecodeAttrsResult decode_attrs(const ceph::buffer::list* ux_key1,
 733                                    const ceph::buffer::list* ux_attrs1);
 734
 735     void invalidate();
 736
 737     bool reclaim(const cohort::lru::ObjectFactory* newobj_fac) override;
 738
 739     typedef cohort::lru::LRU<std::mutex> FhLRU;
 740
 741     struct FhLT
 742     {
 743       // for internal ordering
 744       bool operator()(const RGWFileHandle& lhs, const RGWFileHandle& rhs) const
 745         { return (lhs.get_key() < rhs.get_key()); }
 746
 747       // for external search by fh_key
 748       bool operator()(const fh_key& k, const RGWFileHandle& fh) const
 749         { return k < fh.get_key(); }
 750
 751       bool operator()(const RGWFileHandle& fh, const fh_key& k) const
 752         { return fh.get_key() < k; }
 753     };
 754
 755     struct FhEQ
 756     {
 757       bool operator()(const RGWFileHandle& lhs, const RGWFileHandle& rhs) const
 758         { return (lhs.get_key() == rhs.get_key()); }
 759
 760       bool operator()(const fh_key& k, const RGWFileHandle& fh) const
 761         { return k == fh.get_key(); }
 762
 763       bool operator()(const RGWFileHandle& fh, const fh_key& k) const
 764         { return fh.get_key() == k; }
 765     };
 766
 767     typedef bi::link_mode<bi::safe_link> link_mode; /* XXX normal */
 768 #if defined(FHCACHE_AVL)
 769     typedef bi::avl_set_member_hook<link_mode> tree_hook_type;
 770 #else
 771     /* RBT */
 772     typedef bi::set_member_hook<link_mode> tree_hook_type;
 773 #endif
 774     tree_hook_type fh_hook;
 775
 776     typedef bi::member_hook<
 777       RGWFileHandle, tree_hook_type, &RGWFileHandle::fh_hook> FhHook;
 778
 779 #if defined(FHCACHE_AVL)
 780     typedef bi::avltree<RGWFileHandle, bi::compare<FhLT>, FhHook> FHTree;
 781 #else
 782     typedef bi::rbtree<RGWFileHandle, bi::compare<FhLT>, FhHook> FhTree;
 783 #endif
 784     typedef cohort::lru::TreeX<RGWFileHandle, FhTree, FhLT, FhEQ, fh_key,
 785                                std::mutex> FHCache;
 786
 787     ~RGWFileHandle() override;
 788
 789     friend std::ostream& operator<<(std::ostream &os,
 790                                     RGWFileHandle const &rgw_fh);
 791
 792     class Factory : public cohort::lru::ObjectFactory
 793     {
 794     public:
 795       RGWLibFS* fs;
 796       RGWFileHandle* parent;
 797       const fh_key& fhk;
 798       std::string& name;
 799       uint32_t flags;
 800
 801       Factory() = delete;
 802
 803       Factory(RGWLibFS* _fs, RGWFileHandle* _parent,
 804               const fh_key& _fhk, std::string& _name, uint32_t _flags)
 805         : fs(_fs), parent(_parent), fhk(_fhk), name(_name),
 806           flags(_flags) {}
 807
 808       void recycle (cohort::lru::Object* o) override {
 809         /* re-use an existing object */
 810         o->~Object(); // call lru::Object virtual dtor
 811         // placement new!
 812         new (o) RGWFileHandle(fs, parent, fhk, name, flags);
 813       }
 814
 815       cohort::lru::Object* alloc() override {
 816         return new RGWFileHandle(fs, parent, fhk, name, flags);
 817       }
 818     }; /* Factory */
 819
 820   }; /* RGWFileHandle */
 821
 822   WRITE_CLASS_ENCODER(RGWFileHandle);
 823
 824   inline RGWFileHandle* get_rgwfh(struct rgw_file_handle* fh) {
 825     return static_cast<RGWFileHandle*>(fh->fh_private);
 826   }
 827
 828   inline enum rgw_fh_type fh_type_of(uint32_t flags) {
 829     enum rgw_fh_type fh_type;
 830     switch(flags & RGW_LOOKUP_TYPE_FLAGS)
 831     {
 832     case RGW_LOOKUP_FLAG_DIR:
 833       fh_type = RGW_FS_TYPE_DIRECTORY;
 834       break;
 835     case RGW_LOOKUP_FLAG_FILE:
 836       fh_type = RGW_FS_TYPE_FILE;
 837       break;
 838     default:
 839       fh_type = RGW_FS_TYPE_NIL;
 840     };
 841     return fh_type;
 842   }
 843
 844   typedef std::tuple<RGWFileHandle*, uint32_t> LookupFHResult;
 845   typedef std::tuple<RGWFileHandle*, int> MkObjResult;
 846
 847   class RGWLibFS
 848   {
 849     CephContext* cct;
 850     struct rgw_fs fs{};
 851     RGWFileHandle root_fh;
 852     rgw_fh_callback_t invalidate_cb;
 853     void *invalidate_arg;
 854     bool shutdown;
 855
 856     mutable std::atomic<uint64_t> refcnt;
 857
 858     RGWFileHandle::FHCache fh_cache;
 859     RGWFileHandle::FhLRU fh_lru;
 860
 861     std::string uid; // should match user.user_id, iiuc
 862
 863     std::unique_ptr<rgw::sal::User> user;
 864     RGWAccessKey key; // XXXX acc_key
 865
 866     static std::atomic<uint32_t> fs_inst_counter;
 867
 868     static uint32_t write_completion_interval_s;
 869
 870     using lock_guard = std::lock_guard<std::mutex>;
 871     using unique_lock = std::unique_lock<std::mutex>;
 872
 873     struct event
 874     {
 875       enum class type : uint8_t { READDIR } ;
 876       type t;
 877       const fh_key fhk;
 878       struct timespec ts;
 879       event(type t, const fh_key& k, const struct timespec& ts)
 880         : t(t), fhk(k), ts(ts) {}
 881     };
 882
 883     friend std::ostream& operator<<(std::ostream &os,
 884                                     RGWLibFS::event const &ev);
 885
 886     using event_vector = /* boost::small_vector<event, 16> */
 887       std::vector<event>;
 888
 889     struct WriteCompletion
 890     {
 891       RGWFileHandle& rgw_fh;
 892
 893       explicit WriteCompletion(RGWFileHandle& _fh) : rgw_fh(_fh) {
 894         rgw_fh.get_fs()->ref(&rgw_fh);
 895       }
 896
 897       void operator()() {
 898         rgw_fh.close(); /* will finish in-progress write */
 899         rgw_fh.get_fs()->unref(&rgw_fh);
 900       }
 901     };
 902
 903     static ceph::timer<ceph::mono_clock> write_timer;
 904
 905     struct State {
 906       std::mutex mtx;
 907       std::atomic<uint32_t> flags;
 908       std::deque<event> events;
 909
 910       State() : flags(0) {}
 911
 912       void push_event(const event& ev) {
 913         events.push_back(ev);
 914       }
 915     } state;
 916
 917     uint32_t new_inst() {
 918       return ++fs_inst_counter;
 919     }
 920
 921     friend class RGWFileHandle;
 922     friend class RGWLibProcess;
 923
 924   public:
 925
 926     static constexpr uint32_t FLAG_NONE =      0x0000;
 927     static constexpr uint32_t FLAG_CLOSED =    0x0001;
 928
 929     struct BucketStats {
 930       size_t size;
 931       size_t size_rounded;
 932       real_time creation_time;
 933       uint64_t num_entries;
 934     };
 935
 936     RGWLibFS(CephContext* _cct, const char *_uid, const char *_user_id,
 937             const char* _key, const char *root)
 938       : cct(_cct), root_fh(this), invalidate_cb(nullptr),
 939         invalidate_arg(nullptr), shutdown(false), refcnt(1),
 940         fh_cache(cct->_conf->rgw_nfs_fhcache_partitions,
 941                  cct->_conf->rgw_nfs_fhcache_size),
 942         fh_lru(cct->_conf->rgw_nfs_lru_lanes,
 943                cct->_conf->rgw_nfs_lru_lane_hiwat),
 944         uid(_uid), key(_user_id, _key) {
 945
 946       if (!root || !strcmp(root, "/")) {
 947         root_fh.init_rootfs(uid, RGWFileHandle::root_name, false);
 948       } else {
 949         root_fh.init_rootfs(uid, root, true);
 950       }
 951
 952       /* pointer to self */
 953       fs.fs_private = this;
 954
 955       /* expose public root fh */
 956       fs.root_fh = root_fh.get_fh();
 957
 958       new_inst();
 959     }
 960
 961     friend void intrusive_ptr_add_ref(const RGWLibFS* fs) {
 962       fs->refcnt.fetch_add(1, std::memory_order_relaxed);
 963     }
 964
 965     friend void intrusive_ptr_release(const RGWLibFS* fs) {
 966       if (fs->refcnt.fetch_sub(1, std::memory_order_release) == 0) {
 967         std::atomic_thread_fence(std::memory_order_acquire);
 968         delete fs;
 969       }
 970     }
 971
 972     RGWLibFS* ref() {
 973       intrusive_ptr_add_ref(this);
 974       return this;
 975     }
 976
 977     inline void rele() {
 978       intrusive_ptr_release(this);
 979     }
 980
 981     void stop() { shutdown = true; }
 982
 983     void release_evict(RGWFileHandle* fh) {
 984       /* remove from cache, releases sentinel ref */
 985       fh_cache.remove(fh->fh.fh_hk.object, fh,
 986                       RGWFileHandle::FHCache::FLAG_LOCK);
 987       /* release call-path ref */
 988       (void) fh_lru.unref(fh, cohort::lru::FLAG_NONE);
 989     }
 990
 991     int authorize(const DoutPrefixProvider *dpp, rgw::sal::Store* store) {
 992       int ret = store->get_user_by_access_key(dpp, key.id, null_yield, &user);
 993       if (ret == 0) {
 994         RGWAccessKey* k = user->get_info().get_key(key.id);
 995         if (!k || (k->key != key.key))
 996           return -EINVAL;
 997         if (user->get_info().suspended)
 998           return -ERR_USER_SUSPENDED;
 999       } else {
1000         /* try external authenticators (ldap for now) */
1001         rgw::LDAPHelper* ldh = rgwlib.get_ldh(); /* !nullptr */
1002         RGWToken token;
1003         /* boost filters and/or string_ref may throw on invalid input */
1004         try {
1005           token = rgw::from_base64(key.id);
1006         } catch(...) {
1007           token = std::string("");
1008         }
1009         if (token.valid() && (ldh->auth(token.id, token.key) == 0)) {
1010           /* try to store user if it doesn't already exist */
1011           if (user->load_user(dpp, null_yield) < 0) {
1012             int ret = user->store_user(dpp, null_yield, true);
1013             if (ret < 0) {
1014               lsubdout(get_context(), rgw, 10)
1015                 << "NOTICE: failed to store new user's info: ret=" << ret
1016                 << dendl;
1017             }
1018           }
1019         } /* auth success */
1020       }
1021       return ret;
1022     } /* authorize */
1023
1024     int register_invalidate(rgw_fh_callback_t cb, void *arg, uint32_t flags) {
1025       invalidate_cb = cb;
1026       invalidate_arg = arg;
1027       return 0;
1028     }
1029
1030     /* find RGWFileHandle by id  */
1031     LookupFHResult lookup_fh(const fh_key& fhk,
1032                              const uint32_t flags = RGWFileHandle::FLAG_NONE) {
1033       using std::get;
1034
1035       // cast int32_t(RGWFileHandle::FLAG_NONE) due to strictness of Clang
1036       // the cast transfers a lvalue into a rvalue  in the ctor
1037       // check the commit message for the full details
1038       LookupFHResult fhr { nullptr, uint32_t(RGWFileHandle::FLAG_NONE) };
1039
1040       RGWFileHandle::FHCache::Latch lat;
1041       bool fh_locked = flags & RGWFileHandle::FLAG_LOCKED;
1042
1043     retry:
1044       RGWFileHandle* fh =
1045         fh_cache.find_latch(fhk.fh_hk.object /* partition selector*/,
1046                             fhk /* key */, lat /* serializer */,
1047                             RGWFileHandle::FHCache::FLAG_LOCK);
1048       /* LATCHED */
1049       if (fh) {
1050         if (likely(! fh_locked))
1051             fh->mtx.lock(); // XXX !RAII because may-return-LOCKED
1052         /* need initial ref from LRU (fast path) */
1053         if (! fh_lru.ref(fh, cohort::lru::FLAG_INITIAL)) {
1054           lat.lock->unlock();
1055           if (likely(! fh_locked))
1056             fh->mtx.unlock();
1057           goto retry; /* !LATCHED */
1058         }
1059         /* LATCHED, LOCKED */
1060         if (! (flags & RGWFileHandle::FLAG_LOCK))
1061           fh->mtx.unlock(); /* ! LOCKED */
1062       }
1063       lat.lock->unlock(); /* !LATCHED */
1064       get<0>(fhr) = fh;
1065       if (fh) {
1066             lsubdout(get_context(), rgw, 17)
1067               << __func__ << " 1 " << *fh
1068               << dendl;
1069       }
1070       return fhr;
1071     } /* lookup_fh(const fh_key&) */
1072
1073     /* find or create an RGWFileHandle */
1074     LookupFHResult lookup_fh(RGWFileHandle* parent, const char *name,
1075                              const uint32_t flags = RGWFileHandle::FLAG_NONE) {
1076       using std::get;
1077
1078       // cast int32_t(RGWFileHandle::FLAG_NONE) due to strictness of Clang
1079       // the cast transfers a lvalue into a rvalue  in the ctor
1080       // check the commit message for the full details
1081       LookupFHResult fhr { nullptr, uint32_t(RGWFileHandle::FLAG_NONE) };
1082
1083       /* mount is stale? */
1084       if (state.flags & FLAG_CLOSED)
1085         return fhr;
1086
1087       RGWFileHandle::FHCache::Latch lat;
1088       bool fh_locked = flags & RGWFileHandle::FLAG_LOCKED;
1089
1090       std::string obj_name{name};
1091       std::string key_name{parent->make_key_name(name)};
1092       fh_key fhk = parent->make_fhk(obj_name);
1093
1094       lsubdout(get_context(), rgw, 10)
1095         << __func__ << " called on "
1096         << parent->object_name() << " for " << key_name
1097         << " (" << obj_name << ")"
1098         << " -> " << fhk
1099         << dendl;
1100
1101     retry:
1102       RGWFileHandle* fh =
1103         fh_cache.find_latch(fhk.fh_hk.object /* partition selector*/,
1104                             fhk /* key */, lat /* serializer */,
1105                             RGWFileHandle::FHCache::FLAG_LOCK);
1106       /* LATCHED */
1107       if (fh) {
1108         if (likely(! fh_locked))
1109           fh->mtx.lock(); // XXX !RAII because may-return-LOCKED
1110         if (fh->flags & RGWFileHandle::FLAG_DELETED) {
1111           /* for now, delay briefly and retry */
1112           lat.lock->unlock();
1113           if (likely(! fh_locked))
1114             fh->mtx.unlock();
1115           std::this_thread::sleep_for(std::chrono::milliseconds(20));
1116           goto retry; /* !LATCHED */
1117         }
1118         /* need initial ref from LRU (fast path) */
1119         if (! fh_lru.ref(fh, cohort::lru::FLAG_INITIAL)) {
1120           lat.lock->unlock();
1121           if (likely(! fh_locked))
1122             fh->mtx.unlock();
1123           goto retry; /* !LATCHED */
1124         }
1125         /* LATCHED, LOCKED */
1126         if (! (flags & RGWFileHandle::FLAG_LOCK))
1127           if (likely(! fh_locked))
1128             fh->mtx.unlock(); /* ! LOCKED */
1129       } else {
1130         /* make or re-use handle */
1131         RGWFileHandle::Factory prototype(this, parent, fhk,
1132                                          obj_name, CREATE_FLAGS(flags));
1133         uint32_t iflags{cohort::lru::FLAG_INITIAL};
1134         fh = static_cast<RGWFileHandle*>(
1135           fh_lru.insert(&prototype,
1136                         cohort::lru::Edge::MRU,
1137                         iflags));
1138         if (fh) {
1139           /* lock fh (LATCHED) */
1140           if (flags & RGWFileHandle::FLAG_LOCK)
1141             fh->mtx.lock();
1142           if (likely(! (iflags & cohort::lru::FLAG_RECYCLE))) {
1143             /* inserts at cached insert iterator, releasing latch */
1144             fh_cache.insert_latched(
1145               fh, lat, RGWFileHandle::FHCache::FLAG_UNLOCK);
1146           } else {
1147             /* recycle step invalidates Latch */
1148             fh_cache.insert(
1149               fhk.fh_hk.object, fh, RGWFileHandle::FHCache::FLAG_NONE);
1150             lat.lock->unlock(); /* !LATCHED */
1151           }
1152           get<1>(fhr) |= RGWFileHandle::FLAG_CREATE;
1153           /* ref parent (non-initial ref cannot fail on valid object) */
1154           if (! parent->is_mount()) {
1155             (void) fh_lru.ref(parent, cohort::lru::FLAG_NONE);
1156           }
1157           goto out; /* !LATCHED */
1158         } else {
1159           lat.lock->unlock();
1160           goto retry; /* !LATCHED */
1161         }
1162       }
1163       lat.lock->unlock(); /* !LATCHED */
1164     out:
1165       get<0>(fhr) = fh;
1166       if (fh) {
1167             lsubdout(get_context(), rgw, 17)
1168               << __func__ << " 2 " << *fh
1169               << dendl;
1170       }
1171       return fhr;
1172     } /*  lookup_fh(RGWFileHandle*, const char *, const uint32_t) */
1173
1174     inline void unref(RGWFileHandle* fh) {
1175       if (likely(! fh->is_mount())) {
1176         (void) fh_lru.unref(fh, cohort::lru::FLAG_NONE);
1177       }
1178     }
1179
1180     inline RGWFileHandle* ref(RGWFileHandle* fh) {
1181       if (likely(! fh->is_mount())) {
1182         fh_lru.ref(fh, cohort::lru::FLAG_NONE);
1183       }
1184       return fh;
1185     }
1186
1187     int getattr(RGWFileHandle* rgw_fh, struct stat* st);
1188
1189     int setattr(RGWFileHandle* rgw_fh, struct stat* st, uint32_t mask,
1190                 uint32_t flags);
1191
1192     int getxattrs(RGWFileHandle* rgw_fh, rgw_xattrlist* attrs,
1193                   rgw_getxattr_cb cb, void *cb_arg, uint32_t flags);
1194
1195     int lsxattrs(RGWFileHandle* rgw_fh, rgw_xattrstr *filter_prefix,
1196                  rgw_getxattr_cb cb, void *cb_arg, uint32_t flags);
1197
1198     int setxattrs(RGWFileHandle* rgw_fh, rgw_xattrlist* attrs, uint32_t flags);
1199
1200     int rmxattrs(RGWFileHandle* rgw_fh, rgw_xattrlist* attrs, uint32_t flags);
1201
1202     void update_fh(RGWFileHandle *rgw_fh);
1203
1204     LookupFHResult stat_bucket(RGWFileHandle* parent, const char *path,
1205                                RGWLibFS::BucketStats& bs,
1206                                uint32_t flags);
1207
1208     LookupFHResult fake_leaf(RGWFileHandle* parent, const char *path,
1209                              enum rgw_fh_type type = RGW_FS_TYPE_NIL,
1210                              struct stat *st = nullptr, uint32_t mask = 0,
1211                              uint32_t flags = RGWFileHandle::FLAG_NONE);
1212
1213     LookupFHResult stat_leaf(RGWFileHandle* parent, const char *path,
1214                              enum rgw_fh_type type = RGW_FS_TYPE_NIL,
1215                              uint32_t flags = RGWFileHandle::FLAG_NONE);
1216
1217     int read(RGWFileHandle* rgw_fh, uint64_t offset, size_t length,
1218              size_t* bytes_read, void* buffer, uint32_t flags);
1219
1220     int readlink(RGWFileHandle* rgw_fh, uint64_t offset, size_t length,
1221              size_t* bytes_read, void* buffer, uint32_t flags);
1222
1223     int rename(RGWFileHandle* old_fh, RGWFileHandle* new_fh,
1224                const char *old_name, const char *new_name);
1225
1226     MkObjResult create(RGWFileHandle* parent, const char *name, struct stat *st,
1227                       uint32_t mask, uint32_t flags);
1228
1229     MkObjResult symlink(RGWFileHandle* parent, const char *name,
1230                const char *link_path, struct stat *st, uint32_t mask, uint32_t flags);
1231
1232     MkObjResult mkdir(RGWFileHandle* parent, const char *name, struct stat *st,
1233                       uint32_t mask, uint32_t flags);
1234
1235     int unlink(RGWFileHandle* rgw_fh, const char *name,
1236                uint32_t flags = FLAG_NONE);
1237
1238     /* find existing RGWFileHandle */
1239     RGWFileHandle* lookup_handle(struct rgw_fh_hk fh_hk) {
1240
1241       if (state.flags & FLAG_CLOSED)
1242         return nullptr;
1243
1244       RGWFileHandle::FHCache::Latch lat;
1245       fh_key fhk(fh_hk);
1246
1247     retry:
1248       RGWFileHandle* fh =
1249         fh_cache.find_latch(fhk.fh_hk.object /* partition selector*/,
1250                             fhk /* key */, lat /* serializer */,
1251                             RGWFileHandle::FHCache::FLAG_LOCK);
1252       /* LATCHED */
1253       if (! fh) {
1254         if (unlikely(fhk == root_fh.fh.fh_hk)) {
1255           /* lookup for root of this fs */
1256           fh = &root_fh;
1257           goto out;
1258         }
1259         lsubdout(get_context(), rgw, 0)
1260           << __func__ << " handle lookup failed " << fhk
1261           << dendl;
1262         goto out;
1263       }
1264       fh->mtx.lock();
1265       if (fh->flags & RGWFileHandle::FLAG_DELETED) {
1266         /* for now, delay briefly and retry */
1267         lat.lock->unlock();
1268         fh->mtx.unlock(); /* !LOCKED */
1269         std::this_thread::sleep_for(std::chrono::milliseconds(20));
1270         goto retry; /* !LATCHED */
1271       }
1272       if (! fh_lru.ref(fh, cohort::lru::FLAG_INITIAL)) {
1273         lat.lock->unlock();
1274         fh->mtx.unlock();
1275         goto retry; /* !LATCHED */
1276       }
1277       /* LATCHED */
1278       fh->mtx.unlock(); /* !LOCKED */
1279     out:
1280       lat.lock->unlock(); /* !LATCHED */
1281
1282       /* special case:  lookup root_fh */
1283       if (! fh) {
1284         if (unlikely(fh_hk == root_fh.fh.fh_hk)) {
1285           fh = &root_fh;
1286         }
1287       }
1288
1289       return fh;
1290     }
1291
1292     CephContext* get_context() {
1293       return cct;
1294     }
1295
1296     struct rgw_fs* get_fs() { return &fs; }
1297
1298     RGWFileHandle& get_fh() { return root_fh; }
1299
1300     uint64_t get_fsid() { return root_fh.state.dev; }
1301
1302     RGWUserInfo* get_user() { return &user->get_info(); }
1303
1304     void update_user(const DoutPrefixProvider *dpp) {
1305       (void) rgwlib.get_store()->get_user_by_access_key(dpp, key.id, null_yield, &user);
1306     }
1307
1308     void close();
1309     void gc();
1310   }; /* RGWLibFS */
1311
1312 static inline std::string make_uri(const std::string& bucket_name,
1313                                    const std::string& object_name) {
1314   std::string uri("/");
1315   uri.reserve(bucket_name.length() + object_name.length() + 2);
1316   uri += bucket_name;
1317   uri += "/";
1318   uri += object_name;
1319   return uri;
1320 }
1321
1322 /*
1323   read directory content (buckets)
1324 */
1325
1326 class RGWListBucketsRequest : public RGWLibRequest,
1327                               public RGWListBuckets /* RGWOp */
1328 {
1329 public:
1330   RGWFileHandle* rgw_fh;
1331   RGWFileHandle::readdir_offset offset;
1332   void* cb_arg;
1333   rgw_readdir_cb rcb;
1334   uint64_t* ioff;
1335   size_t ix;
1336   uint32_t d_count;
1337   bool rcb_eof; // caller forced early stop in readdir cycle
1338
1339   RGWListBucketsRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
1340                         RGWFileHandle* _rgw_fh, rgw_readdir_cb _rcb,
1341                         void* _cb_arg, RGWFileHandle::readdir_offset& _offset)
1342     : RGWLibRequest(_cct, std::move(_user)), rgw_fh(_rgw_fh), offset(_offset),
1343       cb_arg(_cb_arg), rcb(_rcb), ioff(nullptr), ix(0), d_count(0),
1344       rcb_eof(false) {
1345
1346     using boost::get;
1347
1348     if (unlikely(!! get<uint64_t*>(&offset))) {
1349       ioff = get<uint64_t*>(offset);
1350       const auto& mk = rgw_fh->find_marker(*ioff);
1351       if (mk) {
1352         marker = mk->name;
1353       }
1354     } else {
1355       const char* mk = get<const char*>(offset);
1356       if (mk) {
1357         marker = mk;
1358       }
1359     }
1360     op = this;
1361   }
1362
1363   bool only_bucket() override { return false; }
1364
1365   int op_init() override {
1366     // assign store, s, and dialect_handler
1367     RGWObjectCtx* rados_ctx
1368       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
1369     // framework promises to call op_init after parent init
1370     ceph_assert(rados_ctx);
1371     RGWOp::init(rados_ctx->get_store(), get_state(), this);
1372     op = this; // assign self as op: REQUIRED
1373     return 0;
1374   }
1375
1376   int header_init() override {
1377     struct req_state* state = get_state();
1378     state->info.method = "GET";
1379     state->op = OP_GET;
1380
1381     /* XXX derp derp derp */
1382     state->relative_uri = "/";
1383     state->info.request_uri = "/"; // XXX
1384     state->info.effective_uri = "/";
1385     state->info.request_params = "";
1386     state->info.domain = ""; /* XXX ? */
1387
1388     return 0;
1389   }
1390
1391   int get_params(optional_yield) override {
1392     limit = -1; /* no limit */
1393     return 0;
1394   }
1395
1396   void send_response_begin(bool has_buckets) override {
1397     sent_data = true;
1398   }
1399
1400   void send_response_data(rgw::sal::BucketList& buckets) override {
1401     if (!sent_data)
1402       return;
1403     auto& m = buckets.get_buckets();
1404     for (const auto& iter : m) {
1405       std::string_view marker{iter.first};
1406       auto& ent = iter.second;
1407       if (! this->operator()(ent->get_name(), marker)) {
1408         /* caller cannot accept more */
1409         lsubdout(cct, rgw, 5) << "ListBuckets rcb failed"
1410                               << " dirent=" << ent->get_name()
1411                               << " call count=" << ix
1412                               << dendl;
1413         rcb_eof = true;
1414         return;
1415       }
1416       ++ix;
1417     }
1418   } /* send_response_data */
1419
1420   void send_response_end() override {
1421     // do nothing
1422   }
1423
1424   int operator()(const std::string_view& name,
1425                  const std::string_view& marker) {
1426     uint64_t off = XXH64(name.data(), name.length(), fh_key::seed);
1427     if (!! ioff) {
1428       *ioff = off;
1429     }
1430     /* update traversal cache */
1431     rgw_fh->add_marker(off, rgw_obj_key{marker.data(), ""},
1432                        RGW_FS_TYPE_DIRECTORY);
1433     ++d_count;
1434     return rcb(name.data(), cb_arg, off, nullptr, 0, RGW_LOOKUP_FLAG_DIR);
1435   }
1436
1437   bool eof() {
1438     using boost::get;
1439
1440     if (unlikely(cct->_conf->subsys.should_gather(ceph_subsys_rgw, 15))) {
1441       bool is_offset =
1442         unlikely(! get<const char*>(&offset)) ||
1443         !! get<const char*>(offset);
1444       lsubdout(cct, rgw, 15) << "READDIR offset: " <<
1445         ((is_offset) ? offset : "(nil)")
1446                              << " is_truncated: " << is_truncated
1447                              << dendl;
1448     }
1449     return !is_truncated && !rcb_eof;
1450   }
1451
1452 }; /* RGWListBucketsRequest */
1453
1454 /*
1455   read directory content (bucket objects)
1456 */
1457
1458 class RGWReaddirRequest : public RGWLibRequest,
1459                           public RGWListBucket /* RGWOp */
1460 {
1461 public:
1462   RGWFileHandle* rgw_fh;
1463   RGWFileHandle::readdir_offset offset;
1464   void* cb_arg;
1465   rgw_readdir_cb rcb;
1466   uint64_t* ioff;
1467   size_t ix;
1468   uint32_t d_count;
1469   bool rcb_eof; // caller forced early stop in readdir cycle
1470
1471   RGWReaddirRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
1472                     RGWFileHandle* _rgw_fh, rgw_readdir_cb _rcb,
1473                     void* _cb_arg, RGWFileHandle::readdir_offset& _offset)
1474     : RGWLibRequest(_cct, std::move(_user)), rgw_fh(_rgw_fh), offset(_offset),
1475       cb_arg(_cb_arg), rcb(_rcb), ioff(nullptr), ix(0), d_count(0),
1476       rcb_eof(false) {
1477
1478     using boost::get;
1479
1480     if (unlikely(!! get<uint64_t*>(&offset))) {
1481       ioff = get<uint64_t*>(offset);
1482       const auto& mk = rgw_fh->find_marker(*ioff);
1483       if (mk) {
1484         marker = *mk;
1485       }
1486     } else {
1487       const char* mk = get<const char*>(offset);
1488       if (mk) {
1489         std::string tmark{rgw_fh->relative_object_name()};
1490         if (tmark.length() > 0)
1491           tmark += "/";
1492         tmark += mk;
1493         marker = rgw_obj_key{std::move(tmark), "", ""};
1494       }
1495     }
1496
1497     default_max = 1000; // XXX was being omitted
1498     op = this;
1499   }
1500
1501   bool only_bucket() override { return true; }
1502
1503   int op_init() override {
1504     // assign store, s, and dialect_handler
1505     RGWObjectCtx* rados_ctx
1506       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
1507     // framework promises to call op_init after parent init
1508     ceph_assert(rados_ctx);
1509     RGWOp::init(rados_ctx->get_store(), get_state(), this);
1510     op = this; // assign self as op: REQUIRED
1511     return 0;
1512   }
1513
1514   int header_init() override {
1515     struct req_state* state = get_state();
1516     state->info.method = "GET";
1517     state->op = OP_GET;
1518
1519     /* XXX derp derp derp */
1520     std::string uri = "/" + rgw_fh->bucket_name() + "/";
1521     state->relative_uri = uri;
1522     state->info.request_uri = uri; // XXX
1523     state->info.effective_uri = uri;
1524     state->info.request_params = "";
1525     state->info.domain = ""; /* XXX ? */
1526
1527     prefix = rgw_fh->relative_object_name();
1528     if (prefix.length() > 0)
1529       prefix += "/";
1530     delimiter = '/';
1531
1532     return 0;
1533   }
1534
1535   int operator()(const std::string_view name, const rgw_obj_key& marker,
1536                  const ceph::real_time& t, const uint64_t fsz, uint8_t type) {
1537
1538     assert(name.length() > 0); // all cases handled in callers
1539
1540     /* hash offset of name in parent (short name) for NFS readdir cookie */
1541     uint64_t off = XXH64(name.data(), name.length(), fh_key::seed);
1542     if (unlikely(!! ioff)) {
1543       *ioff = off;
1544     }
1545
1546     /* update traversal cache */
1547     rgw_fh->add_marker(off, marker, type);
1548     ++d_count;
1549
1550     /* set c/mtime and size from bucket index entry */
1551     struct stat st = {};
1552 #ifdef HAVE_STAT_ST_MTIMESPEC_TV_NSEC
1553     st.st_atimespec = ceph::real_clock::to_timespec(t);
1554     st.st_mtimespec = st.st_atimespec;
1555     st.st_ctimespec = st.st_atimespec;
1556 #else
1557     st.st_atim = ceph::real_clock::to_timespec(t);
1558     st.st_mtim = st.st_atim;
1559     st.st_ctim = st.st_atim;
1560 #endif
1561     st.st_size = fsz;
1562
1563     return rcb(name.data(), cb_arg, off, &st, RGWFileHandle::RCB_MASK,
1564                (type == RGW_FS_TYPE_DIRECTORY) ?
1565                RGW_LOOKUP_FLAG_DIR :
1566                RGW_LOOKUP_FLAG_FILE);
1567   }
1568
1569   int get_params(optional_yield) override {
1570     max = default_max;
1571     return 0;
1572   }
1573
1574   void send_response() override {
1575     struct req_state* state = get_state();
1576     auto cnow = real_clock::now();
1577
1578     /* enumerate objs and common_prefixes in parallel,
1579      * avoiding increment on and end iterator, which is
1580      * undefined */
1581
1582     class DirIterator
1583     {
1584       std::vector<rgw_bucket_dir_entry>& objs;
1585       std::vector<rgw_bucket_dir_entry>::iterator obj_iter;
1586
1587       std::map<std::string, bool>& common_prefixes;
1588       std::map<string, bool>::iterator cp_iter;
1589
1590       boost::optional<std::string_view> obj_sref;
1591       boost::optional<std::string_view> cp_sref;
1592       bool _skip_cp;
1593
1594     public:
1595
1596       DirIterator(std::vector<rgw_bucket_dir_entry>& objs,
1597                   std::map<string, bool>& common_prefixes)
1598         : objs(objs), common_prefixes(common_prefixes), _skip_cp(false)
1599         {
1600           obj_iter = objs.begin();
1601           parse_obj();
1602           cp_iter = common_prefixes.begin();
1603           parse_cp();
1604         }
1605
1606       bool is_obj() {
1607         return (obj_iter != objs.end());
1608       }
1609
1610       bool is_cp(){
1611         return (cp_iter != common_prefixes.end());
1612       }
1613
1614       bool eof() {
1615         return ((!is_obj()) && (!is_cp()));
1616       }
1617
1618       void parse_obj() {
1619         if (is_obj()) {
1620           std::string_view sref{obj_iter->key.name};
1621           size_t last_del = sref.find_last_of('/');
1622           if (last_del != string::npos)
1623             sref.remove_prefix(last_del+1);
1624           obj_sref = sref;
1625         }
1626       } /* parse_obj */
1627
1628       void next_obj() {
1629         ++obj_iter;
1630         parse_obj();
1631       }
1632
1633       void parse_cp() {
1634         if (is_cp()) {
1635           /* leading-/ skip case */
1636           if (cp_iter->first == "/") {
1637             _skip_cp = true;
1638             return;
1639           } else
1640             _skip_cp = false;
1641
1642           /* it's safest to modify the element in place--a suffix-modifying
1643            * string_ref operation is problematic since ULP rgw_file callers
1644            * will ultimately need a c-string */
1645           if (cp_iter->first.back() == '/')
1646             const_cast<std::string&>(cp_iter->first).pop_back();
1647
1648           std::string_view sref{cp_iter->first};
1649           size_t last_del = sref.find_last_of('/');
1650           if (last_del != string::npos)
1651             sref.remove_prefix(last_del+1);
1652           cp_sref = sref;
1653         } /* is_cp */
1654       } /* parse_cp */
1655
1656       void next_cp() {
1657         ++cp_iter;
1658         parse_cp();
1659       }
1660
1661       bool skip_cp() {
1662         return _skip_cp;
1663       }
1664
1665       bool entry_is_obj() {
1666         return (is_obj() &&
1667                 ((! is_cp()) ||
1668                  (obj_sref.get() < cp_sref.get())));
1669       }
1670
1671       std::string_view get_obj_sref() {
1672         return obj_sref.get();
1673       }
1674
1675       std::string_view get_cp_sref() {
1676         return cp_sref.get();
1677       }
1678
1679       std::vector<rgw_bucket_dir_entry>::iterator& get_obj_iter() {
1680         return obj_iter;
1681       }
1682
1683       std::map<string, bool>::iterator& get_cp_iter() {
1684         return cp_iter;
1685       }
1686
1687     }; /* DirIterator */
1688
1689     DirIterator di{objs, common_prefixes};
1690
1691     for (;;) {
1692
1693       if (di.eof()) {
1694         break; // done
1695       }
1696
1697       /* assert: one of is_obj() || is_cp() holds */
1698       if (di.entry_is_obj()) {
1699         auto sref = di.get_obj_sref();
1700         if (sref.empty()) {
1701           /* recursive list of a leaf dir (iirc), do nothing */
1702         } else {
1703           /* send a file entry */
1704           auto obj_entry = *(di.get_obj_iter());
1705
1706           lsubdout(cct, rgw, 15) << "RGWReaddirRequest "
1707                                  << __func__ << " "
1708                                  << "list uri=" << state->relative_uri << " "
1709                                  << " prefix=" << prefix << " "
1710                                  << " obj path=" << obj_entry.key.name
1711                                  << " (" << sref << ")" << ""
1712                                  << " mtime="
1713                                  << real_clock::to_time_t(obj_entry.meta.mtime)
1714                                  << " size=" << obj_entry.meta.accounted_size
1715                                  << dendl;
1716
1717           if (! this->operator()(sref, next_marker, obj_entry.meta.mtime,
1718                                  obj_entry.meta.accounted_size,
1719                                  RGW_FS_TYPE_FILE)) {
1720             /* caller cannot accept more */
1721             lsubdout(cct, rgw, 5) << "readdir rcb caller signalled stop"
1722                                   << " dirent=" << sref.data()
1723                                   << " call count=" << ix
1724                                   << dendl;
1725             rcb_eof = true;
1726             return;
1727           }
1728         }
1729         di.next_obj(); // and advance object
1730       } else {
1731         /* send a dir entry */
1732         if (! di.skip_cp()) {
1733           auto sref = di.get_cp_sref();
1734
1735           lsubdout(cct, rgw, 15) << "RGWReaddirRequest "
1736                                  << __func__ << " "
1737                                  << "list uri=" << state->relative_uri << " "
1738                                  << " prefix=" << prefix << " "
1739                                  << " cpref=" << sref
1740                                  << dendl;
1741
1742           if (sref.empty()) {
1743             /* null path segment--could be created in S3 but has no NFS
1744              * interpretation */
1745           } else {
1746             if (! this->operator()(sref, next_marker, cnow, 0,
1747                                    RGW_FS_TYPE_DIRECTORY)) {
1748               /* caller cannot accept more */
1749               lsubdout(cct, rgw, 5) << "readdir rcb caller signalled stop"
1750                                     << " dirent=" << sref.data()
1751                                     << " call count=" << ix
1752                                     << dendl;
1753               rcb_eof = true;
1754               return;
1755             }
1756           }
1757         }
1758         di.next_cp(); // and advance common_prefixes
1759       } /* ! di.entry_is_obj() */
1760     } /* for (;;) */
1761   }
1762
1763   virtual void send_versioned_response() {
1764     send_response();
1765   }
1766
1767   bool eof() {
1768     using boost::get;
1769
1770     if (unlikely(cct->_conf->subsys.should_gather(ceph_subsys_rgw, 15))) {
1771       bool is_offset =
1772         unlikely(! get<const char*>(&offset)) ||
1773         !! get<const char*>(offset);
1774       lsubdout(cct, rgw, 15) << "READDIR offset: " <<
1775         ((is_offset) ? offset : "(nil)")
1776                              << " next marker: " << next_marker
1777                              << " is_truncated: " << is_truncated
1778                              << dendl;
1779     }
1780     return !is_truncated && !rcb_eof;
1781   }
1782
1783 }; /* RGWReaddirRequest */
1784
1785 /*
1786   dir has-children predicate (bucket objects)
1787 */
1788
1789 class RGWRMdirCheck : public RGWLibRequest,
1790                       public RGWListBucket /* RGWOp */
1791 {
1792 public:
1793   const RGWFileHandle* rgw_fh;
1794   bool valid;
1795   bool has_children;
1796
1797   RGWRMdirCheck (CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
1798                  const RGWFileHandle* _rgw_fh)
1799     : RGWLibRequest(_cct, std::move(_user)), rgw_fh(_rgw_fh), valid(false),
1800       has_children(false) {
1801     default_max = 2;
1802     op = this;
1803   }
1804
1805   bool only_bucket() override { return true; }
1806
1807   int op_init() override {
1808     // assign store, s, and dialect_handler
1809     RGWObjectCtx* rados_ctx
1810       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
1811     // framework promises to call op_init after parent init
1812     ceph_assert(rados_ctx);
1813     RGWOp::init(rados_ctx->get_store(), get_state(), this);
1814     op = this; // assign self as op: REQUIRED
1815     return 0;
1816   }
1817
1818   int header_init() override {
1819     struct req_state* state = get_state();
1820     state->info.method = "GET";
1821     state->op = OP_GET;
1822
1823     std::string uri = "/" + rgw_fh->bucket_name() + "/";
1824     state->relative_uri = uri;
1825     state->info.request_uri = uri;
1826     state->info.effective_uri = uri;
1827     state->info.request_params = "";
1828     state->info.domain = ""; /* XXX ? */
1829
1830     prefix = rgw_fh->relative_object_name();
1831     if (prefix.length() > 0)
1832       prefix += "/";
1833     delimiter = '/';
1834
1835     return 0;
1836   }
1837
1838   int get_params(optional_yield) override {
1839     max = default_max;
1840     return 0;
1841   }
1842
1843   void send_response() override {
1844     valid = true;
1845     if ((objs.size() > 1) ||
1846         (! objs.empty() &&
1847          (objs.front().key.name != prefix))) {
1848       has_children = true;
1849       return;
1850     }
1851     for (auto& iter : common_prefixes) {
1852       /* readdir never produces a name for this case */
1853       if (iter.first == "/")
1854         continue;
1855       has_children = true;
1856       break;
1857     }
1858   }
1859
1860   virtual void send_versioned_response() {
1861     send_response();
1862   }
1863
1864 }; /* RGWRMdirCheck */
1865
1866 /*
1867   create bucket
1868 */
1869
1870 class RGWCreateBucketRequest : public RGWLibRequest,
1871                                public RGWCreateBucket /* RGWOp */
1872 {
1873 public:
1874   const std::string& bucket_name;
1875
1876   RGWCreateBucketRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
1877                         std::string& _bname)
1878     : RGWLibRequest(_cct, std::move(_user)), bucket_name(_bname) {
1879     op = this;
1880   }
1881
1882   bool only_bucket() override { return false; }
1883
1884   int read_permissions(RGWOp* op_obj, optional_yield) override {
1885     /* we ARE a 'create bucket' request (cf. rgw_rest.cc, ll. 1305-6) */
1886     return 0;
1887   }
1888
1889   int op_init() override {
1890     // assign store, s, and dialect_handler
1891     RGWObjectCtx* rados_ctx
1892       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
1893     // framework promises to call op_init after parent init
1894     ceph_assert(rados_ctx);
1895     RGWOp::init(rados_ctx->get_store(), get_state(), this);
1896     op = this; // assign self as op: REQUIRED
1897     return 0;
1898   }
1899
1900   int header_init() override {
1901
1902     struct req_state* state = get_state();
1903     state->info.method = "PUT";
1904     state->op = OP_PUT;
1905
1906     string uri = "/" + bucket_name;
1907     /* XXX derp derp derp */
1908     state->relative_uri = uri;
1909     state->info.request_uri = uri; // XXX
1910     state->info.effective_uri = uri;
1911     state->info.request_params = "";
1912     state->info.domain = ""; /* XXX ? */
1913
1914     return 0;
1915   }
1916
1917   int get_params(optional_yield) override {
1918     struct req_state* state = get_state();
1919     RGWAccessControlPolicy_S3 s3policy(state->cct);
1920     /* we don't have (any) headers, so just create canned ACLs */
1921     int ret = s3policy.create_canned(state->owner, state->bucket_owner, state->canned_acl);
1922     policy = s3policy;
1923     return ret;
1924   }
1925
1926   void send_response() override {
1927     /* TODO: something (maybe) */
1928   }
1929 }; /* RGWCreateBucketRequest */
1930
1931 /*
1932   delete bucket
1933 */
1934
1935 class RGWDeleteBucketRequest : public RGWLibRequest,
1936                                public RGWDeleteBucket /* RGWOp */
1937 {
1938 public:
1939   const std::string& bucket_name;
1940
1941   RGWDeleteBucketRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
1942                         std::string& _bname)
1943     : RGWLibRequest(_cct, std::move(_user)), bucket_name(_bname) {
1944     op = this;
1945   }
1946
1947   bool only_bucket() override { return true; }
1948
1949   int op_init() override {
1950     // assign store, s, and dialect_handler
1951     RGWObjectCtx* rados_ctx
1952       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
1953     // framework promises to call op_init after parent init
1954     ceph_assert(rados_ctx);
1955     RGWOp::init(rados_ctx->get_store(), get_state(), this);
1956     op = this; // assign self as op: REQUIRED
1957     return 0;
1958   }
1959
1960   int header_init() override {
1961
1962     struct req_state* state = get_state();
1963     state->info.method = "DELETE";
1964     state->op = OP_DELETE;
1965
1966     string uri = "/" + bucket_name;
1967     /* XXX derp derp derp */
1968     state->relative_uri = uri;
1969     state->info.request_uri = uri; // XXX
1970     state->info.effective_uri = uri;
1971     state->info.request_params = "";
1972     state->info.domain = ""; /* XXX ? */
1973
1974     return 0;
1975   }
1976
1977   void send_response() override {}
1978
1979 }; /* RGWDeleteBucketRequest */
1980
1981 /*
1982   put object
1983 */
1984 class RGWPutObjRequest : public RGWLibRequest,
1985                          public RGWPutObj /* RGWOp */
1986 {
1987 public:
1988   const std::string& bucket_name;
1989   const std::string& obj_name;
1990   buffer::list& bl; /* XXX */
1991   size_t bytes_written;
1992
1993   RGWPutObjRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
1994                   const std::string& _bname, const std::string& _oname,
1995                   buffer::list& _bl)
1996     : RGWLibRequest(_cct, std::move(_user)), bucket_name(_bname), obj_name(_oname),
1997       bl(_bl), bytes_written(0) {
1998     op = this;
1999   }
2000
2001   bool only_bucket() override { return true; }
2002
2003   int op_init() override {
2004     // assign store, s, and dialect_handler
2005     RGWObjectCtx* rados_ctx
2006       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2007     // framework promises to call op_init after parent init
2008     ceph_assert(rados_ctx);
2009     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2010     op = this; // assign self as op: REQUIRED
2011
2012     int rc = valid_s3_object_name(obj_name);
2013     if (rc != 0)
2014       return rc;
2015
2016     return 0;
2017   }
2018
2019   int header_init() override {
2020
2021     struct req_state* state = get_state();
2022     state->info.method = "PUT";
2023     state->op = OP_PUT;
2024
2025     /* XXX derp derp derp */
2026     std::string uri = make_uri(bucket_name, obj_name);
2027     state->relative_uri = uri;
2028     state->info.request_uri = uri; // XXX
2029     state->info.effective_uri = uri;
2030     state->info.request_params = "";
2031     state->info.domain = ""; /* XXX ? */
2032
2033     /* XXX required in RGWOp::execute() */
2034     state->content_length = bl.length();
2035
2036     return 0;
2037   }
2038
2039   int get_params(optional_yield) override {
2040     struct req_state* state = get_state();
2041     RGWAccessControlPolicy_S3 s3policy(state->cct);
2042     /* we don't have (any) headers, so just create canned ACLs */
2043     int ret = s3policy.create_canned(state->owner, state->bucket_owner, state->canned_acl);
2044     policy = s3policy;
2045     return ret;
2046   }
2047
2048   int get_data(buffer::list& _bl) override {
2049     /* XXX for now, use sharing semantics */
2050     _bl = std::move(bl);
2051     uint32_t len = _bl.length();
2052     bytes_written += len;
2053     return len;
2054   }
2055
2056   void send_response() override {}
2057
2058   int verify_params() override {
2059     if (bl.length() > cct->_conf->rgw_max_put_size)
2060       return -ERR_TOO_LARGE;
2061     return 0;
2062   }
2063
2064   buffer::list* get_attr(const std::string& k) {
2065     auto iter = attrs.find(k);
2066     return (iter != attrs.end()) ? &(iter->second) : nullptr;
2067   }
2068
2069 }; /* RGWPutObjRequest */
2070
2071 /*
2072   get object
2073 */
2074
2075 class RGWReadRequest : public RGWLibRequest,
2076                        public RGWGetObj /* RGWOp */
2077 {
2078 public:
2079   RGWFileHandle* rgw_fh;
2080   void *ulp_buffer;
2081   size_t nread;
2082   size_t read_resid; /* initialize to len, <= sizeof(ulp_buffer) */
2083   bool do_hexdump = false;
2084
2085   RGWReadRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
2086                  RGWFileHandle* _rgw_fh, uint64_t off, uint64_t len,
2087                  void *_ulp_buffer)
2088     : RGWLibRequest(_cct, std::move(_user)), rgw_fh(_rgw_fh), ulp_buffer(_ulp_buffer),
2089       nread(0), read_resid(len) {
2090     op = this;
2091
2092     /* fixup RGWGetObj (already know range parameters) */
2093     RGWGetObj::range_parsed = true;
2094     RGWGetObj::get_data = true; // XXX
2095     RGWGetObj::partial_content = true;
2096     RGWGetObj::ofs = off;
2097     RGWGetObj::end = off + len;
2098   }
2099
2100   bool only_bucket() override { return false; }
2101
2102   int op_init() override {
2103     // assign store, s, and dialect_handler
2104     RGWObjectCtx* rados_ctx
2105       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2106     // framework promises to call op_init after parent init
2107     ceph_assert(rados_ctx);
2108     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2109     op = this; // assign self as op: REQUIRED
2110     return 0;
2111   }
2112
2113   int header_init() override {
2114
2115     struct req_state* state = get_state();
2116     state->info.method = "GET";
2117     state->op = OP_GET;
2118
2119     /* XXX derp derp derp */
2120     state->relative_uri = make_uri(rgw_fh->bucket_name(),
2121                                rgw_fh->relative_object_name());
2122     state->info.request_uri = state->relative_uri; // XXX
2123     state->info.effective_uri = state->relative_uri;
2124     state->info.request_params = "";
2125     state->info.domain = ""; /* XXX ? */
2126
2127     return 0;
2128   }
2129
2130   int get_params(optional_yield) override {
2131     return 0;
2132   }
2133
2134   int send_response_data(ceph::buffer::list& bl, off_t bl_off,
2135                          off_t bl_len) override {
2136     size_t bytes;
2137     for (auto& bp : bl.buffers()) {
2138       /* if for some reason bl_off indicates the start-of-data is not at
2139        * the current buffer::ptr, skip it and account */
2140       if (bl_off > bp.length()) {
2141         bl_off -= bp.length();
2142         continue;
2143       }
2144       /* read no more than read_resid */
2145       bytes = std::min(read_resid, size_t(bp.length()-bl_off));
2146       memcpy(static_cast<char*>(ulp_buffer)+nread, bp.c_str()+bl_off, bytes);
2147       read_resid -= bytes; /* reduce read_resid by bytes read */
2148       nread += bytes;
2149       bl_off = 0;
2150       /* stop if we have no residual ulp_buffer */
2151       if (! read_resid)
2152         break;
2153     }
2154     return 0;
2155   }
2156
2157   int send_response_data_error(optional_yield) override {
2158     /* S3 implementation just sends nothing--there is no side effect
2159      * to simulate here */
2160     return 0;
2161   }
2162
2163   bool prefetch_data() override { return false; }
2164
2165 }; /* RGWReadRequest */
2166
2167 /*
2168   delete object
2169 */
2170
2171 class RGWDeleteObjRequest : public RGWLibRequest,
2172                             public RGWDeleteObj /* RGWOp */
2173 {
2174 public:
2175   const std::string& bucket_name;
2176   const std::string& obj_name;
2177
2178   RGWDeleteObjRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
2179                       const std::string& _bname, const std::string& _oname)
2180     : RGWLibRequest(_cct, std::move(_user)), bucket_name(_bname), obj_name(_oname) {
2181     op = this;
2182   }
2183
2184   bool only_bucket() override { return true; }
2185
2186   int op_init() override {
2187     // assign store, s, and dialect_handler
2188     RGWObjectCtx* rados_ctx
2189       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2190     // framework promises to call op_init after parent init
2191     ceph_assert(rados_ctx);
2192     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2193     op = this; // assign self as op: REQUIRED
2194     return 0;
2195   }
2196
2197   int header_init() override {
2198
2199     struct req_state* state = get_state();
2200     state->info.method = "DELETE";
2201     state->op = OP_DELETE;
2202
2203     /* XXX derp derp derp */
2204     std::string uri = make_uri(bucket_name, obj_name);
2205     state->relative_uri = uri;
2206     state->info.request_uri = uri; // XXX
2207     state->info.effective_uri = uri;
2208     state->info.request_params = "";
2209     state->info.domain = ""; /* XXX ? */
2210
2211     return 0;
2212   }
2213
2214   void send_response() override {}
2215
2216 }; /* RGWDeleteObjRequest */
2217
2218 class RGWStatObjRequest : public RGWLibRequest,
2219                           public RGWGetObj /* RGWOp */
2220 {
2221 public:
2222   const std::string& bucket_name;
2223   const std::string& obj_name;
2224   uint64_t _size;
2225   uint32_t flags;
2226
2227   static constexpr uint32_t FLAG_NONE = 0x000;
2228
2229   RGWStatObjRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
2230                     const std::string& _bname, const std::string& _oname,
2231                     uint32_t _flags)
2232     : RGWLibRequest(_cct, std::move(_user)), bucket_name(_bname), obj_name(_oname),
2233       _size(0), flags(_flags) {
2234     op = this;
2235
2236     /* fixup RGWGetObj (already know range parameters) */
2237     RGWGetObj::range_parsed = true;
2238     RGWGetObj::get_data = false; // XXX
2239     RGWGetObj::partial_content = true;
2240     RGWGetObj::ofs = 0;
2241     RGWGetObj::end = UINT64_MAX;
2242   }
2243
2244   const char* name() const override { return "stat_obj"; }
2245   RGWOpType get_type() override { return RGW_OP_STAT_OBJ; }
2246
2247   real_time get_mtime() const {
2248     return lastmod;
2249   }
2250
2251   /* attributes */
2252   uint64_t get_size() { return _size; }
2253   real_time ctime() { return mod_time; } // XXX
2254   real_time mtime() { return mod_time; }
2255   std::map<string, bufferlist>& get_attrs() { return attrs; }
2256
2257   buffer::list* get_attr(const std::string& k) {
2258     auto iter = attrs.find(k);
2259     return (iter != attrs.end()) ? &(iter->second) : nullptr;
2260   }
2261
2262   bool only_bucket() override { return false; }
2263
2264   int op_init() override {
2265     // assign store, s, and dialect_handler
2266     RGWObjectCtx* rados_ctx
2267       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2268     // framework promises to call op_init after parent init
2269     ceph_assert(rados_ctx);
2270     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2271     op = this; // assign self as op: REQUIRED
2272     return 0;
2273   }
2274
2275   int header_init() override {
2276
2277     struct req_state* state = get_state();
2278     state->info.method = "GET";
2279     state->op = OP_GET;
2280
2281     /* XXX derp derp derp */
2282     state->relative_uri = make_uri(bucket_name, obj_name);
2283     state->info.request_uri = state->relative_uri; // XXX
2284     state->info.effective_uri = state->relative_uri;
2285     state->info.request_params = "";
2286     state->info.domain = ""; /* XXX ? */
2287
2288     return 0;
2289   }
2290
2291   int get_params(optional_yield) override {
2292     return 0;
2293   }
2294
2295   int send_response_data(ceph::buffer::list& _bl, off_t s_off,
2296                          off_t e_off) override {
2297     /* NOP */
2298     /* XXX save attrs? */
2299     return 0;
2300   }
2301
2302   int send_response_data_error(optional_yield) override {
2303     /* NOP */
2304     return 0;
2305   }
2306
2307   void execute(optional_yield y) override {
2308     RGWGetObj::execute(y);
2309     _size = get_state()->obj_size;
2310   }
2311
2312 }; /* RGWStatObjRequest */
2313
2314 class RGWStatBucketRequest : public RGWLibRequest,
2315                              public RGWStatBucket /* RGWOp */
2316 {
2317 public:
2318   std::string uri;
2319   std::map<std::string, buffer::list> attrs;
2320   RGWLibFS::BucketStats& bs;
2321
2322   RGWStatBucketRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
2323                        const std::string& _path,
2324                        RGWLibFS::BucketStats& _stats)
2325     : RGWLibRequest(_cct, std::move(_user)), bs(_stats) {
2326     uri = "/" + _path;
2327     op = this;
2328   }
2329
2330   buffer::list* get_attr(const std::string& k) {
2331     auto iter = attrs.find(k);
2332     return (iter != attrs.end()) ? &(iter->second) : nullptr;
2333   }
2334
2335   real_time get_ctime() const {
2336     return bucket->get_creation_time();
2337   }
2338
2339   bool only_bucket() override { return false; }
2340
2341   int op_init() override {
2342     // assign store, s, and dialect_handler
2343     RGWObjectCtx* rados_ctx
2344       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2345     // framework promises to call op_init after parent init
2346     ceph_assert(rados_ctx);
2347     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2348     op = this; // assign self as op: REQUIRED
2349     return 0;
2350   }
2351
2352   int header_init() override {
2353
2354     struct req_state* state = get_state();
2355     state->info.method = "GET";
2356     state->op = OP_GET;
2357
2358     /* XXX derp derp derp */
2359     state->relative_uri = uri;
2360     state->info.request_uri = uri; // XXX
2361     state->info.effective_uri = uri;
2362     state->info.request_params = "";
2363     state->info.domain = ""; /* XXX ? */
2364
2365     return 0;
2366   }
2367
2368   virtual int get_params() {
2369     return 0;
2370   }
2371
2372   void send_response() override {
2373     bucket->get_creation_time() = get_state()->bucket->get_info().creation_time;
2374     bs.size = bucket->get_size();
2375     bs.size_rounded = bucket->get_size_rounded();
2376     bs.creation_time = bucket->get_creation_time();
2377     bs.num_entries = bucket->get_count();
2378     std::swap(attrs, get_state()->bucket_attrs);
2379   }
2380
2381   bool matched() {
2382     return (bucket->get_name().length() > 0);
2383   }
2384
2385 }; /* RGWStatBucketRequest */
2386
2387 class RGWStatLeafRequest : public RGWLibRequest,
2388                            public RGWListBucket /* RGWOp */
2389 {
2390 public:
2391   RGWFileHandle* rgw_fh;
2392   std::string path;
2393   bool matched;
2394   bool is_dir;
2395   bool exact_matched;
2396
2397   RGWStatLeafRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
2398                      RGWFileHandle* _rgw_fh, const std::string& _path)
2399     : RGWLibRequest(_cct, std::move(_user)), rgw_fh(_rgw_fh), path(_path),
2400       matched(false), is_dir(false), exact_matched(false) {
2401     default_max = 1000; // logical max {"foo", "foo/"}
2402     op = this;
2403   }
2404
2405   bool only_bucket() override { return true; }
2406
2407   int op_init() override {
2408     // assign store, s, and dialect_handler
2409     RGWObjectCtx* rados_ctx
2410       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2411     // framework promises to call op_init after parent init
2412     ceph_assert(rados_ctx);
2413     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2414     op = this; // assign self as op: REQUIRED
2415     return 0;
2416   }
2417
2418   int header_init() override {
2419
2420     struct req_state* state = get_state();
2421     state->info.method = "GET";
2422     state->op = OP_GET;
2423
2424     /* XXX derp derp derp */
2425     std::string uri = "/" + rgw_fh->bucket_name() + "/";
2426     state->relative_uri = uri;
2427     state->info.request_uri = uri; // XXX
2428     state->info.effective_uri = uri;
2429     state->info.request_params = "";
2430     state->info.domain = ""; /* XXX ? */
2431
2432     prefix = rgw_fh->relative_object_name();
2433     if (prefix.length() > 0)
2434       prefix += "/";
2435     prefix += path;
2436     delimiter = '/';
2437
2438     return 0;
2439   }
2440
2441   int get_params(optional_yield) override {
2442     max = default_max;
2443     return 0;
2444   }
2445
2446   void send_response() override {
2447     struct req_state* state = get_state();
2448     // try objects
2449     for (const auto& iter : objs) {
2450       auto& name = iter.key.name;
2451       lsubdout(cct, rgw, 15) << "RGWStatLeafRequest "
2452                              << __func__ << " "
2453                              << "list uri=" << state->relative_uri << " "
2454                              << " prefix=" << prefix << " "
2455                              << " obj path=" << name << ""
2456                              << " target = " << path << ""
2457                              << dendl;
2458       /* XXX is there a missing match-dir case (trailing '/')? */
2459       matched = true;
2460       if (name == path) {
2461         exact_matched = true;
2462         return;
2463       }
2464     }
2465     // try prefixes
2466     for (auto& iter : common_prefixes) {
2467       auto& name = iter.first;
2468       lsubdout(cct, rgw, 15) << "RGWStatLeafRequest "
2469                              << __func__ << " "
2470                              << "list uri=" << state->relative_uri << " "
2471                              << " prefix=" << prefix << " "
2472                              << " pref path=" << name << " (not chomped)"
2473                              << " target = " << path << ""
2474                              << dendl;
2475       matched = true;
2476       /* match-dir case (trailing '/') */
2477       if (name == prefix + "/") {
2478         exact_matched = true;
2479         is_dir = true;
2480         return;
2481       }
2482     }
2483   }
2484
2485   virtual void send_versioned_response() {
2486     send_response();
2487   }
2488 }; /* RGWStatLeafRequest */
2489
2490 /*
2491   put object
2492 */
2493
2494 class RGWWriteRequest : public RGWLibContinuedReq,
2495                         public RGWPutObj /* RGWOp */
2496 {
2497 public:
2498   const std::string& bucket_name;
2499   const std::string& obj_name;
2500   RGWFileHandle* rgw_fh;
2501   std::optional<rgw::BlockingAioThrottle> aio;
2502   std::unique_ptr<rgw::sal::Writer> processor;
2503   rgw::sal::DataProcessor* filter;
2504   boost::optional<RGWPutObj_Compress> compressor;
2505   CompressorRef plugin;
2506   buffer::list data;
2507   uint64_t timer_id;
2508   MD5 hash;
2509   off_t real_ofs;
2510   size_t bytes_written;
2511   bool eio;
2512
2513   RGWWriteRequest(rgw::sal::Store* store,
2514                   std::unique_ptr<rgw::sal::User> _user,
2515                   RGWFileHandle* _fh, const std::string& _bname,
2516                   const std::string& _oname)
2517     : RGWLibContinuedReq(store->ctx(), std::move(_user)),
2518       bucket_name(_bname), obj_name(_oname),
2519       rgw_fh(_fh), filter(nullptr), timer_id(0), real_ofs(0),
2520       bytes_written(0), eio(false) {
2521
2522     // in ctr this is not a virtual call
2523     // invoking this classes's header_init()
2524     (void) RGWWriteRequest::header_init();
2525     op = this;
2526     // Allow use of MD5 digest in FIPS mode for non-cryptographic purposes
2527     hash.SetFlags(EVP_MD_CTX_FLAG_NON_FIPS_ALLOW);
2528   }
2529
2530   bool only_bucket() override { return true; }
2531
2532   int op_init() override {
2533     // assign store, s, and dialect_handler
2534     RGWObjectCtx* rados_ctx
2535       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2536     // framework promises to call op_init after parent init
2537     ceph_assert(rados_ctx);
2538     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2539     op = this; // assign self as op: REQUIRED
2540     return 0;
2541   }
2542
2543   int header_init() override {
2544
2545     struct req_state* state = get_state();
2546     state->info.method = "PUT";
2547     state->op = OP_PUT;
2548
2549     /* XXX derp derp derp */
2550     std::string uri = make_uri(bucket_name, obj_name);
2551     state->relative_uri = uri;
2552     state->info.request_uri = uri; // XXX
2553     state->info.effective_uri = uri;
2554     state->info.request_params = "";
2555     state->info.domain = ""; /* XXX ? */
2556
2557     return 0;
2558   }
2559
2560   int get_params(optional_yield) override {
2561     struct req_state* state = get_state();
2562     RGWAccessControlPolicy_S3 s3policy(state->cct);
2563     /* we don't have (any) headers, so just create canned ACLs */
2564     int ret = s3policy.create_canned(state->owner, state->bucket_owner, state->canned_acl);
2565     policy = s3policy;
2566     return ret;
2567   }
2568
2569   int get_data(buffer::list& _bl) override {
2570     /* XXX for now, use sharing semantics */
2571     uint32_t len = data.length();
2572     _bl = std::move(data);
2573     bytes_written += len;
2574     return len;
2575   }
2576
2577   void put_data(off_t off, buffer::list& _bl) {
2578     if (off != real_ofs) {
2579       eio = true;
2580     }
2581     data = std::move(_bl);
2582     real_ofs += data.length();
2583     ofs = off; /* consumed in exec_continue() */
2584   }
2585
2586   int exec_start() override;
2587   int exec_continue() override;
2588   int exec_finish() override;
2589
2590   void send_response() override {}
2591
2592   int verify_params() override {
2593     return 0;
2594   }
2595 }; /* RGWWriteRequest */
2596
2597 /*
2598   copy object
2599 */
2600 class RGWCopyObjRequest : public RGWLibRequest,
2601                           public RGWCopyObj /* RGWOp */
2602 {
2603 public:
2604   RGWFileHandle* src_parent;
2605   RGWFileHandle* dst_parent;
2606   const std::string& src_name;
2607   const std::string& dst_name;
2608
2609   RGWCopyObjRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
2610                     RGWFileHandle* _src_parent, RGWFileHandle* _dst_parent,
2611                     const std::string& _src_name, const std::string& _dst_name)
2612     : RGWLibRequest(_cct, std::move(_user)), src_parent(_src_parent),
2613       dst_parent(_dst_parent), src_name(_src_name), dst_name(_dst_name) {
2614     /* all requests have this */
2615     op = this;
2616
2617     /* allow this request to replace selected attrs */
2618     attrs_mod = rgw::sal::ATTRSMOD_MERGE;
2619   }
2620
2621   bool only_bucket() override { return true; }
2622
2623   int op_init() override {
2624     // assign store, s, and dialect_handler
2625     RGWObjectCtx* rados_ctx
2626       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2627     // framework promises to call op_init after parent init
2628     ceph_assert(rados_ctx);
2629     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2630     op = this; // assign self as op: REQUIRED
2631
2632     return 0;
2633   }
2634
2635   int header_init() override {
2636
2637     struct req_state* state = get_state();
2638     state->info.method = "PUT"; // XXX check
2639     state->op = OP_PUT;
2640
2641     src_bucket_name = src_parent->bucket_name();
2642     state->src_bucket_name = src_bucket_name;
2643     dest_bucket_name = dst_parent->bucket_name();
2644     state->bucket_name = dest_bucket_name;
2645     dest_obj_name = dst_parent->format_child_name(dst_name, false);
2646
2647     int rc = valid_s3_object_name(dest_obj_name);
2648     if (rc != 0)
2649       return rc;
2650
2651     /* XXX and fixup key attr (could optimize w/string ref and
2652      * dest_obj_name) */
2653     buffer::list ux_key;
2654     fh_key fhk = dst_parent->make_fhk(dst_name);
2655     rgw::encode(fhk, ux_key);
2656     emplace_attr(RGW_ATTR_UNIX_KEY1, std::move(ux_key));
2657
2658 #if 0 /* XXX needed? */
2659     state->relative_uri = uri;
2660     state->info.request_uri = uri; // XXX
2661     state->info.effective_uri = uri;
2662     state->info.request_params = "";
2663     state->info.domain = ""; /* XXX ? */
2664 #endif
2665
2666     return 0;
2667   }
2668
2669   int get_params(optional_yield) override {
2670     struct req_state* s = get_state();
2671     RGWAccessControlPolicy_S3 s3policy(s->cct);
2672     /* we don't have (any) headers, so just create canned ACLs */
2673     int ret = s3policy.create_canned(s->owner, s->bucket_owner, s->canned_acl);
2674     dest_policy = s3policy;
2675     /* src_object required before RGWCopyObj::verify_permissions() */
2676     rgw_obj_key k = rgw_obj_key(src_name);
2677     s->src_object = s->bucket->get_object(k);
2678     s->object = s->src_object->clone(); // needed to avoid trap at rgw_op.cc:5150
2679     return ret;
2680   }
2681
2682   void send_response() override {}
2683   void send_partial_response(off_t ofs) override {}
2684
2685 }; /* RGWCopyObjRequest */
2686
2687 class RGWGetAttrsRequest : public RGWLibRequest,
2688                            public RGWGetAttrs /* RGWOp */
2689 {
2690 public:
2691   const std::string& bucket_name;
2692   const std::string& obj_name;
2693
2694   RGWGetAttrsRequest(CephContext* _cct,
2695                      std::unique_ptr<rgw::sal::User> _user,
2696                      const std::string& _bname, const std::string& _oname)
2697     : RGWLibRequest(_cct, std::move(_user)), RGWGetAttrs(),
2698       bucket_name(_bname), obj_name(_oname) {
2699     op = this;
2700   }
2701
2702   const flat_map<std::string, std::optional<buffer::list>>& get_attrs() {
2703     return attrs;
2704   }
2705
2706   virtual bool only_bucket() { return false; }
2707
2708   virtual int op_init() {
2709     // assign store, s, and dialect_handler
2710     RGWObjectCtx* rados_ctx
2711       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2712     // framework promises to call op_init after parent init
2713     assert(rados_ctx);
2714     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2715     op = this; // assign self as op: REQUIRED
2716     return 0;
2717   }
2718
2719   virtual int header_init() {
2720
2721     struct req_state* s = get_state();
2722     s->info.method = "GET";
2723     s->op = OP_GET;
2724
2725     std::string uri = make_uri(bucket_name, obj_name);
2726     s->relative_uri = uri;
2727     s->info.request_uri = uri;
2728     s->info.effective_uri = uri;
2729     s->info.request_params = "";
2730     s->info.domain = ""; /* XXX ? */
2731
2732     return 0;
2733   }
2734
2735   virtual int get_params() {
2736     return 0;
2737   }
2738
2739   virtual void send_response() {}
2740
2741 }; /* RGWGetAttrsRequest */
2742
2743 class RGWSetAttrsRequest : public RGWLibRequest,
2744                            public RGWSetAttrs /* RGWOp */
2745 {
2746 public:
2747   const std::string& bucket_name;
2748   const std::string& obj_name;
2749
2750   RGWSetAttrsRequest(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
2751                      const std::string& _bname, const std::string& _oname)
2752     : RGWLibRequest(_cct, std::move(_user)), bucket_name(_bname), obj_name(_oname) {
2753     op = this;
2754   }
2755
2756   const std::map<std::string, buffer::list>& get_attrs() {
2757     return attrs;
2758   }
2759
2760   bool only_bucket() override { return false; }
2761
2762   int op_init() override {
2763     // assign store, s, and dialect_handler
2764     RGWObjectCtx* rados_ctx
2765       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2766     // framework promises to call op_init after parent init
2767     ceph_assert(rados_ctx);
2768     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2769     op = this; // assign self as op: REQUIRED
2770     return 0;
2771   }
2772
2773   int header_init() override {
2774
2775     struct req_state* state = get_state();
2776     state->info.method = "PUT";
2777     state->op = OP_PUT;
2778
2779     /* XXX derp derp derp */
2780     std::string uri = make_uri(bucket_name, obj_name);
2781     state->relative_uri = uri;
2782     state->info.request_uri = uri; // XXX
2783     state->info.effective_uri = uri;
2784     state->info.request_params = "";
2785     state->info.domain = ""; /* XXX ? */
2786
2787     return 0;
2788   }
2789
2790   int get_params(optional_yield) override {
2791     return 0;
2792   }
2793
2794   void send_response() override {}
2795
2796 }; /* RGWSetAttrsRequest */
2797
2798 class RGWRMAttrsRequest : public RGWLibRequest,
2799                           public RGWRMAttrs /* RGWOp */
2800 {
2801 public:
2802   const std::string& bucket_name;
2803   const std::string& obj_name;
2804
2805   RGWRMAttrsRequest(CephContext* _cct,
2806                      std::unique_ptr<rgw::sal::User> _user,
2807                      const std::string& _bname, const std::string& _oname)
2808     : RGWLibRequest(_cct, std::move(_user)), RGWRMAttrs(),
2809       bucket_name(_bname), obj_name(_oname) {
2810     op = this;
2811   }
2812
2813   const rgw::sal::Attrs& get_attrs() {
2814     return attrs;
2815   }
2816
2817   virtual bool only_bucket() { return false; }
2818
2819   virtual int op_init() {
2820     // assign store, s, and dialect_handler
2821     RGWObjectCtx* rados_ctx
2822       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2823     // framework promises to call op_init after parent init
2824     assert(rados_ctx);
2825     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2826     op = this; // assign self as op: REQUIRED
2827     return 0;
2828   }
2829
2830   virtual int header_init() {
2831
2832     struct req_state* s = get_state();
2833     s->info.method = "DELETE";
2834     s->op = OP_PUT;
2835
2836     std::string uri = make_uri(bucket_name, obj_name);
2837     s->relative_uri = uri;
2838     s->info.request_uri = uri;
2839     s->info.effective_uri = uri;
2840     s->info.request_params = "";
2841     s->info.domain = ""; /* XXX ? */
2842
2843     return 0;
2844   }
2845
2846   virtual int get_params() {
2847     return 0;
2848   }
2849
2850   virtual void send_response() {}
2851
2852 }; /* RGWRMAttrsRequest */
2853
2854 /*
2855  * Send request to get the rados cluster stats
2856  */
2857 class RGWGetClusterStatReq : public RGWLibRequest,
2858         public RGWGetClusterStat {
2859 public:
2860   struct rados_cluster_stat_t& stats_req;
2861   RGWGetClusterStatReq(CephContext* _cct, std::unique_ptr<rgw::sal::User> _user,
2862                        rados_cluster_stat_t& _stats):
2863   RGWLibRequest(_cct, std::move(_user)), stats_req(_stats){
2864     op = this;
2865   }
2866
2867   int op_init() override {
2868     // assign store, s, and dialect_handler
2869     RGWObjectCtx* rados_ctx
2870       = static_cast<RGWObjectCtx*>(get_state()->obj_ctx);
2871     // framework promises to call op_init after parent init
2872     ceph_assert(rados_ctx);
2873     RGWOp::init(rados_ctx->get_store(), get_state(), this);
2874     op = this; // assign self as op: REQUIRED
2875     return 0;
2876   }
2877
2878   int header_init() override {
2879     struct req_state* state = get_state();
2880     state->info.method = "GET";
2881     state->op = OP_GET;
2882     return 0;
2883   }
2884
2885   int get_params(optional_yield) override { return 0; }
2886   bool only_bucket() override { return false; }
2887   void send_response() override {
2888     stats_req.kb = stats_op.kb;
2889     stats_req.kb_avail = stats_op.kb_avail;
2890     stats_req.kb_used = stats_op.kb_used;
2891     stats_req.num_objects = stats_op.num_objects;
2892   }
2893 }; /* RGWGetClusterStatReq */
2894
2895
2896 } /* namespace rgw */
2897
2898 #endif /* RGW_FILE_H */