kclient: deal with osdc abort vs kick races; revoke pages on msg dup

diff --git a/src/kernel/messenger.c b/src/kernel/messenger.c
index 25b75f24c0145ca2b86eda3711a4ed25707f07b3..1306249b66b542a242ced6b0459e729b24c3d549 100644 (file)
--- a/src/kernel/messenger.c
+++ b/src/kernel/messenger.c
@@ -886,6 +886,10 @@ out:
  */
 static void reset_connection(struct ceph_connection *con)
 {
+       derr(1, "%s%d %u.%u.%u.%u:%u connection reset\n", 
+            ENTITY_NAME(con->peer_name),
+            IPQUADPORT(con->peer_addr.ipaddr));
+
        /* reset connection, out_queue, msg_ and connect_seq */
        /* discard existing out_queue and msg_seq */
        while (!list_empty(&con->out_queue)) {
@@ -1365,9 +1369,12 @@ struct ceph_msg *ceph_msg_maybe_dup(struct ceph_msg *old)
        BUG_ON(!dup);
        memcpy(dup->front.iov_base, old->front.iov_base,
               le32_to_cpu(old->hdr.front_len));
-       if (old->pages)
-               derr(0, "WARNING: unsafely referenced old pages for %p\n",
-                    old);
+       
+       /* revoke old message's pages */
+       mutex_lock(&old->page_mutex);
+       old->pages = 0;
+       mutex_unlock(&old->page_mutex);
+
        ceph_msg_put(old);
        return dup;
 }

diff --git a/src/kernel/osd_client.c b/src/kernel/osd_client.c
index 9a4848fe94d4efbc829b13a5ace6dcdac00e2b0a..fd7ad1d97481960a7fe87241ccdfd7ac1505c611 100644 (file)
--- a/src/kernel/osd_client.c
+++ b/src/kernel/osd_client.c
@@ -72,6 +72,7 @@ static struct ceph_osd_request *alloc_request(int nr_pages,
        req = kmalloc(sizeof(*req) + nr_pages*sizeof(void *), GFP_NOFS);
        if (req == NULL)
                return ERR_PTR(-ENOMEM);
+       req->r_aborted = 0;
        req->r_request = msg;
        req->r_nr_pages = nr_pages;
        atomic_set(&req->r_ref, 1);
@@ -253,8 +254,8 @@ static int kick_requests(struct ceph_osd_client *osdc)
        int osd;
        int ret = 0;
        
-more:
        spin_lock(&osdc->request_lock);
+more:
        got = radix_tree_gang_lookup(&osdc->request_tree, (void **)&req, 
                                     next_tid, 1);
        if (got == 0)
@@ -270,9 +271,11 @@ more:
                dout(20, "kicking tid %llu osd%d\n", req->r_tid, osd);
                get_request(req);
                spin_unlock(&osdc->request_lock);
-               req->r_request = ceph_msg_maybe_dup(req->r_request);  
-               req->r_flags |= CEPH_OSD_OP_RETRY;
-               send_request(osdc, req, osd);
+               req->r_request = ceph_msg_maybe_dup(req->r_request);
+               if (req->r_aborted) {
+                       req->r_flags |= CEPH_OSD_OP_RETRY;
+                       send_request(osdc, req, osd);
+               }
                put_request(req);
                goto more;
        }
@@ -451,9 +454,17 @@ int do_request(struct ceph_osd_client *osdc, struct ceph_osd_request *req)
 
        unregister_request(osdc, req);
        if (rc < 0) {
-               struct ceph_msg *msg = req->r_request;
+               struct ceph_msg *msg;
                dout(0, "tid %llu err %d, revoking %p pages\n", req->r_tid, 
-                    rc, msg);
+                    rc, req->r_request);
+               /* 
+                * mark req aborted _before_ revoking pages, so that
+                * if a racing kick_request _does_ dup the page vec
+                * pointer, it will definitely then see the aborted
+                * flag and not send the request.
+                */
+               req->r_aborted = 1;
+               msg = req->r_request;
                mutex_lock(&msg->page_mutex);
                msg->pages = 0;
                mutex_unlock(&msg->page_mutex);

diff --git a/src/kernel/osd_client.h b/src/kernel/osd_client.h
index c120cf7fa87d045bbc3fcaf7e221d3a4089bc881..fc106e66f723bb949f20f6b20918c693720c9815 100644 (file)
--- a/src/kernel/osd_client.h
+++ b/src/kernel/osd_client.h
@@ -22,6 +22,7 @@ enum {
 
 struct ceph_osd_request {
        __u64             r_tid;
+       int               r_aborted;
        int               r_flags;
        struct ceph_msg  *r_request;
        struct ceph_entity_addr r_last_osd;  /* last osd we sent request to */