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