From 54ab3b24c536bc9b45ab444830974c6bea57778e Mon Sep 17 00:00:00 2001
From: Ilya Dryomov <idryomov@gmail.com>
Date: Sat, 11 May 2019 16:21:49 +0200
Subject: rbd: get rid of obj_req->xferred, obj_req->result and
 img_req->xferred

obj_req->xferred and img_req->xferred don't bring any value.  The
former is used for short reads and has to be set to obj_req->ex.oe_len
after that and elsewhere.  The latter is just an aggregate.

Use result for short reads (>=0 - number of bytes read, <0 - error) and
pass it around explicitly.  No need to store it in obj_req.

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>
Reviewed-by: Dongsheng Yang <dongsheng.yang@easystack.cn>
---
 drivers/block/rbd.c | 149 ++++++++++++++++++++--------------------------------
 1 file changed, 58 insertions(+), 91 deletions(-)

(limited to 'drivers/block')

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index e5009a34f9c2..a9b0b23148f9 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -276,9 +276,6 @@ struct rbd_obj_request {
 
 	struct ceph_osd_request	*osd_req;
 
-	u64			xferred;	/* bytes transferred */
-	int			result;
-
 	struct kref		kref;
 };
 
@@ -301,7 +298,6 @@ struct rbd_img_request {
 		struct rbd_obj_request	*obj_request;	/* obj req initiator */
 	};
 	spinlock_t		completion_lock;
-	u64			xferred;/* aggregate bytes transferred */
 	int			result;	/* first nonzero obj_request result */
 
 	struct list_head	object_extents;	/* obj_req.ex structs */
@@ -584,6 +580,8 @@ static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
 		u64 *snap_features);
 
+static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result);
+
 static int rbd_open(struct block_device *bdev, fmode_t mode)
 {
 	struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
@@ -1317,6 +1315,8 @@ static void zero_bvecs(struct ceph_bvec_iter *bvec_pos, u32 off, u32 bytes)
 static void rbd_obj_zero_range(struct rbd_obj_request *obj_req, u32 off,
 			       u32 bytes)
 {
+	dout("%s %p data buf %u~%u\n", __func__, obj_req, off, bytes);
+
 	switch (obj_req->img_request->data_type) {
 	case OBJ_REQUEST_BIO:
 		zero_bios(&obj_req->bio_pos, off, bytes);
@@ -1457,28 +1457,26 @@ static bool rbd_img_is_write(struct rbd_img_request *img_req)
 	}
 }
 
-static void rbd_obj_handle_request(struct rbd_obj_request *obj_req);
-
 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
 {
 	struct rbd_obj_request *obj_req = osd_req->r_priv;
+	int result;
 
 	dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
 	     osd_req->r_result, obj_req);
 	rbd_assert(osd_req == obj_req->osd_req);
 
-	obj_req->result = osd_req->r_result < 0 ? osd_req->r_result : 0;
-	if (!obj_req->result && !rbd_img_is_write(obj_req->img_request))
-		obj_req->xferred = osd_req->r_result;
+	/*
+	 * Writes aren't allowed to return a data payload.  In some
+	 * guarded write cases (e.g. stat + zero on an empty object)
+	 * a stat response makes it through, but we don't care.
+	 */
+	if (osd_req->r_result > 0 && rbd_img_is_write(obj_req->img_request))
+		result = 0;
 	else
-		/*
-		 * Writes aren't allowed to return a data payload.  In some
-		 * guarded write cases (e.g. stat + zero on an empty object)
-		 * a stat response makes it through, but we don't care.
-		 */
-		obj_req->xferred = 0;
+		result = osd_req->r_result;
 
-	rbd_obj_handle_request(obj_req);
+	rbd_obj_handle_request(obj_req, result);
 }
 
 static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request)
@@ -2041,7 +2039,6 @@ static int __rbd_img_fill_request(struct rbd_img_request *img_req)
 		if (ret < 0)
 			return ret;
 		if (ret > 0) {
-			img_req->xferred += obj_req->ex.oe_len;
 			img_req->pending_count--;
 			rbd_img_obj_request_del(img_req, obj_req);
 			continue;
@@ -2400,17 +2397,17 @@ static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
 	return 0;
 }
 
