opencrypto: Introduce crypto_dispatch_async()

Currently, OpenCrypto consumers can request asynchronous dispatch by
setting a flag in the cryptop.  (Currently only IPSec may do this.)   I
think this is a bit confusing: we (conditionally) set cryptop flags to
request async dispatch, and then crypto_dispatch() immediately examines
those flags to see if the consumer wants async dispatch. The flag names
are also confusing since they don't specify what "async" applies to:
dispatch or completion.

Add a new KPI, crypto_dispatch_async(), rather than encoding the
requested dispatch type in each cryptop. crypto_dispatch_async() falls
back to crypto_dispatch() if the session's driver provides asynchronous
dispatch. Get rid of CRYPTOP_ASYNC() and CRYPTOP_ASYNC_KEEPORDER().

Similarly, add crypto_dispatch_batch() to request processing of a tailq
of cryptops, rather than encoding the scheduling policy using cryptop
flags.  Convert GELI, the only user of this interface (disabled by
default) to use the new interface.

Add CRYPTO_SESS_SYNC(), which can be used by consumers to determine
whether crypto requests will be dispatched synchronously. This is just
a helper macro. Use it instead of looking at cap flags directly.

Fix style in crypto_done(). Also get rid of CRYPTO_RETW_EMPTY() and
just check the relevant queues directly. This could result in some
unnecessary wakeups but I think it's very uncommon to be using more than
one queue per worker in a given workload, so checking all three queues
is a waste of cycles.

Reviewed by:	jhb
Sponsored by:	Ampere Computing
Submitted by:	Klara, Inc.
MFC after:	2 weeks
Differential Revision:	https://reviews.freebsd.org/D28194

2 files changed, 93 insertions, 76 deletions

diff --git a/sys/opencrypto/crypto.c b/sys/opencrypto/crypto.c
index 0316eb35361a..3b489739f067 100644
--- a/sys/opencrypto/crypto.c
+++ b/sys/opencrypto/crypto.c
@@ -188,8 +188,6 @@ static struct crypto_ret_worker *crypto_ret_workers = NULL;
 
 #define	CRYPTO_RETW_LOCK(w)	mtx_lock(&w->crypto_ret_mtx)
 #define	CRYPTO_RETW_UNLOCK(w)	mtx_unlock(&w->crypto_ret_mtx)
-#define	CRYPTO_RETW_EMPTY(w) \
-	(TAILQ_EMPTY(&w->crp_ret_q) && TAILQ_EMPTY(&w->crp_ret_kq) && TAILQ_EMPTY(&w->crp_ordered_ret_q))
 
 static int crypto_workers_num = 0;
 SYSCTL_INT(_kern_crypto, OID_AUTO, num_workers, CTLFLAG_RDTUN,
@@ -1406,11 +1404,8 @@ crp_sanity(struct cryptop *crp)
 }
 #endif
 
