git: 7381bbee29df - main - cam: Run all XPT_ASYNC ccbs in a dedicated thread

[Warner Losh]

The branch main has been updated by imp:

URL: https://cgit.FreeBSD.org/src/commit/?id=7381bbee29df959e88ec59866cf2878263e7f3b2

commit 7381bbee29df959e88ec59866cf2878263e7f3b2
Author:     Warner Losh <imp at FreeBSD.org>
AuthorDate: 2021-03-12 20:20:52 +0000
Commit:     Warner Losh <imp at FreeBSD.org>
CommitDate: 2021-03-12 20:29:42 +0000

    cam: Run all XPT_ASYNC ccbs in a dedicated thread
    
    Queue all XPT_ASYNC ccb's and run those in a new cam async thread. This thread
    is allowed to sleep for things like memory. This should allow us to make all the
    registration routines for cam periph drivers simpler since they can assume they
    can always allocate memory. This is a separate thread so that any I/O that's
    completed in xpt_done_td isn't held up.
    
    This should fix the panics for WAITOK alloations that are elsewhere in the
    storage stack that aren't so easy to convert to NOWAIT. Additional future work
    will convert other allocations in the registration path to WAITOK should
    detailed analysis show it to be safe.
    
    Reviewed by: chs@, rpokala@
    Differential Revision:  https://reviews.freebsd.org/D29210
---
 sys/cam/cam.h     |  5 +++--
 sys/cam/cam_xpt.c | 49 ++++++++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 51 insertions(+), 3 deletions(-)

diff --git a/sys/cam/cam.h b/sys/cam/cam.h
index 2aaa2e3a5775..70813f92ad10 100644
--- a/sys/cam/cam.h
+++ b/sys/cam/cam.h
@@ -94,8 +94,9 @@ typedef struct {
 	u_int32_t generation;
 	int       index;
 #define CAM_UNQUEUED_INDEX	-1
-#define CAM_ACTIVE_INDEX	-2	
-#define CAM_DONEQ_INDEX		-3	
+#define CAM_ACTIVE_INDEX	-2
+#define CAM_DONEQ_INDEX		-3
+#define CAM_ASYNC_INDEX		-4
 #define CAM_EXTRAQ_INDEX	INT_MAX
 } cam_pinfo;
 
diff --git a/sys/cam/cam_xpt.c b/sys/cam/cam_xpt.c
index 9cfee80a6049..24e16bebb2e7 100644
--- a/sys/cam/cam_xpt.c
+++ b/sys/cam/cam_xpt.c
@@ -180,6 +180,7 @@ struct cam_doneq {
 static struct cam_doneq cam_doneqs[MAXCPU];
 static u_int __read_mostly cam_num_doneqs;
 static struct proc *cam_proc;
+static struct cam_doneq cam_async;
 
 SYSCTL_INT(_kern_cam, OID_AUTO, num_doneqs, CTLFLAG_RDTUN,
            &cam_num_doneqs, 0, "Number of completion queues/threads");
@@ -271,6 +272,7 @@ static void	 xptpoll(struct cam_sim *sim);
 static void	 camisr_runqueue(void);
 static void	 xpt_done_process(struct ccb_hdr *ccb_h);
 static void	 xpt_done_td(void *);
+static void	 xpt_async_td(void *);
 static dev_match_ret	xptbusmatch(struct dev_match_pattern *patterns,
 				    u_int num_patterns, struct cam_eb *bus);
 static dev_match_ret	xptdevicematch(struct dev_match_pattern *patterns,
@@ -972,6 +974,15 @@ xpt_init(void *dummy)
 		return (ENOMEM);
 	}
 
+	mtx_init(&cam_async.cam_doneq_mtx, "CAM async", NULL, MTX_DEF);
+	STAILQ_INIT(&cam_async.cam_doneq);
+	if (kproc_kthread_add(xpt_async_td, &cam_async,
+		&cam_proc, NULL, 0, 0, "cam", "async") != 0) {
+		printf("xpt_init: Cannot init async thread "
+		       "- failing attach\n");
+		return (ENOMEM);
+	}
+
 	/*
 	 * Register a callback for when interrupts are enabled.
 	 */
@@ -3146,8 +3157,16 @@ call_sim:
 		xpt_done(start_ccb);
 		break;
 	case XPT_ASYNC:
+		/*
+		 * Queue the async operation so it can be run from a sleepable
+		 * context.
+		 */
 		start_ccb->ccb_h.status = CAM_REQ_CMP;
-		xpt_done(start_ccb);
+		mtx_lock(&cam_async.cam_doneq_mtx);
+		STAILQ_INSERT_TAIL(&cam_async.cam_doneq, &start_ccb->ccb_h, sim_links.stqe);
+		start_ccb->ccb_h.pinfo.index = CAM_ASYNC_INDEX;
+		mtx_unlock(&cam_async.cam_doneq_mtx);
+		wakeup(&cam_async.cam_doneq);
 		break;
 	default:
 	case XPT_SDEV_TYPE:
@@ -5472,6 +5491,34 @@ xpt_done_process(struct ccb_hdr *ccb_h)
 		mtx_unlock(mtx);
 }
 
+/*
+ * Parameterize instead and use xpt_done_td?
+ */
+static void
+xpt_async_td(void *arg)
+{
+	struct cam_doneq *queue = arg;
+	struct ccb_hdr *ccb_h;
+	STAILQ_HEAD(, ccb_hdr)	doneq;
+
+	STAILQ_INIT(&doneq);
+	mtx_lock(&queue->cam_doneq_mtx);
+	while (1) {
+		while (STAILQ_EMPTY(&queue->cam_doneq))
+			msleep(&queue->cam_doneq, &queue->cam_doneq_mtx,
+			    PRIBIO, "-", 0);
+		STAILQ_CONCAT(&doneq, &queue->cam_doneq);
+		mtx_unlock(&queue->cam_doneq_mtx);
+
+		while ((ccb_h = STAILQ_FIRST(&doneq)) != NULL) {
+			STAILQ_REMOVE_HEAD(&doneq, sim_links.stqe);
+			xpt_done_process(ccb_h);
+		}
+
+		mtx_lock(&queue->cam_doneq_mtx);
+	}
+}
+
 void
 xpt_done_td(void *arg)
 {