-static bool rbd_obj_handle_read(struct rbd_obj_request *obj_req)
+static bool rbd_obj_handle_read(struct rbd_obj_request *obj_req, int *result)
 {
 	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
 	int ret;
 
-	if (obj_req->result == -ENOENT &&
+	if (*result == -ENOENT &&
 	    rbd_dev->parent_overlap && !obj_req->tried_parent) {
 		/* reverse map this object extent onto the parent */
 		ret = rbd_obj_calc_img_extents(obj_req, false);
 		if (ret) {
-			obj_req->result = ret;
+			*result = ret;
 			return true;
 		}
 
@@ -2418,7 +2415,7 @@ static bool rbd_obj_handle_read(struct rbd_obj_request *obj_req)
 			obj_req->tried_parent = true;
 			ret = rbd_obj_read_from_parent(obj_req);
 			if (ret) {
-				obj_req->result = ret;
+				*result = ret;
 				return true;
 			}
 			return false;
@@ -2428,16 +2425,18 @@ static bool rbd_obj_handle_read(struct rbd_obj_request *obj_req)
 	/*
 	 * -ENOENT means a hole in the image -- zero-fill the entire
 	 * length of the request.  A short read also implies zero-fill
-	 * to the end of the request.  In both cases we update xferred
-	 * count to indicate the whole request was satisfied.
+	 * to the end of the request.
 	 */
-	if (obj_req->result == -ENOENT ||
-	    (!obj_req->result && obj_req->xferred < obj_req->ex.oe_len)) {
-		rbd_assert(!obj_req->xferred || !obj_req->result);
-		rbd_obj_zero_range(obj_req, obj_req->xferred,
-				   obj_req->ex.oe_len - obj_req->xferred);
-		obj_req->result = 0;
-		obj_req->xferred = obj_req->ex.oe_len;
+	if (*result == -ENOENT) {
+		rbd_obj_zero_range(obj_req, 0, obj_req->ex.oe_len);
+		*result = 0;
+	} else if (*result >= 0) {
+		if (*result < obj_req->ex.oe_len)
+			rbd_obj_zero_range(obj_req, *result,
+					   obj_req->ex.oe_len - *result);
+		else
+			rbd_assert(*result == obj_req->ex.oe_len);
+		*result = 0;
 	}
 
 	return true;
@@ -2635,14 +2634,13 @@ static int rbd_obj_handle_write_guard(struct rbd_obj_request *obj_req)
 	return rbd_obj_read_from_parent(obj_req);
 }
 
-static bool rbd_obj_handle_write(struct rbd_obj_request *obj_req)
+static bool rbd_obj_handle_write(struct rbd_obj_request *obj_req, int *result)
 {
 	int ret;
 
 	switch (obj_req->write_state) {
 	case RBD_OBJ_WRITE_GUARD:
-		rbd_assert(!obj_req->xferred);
-		if (obj_req->result == -ENOENT) {
+		if (*result == -ENOENT) {
 			/*
 			 * The target object doesn't exist.  Read the data for
 			 * the entire target object up to the overlap point (if
@@ -2650,7 +2648,7 @@ static bool rbd_obj_handle_write(struct rbd_obj_request *obj_req)
 			 */
 			ret = rbd_obj_handle_write_guard(obj_req);
 			if (ret) {
-				obj_req->result = ret;
+				*result = ret;
 				return true;
 			}
 			return false;
@@ -2658,33 +2656,26 @@ static bool rbd_obj_handle_write(struct rbd_obj_request *obj_req)
 		/* fall through */
 	case RBD_OBJ_WRITE_FLAT:
 	case RBD_OBJ_WRITE_COPYUP_OPS:
-		if (!obj_req->result)
-			/*
-			 * There is no such thing as a successful short
-			 * write -- indicate the whole request was satisfied.
-			 */
-			obj_req->xferred = obj_req->ex.oe_len;
 		return true;
 	case RBD_OBJ_WRITE_READ_FROM_PARENT:
-		if (obj_req->result)
+		if (*result < 0)
 			return true;
 
-		rbd_assert(obj_req->xferred);
-		ret = rbd_obj_issue_copyup(obj_req, obj_req->xferred);
+		rbd_assert(*result);
+		ret = rbd_obj_issue_copyup(obj_req, *result);
 		if (ret) {
-			obj_req->result = ret;
-			obj_req->xferred = 0;
+			*result = ret;
 			return true;
 		}
 		return false;
 	case RBD_OBJ_WRITE_COPYUP_EMPTY_SNAPC:
-		if (obj_req->result)
+		if (*result)
 			return true;
 
 		obj_req->write_state = RBD_OBJ_WRITE_COPYUP_OPS;
 		ret = rbd_obj_issue_copyup_ops(obj_req, MODS_ONLY);
 		if (ret) {
-			obj_req->result = ret;
+			*result = ret;
 			return true;
 		}
 		return false;
@@ -2696,24 +2687,23 @@ static bool rbd_obj_handle_write(struct rbd_obj_request *obj_req)
 /*
  * Returns true if @obj_req is completed, or false otherwise.
  */
-static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req)
+static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
+				     int *result)
 {
 	switch (obj_req->img_request->op_type) {
 	case OBJ_OP_READ:
-		return rbd_obj_handle_read(obj_req);
+		return rbd_obj_handle_read(obj_req, result);
 	case OBJ_OP_WRITE:
-		return rbd_obj_handle_write(obj_req);
+		return rbd_obj_handle_write(obj_req, result);
 	case OBJ_OP_DISCARD:
 	case OBJ_OP_ZEROOUT:
-		if (rbd_obj_handle_write(obj_req)) {
+		if (rbd_obj_handle_write(obj_req, result)) {
 			/*
 			 * Hide -ENOENT from delete/truncate/zero -- discarding
 			 * a non-existent object is not a problem.
 			 */
-			if (obj_req->result == -ENOENT) {
-				obj_req->result = 0;
-				obj_req->xferred = obj_req->ex.oe_len;
-			}
+			if (*result == -ENOENT)
+				*result = 0;
 			return true;
 		}
 		return false;
@@ -2722,66 +2712,41 @@ static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req)
 	}
 }
 
-static void rbd_obj_end_request(struct rbd_obj_request *obj_req)
+static void rbd_obj_end_request(struct rbd_obj_request *obj_req, int result)
 {
 	struct rbd_img_request *img_req = obj_req->img_request;
 
-	rbd_assert((!obj_req->result &&
-		    obj_req->xferred == obj_req->ex.oe_len) ||
-		   (obj_req->result < 0 && !obj_req->xferred));
-	if (!obj_req->result) {
-		img_req->xferred += obj_req->xferred;
+	rbd_assert(result <= 0);
+	if (!result)
 		return;
-	}
 
-	rbd_warn(img_req->rbd_dev,
-		 "%s at objno %llu %llu~%llu result %d xferred %llu",
+	rbd_warn(img_req->rbd_dev, "%s at objno %llu %llu~%llu result %d",
 		 obj_op_name(img_req->op_type), obj_req->ex.oe_objno,
-		 obj_req->ex.oe_off, obj_req->ex.oe_len, 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;
-
-	rbd_assert(test_bit(IMG_REQ_CHILD, &img_req->flags));
-	rbd_assert((!img_req->result &&
-		    img_req->xferred == rbd_obj_img_extents_bytes(obj_req)) ||
-		   (img_req->result < 0 && !img_req->xferred));
-
-	obj_req->result = img_req->result;
-	obj_req->xferred = img_req->xferred;
-	rbd_img_request_put(img_req);
+		 obj_req->ex.oe_off, obj_req->ex.oe_len, result);
+	if (!img_req->result)
+		img_req->result = result;
 }
 
 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)
+static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result)
 {
 	struct rbd_img_request *img_req;
 
 again:
-	if (!__rbd_obj_handle_request(obj_req))
+	if (!__rbd_obj_handle_request(obj_req, &result))
 		return;
 
 	img_req = obj_req->img_request;
 	spin_lock(&img_req->completion_lock);
-	rbd_obj_end_request(obj_req);
+	rbd_obj_end_request(obj_req, result);
 	rbd_assert(img_req->pending_count);
 	if (--img_req->pending_count) {
 		spin_unlock(&img_req->completion_lock);
@@ -2789,9 +2754,11 @@ again:
 	}
 
 	spin_unlock(&img_req->completion_lock);
+	rbd_assert(img_req->result <= 0);
 	if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
 		obj_req = img_req->obj_request;
-		rbd_img_end_child_request(img_req);
+		result = img_req->result ?: rbd_obj_img_extents_bytes(obj_req);
+		rbd_img_request_put(img_req);
 		goto again;
 	}
 	rbd_img_end_request(img_req);
-- 
cgit v1.2.3


From a9b67e69949d20dcb38ea6aaed5500318c7c91f6 Mon Sep 17 00:00:00 2001
From: Ilya Dryomov <idryomov@gmail.com>
Date: Wed, 8 May 2019 13:35:57 +0200
Subject: rbd: replace obj_req->tried_parent with obj_req->read_state

Make rbd_obj_handle_read() look like a state machine and get rid of
the necessity to patch result in rbd_obj_handle_request(), completing
the removal of obj_req->xferred and img_req->xferred.

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>
Reviewed-by: Dongsheng Yang <dongsheng.yang@easystack.cn>
---
 drivers/block/rbd.c | 82 ++++++++++++++++++++++++++++++-----------------------
 1 file changed, 46 insertions(+), 36 deletions(-)

(limited to 'drivers/block')

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index a9b0b23148f9..7925b2fdde79 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -219,6 +219,11 @@ enum obj_operation_type {
 	OBJ_OP_ZEROOUT,
 };
 
+enum rbd_obj_read_state {
+	RBD_OBJ_READ_OBJECT = 1,
+	RBD_OBJ_READ_PARENT,
+};
+
 /*
  * Writes go through the following state machine to deal with
  * layering:
@@ -255,7 +260,7 @@ enum rbd_obj_write_state {
 struct rbd_obj_request {
 	struct ceph_object_extent ex;
 	union {
-		bool			tried_parent;	/* for reads */
+		enum rbd_obj_read_state	 read_state;	/* for reads */
 		enum rbd_obj_write_state write_state;	/* for writes */
 	};
 
@@ -1794,6 +1799,7 @@ static int rbd_obj_setup_read(struct rbd_obj_request *obj_req)
 	rbd_osd_req_setup_data(obj_req, 0);
 
 	rbd_osd_req_format_read(obj_req);
+	obj_req->read_state = RBD_OBJ_READ_OBJECT;
 	return 0;
 }
 
@@ -2402,44 +2408,48 @@ static bool rbd_obj_handle_read(struct rbd_obj_request *obj_req, int *result)
 	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
 	int ret;
 
-	if (*result == -ENOENT &&
-	    rbd_dev->parent_overlap && !obj_req->tried_parent) {
-		/* reverse map this object extent onto the parent */
-		ret = rbd_obj_calc_img_extents(obj_req, false);
-		if (ret) {
-			*result = ret;
-			return true;
-		}
-
-		if (obj_req->num_img_extents) {
-			obj_req->tried_parent = true;
-			ret = rbd_obj_read_from_parent(obj_req);
+	switch (obj_req->read_state) {
+	case RBD_OBJ_READ_OBJECT:
+		if (*result == -ENOENT && rbd_dev->parent_overlap) {
+			/* reverse map this object extent onto the parent */
+			ret = rbd_obj_calc_img_extents(obj_req, false);
 			if (ret) {
 				*result = ret;
 				return true;
 			}
-			return false;
+			if (obj_req->num_img_extents) {
+				ret = rbd_obj_read_from_parent(obj_req);
+				if (ret) {
+					*result = ret;
+					return true;
+				}
+				obj_req->read_state = RBD_OBJ_READ_PARENT;
+				return false;
+			}
 		}
-	}
 
-	/*
-	 * -ENOENT means a hole in the image -- zero-fill the entire
-	 * length of the request.  A short read also implies zero-fill
-	 * to the end of the request.
-	 */
-	if (*result == -ENOENT) {
-		rbd_obj_zero_range(obj_req, 0, obj_req->ex.oe_len);
-		*result = 0;
-	} else if (*result >= 0) {
-		if (*result < obj_req->ex.oe_len)
-			rbd_obj_zero_range(obj_req, *result,
-					   obj_req->ex.oe_len - *result);
-		else
-			rbd_assert(*result == obj_req->ex.oe_len);
-		*result = 0;
+		/*
+		 * -ENOENT means a hole in the image -- zero-fill the entire
+		 * length of the request.  A short read also implies zero-fill
+		 * to the end of the request.
+		 */
+		if (*result == -ENOENT) {
+			rbd_obj_zero_range(obj_req, 0, obj_req->ex.oe_len);
+			*result = 0;
+		} else if (*result >= 0) {
+			if (*result < obj_req->ex.oe_len)
+				rbd_obj_zero_range(obj_req, *result,
+						obj_req->ex.oe_len - *result);
+			else
+				rbd_assert(*result == obj_req->ex.oe_len);
+			*result = 0;
+		}
+		return true;
+	case RBD_OBJ_READ_PARENT:
+		return true;
+	default:
+		BUG();
 	}
-
-	return true;
 }
 
 /*
@@ -2658,11 +2668,11 @@ static bool rbd_obj_handle_write(struct rbd_obj_request *obj_req, int *result)
 	case RBD_OBJ_WRITE_COPYUP_OPS:
 		return true;
 	case RBD_OBJ_WRITE_READ_FROM_PARENT:
-		if (*result < 0)
+		if (*result)
 			return true;
 
-		rbd_assert(*result);
-		ret = rbd_obj_issue_copyup(obj_req, *result);
+		ret = rbd_obj_issue_copyup(obj_req,
+					   rbd_obj_img_extents_bytes(obj_req));
 		if (ret) {
 			*result = ret;
 			return true;
@@ -2757,7 +2767,7 @@ again:
 	rbd_assert(img_req->result <= 0);
 	if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
 		obj_req = img_req->obj_request;
-		result = img_req->result ?: rbd_obj_img_extents_bytes(obj_req);
+		result = img_req->result;
 		rbd_img_request_put(img_req);
 		goto again;
 	}
-- 
cgit v1.2.3


From 0ad5d953548fe336134026b039ad4fd9e6594f16 Mon Sep 17 00:00:00 2001
From: Ilya Dryomov <idryomov@gmail.com>
Date: Tue, 14 May 2019 20:45:38 +0200
Subject: rbd: get rid of RBD_OBJ_WRITE_{FLAT,GUARD}

In preparation for moving OSD request allocation and submission into
object request state machines, get rid of RBD_OBJ_WRITE_{FLAT,GUARD}.
We would need to start in a new state, whether the request is guarded
or not.  Unify them into RBD_OBJ_WRITE_OBJECT and pass guard info
through obj_req->flags.

While at it, make our ENOENT handling a little more precise: only hide
ENOENT when it is actually expected, that is on delete.

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>
Reviewed-by: Dongsheng Yang <dongsheng.yang@easystack.cn>
---
 drivers/block/rbd.c | 112 ++++++++++++++++++++++++++++------------------------
 1 file changed, 60 insertions(+), 52 deletions(-)

(limited to 'drivers/block')

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 7925b2fdde79..488da877a2bb 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -219,6 +219,9 @@ enum obj_operation_type {
 	OBJ_OP_ZEROOUT,
 };
 
+#define RBD_OBJ_FLAG_DELETION			(1U << 0)
+#define RBD_OBJ_FLAG_COPYUP_ENABLED		(1U << 1)
+
 enum rbd_obj_read_state {
 	RBD_OBJ_READ_OBJECT = 1,
 	RBD_OBJ_READ_PARENT,
@@ -250,8 +253,7 @@ enum rbd_obj_read_state {
  * even if there is a parent).
  */
 enum rbd_obj_write_state {
-	RBD_OBJ_WRITE_FLAT = 1,
-	RBD_OBJ_WRITE_GUARD,
+	RBD_OBJ_WRITE_OBJECT = 1,
 	RBD_OBJ_WRITE_READ_FROM_PARENT,
 	RBD_OBJ_WRITE_COPYUP_EMPTY_SNAPC,
 	RBD_OBJ_WRITE_COPYUP_OPS,
@@ -259,6 +261,7 @@ enum rbd_obj_write_state {
 
 struct rbd_obj_request {
 	struct ceph_object_extent ex;
+	unsigned int		flags;	/* RBD_OBJ_FLAG_* */
 	union {
 		enum rbd_obj_read_state	 read_state;	/* for reads */
 		enum rbd_obj_write_state write_state;	/* for writes */
@@ -1858,7 +1861,6 @@ static void __rbd_obj_setup_write(struct rbd_obj_request *obj_req,
 static int rbd_obj_setup_write(struct rbd_obj_request *obj_req)
 {
 	unsigned int num_osd_ops, which = 0;
-	bool need_guard;
 	int ret;
 
 	/* reverse map the entire object onto the parent */
@@ -1866,23 +1868,24 @@ static int rbd_obj_setup_write(struct rbd_obj_request *obj_req)
 	if (ret)
 		return ret;
 
-	need_guard = rbd_obj_copyup_enabled(obj_req);
-	num_osd_ops = need_guard + count_write_ops(obj_req);
+	if (rbd_obj_copyup_enabled(obj_req))
+		obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
+
+	num_osd_ops = count_write_ops(obj_req);
+	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
+		num_osd_ops++; /* stat */
 
 	obj_req->osd_req = rbd_osd_req_create(obj_req, num_osd_ops);
 	if (!obj_req->osd_req)
 		return -ENOMEM;
 
-	if (need_guard) {
+	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
 		ret = __rbd_obj_setup_stat(obj_req, which++);
 		if (ret)
 			return ret;
-
-		obj_req->write_state = RBD_OBJ_WRITE_GUARD;
-	} else {
-		obj_req->write_state = RBD_OBJ_WRITE_FLAT;
 	}
 
+	obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
 	__rbd_obj_setup_write(obj_req, which);
 	return 0;
 }
@@ -1921,11 +1924,15 @@ static int rbd_obj_setup_discard(struct rbd_obj_request *obj_req)
 	if (ret)
 		return ret;
 
+	if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents)
+		obj_req->flags |= RBD_OBJ_FLAG_DELETION;
+
 	obj_req->osd_req = rbd_osd_req_create(obj_req, 1);
 	if (!obj_req->osd_req)
 		return -ENOMEM;
 
 	if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
+		rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
 		osd_req_op_init(obj_req->osd_req, 0, CEPH_OSD_OP_DELETE, 0);
 	} else {
 		dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
@@ -1936,7 +1943,7 @@ static int rbd_obj_setup_discard(struct rbd_obj_request *obj_req)
 				       off, next_off - off, 0, 0);
 	}
 
-	obj_req->write_state = RBD_OBJ_WRITE_FLAT;
+	obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
 	rbd_osd_req_format_write(obj_req);
 	return 0;
 }
@@ -1961,11 +1968,12 @@ static void __rbd_obj_setup_zeroout(struct rbd_obj_request *obj_req,
 
 	if (rbd_obj_is_entire(obj_req)) {
 		if (obj_req->num_img_extents) {
-			if (!rbd_obj_copyup_enabled(obj_req))
+			if (!(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
 				osd_req_op_init(obj_req->osd_req, which++,
 						CEPH_OSD_OP_CREATE, 0);
 			opcode = CEPH_OSD_OP_TRUNCATE;
 		} else {
+			rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
 			osd_req_op_init(obj_req->osd_req, which++,
 					CEPH_OSD_OP_DELETE, 0);
 			opcode = 0;
@@ -1986,7 +1994,6 @@ static void __rbd_obj_setup_zeroout(struct rbd_obj_request *obj_req,
 static int rbd_obj_setup_zeroout(struct rbd_obj_request *obj_req)
 {
 	unsigned int num_osd_ops, which = 0;
-	bool need_guard;
 	int ret;
 
 	/* reverse map the entire object onto the parent */
@@ -1994,23 +2001,28 @@ static int rbd_obj_setup_zeroout(struct rbd_obj_request *obj_req)
 	if (ret)
 		return ret;
 
-	need_guard = rbd_obj_copyup_enabled(obj_req);
-	num_osd_ops = need_guard + count_zeroout_ops(obj_req);
+	if (rbd_obj_copyup_enabled(obj_req))
+		obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
+	if (!obj_req->num_img_extents) {
+		if (rbd_obj_is_entire(obj_req))
+			obj_req->flags |= RBD_OBJ_FLAG_DELETION;
+	}
+
+	num_osd_ops = count_zeroout_ops(obj_req);
+	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
+		num_osd_ops++; /* stat */
 
 	obj_req->osd_req = rbd_osd_req_create(obj_req, num_osd_ops);
 	if (!obj_req->osd_req)
 		return -ENOMEM;
 
-	if (need_guard) {
+	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
 		ret = __rbd_obj_setup_stat(obj_req, which++);
 		if (ret)
 			return ret;
-
-		obj_req->write_state = RBD_OBJ_WRITE_GUARD;
-	} else {
-		obj_req->write_state = RBD_OBJ_WRITE_FLAT;
 	}
 
+	obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
 	__rbd_obj_setup_zeroout(obj_req, which);
 	return 0;
 }
@@ -2617,6 +2629,11 @@ static int setup_copyup_bvecs(struct rbd_obj_request *obj_req, u64 obj_overlap)
 	return 0;
 }
 
+/*
+ * The target object doesn't exist.  Read the data for the entire
+ * target object up to the overlap point (if any) from the parent,
+ * so we can use it for a copyup.
+ */
 static int rbd_obj_handle_write_guard(struct rbd_obj_request *obj_req)
 {
 	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
@@ -2649,22 +2666,24 @@ static bool rbd_obj_handle_write(struct rbd_obj_request *obj_req, int *result)
 	int ret;
 
 	switch (obj_req->write_state) {
-	case RBD_OBJ_WRITE_GUARD:
+	case RBD_OBJ_WRITE_OBJECT:
 		if (*result == -ENOENT) {
+			if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
+				ret = rbd_obj_handle_write_guard(obj_req);
+				if (ret) {
+					*result = ret;
+					return true;
+				}
+				return false;
+			}
 			/*
-			 * The target object doesn't exist.  Read the data for
-			 * the entire target object up to the overlap point (if
-			 * any) from the parent, so we can use it for a copyup.
+			 * On a non-existent object:
+			 *   delete - -ENOENT, truncate/zero - 0
 			 */
-			ret = rbd_obj_handle_write_guard(obj_req);
-			if (ret) {
-				*result = ret;
-				return true;
-			}
-			return false;
+			if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
+				*result = 0;
 		}
 		/* fall through */
-	case RBD_OBJ_WRITE_FLAT:
 	case RBD_OBJ_WRITE_COPYUP_OPS:
 		return true;
 	case RBD_OBJ_WRITE_READ_FROM_PARENT:
@@ -2695,31 +2714,20 @@ static bool rbd_obj_handle_write(struct rbd_obj_request *obj_req, int *result)
 }
 
 /*
- * Returns true if @obj_req is completed, or false otherwise.
+ * Return true if @obj_req is completed.
  */
 static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
 				     int *result)
 {
-	switch (obj_req->img_request->op_type) {
-	case OBJ_OP_READ:
-		return rbd_obj_handle_read(obj_req, result);
-	case OBJ_OP_WRITE:
-		return rbd_obj_handle_write(obj_req, result);
-	case OBJ_OP_DISCARD:
-	case OBJ_OP_ZEROOUT:
-		if (rbd_obj_handle_write(obj_req, result)) {
-			/*
-			 * Hide -ENOENT from delete/truncate/zero -- discarding
-			 * a non-existent object is not a problem.
-			 */
-			if (*result == -ENOENT)
-				*result = 0;
-			return true;
-		}
-		return false;
-	default:
-		BUG();
-	}
+	struct rbd_img_request *img_req = obj_req->img_request;
+	bool done;
+
+	if (!rbd_img_is_write(img_req))
+		done = rbd_obj_handle_read(obj_req, result);
+	else
+		done = rbd_obj_handle_write(obj_req, result);
+
+	return done;
 }
 
 static void rbd_obj_end_request(struct rbd_obj_request *obj_req, int result)
-- 
cgit v1.2.3


From 85b5e6d11898fddfcb2832024f73454d448f76e0 Mon Sep 17 00:00:00 2001
From: Ilya Dryomov <idryomov@gmail.com>
Date: Tue, 14 May 2019 21:06:07 +0200
Subject: rbd: move OSD request submission into object request state machines

Start eliminating asymmetry where the initial OSD request is allocated
and submitted from outside the state machine, making error handling and
restarts harder than they could be.  This commit deals with submission,
a commit that deals with allocation will follow.

Note that this commit adds parent chain recursion on the submission
side:

  rbd_img_request_submit
    rbd_obj_handle_request
      __rbd_obj_handle_request
        rbd_obj_handle_read
          rbd_obj_handle_write_guard
            rbd_obj_read_from_parent
              rbd_img_request_submit

This will be fixed in the next commit.

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>
Reviewed-by: Dongsheng Yang <dongsheng.yang@easystack.cn>
---
 drivers/block/rbd.c | 60 +++++++++++++++++++++++++++++++++++++++++++----------
 1 file changed, 49 insertions(+), 11 deletions(-)

(limited to 'drivers/block')

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 488da877a2bb..9c6be82353c0 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -223,7 +223,8 @@ enum obj_operation_type {
 #define RBD_OBJ_FLAG_COPYUP_ENABLED		(1U << 1)
 
 enum rbd_obj_read_state {
-	RBD_OBJ_READ_OBJECT = 1,
+	RBD_OBJ_READ_START = 1,
+	RBD_OBJ_READ_OBJECT,
 	RBD_OBJ_READ_PARENT,
 };
 
@@ -253,7 +254,8 @@ enum rbd_obj_read_state {
  * even if there is a parent).
  */
 enum rbd_obj_write_state {
-	RBD_OBJ_WRITE_OBJECT = 1,
+	RBD_OBJ_WRITE_START = 1,
+	RBD_OBJ_WRITE_OBJECT,
 	RBD_OBJ_WRITE_READ_FROM_PARENT,
 	RBD_OBJ_WRITE_COPYUP_EMPTY_SNAPC,
 	RBD_OBJ_WRITE_COPYUP_OPS,
@@ -284,6 +286,7 @@ struct rbd_obj_request {
 
 	struct ceph_osd_request	*osd_req;
 
+	struct mutex		state_mutex;
 	struct kref		kref;
 };
 
@@ -1560,6 +1563,7 @@ static struct rbd_obj_request *rbd_obj_request_create(void)
 		return NULL;
 
 	ceph_object_extent_init(&obj_request->ex);
+	mutex_init(&obj_request->state_mutex);
 	kref_init(&obj_request->kref);
 
 	dout("%s %p\n", __func__, obj_request);
@@ -1802,7 +1806,7 @@ static int rbd_obj_setup_read(struct rbd_obj_request *obj_req)
 	rbd_osd_req_setup_data(obj_req, 0);
 
 	rbd_osd_req_format_read(obj_req);
-	obj_req->read_state = RBD_OBJ_READ_OBJECT;
+	obj_req->read_state = RBD_OBJ_READ_START;
 	return 0;
 }
 
@@ -1885,7 +1889,7 @@ static int rbd_obj_setup_write(struct rbd_obj_request *obj_req)
 			return ret;
 	}
 
-	obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
+	obj_req->write_state = RBD_OBJ_WRITE_START;
 	__rbd_obj_setup_write(obj_req, which);
 	return 0;
 }
@@ -1943,7 +1947,7 @@ static int rbd_obj_setup_discard(struct rbd_obj_request *obj_req)
 				       off, next_off - off, 0, 0);
 	}
 
-	obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
+	obj_req->write_state = RBD_OBJ_WRITE_START;
 	rbd_osd_req_format_write(obj_req);
 	return 0;
 }
@@ -2022,7 +2026,7 @@ static int rbd_obj_setup_zeroout(struct rbd_obj_request *obj_req)
 			return ret;
 	}
 
-	obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
+	obj_req->write_state = RBD_OBJ_WRITE_START;
 	__rbd_obj_setup_zeroout(obj_req, which);
 	return 0;
 }
@@ -2363,11 +2367,17 @@ static void rbd_img_request_submit(struct rbd_img_request *img_request)
 
 	rbd_img_request_get(img_request);
 	for_each_obj_request(img_request, obj_request)
-		rbd_obj_request_submit(obj_request);
+		rbd_obj_handle_request(obj_request, 0);
 
 	rbd_img_request_put(img_request);
 }
 
+static int rbd_obj_read_object(struct rbd_obj_request *obj_req)
+{
+	rbd_obj_request_submit(obj_req);
+	return 0;
+}
+
 static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
 {
 	struct rbd_img_request *img_req = obj_req->img_request;
@@ -2415,12 +2425,22 @@ static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
 	return 0;
 }
 
-static bool rbd_obj_handle_read(struct rbd_obj_request *obj_req, int *result)
+static bool rbd_obj_advance_read(struct rbd_obj_request *obj_req, int *result)
 {
 	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
 	int ret;
 
 	switch (obj_req->read_state) {
+	case RBD_OBJ_READ_START:
+		rbd_assert(!*result);
+
+		ret = rbd_obj_read_object(obj_req);
+		if (ret) {
+			*result = ret;
+			return true;
+		}
+		obj_req->read_state = RBD_OBJ_READ_OBJECT;
+		return false;
 	case RBD_OBJ_READ_OBJECT:
 		if (*result == -ENOENT && rbd_dev->parent_overlap) {
 			/* reverse map this object extent onto the parent */
@@ -2464,6 +2484,12 @@ static bool rbd_obj_handle_read(struct rbd_obj_request *obj_req, int *result)
 	}
 }
 
+static int rbd_obj_write_object(struct rbd_obj_request *obj_req)
+{
+	rbd_obj_request_submit(obj_req);
+	return 0;
+}
+
 /*
  * copyup_bvecs pages are never highmem pages
  */
@@ -2661,11 +2687,21 @@ static int rbd_obj_handle_write_guard(struct rbd_obj_request *obj_req)
 	return rbd_obj_read_from_parent(obj_req);
 }
 
-static bool rbd_obj_handle_write(struct rbd_obj_request *obj_req, int *result)
+static bool rbd_obj_advance_write(struct rbd_obj_request *obj_req, int *result)
 {
 	int ret;
 
 	switch (obj_req->write_state) {
+	case RBD_OBJ_WRITE_START:
+		rbd_assert(!*result);
+
+		ret = rbd_obj_write_object(obj_req);
+		if (ret) {
+			*result = ret;
+			return true;
+		}
+		obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
+		return false;
 	case RBD_OBJ_WRITE_OBJECT:
 		if (*result == -ENOENT) {
 			if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
@@ -2722,10 +2758,12 @@ static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
 	struct rbd_img_request *img_req = obj_req->img_request;
 	bool done;
 
+	mutex_lock(&obj_req->state_mutex);
 	if (!rbd_img_is_write(img_req))
-		done = rbd_obj_handle_read(obj_req, result);
+		done = rbd_obj_advance_read(obj_req, result);
 	else
-		done = rbd_obj_handle_write(obj_req, result);
+		done = rbd_obj_advance_write(obj_req, result);
+	mutex_unlock(&obj_req->state_mutex);
 
 	return done;
 }
-- 
cgit v1.2.3


From 0192ce2ee68b54394ba62e5668067bf01f7bc609 Mon Sep 17 00:00:00 2001
From: Ilya Dryomov <idryomov@gmail.com>
Date: Thu, 16 May 2019 15:06:56 +0200
Subject: rbd: introduce image request state machine

Make it possible to schedule image requests on a workqueue.  This fixes
parent chain recursion added in the previous commit and lays the ground
for exclusive lock wait/wake improvements.

The "wait for pending subrequests and report first nonzero result" code
is generalized to be used by object request state machine.

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>
Reviewed-by: Dongsheng Yang <dongsheng.yang@easystack.cn>
---
 drivers/block/rbd.c | 194 +++++++++++++++++++++++++++++++++++++---------------
 1 file changed, 137 insertions(+), 57 deletions(-)

(limited to 'drivers/block')

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 9c6be82353c0..51dd1b99c242 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -203,6 +203,11 @@ struct rbd_client {
 	struct list_head	node;
 };
 
+struct pending_result {
+	int			result;		/* first nonzero result */
+	int			num_pending;
+};
+
 struct rbd_img_request;
 
 enum obj_request_type {
@@ -295,11 +300,18 @@ enum img_req_flags {
 	IMG_REQ_LAYERED,	/* ENOENT handling: normal = 0, layered = 1 */
 };
 
+enum rbd_img_state {
+	RBD_IMG_START = 1,
+	__RBD_IMG_OBJECT_REQUESTS,
+	RBD_IMG_OBJECT_REQUESTS,
+};
+
 struct rbd_img_request {
 	struct rbd_device	*rbd_dev;
 	enum obj_operation_type	op_type;
 	enum obj_request_type	data_type;
 	unsigned long		flags;
+	enum rbd_img_state	state;
 	union {
 		u64			snap_id;	/* for reads */
 		struct ceph_snap_context *snapc;	/* for writes */
@@ -308,12 +320,13 @@ struct rbd_img_request {
 		struct request		*rq;		/* block request */
 		struct rbd_obj_request	*obj_request;	/* obj req initiator */
 	};
-	spinlock_t		completion_lock;
-	int			result;	/* first nonzero obj_request result */
 
 	struct list_head	object_extents;	/* obj_req.ex structs */
-	u32			pending_count;
 
+	struct mutex		state_mutex;
+	struct pending_result	pending;
+	struct work_struct	work;
+	int			work_result;
 	struct kref		kref;
 };
 
@@ -592,6 +605,23 @@ static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
 		u64 *snap_features);
 
 static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result);
+static void rbd_img_handle_request(struct rbd_img_request *img_req, int result);
+
+/*
+ * Return true if nothing else is pending.
+ */
+static bool pending_result_dec(struct pending_result *pending, int *result)
+{
+	rbd_assert(pending->num_pending > 0);
+
+	if (*result && !pending->result)
+		pending->result = *result;
+	if (--pending->num_pending)
+		return false;
+
+	*result = pending->result;
+	return true;
+}
 
 static int rbd_open(struct block_device *bdev, fmode_t mode)
 {
@@ -1350,13 +1380,6 @@ static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
 	kref_put(&obj_request->kref, rbd_obj_request_destroy);
 }
 
-static void rbd_img_request_get(struct rbd_img_request *img_request)
-{
-	dout("%s: img %p (was %d)\n", __func__, img_request,
-	     kref_read(&img_request->kref));
-	kref_get(&img_request->kref);
-}
-
 static void rbd_img_request_destroy(struct kref *kref);
 static void rbd_img_request_put(struct rbd_img_request *img_request)
 {
@@ -1373,7 +1396,6 @@ static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
 
 	/* Image request now owns object's original reference */
 	obj_request->img_request = img_request;
-	img_request->pending_count++;
 	dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
 }
 
@@ -1694,8 +1716,8 @@ static struct rbd_img_request *rbd_img_request_create(
 	if (rbd_dev_parent_get(rbd_dev))
 		img_request_layered_set(img_request);
 
-	spin_lock_init(&img_request->completion_lock);
 	INIT_LIST_HEAD(&img_request->object_extents);
+	mutex_init(&img_request->state_mutex);
 	kref_init(&img_request->kref);
 
 	dout("%s: rbd_dev %p %s -> img %p\n", __func__, rbd_dev,
@@ -2061,7 +2083,6 @@ static int __rbd_img_fill_request(struct rbd_img_request *img_req)
 		if (ret < 0)
 			return ret;
 		if (ret > 0) {
-			img_req->pending_count--;
 			rbd_img_obj_request_del(img_req, obj_req);
 			continue;
 		}
@@ -2071,6 +2092,7 @@ static int __rbd_img_fill_request(struct rbd_img_request *img_req)
 			return ret;
 	}
 
+	img_req->state = RBD_IMG_START;
 	return 0;
 }
 
@@ -2359,17 +2381,19 @@ static int rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
 					 &it);
 }
 
-static void rbd_img_request_submit(struct rbd_img_request *img_request)
+static void rbd_img_handle_request_work(struct work_struct *work)
 {
-	struct rbd_obj_request *obj_request;
+	struct rbd_img_request *img_req =
+	    container_of(work, struct rbd_img_request, work);
 
-	dout("%s: img %p\n", __func__, img_request);
-
-	rbd_img_request_get(img_request);
-	for_each_obj_request(img_request, obj_request)
-		rbd_obj_handle_request(obj_request, 0);
+	rbd_img_handle_request(img_req, img_req->work_result);
+}
 
-	rbd_img_request_put(img_request);
+static void rbd_img_schedule(struct rbd_img_request *img_req, int result)
+{
+	INIT_WORK(&img_req->work, rbd_img_handle_request_work);
+	img_req->work_result = result;
+	queue_work(rbd_wq, &img_req->work);
 }
 
 static int rbd_obj_read_object(struct rbd_obj_request *obj_req)
@@ -2421,7 +2445,8 @@ static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
 		return ret;
 	}
 
