diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 63f73e8..f008b8e 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -376,6 +376,8 @@ struct rbd_device {
 	atomic_t		parent_ref;
 	struct rbd_device	*parent;
 
+	atomic_t		budget_reserved;
+
 	/* Block layer tags. */
 	struct blk_mq_tag_set	tag_set;
 
@@ -734,6 +736,7 @@ static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
 enum {
 	Opt_queue_depth,
 	Opt_lock_timeout,
+	Opt_busy_threshold,
 	Opt_last_int,
 	/* int args above */
 	Opt_pool_ns,
@@ -750,6 +753,7 @@ enum {
 static match_table_t rbd_opts_tokens = {
 	{Opt_queue_depth, "queue_depth=%d"},
 	{Opt_lock_timeout, "lock_timeout=%d"},
+	{Opt_busy_threshold, "busy_threshold=%d"},
 	/* int args above */
 	{Opt_pool_ns, "_pool_ns=%s"},
 	/* string args above */
@@ -765,6 +769,7 @@ enum {
 
 struct rbd_options {
 	int	queue_depth;
+	unsigned long	busy_threshold;
 	unsigned long	lock_timeout;
 	bool	read_only;
 	bool	lock_on_read;
@@ -773,6 +778,7 @@ struct rbd_options {
 };
 
 #define RBD_QUEUE_DEPTH_DEFAULT	BLKDEV_MAX_RQ
+#define RBD_BUSY_THRESHOLD_DEFAULT	0 /* never be busy */
 #define RBD_LOCK_TIMEOUT_DEFAULT 0  /* no timeout */
 #define RBD_READ_ONLY_DEFAULT	false
 #define RBD_LOCK_ON_READ_DEFAULT false
@@ -820,6 +826,16 @@ static int parse_rbd_opts_token(char *c, void *private)
 		}
 		pctx->opts->lock_timeout = msecs_to_jiffies(intval * 1000);
 		break;
+	case Opt_busy_threshold:
+		/* 0 means .get_budget in blk_mq_ops will always return true.
+		 * Then request will dequeue from SQ immediately.
+		 */
+		if (intval < 0) {
+			pr_err("busy_threashold out of range\n");
+			return -EINVAL;
+		}
+		pctx->opts->busy_threshold = intval;
+		break;
 	case Opt_pool_ns:
 		kfree(pctx->spec->pool_ns);
 		pctx->spec->pool_ns = match_strdup(argstr);
@@ -2590,6 +2606,7 @@ static void rbd_img_end_request(struct rbd_img_request *img_req)
 		    img_req->xferred == blk_rq_bytes(img_req->rq)) ||
 		   (img_req->result < 0 && !img_req->xferred));
 
+	atomic_dec(&img_req->rbd_dev->budget_reserved);
 	blk_mq_end_request(img_req->rq,
 			   errno_to_blk_status(img_req->result));
 	rbd_img_request_put(img_req);
@@ -3747,6 +3764,7 @@ static void rbd_queue_workfn(struct work_struct *work)
 			 obj_op_name(op_type), length, offset, result);
 	ceph_put_snap_context(snapc);
 err:
+	atomic_dec(&rbd_dev->budget_reserved);
 	blk_mq_end_request(rq, errno_to_blk_status(result));
 }
 
@@ -3955,9 +3973,41 @@ static int rbd_init_request(struct blk_mq_tag_set *set, struct request *rq,
 	return 0;
 }
 
+static bool rbd_get_budget(struct blk_mq_hw_ctx *hctx)
+{
+	struct request_queue *q = hctx->queue;
+	struct rbd_device *rbd_dev = q->queuedata;
+	unsigned long busy_threshold = rbd_dev->opts->busy_threshold;
+
+	if (!busy_threshold)
+		return true;
+
+	if (atomic_inc_return(&rbd_dev->budget_reserved) <= busy_threshold)
+		return true;
+
+	atomic_dec(&rbd_dev->budget_reserved);
+	/* use 3ms as same with BLK_MQ_RESOURCE_DELAY */
+	blk_mq_delay_run_hw_queue(hctx, 3);
+	return false;
+}
+
+static void rbd_put_budget(struct blk_mq_hw_ctx *hctx)
+{
+	struct request_queue *q = hctx->queue;
+	struct rbd_device *rbd_dev = q->queuedata;
+	unsigned long busy_threshold = rbd_dev->opts->busy_threshold;
+
+	if (!busy_threshold)
+		return;
+
+	atomic_dec(&rbd_dev->budget_reserved);
+}
+
 static const struct blk_mq_ops rbd_mq_ops = {
 	.queue_rq	= rbd_queue_rq,
 	.init_request	= rbd_init_request,
+	.get_budget	= rbd_get_budget,
+	.put_budget	= rbd_put_budget,
 };
 
 static int rbd_init_disk(struct rbd_device *rbd_dev)
@@ -5390,6 +5440,7 @@ static int rbd_add_parse_args(const char *buf,
 	pctx.opts->read_only = RBD_READ_ONLY_DEFAULT;
 	pctx.opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
 	pctx.opts->lock_timeout = RBD_LOCK_TIMEOUT_DEFAULT;
+	pctx.opts->busy_threshold = RBD_BUSY_THRESHOLD_DEFAULT;
 	pctx.opts->lock_on_read = RBD_LOCK_ON_READ_DEFAULT;
 	pctx.opts->exclusive = RBD_EXCLUSIVE_DEFAULT;
 	pctx.opts->trim = RBD_TRIM_DEFAULT;
@@ -5630,6 +5681,7 @@ static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
 
 	rbd_dev->parent = parent;
 	atomic_set(&rbd_dev->parent_ref, 1);
+	atomic_set(&rbd_dev->budget_reserved, 0);
 	return 0;
 
 out_err:
