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