kclient: cap import/export, some cleanups

diff --git a/src/kernel/inode.c b/src/kernel/inode.c
index 8c9df49c5bd34a433e94f63e28c86ea057414a11..1b6504a2ba59215bfc8d68134c826925136f3005 100644 (file)
--- a/src/kernel/inode.c
+++ b/src/kernel/inode.c
@@ -566,7 +566,7 @@ int ceph_dentry_lease_valid(struct dentry *dentry)
        struct ceph_dentry_info *di;
        struct ceph_mds_session *session;
        int valid = 0;
-       u64 gen;
+       u32 gen;
        unsigned long ttl;
 
        spin_lock(&dentry->d_lock);
@@ -942,12 +942,14 @@ int ceph_get_cap_mds(struct inode *inode)
 }
 
 /*
- * caller shoudl hold session s_mutex.
+ * caller should hold session s_mutex.
+ *
+ * @fmode can be negative, in which case it is ignored.
  */
-struct ceph_inode_cap *ceph_add_cap(struct inode *inode,
-                                   struct ceph_mds_session *session,
-                                   int fmode,
-                                   u32 issued, u32 seq)
+int ceph_add_cap(struct inode *inode,
+                struct ceph_mds_session *session,
+                int fmode, unsigned issued,
+                unsigned seq, unsigned mseq)
 {
        int mds = session->s_mds;
        struct ceph_inode_info *ci = ceph_inode(inode);
@@ -972,7 +974,7 @@ struct ceph_inode_cap *ceph_add_cap(struct inode *inode,
                                spin_unlock(&inode->i_lock);
                                new_cap = kmalloc(sizeof(*cap), GFP_NOFS);
                                if (new_cap == 0)
-                                       return ERR_PTR(-ENOMEM);
+                                       return -ENOMEM;
                                spin_lock(&inode->i_lock);
                        }
                }
