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