Ceph » Linux kernel client

The reason is that ceph_reserve_caps may not reserve enough caps under high memory pressure, but it saved the needed caps number that expected to be reserved. When getting caps, crash would happen due to number mismatch.

I try to fix it in this way that only saves the caps number that successfully obtained the reservation in ceph_reserve_caps. When getting caps, we give the unreserved caps another chance to allocate memory again.

    ceph: try to allocate memory for unreserved caps

    ceph_reserve_caps may not reserve enough caps under high memory
    pressure, but it saved the needed caps number that expected to
    be reserved. When getting caps, crash would happen due to number
    mismatch.

    Now we only saves the caps number that successfully obtained the
    reservation in ceph_reserve_caps. When getting caps, we give the
    unreserved caps another chance to allocate memory again.

    Signed-off-by: Zhi Zhang <zhang.david2011@gmail.com>
---
 fs/ceph/caps.c  | 12 +++++++++---
 fs/ceph/inode.c |  9 ++++++++-
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index a14b2c9..9d4e27d 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -200,7 +200,7 @@ void ceph_reserve_caps(struct ceph_mds_client *mdsc,
                                         mdsc->caps_avail_count);
        spin_unlock(&mdsc->caps_list_lock);

-       ctx->count = need;
+       ctx->count = have + alloc;
        dout("reserve caps ctx=%p %d = %d used + %d resv + %d avail\n",
             ctx, mdsc->caps_total_count, mdsc->caps_use_count,
             mdsc->caps_reserve_count, mdsc->caps_avail_count);