-	rbd_img_request_submit(child_img_req);
+	/* avoid parent chain recursion */
+	rbd_img_schedule(child_img_req, 0);
 	return 0;
 }
 
@@ -2756,6 +2781,7 @@ static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
 				     int *result)
 {
 	struct rbd_img_request *img_req = obj_req->img_request;
+	struct rbd_device *rbd_dev = img_req->rbd_dev;
 	bool done;
 
 	mutex_lock(&obj_req->state_mutex);
@@ -2765,59 +2791,113 @@ static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
 		done = rbd_obj_advance_write(obj_req, result);
 	mutex_unlock(&obj_req->state_mutex);
 
+	if (done && *result) {
+		rbd_assert(*result < 0);
+		rbd_warn(rbd_dev, "%s at objno %llu %llu~%llu result %d",
+			 obj_op_name(img_req->op_type), obj_req->ex.oe_objno,
+			 obj_req->ex.oe_off, obj_req->ex.oe_len, *result);
+	}
 	return done;
 }
 
-static void rbd_obj_end_request(struct rbd_obj_request *obj_req, int result)
+/*
+ * This is open-coded in rbd_img_handle_request() to avoid parent chain
+ * recursion.
+ */
+static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result)
+{
+	if (__rbd_obj_handle_request(obj_req, &result))
+		rbd_img_handle_request(obj_req->img_request, result);
+}
+
+static void rbd_img_object_requests(struct rbd_img_request *img_req)
 {
-	struct rbd_img_request *img_req = obj_req->img_request;
+	struct rbd_obj_request *obj_req;
 
-	rbd_assert(result <= 0);
-	if (!result)
-		return;
+	rbd_assert(!img_req->pending.result && !img_req->pending.num_pending);
+
+	for_each_obj_request(img_req, obj_req) {
+		int result = 0;
 
-	rbd_warn(img_req->rbd_dev, "%s at objno %llu %llu~%llu result %d",
-		 obj_op_name(img_req->op_type), obj_req->ex.oe_objno,
-		 obj_req->ex.oe_off, obj_req->ex.oe_len, result);
-	if (!img_req->result)
-		img_req->result = result;
+		if (__rbd_obj_handle_request(obj_req, &result)) {
+			if (result) {
+				img_req->pending.result = result;
+				return;
+			}
+		} else {
+			img_req->pending.num_pending++;
+		}
+	}
 }
 
