ceph/src/client/Inode.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 "Client.h"
   5 #include "Inode.h"
   6 #include "Dentry.h"
   7 #include "Dir.h"
   8 #include "Fh.h"
   9 #include "MetaSession.h"
  10 #include "ClientSnapRealm.h"
  11 #include "Delegation.h"
  12
  13 #include "mds/flock.h"
  14
  15 Inode::~Inode()
  16 {
  17   delay_cap_item.remove_myself();
  18   dirty_cap_item.remove_myself();
  19   snaprealm_item.remove_myself();
  20
  21   if (snapdir_parent) {
  22     snapdir_parent->flags &= ~I_SNAPDIR_OPEN;
  23     snapdir_parent.reset();
  24   }
  25
  26   if (!oset.objects.empty()) {
  27     lsubdout(client->cct, client, 0) << __func__ << ": leftover objects on inode 0x"
  28       << std::hex << ino << std::dec << dendl;
  29     ceph_assert(oset.objects.empty());
  30   }
  31
  32   if (!delegations.empty()) {
  33     lsubdout(client->cct, client, 0) << __func__ << ": leftover delegations on inode 0x"
  34       << std::hex << ino << std::dec << dendl;
  35     ceph_assert(delegations.empty());
  36   }
  37 }
  38
  39 ostream& operator<<(ostream &out, const Inode &in)
  40 {
  41   out << in.vino() << "("
  42       << "faked_ino=" << in.faked_ino
  43       << " ref=" << in._ref
  44       << " ll_ref=" << in.ll_ref
  45       << " cap_refs=" << in.cap_refs
  46       << " open=" << in.open_by_mode
  47       << " mode=" << oct << in.mode << dec
  48       << " size=" << in.size << "/" << in.max_size
  49       << " nlink=" << in.nlink
  50       << " btime=" << in.btime
  51       << " mtime=" << in.mtime
  52       << " ctime=" << in.ctime
  53       << " caps=" << ccap_string(in.caps_issued());
  54   if (!in.caps.empty()) {
  55     out << "(";
  56     bool first = true;
  57     for (const auto &pair : in.caps) {
  58       if (!first)
  59         out << ',';
  60       out << pair.first << '=' << ccap_string(pair.second.issued);
  61       first = false;
  62     }
  63     out << ")";
  64   }
  65   if (in.dirty_caps)
  66     out << " dirty_caps=" << ccap_string(in.dirty_caps);
  67   if (in.flushing_caps)
  68     out << " flushing_caps=" << ccap_string(in.flushing_caps);
  69
  70   if (in.flags & I_COMPLETE)
  71     out << " COMPLETE";
  72
  73   if (in.is_file())
  74     out << " " << in.oset;
  75
  76   if (!in.dentries.empty())
  77     out << " parents=" << in.dentries;
  78
  79   if (in.is_dir() && in.has_dir_layout())
  80     out << " has_dir_layout";
  81
  82   if (in.quota.is_enable())
  83     out << " " << in.quota;
  84
  85   out << ' ' << &in << ")";
  86   return out;
  87 }
  88
  89
  90 void Inode::make_long_path(filepath& p)
  91 {
  92   if (!dentries.empty()) {
  93     Dentry *dn = get_first_parent();
  94     ceph_assert(dn->dir && dn->dir->parent_inode);
  95     dn->dir->parent_inode->make_long_path(p);
  96     p.push_dentry(dn->name);
  97   } else if (snapdir_parent) {
  98     snapdir_parent->make_nosnap_relative_path(p);
  99     string empty;
 100     p.push_dentry(empty);
 101   } else
 102     p = filepath(ino);
 103 }
 104
 105 /*
 106  * make a filepath suitable for an mds request:
 107  *  - if we are non-snapped/live, the ino is sufficient, e.g. #1234
 108  *  - if we are snapped, make filepath relative to first non-snapped parent.
 109  */
 110 void Inode::make_nosnap_relative_path(filepath& p)
 111 {
 112   if (snapid == CEPH_NOSNAP) {
 113     p = filepath(ino);
 114   } else if (snapdir_parent) {
 115     snapdir_parent->make_nosnap_relative_path(p);
 116     string empty;
 117     p.push_dentry(empty);
 118   } else if (!dentries.empty()) {
 119     Dentry *dn = get_first_parent();
 120     ceph_assert(dn->dir && dn->dir->parent_inode);
 121     dn->dir->parent_inode->make_nosnap_relative_path(p);
 122     p.push_dentry(dn->name);
 123   } else {
 124     p = filepath(ino);
 125   }
 126 }
 127
 128 void Inode::get_open_ref(int mode)
 129 {
 130   open_by_mode[mode]++;
 131   break_deleg(!(mode & CEPH_FILE_MODE_WR));
 132 }
 133
 134 bool Inode::put_open_ref(int mode)
 135 {
 136   //cout << "open_by_mode[" << mode << "] " << open_by_mode[mode] << " -> " << (open_by_mode[mode]-1) << std::endl;
 137   auto& ref = open_by_mode.at(mode);
 138   ceph_assert(ref > 0);
 139   if (--ref == 0)
 140     return true;
 141   return false;
 142 }
 143
 144 void Inode::get_cap_ref(int cap)
 145 {
 146   int n = 0;
 147   while (cap) {
 148     if (cap & 1) {
 149       int c = 1 << n;
 150       cap_refs[c]++;
 151       //cout << "inode " << *this << " get " << cap_string(c) << " " << (cap_refs[c]-1) << " -> " << cap_refs[c] << std::endl;
 152     }
 153     cap >>= 1;
 154     n++;
 155   }
 156 }
 157
 158 int Inode::put_cap_ref(int cap)
 159 {
 160   int last = 0;
 161   int n = 0;
 162   while (cap) {
 163     if (cap & 1) {
 164       int c = 1 << n;
 165       if (cap_refs[c] <= 0) {
 166         lderr(client->cct) << "put_cap_ref " << ccap_string(c) << " went negative on " << *this << dendl;
 167         ceph_assert(cap_refs[c] > 0);
 168       }
 169       if (--cap_refs[c] == 0)
 170         last |= c;
 171       //cout << "inode " << *this << " put " << cap_string(c) << " " << (cap_refs[c]+1) << " -> " << cap_refs[c] << std::endl;
 172     }
 173     cap >>= 1;
 174     n++;
 175   }
 176   return last;
 177 }
 178
 179 bool Inode::is_any_caps()
 180 {
 181   return !caps.empty() || snap_caps;
 182 }
 183
 184 bool Inode::cap_is_valid(const Cap &cap) const
 185 {
 186   /*cout << "cap_gen     " << cap->session-> cap_gen << std::endl
 187     << "session gen " << cap->gen << std::endl
 188     << "cap expire  " << cap->session->cap_ttl << std::endl
 189     << "cur time    " << ceph_clock_now(cct) << std::endl;*/
 190   if ((cap.session->cap_gen <= cap.gen)
 191       && (ceph_clock_now() < cap.session->cap_ttl)) {
 192     return true;
 193   }
 194   return false;
 195 }
 196
 197 int Inode::caps_issued(int *implemented) const
 198 {
 199   int c = snap_caps;
 200   int i = 0;
 201   for (const auto &pair : caps) {
 202     const Cap &cap = pair.second;
 203     if (cap_is_valid(cap)) {
 204       c |= cap.issued;
 205       i |= cap.implemented;
 206     }
 207   }
 208   // exclude caps issued by non-auth MDS, but are been revoking by
 209   // the auth MDS. The non-auth MDS should be revoking/exporting
 210   // these caps, but the message is delayed.
 211   if (auth_cap)
 212     c &= ~auth_cap->implemented | auth_cap->issued;
 213
 214   if (implemented)
 215     *implemented = i;
 216   return c;
 217 }
 218
 219 void Inode::try_touch_cap(mds_rank_t mds)
 220 {
 221   auto it = caps.find(mds);
 222   if (it != caps.end()) {
 223     it->second.touch();
 224   }
 225 }
 226
 227 /**
 228  * caps_issued_mask - check whether we have all of the caps in the mask
 229  * @mask: mask to check against
 230  * @allow_impl: whether the caller can also use caps that are implemented but not issued
 231  *
 232  * This is the bog standard "check whether we have the required caps" operation.
 233  * Typically, we only check against the capset that is currently "issued".
 234  * In other words, we ignore caps that have been revoked but not yet released.
 235  *
 236  * Some callers (particularly those doing attribute retrieval) can also make
 237  * use of the full set of "implemented" caps to satisfy requests from the
 238  * cache.
 239  *
 240  * Those callers should refrain from taking new references to implemented
 241  * caps!
 242  */
 243 bool Inode::caps_issued_mask(unsigned mask, bool allow_impl)
 244 {
 245   int c = snap_caps;
 246   int i = 0;
 247
 248   if ((c & mask) == mask)
 249     return true;
 250   // prefer auth cap
 251   if (auth_cap &&
 252       cap_is_valid(*auth_cap) &&
 253       (auth_cap->issued & mask) == mask) {
 254     auth_cap->touch();
 255     return true;
 256   }
 257   // try any cap
 258   for (auto &pair : caps) {
 259     Cap &cap = pair.second;
 260     if (cap_is_valid(cap)) {
 261       if ((cap.issued & mask) == mask) {
 262         cap.touch();
 263         return true;
 264       }
 265       c |= cap.issued;
 266       i |= cap.implemented;
 267     }
 268   }
 269
 270   if (allow_impl)
 271     c |= i;
 272
 273   if ((c & mask) == mask) {
 274     // bah.. touch them all
 275     for (auto &pair : caps) {
 276       pair.second.touch();
 277     }
 278     return true;
 279   }
 280   return false;
 281 }
 282
 283 int Inode::caps_used()
 284 {
 285   int w = 0;
 286   for (map<int,int>::iterator p = cap_refs.begin();
 287        p != cap_refs.end();
 288        ++p)
 289     if (p->second)
 290       w |= p->first;
 291   return w;
 292 }
 293
 294 int Inode::caps_file_wanted()
 295 {
 296   int want = 0;
 297   for (map<int,int>::iterator p = open_by_mode.begin();
 298        p != open_by_mode.end();
 299        ++p)
 300     if (p->second)
 301       want |= ceph_caps_for_mode(p->first);
 302   return want;
 303 }
 304
 305 int Inode::caps_wanted()
 306 {
 307   int want = caps_file_wanted() | caps_used();
 308   if (want & CEPH_CAP_FILE_BUFFER)
 309     want |= CEPH_CAP_FILE_EXCL;
 310   return want;
 311 }
 312
 313 int Inode::caps_mds_wanted()
 314 {
 315   int want = 0;
 316   for (const auto &pair : caps) {
 317     want |= pair.second.wanted;
 318   }
 319   return want;
 320 }
 321
 322 int Inode::caps_dirty()
 323 {
 324   return dirty_caps | flushing_caps;
 325 }
 326
 327 const UserPerm* Inode::get_best_perms()
 328 {
 329   const UserPerm *perms = NULL;
 330   for (const auto &pair : caps) {
 331     const UserPerm& iperm = pair.second.latest_perms;
 332     if (!perms) { // we don't have any, take what's present
 333       perms = &iperm;
 334     } else if (iperm.uid() == uid) {
 335       if (iperm.gid() == gid) { // we have the best possible, return
 336         return &iperm;
 337       }
 338       if (perms->uid() != uid) { // take uid > gid every time
 339         perms = &iperm;
 340       }
 341     } else if (perms->uid() != uid && iperm.gid() == gid) {
 342       perms = &iperm; // a matching gid is better than nothing
 343     }
 344   }
 345   return perms;
 346 }
 347
 348 bool Inode::have_valid_size()
 349 {
 350   // RD+RDCACHE or WR+WRBUFFER => valid size
 351   if (caps_issued() & (CEPH_CAP_FILE_SHARED | CEPH_CAP_FILE_EXCL))
 352     return true;
 353   return false;
 354 }
 355
 356 // open Dir for an inode.  if it's not open, allocated it (and pin dentry in memory).
 357 Dir *Inode::open_dir()
 358 {
 359   if (!dir) {
 360     dir = new Dir(this);
 361     lsubdout(client->cct, client, 15) << "open_dir " << dir << " on " << this << dendl;
 362     ceph_assert(dentries.size() < 2); // dirs can't be hard-linked
 363     if (!dentries.empty())
 364       get_first_parent()->get();      // pin dentry
 365     get();                  // pin inode
 366   }
 367   return dir;
 368 }
 369
 370 bool Inode::check_mode(const UserPerm& perms, unsigned want)
 371 {
 372   if (uid == perms.uid()) {
 373     // if uid is owner, owner entry determines access
 374     want = want << 6;
 375   } else if (perms.gid_in_groups(gid)) {
 376     // if a gid or sgid matches the owning group, group entry determines access
 377     want = want << 3;
 378   }
 379
 380   return (mode & want) == want;
 381 }
 382
 383 void Inode::get() {
 384   _ref++;
 385   lsubdout(client->cct, client, 15) << "inode.get on " << this << " " <<  ino << '.' << snapid
 386                                     << " now " << _ref << dendl;
 387 }
 388
 389 //private method to put a reference; see Client::put_inode()
 390 int Inode::_put(int n) {
 391   _ref -= n;
 392   lsubdout(client->cct, client, 15) << "inode.put on " << this << " " << ino << '.' << snapid
 393                                     << " now " << _ref << dendl;
 394   ceph_assert(_ref >= 0);
 395   return _ref;
 396 }
 397
 398
 399 void Inode::dump(Formatter *f) const
 400 {
 401   f->dump_stream("ino") << ino;
 402   f->dump_stream("snapid") << snapid;
 403   if (rdev)
 404     f->dump_unsigned("rdev", rdev);
 405   f->dump_stream("ctime") << ctime;
 406   f->dump_stream("btime") << btime;
 407   f->dump_stream("mode") << '0' << std::oct << mode << std::dec;
 408   f->dump_unsigned("uid", uid);
 409   f->dump_unsigned("gid", gid);
 410   f->dump_int("nlink", nlink);
 411
 412   f->dump_unsigned("size", size);
 413   f->dump_unsigned("max_size", max_size);
 414   f->dump_unsigned("truncate_seq", truncate_seq);
 415   f->dump_unsigned("truncate_size", truncate_size);
 416   f->dump_stream("mtime") << mtime;
 417   f->dump_stream("atime") << atime;
 418   f->dump_unsigned("time_warp_seq", time_warp_seq);
 419   f->dump_unsigned("change_attr", change_attr);
 420
 421   f->dump_object("layout", layout);
 422   if (is_dir()) {
 423     f->open_object_section("dir_layout");
 424     ::dump(dir_layout, f);
 425     f->close_section();
 426
 427     f->dump_bool("complete", flags & I_COMPLETE);
 428     f->dump_bool("ordered", flags & I_DIR_ORDERED);
 429
 430     /* FIXME when wip-mds-encoding is merged ***
 431     f->open_object_section("dir_stat");
 432     dirstat.dump(f);
 433     f->close_section();
 434
 435     f->open_object_section("rstat");
 436     rstat.dump(f);
 437     f->close_section();
 438     */
 439   }
 440
 441   f->dump_unsigned("version", version);
 442   f->dump_unsigned("xattr_version", xattr_version);
 443   f->dump_unsigned("flags", flags);
 444
 445   if (is_dir()) {
 446     if (!dir_contacts.empty()) {
 447       f->open_object_section("dir_contants");
 448       for (set<int>::iterator p = dir_contacts.begin(); p != dir_contacts.end(); ++p)
 449         f->dump_int("mds", *p);
 450       f->close_section();
 451     }
 452     f->dump_int("dir_hashed", (int)dir_hashed);
 453     f->dump_int("dir_replicated", (int)dir_replicated);
 454   }
 455
 456   f->open_array_section("caps");
 457   for (const auto &pair : caps) {
 458     f->open_object_section("cap");
 459     f->dump_int("mds", pair.first);
 460     if (&pair.second == auth_cap)
 461       f->dump_int("auth", 1);
 462     pair.second.dump(f);
 463     f->close_section();
 464   }
 465   f->close_section();
 466   if (auth_cap)
 467     f->dump_int("auth_cap", auth_cap->session->mds_num);
 468
 469   f->dump_stream("dirty_caps") << ccap_string(dirty_caps);
 470   if (flushing_caps) {
 471     f->dump_stream("flushings_caps") << ccap_string(flushing_caps);
 472     f->open_object_section("flushing_cap_tid");
 473     for (map<ceph_tid_t, int>::const_iterator p = flushing_cap_tids.begin();
 474          p != flushing_cap_tids.end();
 475          ++p) {
 476       string n(ccap_string(p->second));
 477       f->dump_unsigned(n.c_str(), p->first);
 478     }
 479     f->close_section();
 480   }
 481   f->dump_int("shared_gen", shared_gen);
 482   f->dump_int("cache_gen", cache_gen);
 483   if (snap_caps) {
 484     f->dump_int("snap_caps", snap_caps);
 485     f->dump_int("snap_cap_refs", snap_cap_refs);
 486   }
 487
 488   f->dump_stream("hold_caps_until") << hold_caps_until;
 489
 490   if (snaprealm) {
 491     f->open_object_section("snaprealm");
 492     snaprealm->dump(f);
 493     f->close_section();
 494   }
 495   if (!cap_snaps.empty()) {
 496     for (const auto &p : cap_snaps) {
 497       f->open_object_section("cap_snap");
 498       f->dump_stream("follows") << p.first;
 499       p.second.dump(f);
 500       f->close_section();
 501     }
 502   }
 503
 504   // open
 505   if (!open_by_mode.empty()) {
 506     f->open_array_section("open_by_mode");
 507     for (map<int,int>::const_iterator p = open_by_mode.begin(); p != open_by_mode.end(); ++p) {
 508       f->open_object_section("ref");
 509       f->dump_int("mode", p->first);
 510       f->dump_int("refs", p->second);
 511       f->close_section();
 512     }
 513     f->close_section();
 514   }
 515   if (!cap_refs.empty()) {
 516     f->open_array_section("cap_refs");
 517     for (map<int,int>::const_iterator p = cap_refs.begin(); p != cap_refs.end(); ++p) {
 518       f->open_object_section("cap_ref");
 519       f->dump_stream("cap") << ccap_string(p->first);
 520       f->dump_int("refs", p->second);
 521       f->close_section();
 522     }
 523     f->close_section();
 524   }
 525
 526   f->dump_unsigned("reported_size", reported_size);
 527   if (wanted_max_size != max_size)
 528     f->dump_unsigned("wanted_max_size", wanted_max_size);
 529   if (requested_max_size != max_size)
 530     f->dump_unsigned("requested_max_size", requested_max_size);
 531
 532   f->dump_int("ref", _ref);
 533   f->dump_int("ll_ref", ll_ref);
 534
 535   if (!dentries.empty()) {
 536     f->open_array_section("parents");
 537     for (const auto &dn : dentries) {
 538       f->open_object_section("dentry");
 539       f->dump_stream("dir_ino") << dn->dir->parent_inode->ino;
 540       f->dump_string("name", dn->name);
 541       f->close_section();
 542     }
 543     f->close_section();
 544   }
 545 }
 546
 547 void Cap::dump(Formatter *f) const
 548 {
 549   f->dump_int("mds", session->mds_num);
 550   f->dump_stream("ino") << inode.ino;
 551   f->dump_unsigned("cap_id", cap_id);
 552   f->dump_stream("issued") << ccap_string(issued);
 553   if (implemented != issued)
 554     f->dump_stream("implemented") << ccap_string(implemented);
 555   f->dump_stream("wanted") << ccap_string(wanted);
 556   f->dump_unsigned("seq", seq);
 557   f->dump_unsigned("issue_seq", issue_seq);
 558   f->dump_unsigned("mseq", mseq);
 559   f->dump_unsigned("gen", gen);
 560 }
 561
 562 void CapSnap::dump(Formatter *f) const
 563 {
 564   f->dump_stream("ino") << in->ino;
 565   f->dump_stream("issued") << ccap_string(issued);
 566   f->dump_stream("dirty") << ccap_string(dirty);
 567   f->dump_unsigned("size", size);
 568   f->dump_stream("ctime") << ctime;
 569   f->dump_stream("mtime") << mtime;
 570   f->dump_stream("atime") << atime;
 571   f->dump_int("time_warp_seq", time_warp_seq);
 572   f->dump_stream("mode") << '0' << std::oct << mode << std::dec;
 573   f->dump_unsigned("uid", uid);
 574   f->dump_unsigned("gid", gid);
 575   if (!xattrs.empty()) {
 576     f->open_object_section("xattr_lens");
 577     for (map<string,bufferptr>::const_iterator p = xattrs.begin(); p != xattrs.end(); ++p)
 578       f->dump_int(p->first.c_str(), p->second.length());
 579     f->close_section();
 580   }
 581   f->dump_unsigned("xattr_version", xattr_version);
 582   f->dump_int("writing", (int)writing);
 583   f->dump_int("dirty_data", (int)dirty_data);
 584   f->dump_unsigned("flush_tid", flush_tid);
 585 }
 586
 587 void Inode::set_async_err(int r)
 588 {
 589   for (const auto &fh : fhs) {
 590     fh->async_err = r;
 591   }
 592 }
 593
 594 bool Inode::has_recalled_deleg()
 595 {
 596   if (delegations.empty())
 597     return false;
 598
 599   // Either all delegations are recalled or none are. Just check the first.
 600   Delegation& deleg = delegations.front();
 601   return deleg.is_recalled();
 602 }
 603
 604 void Inode::recall_deleg(bool skip_read)
 605 {
 606   if (delegations.empty())
 607     return;
 608
 609   // Issue any recalls
 610   for (list<Delegation>::iterator d = delegations.begin();
 611        d != delegations.end(); ++d) {
 612
 613     Delegation& deleg = *d;
 614     deleg.recall(skip_read);
 615   }
 616 }
 617
 618 bool Inode::delegations_broken(bool skip_read)
 619 {
 620   if (delegations.empty()) {
 621     lsubdout(client->cct, client, 10) <<
 622           __func__ << ": delegations empty on " << *this << dendl;
 623     return true;
 624   }
 625
 626   if (skip_read) {
 627     Delegation& deleg = delegations.front();
 628     lsubdout(client->cct, client, 10) <<
 629         __func__ << ": read delegs only on " << *this << dendl;
 630     if (deleg.get_type() == CEPH_FILE_MODE_RD) {
 631         return true;
 632     }
 633   }
 634   lsubdout(client->cct, client, 10) <<
 635         __func__ << ": not broken" << *this << dendl;
 636   return false;
 637 }
 638
 639 void Inode::break_deleg(bool skip_read)
 640 {
 641   lsubdout(client->cct, client, 10) <<
 642           __func__ << ": breaking delegs on " << *this << dendl;
 643
 644   recall_deleg(skip_read);
 645
 646   while (!delegations_broken(skip_read))
 647     client->wait_on_list(waitfor_deleg);
 648 }
 649
 650 /**
 651  * set_deleg: request a delegation on an open Fh
 652  * @fh: filehandle on which to acquire it
 653  * @type: delegation request type
 654  * @cb: delegation recall callback function
 655  * @priv: private pointer to be passed to callback
 656  *
 657  * Attempt to acquire a delegation on an open file handle. If there are no
 658  * conflicts and we have the right caps, allocate a new delegation, fill it
 659  * out and return 0. Return an error if we can't get one for any reason.
 660  */
 661 int Inode::set_deleg(Fh *fh, unsigned type, ceph_deleg_cb_t cb, void *priv)
 662 {
 663   lsubdout(client->cct, client, 10) <<
 664           __func__ << ": inode " << *this << dendl;
 665
 666   /*
 667    * 0 deleg timeout means that they haven't been explicitly enabled. Don't
 668    * allow it, with an unusual error to make it clear.
 669    */
 670   if (!client->get_deleg_timeout())
 671     return -ETIME;
 672
 673   // Just say no if we have any recalled delegs still outstanding
 674   if (has_recalled_deleg()) {
 675     lsubdout(client->cct, client, 10) << __func__ <<
 676           ": has_recalled_deleg" << dendl;
 677     return -EAGAIN;
 678   }
 679
 680   // check vs. currently open files on this inode
 681   switch (type) {
 682   case CEPH_DELEGATION_RD:
 683     if (open_count_for_write()) {
 684       lsubdout(client->cct, client, 10) << __func__ <<
 685             ": open for write" << dendl;
 686       return -EAGAIN;
 687     }
 688     break;
 689   case CEPH_DELEGATION_WR:
 690     if (open_count() > 1) {
 691       lsubdout(client->cct, client, 10) << __func__ << ": open" << dendl;
 692       return -EAGAIN;
 693     }
 694     break;
 695   default:
 696     return -EINVAL;
 697   }
 698
 699   /*
 700    * A delegation is essentially a long-held container for cap references that
 701    * we delegate to the client until recalled. The caps required depend on the
 702    * type of delegation (read vs. rw). This is entirely an opportunistic thing.
 703    * If we don't have the necessary caps for the delegation, then we just don't
 704    * grant one.
 705    *
 706    * In principle we could request the caps from the MDS, but a delegation is
 707    * usually requested just after an open. If we don't have the necessary caps
 708    * already, then it's likely that there is some sort of conflicting access.
 709    *
 710    * In the future, we may need to add a way to have this request caps more
 711    * aggressively -- for instance, to handle WANT_DELEGATION for NFSv4.1+.
 712    */
 713   int need = ceph_deleg_caps_for_type(type);
 714   if (!caps_issued_mask(need)) {
 715     lsubdout(client->cct, client, 10) << __func__ << ": cap mismatch, have="
 716       << ccap_string(caps_issued()) << " need=" << ccap_string(need) << dendl;
 717     return -EAGAIN;
 718   }
 719
 720   for (list<Delegation>::iterator d = delegations.begin();
 721        d != delegations.end(); ++d) {
 722     Delegation& deleg = *d;
 723     if (deleg.get_fh() == fh) {
 724       deleg.reinit(type, cb, priv);
 725       return 0;
 726     }
 727   }
 728
 729   delegations.emplace_back(fh, type, cb, priv);
 730   return 0;
 731 }
 732
 733 /**
 734  * unset_deleg - remove a delegation that was previously set
 735  * @fh: file handle to clear delegation of
 736  *
 737  * Unlink delegation from the Inode (if there is one), put caps and free it.
 738  */
 739 void Inode::unset_deleg(Fh *fh)
 740 {
 741   for (list<Delegation>::iterator d = delegations.begin();
 742        d != delegations.end(); ++d) {
 743     Delegation& deleg = *d;
 744     if (deleg.get_fh() == fh) {
 745       delegations.erase(d);
 746       client->signal_cond_list(waitfor_deleg);
 747       break;
 748     }
 749   }
 750 }
 751
 752 /**
 753 * mark_caps_dirty - mark some caps dirty
 754 * @caps: the dirty caps
 755 *
 756 * note that if there is no dirty and flushing caps before, we need to pin this inode.
 757 * it will be unpined by handle_cap_flush_ack when there are no dirty and flushing caps.
 758 */
 759 void Inode::mark_caps_dirty(int caps)
 760 {
 761   lsubdout(client->cct, client, 10) << __func__ << " " << *this << " " << ccap_string(dirty_caps) << " -> "
 762            << ccap_string(dirty_caps | caps) << dendl;
 763   if (caps && !caps_dirty())
 764     get();
 765   dirty_caps |= caps;
 766   client->get_dirty_list().push_back(&dirty_cap_item);
 767 }
 768
 769 /**
 770 * mark_caps_clean - only clean the dirty_caps and caller should start flushing the dirty caps.
 771 */
 772 void Inode::mark_caps_clean()
 773 {
 774   lsubdout(client->cct, client, 10) << __func__ << " " << *this << dendl;
 775   dirty_caps = 0;
 776   dirty_cap_item.remove_myself();
 777 }
 778
 779