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