libceph: fix alloc_msg_with_page_vector() memory leaks

commit e886274031200bb60965c1b9c49b7acda56a93bd upstream.

Make it so that CEPH_MSG_DATA_PAGES data item can own pages,
fixing a bunch of memory leaks for a page vector allocated in
alloc_msg_with_page_vector().  Currently, only watch-notify
messages trigger this allocation, and normally the page vector
is freed either in handle_watch_notify() or by the caller of
ceph_osdc_notify().  But if the message is freed before that
(e.g. if the session faults while reading in the message or
if the notify is stale), we leak the page vector.

This was supposed to be fixed by switching to a message-owned
pagelist, but that never happened.

Fixes: 1907920324f1 ("libceph: support for sending notifies")
Reported-by: Roman Penyaev <rpenyaev@suse.de>
Signed-off-by: Ilya Dryomov <idryomov@gmail.com>
Reviewed-by: Roman Penyaev <rpenyaev@suse.de>
Signed-off-by: Greg Kroah-Hartman <gregkh@linuxfoundation.org>

diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
index c4458dc6a7576260ef4981499c60e048768f307d..76371aaae2d1cf52fb758fae651a566d6672c068 100644 (file)
--- a/include/linux/ceph/messenger.h
+++ b/include/linux/ceph/messenger.h
@@ -175,9 +175,10 @@ struct ceph_msg_data {
 #endif /* CONFIG_BLOCK */
                struct ceph_bvec_iter   bvec_pos;
                struct {
-                       struct page     **pages;        /* NOT OWNER. */
+                       struct page     **pages;
                        size_t          length;         /* total # bytes */
                        unsigned int    alignment;      /* first page */
+                       bool            own_pages;
                };
                struct ceph_pagelist    *pagelist;
        };
@@ -356,8 +357,8 @@ extern void ceph_con_keepalive(struct ceph_connection *con);
 extern bool ceph_con_keepalive_expired(struct ceph_connection *con,
                                       unsigned long interval);
 
-extern void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
-                               size_t length, size_t alignment);
+void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
+                            size_t length, size_t alignment, bool own_pages);
 extern void ceph_msg_data_add_pagelist(struct ceph_msg *msg,
                                struct ceph_pagelist *pagelist);
 #ifdef CONFIG_BLOCK

diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index 5b4bd8261002e449396e5a808ce2bac4734a1b39..f8ca5edc5f2c9d7c20aab13920b71821f3aafcb2 100644 (file)
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -3248,12 +3248,16 @@ static struct ceph_msg_data *ceph_msg_data_add(struct ceph_msg *msg)
 
 static void ceph_msg_data_destroy(struct ceph_msg_data *data)
 {
-       if (data->type == CEPH_MSG_DATA_PAGELIST)
+       if (data->type == CEPH_MSG_DATA_PAGES && data->own_pages) {
+               int num_pages = calc_pages_for(data->alignment, data->length);
+               ceph_release_page_vector(data->pages, num_pages);
+       } else if (data->type == CEPH_MSG_DATA_PAGELIST) {
                ceph_pagelist_release(data->pagelist);
+       }
 }
 
 void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
-               size_t length, size_t alignment)
+                            size_t length, size_t alignment, bool own_pages)
 {
        struct ceph_msg_data *data;
 
@@ -3265,6 +3269,7 @@ void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
        data->pages = pages;
        data->length = length;
        data->alignment = alignment & ~PAGE_MASK;
+       data->own_pages = own_pages;
 
        msg->data_length += length;
 }

diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index ba45b074a362857f18a688fddf5459cf487cbd5f..2352afa62d1fa96d0a7626f8bb98598ef249ed67 100644 (file)
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -962,7 +962,7 @@ static void ceph_osdc_msg_data_add(struct ceph_msg *msg,
                BUG_ON(length > (u64) SIZE_MAX);
                if (length)
                        ceph_msg_data_add_pages(msg, osd_data->pages,
-                                       length, osd_data->alignment);
+                                       length, osd_data->alignment, false);
        } else if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGELIST) {
                BUG_ON(!length);
                ceph_msg_data_add_pagelist(msg, osd_data->pagelist);
@@ -4436,9 +4436,7 @@ static void handle_watch_notify(struct ceph_osd_client *osdc,
                                                        CEPH_MSG_DATA_PAGES);
                                        *lreq->preply_pages = data->pages;
                                        *lreq->preply_len = data->length;
-                               } else {
-                                       ceph_release_page_vector(data->pages,
-                                              calc_pages_for(0, data->length));
+                                       data->own_pages = false;
                                }
                        }
                        lreq->notify_finish_error = return_code;
@@ -5500,9 +5498,6 @@ out_unlock_osdc:
        return m;
 }
 
-/*
- * TODO: switch to a msg-owned pagelist
- */
 static struct ceph_msg *alloc_msg_with_page_vector(struct ceph_msg_header *hdr)
 {
        struct ceph_msg *m;
@@ -5516,7 +5511,6 @@ static struct ceph_msg *alloc_msg_with_page_vector(struct ceph_msg_header *hdr)
 
        if (data_len) {
                struct page **pages;
-               struct ceph_osd_data osd_data;
 
                pages = ceph_alloc_page_vector(calc_pages_for(0, data_len),
                                               GFP_NOIO);
@@ -5525,9 +5519,7 @@ static struct ceph_msg *alloc_msg_with_page_vector(struct ceph_msg_header *hdr)
                        return NULL;
                }
 
-               ceph_osd_data_pages_init(&osd_data, pages, data_len, 0, false,
-                                        false);
-               ceph_osdc_msg_data_add(m, &osd_data);
+               ceph_msg_data_add_pages(m, pages, data_len, 0, true);
        }
 
        return m;