ceph: Check for created flag in response from mds

The mds now sends back a created inode if the create request
performed the create.  If the file already existed, no inode is
returned in the reply.  This allows ceph to set the created flag
in atomic_open so that permissions are properly checked in the case
that the file wasn't created by the create call to the mds.

To ensure compability with previous kernels, a feature for sending
back the inode in the create reply was added, so that the mds will
only send back the inode if the client indicates it supports the
feature.

Signed-off-by: Sam Lang <sam.lang@inktank.com>
Reviewed-by: Sage Weil <sage@inktank.com>

1 files changed, 31 insertions, 2 deletions

diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 9165eb8309eb..d95842036c8b 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -233,6 +233,30 @@ bad:
 }
 
 /*
+ * parse create results
+ */
+static int parse_reply_info_create(void **p, void *end,
+                                  struct ceph_mds_reply_info_parsed *info,
+                                  int features)
+{
+        if (features & CEPH_FEATURE_REPLY_CREATE_INODE) {
+                if (*p == end) {
+                        info->has_create_ino = false;
+                } else {
+                        info->has_create_ino = true;
+                        info->ino = ceph_decode_64(p);
+                }
+        }
+
+        if (unlikely(*p != end))
+                goto bad;
+        return 0;
+
+bad:
+        return -EIO;
+}
+
+/*
  * parse extra results
  */
 static int parse_reply_info_extra(void **p, void *end,
@@ -241,8 +265,12 @@ static int parse_reply_info_extra(void **p, void *end,
 {
         if (info->head->op == CEPH_MDS_OP_GETFILELOCK)
                 return parse_reply_info_filelock(p, end, info, features);
-        else
+        else if (info->head->op == CEPH_MDS_OP_READDIR)
                 return parse_reply_info_dir(p, end, info, features);
+        else if (info->head->op == CEPH_MDS_OP_CREATE)
+                return parse_reply_info_create(p, end, info, features);
+        else
+                return -EIO;
 }
 
 /*
@@ -2170,7 +2198,8 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
         mutex_lock(&req->r_fill_mutex);
         err = ceph_fill_trace(mdsc->fsc->sb, req, req->r_session);
         if (err == 0) {
-                if (result == 0 && req->r_op != CEPH_MDS_OP_GETFILELOCK &&
+                if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR ||
+                                    req->r_op == CEPH_MDS_OP_LSSNAP) &&
                     rinfo->dir_nr)
                         ceph_readdir_prepopulate(req, req->r_session);
                 ceph_unreserve_caps(mdsc, &req->r_caps_reservation);