ceph: release cap on import if we don't have the inode

If we get an IMPORT that give us a cap, but we don't have the inode, queue
a release (and try to send it immediately) so that the MDS doesn't get
stuck waiting for us.

Signed-off-by: Sage Weil <sage@newdream.net>

1 files changed, 55 insertions, 35 deletions

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index da2a0e3cb200..7c692e746237 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -981,6 +981,46 @@ static int send_cap_msg(struct ceph_mds_session *session,
         return 0;
 }
 
+static void __queue_cap_release(struct ceph_mds_session *session,
+                                u64 ino, u64 cap_id, u32 migrate_seq,
+                                u32 issue_seq)
+{
+        struct ceph_msg *msg;
+        struct ceph_mds_cap_release *head;
+        struct ceph_mds_cap_item *item;
+
+        spin_lock(&session->s_cap_lock);
+        BUG_ON(!session->s_num_cap_releases);
+        msg = list_first_entry(&session->s_cap_releases,
+                               struct ceph_msg, list_head);
+
+        dout(" adding %llx release to mds%d msg %p (%d left)\n",
+             ino, session->s_mds, msg, session->s_num_cap_releases);
+
+        BUG_ON(msg->front.iov_len + sizeof(*item) > PAGE_CACHE_SIZE);
+        head = msg->front.iov_base;
+        head->num = cpu_to_le32(le32_to_cpu(head->num) + 1);
+        item = msg->front.iov_base + msg->front.iov_len;
+        item->ino = cpu_to_le64(ino);
+        item->cap_id = cpu_to_le64(cap_id);
+        item->migrate_seq = cpu_to_le32(migrate_seq);
+        item->seq = cpu_to_le32(issue_seq);
+
+        session->s_num_cap_releases--;
+
+        msg->front.iov_len += sizeof(*item);
+        if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) {
+                dout(" release msg %p full\n", msg);
+                list_move_tail(&msg->list_head, &session->s_cap_releases_done);
+        } else {
+                dout(" release msg %p at %d/%d (%d)\n", msg,
+                     (int)le32_to_cpu(head->num),
+                     (int)CEPH_CAPS_PER_RELEASE,
+                     (int)msg->front.iov_len);
+        }
+        spin_unlock(&session->s_cap_lock);
+}
+
 /*
  * Queue cap releases when an inode is dropped from our cache.  Since
  * inode is about to be destroyed, there is no need for i_lock.
@@ -994,41 +1034,9 @@ void ceph_queue_caps_release(struct inode *inode)
         while (p) {
                 struct ceph_cap *cap = rb_entry(p, struct ceph_cap, ci_node);
                 struct ceph_mds_session *session = cap->session;
-                struct ceph_msg *msg;
-                struct ceph_mds_cap_release *head;
-                struct ceph_mds_cap_item *item;
 
-                spin_lock(&session->s_cap_lock);
-                BUG_ON(!session->s_num_cap_releases);
-                msg = list_first_entry(&session->s_cap_releases,
-                                       struct ceph_msg, list_head);
-
-                dout(" adding %p release to mds%d msg %p (%d left)\n",
-                     inode, session->s_mds, msg, session->s_num_cap_releases);
-
-                BUG_ON(msg->front.iov_len + sizeof(*item) > PAGE_CACHE_SIZE);
-                head = msg->front.iov_base;
-                head->num = cpu_to_le32(le32_to_cpu(head->num) + 1);
-                item = msg->front.iov_base + msg->front.iov_len;
-                item->ino = cpu_to_le64(ceph_ino(inode));
-                item->cap_id = cpu_to_le64(cap->cap_id);
-                item->migrate_seq = cpu_to_le32(cap->mseq);
-                item->seq = cpu_to_le32(cap->issue_seq);
-
-                session->s_num_cap_releases--;
-
-                msg->front.iov_len += sizeof(*item);
-                if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) {
-                        dout(" release msg %p full\n", msg);
-                        list_move_tail(&msg->list_head,
-                                       &session->s_cap_releases_done);
-                } else {
-                        dout(" release msg %p at %d/%d (%d)\n", msg,
-                             (int)le32_to_cpu(head->num),
-                             (int)CEPH_CAPS_PER_RELEASE,
-                             (int)msg->front.iov_len);
-                }
-                spin_unlock(&session->s_cap_lock);
+                __queue_cap_release(session, ceph_ino(inode), cap->cap_id,
+                                    cap->mseq, cap->issue_seq);
                 p = rb_next(p);
                 __ceph_remove_cap(cap);
         }
@@ -2655,7 +2663,7 @@ void ceph_handle_caps(struct ceph_mds_session *session,
         struct ceph_mds_caps *h;
         int mds = session->s_mds;
         int op;
-        u32 seq;
+        u32 seq, mseq;
         struct ceph_vino vino;
         u64 cap_id;
         u64 size, max_size;
@@ -2675,6 +2683,7 @@ void ceph_handle_caps(struct ceph_mds_session *session,
         vino.snap = CEPH_NOSNAP;
         cap_id = le64_to_cpu(h->cap_id);
         seq = le32_to_cpu(h->seq);
+        mseq = le32_to_cpu(h->migrate_seq);
         size = le64_to_cpu(h->size);
         max_size = le64_to_cpu(h->max_size);
 
@@ -2689,6 +2698,17 @@ void ceph_handle_caps(struct ceph_mds_session *session,
              vino.snap, inode);
         if (!inode) {
                 dout(" i don't have ino %llx\n", vino.ino);
+
+                if (op == CEPH_CAP_OP_IMPORT)
+                        __queue_cap_release(session, vino.ino, cap_id,
+                                            mseq, seq);
+
+                /*
+                 * send any full release message to try to move things
+                 * along for the mds (who clearly thinks we still have this
+                 * cap).
+                 */
+                ceph_send_cap_releases(mdsc, session);
                 goto done;
         }