Merge branch 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/sage/ceph-client

Pull Ceph updates from Sage Weil:
 "This has a mix of bug fixes and cleanups.

  Alex's patch fixes a rare race in RBD.  Ilya's patches fix an ENOENT
  check when a second rbd image is mapped and a couple memory leaks.
  Zheng fixes several issues with fragmented directories and multiple
  MDSs.  Josh fixes a spin/sleep issue, and Josh and Guangliang's
  patches fix setting and unsetting RBD images read-only.

  Naturally there are several other cleanups mixed in for good measure"

* 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/sage/ceph-client: (23 commits)
  rbd: only set disk to read-only once
  rbd: move calls that may sleep out of spin lock range
  rbd: add ioctl for rbd
  ceph: use truncate_pagecache() instead of truncate_inode_pages()
  ceph: include time stamp in every MDS request
  rbd: fix ida/idr memory leak
  rbd: use reference counts for image requests
  rbd: fix osd_request memory leak in __rbd_dev_header_watch_sync()
  rbd: make sure we have latest osdmap on 'rbd map'
  libceph: add ceph_monc_wait_osdmap()
  libceph: mon_get_version request infrastructure
  libceph: recognize poolop requests in debugfs
  ceph: refactor readpage_nounlock() to make the logic clearer
  mds: check cap ID when handling cap export message
  ceph: remember subtree root dirfrag's auth MDS
  ceph: introduce ceph_fill_fragtree()
  ceph: handle cap import atomically
  ceph: pre-allocate ceph_cap struct for ceph_add_cap()
  ceph: update inode fields according to issued caps
  rbd: replace IS_ERR and PTR_ERR with PTR_ERR_OR_ZERO
  ...

3 files changed, 153 insertions, 7 deletions

diff --git a/net/ceph/ceph_common.c b/net/ceph/ceph_common.c
index 67d7721d237e..1675021d8c12 100644
--- a/net/ceph/ceph_common.c
+++ b/net/ceph/ceph_common.c
@@ -72,6 +72,8 @@ const char *ceph_msg_type_name(int type)
         case CEPH_MSG_MON_SUBSCRIBE_ACK: return "mon_subscribe_ack";
         case CEPH_MSG_STATFS: return "statfs";
         case CEPH_MSG_STATFS_REPLY: return "statfs_reply";
+        case CEPH_MSG_MON_GET_VERSION: return "mon_get_version";
+        case CEPH_MSG_MON_GET_VERSION_REPLY: return "mon_get_version_reply";
         case CEPH_MSG_MDS_MAP: return "mds_map";
         case CEPH_MSG_CLIENT_SESSION: return "client_session";
         case CEPH_MSG_CLIENT_RECONNECT: return "client_reconnect";
diff --git a/net/ceph/debugfs.c b/net/ceph/debugfs.c
index 10421a4b76f8..d1a62c69a9f4 100644
--- a/net/ceph/debugfs.c
+++ b/net/ceph/debugfs.c
@@ -126,9 +126,13 @@ static int monc_show(struct seq_file *s, void *p)
                 req = rb_entry(rp, struct ceph_mon_generic_request, node);
                 op = le16_to_cpu(req->request->hdr.type);
                 if (op == CEPH_MSG_STATFS)
-                        seq_printf(s, "%lld statfs\n", req->tid);
+                        seq_printf(s, "%llu statfs\n", req->tid);
+                else if (op == CEPH_MSG_POOLOP)
+                        seq_printf(s, "%llu poolop\n", req->tid);
+                else if (op == CEPH_MSG_MON_GET_VERSION)
+                        seq_printf(s, "%llu mon_get_version", req->tid);
                 else
-                        seq_printf(s, "%lld unknown\n", req->tid);
+                        seq_printf(s, "%llu unknown\n", req->tid);
         }
 
         mutex_unlock(&monc->mutex);
diff --git a/net/ceph/mon_client.c b/net/ceph/mon_client.c
index 2ac9ef35110b..067d3af2eaf6 100644
--- a/net/ceph/mon_client.c
+++ b/net/ceph/mon_client.c
@@ -296,6 +296,33 @@ void ceph_monc_request_next_osdmap(struct ceph_mon_client *monc)
                 __send_subscribe(monc);
         mutex_unlock(&monc->mutex);
 }
+EXPORT_SYMBOL(ceph_monc_request_next_osdmap);
+
+int ceph_monc_wait_osdmap(struct ceph_mon_client *monc, u32 epoch,
+                          unsigned long timeout)
+{
+        unsigned long started = jiffies;
+        int ret;
+
+        mutex_lock(&monc->mutex);
+        while (monc->have_osdmap < epoch) {
+                mutex_unlock(&monc->mutex);
+
+                if (timeout != 0 && time_after_eq(jiffies, started + timeout))
+                        return -ETIMEDOUT;
+
+                ret = wait_event_interruptible_timeout(monc->client->auth_wq,
+                                         monc->have_osdmap >= epoch, timeout);
+                if (ret < 0)
+                        return ret;
+
+                mutex_lock(&monc->mutex);
+        }
+
+        mutex_unlock(&monc->mutex);
+        return 0;
+}
+EXPORT_SYMBOL(ceph_monc_wait_osdmap);
 
 /*
  *
@@ -477,14 +504,13 @@ static struct ceph_msg *get_generic_reply(struct ceph_connection *con,
         return m;
 }
 
-static int do_generic_request(struct ceph_mon_client *monc,
-                              struct ceph_mon_generic_request *req)
+static int __do_generic_request(struct ceph_mon_client *monc, u64 tid,
+                                struct ceph_mon_generic_request *req)
 {
         int err;
 
         /* register request */
