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