ceph: control access to page vector for incoming data

When we issue an OSD read, we specify a vector of pages that the data is to
be read into.  The request may be sent multiple times, to multiple OSDs, if
the osdmap changes, which means we can get more than one reply.

Only read data into the page vector if the reply is coming from the
OSD we last sent the request to.  Keep track of which connection is using
the vector by taking a reference.  If another connection was already
using the vector before and a new reply comes in on the right connection,
revoke the pages from the other connection.

Signed-off-by: Sage Weil <sage@newdream.net>

1 files changed, 29 insertions, 0 deletions

diff --git a/fs/ceph/messenger.c b/fs/ceph/messenger.c
index c03b4185c143..506b638a023b 100644
--- a/fs/ceph/messenger.c
+++ b/fs/ceph/messenger.c
@@ -1976,6 +1976,35 @@ void ceph_con_revoke(struct ceph_connection *con, struct ceph_msg *msg)
 }
 
 /*
+ * Revoke a page vector that we may be reading data into
+ */
+void ceph_con_revoke_pages(struct ceph_connection *con, struct page **pages)
+{
+        mutex_lock(&con->mutex);
+        if (con->in_msg && con->in_msg->pages == pages) {
+                unsigned data_len = le32_to_cpu(con->in_hdr.data_len);
+
+                /* skip rest of message */
+                dout("con_revoke_pages %p msg %p pages %p revoked\n", con,
+                     con->in_msg, pages);
+                if (con->in_msg_pos.data_pos < data_len)
+                        con->in_base_pos = con->in_msg_pos.data_pos - data_len;
+                else
+                        con->in_base_pos = con->in_base_pos -
+                                sizeof(struct ceph_msg_header) -
+                                sizeof(struct ceph_msg_footer);
+                con->in_msg->pages = NULL;
+                ceph_msg_put(con->in_msg);
+                con->in_msg = NULL;
+                con->in_tag = CEPH_MSGR_TAG_READY;
+        } else {
+                dout("con_revoke_pages %p msg %p pages %p no-op\n",
+                     con, con->in_msg, pages);
+        }
+        mutex_unlock(&con->mutex);
+}
+
+/*
  * Queue a keepalive byte to ensure the tcp connection is alive.
  */
 void ceph_con_keepalive(struct ceph_connection *con)