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