ceph: perform lazy reads when file mode and caps permit

[mv-sheeva.git] / fs / ceph / caps.c

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index 619b61655ee5dd4496ca4ec6374849c09f892817..b28915d5f40476f72cf05c9881031ec70a05941a 100644 (file)
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -244,8 +244,14 @@ static struct ceph_cap *get_cap(struct ceph_cap_reservation *ctx)
        struct ceph_cap *cap = NULL;
 
        /* temporary, until we do something about cap import/export */
-       if (!ctx)
-               return kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
+       if (!ctx) {
+               cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
+               if (cap) {
+                       caps_use_count++;
+                       caps_total_count++;
+               }
+               return cap;
+       }
 
        spin_lock(&caps_list_lock);
        dout("get_cap ctx=%p (%d) %d = %d used + %d resv + %d avail\n",
@@ -331,8 +337,7 @@ static struct ceph_cap *__get_cap_for_mds(struct ceph_inode_info *ci, int mds)
 }
 
 /*
- * Return id of any MDS with a cap, preferably FILE_WR|WRBUFFER|EXCL, else
- * -1.
+ * Return id of any MDS with a cap, preferably FILE_WR|BUFFER|EXCL, else -1.
  */
 static int __ceph_get_cap_mds(struct ceph_inode_info *ci, u32 *mseq)
 {
@@ -340,7 +345,7 @@ static int __ceph_get_cap_mds(struct ceph_inode_info *ci, u32 *mseq)
        int mds = -1;
        struct rb_node *p;
 
-       /* prefer mds with WR|WRBUFFER|EXCL caps */
+       /* prefer mds with WR|BUFFER|EXCL caps */
        for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
                cap = rb_entry(p, struct ceph_cap, ci_node);
                mds = cap->mds;
@@ -477,8 +482,8 @@ static void __check_cap_issue(struct ceph_inode_info *ci, struct ceph_cap *cap,
         * Each time we receive FILE_CACHE anew, we increment
         * i_rdcache_gen.
         */
-       if ((issued & CEPH_CAP_FILE_CACHE) &&
-           (had & CEPH_CAP_FILE_CACHE) == 0)
+       if ((issued & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) &&
+           (had & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0)
                ci->i_rdcache_gen++;
 
        /*
@@ -621,7 +626,7 @@ retry:
        if (fmode >= 0)
                __ceph_get_fmode(ci, fmode);
        spin_unlock(&inode->i_lock);
-       wake_up(&ci->i_cap_wq);
+       wake_up_all(&ci->i_cap_wq);
        return 0;
 }
 
@@ -825,7 +830,7 @@ int __ceph_caps_file_wanted(struct ceph_inode_info *ci)
 {
        int want = 0;
        int mode;
-       for (mode = 0; mode < 4; mode++)
+       for (mode = 0; mode < CEPH_FILE_MODE_NUM; mode++)
                if (ci->i_nr_by_mode[mode])
                        want |= ceph_caps_for_mode(mode);
        return want;
@@ -1175,7 +1180,7 @@ static int __send_cap(struct ceph_mds_client *mdsc, struct ceph_cap *cap,
        }
 
        if (wake)
-               wake_up(&ci->i_cap_wq);
+               wake_up_all(&ci->i_cap_wq);
 
        return delayed;
 }
@@ -1504,11 +1509,13 @@ retry_locked:
            ci->i_wrbuffer_ref == 0 &&               /* no dirty pages... */
            ci->i_rdcache_gen &&                     /* may have cached pages */
            (file_wanted == 0 ||                     /* no open files */
-            (revoking & CEPH_CAP_FILE_CACHE)) &&     /*  or revoking cache */
+            (revoking & (CEPH_CAP_FILE_CACHE|
+                         CEPH_CAP_FILE_LAZYIO))) && /*  or revoking cache */
            !tried_invalidate) {
                dout("check_caps trying to invalidate on %p\n", inode);
                if (try_nonblocking_invalidate(inode) < 0) {
-                       if (revoking & CEPH_CAP_FILE_CACHE) {
+                       if (revoking & (CEPH_CAP_FILE_CACHE|
+                                       CEPH_CAP_FILE_LAZYIO)) {
                                dout("check_caps queuing invalidate\n");
                                queue_invalidate = 1;
                                ci->i_rdcache_revoking = ci->i_rdcache_gen;
@@ -2147,7 +2154,7 @@ void ceph_put_cap_refs(struct ceph_inode_info *ci, int had)
        else if (flushsnaps)
                ceph_flush_snaps(ci);
        if (wake)
-               wake_up(&ci->i_cap_wq);
+               wake_up_all(&ci->i_cap_wq);
        if (put)
                iput(inode);
 }
@@ -2223,7 +2230,7 @@ void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
                iput(inode);
        } else if (complete_capsnap) {
                ceph_flush_snaps(ci);
-               wake_up(&ci->i_cap_wq);
+               wake_up_all(&ci->i_cap_wq);
        }
        if (drop_capsnap)
                iput(inode);
@@ -2271,7 +2278,8 @@ static void handle_cap_grant(struct inode *inode, struct ceph_mds_caps *grant,
         * try to invalidate (once).  (If there are dirty buffers, we
         * will invalidate _after_ writeback.)
         */
-       if (((cap->issued & ~newcaps) & CEPH_CAP_FILE_CACHE) &&
+       if (((cap->issued & ~newcaps) & (CEPH_CAP_FILE_CACHE|
+                                        CEPH_CAP_FILE_LAZYIO)) &&
            !ci->i_wrbuffer_ref) {
                if (try_nonblocking_invalidate(inode) == 0) {
                        revoked_rdcache = 1;
@@ -2369,7 +2377,8 @@ static void handle_cap_grant(struct inode *inode, struct ceph_mds_caps *grant,
                        writeback = 1; /* will delay ack */
                else if (dirty & ~newcaps)
                        check_caps = 1;  /* initiate writeback in check_caps */
-               else if (((used & ~newcaps) & CEPH_CAP_FILE_CACHE) == 0 ||
+               else if (((used & ~newcaps) & (CEPH_CAP_FILE_CACHE|
+                                              CEPH_CAP_FILE_LAZYIO)) == 0 ||
                           revoked_rdcache)
                        check_caps = 2;     /* send revoke ack in check_caps */
                cap->issued = newcaps;
@@ -2399,7 +2408,7 @@ static void handle_cap_grant(struct inode *inode, struct ceph_mds_caps *grant,
        if (queue_invalidate)
                ceph_queue_invalidate(inode);
        if (wake)
-               wake_up(&ci->i_cap_wq);
+               wake_up_all(&ci->i_cap_wq);
 
        if (check_caps == 1)
                ceph_check_caps(ci, CHECK_CAPS_NODELAY|CHECK_CAPS_AUTHONLY,
@@ -2454,7 +2463,7 @@ static void handle_cap_flush_ack(struct inode *inode, u64 flush_tid,
                                         struct ceph_inode_info,
                                         i_flushing_item)->vfs_inode);
                mdsc->num_cap_flushing--;
-               wake_up(&mdsc->cap_flushing_wq);
+               wake_up_all(&mdsc->cap_flushing_wq);
                dout(" inode %p now !flushing\n", inode);
 
                if (ci->i_dirty_caps == 0) {
@@ -2466,7 +2475,7 @@ static void handle_cap_flush_ack(struct inode *inode, u64 flush_tid,
                }
        }
        spin_unlock(&mdsc->cap_dirty_lock);
-       wake_up(&ci->i_cap_wq);
+       wake_up_all(&ci->i_cap_wq);
 
 out:
        spin_unlock(&inode->i_lock);
@@ -2886,18 +2895,19 @@ int ceph_encode_inode_release(void **p, struct inode *inode,
        struct ceph_inode_info *ci = ceph_inode(inode);
        struct ceph_cap *cap;
        struct ceph_mds_request_release *rel = *p;
+       int used, dirty;
        int ret = 0;
-       int used = 0;
 
        spin_lock(&inode->i_lock);
        used = __ceph_caps_used(ci);
+       dirty = __ceph_caps_dirty(ci);
 
-       dout("encode_inode_release %p mds%d used %s drop %s unless %s\n", inode,
-            mds, ceph_cap_string(used), ceph_cap_string(drop),
+       dout("encode_inode_release %p mds%d used|dirty %s drop %s unless %s\n",
+            inode, mds, ceph_cap_string(used|dirty), ceph_cap_string(drop),
             ceph_cap_string(unless));
 
-       /* only drop unused caps */
-       drop &= ~used;
+       /* only drop unused, clean caps */
+       drop &= ~(used | dirty);
 
        cap = __get_cap_for_mds(ci, mds);
        if (cap && __cap_is_valid(cap)) {
@@ -2977,6 +2987,7 @@ int ceph_encode_dentry_release(void **p, struct dentry *dentry,
                memcpy(*p, dentry->d_name.name, dentry->d_name.len);
                *p += dentry->d_name.len;
                rel->dname_seq = cpu_to_le32(di->lease_seq);
+               __ceph_mdsc_drop_dentry_lease(dentry);
        }
        spin_unlock(&dentry->d_lock);
        return ret;