@@ -232,14 +232,20 @@ struct ceph_cap *ceph_get_cap(struct ceph_mds_client *mdsc,
 {
        struct ceph_cap *cap = NULL;

-       /* temporary, until we do something about cap import/export */
-       if (!ctx) {
+       /* 
+        * 1. temporary, until we do something about cap import/export 
+        * 2. try to allocate memory for new cap,
+        *    if there is no reservation for it
+        */
+       if (!ctx || ctx->count == 0) {
                cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
                if (cap) {
                        spin_lock(&mdsc->caps_list_lock);
                        mdsc->caps_use_count++;
                        mdsc->caps_total_count++;
                        spin_unlock(&mdsc->caps_list_lock);
+               } else {
+                       pr_warn("get_cap ENOMEM on allocating new cap\n");
                }
                return cap;
        }
diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index ab81652..c542ac1 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -744,8 +744,15 @@ static int fill_inode(struct inode *inode, struct page *locked_page,
             ci->i_version);

        /* prealloc new cap struct */
-       if (info->cap.caps && ceph_snap(inode) == CEPH_NOSNAP)
+       if (info->cap.caps && ceph_snap(inode) == CEPH_NOSNAP) {
                new_cap = ceph_get_cap(mdsc, caps_reservation);
+               if (!new_cap) {
+                        pr_err("fill_inode %p ino %llx.%llx ENOMEM new_cap\n",
+                                inode, ceph_vinop(inode));
+                        err = -ENOMEM;
+                        goto out;
+                }
+       }

        /*
         * prealloc xattr data, if it looks like we'll need it.  only

Hi Yan,

Could you pls help to review this change and let me know if something wrong? Thanks.

failing to get caps in fill_inode() is bad. If ceph_reserve_caps() fails to reserve caps, it should trim some caps and try again. If it still can't reserve caps, it should make caller return -ENOMEM

According to your suggestion, shall we call trim_caps to trim as much caps as possible, or just call trim_caps to trim the expected number that equals to the needed caps number?

And if we do this, the original trim caps behavior will be changed. Originally trim caps was like passive mode which was triggered by MDS. Now it will become both active and passive mode, right?

    ceph: try to allocate enough memory for reserved caps

    ceph_reserve_caps may not reserve enough caps under high memory
    pressure, but it saved the needed caps number that expected to
    be reserved. When getting caps, crash would happen due to number
    mismatch.

    Now we will try to trim more caps when failing to allocate memory
    for caps need to be reserved, then try again. If still failing to
    allocate memory, return ENOMEM.

    Signed-off-by: Zhi Zhang <zhang.david2011@gmail.com>
---
 fs/ceph/caps.c       | 60 ++++++++++++++++++++++++++++++++++++++++++++++++++++--------
 fs/ceph/mds_client.c | 22 ++++++++++++++++------
 fs/ceph/mds_client.h |  3 +++
 fs/ceph/super.h      |  2 +-
 4 files changed, 72 insertions(+), 15 deletions(-)

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index a14b2c9..b1e5d94 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -154,13 +154,15 @@ void ceph_adjust_min_caps(struct ceph_mds_client *mdsc, int delta)
        spin_unlock(&mdsc->caps_list_lock);
 }

-void ceph_reserve_caps(struct ceph_mds_client *mdsc,
+int ceph_reserve_caps(struct ceph_mds_client *mdsc,
                      struct ceph_cap_reservation *ctx, int need)
 {
-       int i;
+       int i, j;
        struct ceph_cap *cap;
        int have;
        int alloc = 0;
+       bool trimmed = false;
+       struct ceph_mds_session *s;
        LIST_HEAD(newcaps);

        dout("reserve caps ctx=%p need=%d\n", ctx, need);
@@ -179,16 +181,40 @@ void ceph_reserve_caps(struct ceph_mds_client *mdsc,
        spin_unlock(&mdsc->caps_list_lock);

        for (i = have; i < need; i++) {
+retry:
                cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
-               if (!cap)
-                       break;
+               if (!cap) {
+                       if (!trimmed) {
+                               mutex_lock(&mdsc->mutex);
+                               for (j = 0; j < mdsc->max_sessions; j++) {
+                                       s = __ceph_lookup_mds_session(mdsc, j);
+                                       if (!s)
+                                               continue;
+                                       mutex_unlock(&mdsc->mutex);     
+
+                                       // trim as much caps as possible to 
+                                       // free more memory
+                                       mutex_lock(&s->s_mutex);
+                                       trim_caps(mdsc, s, 0);
+                                       mutex_unlock(&s->s_mutex);
+                                       
+                                       ceph_put_mds_session(s);
+                                       mutex_lock(&mdsc->mutex);
+                               }
+                               mutex_unlock(&mdsc->mutex);
+                               trimmed = true;
+                               goto retry;
+                       } else {
+                               pr_warn("reserve caps ctx=%p ENOMEM " 
+                                       "need=%d got=%d\n",
+                               ctx, need, have + alloc);
+                               goto out_nomem;
+                       }
+               }
                list_add(&cap->caps_item, &newcaps);
                alloc++;
        }
-       /* we didn't manage to reserve as much as we needed */
-       if (have + alloc != need)
-               pr_warn("reserve caps ctx=%p ENOMEM need=%d got=%d\n",
-                       ctx, need, have + alloc);
+       BUG_ON(have + alloc != need);

        spin_lock(&mdsc->caps_list_lock);
        mdsc->caps_total_count += alloc;
@@ -204,6 +230,24 @@ void ceph_reserve_caps(struct ceph_mds_client *mdsc,
        dout("reserve caps ctx=%p %d = %d used + %d resv + %d avail\n",
             ctx, mdsc->caps_total_count, mdsc->caps_use_count,
             mdsc->caps_reserve_count, mdsc->caps_avail_count);
+       return 0;
+
+out_nomem:
+       while (!list_empty(&newcaps)) {
+               cap = list_first_entry(&newcaps,
+                                       struct ceph_cap, caps_item);
+               list_del(&cap->caps_item);
+               kmem_cache_free(ceph_cap_cachep, cap);
+       }       
+
+       spin_lock(&mdsc->caps_list_lock);
+       mdsc->caps_avail_count += have;
+       mdsc->caps_reserve_count -= have;
+       BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
+                                        mdsc->caps_reserve_count +
+                                        mdsc->caps_avail_count);
+       spin_unlock(&mdsc->caps_list_lock);
+       return -ENOMEM;
 }

 int ceph_unreserve_caps(struct ceph_mds_client *mdsc,
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 1b46825..edea9a0 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -604,10 +604,20 @@ static void __register_request(struct ceph_mds_client *mdsc,
                               struct ceph_mds_request *req,
                               struct inode *dir)
 {
+       int ret = 0;
+
        req->r_tid = ++mdsc->last_tid;
-       if (req->r_num_caps)
-               ceph_reserve_caps(mdsc, &req->r_caps_reservation,
-                                 req->r_num_caps);
+       if (req->r_num_caps) {
+               ret = ceph_reserve_caps(mdsc, &req->r_caps_reservation,
+                                       req->r_num_caps);
+               if (ret) {
+                       pr_err("__register_request %p " 
+                              "failed to reserve caps: %d\n", req, ret);
+                       // set req->r_err to fail early from __do_request
+                       req->r_err = ret;
+                       return;
+               }
+       }
        dout("__register_request %p tid %lld\n", req, req->r_tid);
        ceph_mdsc_get_request(req);
        insert_request(&mdsc->request_tree, req);
@@ -1545,9 +1555,9 @@ static int trim_caps_cb(struct inode *inode, struct ceph_cap *cap, void *arg)
 /*
  * Trim session cap count down to some max number.
  */
-static int trim_caps(struct ceph_mds_client *mdsc,
-                    struct ceph_mds_session *session,
-                    int max_caps)
+int trim_caps(struct ceph_mds_client *mdsc,
+              struct ceph_mds_session *session,
+              int max_caps)
 {
        int trim_caps = session->s_nr_caps - max_caps;

diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 837ac4b..63b19e4 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -444,4 +444,7 @@ extern void ceph_mdsc_handle_fsmap(struct ceph_mds_client *mdsc,
 extern void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,
                                          struct ceph_mds_session *session);

+extern int trim_caps(struct ceph_mds_client *mdsc,
+                     struct ceph_mds_session *session,
+                     int max_caps);
 #endif
diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 2beeec0..e5fee4f 100644
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -648,7 +648,7 @@ static inline int __ceph_caps_wanted(struct ceph_inode_info *ci)
 extern void ceph_caps_init(struct ceph_mds_client *mdsc);
 extern void ceph_caps_finalize(struct ceph_mds_client *mdsc);
 extern void ceph_adjust_min_caps(struct ceph_mds_client *mdsc, int delta);
-extern void ceph_reserve_caps(struct ceph_mds_client *mdsc,
+extern int ceph_reserve_caps(struct ceph_mds_client *mdsc,
                             struct ceph_cap_reservation *ctx, int need);
 extern int ceph_unreserve_caps(struct ceph_mds_client *mdsc,
                               struct ceph_cap_reservation *ctx);

Hi Yan,

Here are the new changes according to your suggestion. Could you pls help to review again?
This change is testing on our env, so far so good.

BTW, I still prefer the first change, because it is simple and clear. And as you mentioned "failing to get caps in fill_inode() is bad", I saw there was already other ENOMEM failure in fill_inode() and cleanup logic is good, so I think it is fine to add this one too. What do you think?

Thanks.

I found several issues

1. caller of ceph_reserve_caps() should have already locked mdsc->mutex.

2. trim_caps(..., max_caps = 0) is too aggressive, maybe set it to s->s_nr_caps - (need - i);

3. should rename trim_caps() to ceph_trim_caps()

The ENOMEM cleanup code in fill_inode() is not good. Dropping caps silently can causes mds malfunction.

Ok, will update them soon. Thanks.

    ceph: try to allocate enough memory for reserved caps

    ceph_reserve_caps may not reserve enough caps under high memory
    pressure, but it saved the needed caps number that expected to
    be reserved. When getting caps, crash would happen due to number
    mismatch.

    Now we will try to trim more caps when failing to allocate memory
    for caps need to be reserved, then try again. If still failing to
    allocate memory, return ENOMEM.

    Signed-off-by: Zhi Zhang <zhang.david2011@gmail.com>
---
 fs/ceph/caps.c       | 62 ++++++++++++++++++++++++++++++++++++++++++++++++++++++--------
 fs/ceph/mds_client.c | 22 ++++++++++++++++------
 fs/ceph/mds_client.h |  3 +++
 fs/ceph/super.h      |  2 +-
 4 files changed, 74 insertions(+), 15 deletions(-)

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index a14b2c9..d00ca7b 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -154,13 +154,19 @@ void ceph_adjust_min_caps(struct ceph_mds_client *mdsc, int delta)
        spin_unlock(&mdsc->caps_list_lock);
 }

-void ceph_reserve_caps(struct ceph_mds_client *mdsc,
+/*
+ * Called under mdsc->mutex.
+ */
+int ceph_reserve_caps(struct ceph_mds_client *mdsc,
                      struct ceph_cap_reservation *ctx, int need)
 {
-       int i;
+       int i, j;
        struct ceph_cap *cap;
        int have;
        int alloc = 0;
+       int max_caps;
+       bool trimmed = false;
+       struct ceph_mds_session *s;
        LIST_HEAD(newcaps);

        dout("reserve caps ctx=%p need=%d\n", ctx, need);
@@ -179,16 +185,38 @@ void ceph_reserve_caps(struct ceph_mds_client *mdsc,
        spin_unlock(&mdsc->caps_list_lock);

        for (i = have; i < need; i++) {
+retry:
                cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
-               if (!cap)
-                       break;
+               if (!cap) {
+                       if (!trimmed) {
+                               for (j = 0; j < mdsc->max_sessions; j++) {
+                                       s = __ceph_lookup_mds_session(mdsc, j);
+                                       if (!s)
+                                               continue;
+                                       mutex_unlock(&mdsc->mutex);     
+
+                                       // trim needed caps to free memory
+                                       mutex_lock(&s->s_mutex);
+                                       max_caps = s->s_nr_caps - (need - i);
+                                       ceph_trim_caps(mdsc, s, max_caps);
+                                       mutex_unlock(&s->s_mutex);
+                                       
+                                       ceph_put_mds_session(s);
+                                       mutex_lock(&mdsc->mutex);
+                               }
+                               trimmed = true;
+                               goto retry;
+                       } else {
+                               pr_warn("reserve caps ctx=%p ENOMEM " 
+                                       "need=%d got=%d\n",
+                               ctx, need, have + alloc);
+                               goto out_nomem;
+                       }
+               }
                list_add(&cap->caps_item, &newcaps);
                alloc++;
        }
-       /* we didn't manage to reserve as much as we needed */
-       if (have + alloc != need)
-               pr_warn("reserve caps ctx=%p ENOMEM need=%d got=%d\n",
-                       ctx, need, have + alloc);
+       BUG_ON(have + alloc != need);

        spin_lock(&mdsc->caps_list_lock);
        mdsc->caps_total_count += alloc;
@@ -204,6 +232,24 @@ void ceph_reserve_caps(struct ceph_mds_client *mdsc,
        dout("reserve caps ctx=%p %d = %d used + %d resv + %d avail\n",
             ctx, mdsc->caps_total_count, mdsc->caps_use_count,
             mdsc->caps_reserve_count, mdsc->caps_avail_count);
+       return 0;
+
+out_nomem:
+       while (!list_empty(&newcaps)) {
+               cap = list_first_entry(&newcaps,
+                                       struct ceph_cap, caps_item);
+               list_del(&cap->caps_item);
+               kmem_cache_free(ceph_cap_cachep, cap);
+       }       
+
+       spin_lock(&mdsc->caps_list_lock);
+       mdsc->caps_avail_count += have;
+       mdsc->caps_reserve_count -= have;
+       BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
+                                        mdsc->caps_reserve_count +
+                                        mdsc->caps_avail_count);
+       spin_unlock(&mdsc->caps_list_lock);
+       return -ENOMEM;
 }

 int ceph_unreserve_caps(struct ceph_mds_client *mdsc,
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 1b46825..6885669 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -604,10 +604,20 @@ static void __register_request(struct ceph_mds_client *mdsc,
                               struct ceph_mds_request *req,
                               struct inode *dir)
 {
+       int ret = 0;
+
        req->r_tid = ++mdsc->last_tid;
-       if (req->r_num_caps)
-               ceph_reserve_caps(mdsc, &req->r_caps_reservation,
-                                 req->r_num_caps);
+       if (req->r_num_caps) {
+               ret = ceph_reserve_caps(mdsc, &req->r_caps_reservation,
+                                       req->r_num_caps);
+               if (ret) {
+                       pr_err("__register_request %p " 
+                              "failed to reserve caps: %d\n", req, ret);
+                       // set req->r_err to fail early from __do_request
+                       req->r_err = ret;
+                       return;
+               }
+       }
        dout("__register_request %p tid %lld\n", req, req->r_tid);
        ceph_mdsc_get_request(req);
        insert_request(&mdsc->request_tree, req);
@@ -1545,9 +1555,9 @@ static int trim_caps_cb(struct inode *inode, struct ceph_cap *cap, void *arg)
 /*
  * Trim session cap count down to some max number.
  */
-static int trim_caps(struct ceph_mds_client *mdsc,
-                    struct ceph_mds_session *session,
-                    int max_caps)
+int ceph_trim_caps(struct ceph_mds_client *mdsc,
+                   struct ceph_mds_session *session,
+                   int max_caps)
 {
        int trim_caps = session->s_nr_caps - max_caps;

diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 837ac4b..f9160f4 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -444,4 +444,7 @@ extern void ceph_mdsc_handle_fsmap(struct ceph_mds_client *mdsc,
 extern void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,
                                          struct ceph_mds_session *session);

+extern int ceph_trim_caps(struct ceph_mds_client *mdsc,
+                          struct ceph_mds_session *session,
+                          int max_caps);
 #endif
diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 2beeec0..e5fee4f 100644
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -648,7 +648,7 @@ static inline int __ceph_caps_wanted(struct ceph_inode_info *ci)
 extern void ceph_caps_init(struct ceph_mds_client *mdsc);
 extern void ceph_caps_finalize(struct ceph_mds_client *mdsc);
 extern void ceph_adjust_min_caps(struct ceph_mds_client *mdsc, int delta);
-extern void ceph_reserve_caps(struct ceph_mds_client *mdsc,
+extern int ceph_reserve_caps(struct ceph_mds_client *mdsc,
                             struct ceph_cap_reservation *ctx, int need);
 extern int ceph_unreserve_caps(struct ceph_mds_client *mdsc,
                               struct ceph_cap_reservation *ctx);

Hi Yan,

Pls help to review the new changes again, thanks.

this patch does not compile. you should test your patch before submitting it

Sorry, I will check it right now. This patch is already testing on our env. I think it could be some mistake when I merge it into Linux master branch.

    ceph: try to allocate enough memory for reserved caps

    ceph_reserve_caps may not reserve enough caps under high memory
    pressure, but it saved the needed caps number that expected to
    be reserved. When getting caps, crash would happen due to number
    mismatch.

    Now we will try to trim more caps when failing to allocate memory
    for caps need to be reserved, then try again. If still failing to
    allocate memory, return ENOMEM.

    Signed-off-by: Zhi Zhang <zhang.david2011@gmail.com>
---
 fs/ceph/caps.c       | 62 ++++++++++++++++++++++++++++++++++++++++++++++++++++++--------
 fs/ceph/mds_client.c | 24 +++++++++++++++++-------
 fs/ceph/mds_client.h |  3 +++
 fs/ceph/super.h      |  2 +-
 4 files changed, 75 insertions(+), 16 deletions(-)

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index a14b2c9..6784f59 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -154,13 +154,19 @@ void ceph_adjust_min_caps(struct ceph_mds_client *mdsc, int delta)
        spin_unlock(&mdsc->caps_list_lock);
 }

-void ceph_reserve_caps(struct ceph_mds_client *mdsc,
+/*
+ * Called under mdsc->mutex.
+ */
+int ceph_reserve_caps(struct ceph_mds_client *mdsc,
                      struct ceph_cap_reservation *ctx, int need)
 {
-       int i;
+       int i, j;
        struct ceph_cap *cap;
        int have;
        int alloc = 0;
+       int max_caps;
+       bool trimmed = false;
+       struct ceph_mds_session *s;
        LIST_HEAD(newcaps);

        dout("reserve caps ctx=%p need=%d\n", ctx, need);
@@ -179,16 +185,38 @@ void ceph_reserve_caps(struct ceph_mds_client *mdsc,
        spin_unlock(&mdsc->caps_list_lock);

        for (i = have; i < need; i++) {
+retry:
                cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
-               if (!cap)
-                       break;
+               if (!cap) {
+                       if (!trimmed) {
+                               for (j = 0; j < mdsc->max_sessions; j++) {
+                                       s = __ceph_lookup_mds_session(mdsc, j);
+                                       if (!s)
+                                               continue;
+                                       mutex_unlock(&mdsc->mutex);     
+
+                                       // trim needed caps to free memory
+                                       mutex_lock(&s->s_mutex);
+                                       max_caps = s->s_nr_caps - (need - i);
+                                       ceph_trim_caps(mdsc, s, max_caps);
+                                       mutex_unlock(&s->s_mutex);
+                                       
+                                       ceph_put_mds_session(s);
+                                       mutex_lock(&mdsc->mutex);
+                               }
+                               trimmed = true;
+                               goto retry;
+                       } else {
+                               pr_warn("reserve caps ctx=%p ENOMEM " 
+                                       "need=%d got=%d\n", 
+                                       ctx, need, have + alloc);
+                               goto out_nomem;
+                       }
+               }
                list_add(&cap->caps_item, &newcaps);
                alloc++;
        }
-       /* we didn't manage to reserve as much as we needed */
-       if (have + alloc != need)
-               pr_warn("reserve caps ctx=%p ENOMEM need=%d got=%d\n",
-                       ctx, need, have + alloc);
+       BUG_ON(have + alloc != need);

        spin_lock(&mdsc->caps_list_lock);
        mdsc->caps_total_count += alloc;
@@ -204,6 +232,24 @@ void ceph_reserve_caps(struct ceph_mds_client *mdsc,
        dout("reserve caps ctx=%p %d = %d used + %d resv + %d avail\n",
             ctx, mdsc->caps_total_count, mdsc->caps_use_count,
             mdsc->caps_reserve_count, mdsc->caps_avail_count);
+       return 0;
+
+out_nomem:
+       while (!list_empty(&newcaps)) {
+               cap = list_first_entry(&newcaps,
+                                       struct ceph_cap, caps_item);
+               list_del(&cap->caps_item);
+               kmem_cache_free(ceph_cap_cachep, cap);
+       }       
+
+       spin_lock(&mdsc->caps_list_lock);
+       mdsc->caps_avail_count += have;
+       mdsc->caps_reserve_count -= have;
+       BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
+                                        mdsc->caps_reserve_count +
+                                        mdsc->caps_avail_count);
+       spin_unlock(&mdsc->caps_list_lock);
+       return -ENOMEM;
 }

 int ceph_unreserve_caps(struct ceph_mds_client *mdsc,
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 1b46825..abc0375 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -604,10 +604,20 @@ static void __register_request(struct ceph_mds_client *mdsc,
                               struct ceph_mds_request *req,
                               struct inode *dir)
 {
+       int ret = 0;
+
        req->r_tid = ++mdsc->last_tid;
-       if (req->r_num_caps)
-               ceph_reserve_caps(mdsc, &req->r_caps_reservation,
-                                 req->r_num_caps);
+       if (req->r_num_caps) {
+               ret = ceph_reserve_caps(mdsc, &req->r_caps_reservation,
+                                       req->r_num_caps);
+               if (ret) {
+                       pr_err("__register_request %p " 
+                              "failed to reserve caps: %d\n", req, ret);
+                       // set req->r_err to fail early from __do_request
+                       req->r_err = ret;
+                       return;
+               }
+       }
        dout("__register_request %p tid %lld\n", req, req->r_tid);
        ceph_mdsc_get_request(req);
        insert_request(&mdsc->request_tree, req);
@@ -1545,9 +1555,9 @@ static int trim_caps_cb(struct inode *inode, struct ceph_cap *cap, void *arg)
 /*
  * Trim session cap count down to some max number.
  */
-static int trim_caps(struct ceph_mds_client *mdsc,
-                    struct ceph_mds_session *session,
-                    int max_caps)
+int ceph_trim_caps(struct ceph_mds_client *mdsc,
+                   struct ceph_mds_session *session,
+                   int max_caps)
 {
        int trim_caps = session->s_nr_caps - max_caps;

@@ -2773,7 +2783,7 @@ static void handle_session(struct ceph_mds_session *session,
                break;

        case CEPH_SESSION_RECALL_STATE:
-               trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
+               ceph_trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
                break;

        case CEPH_SESSION_FLUSHMSG:
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 837ac4b..f9160f4 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -444,4 +444,7 @@ extern void ceph_mdsc_handle_fsmap(struct ceph_mds_client *mdsc,
 extern void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,
                                          struct ceph_mds_session *session);

+extern int ceph_trim_caps(struct ceph_mds_client *mdsc,
+                          struct ceph_mds_session *session,
+                          int max_caps);
 #endif
diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 2beeec0..e5fee4f 100644
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -648,7 +648,7 @@ static inline int __ceph_caps_wanted(struct ceph_inode_info *ci)
 extern void ceph_caps_init(struct ceph_mds_client *mdsc);
 extern void ceph_caps_finalize(struct ceph_mds_client *mdsc);
 extern void ceph_adjust_min_caps(struct ceph_mds_client *mdsc, int delta);
-extern void ceph_reserve_caps(struct ceph_mds_client *mdsc,
+extern int ceph_reserve_caps(struct ceph_mds_client *mdsc,
                             struct ceph_cap_reservation *ctx, int need);
 extern int ceph_unreserve_caps(struct ceph_mds_client *mdsc,
                               struct ceph_cap_reservation *ctx);

Pls review it again. Last wrong one is due to merging mistake.

Thanks.

LGTM

Hi Yan, Zhi

I think current fix for the issue still has a small nit.
When caps_avail_count is in a low level, most newly
trimmed caps will probably go into ->caps_list and can't
reuse properly.

Plese see below title in the ceph-devel maillist for detail.
[PATCH] ceph: optimizing cap reservation