ceph: include transaction id in ceph_msg_header (protocol change)

Many (most?) message types include a transaction id.  By including it in
the fixed size header, we always have it available even when we are unable
to allocate memory for the (larger, variable sized) message body.  This
will allow us to error out the appropriate request instead of (silently)
dropping the reply.

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

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index 93c1afe3f0b3ecb395af080bdef0b7f046eadf1a..847ae64346fec784d2c18594109b26b7f410adc7 100644 (file)
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -922,14 +922,14 @@ static int send_cap_msg(struct ceph_mds_session *session,
        if (IS_ERR(msg))
                return PTR_ERR(msg);
 
-       fc = msg->front.iov_base;
+       msg->hdr.tid = cpu_to_le64(flush_tid);
 
+       fc = msg->front.iov_base;
        memset(fc, 0, sizeof(*fc));
 
        fc->cap_id = cpu_to_le64(cid);
        fc->op = cpu_to_le32(op);
        fc->seq = cpu_to_le32(seq);
-       fc->client_tid = cpu_to_le64(flush_tid);
        fc->issue_seq = cpu_to_le32(issue_seq);
        fc->migrate_seq = cpu_to_le32(mseq);
        fc->caps = cpu_to_le32(caps);
@@ -2329,7 +2329,7 @@ restart:
  * Handle FLUSH_ACK from MDS, indicating that metadata we sent to the
  * MDS has been safely committed.
  */
-static void handle_cap_flush_ack(struct inode *inode,
+static void handle_cap_flush_ack(struct inode *inode, u64 flush_tid,
                                 struct ceph_mds_caps *m,
                                 struct ceph_mds_session *session,
                                 struct ceph_cap *cap)
@@ -2340,7 +2340,6 @@ static void handle_cap_flush_ack(struct inode *inode,
        unsigned seq = le32_to_cpu(m->seq);
        int dirty = le32_to_cpu(m->dirty);
        int cleaned = 0;
-       u64 flush_tid = le64_to_cpu(m->client_tid);
        int drop = 0;
        int i;
 
@@ -2396,13 +2395,12 @@ out:
  *
  * Caller hold s_mutex.
  */
-static void handle_cap_flushsnap_ack(struct inode *inode,
+static void handle_cap_flushsnap_ack(struct inode *inode, u64 flush_tid,
                                     struct ceph_mds_caps *m,
                                     struct ceph_mds_session *session)
 {
        struct ceph_inode_info *ci = ceph_inode(inode);
        u64 follows = le64_to_cpu(m->snap_follows);
-       u64 flush_tid = le64_to_cpu(m->client_tid);
        struct ceph_cap_snap *capsnap;
        int drop = 0;
 
@@ -2587,12 +2585,14 @@ void ceph_handle_caps(struct ceph_mds_session *session,
        struct ceph_vino vino;
        u64 cap_id;
        u64 size, max_size;
+       u64 tid;
        int check_caps = 0;
        int r;
 
        dout("handle_caps from mds%d\n", mds);
 
        /* decode */
+       tid = le64_to_cpu(msg->hdr.tid);
        if (msg->front.iov_len < sizeof(*h))
                goto bad;
        h = msg->front.iov_base;
@@ -2621,7 +2621,7 @@ void ceph_handle_caps(struct ceph_mds_session *session,
        /* these will work even if we don't have a cap yet */
        switch (op) {
        case CEPH_CAP_OP_FLUSHSNAP_ACK:
-               handle_cap_flushsnap_ack(inode, h, session);
+               handle_cap_flushsnap_ack(inode, tid, h, session);
                goto done;
 
        case CEPH_CAP_OP_EXPORT:
@@ -2662,7 +2662,7 @@ void ceph_handle_caps(struct ceph_mds_session *session,
                break;
 
        case CEPH_CAP_OP_FLUSH_ACK:
-               handle_cap_flush_ack(inode, h, session, cap);
+               handle_cap_flush_ack(inode, tid, h, session, cap);
                break;
 
        case CEPH_CAP_OP_TRUNC:

diff --git a/fs/ceph/ceph_fs.h b/fs/ceph/ceph_fs.h
index e2fd0247827e8eaf1336b5a5846e8385e6e51f85..e87dfa6ec8e5286af1b532ae344557254a00e9d9 100644 (file)
--- a/fs/ceph/ceph_fs.h
+++ b/fs/ceph/ceph_fs.h
@@ -35,7 +35,7 @@
  * internal cluster protocols separately from the public,
  * client-facing protocol.
  */
-#define CEPH_OSD_PROTOCOL     7 /* cluster internal */
+#define CEPH_OSD_PROTOCOL     8 /* cluster internal */
 #define CEPH_MDS_PROTOCOL     9 /* cluster internal */
 #define CEPH_MON_PROTOCOL     5 /* cluster internal */
 #define CEPH_OSDC_PROTOCOL   22 /* server/client */
@@ -136,7 +136,6 @@ struct ceph_mon_request_header {
 struct ceph_mon_statfs {
        struct ceph_mon_request_header monhdr;
        struct ceph_fsid fsid;
-       __le64 tid;
 } __attribute__ ((packed));
 
 struct ceph_statfs {
@@ -146,7 +145,6 @@ struct ceph_statfs {
 
 struct ceph_mon_statfs_reply {
        struct ceph_fsid fsid;
-       __le64 tid;
        __le64 version;
        struct ceph_statfs st;
 } __attribute__ ((packed));
@@ -333,7 +331,7 @@ union ceph_mds_request_args {
 #define CEPH_MDS_FLAG_WANT_DENTRY   2  /* want dentry in reply */
 
 struct ceph_mds_request_head {
-       __le64 tid, oldest_client_tid;
+       __le64 oldest_client_tid;
        __le32 mdsmap_epoch;           /* on client */
        __le32 flags;                  /* CEPH_MDS_FLAG_* */
        __u8 num_retry, num_fwd;       /* count retry, fwd attempts */
@@ -356,7 +354,6 @@ struct ceph_mds_request_release {
 
 /* client reply */
 struct ceph_mds_reply_head {
-       __le64 tid;
        __le32 op;
        __le32 result;
        __le32 mdsmap_epoch;
@@ -542,7 +539,6 @@ struct ceph_mds_caps {
        __le32 migrate_seq;
        __le64 snap_follows;
        __le32 snap_trace_len;
-       __le64 client_tid;          /* for FLUSH(SNAP) -> FLUSH(SNAP)_ACK */
 
        /* authlock */
        __le32 uid, gid, mode;

diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 63ca3b1ad45fc69feb0c87d1b7598c9264239466..ec884e2845dbc21d50b55bf297a8a13857ceaa10 100644 (file)
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -1339,6 +1339,8 @@ static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
        if (IS_ERR(msg))
                goto out_free2;
 
+       msg->hdr.tid = cpu_to_le64(req->r_tid);
+
        head = msg->front.iov_base;
        p = msg->front.iov_base + sizeof(*head);
        end = msg->front.iov_base + msg->front.iov_len;
@@ -1431,7 +1433,6 @@ static int __prepare_send_request(struct ceph_mds_client *mdsc,
        req->r_request = msg;
 
        rhead = msg->front.iov_base;
-       rhead->tid = cpu_to_le64(req->r_tid);
        rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
        if (req->r_got_unsafe)
                flags |= CEPH_MDS_FLAG_REPLAY;
@@ -1664,7 +1665,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
        }
 
        /* get request, session */
-       tid = le64_to_cpu(head->tid);
+       tid = le64_to_cpu(msg->hdr.tid);
        mutex_lock(&mdsc->mutex);
        req = __lookup_request(mdsc, tid);
        if (!req) {

diff --git a/fs/ceph/mon_client.c b/fs/ceph/mon_client.c
index 775a9c029c514d18b64d87dbd5855a6cfbefef8b..bb94006fc686c366b3ed2f01df298044477bed57 100644 (file)
--- a/fs/ceph/mon_client.c
+++ b/fs/ceph/mon_client.c
@@ -349,7 +349,7 @@ static void handle_statfs_reply(struct ceph_mon_client *monc,
 
        if (msg->front.iov_len != sizeof(*reply))
                goto bad;
-       tid = le64_to_cpu(reply->tid);
+       tid = le64_to_cpu(msg->hdr.tid);
        dout("handle_statfs_reply %p tid %llu\n", msg, tid);
 
        mutex_lock(&monc->mutex);
@@ -382,12 +382,12 @@ static int send_statfs(struct ceph_mon_client *monc,
        if (IS_ERR(msg))
                return PTR_ERR(msg);
        req->request = msg;
+       msg->hdr.tid = cpu_to_le64(req->tid);
        h = msg->front.iov_base;
        h->monhdr.have_version = 0;
        h->monhdr.session_mon = cpu_to_le16(-1);
        h->monhdr.session_mon_tid = 0;
        h->fsid = monc->monmap->fsid;
-       h->tid = cpu_to_le64(req->tid);
        ceph_con_send(monc->con, msg);
        return 0;
 }

diff --git a/fs/ceph/msgr.h b/fs/ceph/msgr.h
index c758e8f8f71bf69e26a51e1d21a3fcf41126218b..e46d8b806dea7b5712562f2ae0b26c9bb2fddcac 100644 (file)
--- a/fs/ceph/msgr.h
+++ b/fs/ceph/msgr.h
@@ -21,7 +21,7 @@
  * whenever the wire protocol changes.  try to keep this string length
  * constant.
  */
-#define CEPH_BANNER "ceph v024"
+#define CEPH_BANNER "ceph v025"
 #define CEPH_BANNER_MAX_LEN 30
 
 
@@ -132,6 +132,7 @@ struct ceph_msg_connect_reply {
  */
 struct ceph_msg_header {
        __le64 seq;       /* message seq# for this session */
+       __le64 tid;       /* transaction id */
        __le16 type;      /* message type */
        __le16 priority;  /* priority.  higher value == higher priority */
        __le16 version;   /* version of message encoding */

diff --git a/fs/ceph/osd_client.c b/fs/ceph/osd_client.c
index 374f0013956cb7522f7e361ca8be65694a208fa0..a0aac436d5d4431dd1f79b9e7fd8a11326d1b48e 100644 (file)
--- a/fs/ceph/osd_client.c
+++ b/fs/ceph/osd_client.c
@@ -439,11 +439,9 @@ static struct ceph_osd *__lookup_osd(struct ceph_osd_client *osdc, int o)
 static void register_request(struct ceph_osd_client *osdc,
                             struct ceph_osd_request *req)
 {
-       struct ceph_osd_request_head *head = req->r_request->front.iov_base;
-
        mutex_lock(&osdc->request_mutex);
        req->r_tid = ++osdc->last_tid;
-       head->tid = cpu_to_le64(req->r_tid);
+       req->r_request->hdr.tid = cpu_to_le64(req->r_tid);
 
        dout("register_request %p tid %lld\n", req, req->r_tid);
        __insert_request(osdc, req);
@@ -702,9 +700,9 @@ static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg,
        u64 tid;
        int numops, object_len, flags;
 
+       tid = le64_to_cpu(msg->hdr.tid);
        if (msg->front.iov_len < sizeof(*rhead))
                goto bad;
-       tid = le64_to_cpu(rhead->tid);
        numops = le32_to_cpu(rhead->num_ops);
        object_len = le32_to_cpu(rhead->object_len);
        if (msg->front.iov_len != sizeof(*rhead) + object_len +
@@ -1002,7 +1000,6 @@ static int prepare_pages(struct ceph_connection *con, struct ceph_msg *m,
 {
        struct ceph_osd *osd = con->private;
        struct ceph_osd_client *osdc;
-       struct ceph_osd_reply_head *rhead = m->front.iov_base;
        struct ceph_osd_request *req;
        u64 tid;
        int ret = -1;
@@ -1016,7 +1013,7 @@ static int prepare_pages(struct ceph_connection *con, struct ceph_msg *m,
        if (unlikely(type != CEPH_MSG_OSD_OPREPLY))
                return -1;  /* hmm! */
 
-       tid = le64_to_cpu(rhead->tid);
+       tid = le64_to_cpu(m->hdr.tid);
        mutex_lock(&osdc->request_mutex);
        req = __lookup_request(osdc, tid);
        if (!req) {

diff --git a/fs/ceph/rados.h b/fs/ceph/rados.h
index 12bfb2f7c27577c746994f84d73f16877c07c9db..c5614d4ae34a6864d780e999d1e8214eef06287d 100644 (file)
--- a/fs/ceph/rados.h
+++ b/fs/ceph/rados.h
@@ -331,7 +331,6 @@ struct ceph_osd_op {
  * ceph_osd_op object operations.
  */
 struct ceph_osd_request_head {
-       __le64 tid;                        /* transaction id */
        __le32 client_inc;                 /* client incarnation */
        struct ceph_object_layout layout;  /* pgid */
        __le32 osdmap_epoch;               /* client's osdmap epoch */
@@ -352,7 +351,6 @@ struct ceph_osd_request_head {
 } __attribute__ ((packed));
 
 struct ceph_osd_reply_head {
-       __le64 tid;                       /* transaction id */
        __le32 client_inc;                /* client incarnation */
        __le32 flags;
        struct ceph_object_layout layout;