-static void rbd_img_end_request(struct rbd_img_request *img_req)
+static bool rbd_img_advance(struct rbd_img_request *img_req, int *result)
 {
-	rbd_assert(!test_bit(IMG_REQ_CHILD, &img_req->flags));
+again:
+	switch (img_req->state) {
+	case RBD_IMG_START:
+		rbd_assert(!*result);
 
-	blk_mq_end_request(img_req->rq,
-			   errno_to_blk_status(img_req->result));
-	rbd_img_request_put(img_req);
+		rbd_img_object_requests(img_req);
+		if (!img_req->pending.num_pending) {
+			*result = img_req->pending.result;
+			img_req->state = RBD_IMG_OBJECT_REQUESTS;
+			goto again;
+		}
+		img_req->state = __RBD_IMG_OBJECT_REQUESTS;
+		return false;
+	case __RBD_IMG_OBJECT_REQUESTS:
+		if (!pending_result_dec(&img_req->pending, result))
+			return false;
+		/* fall through */
+	case RBD_IMG_OBJECT_REQUESTS:
+		return true;
+	default:
+		BUG();
+	}
 }
 
-static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result)
+/*
+ * Return true if @img_req is completed.
+ */
+static bool __rbd_img_handle_request(struct rbd_img_request *img_req,
+				     int *result)
 {
-	struct rbd_img_request *img_req;
+	struct rbd_device *rbd_dev = img_req->rbd_dev;
+	bool done;
 
-again:
-	if (!__rbd_obj_handle_request(obj_req, &result))
-		return;
+	mutex_lock(&img_req->state_mutex);
+	done = rbd_img_advance(img_req, result);
+	mutex_unlock(&img_req->state_mutex);
 
-	img_req = obj_req->img_request;
-	spin_lock(&img_req->completion_lock);
-	rbd_obj_end_request(obj_req, result);
-	rbd_assert(img_req->pending_count);
-	if (--img_req->pending_count) {
-		spin_unlock(&img_req->completion_lock);
-		return;
+	if (done && *result) {
+		rbd_assert(*result < 0);
+		rbd_warn(rbd_dev, "%s%s result %d",
+		      test_bit(IMG_REQ_CHILD, &img_req->flags) ? "child " : "",
+		      obj_op_name(img_req->op_type), *result);
 	}
+	return done;
+}
+
+static void rbd_img_handle_request(struct rbd_img_request *img_req, int result)
+{
+again:
+	if (!__rbd_img_handle_request(img_req, &result))
+		return;
 
-	spin_unlock(&img_req->completion_lock);
-	rbd_assert(img_req->result <= 0);
 	if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
-		obj_req = img_req->obj_request;
-		result = img_req->result;
+		struct rbd_obj_request *obj_req = img_req->obj_request;
+
 		rbd_img_request_put(img_req);
-		goto again;
+		if (__rbd_obj_handle_request(obj_req, &result)) {
+			img_req = obj_req->img_request;
+			goto again;
+		}
+	} else {
+		struct request *rq = img_req->rq;
+
+		rbd_img_request_put(img_req);
+		blk_mq_end_request(rq, errno_to_blk_status(result));
 	}
-	rbd_img_end_request(img_req);
 }
 
 static const struct rbd_client_id rbd_empty_cid;
