nvme: Queue aborts beyond the abort command limit

Queue aborts that would exceed the abort command limit
in software as a convenience for the user.

Change-Id: I8c1f0380984cc6c0cdb453db961939a7f571b336
Signed-off-by: Ben Walker <benjamin.walker@intel.com>

lib/nvme/nvme_ctrlr.c

@@ -514,7 +514,8 @@ int
 spdk_nvme_ctrlr_reset(struct spdk_nvme_ctrlr *ctrlr)
 {
 	int rc = 0;
-	struct spdk_nvme_qpair *qpair;
+	struct spdk_nvme_qpair	*qpair;
+	struct nvme_request	*req, *tmp;
 
 	nvme_robust_mutex_lock(&ctrlr->ctrlr_lock);
 
@@ -532,6 +533,13 @@ spdk_nvme_ctrlr_reset(struct spdk_nvme_ctrlr *ctrlr)
 
 	SPDK_NOTICELOG("resetting controller\n");
 
+	/* Free all of the queued abort requests */
+	STAILQ_FOREACH_SAFE(req, &ctrlr->queued_aborts, stailq, tmp) {
+		STAILQ_REMOVE_HEAD(&ctrlr->queued_aborts, stailq);
+		nvme_free_request(req);
+		ctrlr->outstanding_aborts--;
+	}
+
 	/* Disable all queues before disabling the controller hardware. */
 	nvme_qpair_disable(ctrlr->adminq);
 	TAILQ_FOREACH(qpair, &ctrlr->active_io_qpairs, tailq) {
@@ -1313,6 +1321,8 @@ nvme_ctrlr_construct(struct spdk_nvme_ctrlr *ctrlr)
 	ctrlr->is_failed = false;
 
 	TAILQ_INIT(&ctrlr->active_io_qpairs);
+	STAILQ_INIT(&ctrlr->queued_aborts);
+	ctrlr->outstanding_aborts = 0;
 
 	rc = nvme_robust_mutex_init_recursive_shared(&ctrlr->ctrlr_lock);
 	if (rc != 0) {

lib/nvme/nvme_ctrlr_cmd.c

@@ -392,6 +392,38 @@ spdk_nvme_ctrlr_cmd_get_log_page(struct spdk_nvme_ctrlr *ctrlr, uint8_t log_page
 	return rc;
 }
 
+static void
+spdk_nvme_ctrlr_cmd_abort_cpl(void *ctx, const struct spdk_nvme_cpl *cpl)
+{
+	struct nvme_request	*req, *next, *tmp;
+	struct spdk_nvme_ctrlr	*ctrlr;
+	int			rc;
+
+	req = ctx;
+	ctrlr = (struct spdk_nvme_ctrlr *)req->user_buffer;
+
+	ctrlr->outstanding_aborts--;
+	STAILQ_FOREACH_SAFE(next, &ctrlr->queued_aborts, stailq, tmp) {
+		STAILQ_REMOVE_HEAD(&ctrlr->queued_aborts, stailq);
+		ctrlr->outstanding_aborts++;
+		rc = nvme_ctrlr_submit_admin_request(ctrlr, next);
+		if (rc < 0) {
+			SPDK_ERRLOG("Failed to submit queued abort.\n");
+			next->cpl.status.sct = SPDK_NVME_SCT_GENERIC;
+			next->cpl.status.sc = SPDK_NVME_SC_INTERNAL_DEVICE_ERROR;
+			next->cpl.status.dnr = 1;
+			next->cb_fn(next->cb_arg, &req->cpl);
+
+			nvme_free_request(next);
+		} else {
+			/* If the first abort succeeds, stop iterating. */
+			break;
+		}
+	}
+
+	req->user_cb_fn(req->user_cb_arg, cpl);
+}
+
 int
 spdk_nvme_ctrlr_cmd_abort(struct spdk_nvme_ctrlr *ctrlr, struct spdk_nvme_qpair *qpair,
 			  uint16_t cid, spdk_nvme_cmd_cb cb_fn, void *cb_arg)
@@ -408,17 +440,29 @@ spdk_nvme_ctrlr_cmd_abort(struct spdk_nvme_ctrlr *ctrlr, struct spdk_nvme_qpair
 	}
 
 	nvme_robust_mutex_lock(&ctrlr->ctrlr_lock);
-	req = nvme_allocate_request_null(cb_fn, cb_arg);
+	req = nvme_allocate_request_null(spdk_nvme_ctrlr_cmd_abort_cpl, NULL);
 	if (req == NULL) {
 		nvme_robust_mutex_unlock(&ctrlr->ctrlr_lock);
 		return -ENOMEM;
 	}
+	req->cb_arg = req;
+	req->user_cb_fn = cb_fn;
+	req->user_cb_arg = cb_arg;
+	req->user_buffer = ctrlr; /* This is a hack to get to the ctrlr in the
+				   * completion handler. */
 
 	cmd = &req->cmd;
 	cmd->opc = SPDK_NVME_OPC_ABORT;
 	cmd->cdw10 = (cid << 16) | sqid;
 
-	rc = nvme_ctrlr_submit_admin_request(ctrlr, req);
+	if (ctrlr->outstanding_aborts >= ctrlr->cdata.acl) {
+		STAILQ_INSERT_TAIL(&ctrlr->queued_aborts, req, stailq);
+		rc = 0;
+	} else {
+		ctrlr->outstanding_aborts++;
+		rc = nvme_ctrlr_submit_admin_request(ctrlr, req);
+	}
+
 	nvme_robust_mutex_unlock(&ctrlr->ctrlr_lock);
 	return rc;
 }

lib/nvme/nvme_internal.h

@@ -447,6 +447,9 @@ struct spdk_nvme_ctrlr {
 	spdk_nvme_timeout_cb		timeout_cb_fn;
 	void				*timeout_cb_arg;
 	uint64_t			timeout_ticks;
+
+	STAILQ_HEAD(, nvme_request)	queued_aborts;
+	uint32_t			outstanding_aborts;
 };
 
 struct nvme_driver {

test/lib/nvme/unit/nvme_ctrlr_cmd_c/nvme_ctrlr_cmd_ut.c

@@ -285,6 +285,12 @@ nvme_allocate_request_user_copy(void *buffer, uint32_t payload_size, spdk_nvme_c
 	return nvme_allocate_request_contig(buffer, payload_size, cb_fn, cb_arg);
 }
 
+void
+nvme_free_request(struct nvme_request *req)
+{
+	return;
+}
+
 int
 nvme_qpair_submit_request(struct spdk_nvme_qpair *qpair, struct nvme_request *req)
 {
@@ -461,6 +467,8 @@ test_abort_cmd(void)
 	struct spdk_nvme_ctrlr	ctrlr = {};
 	struct spdk_nvme_qpair	qpair = {};
 
+	STAILQ_INIT(&ctrlr.queued_aborts);
+
 	verify_fn = verify_abort_cmd;
 
 	qpair.id = abort_sqid;