@@ -980,7 +982,6 @@ struct ceph_inode_cap *ceph_add_cap(struct inode *inode,
                is_new = 1;    /* grab inode later */
                cap->issued = cap->implemented = 0;
                cap->mds = mds;
-               cap->seq = 0;
                cap->flags = 0;
 
                cap->ci = ci;
@@ -990,6 +991,13 @@ struct ceph_inode_cap *ceph_add_cap(struct inode *inode,
                cap->session = session;
                list_add(&cap->session_caps, &session->s_caps);
                session->s_nr_caps++;
+
+               /* clear out old exporting info? */
+               if (ci->i_cap_exporting_mds == mds) {
+                       ci->i_cap_exporting_issued = 0;
+                       ci->i_cap_exporting_mseq = 0;
+                       ci->i_cap_exporting_mds = -1;
+               }
        }
 
        dout(10, "add_cap inode %p (%llx) got cap %xh now %xh seq %d from %d\n",
@@ -997,12 +1005,14 @@ struct ceph_inode_cap *ceph_add_cap(struct inode *inode,
        cap->issued |= issued;
        cap->implemented |= issued;
        cap->seq = seq;
+       cap->mseq = mseq;
        cap->gen = session->s_cap_gen;
-       __ceph_get_fmode(ci, fmode);
+       if (fmode >= 0)
+               __ceph_get_fmode(ci, fmode);
        spin_unlock(&inode->i_lock);
        if (is_new)
                igrab(inode);
-       return cap;
+       return 0;
 }
 
 int __ceph_caps_issued(struct ceph_inode_info *ci)
@@ -1010,7 +1020,7 @@ int __ceph_caps_issued(struct ceph_inode_info *ci)
        int have = 0;
        struct ceph_inode_cap *cap;
        struct list_head *p;
-       u64 gen;
+       u32 gen;
        unsigned long ttl;
 
        list_for_each(p, &ci->i_caps) {
@@ -1023,7 +1033,7 @@ int __ceph_caps_issued(struct ceph_inode_info *ci)
 
                if (cap->gen < gen || time_after_eq(jiffies, ttl)) {
                        dout(30, "__ceph_caps_issued %p cap %p issued %d "
-                            "but STALE (gen %llu vs %llu)\n", &ci->vfs_inode,
+                            "but STALE (gen %u vs %u)\n", &ci->vfs_inode,
                             cap, cap->issued, cap->gen, gen);
                        continue;
                }
@@ -1406,8 +1416,9 @@ void __ceph_do_pending_vmtruncate(struct inode *inode)
                dout(10, "__do_pending_vmtruncate %p nothing to do\n", inode);
 }
 
-int ceph_handle_cap_trunc(struct inode *inode, struct ceph_mds_file_caps *trunc,
-                         struct ceph_mds_session *session)
+void ceph_handle_cap_trunc(struct inode *inode,
+                          struct ceph_mds_file_caps *trunc,
+                          struct ceph_mds_session *session)
 {
        struct ceph_inode_info *ci = ceph_inode(inode);
        int mds = session->s_mds;
@@ -1447,9 +1458,73 @@ int ceph_handle_cap_trunc(struct inode *inode, struct ceph_mds_file_caps *trunc,
        if (queue_trunc)
                queue_work(ceph_client(inode->i_sb)->trunc_wq,
                           &ci->i_vmtruncate_work);
-       return 0;
 }
 
+void ceph_handle_cap_export(struct inode *inode, struct ceph_mds_file_caps *ex,
+                           struct ceph_mds_session *session)
+{
+       struct ceph_inode_info *ci = ceph_inode(inode);
+       int mds = session->s_mds;
+       unsigned mseq = le32_to_cpu(ex->migrate_seq);
+       struct ceph_inode_cap *cap = 0, *t;
+       struct list_head *p;
+
+       dout(10, "handle_cap_export inode %p ci %p mds%d mseq %d\n",
+            inode, ci, mds, mseq);
+
+       spin_lock(&inode->i_lock);
+
+       /* make sure we haven't seen a higher mseq */
+       list_for_each(p, &ci->i_caps) {
+               t = list_entry(p, struct ceph_inode_cap, ci_caps);
+               if (t->mseq > mseq) {
+                       dout(10, " higher mseq on cap from mds%d\n",
+                            t->session->s_mds);
+                       goto out;
+               }
+               if (t->session->s_mds == mds)
+                       cap = t;
+       }
+
+       if (cap) {
+               /* make note, and remove */
+               ci->i_cap_exporting_mds = mds;
+               ci->i_cap_exporting_mseq = mseq;
+               ci->i_cap_exporting_issued = cap->issued;
+               __ceph_remove_cap(cap);
+       }
+
+out:
+       spin_unlock(&inode->i_lock);
+}
+
+void ceph_handle_cap_import(struct inode *inode, struct ceph_mds_file_caps *im,
+                           struct ceph_mds_session *session)
+{
+       struct ceph_inode_info *ci = ceph_inode(inode);
+       int mds = session->s_mds;
+       unsigned issued = le32_to_cpu(im->caps);
+       unsigned seq = le32_to_cpu(im->seq);
+       unsigned mseq = le32_to_cpu(im->migrate_seq);
+
+       if (ci->i_cap_exporting_mds >= 0 &&
+           ci->i_cap_exporting_mseq < mseq) {
+               dout(10, "handle_cap_import inode %p ci %p mds%d mseq %d"
+                    " - cleared exporting from mds%d\n",
+                    inode, ci, mds, mseq,
+                    ci->i_cap_exporting_mds);
+               ci->i_cap_exporting_issued = 0;
+               ci->i_cap_exporting_mseq = 0;
+               ci->i_cap_exporting_mds = -1;
+       } else {
+               dout(10, "handle_cap_import inode %p ci %p mds%d mseq %d\n",
+                    inode, ci, mds, mseq);
+       }
+
+       ceph_add_cap(inode, session, -1, issued, seq, mseq);
+}
+
+
 static void __take_cap_refs(struct ceph_inode_info *ci, int got)
 {
        if (got & CEPH_CAP_RD)

diff --git a/src/kernel/mds_client.c b/src/kernel/mds_client.c
index 75b984c5184b904bf7e1d31253b5aeba296920b4..f79d610b1d2ebfa033c4d85d3ba178b620591688 100644 (file)
--- a/src/kernel/mds_client.c
+++ b/src/kernel/mds_client.c
@@ -400,7 +400,6 @@ static struct ceph_mds_request *new_request(struct ceph_msg *msg)
        req->r_old_dentry = 0;
        req->r_expects_cap = 0;
        req->r_fmode = 0;
-       req->r_cap = 0;
        req->r_session = 0;
        req->r_fwd_session = 0;
        req->r_attempts = 0;
@@ -1105,10 +1104,10 @@ void ceph_mdsc_handle_reply(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
        struct ceph_mds_request *req;
        struct ceph_mds_reply_head *head = msg->front.iov_base;
        struct ceph_mds_reply_info *rinfo;
-       __u64 tid;
+       u64 tid;
        int err, result;
        int mds;
-       __u32 cap, capseq;
+       u32 cap, capseq, mseq;
 
        /* extract tid */
        if (msg->front.iov_len < sizeof(*head)) {
@@ -1160,15 +1159,13 @@ void ceph_mdsc_handle_reply(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
                if (req->r_expects_cap && req->r_last_inode) {
                        cap = le32_to_cpu(rinfo->head->file_caps);
                        capseq = le32_to_cpu(rinfo->head->file_caps_seq);
-                       req->r_cap = ceph_add_cap(req->r_last_inode,
-                                                 req->r_session,
-                                                 req->r_fmode,
-                                                 cap, capseq);
-                       if (IS_ERR(req->r_cap)) {
-                               err = PTR_ERR(req->r_cap);
-                               req->r_cap = 0;
+                       mseq = le32_to_cpu(rinfo->head->file_caps_mseq);
+                       err = ceph_add_cap(req->r_last_inode,
+                                          req->r_session,
+                                          req->r_fmode,
+                                          cap, capseq, mseq);
+                       if (err)
                                goto done;
-                       }
                }
 
                /* readdir result? */
@@ -1556,8 +1553,11 @@ void ceph_mdsc_handle_filecaps(struct ceph_mds_client *mdsc,
                break;
 
        case CEPH_CAP_OP_EXPORT:
+               ceph_handle_cap_export(inode, h, session);
+               break;
+
        case CEPH_CAP_OP_IMPORT:
-               dout(10, "cap export/import -- IMPLEMENT ME\n");
+               ceph_handle_cap_import(inode, h, session);
                break;
        }
 

diff --git a/src/kernel/mds_client.h b/src/kernel/mds_client.h
index dfeca70a875dd9ddf57b88d6e12562b90d608bef..d7d12e8f5a43e3333a3a660e1f64c56b1f6f9e11 100644 (file)
--- a/src/kernel/mds_client.h
+++ b/src/kernel/mds_client.h
@@ -10,6 +10,7 @@
 #include "mdsmap.h"
 
 struct ceph_client;
+struct ceph_inode_cap;
 
 /*
  * for mds reply parsing
@@ -60,7 +61,7 @@ struct ceph_mds_session {
        u64               s_seq;      /* incoming msg seq # */
        struct mutex      s_mutex;
        spinlock_t        s_cap_lock; /* protects s_cap_gen, s_cap_ttl */
-       u64               s_cap_gen;  /* inc each time we get mds stale msg */
+       u32               s_cap_gen;  /* inc each time we get mds stale msg */
        unsigned long     s_cap_ttl, s_renew_requested;
        struct list_head  s_caps;
        struct list_head  s_inode_leases, s_dentry_leases;
@@ -97,7 +98,6 @@ struct ceph_mds_request {
        int                     r_expects_cap;
        int                     r_fmode;  /* if expecting cap */
        unsigned long           r_from_time;
-       struct ceph_inode_cap  *r_cap;
        struct ceph_mds_session *r_session;
        struct ceph_mds_session *r_fwd_session;  /* forwarded from */
 

diff --git a/src/kernel/super.h b/src/kernel/super.h
index c8bbeb6abb4d74a662fd7be62fbfc8f26be2f34c..6cff52b61c989b9d84374c203847fc9597116779 100644 (file)
--- a/src/kernel/super.h
+++ b/src/kernel/super.h
@@ -153,7 +153,7 @@ struct ceph_inode_cap {
        int mds;    /* -1 if not used */
        int issued;       /* latest, from the mds */
        int implemented;  /* what we've implemneted (for tracking revocation) */
-       u64 seq, gen;
+       u32 seq, mseq, gen;
        int flags;  /* stale, etc.? */
        struct ceph_inode_info *ci;
        struct list_head ci_caps;       /* per-ci caplist */
@@ -204,7 +204,7 @@ struct ceph_inode_info {
        int i_lease_mask;
        struct ceph_mds_session *i_lease_session;
        long unsigned i_lease_ttl;  /* jiffies */
-       u64 i_lease_gen;
+       u32 i_lease_gen;
        struct list_head i_lease_item; /* mds session list */
 
        struct rb_root i_fragtree;
@@ -217,6 +217,9 @@ struct ceph_inode_info {
        wait_queue_head_t i_cap_wq;
        unsigned long i_hold_caps_until; /* jiffies */
        struct list_head i_cap_delay_list;
+       int i_cap_exporting_mds;
+       unsigned i_cap_exporting_mseq;
+       unsigned i_cap_exporting_issued;
 
        int i_nr_by_mode[CEPH_FILE_MODE_NUM];
        loff_t i_max_size;      /* size authorized by mds */
@@ -269,7 +272,7 @@ extern __u32 ceph_choose_frag(struct ceph_inode_info *ci, u32 v,
 struct ceph_dentry_info {
        struct dentry *dentry;
        struct ceph_mds_session *lease_session;
-       u64 lease_gen;
+       u32 lease_gen;
        struct list_head lease_item; /* mds session list */
 };
 
@@ -440,10 +443,10 @@ extern void ceph_update_dentry_lease(struct dentry *dentry,
 extern int ceph_inode_lease_valid(struct inode *inode, int mask);
 extern int ceph_dentry_lease_valid(struct dentry *dentry);
 
-extern struct ceph_inode_cap *ceph_add_cap(struct inode *inode,
-                                          struct ceph_mds_session *session,
-                                          int fmode,
-                                          u32 cap, u32 seq);
+extern int ceph_add_cap(struct inode *inode,
+                       struct ceph_mds_session *session,
+                       int fmode, unsigned issued,
+                       unsigned cap, unsigned seq);
 extern void __ceph_remove_cap(struct ceph_inode_cap *cap);
 extern void ceph_remove_cap(struct ceph_inode_cap *cap);
 extern void ceph_remove_all_caps(struct ceph_inode_info *ci);
@@ -451,9 +454,15 @@ extern int ceph_get_cap_mds(struct inode *inode);
 extern int ceph_handle_cap_grant(struct inode *inode,
                                 struct ceph_mds_file_caps *grant,
                                 struct ceph_mds_session *session);
-extern int ceph_handle_cap_trunc(struct inode *inode,
-                                struct ceph_mds_file_caps *grant,
-                                struct ceph_mds_session *session);
+extern void ceph_handle_cap_trunc(struct inode *inode,
+                                 struct ceph_mds_file_caps *trunc,
+                                 struct ceph_mds_session *session);
+extern void ceph_handle_cap_export(struct inode *inode,
+                                  struct ceph_mds_file_caps *ex,
+                                  struct ceph_mds_session *session);
+extern void ceph_handle_cap_import(struct inode *inode,
+                                  struct ceph_mds_file_caps *im,
+                                  struct ceph_mds_session *session);
 extern int ceph_get_cap_refs(struct ceph_inode_info *ci, int need, int want, int *got, loff_t offset);
 extern void ceph_take_cap_refs(struct ceph_inode_info *ci, int got);
 extern void ceph_put_cap_refs(struct ceph_inode_info *ci, int had);