-/*
- * Add a crypto request to a queue, to be processed by the kernel thread.
- */
-int
-crypto_dispatch(struct cryptop *crp)
+static int
+crypto_dispatch_one(struct cryptop *crp, int hint)
 {
 	struct cryptocap *cap;
 	int result;
@@ -1418,49 +1413,82 @@ crypto_dispatch(struct cryptop *crp)
 #ifdef INVARIANTS
 	crp_sanity(crp);
 #endif
-
 	CRYPTOSTAT_INC(cs_ops);
 
 	crp->crp_retw_id = crp->crp_session->id % crypto_workers_num;
 
-	if (CRYPTOP_ASYNC(crp)) {
-		if (crp->crp_flags & CRYPTO_F_ASYNC_KEEPORDER) {
-			struct crypto_ret_worker *ret_worker;
+	/*
+	 * Caller marked the request to be processed immediately; dispatch it
+	 * directly to the driver unless the driver is currently blocked, in
+	 * which case it is queued for deferred dispatch.
+	 */
+	cap = crp->crp_session->cap;
+	if (!atomic_load_int(&cap->cc_qblocked)) {
+		result = crypto_invoke(cap, crp, hint);
+		if (result != ERESTART)
+			return (result);
 
-			ret_worker = CRYPTO_RETW(crp->crp_retw_id);
+		/*
+		 * The driver ran out of resources, put the request on the
+		 * queue.
+		 */
+	}
+	crypto_batch_enqueue(crp);
+	return (0);
+}
 
-			CRYPTO_RETW_LOCK(ret_worker);
-			crp->crp_seq = ret_worker->reorder_ops++;
-			CRYPTO_RETW_UNLOCK(ret_worker);
-		}
+int
+crypto_dispatch(struct cryptop *crp)
+{
+	return (crypto_dispatch_one(crp, 0));
+}
 
-		TASK_INIT(&crp->crp_task, 0, crypto_task_invoke, crp);
-		taskqueue_enqueue(crypto_tq, &crp->crp_task);
-		return (0);
-	}
+int
+crypto_dispatch_async(struct cryptop *crp, int flags)
+{
+	struct crypto_ret_worker *ret_worker;
 
-	if ((crp->crp_flags & CRYPTO_F_BATCH) == 0) {
+	if (!CRYPTO_SESS_SYNC(crp->crp_session)) {
 		/*
-		 * Caller marked the request to be processed
-		 * immediately; dispatch it directly to the
-		 * driver unless the driver is currently blocked.
+		 * The driver issues completions asynchonously, don't bother
+		 * deferring dispatch to a worker thread.
 		 */
-		cap = crp->crp_session->cap;
-		if (!cap->cc_qblocked) {
-			result = crypto_invoke(cap, crp, 0);
-			if (result != ERESTART)
-				return (result);
-			/*
-			 * The driver ran out of resources, put the request on
-			 * the queue.
-			 */
-		}
+		return (crypto_dispatch(crp));
 	}
-	crypto_batch_enqueue(crp);
-	return 0;
+
+#ifdef INVARIANTS
+	crp_sanity(crp);
+#endif
+	CRYPTOSTAT_INC(cs_ops);
+
+	crp->crp_retw_id = crp->crp_session->id % crypto_workers_num;
+	if ((flags & CRYPTO_ASYNC_ORDERED) != 0) {
+		crp->crp_flags |= CRYPTO_F_ASYNC_ORDERED;
+		ret_worker = CRYPTO_RETW(crp->crp_retw_id);
+		CRYPTO_RETW_LOCK(ret_worker);
+		crp->crp_seq = ret_worker->reorder_ops++;
+		CRYPTO_RETW_UNLOCK(ret_worker);
+	}
+	TASK_INIT(&crp->crp_task, 0, crypto_task_invoke, crp);
+	taskqueue_enqueue(crypto_tq, &crp->crp_task);
+	return (0);
 }
 
 void
+crypto_dispatch_batch(struct cryptopq *crpq, int flags)
+{
+	struct cryptop *crp;
+	int hint;
+
+	while ((crp = TAILQ_FIRST(crpq)) != NULL) {
+		hint = TAILQ_NEXT(crp, crp_next) != NULL ? CRYPTO_HINT_MORE : 0;
+		TAILQ_REMOVE(crpq, crp, crp_next);
+		if (crypto_dispatch_one(crp, hint) != 0)
+			crypto_batch_enqueue(crp);
+	}
+}
+
+static void
 crypto_batch_enqueue(struct cryptop *crp)
 {
 
@@ -1814,10 +1842,10 @@ crypto_done(struct cryptop *crp)
 	 * doing extraneous context switches; the latter is mostly
 	 * used with the software crypto driver.
 	 */
-	if (!CRYPTOP_ASYNC_KEEPORDER(crp) &&
-	    ((crp->crp_flags & CRYPTO_F_CBIMM) ||
-	    ((crp->crp_flags & CRYPTO_F_CBIFSYNC) &&
-	     (crypto_ses2caps(crp->crp_session) & CRYPTOCAP_F_SYNC)))) {
+	if ((crp->crp_flags & CRYPTO_F_ASYNC_ORDERED) == 0 &&
+	    ((crp->crp_flags & CRYPTO_F_CBIMM) != 0 ||
+	    ((crp->crp_flags & CRYPTO_F_CBIFSYNC) != 0 &&
+	    CRYPTO_SESS_SYNC(crp->crp_session)))) {
 		/*
 		 * Do the callback directly.  This is ok when the
 		 * callback routine does very little (e.g. the
@@ -1829,36 +1857,35 @@ crypto_done(struct cryptop *crp)
 		bool wake;
 
 		ret_worker = CRYPTO_RETW(crp->crp_retw_id);
-		wake = false;
 
 		/*
 		 * Normal case; queue the callback for the thread.
 		 */
 		CRYPTO_RETW_LOCK(ret_worker);
-		if (CRYPTOP_ASYNC_KEEPORDER(crp)) {
+		if ((crp->crp_flags & CRYPTO_F_ASYNC_ORDERED) != 0) {
 			struct cryptop *tmp;
 
-			TAILQ_FOREACH_REVERSE(tmp, &ret_worker->crp_ordered_ret_q,
-					cryptop_q, crp_next) {
+			TAILQ_FOREACH_REVERSE(tmp,
+			    &ret_worker->crp_ordered_ret_q, cryptop_q,
+			    crp_next) {
 				if (CRYPTO_SEQ_GT(crp->crp_seq, tmp->crp_seq)) {
-					TAILQ_INSERT_AFTER(&ret_worker->crp_ordered_ret_q,
-							tmp, crp, crp_next);
+					TAILQ_INSERT_AFTER(
+					    &ret_worker->crp_ordered_ret_q, tmp,
+					    crp, crp_next);
 					break;
 				}
 			}
 			if (tmp == NULL) {
-				TAILQ_INSERT_HEAD(&ret_worker->crp_ordered_ret_q,
-						crp, crp_next);
+				TAILQ_INSERT_HEAD(
+				    &ret_worker->crp_ordered_ret_q, crp,
+				    crp_next);
 			}
 
-			if (crp->crp_seq == ret_worker->reorder_cur_seq)
-				wake = true;
-		}
-		else {
-			if (CRYPTO_RETW_EMPTY(ret_worker))
-				wake = true;
-
-			TAILQ_INSERT_TAIL(&ret_worker->crp_ret_q, crp, crp_next);
+			wake = crp->crp_seq == ret_worker->reorder_cur_seq;
+		} else {
+			wake = TAILQ_EMPTY(&ret_worker->crp_ret_q);
+			TAILQ_INSERT_TAIL(&ret_worker->crp_ret_q, crp,
+			    crp_next);
 		}
 
 		if (wake)
@@ -1894,7 +1921,7 @@ crypto_kdone(struct cryptkop *krp)
 	ret_worker = CRYPTO_RETW(0);
 
 	CRYPTO_RETW_LOCK(ret_worker);
-	if (CRYPTO_RETW_EMPTY(ret_worker))
+	if (TAILQ_EMPTY(&ret_worker->crp_ret_kq))
 		wakeup_one(&ret_worker->crp_ret_q);		/* shared wait channel */
 	TAILQ_INSERT_TAIL(&ret_worker->crp_ret_kq, krp, krp_next);
 	CRYPTO_RETW_UNLOCK(ret_worker);
@@ -1991,13 +2018,10 @@ crypto_proc(void)
 					 */
 					if (submit->crp_session->cap == cap)
 						hint = CRYPTO_HINT_MORE;
-					break;
 				} else {
 					submit = crp;
-					if ((submit->crp_flags & CRYPTO_F_BATCH) == 0)
-						break;
-					/* keep scanning for more are q'd */
 				}
+				break;
 			}
 		}
 		if (submit != NULL) {
diff --git a/sys/opencrypto/cryptodev.h b/sys/opencrypto/cryptodev.h
index ecb1d929d1db..659599cb7d60 100644
--- a/sys/opencrypto/cryptodev.h
+++ b/sys/opencrypto/cryptodev.h
@@ -455,18 +455,10 @@ struct cryptop {
 					 */
 	int		crp_flags;
 
-#define	CRYPTO_F_BATCH		0x0008	/* Batch op if possible */
 #define	CRYPTO_F_CBIMM		0x0010	/* Do callback immediately */
 #define	CRYPTO_F_DONE		0x0020	/* Operation completed */
 #define	CRYPTO_F_CBIFSYNC	0x0040	/* Do CBIMM if op is synchronous */
-#define	CRYPTO_F_ASYNC		0x0080	/* Dispatch crypto jobs on several threads
-					 * if op is synchronous
-					 */
-#define	CRYPTO_F_ASYNC_KEEPORDER	0x0100	/*
-					 * Dispatch the crypto jobs in the same
-					 * order there are submitted. Applied only
-					 * if CRYPTO_F_ASYNC flags is set
-					 */
+#define	CRYPTO_F_ASYNC_ORDERED	0x0100	/* Completions must happen in order */
 #define	CRYPTO_F_IV_SEPARATE	0x0200	/* Use crp_iv[] as IV. */
 
 	int		crp_op;
@@ -506,6 +498,8 @@ struct cryptop {
 					 */
 };
 
+TAILQ_HEAD(cryptopq, cryptop);
+
 static __inline void
 _crypto_use_buf(struct crypto_buffer *cb, void *buf, int len)
 {
@@ -587,12 +581,6 @@ crypto_use_output_uio(struct cryptop *crp, struct uio *uio)
 	_crypto_use_uio(&crp->crp_obuf, uio);
 }
 
-#define	CRYPTOP_ASYNC(crp)			\
-	(((crp)->crp_flags & CRYPTO_F_ASYNC) && \
-	crypto_ses2caps((crp)->crp_session) & CRYPTOCAP_F_SYNC)
-#define	CRYPTOP_ASYNC_KEEPORDER(crp) \
-	(CRYPTOP_ASYNC(crp) && \
-	(crp)->crp_flags & CRYPTO_F_ASYNC_KEEPORDER)
 #define	CRYPTO_HAS_OUTPUT_BUFFER(crp)					\
 	((crp)->crp_obuf.cb_type != CRYPTO_BUF_NONE)
 
@@ -642,6 +630,8 @@ extern	void crypto_freesession(crypto_session_t cses);
 #define	CRYPTOCAP_F_SOFTWARE	CRYPTO_FLAG_SOFTWARE
 #define	CRYPTOCAP_F_SYNC	0x04000000	/* operates synchronously */
 #define	CRYPTOCAP_F_ACCEL_SOFTWARE 0x08000000
+#define	CRYPTO_SESS_SYNC(sess)	\
+	((crypto_ses2caps(sess) & CRYPTOCAP_F_SYNC) != 0)
 extern	int32_t crypto_get_driverid(device_t dev, size_t session_size,
     int flags);
 extern	int crypto_find_driver(const char *);
@@ -650,6 +640,9 @@ extern	int crypto_getcaps(int hid);
 extern	int crypto_kregister(uint32_t, int, uint32_t);
 extern	int crypto_unregister_all(uint32_t driverid);
 extern	int crypto_dispatch(struct cryptop *crp);
+#define	CRYPTO_ASYNC_ORDERED	0x1	/* complete in order dispatched */
+extern	int crypto_dispatch_async(struct cryptop *crp, int flags);
+extern	void crypto_dispatch_batch(struct cryptopq *crpq, int flags);
 extern	int crypto_kdispatch(struct cryptkop *);
 #define	CRYPTO_SYMQ	0x1
 #define	CRYPTO_ASYMQ	0x2