@@ -3933,10 +4013,10 @@ static void rbd_queue_workfn(struct work_struct *work)
 	else
 		result = rbd_img_fill_from_bio(img_request, offset, length,
 					       rq->bio);
-	if (result || !img_request->pending_count)
+	if (result)
 		goto err_img_request;
 
-	rbd_img_request_submit(img_request);
+	rbd_img_handle_request(img_request, 0);
 	if (must_be_locked)
 		up_read(&rbd_dev->lock_rwsem);
 	return;
-- 
cgit v1.2.3


From bcbab1db6c95a9abe7793c186f87fd41f1b8d4ea Mon Sep 17 00:00:00 2001
From: Ilya Dryomov <idryomov@gmail.com>
Date: Mon, 27 May 2019 11:41:36 +0200
Subject: rbd: introduce obj_req->osd_reqs list

Since the dawn of time it had been assumed that a single object request
spawns a single OSD request.  This is already impacting copyup: instead
of sending empty and current snapc copyups together, we wait for empty
snapc OSD request to complete in order to reassign obj_req->osd_req
with current snapc OSD request.  Looking further, updating potentially
hundreds of snapshot object maps serially is a non-starter.

Replace obj_req->osd_req pointer with obj_req->osd_reqs list.  Use
osd_req->r_private_item for linkage.

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>
Reviewed-by: Dongsheng Yang <dongsheng.yang@easystack.cn>
---
 drivers/block/rbd.c | 191 +++++++++++++++++++++++++++-------------------------
 1 file changed, 100 insertions(+), 91 deletions(-)

