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