rbd: new request completion code

Do away with partial request completions and all the associated
complexity.  Individual object requests no longer need to be completed
in order -- when the last one becomes ready, we complete the entire
higher level request all at once.

This also wraps up the conversion to a state machine model and
eliminates the recursion described in commit 6d69bb536bac ("rbd:
prevent kernel stack blow up on rbd map").

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 015bd53..2eb0abd 100644 (file)
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -327,6 +327,7 @@ struct rbd_img_request {
        int                     result; /* first nonzero obj_request result */
 
        u32                     obj_request_count;
+       u32                     pending_count;
        struct list_head        obj_requests;   /* rbd_obj_request structs */
 
        struct kref             kref;
@@ -1406,6 +1407,7 @@ static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
        obj_request_img_data_set(obj_request);
        rbd_assert(obj_request->which != BAD_WHICH);
        img_request->obj_request_count++;
+       img_request->pending_count++;
        list_add_tail(&obj_request->links, &img_request->obj_requests);
        dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
                obj_request->which);
@@ -1451,10 +1453,6 @@ static void rbd_obj_request_submit(struct rbd_obj_request *obj_request)
        dout("%s %p object_no %016llx %llu~%llu osd_req %p\n", __func__,
             obj_request, obj_request->object_no, obj_request->offset,
             obj_request->length, osd_req);
-       if (obj_request_img_data_test(obj_request)) {
-               WARN_ON(obj_request->callback != rbd_img_obj_callback);
-               rbd_img_request_get(obj_request->img_request);
-       }
        ceph_osdc_start_request(osd_req->r_osdc, osd_req, false);
 }
 
@@ -2236,8 +2234,6 @@ static void rbd_img_request_submit(struct rbd_img_request *img_request)
        rbd_img_request_put(img_request);
 }
 
-static void rbd_img_end_child_request(struct rbd_img_request *img_req);
-
 static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req,
                                    u64 img_offset, u32 bytes)
 {
@@ -2249,8 +2245,6 @@ static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req,
        if (!child_img_req)
                return -ENOMEM;
 
-       child_img_req->callback = rbd_img_end_child_request;
-
        if (!rbd_img_is_write(img_req)) {
                switch (obj_req->type) {
                case OBJ_REQUEST_BIO:
@@ -2386,8 +2380,6 @@ static int rbd_obj_issue_copyup(struct rbd_obj_request *obj_req, u32 bytes)
        }
 
        rbd_obj_request_submit(obj_req);
-       /* FIXME: in lieu of rbd_img_obj_callback() */
-       rbd_img_request_put(obj_req->img_request);
        return 0;
 }
 
@@ -2540,6 +2532,29 @@ static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req)
        }
 }
 
+static void rbd_obj_end_request(struct rbd_obj_request *obj_req)
+{
+       struct rbd_img_request *img_req = obj_req->img_request;
+
+       rbd_assert((!obj_req->result &&
+                   obj_req->xferred == obj_req->length) ||
+                  (obj_req->result < 0 && !obj_req->xferred));
+       if (!obj_req->result) {
+               img_req->xferred += obj_req->xferred;
+               return;
+       }
+
+       rbd_warn(img_req->rbd_dev,
+                "%s at objno %llu %llu~%llu result %d xferred %llu",
+                obj_op_name(img_req->op_type), obj_req->object_no,
+                obj_req->offset, obj_req->length, obj_req->result,
+                obj_req->xferred);
+       if (!img_req->result) {
+               img_req->result = obj_req->result;
+               img_req->xferred = 0;
+       }
+}
+
 static void rbd_img_end_child_request(struct rbd_img_request *img_req)
 {
        struct rbd_obj_request *obj_req = img_req->obj_request;
@@ -2549,17 +2564,44 @@ static void rbd_img_end_child_request(struct rbd_img_request *img_req)
        obj_req->result = img_req->result;
        obj_req->xferred = img_req->xferred;
        rbd_img_request_put(img_req);
+}
 
-       rbd_obj_handle_request(obj_req);
+static void rbd_img_end_request(struct rbd_img_request *img_req)
+{
+       rbd_assert(!test_bit(IMG_REQ_CHILD, &img_req->flags));
+       rbd_assert((!img_req->result &&
+                   img_req->xferred == blk_rq_bytes(img_req->rq)) ||
+                  (img_req->result < 0 && !img_req->xferred));
+
+       blk_mq_end_request(img_req->rq,
+                          errno_to_blk_status(img_req->result));
+       rbd_img_request_put(img_req);
 }
 
 static void rbd_obj_handle_request(struct rbd_obj_request *obj_req)
 {
+       struct rbd_img_request *img_req;
+
+again:
        if (!__rbd_obj_handle_request(obj_req))
                return;
 
-       obj_request_done_set(obj_req);
-       rbd_obj_request_complete(obj_req);
+       img_req = obj_req->img_request;
+       spin_lock(&img_req->completion_lock);
+       rbd_obj_end_request(obj_req);
+       rbd_assert(img_req->pending_count);
+       if (--img_req->pending_count) {
+               spin_unlock(&img_req->completion_lock);
+               return;
+       }
+
+       spin_unlock(&img_req->completion_lock);
+       if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
+               obj_req = img_req->obj_request;
+               rbd_img_end_child_request(img_req);
+               goto again;
+       }
+       rbd_img_end_request(img_req);
 }
 
 static const struct rbd_client_id rbd_empty_cid;