ceph: handle SESSION_FORCE_RO message

[deliverable/linux.git] / fs / ceph / mds_client.c

diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index a92d3f5c6c12fa3ae0b7308c9fcd379ac821cc67..c6c33b411a2f7f7679af20038bcecf5cad5ed4a6 100644 (file)
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -89,6 +89,16 @@ static int parse_reply_info_in(void **p, void *end,
        ceph_decode_need(p, end, info->xattr_len, bad);
        info->xattr_data = *p;
        *p += info->xattr_len;
+
+       if (features & CEPH_FEATURE_MDS_INLINE_DATA) {
+               ceph_decode_64_safe(p, end, info->inline_version, bad);
+               ceph_decode_32_safe(p, end, info->inline_len, bad);
+               ceph_decode_need(p, end, info->inline_len, bad);
+               info->inline_data = *p;
+               *p += info->inline_len;
+       } else
+               info->inline_version = CEPH_INLINE_NONE;
+
        return 0;
 bad:
        return err;
@@ -524,8 +534,7 @@ void ceph_mdsc_release_request(struct kref *kref)
        }
        if (req->r_locked_dir)
                ceph_put_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
-       if (req->r_target_inode)
-               iput(req->r_target_inode);
+       iput(req->r_target_inode);
        if (req->r_dentry)
                dput(req->r_dentry);
        if (req->r_old_dentry)
@@ -861,8 +870,11 @@ static struct ceph_msg *create_session_open_msg(struct ceph_mds_client *mdsc, u6
        /*
         * Serialize client metadata into waiting buffer space, using
         * the format that userspace expects for map<string, string>
+        *
+        * ClientSession messages with metadata are v2
         */
-       msg->hdr.version = 2;  /* ClientSession messages with metadata are v2 */
+       msg->hdr.version = cpu_to_le16(2);
+       msg->hdr.compat_version = cpu_to_le16(1);
 
        /* The write pointer, following the session_head structure */
        p = msg->front.iov_base + sizeof(*h);
@@ -1066,8 +1078,7 @@ out:
        session->s_cap_iterator = NULL;
        spin_unlock(&session->s_cap_lock);
 
-       if (last_inode)
-               iput(last_inode);
+       iput(last_inode);
        if (old_cap)
                ceph_put_cap(session->s_mdsc, old_cap);
 
@@ -1874,7 +1885,7 @@ static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
                goto out_free2;
        }
 
-       msg->hdr.version = 2;
+       msg->hdr.version = cpu_to_le16(2);
        msg->hdr.tid = cpu_to_le64(req->r_tid);
 
        head = msg->front.iov_base;
@@ -2208,6 +2219,8 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
                        &req->r_completion, req->r_timeout);
                if (err == 0)
                        err = -EIO;
+       } else if (req->r_wait_for_completion) {
+               err = req->r_wait_for_completion(mdsc, req);
        } else {
                err = wait_for_completion_killable(&req->r_completion);
        }
@@ -2567,6 +2580,14 @@ static void handle_session(struct ceph_mds_session *session,
                send_flushmsg_ack(mdsc, session, seq);
                break;
 
+       case CEPH_SESSION_FORCE_RO:
+               dout("force_session_readonly %p\n", session);
+               spin_lock(&session->s_cap_lock);
+               session->s_readonly = true;
+               spin_unlock(&session->s_cap_lock);
+               wake_up_session_caps(session, 0);
+               break;
+
        default:
                pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds);
                WARN_ON(1);
@@ -2778,6 +2799,8 @@ static void send_mds_reconnect(struct ceph_mds_client *mdsc,
        spin_unlock(&session->s_gen_ttl_lock);
 
        spin_lock(&session->s_cap_lock);
+       /* don't know if session is readonly */
+       session->s_readonly = 0;
        /*
         * notify __ceph_remove_cap() that we are composing cap reconnect.
         * If a cap get released before being added to the cap reconnect,
@@ -3744,6 +3767,20 @@ static struct ceph_msg *mds_alloc_msg(struct ceph_connection *con,
        return msg;
 }
 
+static int sign_message(struct ceph_connection *con, struct ceph_msg *msg)
+{
+       struct ceph_mds_session *s = con->private;
+       struct ceph_auth_handshake *auth = &s->s_auth;
+       return ceph_auth_sign_message(auth, msg);
+}
+
+static int check_message_signature(struct ceph_connection *con, struct ceph_msg *msg)
+{
+       struct ceph_mds_session *s = con->private;
+       struct ceph_auth_handshake *auth = &s->s_auth;
+       return ceph_auth_check_message_signature(auth, msg);
+}
+
 static const struct ceph_connection_operations mds_con_ops = {
        .get = con_get,
        .put = con_put,
@@ -3753,6 +3790,8 @@ static const struct ceph_connection_operations mds_con_ops = {
        .invalidate_authorizer = invalidate_authorizer,
        .peer_reset = peer_reset,
        .alloc_msg = mds_alloc_msg,
+       .sign_message = sign_message,
+       .check_message_signature = check_message_signature,
 };
 
 /* eof */