(limited to 'drivers/block')

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 51dd1b99c242..c6e0d4ace8ac 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -289,7 +289,7 @@ struct rbd_obj_request {
 	struct bio_vec		*copyup_bvecs;
 	u32			copyup_bvec_count;
 
-	struct ceph_osd_request	*osd_req;
+	struct list_head	osd_reqs;	/* w/ r_private_item */
 
 	struct mutex		state_mutex;
 	struct kref		kref;
@@ -1410,7 +1410,9 @@ static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
 
 static void rbd_obj_request_submit(struct rbd_obj_request *obj_request)
 {
-	struct ceph_osd_request *osd_req = obj_request->osd_req;
+	struct ceph_osd_request *osd_req =
+	    list_last_entry(&obj_request->osd_reqs, struct ceph_osd_request,
+			    r_private_item);
 
 	dout("%s %p object_no %016llx %llu~%llu osd_req %p\n", __func__,
 	     obj_request, obj_request->ex.oe_objno, obj_request->ex.oe_off,
@@ -1497,7 +1499,6 @@ static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
 
 	dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
 	     osd_req->r_result, obj_req);
-	rbd_assert(osd_req == obj_req->osd_req);
 
 	/*
 	 * Writes aren't allowed to return a data payload.  In some
@@ -1512,17 +1513,17 @@ static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
 	rbd_obj_handle_request(obj_req, result);
 }
 
-static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request)
+static void rbd_osd_format_read(struct ceph_osd_request *osd_req)
 {
-	struct ceph_osd_request *osd_req = obj_request->osd_req;
+	struct rbd_obj_request *obj_request = osd_req->r_priv;
 
 	osd_req->r_flags = CEPH_OSD_FLAG_READ;
 	osd_req->r_snapid = obj_request->img_request->snap_id;
 }
 
-static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request)
+static void rbd_osd_format_write(struct ceph_osd_request *osd_req)
 {
-	struct ceph_osd_request *osd_req = obj_request->osd_req;
+	struct rbd_obj_request *obj_request = osd_req->r_priv;
 
 	osd_req->r_flags = CEPH_OSD_FLAG_WRITE;
 	ktime_get_real_ts64(&osd_req->r_mtime);
@@ -1530,19 +1531,21 @@ static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request)
 }
 
 static struct ceph_osd_request *
-__rbd_osd_req_create(struct rbd_obj_request *obj_req,
-		     struct ceph_snap_context *snapc, unsigned int num_ops)
+__rbd_obj_add_osd_request(struct rbd_obj_request *obj_req,
+			  struct ceph_snap_context *snapc, int num_ops)
 {
 	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
 	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
 	struct ceph_osd_request *req;
 	const char *name_format = rbd_dev->image_format == 1 ?
 				      RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
+	int ret;
 
 	req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, GFP_NOIO);
 	if (!req)
-		return NULL;
+		return ERR_PTR(-ENOMEM);
 
+	list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
 	req->r_callback = rbd_osd_req_callback;
 	req->r_priv = obj_req;
 
@@ -1553,27 +1556,20 @@ __rbd_osd_req_create(struct rbd_obj_request *obj_req,
 	ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
 	req->r_base_oloc.pool = rbd_dev->layout.pool_id;
 
-	if (ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
-			rbd_dev->header.object_prefix, obj_req->ex.oe_objno))
-		goto err_req;
+	ret = ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
+			       rbd_dev->header.object_prefix,
+			       obj_req->ex.oe_objno);
+	if (ret)
+		return ERR_PTR(ret);
 
 	return req;
-
-err_req:
-	ceph_osdc_put_request(req);
-	return NULL;
 }
 
 static struct ceph_osd_request *
-rbd_osd_req_create(struct rbd_obj_request *obj_req, unsigned int num_ops)
+rbd_obj_add_osd_request(struct rbd_obj_request *obj_req, int num_ops)
 {
-	return __rbd_osd_req_create(obj_req, obj_req->img_request->snapc,
-				    num_ops);
-}
-
-static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req)
-{
-	ceph_osdc_put_request(osd_req);
+	return __rbd_obj_add_osd_request(obj_req, obj_req->img_request->snapc,
+					 num_ops);
 }
 
 static struct rbd_obj_request *rbd_obj_request_create(void)
@@ -1585,6 +1581,7 @@ static struct rbd_obj_request *rbd_obj_request_create(void)
 		return NULL;
 
 	ceph_object_extent_init(&obj_request->ex);
+	INIT_LIST_HEAD(&obj_request->osd_reqs);
 	mutex_init(&obj_request->state_mutex);
 	kref_init(&obj_request->kref);
 
@@ -1595,14 +1592,19 @@ static struct rbd_obj_request *rbd_obj_request_create(void)
 static void rbd_obj_request_destroy(struct kref *kref)
 {
 	struct rbd_obj_request *obj_request;
+	struct ceph_osd_request *osd_req;
 	u32 i;
 
 	obj_request = container_of(kref, struct rbd_obj_request, kref);
 
 	dout("%s: obj %p\n", __func__, obj_request);
 
-	if (obj_request->osd_req)
-		rbd_osd_req_destroy(obj_request->osd_req);
+	while (!list_empty(&obj_request->osd_reqs)) {
+		osd_req = list_first_entry(&obj_request->osd_reqs,
+				    struct ceph_osd_request, r_private_item);
+		list_del_init(&osd_req->r_private_item);
+		ceph_osdc_put_request(osd_req);
+	}
 
 	switch (obj_request->img_request->data_type) {
 	case OBJ_REQUEST_NODATA:
@@ -1796,11 +1798,13 @@ static int rbd_obj_calc_img_extents(struct rbd_obj_request *obj_req,
 	return 0;
 }
 
-static void rbd_osd_req_setup_data(struct rbd_obj_request *obj_req, u32 which)
+static void rbd_osd_setup_data(struct ceph_osd_request *osd_req, int which)
 {
+	struct rbd_obj_request *obj_req = osd_req->r_priv;
+
 	switch (obj_req->img_request->data_type) {
 	case OBJ_REQUEST_BIO:
-		osd_req_op_extent_osd_data_bio(obj_req->osd_req, which,
+		osd_req_op_extent_osd_data_bio(osd_req, which,
 					       &obj_req->bio_pos,
 					       obj_req->ex.oe_len);
 		break;
@@ -1809,7 +1813,7 @@ static void rbd_osd_req_setup_data(struct rbd_obj_request *obj_req, u32 which)
 		rbd_assert(obj_req->bvec_pos.iter.bi_size ==
 							obj_req->ex.oe_len);
 		rbd_assert(obj_req->bvec_idx == obj_req->bvec_count);
-		osd_req_op_extent_osd_data_bvec_pos(obj_req->osd_req, which,
+		osd_req_op_extent_osd_data_bvec_pos(osd_req, which,
 						    &obj_req->bvec_pos);
 		break;
 	default:
@@ -1819,21 +1823,22 @@ static void rbd_osd_req_setup_data(struct rbd_obj_request *obj_req, u32 which)
 
 static int rbd_obj_setup_read(struct rbd_obj_request *obj_req)
 {
-	obj_req->osd_req = __rbd_osd_req_create(obj_req, NULL, 1);
-	if (!obj_req->osd_req)
-		return -ENOMEM;
+	struct ceph_osd_request *osd_req;
 
-	osd_req_op_extent_init(obj_req->osd_req, 0, CEPH_OSD_OP_READ,
+	osd_req = __rbd_obj_add_osd_request(obj_req, NULL, 1);
+	if (IS_ERR(osd_req))
+		return PTR_ERR(osd_req);
+
+	osd_req_op_extent_init(osd_req, 0, CEPH_OSD_OP_READ,
 			       obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
-	rbd_osd_req_setup_data(obj_req, 0);
+	rbd_osd_setup_data(osd_req, 0);
 
-	rbd_osd_req_format_read(obj_req);
+	rbd_osd_format_read(osd_req);
 	obj_req->read_state = RBD_OBJ_READ_START;
 	return 0;
 }
 
-static int __rbd_obj_setup_stat(struct rbd_obj_request *obj_req,
-				unsigned int which)
+static int rbd_osd_setup_stat(struct ceph_osd_request *osd_req, int which)
 {
 	struct page **pages;
 
@@ -1849,8 +1854,8 @@ static int __rbd_obj_setup_stat(struct rbd_obj_request *obj_req,
 	if (IS_ERR(pages))
 		return PTR_ERR(pages);
 
-	osd_req_op_init(obj_req->osd_req, which, CEPH_OSD_OP_STAT, 0);
-	osd_req_op_raw_data_in_pages(obj_req->osd_req, which, pages,
+	osd_req_op_init(osd_req, which, CEPH_OSD_OP_STAT, 0);
+	osd_req_op_raw_data_in_pages(osd_req, which, pages,
 				     8 + sizeof(struct ceph_timespec),
 				     0, false, true);
 	return 0;
@@ -1861,13 +1866,14 @@ static int count_write_ops(struct rbd_obj_request *obj_req)
 	return 2; /* setallochint + write/writefull */
 }
 
-static void __rbd_obj_setup_write(struct rbd_obj_request *obj_req,
-				  unsigned int which)
+static void __rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
+				      int which)
 {
+	struct rbd_obj_request *obj_req = osd_req->r_priv;
 	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
 	u16 opcode;
 
-	osd_req_op_alloc_hint_init(obj_req->osd_req, which++,
+	osd_req_op_alloc_hint_init(osd_req, which++,
 				   rbd_dev->layout.object_size,
 				   rbd_dev->layout.object_size);
 
@@ -1876,16 +1882,16 @@ static void __rbd_obj_setup_write(struct rbd_obj_request *obj_req,
 	else
 		opcode = CEPH_OSD_OP_WRITE;
 
-	osd_req_op_extent_init(obj_req->osd_req, which, opcode,
+	osd_req_op_extent_init(osd_req, which, opcode,
 			       obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
-	rbd_osd_req_setup_data(obj_req, which++);
+	rbd_osd_setup_data(osd_req, which);
 
-	rbd_assert(which == obj_req->osd_req->r_num_ops);
-	rbd_osd_req_format_write(obj_req);
+	rbd_osd_format_write(osd_req);
 }
 
 static int rbd_obj_setup_write(struct rbd_obj_request *obj_req)
 {
+	struct ceph_osd_request *osd_req;
 	unsigned int num_osd_ops, which = 0;
 	int ret;
 
@@ -1901,18 +1907,18 @@ static int rbd_obj_setup_write(struct rbd_obj_request *obj_req)
 	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
 		num_osd_ops++; /* stat */
 
-	obj_req->osd_req = rbd_osd_req_create(obj_req, num_osd_ops);
-	if (!obj_req->osd_req)
-		return -ENOMEM;
+	osd_req = rbd_obj_add_osd_request(obj_req, num_osd_ops);
+	if (IS_ERR(osd_req))
+		return PTR_ERR(osd_req);
 
 	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
-		ret = __rbd_obj_setup_stat(obj_req, which++);
+		ret = rbd_osd_setup_stat(osd_req, which++);
 		if (ret)
 			return ret;
 	}
 
 	obj_req->write_state = RBD_OBJ_WRITE_START;
-	__rbd_obj_setup_write(obj_req, which);
+	__rbd_osd_setup_write_ops(osd_req, which);
 	return 0;
 }
 
@@ -1925,6 +1931,7 @@ static u16 truncate_or_zero_opcode(struct rbd_obj_request *obj_req)
 static int rbd_obj_setup_discard(struct rbd_obj_request *obj_req)
 {
 	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
+	struct ceph_osd_request *osd_req;
 	u64 off = obj_req->ex.oe_off;
 	u64 next_off = obj_req->ex.oe_off + obj_req->ex.oe_len;
 	int ret;
@@ -1953,24 +1960,24 @@ static int rbd_obj_setup_discard(struct rbd_obj_request *obj_req)
 	if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents)
 		obj_req->flags |= RBD_OBJ_FLAG_DELETION;
 
-	obj_req->osd_req = rbd_osd_req_create(obj_req, 1);
-	if (!obj_req->osd_req)
-		return -ENOMEM;
+	osd_req = rbd_obj_add_osd_request(obj_req, 1);
+	if (IS_ERR(osd_req))
+		return PTR_ERR(osd_req);
 
 	if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
 		rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
-		osd_req_op_init(obj_req->osd_req, 0, CEPH_OSD_OP_DELETE, 0);
+		osd_req_op_init(osd_req, 0, CEPH_OSD_OP_DELETE, 0);
 	} else {
 		dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
 		     obj_req, obj_req->ex.oe_off, obj_req->ex.oe_len,
 		     off, next_off - off);
-		osd_req_op_extent_init(obj_req->osd_req, 0,
+		osd_req_op_extent_init(osd_req, 0,
 				       truncate_or_zero_opcode(obj_req),
 				       off, next_off - off, 0, 0);
 	}
 
 	obj_req->write_state = RBD_OBJ_WRITE_START;
-	rbd_osd_req_format_write(obj_req);
+	rbd_osd_format_write(osd_req);
 	return 0;
 }
 
@@ -1987,20 +1994,21 @@ static int count_zeroout_ops(struct rbd_obj_request *obj_req)
 	return num_osd_ops;
 }
 