-        mutex_lock(&monc->mutex);
-        req->tid = ++monc->last_tid;
+        req->tid = tid != 0 ? tid : ++monc->last_tid;
         req->request->hdr.tid = cpu_to_le64(req->tid);
         __insert_generic_request(monc, req);
         monc->num_generic_requests++;
@@ -496,13 +522,24 @@ static int do_generic_request(struct ceph_mon_client *monc,
         mutex_lock(&monc->mutex);
         rb_erase(&req->node, &monc->generic_request_tree);
         monc->num_generic_requests--;
-        mutex_unlock(&monc->mutex);
 
         if (!err)
                 err = req->result;
         return err;
 }
 
+static int do_generic_request(struct ceph_mon_client *monc,
+                              struct ceph_mon_generic_request *req)
+{
+        int err;
+
+        mutex_lock(&monc->mutex);
+        err = __do_generic_request(monc, 0, req);
+        mutex_unlock(&monc->mutex);
+
+        return err;
+}
+
 /*
  * statfs
  */
@@ -579,6 +616,96 @@ out:
 }
 EXPORT_SYMBOL(ceph_monc_do_statfs);
 
+static void handle_get_version_reply(struct ceph_mon_client *monc,
+                                     struct ceph_msg *msg)
+{
+        struct ceph_mon_generic_request *req;
+        u64 tid = le64_to_cpu(msg->hdr.tid);
+        void *p = msg->front.iov_base;
+        void *end = p + msg->front_alloc_len;
+        u64 handle;
+
+        dout("%s %p tid %llu\n", __func__, msg, tid);
+
+        ceph_decode_need(&p, end, 2*sizeof(u64), bad);
+        handle = ceph_decode_64(&p);
+        if (tid != 0 && tid != handle)
+                goto bad;
+
+        mutex_lock(&monc->mutex);
+        req = __lookup_generic_req(monc, handle);
+        if (req) {
+                *(u64 *)req->buf = ceph_decode_64(&p);
+                req->result = 0;
+                get_generic_request(req);
+        }
+        mutex_unlock(&monc->mutex);
+        if (req) {
+                complete_all(&req->completion);
+                put_generic_request(req);
+        }
+
+        return;
+bad:
+        pr_err("corrupt mon_get_version reply\n");
+        ceph_msg_dump(msg);
+}
+
+/*
+ * Send MMonGetVersion and wait for the reply.
+ *
+ * @what: one of "mdsmap", "osdmap" or "monmap"
+ */
+int ceph_monc_do_get_version(struct ceph_mon_client *monc, const char *what,
+                             u64 *newest)
+{
+        struct ceph_mon_generic_request *req;
+        void *p, *end;
+        u64 tid;
+        int err;
+
+        req = kzalloc(sizeof(*req), GFP_NOFS);
+        if (!req)
+                return -ENOMEM;
+
+        kref_init(&req->kref);
+        req->buf = newest;
+        req->buf_len = sizeof(*newest);
+        init_completion(&req->completion);
+
+        req->request = ceph_msg_new(CEPH_MSG_MON_GET_VERSION,
+                                    sizeof(u64) + sizeof(u32) + strlen(what),
+                                    GFP_NOFS, true);
+        if (!req->request) {
+                err = -ENOMEM;
+                goto out;
+        }
+
+        req->reply = ceph_msg_new(CEPH_MSG_MON_GET_VERSION_REPLY, 1024,
+                                  GFP_NOFS, true);
+        if (!req->reply) {
+                err = -ENOMEM;
+                goto out;
+        }
+
+        p = req->request->front.iov_base;
+        end = p + req->request->front_alloc_len;
+
+        /* fill out request */
+        mutex_lock(&monc->mutex);
+        tid = ++monc->last_tid;
+        ceph_encode_64(&p, tid); /* handle */
+        ceph_encode_string(&p, end, what, strlen(what));
+
+        err = __do_generic_request(monc, tid, req);
+
+        mutex_unlock(&monc->mutex);
+out:
+        kref_put(&req->kref, release_generic_request);
+        return err;
+}
+EXPORT_SYMBOL(ceph_monc_do_get_version);
+
 /*
  * pool ops
  */
@@ -981,6 +1108,10 @@ static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
                 handle_statfs_reply(monc, msg);
                 break;
 
+        case CEPH_MSG_MON_GET_VERSION_REPLY:
+                handle_get_version_reply(monc, msg);
+                break;
+
         case CEPH_MSG_POOLOP_REPLY:
                 handle_poolop_reply(monc, msg);
                 break;
@@ -1029,6 +1160,15 @@ static struct ceph_msg *mon_alloc_msg(struct ceph_connection *con,
         case CEPH_MSG_AUTH_REPLY:
                 m = ceph_msg_get(monc->m_auth_reply);
                 break;
+        case CEPH_MSG_MON_GET_VERSION_REPLY:
+                if (le64_to_cpu(hdr->tid) != 0)
+                        return get_generic_reply(con, hdr, skip);
+
+                /*
+                 * Older OSDs don't set reply tid even if the orignal
+                 * request had a non-zero tid.  Workaround this weirdness
+                 * by falling through to the allocate case.
+                 */
         case CEPH_MSG_MON_MAP:
         case CEPH_MSG_MDS_MAP:
         case CEPH_MSG_OSD_MAP: