ceph: fix memory leak due to possible dentry init race

[safe/jmp/linux-2.6] / fs / ceph / osd_client.c

diff --git a/fs/ceph/osd_client.c b/fs/ceph/osd_client.c
index 3514f71..97a3a57 100644 (file)
--- a/fs/ceph/osd_client.c
+++ b/fs/ceph/osd_client.c
@@ -147,7 +147,7 @@ struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
                req = kzalloc(sizeof(*req), GFP_NOFS);
        }
        if (req == NULL)
-               return ERR_PTR(-ENOMEM);
+               return NULL;
 
        req->r_osdc = osdc;
        req->r_mempool = use_mempool;
@@ -164,10 +164,10 @@ struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
                msg = ceph_msgpool_get(&osdc->msgpool_op_reply, 0);
        else
                msg = ceph_msg_new(CEPH_MSG_OSD_OPREPLY,
-                                  OSD_OPREPLY_FRONT_LEN, 0, 0, NULL);
-       if (IS_ERR(msg)) {
+                                  OSD_OPREPLY_FRONT_LEN);
+       if (!msg) {
                ceph_osdc_put_request(req);
-               return ERR_PTR(PTR_ERR(msg));
+               return NULL;
        }
        req->r_reply = msg;
 
@@ -178,10 +178,10 @@ struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
        if (use_mempool)
                msg = ceph_msgpool_get(&osdc->msgpool_op, 0);
        else
-               msg = ceph_msg_new(CEPH_MSG_OSD_OP, msg_size, 0, 0, NULL);
-       if (IS_ERR(msg)) {
+               msg = ceph_msg_new(CEPH_MSG_OSD_OP, msg_size);
+       if (!msg) {
                ceph_osdc_put_request(req);
-               return ERR_PTR(PTR_ERR(msg));
+               return NULL;
        }
        msg->hdr.type = cpu_to_le16(CEPH_MSG_OSD_OP);
        memset(msg->front.iov_base, 0, msg->front.iov_len);
@@ -715,7 +715,7 @@ static void handle_timeout(struct work_struct *work)
         * should mark the osd as failed and we should find out about
         * it from an updated osd map.
         */
-       while (!list_empty(&osdc->req_lru)) {
+       while (timeout && !list_empty(&osdc->req_lru)) {
                req = list_entry(osdc->req_lru.next, struct ceph_osd_request,
                                 r_req_lru_item);
 
@@ -1078,6 +1078,7 @@ done:
        if (newmap)
                kick_requests(osdc, NULL);
        up_read(&osdc->map_sem);
+       wake_up(&osdc->client->auth_wq);
        return;
 
 bad:
@@ -1087,45 +1088,6 @@ bad:
        return;
 }
 
-
-/*
- * A read request prepares specific pages that data is to be read into.
- * When a message is being read off the wire, we call prepare_pages to
- * find those pages.
- *  0 = success, -1 failure.
- */
-static int __prepare_pages(struct ceph_connection *con,
-                        struct ceph_msg_header *hdr,
-                        struct ceph_osd_request *req,
-                        u64 tid,
-                        struct ceph_msg *m)
-{
-       struct ceph_osd *osd = con->private;
-       struct ceph_osd_client *osdc;
-       int ret = -1;
-       int data_len = le32_to_cpu(hdr->data_len);
-       unsigned data_off = le16_to_cpu(hdr->data_off);
-
-       int want = calc_pages_for(data_off & ~PAGE_MASK, data_len);
-
-       if (!osd)
-               return -1;
-
-       osdc = osd->o_osdc;
-
-       dout("__prepare_pages on msg %p tid %llu, has %d pages, want %d\n", m,
-            tid, req->r_num_pages, want);
-       if (unlikely(req->r_num_pages < want))
-               goto out;
-       m->pages = req->r_pages;
-       m->nr_pages = req->r_num_pages;
-       ret = 0; /* success */
-out:
-       BUG_ON(ret < 0 || m->nr_pages < want);
-
-       return ret;
-}
-
 /*
  * Register request, send initial attempt.
  */
@@ -1302,8 +1264,8 @@ int ceph_osdc_readpages(struct ceph_osd_client *osdc,
                                    CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ,
                                    NULL, 0, truncate_seq, truncate_size, NULL,
                                    false, 1);
-       if (IS_ERR(req))
-               return PTR_ERR(req);
+       if (!req)
+               return -ENOMEM;
 
        /* it may be a short read due to an object boundary */
        req->r_pages = pages;
@@ -1345,8 +1307,8 @@ int ceph_osdc_writepages(struct ceph_osd_client *osdc, struct ceph_vino vino,
                                    snapc, do_sync,
                                    truncate_seq, truncate_size, mtime,
                                    nofail, 1);
-       if (IS_ERR(req))
-               return PTR_ERR(req);
+       if (!req)
+               return -ENOMEM;
 
        /* it may be a short write due to an object boundary */
        req->r_pages = pages;
@@ -1394,7 +1356,8 @@ static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
 }
 
 /*
- * lookup and return message for incoming reply
+ * lookup and return message for incoming reply.  set up reply message
+ * pages.
  */
 static struct ceph_msg *get_reply(struct ceph_connection *con,
                                  struct ceph_msg_header *hdr,
@@ -1407,7 +1370,6 @@ static struct ceph_msg *get_reply(struct ceph_connection *con,
        int front = le32_to_cpu(hdr->front_len);
        int data_len = le32_to_cpu(hdr->data_len);
        u64 tid;
-       int err;
 
        tid = le64_to_cpu(hdr->tid);
        mutex_lock(&osdc->request_mutex);
@@ -1425,13 +1387,14 @@ static struct ceph_msg *get_reply(struct ceph_connection *con,
                     req->r_reply, req->r_con_filling_msg);
                ceph_con_revoke_message(req->r_con_filling_msg, req->r_reply);
                ceph_con_put(req->r_con_filling_msg);
+               req->r_con_filling_msg = NULL;
        }
 
        if (front > req->r_reply->front.iov_len) {
                pr_warning("get_reply front %d > preallocated %d\n",
                           front, (int)req->r_reply->front.iov_len);
-               m = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, front, 0, 0, NULL);
-               if (IS_ERR(m))
+               m = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, front);
+               if (!m)
                        goto out;
                ceph_msg_put(req->r_reply);
                req->r_reply = m;
@@ -1439,12 +1402,19 @@ static struct ceph_msg *get_reply(struct ceph_connection *con,
        m = ceph_msg_get(req->r_reply);
 
        if (data_len > 0) {
-               err = __prepare_pages(con, hdr, req, tid, m);
-               if (err < 0) {
+               unsigned data_off = le16_to_cpu(hdr->data_off);
+               int want = calc_pages_for(data_off & ~PAGE_MASK, data_len);
+
+               if (unlikely(req->r_num_pages < want)) {
+                       pr_warning("tid %lld reply %d > expected %d pages\n",
+                                  tid, want, m->nr_pages);
                        *skip = 1;
                        ceph_msg_put(m);
-                       m = ERR_PTR(err);
+                       m = NULL;
+                       goto out;
                }
+               m->pages = req->r_pages;
+               m->nr_pages = req->r_num_pages;
        }
        *skip = 0;
        req->r_con_filling_msg = ceph_con_get(con);
@@ -1466,7 +1436,7 @@ static struct ceph_msg *alloc_msg(struct ceph_connection *con,
 
        switch (type) {
        case CEPH_MSG_OSD_MAP:
-               return ceph_msg_new(type, front, 0, 0, NULL);
+               return ceph_msg_new(type, front);
        case CEPH_MSG_OSD_OPREPLY:
                return get_reply(con, hdr, skip);
        default: