ceph: fetch inline data when getting Fcr cap refs

we can't use getattr to fetch inline data after getting Fcr caps,
because it can cause deadlock. The solution is try bringing inline
data to page cache when not holding any cap, and hope the inline
data page is still there after getting the Fcr caps. If the page
is still there, pin it in page cache for later IO.

Signed-off-by: Yan, Zheng <zyan@redhat.com>

1 files changed, 47 insertions, 13 deletions

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index 6372eb9ce491..795afe304871 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -2057,15 +2057,17 @@ static void __take_cap_refs(struct ceph_inode_info *ci, int got)
  * requested from the MDS.
  */
 static int try_get_cap_refs(struct ceph_inode_info *ci, int need, int want,
-                            int *got, loff_t endoff, int *check_max, int *err)
+                            loff_t endoff, int *got, struct page **pinned_page,
+                            int *check_max, int *err)
 {
         struct inode *inode = &ci->vfs_inode;
         int ret = 0;
-        int have, implemented;
+        int have, implemented, _got = 0;
         int file_wanted;
 
         dout("get_cap_refs %p need %s want %s\n", inode,
              ceph_cap_string(need), ceph_cap_string(want));
+again:
         spin_lock(&ci->i_ceph_lock);
 
         /* make sure file is actually open */
@@ -2075,7 +2077,7 @@ static int try_get_cap_refs(struct ceph_inode_info *ci, int need, int want,
                      ceph_cap_string(need), ceph_cap_string(file_wanted));
                 *err = -EBADF;
                 ret = 1;
-                goto out;
+                goto out_unlock;
         }
 
         /* finish pending truncate */
@@ -2095,7 +2097,7 @@ static int try_get_cap_refs(struct ceph_inode_info *ci, int need, int want,
                                 *check_max = 1;
                                 ret = 1;
                         }
-                        goto out;
+                        goto out_unlock;
                 }
                 /*
                  * If a sync write is in progress, we must wait, so that we
@@ -2103,7 +2105,7 @@ static int try_get_cap_refs(struct ceph_inode_info *ci, int need, int want,
                  */
                 if (__ceph_have_pending_cap_snap(ci)) {
                         dout("get_cap_refs %p cap_snap_pending\n", inode);
-                        goto out;
+                        goto out_unlock;
                 }
         }
 
@@ -2120,18 +2122,50 @@ static int try_get_cap_refs(struct ceph_inode_info *ci, int need, int want,
                      inode, ceph_cap_string(have), ceph_cap_string(not),
                      ceph_cap_string(revoking));
                 if ((revoking & not) == 0) {
-                        *got = need | (have & want);
-                        __take_cap_refs(ci, *got);
+                        _got = need | (have & want);
+                        __take_cap_refs(ci, _got);
                         ret = 1;
                 }
         } else {
                 dout("get_cap_refs %p have %s needed %s\n", inode,
                      ceph_cap_string(have), ceph_cap_string(need));
         }
-out:
+out_unlock:
         spin_unlock(&ci->i_ceph_lock);
+
+        if (ci->i_inline_version != CEPH_INLINE_NONE &&
+            (_got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) &&
+            i_size_read(inode) > 0) {
+                int ret1;
+                struct page *page = find_get_page(inode->i_mapping, 0);
+                if (page) {
+                        if (PageUptodate(page)) {
+                                *pinned_page = page;
+                                goto out;
+                        }
+                        page_cache_release(page);
+                }
+                /*
+                 * drop cap refs first because getattr while holding
+                 * caps refs can cause deadlock.
+                 */
+                ceph_put_cap_refs(ci, _got);
+                _got = 0;
+
+                /* getattr request will bring inline data into page cache */
+                ret1 = __ceph_do_getattr(inode, NULL,
+                                         CEPH_STAT_CAP_INLINE_DATA, true);
+                if (ret1 >= 0) {
+                        ret = 0;
+                        goto again;
+                }
+                *err = ret1;
+                ret = 1;
+        }
+out:
         dout("get_cap_refs %p ret %d got %s\n", inode,
-             ret, ceph_cap_string(*got));
+             ret, ceph_cap_string(_got));
+        *got = _got;
         return ret;
 }
 
@@ -2168,8 +2202,8 @@ static void check_max_size(struct inode *inode, loff_t endoff)
  * due to a small max_size, make sure we check_max_size (and possibly
  * ask the mds) so we don't get hung up indefinitely.
  */
-int ceph_get_caps(struct ceph_inode_info *ci, int need, int want, int *got,
-                  loff_t endoff)
+int ceph_get_caps(struct ceph_inode_info *ci, int need, int want,
+                  loff_t endoff, int *got, struct page **pinned_page)
 {
         int check_max, ret, err;
 
@@ -2179,8 +2213,8 @@ retry:
         check_max = 0;
         err = 0;
         ret = wait_event_interruptible(ci->i_cap_wq,
-                                       try_get_cap_refs(ci, need, want,
-                                                        got, endoff,
+                                       try_get_cap_refs(ci, need, want, endoff,
+                                                        got, pinned_page,
                                                         &check_max, &err));
         if (err)
                 ret = err;