-static void __rbd_obj_setup_zeroout(struct rbd_obj_request *obj_req,
-				    unsigned int which)
+static void __rbd_osd_setup_zeroout_ops(struct ceph_osd_request *osd_req,
+					int which)
 {
+	struct rbd_obj_request *obj_req = osd_req->r_priv;
 	u16 opcode;
 
 	if (rbd_obj_is_entire(obj_req)) {
 		if (obj_req->num_img_extents) {
 			if (!(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
-				osd_req_op_init(obj_req->osd_req, which++,
+				osd_req_op_init(osd_req, which++,
 						CEPH_OSD_OP_CREATE, 0);
 			opcode = CEPH_OSD_OP_TRUNCATE;
 		} else {
 			rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
-			osd_req_op_init(obj_req->osd_req, which++,
+			osd_req_op_init(osd_req, which++,
 					CEPH_OSD_OP_DELETE, 0);
 			opcode = 0;
 		}
@@ -2009,16 +2017,16 @@ static void __rbd_obj_setup_zeroout(struct rbd_obj_request *obj_req,
 	}
 
 	if (opcode)
-		osd_req_op_extent_init(obj_req->osd_req, which++, opcode,
+		osd_req_op_extent_init(osd_req, which, opcode,
 				       obj_req->ex.oe_off, obj_req->ex.oe_len,
 				       0, 0);
 
-	rbd_assert(which == obj_req->osd_req->r_num_ops);
-	rbd_osd_req_format_write(obj_req);
+	rbd_osd_format_write(osd_req);
 }
 
 static int rbd_obj_setup_zeroout(struct rbd_obj_request *obj_req)
 {
+	struct ceph_osd_request *osd_req;
 	unsigned int num_osd_ops, which = 0;
 	int ret;
 
@@ -2038,18 +2046,18 @@ static int rbd_obj_setup_zeroout(struct rbd_obj_request *obj_req)
 	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
 		num_osd_ops++; /* stat */
 
-	obj_req->osd_req = rbd_osd_req_create(obj_req, num_osd_ops);
-	if (!obj_req->osd_req)
-		return -ENOMEM;
+	osd_req = rbd_obj_add_osd_request(obj_req, num_osd_ops);
+	if (IS_ERR(osd_req))
+		return PTR_ERR(osd_req);
 
 	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
-		ret = __rbd_obj_setup_stat(obj_req, which++);
+		ret = rbd_osd_setup_stat(osd_req, which++);
 		if (ret)
 			return ret;
 	}
 
 	obj_req->write_state = RBD_OBJ_WRITE_START;
-	__rbd_obj_setup_zeroout(obj_req, which);
+	__rbd_osd_setup_zeroout_ops(osd_req, which);
 	return 0;
 }
 
@@ -2061,6 +2069,7 @@ static int rbd_obj_setup_zeroout(struct rbd_obj_request *obj_req)
 static int __rbd_img_fill_request(struct rbd_img_request *img_req)
 {
 	struct rbd_obj_request *obj_req, *next_obj_req;
+	struct ceph_osd_request *osd_req;
 	int ret;
 
 	for_each_obj_request_safe(img_req, obj_req, next_obj_req) {
@@ -2087,7 +2096,10 @@ static int __rbd_img_fill_request(struct rbd_img_request *img_req)
 			continue;
 		}
 
-		ret = ceph_osdc_alloc_messages(obj_req->osd_req, GFP_NOIO);
+		osd_req = list_last_entry(&obj_req->osd_reqs,
+					  struct ceph_osd_request,
+					  r_private_item);
+		ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
 		if (ret)
 			return ret;
 	}
@@ -2538,28 +2550,27 @@ static bool is_zero_bvecs(struct bio_vec *bvecs, u32 bytes)
 static int rbd_obj_issue_copyup_empty_snapc(struct rbd_obj_request *obj_req,
 					    u32 bytes)
 {
+	struct ceph_osd_request *osd_req;
 	int ret;
 
 	dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
-	rbd_assert(obj_req->osd_req->r_ops[0].op == CEPH_OSD_OP_STAT);
 	rbd_assert(bytes > 0 && bytes != MODS_ONLY);
-	rbd_osd_req_destroy(obj_req->osd_req);
 
-	obj_req->osd_req = __rbd_osd_req_create(obj_req, &rbd_empty_snapc, 1);
-	if (!obj_req->osd_req)
-		return -ENOMEM;
+	osd_req = __rbd_obj_add_osd_request(obj_req, &rbd_empty_snapc, 1);
+	if (IS_ERR(osd_req))
+		return PTR_ERR(osd_req);
 
-	ret = osd_req_op_cls_init(obj_req->osd_req, 0, "rbd", "copyup");
+	ret = osd_req_op_cls_init(osd_req, 0, "rbd", "copyup");
 	if (ret)
 		return ret;
 
-	osd_req_op_cls_request_data_bvecs(obj_req->osd_req, 0,
+	osd_req_op_cls_request_data_bvecs(osd_req, 0,
 					  obj_req->copyup_bvecs,
 					  obj_req->copyup_bvec_count,
 					  bytes);
-	rbd_osd_req_format_write(obj_req);
+	rbd_osd_format_write(osd_req);
 
-	ret = ceph_osdc_alloc_messages(obj_req->osd_req, GFP_NOIO);
+	ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
 	if (ret)
 		return ret;
 
@@ -2570,14 +2581,12 @@ static int rbd_obj_issue_copyup_empty_snapc(struct rbd_obj_request *obj_req,
 static int rbd_obj_issue_copyup_ops(struct rbd_obj_request *obj_req, u32 bytes)
 {
 	struct rbd_img_request *img_req = obj_req->img_request;
+	struct ceph_osd_request *osd_req;
 	unsigned int num_osd_ops = (bytes != MODS_ONLY);
 	unsigned int which = 0;
 	int ret;
 
 	dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
-	rbd_assert(obj_req->osd_req->r_ops[0].op == CEPH_OSD_OP_STAT ||
-		   obj_req->osd_req->r_ops[0].op == CEPH_OSD_OP_CALL);
-	rbd_osd_req_destroy(obj_req->osd_req);
 
 	switch (img_req->op_type) {
 	case OBJ_OP_WRITE:
@@ -2590,17 +2599,17 @@ static int rbd_obj_issue_copyup_ops(struct rbd_obj_request *obj_req, u32 bytes)
 		BUG();
 	}
 
-	obj_req->osd_req = rbd_osd_req_create(obj_req, num_osd_ops);
-	if (!obj_req->osd_req)
-		return -ENOMEM;
+	osd_req = rbd_obj_add_osd_request(obj_req, num_osd_ops);
+	if (IS_ERR(osd_req))
+		return PTR_ERR(osd_req);
 
 	if (bytes != MODS_ONLY) {
-		ret = osd_req_op_cls_init(obj_req->osd_req, which, "rbd",
+		ret = osd_req_op_cls_init(osd_req, which, "rbd",
 					  "copyup");
 		if (ret)
 			return ret;
 
-		osd_req_op_cls_request_data_bvecs(obj_req->osd_req, which++,
+		osd_req_op_cls_request_data_bvecs(osd_req, which++,
 						  obj_req->copyup_bvecs,
 						  obj_req->copyup_bvec_count,
 						  bytes);
@@ -2608,16 +2617,16 @@ static int rbd_obj_issue_copyup_ops(struct rbd_obj_request *obj_req, u32 bytes)
 
 	switch (img_req->op_type) {
 	case OBJ_OP_WRITE:
-		__rbd_obj_setup_write(obj_req, which);
+		__rbd_osd_setup_write_ops(osd_req, which);
 		break;
 	case OBJ_OP_ZEROOUT:
-		__rbd_obj_setup_zeroout(obj_req, which);
+		__rbd_osd_setup_zeroout_ops(osd_req, which);
 		break;
 	default:
 		BUG();
 	}
 
-	ret = ceph_osdc_alloc_messages(obj_req->osd_req, GFP_NOIO);
+	ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
 	if (ret)
 		return ret;
 
-- 
cgit v1.2.3


From b5ae8cbc6e3746c05a5ffe59cb4c03fee1730764 Mon Sep 17 00:00:00 2001
From: Ilya Dryomov <idryomov@gmail.com>
Date: Wed, 29 May 2019 16:53:14 +0200
Subject: rbd: factor out rbd_osd_setup_copyup()

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>
Reviewed-by: Dongsheng Yang <dongsheng.yang@easystack.cn>
---
 drivers/block/rbd.c | 29 +++++++++++++++++------------
 1 file changed, 17 insertions(+), 12 deletions(-)

(limited to 'drivers/block')

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index c6e0d4ace8ac..dd63dbcefdc5 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -1861,6 +1861,21 @@ static int rbd_osd_setup_stat(struct ceph_osd_request *osd_req, int which)
 	return 0;
 }
 
+static int rbd_osd_setup_copyup(struct ceph_osd_request *osd_req, int which,
+				u32 bytes)
+{
+	struct rbd_obj_request *obj_req = osd_req->r_priv;
+	int ret;
+
+	ret = osd_req_op_cls_init(osd_req, which, "rbd", "copyup");
+	if (ret)
+		return ret;
+
+	osd_req_op_cls_request_data_bvecs(osd_req, which, obj_req->copyup_bvecs,
+					  obj_req->copyup_bvec_count, bytes);
+	return 0;
+}
+
 static int count_write_ops(struct rbd_obj_request *obj_req)
 {
 	return 2; /* setallochint + write/writefull */
@@ -2560,14 +2575,10 @@ static int rbd_obj_issue_copyup_empty_snapc(struct rbd_obj_request *obj_req,
 	if (IS_ERR(osd_req))
 		return PTR_ERR(osd_req);
 
-	ret = osd_req_op_cls_init(osd_req, 0, "rbd", "copyup");
+	ret = rbd_osd_setup_copyup(osd_req, 0, bytes);
 	if (ret)
 		return ret;
 
-	osd_req_op_cls_request_data_bvecs(osd_req, 0,
-					  obj_req->copyup_bvecs,
-					  obj_req->copyup_bvec_count,
-					  bytes);
 	rbd_osd_format_write(osd_req);
 
 	ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
@@ -2604,15 +2615,9 @@ static int rbd_obj_issue_copyup_ops(struct rbd_obj_request *obj_req, u32 bytes)
 		return PTR_ERR(osd_req);
 
 	if (bytes != MODS_ONLY) {
-		ret = osd_req_op_cls_init(osd_req, which, "rbd",
-					  "copyup");
+		ret = rbd_osd_setup_copyup(osd_req, which++, bytes);
 		if (ret)
 			return ret;
-
-		osd_req_op_cls_request_data_bvecs(osd_req, which++,
-						  obj_req->copyup_bvecs,
-						  obj_req->copyup_bvec_count,
-						  bytes);
 	}
 
 	switch (img_req->op_type) {
-- 
cgit v1.2.3


From 27bbd911624891ccb17980ff784d68e6f470b92b Mon Sep 17 00:00:00 2001
From: Ilya Dryomov <idryomov@gmail.com>
Date: Wed, 29 May 2019 17:31:37 +0200
Subject: rbd: factor out __rbd_osd_setup_discard_ops()

With obj_req->xferred removed, obj_req->ex.oe_off and obj_req->ex.oe_len
can be updated if required for alignment.  Previously the new offset and
length weren't stored anywhere beyond rbd_obj_setup_discard().

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>
Reviewed-by: Dongsheng Yang <dongsheng.yang@easystack.cn>
---
 drivers/block/rbd.c | 43 +++++++++++++++++++++++++++----------------
 1 file changed, 27 insertions(+), 16 deletions(-)

(limited to 'drivers/block')

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index dd63dbcefdc5..dd7dcf12426f 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -1943,12 +1943,27 @@ static u16 truncate_or_zero_opcode(struct rbd_obj_request *obj_req)
 					  CEPH_OSD_OP_ZERO;
 }
 
+static void __rbd_osd_setup_discard_ops(struct ceph_osd_request *osd_req,
+					int which)
+{
+	struct rbd_obj_request *obj_req = osd_req->r_priv;
+
+	if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
+		rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
+		osd_req_op_init(osd_req, which, CEPH_OSD_OP_DELETE, 0);
+	} else {
+		osd_req_op_extent_init(osd_req, which,
+				       truncate_or_zero_opcode(obj_req),
+				       obj_req->ex.oe_off, obj_req->ex.oe_len,
+				       0, 0);
+	}
+}
+
 static int rbd_obj_setup_discard(struct rbd_obj_request *obj_req)
 {
 	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
 	struct ceph_osd_request *osd_req;
-	u64 off = obj_req->ex.oe_off;
-	u64 next_off = obj_req->ex.oe_off + obj_req->ex.oe_len;
+	u64 off, next_off;
 	int ret;
 
 	/*
@@ -1961,10 +1976,17 @@ static int rbd_obj_setup_discard(struct rbd_obj_request *obj_req)
 	 */
 	if (rbd_dev->opts->alloc_size != rbd_dev->layout.object_size ||
 	    !rbd_obj_is_tail(obj_req)) {
-		off = round_up(off, rbd_dev->opts->alloc_size);
-		next_off = round_down(next_off, rbd_dev->opts->alloc_size);
+		off = round_up(obj_req->ex.oe_off, rbd_dev->opts->alloc_size);
+		next_off = round_down(obj_req->ex.oe_off + obj_req->ex.oe_len,
+				      rbd_dev->opts->alloc_size);
 		if (off >= next_off)
 			return 1;
+
+		dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
+		     obj_req, obj_req->ex.oe_off, obj_req->ex.oe_len,
+		     off, next_off - off);
+		obj_req->ex.oe_off = off;
+		obj_req->ex.oe_len = next_off - off;
 	}
 
 	/* reverse map the entire object onto the parent */
@@ -1979,19 +2001,8 @@ static int rbd_obj_setup_discard(struct rbd_obj_request *obj_req)
 	if (IS_ERR(osd_req))
 		return PTR_ERR(osd_req);
 
-	if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
-		rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
-		osd_req_op_init(osd_req, 0, CEPH_OSD_OP_DELETE, 0);
-	} else {
-		dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
-		     obj_req, obj_req->ex.oe_off, obj_req->ex.oe_len,
-		     off, next_off - off);
-		osd_req_op_extent_init(osd_req, 0,
-				       truncate_or_zero_opcode(obj_req),
-				       off, next_off - off, 0, 0);
-	}
-
 	obj_req->write_state = RBD_OBJ_WRITE_START;
+	__rbd_osd_setup_discard_ops(osd_req, 0);
 	rbd_osd_format_write(osd_req);
 	return 0;
 }
-- 
cgit v1.2.3


From a086a1b8bdbd14a59b8a4cd979d5c7894e3af59e Mon Sep 17 00:00:00 2001
From: Ilya Dryomov <idryomov@gmail.com>
Date: Wed, 12 Jun 2019 18:33:31 +0200
Subject: rbd: move OSD request allocation into object request state machines

Following submission, move initial OSD request allocation into object
request state machines.  Everything that has to do with OSD requests is
now handled inside the state machine, all __rbd_img_fill_request() has
left is initialization.

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