ceph: do not call __mark_dirty_inode under i_lock

[pandora-kernel.git] / fs / ceph / caps.c

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index 60d27bc..9fa0866 100644 (file)
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -765,7 +765,7 @@ int __ceph_caps_issued_mask(struct ceph_inode_info *ci, int mask, int touch)
                        if (touch) {
                                struct rb_node *q;
 
-                               /* touch this + preceeding caps */
+                               /* touch this + preceding caps */
                                __touch_cap(cap);
                                for (q = rb_first(&ci->i_caps); q != p;
                                     q = rb_next(q)) {
@@ -1331,10 +1331,11 @@ static void ceph_flush_snaps(struct ceph_inode_info *ci)
 }
 
 /*
- * Mark caps dirty.  If inode is newly dirty, add to the global dirty
- * list.
+ * Mark caps dirty.  If inode is newly dirty, return the dirty flags.
+ * Caller is then responsible for calling __mark_inode_dirty with the
+ * returned flags value.
  */
-void __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask)
+int __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask)
 {
        struct ceph_mds_client *mdsc =
                ceph_sb_to_client(ci->vfs_inode.i_sb)->mdsc;
@@ -1357,7 +1358,7 @@ void __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask)
                list_add(&ci->i_dirty_item, &mdsc->cap_dirty);
                spin_unlock(&mdsc->cap_dirty_lock);
                if (ci->i_flushing_caps == 0) {
-                       igrab(inode);
+                       ihold(inode);
                        dirty |= I_DIRTY_SYNC;
                }
        }
@@ -1365,9 +1366,8 @@ void __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask)
        if (((was | ci->i_flushing_caps) & CEPH_CAP_FILE_BUFFER) &&
            (mask & CEPH_CAP_FILE_BUFFER))
                dirty |= I_DIRTY_DATASYNC;
-       if (dirty)
-               __mark_inode_dirty(inode, dirty);
        __cap_delay_requeue(mdsc, ci);
+       return dirty;
 }
 
 /*
@@ -1560,9 +1560,10 @@ retry_locked:
                /* NOTE: no side-effects allowed, until we take s_mutex */
 
                revoking = cap->implemented & ~cap->issued;
-               if (revoking)
-                       dout(" mds%d revoking %s\n", cap->mds,
-                            ceph_cap_string(revoking));
+               dout(" mds%d cap %p issued %s implemented %s revoking %s\n",
+                    cap->mds, cap, ceph_cap_string(cap->issued),
+                    ceph_cap_string(cap->implemented),
+                    ceph_cap_string(revoking));
 
                if (cap == ci->i_auth_cap &&
                    (cap->issued & CEPH_CAP_FILE_WR)) {
@@ -1658,6 +1659,8 @@ ack:
 
                if (cap == ci->i_auth_cap && ci->i_dirty_caps)
                        flushing = __mark_caps_flushing(inode, session);
+               else
+                       flushing = 0;
 
                mds = cap->mds;  /* remember mds, so we don't repeat */
                sent++;
@@ -1940,6 +1943,35 @@ void ceph_kick_flushing_caps(struct ceph_mds_client *mdsc,
        }
 }
 
+static void kick_flushing_inode_caps(struct ceph_mds_client *mdsc,
+                                    struct ceph_mds_session *session,
+                                    struct inode *inode)
+{
+       struct ceph_inode_info *ci = ceph_inode(inode);
+       struct ceph_cap *cap;
+       int delayed = 0;
+
+       spin_lock(&inode->i_lock);
+       cap = ci->i_auth_cap;
+       dout("kick_flushing_inode_caps %p flushing %s flush_seq %lld\n", inode,
+            ceph_cap_string(ci->i_flushing_caps), ci->i_cap_flush_seq);
+       __ceph_flush_snaps(ci, &session, 1);
+       if (ci->i_flushing_caps) {
+               delayed = __send_cap(mdsc, cap, CEPH_CAP_OP_FLUSH,
+                                    __ceph_caps_used(ci),
+                                    __ceph_caps_wanted(ci),
+                                    cap->issued | cap->implemented,
+                                    ci->i_flushing_caps, NULL);
+               if (delayed) {
+                       spin_lock(&inode->i_lock);
+                       __cap_delay_requeue(mdsc, ci);
+                       spin_unlock(&inode->i_lock);
+               }
+       } else {
+               spin_unlock(&inode->i_lock);
+       }
+}
+
 
 /*
  * Take references to capabilities we hold, so that we don't release
@@ -1959,7 +1991,7 @@ static void __take_cap_refs(struct ceph_inode_info *ci, int got)
                ci->i_wr_ref++;
        if (got & CEPH_CAP_FILE_BUFFER) {
                if (ci->i_wrbuffer_ref == 0)
-                       igrab(&ci->vfs_inode);
+                       ihold(&ci->vfs_inode);
                ci->i_wrbuffer_ref++;
                dout("__take_cap_refs %p wrbuffer %d -> %d (?)\n",
                     &ci->vfs_inode, ci->i_wrbuffer_ref-1, ci->i_wrbuffer_ref);
@@ -2687,7 +2719,7 @@ static void handle_cap_import(struct ceph_mds_client *mdsc,
        ceph_add_cap(inode, session, cap_id, -1,
                     issued, wanted, seq, mseq, realmino, CEPH_CAP_FLAG_AUTH,
                     NULL /* no caps context */);
-       try_flush_caps(inode, session, NULL);
+       kick_flushing_inode_caps(mdsc, session, inode);
        up_read(&mdsc->snap_rwsem);
 
        /* make sure we re-request max_size, if necessary */
@@ -2785,8 +2817,7 @@ void ceph_handle_caps(struct ceph_mds_session *session,
        case CEPH_CAP_OP_IMPORT:
                handle_cap_import(mdsc, inode, h, session,
                                  snaptrace, snaptrace_len);
-               ceph_check_caps(ceph_inode(inode), CHECK_CAPS_NODELAY,
-                               session);
+               ceph_check_caps(ceph_inode(inode), 0, session);
                goto done_unlocked;
        }