client: allow getattr, lookup, fstatx, and readdir to use implemented caps

caps_issued_mask is a function that we call to see if we have all of
the caps provided in a mask. Usually this ignores caps that have been
revoked by the MDS but that have not yet been released by the client.
These caps are still valid though -- nothing else can use them until
they've been released.

These codepaths call caps_issued_mask only in order to determine
whether the client holds enough caps to trust cached attributes. They
don't take new references to the caps, so it should be safe to use the
"implemented" capset in these functions.

This helps break a deadlock that can occur when the client holds a
delegation that has been recalled, but needs to satisfy a getattr
request before releasing it.

Tracker: http://tracker.ceph.com/issues/23028
Signed-off-by: Jeff Layton <jlayton@redhat.com>
(cherry picked from commit dacc7b6099f5c2e758e2811a7a6722bd0946eb39)

Conflicts:
	src/client/Inode.cc: Resolved in Inode::caps_issued_mask

diff --git a/src/client/Client.cc b/src/client/Client.cc
index e0b0771893a322431506856a6329973be3dd63a3..8bae2a3fac8fd5db1e56ce107bd5699770c1398b 100644 (file)
--- a/src/client/Client.cc
+++ b/src/client/Client.cc
@@ -6151,7 +6151,7 @@ int Client::_lookup(Inode *dir, const string& dname, int mask, InodeRef *target,
             << " seq " << dn->lease_seq
             << dendl;
 
-    if (!dn->inode || dn->inode->caps_issued_mask(mask)) {
+    if (!dn->inode || dn->inode->caps_issued_mask(mask, true)) {
       // is dn lease valid?
       utime_t now = ceph_clock_now();
       if (dn->lease_mds >= 0 &&
@@ -6169,9 +6169,9 @@ int Client::_lookup(Inode *dir, const string& dname, int mask, InodeRef *target,
                       << " vs lease_gen " << dn->lease_gen << dendl;
       }
       // dir lease?
-      if (dir->caps_issued_mask(CEPH_CAP_FILE_SHARED)) {
+      if (dir->caps_issued_mask(CEPH_CAP_FILE_SHARED, true)) {
        if (dn->cap_shared_gen == dir->shared_gen &&
-           (!dn->inode || dn->inode->caps_issued_mask(mask)))
+           (!dn->inode || dn->inode->caps_issued_mask(mask, true)))
              goto hit_dn;
        if (!dn->inode && (dir->flags & I_COMPLETE)) {
          ldout(cct, 10) << "_lookup concluded ENOENT locally for "
@@ -6184,7 +6184,7 @@ int Client::_lookup(Inode *dir, const string& dname, int mask, InodeRef *target,
     }
   } else {
     // can we conclude ENOENT locally?
-    if (dir->caps_issued_mask(CEPH_CAP_FILE_SHARED) &&
+    if (dir->caps_issued_mask(CEPH_CAP_FILE_SHARED, true) &&
        (dir->flags & I_COMPLETE)) {
       ldout(cct, 10) << "_lookup concluded ENOENT locally for " << *dir << " dn '" << dname << "'" << dendl;
       return -ENOENT;
@@ -6648,7 +6648,7 @@ int Client::_readlink(Inode *in, char *buf, size_t size)
 
 int Client::_getattr(Inode *in, int mask, const UserPerm& perms, bool force)
 {
-  bool yes = in->caps_issued_mask(mask);
+  bool yes = in->caps_issued_mask(mask, true);
 
   ldout(cct, 10) << "_getattr mask " << ccap_string(mask) << " issued=" << yes << dendl;
   if (yes && !force)
@@ -7854,7 +7854,7 @@ int Client::readdir_r_cb(dir_result_t *d, add_dirent_cb_t cb, void *p,
           << dendl;
   if (dirp->inode->snapid != CEPH_SNAPDIR &&
       dirp->inode->is_complete_and_ordered() &&
-      dirp->inode->caps_issued_mask(CEPH_CAP_FILE_SHARED)) {
+      dirp->inode->caps_issued_mask(CEPH_CAP_FILE_SHARED, true)) {
     int err = _readdir_cache_cb(dirp, cb, p, caps, getref);
     if (err != -EAGAIN)
       return err;
@@ -9553,7 +9553,7 @@ int Client::fstatx(int fd, struct ceph_statx *stx, const UserPerm& perms,
   unsigned mask = statx_to_mask(flags, want);
 
   int r = 0;
-  if (mask && !f->inode->caps_issued_mask(mask)) {
+  if (mask && !f->inode->caps_issued_mask(mask, true)) {
     r = _getattr(f->inode, mask, perms);
     if (r < 0) {
       ldout(cct, 3) << "fstatx exit on error!" << dendl;
@@ -10571,7 +10571,7 @@ int Client::ll_getattrx(Inode *in, struct ceph_statx *stx, unsigned int want,
   int res = 0;
   unsigned mask = statx_to_mask(flags, want);
 
-  if (mask && !in->caps_issued_mask(mask))
+  if (mask && !in->caps_issued_mask(mask, true))
     res = _ll_getattr(in, mask, perms);
 
   if (res == 0)

diff --git a/src/client/Inode.cc b/src/client/Inode.cc
index d29fb25d396a0ecbec701d001aa166abd3a94977..0e70697600a5c767e83725c7823a5606fdb139e9 100644 (file)
--- a/src/client/Inode.cc
+++ b/src/client/Inode.cc
@@ -215,9 +215,27 @@ void Inode::try_touch_cap(mds_rank_t mds)
     touch_cap(caps[mds]);
 }
 
-bool Inode::caps_issued_mask(unsigned mask)
+/**
+ * caps_issued_mask - check whether we have all of the caps in the mask
+ * @mask: mask to check against
+ * @allow_impl: whether the caller can also use caps that are implemented but not issued
+ *
+ * This is the bog standard "check whether we have the required caps" operation.
+ * Typically, we only check against the capset that is currently "issued".
+ * In other words, we ignore caps that have been revoked but not yet released.
+ *
+ * Some callers (particularly those doing attribute retrieval) can also make
+ * use of the full set of "implemented" caps to satisfy requests from the
+ * cache.
+ *
+ * Those callers should refrain from taking new references to implemented
+ * caps!
+ */
+bool Inode::caps_issued_mask(unsigned mask, bool allow_impl)
 {
   int c = snap_caps;
+  int i = 0;
+
   if ((c & mask) == mask)
     return true;
   // prefer auth cap
@@ -237,8 +255,13 @@ bool Inode::caps_issued_mask(unsigned mask)
        return true;
       }
       c |= it->second->issued;
+      i |= it->second->implemented;
     }
   }
+
+  if (allow_impl)
+    c |= i;
+
   if ((c & mask) == mask) {
     // bah.. touch them all
     for (map<mds_rank_t,Cap*>::iterator it = caps.begin();

diff --git a/src/client/Inode.h b/src/client/Inode.h
index 71586cad78aebde19c953a04d7d43e6ae5339d61..5242630d5d1a23b4954a6f4c27d0185d18a78ab7 100644 (file)
--- a/src/client/Inode.h
+++ b/src/client/Inode.h
@@ -283,7 +283,7 @@ struct Inode {
   int caps_issued(int *implemented = 0) const;
   void touch_cap(Cap *cap);
   void try_touch_cap(mds_rank_t mds);
-  bool caps_issued_mask(unsigned mask);
+  bool caps_issued_mask(unsigned mask, bool allow_impl=false);
   int caps_used();
   int caps_file_wanted();
   int caps_wanted();