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