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>

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index da2a0e3..7c692e7 100644 (file)
--- 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;
        }
 

diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 29b4485..d28b6a9 100644 (file)
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -1176,8 +1176,8 @@ static int check_cap_flush(struct ceph_mds_client *mdsc, u64 want_flush_seq)
 /*
  * called under s_mutex
  */
-static void send_cap_releases(struct ceph_mds_client *mdsc,
-                      struct ceph_mds_session *session)
+void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
+                           struct ceph_mds_session *session)
 {
        struct ceph_msg *msg;
 
@@ -2693,7 +2693,7 @@ static void delayed_work(struct work_struct *work)
                add_cap_releases(mdsc, s, -1);
                if (s->s_state == CEPH_MDS_SESSION_OPEN ||
                    s->s_state == CEPH_MDS_SESSION_HUNG)
-                       send_cap_releases(mdsc, s);
+                       ceph_send_cap_releases(mdsc, s);
                mutex_unlock(&s->s_mutex);
                ceph_put_mds_session(s);
 

diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index d9936c4..e43752b 100644 (file)
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -322,6 +322,9 @@ static inline void ceph_mdsc_put_request(struct ceph_mds_request *req)
        kref_put(&req->r_kref, ceph_mdsc_release_request);
 }
 
+extern void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
+                                  struct ceph_mds_session *session);
+
 extern void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc);
 
 extern char *ceph_mdsc_build_path(struct dentry *dentry, int *plen, u64 *base,