ceph/src/os/FuseStore.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "include/compat.h"
   5 #include "FuseStore.h"
   6 #include "os/ObjectStore.h"
   7 #include "include/stringify.h"
   8 #include "common/errno.h"
   9
  10 #define FUSE_USE_VERSION 30
  11 #include <fuse.h>
  12 #include <fuse_lowlevel.h>
  13 #include "include/ceph_fuse.h"
  14
  15 #include <sys/types.h>
  16 #include <sys/stat.h>
  17 #include <unistd.h>
  18 #include <fcntl.h>           /* Definition of AT_* constants */
  19 #include <sys/stat.h>
  20
  21 #if defined(__APPLE__) || defined(__FreeBSD__)
  22 #include <sys/param.h>
  23 #include <sys/mount.h>
  24 #endif
  25
  26 #define dout_context store->cct
  27 #define dout_subsys ceph_subsys_fuse
  28 #include "common/debug.h"
  29 #undef dout_prefix
  30 #define dout_prefix *_dout << "fuse "
  31
  32 // some fuse-y bits of state
  33 struct fs_info {
  34   struct fuse_args args;
  35   struct fuse *f;
  36 #if FUSE_VERSION < FUSE_MAKE_VERSION(3, 0)
  37   struct fuse_chan *ch;
  38 #endif
  39   char *mountpoint;
  40 };
  41
  42 int FuseStore::open_file(string p, struct fuse_file_info *fi,
  43                          std::function<int(bufferlist *bl)> f)
  44 {
  45   if (open_files.count(p)) {
  46     OpenFile *o = open_files[p];
  47     fi->fh = reinterpret_cast<uint64_t>(o);
  48     ++o->ref;
  49     return 0;
  50   }
  51   bufferlist bl;
  52   int r = f(&bl);
  53   if (r < 0) {
  54     return r;
  55   }
  56   OpenFile *o = new OpenFile;
  57   o->path = p;
  58   o->bl.claim(bl);
  59   open_files[p] = o;
  60   fi->fh = reinterpret_cast<uint64_t>(o);
  61   ++o->ref;
  62   return 0;
  63 }
  64
  65 FuseStore::FuseStore(ObjectStore *s, string p)
  66   : store(s),
  67     mount_point(p),
  68     fuse_thread(this)
  69 {
  70   info = new fs_info();
  71 }
  72
  73 FuseStore::~FuseStore()
  74 {
  75   delete info;
  76 }
  77
  78 /*
  79  * / - root directory
  80  * $cid/
  81  * $cid/type - objectstore type
  82  * $cid/bitwise_hash_start = lowest hash value
  83  * $cid/bitwise_hash_end = highest hash value
  84  * $cid/bitwise_hash_bits - how many bits are significant
  85  * $cid/pgmeta/ - pgmeta object
  86  * $cid/all/ - all objects
  87  * $cid/all/$obj/
  88  * $cid/all/$obj/bitwise_hash
  89  * $cid/all/$obj/data
  90  * $cid/all/$obj/omap/$key
  91  * $cid/all/$obj/attr/$name
  92  * $cid/by_bitwise_hash/$hash/$bits/$obj - all objects with this (bitwise) hash (prefix)
  93  */
  94 enum {
  95   FN_ROOT = 1,
  96   FN_TYPE,
  97   FN_COLLECTION,
  98   FN_HASH_START,
  99   FN_HASH_END,
 100   FN_HASH_BITS,
 101   FN_OBJECT,
 102   FN_OBJECT_HASH,
 103   FN_OBJECT_DATA,
 104   FN_OBJECT_OMAP_HEADER,
 105   FN_OBJECT_OMAP,
 106   FN_OBJECT_OMAP_VAL,
 107   FN_OBJECT_ATTR,
 108   FN_OBJECT_ATTR_VAL,
 109   FN_ALL,
 110   FN_HASH_DIR,
 111   FN_HASH_VAL,
 112 };
 113
 114 static int parse_fn(CephContext* cct, const char *path, coll_t *cid,
 115                     ghobject_t *oid, string *key,
 116                     uint32_t *hash, uint32_t *hash_bits)
 117 {
 118   list<string> v;
 119   for (const char *p = path; *p; ++p) {
 120     if (*p == '/')
 121       continue;
 122     const char *e;
 123     for (e = p + 1; *e && *e != '/'; e++) ;
 124     string c(p, e-p);
 125     v.push_back(c);
 126     p = e;
 127     if (!*p)
 128       break;
 129   }
 130   ldout(cct, 10) << __func__ << " path " << path << " -> " << v << dendl;
 131
 132   if (v.empty())
 133     return FN_ROOT;
 134
 135   if (v.front() == "type")
 136     return FN_TYPE;
 137
 138   if (!cid->parse(v.front())) {
 139     return -ENOENT;
 140   }
 141   if (v.size() == 1)
 142     return FN_COLLECTION;
 143   v.pop_front();
 144
 145   if (v.front() == "bitwise_hash_start")
 146     return FN_HASH_START;
 147   if (v.front() == "bitwise_hash_end")
 148     return FN_HASH_END;
 149   if (v.front() == "bitwise_hash_bits")
 150     return FN_HASH_BITS;
 151   if (v.front() == "pgmeta") {
 152     spg_t pgid;
 153     if (cid->is_pg(&pgid)) {
 154       *oid = pgid.make_pgmeta_oid();
 155       v.pop_front();
 156       if (v.empty())
 157         return FN_OBJECT;
 158       goto do_object;
 159     }
 160     return -ENOENT;
 161   }
 162   if (v.front() == "all") {
 163     v.pop_front();
 164     if (v.empty())
 165       return FN_ALL;
 166     goto do_dir;
 167   }
 168   if (v.front() == "by_bitwise_hash") {
 169     v.pop_front();
 170     if (v.empty())
 171       return FN_HASH_DIR;
 172     unsigned long hv, hm;
 173     int r = sscanf(v.front().c_str(), "%lx", &hv);
 174     if (r != 1)
 175       return -ENOENT;
 176     int shift = 32 - v.front().length() * 4;
 177     v.pop_front();
 178     if (v.empty())
 179       return FN_HASH_DIR;
 180     r = sscanf(v.front().c_str(), "%ld", &hm);
 181     if (r != 1)
 182       return -ENOENT;
 183     if (hm < 1 || hm > 32)
 184       return -ENOENT;
 185     v.pop_front();
 186     *hash = hv << shift;//hobject_t::_reverse_bits(hv << shift);
 187     *hash_bits = hm;
 188     if (v.empty())
 189       return FN_HASH_VAL;
 190     goto do_dir;
 191   }
 192   return -ENOENT;
 193
 194  do_dir:
 195   {
 196     string o = v.front();
 197     if (!oid->parse(o)) {
 198       return -ENOENT;
 199     }
 200     v.pop_front();
 201     if (v.empty())
 202       return FN_OBJECT;
 203   }
 204
 205  do_object:
 206   if (v.front() == "data")
 207     return FN_OBJECT_DATA;
 208   if (v.front() == "omap_header")
 209     return FN_OBJECT_OMAP_HEADER;
 210   if (v.front() == "omap") {
 211     v.pop_front();
 212     if (v.empty())
 213       return FN_OBJECT_OMAP;
 214     *key = v.front();
 215     v.pop_front();
 216     if (v.empty())
 217       return FN_OBJECT_OMAP_VAL;
 218     return -ENOENT;
 219   }
 220   if (v.front() == "attr") {
 221     v.pop_front();
 222     if (v.empty())
 223       return FN_OBJECT_ATTR;
 224     *key = v.front();
 225     v.pop_front();
 226     if (v.empty())
 227       return FN_OBJECT_ATTR_VAL;
 228     return -ENOENT;
 229   }
 230   if (v.front() == "bitwise_hash")
 231     return FN_OBJECT_HASH;
 232   return -ENOENT;
 233 }
 234
 235
 236 static int os_getattr(const char *path, struct stat *stbuf
 237 #if FUSE_VERSION >= FUSE_MAKE_VERSION(3, 0)
 238                       , struct fuse_file_info *fi
 239 #endif
 240                       )
 241 {
 242   fuse_context *fc = fuse_get_context();
 243   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
 244   ldout(fs->store->cct, 10) << __func__ << " " << path << dendl;
 245   coll_t cid;
 246   ghobject_t oid;
 247   string key;
 248   uint32_t hash_value, hash_bits;
 249   int t = parse_fn(fs->store->cct, path, &cid, &oid, &key, &hash_value,
 250                    &hash_bits);
 251   if (t < 0)
 252     return t;
 253
 254   std::lock_guard<std::mutex> l(fs->lock);
 255
 256   stbuf->st_size = 0;
 257   stbuf->st_uid = 0;
 258   stbuf->st_gid = 0;
 259   stbuf->st_mode = S_IFREG | 0700;
 260
 261   auto ch = fs->store->open_collection(cid);
 262
 263   switch (t) {
 264   case FN_OBJECT_OMAP:
 265   case FN_OBJECT_ATTR:
 266   case FN_OBJECT:
 267   case FN_OBJECT_DATA:
 268   case FN_OBJECT_OMAP_HEADER:
 269   case FN_OBJECT_OMAP_VAL:
 270     {
 271       spg_t pgid;
 272       if (cid.is_pg(&pgid)) {
 273         if (!ch) {
 274           return -ENOENT;
 275         }
 276         int bits = fs->store->collection_bits(ch);
 277         if (bits >= 0 && !oid.match(bits, pgid.ps())) {
 278           // sorry, not part of this PG
 279           return -ENOENT;
 280         }
 281       }
 282     }
 283     break;
 284   }
 285
 286   switch (t) {
 287   case FN_OBJECT_OMAP:
 288   case FN_OBJECT_ATTR:
 289   case FN_OBJECT:
 290     if (!fs->store->exists(ch, oid))
 291       return -ENOENT;
 292     // fall-thru
 293   case FN_ALL:
 294   case FN_HASH_DIR:
 295   case FN_HASH_VAL:
 296   case FN_COLLECTION:
 297     if (!fs->store->collection_exists(cid))
 298       return -ENOENT;
 299     // fall-thru
 300   case FN_ROOT:
 301     stbuf->st_mode = S_IFDIR | 0700;
 302     return 0;
 303
 304   case FN_TYPE:
 305     stbuf->st_size = fs->store->get_type().length() + 1;
 306     break;
 307
 308   case FN_OBJECT_HASH:
 309     if (!fs->store->exists(ch, oid))
 310       return -ENOENT;
 311     stbuf->st_size = 9;
 312     return 0;
 313
 314   case FN_HASH_END:
 315     if (!ch)
 316       return -ENOENT;
 317     if (fs->store->collection_bits(ch) < 0)
 318       return -ENOENT;
 319     // fall-thru
 320   case FN_HASH_START:
 321     stbuf->st_size = 9;
 322     return 0;
 323
 324   case FN_HASH_BITS:
 325     {
 326       if (!ch)
 327         return -ENOENT;
 328       int bits = fs->store->collection_bits(ch);
 329       if (bits < 0)
 330         return -ENOENT;
 331       char buf[12];
 332       snprintf(buf, sizeof(buf), "%d\n", bits);
 333       stbuf->st_size = strlen(buf);
 334     }
 335     return 0;
 336
 337   case FN_OBJECT_DATA:
 338     {
 339       if (!fs->store->exists(ch, oid))
 340         return -ENOENT;
 341       int r = fs->store->stat(ch, oid, stbuf);
 342       if (r < 0)
 343         return r;
 344     }
 345     break;
 346
 347   case FN_OBJECT_OMAP_HEADER:
 348     {
 349       if (!fs->store->exists(ch, oid))
 350         return -ENOENT;
 351       bufferlist bl;
 352       fs->store->omap_get_header(ch, oid, &bl);
 353       stbuf->st_size = bl.length();
 354     }
 355     break;
 356
 357   case FN_OBJECT_OMAP_VAL:
 358     {
 359       if (!fs->store->exists(ch, oid))
 360         return -ENOENT;
 361       set<string> k;
 362       k.insert(key);
 363       map<string,bufferlist> v;
 364       fs->store->omap_get_values(ch, oid, k, &v);
 365       if (!v.count(key)) {
 366         return -ENOENT;
 367       }
 368       stbuf->st_size = v[key].length();
 369     }
 370     break;
 371
 372   case FN_OBJECT_ATTR_VAL:
 373     {
 374       if (!fs->store->exists(ch, oid))
 375         return -ENOENT;
 376       bufferptr v;
 377       int r = fs->store->getattr(ch, oid, key.c_str(), v);
 378       if (r == -ENODATA)
 379         r = -ENOENT;
 380       if (r < 0)
 381         return r;
 382       stbuf->st_size = v.length();
 383     }
 384     break;
 385
 386   default:
 387     return -ENOENT;
 388   }
 389
 390   return 0;
 391 }
 392
 393 static int os_readdir(const char *path,
 394                       void *buf,
 395                       fuse_fill_dir_t filler,
 396                       off_t offset,
 397                       struct fuse_file_info *fi
 398 #if FUSE_VERSION >= FUSE_MAKE_VERSION(3, 0)
 399                       , enum fuse_readdir_flags
 400 #endif
 401                       )
 402 {
 403   fuse_context *fc = fuse_get_context();
 404   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
 405   ldout(fs->store->cct, 10) << __func__ << " " << path << " offset " << offset
 406                      << dendl;
 407   coll_t cid;
 408   ghobject_t oid;
 409   string key;
 410   uint32_t hash_value, hash_bits;
 411   int t = parse_fn(fs->store->cct, path, &cid, &oid, &key, &hash_value,
 412                    &hash_bits);
 413   if (t < 0)
 414     return t;
 415
 416   std::lock_guard<std::mutex> l(fs->lock);
 417
 418   auto ch = fs->store->open_collection(cid);
 419
 420   // we can't shift 32 bits or else off_t will go negative
 421   const int hash_shift = 31;
 422
 423   switch (t) {
 424   case FN_ROOT:
 425     {
 426       filler_compat(filler, buf, "type", NULL, 0);
 427       vector<coll_t> cls;
 428       fs->store->list_collections(cls);
 429       for (auto c : cls) {
 430         int r = filler_compat(filler, buf, stringify(c).c_str(), NULL, 0);
 431         if (r > 0)
 432           break;
 433       }
 434     }
 435     break;
 436
 437   case FN_COLLECTION:
 438     {
 439       if (!ch) {
 440         return -ENOENT;
 441       }
 442       filler_compat(filler, buf, "bitwise_hash_start", NULL, 0);
 443       if (fs->store->collection_bits(ch) >= 0) {
 444         filler_compat(filler, buf, "bitwise_hash_end", NULL, 0);
 445         filler_compat(filler, buf, "bitwise_hash_bits", NULL, 0);
 446       }
 447       filler_compat(filler, buf, "all", NULL, 0);
 448       filler_compat(filler, buf, "by_bitwise_hash", NULL, 0);
 449       spg_t pgid;
 450       if (cid.is_pg(&pgid) &&
 451           fs->store->exists(ch, pgid.make_pgmeta_oid())) {
 452         filler_compat(filler, buf, "pgmeta", NULL, 0);
 453       }
 454     }
 455     break;
 456
 457   case FN_OBJECT:
 458     {
 459       filler_compat(filler, buf, "bitwise_hash", NULL, 0);
 460       filler_compat(filler, buf, "data", NULL, 0);
 461       filler_compat(filler, buf, "omap", NULL, 0);
 462       filler_compat(filler, buf, "attr", NULL, 0);
 463       filler_compat(filler, buf, "omap_header", NULL, 0);
 464     }
 465     break;
 466
 467   case FN_HASH_VAL:
 468   case FN_ALL:
 469     {
 470       uint32_t bitwise_hash = (offset >> hash_shift) & 0xffffffff;
 471       uint32_t hashoff = offset - (bitwise_hash << hash_shift);
 472       int skip = hashoff;
 473       ghobject_t next = cid.get_min_hobj();
 474       if (offset) {
 475         // obey the offset
 476         next.hobj.set_hash(hobject_t::_reverse_bits(bitwise_hash));
 477       } else if (t == FN_HASH_VAL) {
 478         next.hobj.set_hash(hobject_t::_reverse_bits(hash_value));
 479       }
 480       ghobject_t last;
 481       if (t == FN_HASH_VAL) {
 482         last = next;
 483         uint64_t rev_end = (hash_value | (0xffffffff >> hash_bits)) + 1;
 484         if (rev_end >= 0x100000000)
 485           last = ghobject_t::get_max();
 486         else
 487           last.hobj.set_hash(hobject_t::_reverse_bits(rev_end));
 488       } else {
 489         last = ghobject_t::get_max();
 490       }
 491       ldout(fs->store->cct, 10) << __func__ << std::hex
 492                          << " offset " << offset << " hash "
 493                          << hobject_t::_reverse_bits(hash_value)
 494                          << std::dec
 495                          << "/" << hash_bits
 496                          << " first " << next << " last " << last
 497                          << dendl;
 498       while (true) {
 499         vector<ghobject_t> ls;
 500         int r = fs->store->collection_list(
 501           ch, next, last, 1000, &ls, &next);
 502         if (r < 0)
 503           return r;
 504         for (auto p : ls) {
 505           if (skip) {
 506             --skip;
 507             continue;
 508           }
 509           uint32_t cur_bitwise_hash = p.hobj.get_bitwise_key_u32();
 510           if (cur_bitwise_hash != bitwise_hash) {
 511             bitwise_hash = cur_bitwise_hash;
 512             hashoff = 0;
 513           }
 514           ++hashoff;
 515           uint64_t cur_off = ((uint64_t)bitwise_hash << hash_shift) |
 516             (uint64_t)hashoff;
 517           string s = stringify(p);
 518           r = filler_compat(filler, buf, s.c_str(), NULL, cur_off);
 519           if (r)
 520             break;
 521         }
 522         if (r)
 523           break;
 524         if (next == ghobject_t::get_max() || next == last)
 525           break;
 526       }
 527     }
 528     break;
 529
 530   case FN_OBJECT_OMAP:
 531     {
 532       set<string> keys;
 533       fs->store->omap_get_keys(ch, oid, &keys);
 534       unsigned skip = offset;
 535       for (auto k : keys) {
 536         if (skip) {
 537           --skip;
 538           continue;
 539         }
 540         ++offset;
 541         int r = filler_compat(filler, buf, k.c_str(), NULL, offset);
 542         if (r)
 543           break;
 544       }
 545     }
 546     break;
 547
 548   case FN_OBJECT_ATTR:
 549     {
 550       map<string,bufferptr> aset;
 551       fs->store->getattrs(ch, oid, aset);
 552       unsigned skip = offset;
 553       for (auto a : aset) {
 554         if (skip) {
 555           --skip;
 556           continue;
 557         }
 558         ++offset;
 559         int r = filler_compat(filler, buf, a.first.c_str(), NULL, offset);
 560         if (r)
 561           break;
 562       }
 563     }
 564     break;
 565   }
 566   return 0;
 567 }
 568
 569 static int os_open(const char *path, struct fuse_file_info *fi)
 570 {
 571   fuse_context *fc = fuse_get_context();
 572   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
 573   ldout(fs->store->cct, 10) << __func__ << " " << path << dendl;
 574   coll_t cid;
 575   ghobject_t oid;
 576   string key;
 577   uint32_t hash_value, hash_bits;
 578   int t = parse_fn(fs->store->cct, path, &cid, &oid, &key, &hash_value,
 579                    &hash_bits);
 580   if (t < 0)
 581     return t;
 582
 583   std::lock_guard<std::mutex> l(fs->lock);
 584
 585   auto ch = fs->store->open_collection(cid);
 586
 587   bufferlist *pbl = 0;
 588   switch (t) {
 589   case FN_TYPE:
 590     pbl = new bufferlist;
 591     pbl->append(fs->store->get_type());
 592     pbl->append("\n");
 593     break;
 594
 595   case FN_HASH_START:
 596     {
 597       pbl = new bufferlist;
 598       spg_t pgid;
 599       if (cid.is_pg(&pgid)) {
 600         unsigned long h;
 601         h = hobject_t::_reverse_bits(pgid.ps());
 602         char buf[10];
 603         snprintf(buf, sizeof(buf), "%08lx\n", h);
 604         pbl->append(buf);
 605       } else {
 606         pbl->append("00000000\n");
 607       }
 608     }
 609     break;
 610
 611   case FN_HASH_END:
 612     {
 613       if (!ch) {
 614         return -ENOENT;
 615       }
 616       spg_t pgid;
 617       unsigned long h;
 618       if (cid.is_pg(&pgid)) {
 619         int hash_bits = fs->store->collection_bits(ch);
 620         if (hash_bits >= 0) {
 621           uint64_t rev_start = hobject_t::_reverse_bits(pgid.ps());
 622           uint64_t rev_end = (rev_start | (0xffffffff >> hash_bits));
 623           h = rev_end;
 624         } else {
 625           return -ENOENT;
 626         }
 627       } else {
 628         h = 0xffffffff;
 629       }
 630       char buf[10];
 631       snprintf(buf, sizeof(buf), "%08lx\n", h);
 632       pbl = new bufferlist;
 633       pbl->append(buf);
 634     }
 635     break;
 636
 637   case FN_HASH_BITS:
 638     {
 639       if (!ch) {
 640         return -ENOENT;
 641       }
 642       int r = fs->store->collection_bits(ch);
 643       if (r < 0)
 644         return r;
 645       char buf[12];
 646       snprintf(buf, sizeof(buf), "%d\n", r);
 647       pbl = new bufferlist;
 648       pbl->append(buf);
 649     }
 650     break;
 651
 652   case FN_OBJECT_HASH:
 653     {
 654       pbl = new bufferlist;
 655       char buf[10];
 656       snprintf(buf, sizeof(buf), "%08x\n",
 657                (unsigned)oid.hobj.get_bitwise_key_u32());
 658       pbl->append(buf);
 659     }
 660     break;
 661
 662   case FN_OBJECT_DATA:
 663     {
 664       int r = fs->open_file(
 665         path, fi,
 666         [&](bufferlist *pbl) {
 667           return fs->store->read(ch, oid, 0, 0, *pbl);
 668         });
 669       if (r < 0) {
 670         return r;
 671       }
 672     }
 673     break;
 674
 675   case FN_OBJECT_ATTR_VAL:
 676     {
 677       int r = fs->open_file(
 678         path, fi,
 679         [&](bufferlist *pbl) {
 680           bufferptr bp;
 681           int r = fs->store->getattr(ch, oid, key.c_str(), bp);
 682           if (r < 0)
 683             return r;
 684           pbl->append(bp);
 685           return 0;
 686         });
 687       if (r < 0)
 688         return r;
 689     }
 690     break;
 691
 692   case FN_OBJECT_OMAP_VAL:
 693     {
 694       int r = fs->open_file(
 695         path, fi,
 696         [&](bufferlist *pbl) {
 697           set<string> k;
 698           k.insert(key);
 699           map<string,bufferlist> v;
 700           int r = fs->store->omap_get_values(ch, oid, k, &v);
 701           if (r < 0)
 702             return r;
 703           *pbl = v[key];
 704           return 0;
 705         });
 706       if (r < 0)
 707         return r;
 708     }
 709     break;
 710
 711   case FN_OBJECT_OMAP_HEADER:
 712     {
 713       int r = fs->open_file(
 714         path, fi,
 715         [&](bufferlist *pbl) {
 716           return fs->store->omap_get_header(ch, oid, pbl);
 717         });
 718       if (r < 0)
 719        return r;
 720     }
 721     break;
 722   }
 723
 724   if (pbl) {
 725     FuseStore::OpenFile *o = new FuseStore::OpenFile;
 726     o->bl.claim(*pbl);
 727     fi->fh = reinterpret_cast<uint64_t>(o);
 728   }
 729   return 0;
 730 }
 731
 732 static int os_mkdir(const char *path, mode_t mode)
 733 {
 734   fuse_context *fc = fuse_get_context();
 735   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
 736   ldout(fs->store->cct, 10) << __func__ << " " << path << dendl;
 737   coll_t cid;
 738   ghobject_t oid;
 739   string key;
 740   uint32_t hash_value, hash_bits;
 741   int f = parse_fn(fs->store->cct, path, &cid, &oid, &key, &hash_value,
 742                    &hash_bits);
 743   if (f < 0)
 744     return f;
 745
 746   std::lock_guard<std::mutex> l(fs->lock);
 747
 748   ObjectStore::CollectionHandle ch;
 749
 750   ObjectStore::Transaction t;
 751   switch (f) {
 752   case FN_OBJECT:
 753     {
 754       ch = fs->store->open_collection(cid);
 755       if (!ch) {
 756         return -ENOENT;
 757       }
 758       spg_t pgid;
 759       if (cid.is_pg(&pgid)) {
 760         int bits = fs->store->collection_bits(ch);
 761         if (bits >= 0 && !oid.match(bits, pgid.ps())) {
 762           // sorry, not part of this PG
 763           return -EINVAL;
 764         }
 765       }
 766       t.touch(cid, oid);
 767       ch = fs->store->open_collection(cid);
 768     }
 769     break;
 770
 771   case FN_COLLECTION:
 772     if (cid.is_pg()) {
 773       // use the mode for the bit count.  e.g., mkdir --mode=0003
 774       // mnt/0.7_head will create 0.7 with bits = 3.
 775       mode &= 0777;
 776       if (mode >= 32)
 777         return -EINVAL;
 778     } else {
 779       mode = 0;
 780     }
 781     t.create_collection(cid, mode);
 782     ch = fs->store->create_new_collection(cid);
 783     break;
 784
 785   default:
 786     return -EPERM;
 787   }
 788
 789   if (!t.empty()) {
 790     fs->store->queue_transaction(ch, std::move(t));
 791   }
 792
 793   return 0;
 794 }
 795
 796 static int os_chmod(const char *path, mode_t mode
 797 #if FUSE_VERSION >= FUSE_MAKE_VERSION(3, 0)
 798                     , struct fuse_file_info *fi
 799 #endif
 800                     )
 801 {
 802   fuse_context *fc = fuse_get_context();
 803   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
 804   ldout(fs->store->cct, 10) << __func__ << " " << path << dendl;
 805   return 0;
 806 }
 807
 808 static int os_create(const char *path, mode_t mode, struct fuse_file_info *fi)
 809 {
 810   fuse_context *fc = fuse_get_context();
 811   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
 812   ldout(fs->store->cct, 10) << __func__ << " " << path << dendl;
 813   coll_t cid;
 814   ghobject_t oid;
 815   string key;
 816   uint32_t hash_value, hash_bits;
 817   int f = parse_fn(fs->store->cct, path, &cid, &oid, &key, &hash_value,
 818                    &hash_bits);
 819   if (f < 0)
 820     return f;
 821
 822   std::lock_guard<std::mutex> l(fs->lock);
 823
 824   ObjectStore::CollectionHandle ch = fs->store->open_collection(cid);
 825
 826   ObjectStore::Transaction t;
 827   bufferlist *pbl = 0;
 828   switch (f) {
 829   case FN_OBJECT_DATA:
 830     {
 831       pbl = new bufferlist;
 832       fs->store->read(ch, oid, 0, 0, *pbl);
 833     }
 834     break;
 835
 836   case FN_OBJECT_ATTR_VAL:
 837     {
 838       pbl = new bufferlist;
 839       bufferptr bp;
 840       int r = fs->store->getattr(ch, oid, key.c_str(), bp);
 841       if (r == -ENODATA) {
 842         bufferlist empty;
 843         t.setattr(cid, oid, key.c_str(), empty);
 844       }
 845       pbl->append(bp);
 846     }
 847     break;
 848
 849   case FN_OBJECT_OMAP_VAL:
 850     {
 851       pbl = new bufferlist;
 852       set<string> k;
 853       k.insert(key);
 854       map<string,bufferlist> v;
 855       fs->store->omap_get_values(ch, oid, k, &v);
 856       if (v.count(key) == 0) {
 857         map<string,bufferlist> aset;
 858         aset[key] = bufferlist();
 859         t.omap_setkeys(cid, oid, aset);
 860       } else {
 861         *pbl = v[key];
 862       }
 863     }
 864     break;
 865   }
 866
 867   if (!t.empty()) {
 868     fs->store->queue_transaction(ch, std::move(t));
 869   }
 870
 871   if (pbl) {
 872     FuseStore::OpenFile *o = new FuseStore::OpenFile;
 873     o->bl.claim(*pbl);
 874     o->dirty = true;
 875     fi->fh = reinterpret_cast<uint64_t>(o);
 876   }
 877   return 0;
 878 }
 879
 880 static int os_release(const char *path, struct fuse_file_info *fi)
 881 {
 882   fuse_context *fc = fuse_get_context();
 883   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
 884   ldout(fs->store->cct, 10) << __func__ << " " << path << dendl;
 885   std::lock_guard<std::mutex> l(fs->lock);
 886   FuseStore::OpenFile *o = reinterpret_cast<FuseStore::OpenFile*>(fi->fh);
 887   if (--o->ref == 0) {
 888     ldout(fs->store->cct, 10) << __func__ << " closing last " << o->path << dendl;
 889     fs->open_files.erase(o->path);
 890     delete o;
 891   }
 892   return 0;
 893 }
 894
 895 static int os_read(const char *path, char *buf, size_t size, off_t offset,
 896                    struct fuse_file_info *fi)
 897 {
 898   fuse_context *fc = fuse_get_context();
 899   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
 900   ldout(fs->store->cct, 10) << __func__ << " " << path << " offset " << offset
 901                      << " size " << size << dendl;
 902   std::lock_guard<std::mutex> l(fs->lock);
 903   FuseStore::OpenFile *o = reinterpret_cast<FuseStore::OpenFile*>(fi->fh);
 904   if (!o)
 905     return 0;
 906   if (offset >= o->bl.length())
 907     return 0;
 908   if (offset + size > o->bl.length())
 909     size = o->bl.length() - offset;
 910   bufferlist r;
 911   r.substr_of(o->bl, offset, size);
 912   memcpy(buf, r.c_str(), r.length());
 913   return r.length();
 914 }
 915
 916 static int os_write(const char *path, const char *buf, size_t size,
 917                     off_t offset, struct fuse_file_info *fi)
 918 {
 919   fuse_context *fc = fuse_get_context();
 920   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
 921   ldout(fs->store->cct, 10) << __func__ << " " << path << " offset " << offset
 922                      << " size " << size << dendl;
 923   std::lock_guard<std::mutex> l(fs->lock);
 924   FuseStore::OpenFile *o = reinterpret_cast<FuseStore::OpenFile*>(fi->fh);
 925   if (!o)
 926     return 0;
 927
 928   bufferlist final;
 929   if (offset) {
 930     if (offset > o->bl.length()) {
 931       final.substr_of(o->bl, 0, offset);
 932     } else {
 933       final.claim_append(o->bl);
 934       size_t zlen = offset - final.length();
 935       final.append_zero(zlen);
 936     }
 937   }
 938   final.append(buf, size);
 939   if (offset + size < o->bl.length()) {
 940     bufferlist rest;
 941     rest.substr_of(o->bl, offset + size, o->bl.length() - offset - size);
 942     final.claim_append(rest);
 943   }
 944   o->bl = final;
 945   o->dirty = true;
 946   return size;
 947 }
 948
 949 int os_flush(const char *path, struct fuse_file_info *fi)
 950 {
 951   fuse_context *fc = fuse_get_context();
 952   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
 953   ldout(fs->store->cct, 10) << __func__ << " " << path << dendl;
 954   coll_t cid;
 955   ghobject_t oid;
 956   string key;
 957   uint32_t hash_value, hash_bits;
 958   int f = parse_fn(fs->store->cct, path, &cid, &oid, &key, &hash_value,
 959                    &hash_bits);
 960   if (f < 0)
 961     return f;
 962
 963   std::lock_guard<std::mutex> l(fs->lock);
 964
 965   FuseStore::OpenFile *o = reinterpret_cast<FuseStore::OpenFile*>(fi->fh);
 966   if (!o)
 967     return 0;
 968   if (!o->dirty)
 969     return 0;
 970
 971   ObjectStore::CollectionHandle ch = fs->store->open_collection(cid);
 972
 973   ObjectStore::Transaction t;
 974
 975   switch (f) {
 976   case FN_OBJECT_DATA:
 977     t.write(cid, oid, 0, o->bl.length(), o->bl);
 978     break;
 979
 980   case FN_OBJECT_ATTR_VAL:
 981     t.setattr(cid, oid, key.c_str(), o->bl);
 982     break;
 983
 984   case FN_OBJECT_OMAP_VAL:
 985     {
 986       map<string,bufferlist> aset;
 987       aset[key] = o->bl;
 988       t.omap_setkeys(cid, oid, aset);
 989       break;
 990     }
 991
 992   case FN_OBJECT_OMAP_HEADER:
 993     t.omap_setheader(cid, oid, o->bl);
 994     break;
 995
 996   default:
 997     return 0;
 998   }
 999
1000   fs->store->queue_transaction(ch, std::move(t));
1001
1002   return 0;
1003 }
1004
1005 static int os_unlink(const char *path)
1006 {
1007   fuse_context *fc = fuse_get_context();
1008   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
1009   ldout(fs->store->cct, 10) << __func__ << " " << path << dendl;
1010   coll_t cid;
1011   ghobject_t oid;
1012   string key;
1013   uint32_t hash_value, hash_bits;
1014   int f = parse_fn(fs->store->cct, path, &cid, &oid, &key, &hash_value,
1015                    &hash_bits);
1016   if (f < 0)
1017     return f;
1018
1019   std::lock_guard<std::mutex> l(fs->lock);
1020
1021   ObjectStore::CollectionHandle ch = fs->store->open_collection(cid);
1022   ObjectStore::Transaction t;
1023
1024   switch (f) {
1025   case FN_OBJECT_OMAP_VAL:
1026     {
1027       t.omap_rmkey(cid, oid, key);
1028     }
1029     break;
1030
1031   case FN_OBJECT_ATTR_VAL:
1032     t.rmattr(cid, oid, key.c_str());
1033     break;
1034
1035   case FN_OBJECT_OMAP_HEADER:
1036     {
1037       bufferlist empty;
1038       t.omap_setheader(cid, oid, empty);
1039     }
1040     break;
1041
1042   case FN_OBJECT:
1043     t.remove(cid, oid);
1044     break;
1045
1046   case FN_COLLECTION:
1047     {
1048       bool empty;
1049       int r = fs->store->collection_empty(ch, &empty);
1050       if (r < 0)
1051         return r;
1052       if (!empty)
1053         return -ENOTEMPTY;
1054       t.remove_collection(cid);
1055     }
1056     break;
1057
1058   case FN_OBJECT_DATA:
1059     t.truncate(cid, oid, 0);
1060     break;
1061
1062   default:
1063     return -EPERM;
1064   }
1065
1066   fs->store->queue_transaction(ch, std::move(t));
1067
1068   return 0;
1069 }
1070
1071 static int os_truncate(const char *path, off_t size
1072 #if FUSE_VERSION >= FUSE_MAKE_VERSION(3, 0)
1073                        , struct fuse_file_info *fi
1074 #endif
1075                        )
1076 {
1077   fuse_context *fc = fuse_get_context();
1078   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
1079   ldout(fs->store->cct, 10) << __func__ << " " << path << " size " << size << dendl;
1080   coll_t cid;
1081   ghobject_t oid;
1082   string key;
1083   uint32_t hash_value, hash_bits;
1084   int f = parse_fn(fs->store->cct, path, &cid, &oid, &key, &hash_value,
1085                    &hash_bits);
1086   if (f < 0)
1087     return f;
1088
1089   if (f == FN_OBJECT_OMAP_VAL ||
1090       f == FN_OBJECT_ATTR_VAL ||
1091       f == FN_OBJECT_OMAP_HEADER) {
1092     if (size)
1093       return -EPERM;
1094     return 0;
1095   }
1096   if (f != FN_OBJECT_DATA)
1097     return -EPERM;
1098
1099   std::lock_guard<std::mutex> l(fs->lock);
1100
1101   if (fs->open_files.count(path)) {
1102     FuseStore::OpenFile *o = fs->open_files[path];
1103     if (o->bl.length() > size) {
1104       bufferlist t;
1105       t.substr_of(o->bl, 0, size);
1106       o->bl.swap(t);
1107     }
1108   }
1109
1110   ObjectStore::CollectionHandle ch = fs->store->open_collection(cid);
1111   ObjectStore::Transaction t;
1112   t.truncate(cid, oid, size);
1113   fs->store->queue_transaction(ch, std::move(t));
1114   return 0;
1115 }
1116
1117 static int os_statfs(const char *path, struct statvfs *stbuf)
1118 {
1119   fuse_context *fc = fuse_get_context();
1120   FuseStore *fs = static_cast<FuseStore*>(fc->private_data);
1121   ldout(fs->store->cct, 10) << __func__ << " " << path << dendl;
1122   std::lock_guard<std::mutex> l(fs->lock);
1123
1124   struct store_statfs_t s;
1125   int r = fs->store->statfs(&s);
1126   if (r < 0)
1127     return r;
1128   stbuf->f_bsize = 4096;   // LIES!
1129   stbuf->f_blocks = s.total / 4096;
1130   stbuf->f_bavail = s.available / 4096;
1131   stbuf->f_bfree = stbuf->f_bavail;
1132
1133   ldout(fs->store->cct, 10) << __func__ << " " << path << ": "
1134     << stbuf->f_bavail << "/" << stbuf->f_blocks << dendl;
1135   return 0;
1136 }
1137
1138 static struct fuse_operations fs_oper = {
1139   getattr: os_getattr,
1140   readlink: 0,
1141 #if FUSE_VERSION < FUSE_MAKE_VERSION(3, 0)
1142   getdir: 0,
1143 #endif
1144   mknod: 0,
1145   mkdir: os_mkdir,
1146   unlink: os_unlink,
1147   rmdir: os_unlink,
1148   symlink: 0,
1149   rename: 0,
1150   link: 0,
1151   chmod: os_chmod,
1152   chown: 0,
1153   truncate: os_truncate,
1154 #if FUSE_VERSION < FUSE_MAKE_VERSION(3, 0)
1155   utime: 0,
1156 #endif
1157   open: os_open,
1158   read: os_read,
1159   write: os_write,
1160   statfs: os_statfs,
1161   flush: os_flush,
1162   release: os_release,
1163   fsync: 0,
1164   setxattr: 0,
1165   getxattr: 0,
1166   listxattr: 0,
1167   removexattr: 0,
1168   opendir: 0,
1169   readdir: os_readdir,
1170   releasedir: 0,
1171   fsyncdir: 0,
1172   init: 0,
1173   destroy: 0,
1174   access: 0,
1175   create: os_create,
1176 };
1177
1178 int FuseStore::main()
1179 {
1180   const char *v[] = {
1181     "foo",
1182     mount_point.c_str(),
1183     "-f",
1184     "-d", // debug
1185   };
1186   int c = 3;
1187   auto fuse_debug = store->cct->_conf.get_val<bool>("fuse_debug");
1188   if (fuse_debug)
1189     ++c;
1190   return fuse_main(c, (char**)v, &fs_oper, (void*)this);
1191 }
1192
1193 int FuseStore::start()
1194 {
1195   dout(10) << __func__ << dendl;
1196
1197   memset(&info->args, 0, sizeof(info->args));
1198   const char *v[] = {
1199     "foo",
1200     mount_point.c_str(),
1201     "-f", // foreground
1202     "-d", // debug
1203   };
1204   int c = 3;
1205 #if FUSE_VERSION >= FUSE_MAKE_VERSION(3, 0)
1206   int rc;
1207   struct fuse_cmdline_opts opts = {};
1208 #endif
1209   auto fuse_debug = store->cct->_conf.get_val<bool>("fuse_debug");
1210   if (fuse_debug)
1211     ++c;
1212   fuse_args a = FUSE_ARGS_INIT(c, (char**)v);
1213   info->args = a;
1214 #if FUSE_VERSION >= FUSE_MAKE_VERSION(3, 0)
1215   if (fuse_parse_cmdline(&info->args, &opts) == -1) {
1216 #else
1217   if (fuse_parse_cmdline(&info->args, &info->mountpoint, NULL, NULL) == -1) {
1218 #endif
1219     derr << __func__ << " failed to parse args" << dendl;
1220     return -EINVAL;
1221   }
1222
1223 #if FUSE_VERSION >= FUSE_MAKE_VERSION(3, 0)
1224   info->mountpoint = opts.mountpoint;
1225   info->f = fuse_new(&info->args, &fs_oper, sizeof(fs_oper), (void*)this);
1226   if (!info->f) {
1227     derr << __func__ << " fuse_new failed" << dendl;
1228     return -EIO;
1229   }
1230
1231   rc = fuse_mount(info->f, info->mountpoint);
1232   if (rc != 0) {
1233     derr << __func__ << " fuse_mount failed" << dendl;
1234     return -EIO;
1235   }
1236 #else
1237   info->ch = fuse_mount(info->mountpoint, &info->args);
1238   if (!info->ch) {
1239     derr << __func__ << " fuse_mount failed" << dendl;
1240     return -EIO;
1241   }
1242
1243   info->f = fuse_new(info->ch, &info->args, &fs_oper, sizeof(fs_oper),
1244                      (void*)this);
1245   if (!info->f) {
1246     fuse_unmount(info->mountpoint, info->ch);
1247     derr << __func__ << " fuse_new failed" << dendl;
1248     return -EIO;
1249   }
1250 #endif
1251
1252   fuse_thread.create("fusestore");
1253   dout(10) << __func__ << " done" << dendl;
1254   return 0;
1255 }
1256
1257 int FuseStore::loop()
1258 {
1259   dout(10) << __func__ << " enter" << dendl;
1260   int r = fuse_loop(info->f);
1261   if (r)
1262     derr << __func__ << " got " << cpp_strerror(r) << dendl;
1263   dout(10) << __func__ << " exit" << dendl;
1264   return r;
1265 }
1266
1267 int FuseStore::stop()
1268 {
1269   dout(10) << __func__ << " enter" << dendl;
1270 #if FUSE_VERSION >= FUSE_MAKE_VERSION(3, 0)
1271   fuse_unmount(info->f);
1272 #else
1273   fuse_unmount(info->mountpoint, info->ch);
1274 #endif
1275   fuse_thread.join();
1276   fuse_destroy(info->f);
1277   dout(10) << __func__ << " exit" << dendl;
1278   return 0;
1279 }