crypto: skcipher - multi-data-unit request splitting

index 4efe2ca8c4d1..2d7805409140 100644
@@ -31,6 +31,17 @@ struct scatterlist;
 /**
  *	struct skcipher_request - Symmetric key cipher request
  *	@cryptlen: Number of bytes to encrypt or decrypt
+ *	@unit_size: Size in bytes of each data unit, or 0 for a
+ *		single-unit request (the default).  When non-zero, must be a
+ *		multiple of the cipher block size and @cryptlen must be a
+ *		positive multiple of it.  The data-unit number is a 64-bit
+ *		little-endian counter in the low 8 bytes of @iv, incremented
+ *		once per unit and wrapping at 2^64; the remaining IV bytes are
+ *		left unchanged (any other on-disk IV layout is produced by a
+ *		template wrapping the algorithm).  @iv itself is not modified
+ *		by the request.  An algorithm that advertises
+ *		CRYPTO_ALG_REQ_SEG handles the whole request natively and must
+ *		follow this same counter and @iv-preservation convention.
  *	@iv: Initialisation Vector
  *	@src: Source SG list
  *	@dst: Destination SG list

@@ -39,6 +50,7 @@ struct scatterlist;
  */
 struct skcipher_request {
 	unsigned int cryptlen;
+	unsigned int unit_size;
 
 	u8 *iv;
 

@@ -225,6 +237,7 @@ struct lskcipher_alg {
 	struct skcipher_request *name = \
 		(((struct skcipher_request *)__##name##_desc)->base.tfm = \
 			crypto_sync_skcipher_tfm((_tfm)), \
+		 ((struct skcipher_request *)__##name##_desc)->unit_size = 0, \
 		 (void *)__##name##_desc)
 
 /**

@@ -819,6 +832,8 @@ static inline void skcipher_request_set_tfm(struct skcipher_request *req,
 					    struct crypto_skcipher *tfm)
 {
 	req->base.tfm = crypto_skcipher_tfm(tfm);
+	/* New tfm, new request: default to single-unit. */
+	req->unit_size = 0;
 }
 
 static inline void skcipher_request_set_sync_tfm(struct skcipher_request *req,

@@ -908,6 +923,10 @@ static inline void skcipher_request_set_callback(struct skcipher_request *req,
 	req->base.complete = compl;
 	req->base.data = data;
 	req->base.flags = flags;
+	/* Reset the per-op multi-unit control; a reused request defaults to
+	 * single-unit until skcipher_request_set_unit_size() opts back in.
+	 */
+	req->unit_size = 0;
 }
 
 /**

@@ -937,5 +956,32 @@ static inline void skcipher_request_set_crypt(
 	req->iv = iv;
 }
 
+/**
+ * skcipher_request_set_unit_size() - submit as multiple data units
+ * @req: request handle
+ * @unit_size: unit size in bytes (a multiple of the cipher block size),
+ *	       or 0 to disable
+ *
+ * Process @req as @cryptlen / @unit_size data units sharing one starting
+ * @iv, with per-unit IVs derived by treating @iv as a wide counter (the
+ * data-unit-number convention).  @cryptlen must be a positive multiple of
+ * @unit_size.  If the algorithm does not handle multiple units natively,
+ * the API transparently splits the request into one call per unit; that
+ * split additionally requires an ivsize that is a non-zero multiple of 8
+ * and at most 32 bytes, and rejects a violating request with -EINVAL.  An
+ * algorithm advertising CRYPTO_ALG_REQ_SEG receives the whole request and
+ * enforces its own constraints.
+ *
+ * This function must be called after skcipher_request_set_tfm() and
+ * skcipher_request_set_callback(), both of which reset @req->unit_size
+ * to 0.
+ */
+static inline void
+skcipher_request_set_unit_size(struct skcipher_request *req,
+			       unsigned int unit_size)
+{
+	req->unit_size = unit_size;
+}
+
 #endif	/* _CRYPTO_SKCIPHER_H */
 

index 05deea9dac5e..7d406cfe5751 100644
@@ -280,6 +280,11 @@ static inline bool crypto_tfm_req_virt(struct crypto_tfm *tfm)
 	return tfm->__crt_alg->cra_flags & CRYPTO_ALG_REQ_VIRT;
 }
 
+static inline bool crypto_tfm_req_seg(struct crypto_tfm *tfm)
+{
+	return tfm->__crt_alg->cra_flags & CRYPTO_ALG_REQ_SEG;
+}
+
 static inline u32 crypto_request_flags(struct crypto_async_request *req)
 {
 	return req->flags & ~CRYPTO_TFM_REQ_ON_STACK;

index 9cd37df32dc4..93ae59623f13 100644
@@ -189,6 +189,11 @@ static inline bool crypto_acomp_req_virt(struct crypto_acomp *tfm)
 	return crypto_tfm_req_virt(&tfm->base);
 }
 
+static inline bool crypto_acomp_req_seg(struct crypto_acomp *tfm)
+{
+	return crypto_tfm_req_seg(&tfm->base);
+}
+
 void crypto_acomp_free_streams(struct crypto_acomp_streams *s);
 int crypto_acomp_alloc_streams(struct crypto_acomp_streams *s);
 

index a2137e19be7d..89b9c3f87f4d 100644
@@ -139,6 +139,9 @@
 /* Set if the algorithm cannot have a fallback (e.g., phmac). */
 #define CRYPTO_ALG_NO_FALLBACK		0x00080000
 
+/* Set if the algorithm supports segmentation. */
+#define CRYPTO_ALG_REQ_SEG		0x00100000
+
 /* The high bits 0xff000000 are reserved for type-specific flags. */
 
 /*

index a965b6aabf61..8e9353f4252e 100644
@@ -249,6 +249,11 @@ static inline bool crypto_skcipher_tested(struct crypto_skcipher *tfm)
 	return tfm_base->__crt_alg->cra_flags & CRYPTO_ALG_TESTED;
 }
 
+static inline bool crypto_skcipher_req_seg(struct crypto_skcipher *tfm)
+{
+	return crypto_tfm_req_seg(&tfm->base);
+}
+
 static inline void *skcipher_request_ctx(struct skcipher_request *req)
 {
 	return req->__ctx;

index 2b31d1d5d268..db1576f6bba0 100644
@@ -432,6 +432,112 @@ int crypto_skcipher_setkey(struct crypto_skcipher *tfm, const u8 *key,
 }
 EXPORT_SYMBOL_GPL(crypto_skcipher_setkey);
 
+/* Bounds the on-stack per-unit IV buffers: 16 covers xts, 32 Adiantum. */
+#define SKCIPHER_MAX_UNIT_IVSIZE	32
+
+/*
+ * Advance the per-unit IV to the next data unit.  The data-unit number is a
+ * 64-bit little-endian counter held in the low 8 bytes of @iv (matching
+ * dm-crypt's plain64/essiv sector generators, which the caller feeds in as a
+ * little-endian sector number).  It wraps at 2^64 and never carries into the
+ * higher IV bytes, so batched output stays bit-identical to the per-unit path
+ * across the counter rollover; any other on-disk IV format is produced by a
+ * template wrapping the algorithm, not here.
+ */
+static void skcipher_unit_iv_next(u8 *iv)
+{
+	__le64 lo;
+
+	memcpy(&lo, iv, sizeof(lo));
+	lo = cpu_to_le64(le64_to_cpu(lo) + 1);
+	memcpy(iv, &lo, sizeof(lo));
+}
+
+/*
+ * Transparently split a multi-unit request for an algorithm with no native
+ * multi-unit support: one call per unit, walking the IV as a wide counter.
+ * The caller's request is reused for each unit (same tfm, so the request
+ * context is already correctly sized) and fully restored before returning.
+ * @req->iv is never modified; each unit gets a private copy the algorithm
+ * may write back in place (e.g. xts).
+ */
+static int skcipher_crypt_unit(struct skcipher_request *req, bool enc)
+{
+	struct crypto_skcipher *tfm = crypto_skcipher_reqtfm(req);
+	struct skcipher_alg *alg = crypto_skcipher_alg(tfm);
+	const unsigned int unit = req->unit_size;
+	const unsigned int total = req->cryptlen;
+	const unsigned int ivsize = crypto_skcipher_ivsize(tfm);
+	bool inplace = req->src == req->dst;
+	struct scatterlist *o_src = req->src, *o_dst = req->dst;
+	struct scatter_walk src_walk, dst_walk;
+	struct scatterlist src_sg[2], dst_sg[2];
+	u8 iv_ctr[SKCIPHER_MAX_UNIT_IVSIZE];
+	/* Becomes req->iv: keep the alignment the caller's IV would have had. */
+	u8 iv_unit[SKCIPHER_MAX_UNIT_IVSIZE] __aligned(MAX_ALGAPI_ALIGNMASK + 1);
+	u8 *o_iv = req->iv;
+	unsigned int off;
+	int err = 0;
+
+	if (!total || !IS_ALIGNED(unit, crypto_skcipher_blocksize(tfm)) ||
+	    (total % unit) || !ivsize ||
+	    !IS_ALIGNED(ivsize, sizeof(__le64)) ||
+	    ivsize > SKCIPHER_MAX_UNIT_IVSIZE)
+		return -EINVAL;
+
+	/* The split is synchronous; only a native (REQ_SEG) alg may be async. */
+	if (alg->co.base.cra_flags & CRYPTO_ALG_ASYNC)
+		return -EOPNOTSUPP;
+
+	/* iv_ctr is the counter; iv_unit is the per-unit copy. */
+	memcpy(iv_ctr, req->iv, ivsize);
+
+	sg_init_table(src_sg, 2);
+	scatterwalk_start(&src_walk, req->src);
+	if (!inplace) {
+		sg_init_table(dst_sg, 2);
+		scatterwalk_start(&dst_walk, req->dst);
+	}
+
+	req->unit_size = 0;
+	req->cryptlen = unit;
+
+	for (off = 0; off < total; off += unit) {
+		scatterwalk_get_sglist(&src_walk, src_sg);
+		scatterwalk_skip(&src_walk, unit);
+		req->src = src_sg;
+		if (inplace) {
+			req->dst = src_sg;
+		} else {
+			scatterwalk_get_sglist(&dst_walk, dst_sg);
+			scatterwalk_skip(&dst_walk, unit);
+			req->dst = dst_sg;
+		}
+
+		memcpy(iv_unit, iv_ctr, ivsize);
+		req->iv = iv_unit;
+		err = enc ? crypto_skcipher_encrypt(req) :
+			    crypto_skcipher_decrypt(req);
+		if (err)
+			break;
+
+		skcipher_unit_iv_next(iv_ctr);
+		/*
+		 * Match dm-crypt's per-sector reschedule, but only when the
+		 * caller allows sleeping (the split can run in atomic context).
+		 */
+		if (req->base.flags & CRYPTO_TFM_REQ_MAY_SLEEP)
+			cond_resched();
+	}
+
+	req->src = o_src;
+	req->dst = o_dst;
+	req->iv = o_iv;
+	req->cryptlen = total;
+	req->unit_size = unit;
+	return err;
+}
+
 int crypto_skcipher_encrypt(struct skcipher_request *req)
 {
 	struct crypto_skcipher *tfm = crypto_skcipher_reqtfm(req);

@@ -439,6 +545,9 @@ int crypto_skcipher_encrypt(struct skcipher_request *req)
 
 	if (crypto_skcipher_get_flags(tfm) & CRYPTO_TFM_NEED_KEY)
 		return -ENOKEY;
+	/* Must precede the lskcipher redirect, which ignores unit_size. */
+	if (unlikely(req->unit_size) && !crypto_skcipher_req_seg(tfm))
+		return skcipher_crypt_unit(req, true);
 	if (alg->co.base.cra_type != &crypto_skcipher_type)
 		return crypto_lskcipher_encrypt_sg(req);
 	return alg->encrypt(req);

@@ -452,6 +561,9 @@ int crypto_skcipher_decrypt(struct skcipher_request *req)
 
 	if (crypto_skcipher_get_flags(tfm) & CRYPTO_TFM_NEED_KEY)
 		return -ENOKEY;
+	/* Must precede the lskcipher redirect, which ignores unit_size. */
+	if (unlikely(req->unit_size) && !crypto_skcipher_req_seg(tfm))
+		return skcipher_crypt_unit(req, false);
 	if (alg->co.base.cra_type != &crypto_skcipher_type)
 		return crypto_lskcipher_decrypt_sg(req);
 	return alg->decrypt(req);

index 4d86efae65b2..225c6a9806cb 100644
@@ -3211,6 +3211,274 @@ static int test_skcipher(int enc, const struct cipher_test_suite *suite,
 	return 0;
 }
 
+/* Upper bound on the IVs the multi-unit split accepts (16: xts; 32: Adiantum). */
+#define TEST_MDU_MAX_IVSIZE	32
+
+/*
+ * Increment the IV as a 64-bit little-endian data-unit-number counter in the
+ * low 8 bytes (byte 0 the LSB), wrapping at 2^64 with no carry above -- the
+ * dm-crypt plain64/essiv convention.  Deliberately independent of
+ * skcipher_unit_iv_next()'s implementation, so the two only agree if the
+ * carry-and-wrap is right.
+ */
+static void test_mdu_iv_inc(u8 *iv)
+{
+	int i;
+
+	for (i = 0; i < 8; i++)
+		if (++iv[i])
+			break;
+}
+
+/*
+ * Seed @iv so the low 64-bit counter (bytes [0,8)) is all-ones but its
+ * least-significant byte: the 2nd increment wraps the counter back to zero,
+ * exercising the rollover.  Bytes outside the low 8 keep their value.
+ */
+static void test_mdu_iv_boundary(u8 *iv)
+{
+	unsigned int i;
+
+	for (i = 0; i < 8; i++)
+		iv[i] = 0xff;
+	iv[0] = 0xfe;
+}
+
+/* Encrypt one du_size block with a plain single-DU request (the reference). */
+static int test_mdu_ref_encrypt(struct crypto_skcipher *tfm, const u8 *in,
+				u8 *out, unsigned int du_size, const u8 *iv,
+				unsigned int ivsize)
+{
+	struct skcipher_request *req;
+	struct scatterlist sg_in;
+	DECLARE_CRYPTO_WAIT(wait);
+	u8 ivbuf[TEST_MDU_MAX_IVSIZE];
+	int err;
+
+	req = skcipher_request_alloc(tfm, GFP_KERNEL);
+	if (!req)
+		return -ENOMEM;
+	memcpy(ivbuf, iv, ivsize);
+	memcpy(out, in, du_size);
+	sg_init_one(&sg_in, out, du_size);
+	skcipher_request_set_callback(req, CRYPTO_TFM_REQ_MAY_BACKLOG |
+				      CRYPTO_TFM_REQ_MAY_SLEEP,
+				      crypto_req_done, &wait);
+	skcipher_request_set_crypt(req, &sg_in, &sg_in, du_size, ivbuf);
+	err = crypto_wait_req(crypto_skcipher_encrypt(req), &wait);
+	skcipher_request_free(req);
+	return err;
+}
+
+/*
+ * Build an SG over @buf with du_size-unaligned entries, so the splitter's
+ * per-DU views cross SG entries and exercise the scatter_walk cursor.
+ */
+static void test_mdu_sg_fragment(struct scatterlist *sg, unsigned int nents,
+				 u8 *buf, unsigned int total)
+{
+	unsigned int chunk = total / nents;
+	unsigned int off = 0, i;
+
+	sg_init_table(sg, nents);
+	for (i = 0; i < nents; i++) {
+		unsigned int len = (i == nents - 1) ? total - off : chunk;
+
+		sg_set_buf(&sg[i], buf + off, len);
+		off += len;
+	}
+}
+
+#define TEST_MDU_NR_UNITS	4
+#define TEST_MDU_NR_FRAGS	5
+/*
+ * Verify a batched (unit_size-set) request on @tfm is byte-equal to an
+ * independent N x single-unit reference on the same tfm with
+ * little-endian-walked IVs, over a fragmented SG, then round-trips.
+ * @iv_orig is the ivsize-byte starting IV (the caller varies it to exercise
+ * both a random IV and one seeded to cross a carry boundary).
+ */
+static int test_skcipher_multi_du_one(struct crypto_skcipher *tfm,
+				      unsigned int du_size,
+				      const u8 *iv_orig)
+{
+	const char *driver = crypto_skcipher_driver_name(tfm);
+	const unsigned int total = du_size * TEST_MDU_NR_UNITS;
+	const unsigned int ivsize = crypto_skcipher_ivsize(tfm);
+	const u32 flags = CRYPTO_TFM_REQ_MAY_BACKLOG | CRYPTO_TFM_REQ_MAY_SLEEP;
+	struct skcipher_request *req = NULL;
+	struct scatterlist sg[TEST_MDU_NR_FRAGS], sg_out[TEST_MDU_NR_FRAGS];
+	DECLARE_CRYPTO_WAIT(wait);
+	u8 iv_work[TEST_MDU_MAX_IVSIZE], iv_ref[TEST_MDU_MAX_IVSIZE];
+	u8 *plain = NULL, *buf = NULL, *ref = NULL, *obuf = NULL;
+	unsigned int u;
+	int err;
+
+	plain = kmalloc(total, GFP_KERNEL);
+	buf = kmalloc(total, GFP_KERNEL);
+	ref = kmalloc(total, GFP_KERNEL);
+	obuf = kmalloc(total, GFP_KERNEL);
+	req = skcipher_request_alloc(tfm, GFP_KERNEL);
+	if (!plain || !buf || !ref || !obuf || !req) {
+		err = -ENOMEM;
+		goto out;
+	}
+
+	get_random_bytes(plain, total);
+
+	/* Reference: per-unit single requests, counter-walked IVs. */
+	memcpy(iv_ref, iv_orig, ivsize);
+	for (u = 0; u < TEST_MDU_NR_UNITS; u++) {
+		err = test_mdu_ref_encrypt(tfm, plain + u * du_size,
+					   ref + u * du_size, du_size, iv_ref,
+					   ivsize);
+		if (err) {
+			pr_err("alg: skcipher: %s multi-DU ref encrypt failed (du=%u): %d\n",
+			       driver, du_size, err);
+			goto out;
+		}
+		test_mdu_iv_inc(iv_ref);
+	}
+
+	/* Batched: one request over a fragmented SG. */
+	memcpy(buf, plain, total);
+	memcpy(iv_work, iv_orig, ivsize);
+	test_mdu_sg_fragment(sg, TEST_MDU_NR_FRAGS, buf, total);
+	skcipher_request_set_callback(req, flags, crypto_req_done, &wait);
+	skcipher_request_set_crypt(req, sg, sg, total, iv_work);
+	skcipher_request_set_unit_size(req, du_size);
+	err = crypto_wait_req(crypto_skcipher_encrypt(req), &wait);
+	if (err) {
+		pr_err("alg: skcipher: %s multi-DU encrypt failed (du=%u): %d\n",
+		       driver, du_size, err);
+		goto out;
+	}
+	if (memcmp(buf, ref, total) != 0) {
+		pr_err("alg: skcipher: %s multi-DU ciphertext differs from single-DU reference (du=%u)\n",
+		       driver, du_size);
+		err = -EBADMSG;
+		goto out;
+	}
+	/* req->iv must be unchanged after multi-DU dispatch. */
+	if (memcmp(iv_work, iv_orig, ivsize) != 0) {
+		pr_err("alg: skcipher: %s multi-DU encrypt mutated caller IV (du=%u)\n",
+		       driver, du_size);
+		err = -EBADMSG;
+		goto out;
+	}
+
+	/* Out-of-place: distinct dst SG, fragmented differently from src. */
+	memcpy(buf, plain, total);
+	memset(obuf, 0, total);
+	test_mdu_sg_fragment(sg, TEST_MDU_NR_FRAGS, buf, total);
+	test_mdu_sg_fragment(sg_out, TEST_MDU_NR_FRAGS - 2, obuf, total);
+	skcipher_request_set_callback(req, flags, crypto_req_done, &wait);
+	skcipher_request_set_crypt(req, sg, sg_out, total, iv_work);
+	skcipher_request_set_unit_size(req, du_size);
+	err = crypto_wait_req(crypto_skcipher_encrypt(req), &wait);
+	if (err) {
+		pr_err("alg: skcipher: %s multi-DU out-of-place encrypt failed (du=%u): %d\n",
+		       driver, du_size, err);
+		goto out;
+	}
+	if (memcmp(obuf, ref, total) != 0) {
+		pr_err("alg: skcipher: %s multi-DU out-of-place ciphertext differs (du=%u)\n",
+		       driver, du_size);
+		err = -EBADMSG;
+		goto out;
+	}
+
+	/* Round-trip the batched ciphertext back to plaintext. */
+	memcpy(buf, ref, total);
+	test_mdu_sg_fragment(sg, TEST_MDU_NR_FRAGS, buf, total);
+	skcipher_request_set_callback(req, flags, crypto_req_done, &wait);
+	skcipher_request_set_crypt(req, sg, sg, total, iv_work);
+	skcipher_request_set_unit_size(req, du_size);
+	err = crypto_wait_req(crypto_skcipher_decrypt(req), &wait);
+	if (err) {
+		pr_err("alg: skcipher: %s multi-DU decrypt failed (du=%u): %d\n",
+		       driver, du_size, err);
+		goto out;
+	}
+	if (memcmp(buf, plain, total) != 0) {
+		pr_err("alg: skcipher: %s multi-DU round-trip mismatch (du=%u)\n",
+		       driver, du_size);
+		err = -EBADMSG;
+	}
+
+out:
+	skcipher_request_free(req);
+	kfree(obuf);
+	kfree(ref);
+	kfree(buf);
+	kfree(plain);
+	return err;
+}
+
+/*
+ * Cross-check multi-unit dispatch against a single-unit reference on @tfm
+ * over all unit sizes.  Returns 0 on success or skip; -EBADMSG on a real
+ * mismatch.
+ */
+static int test_skcipher_multi_du_sizes(struct crypto_skcipher *tfm)
+{
+	static const unsigned int du_sizes[] = { 512, 1024, 2048, 4096 };
+	unsigned int ivsize = crypto_skcipher_ivsize(tfm);
+	u8 iv[TEST_MDU_MAX_IVSIZE];
+	unsigned int j;
+	int err = 0;
+
+	for (j = 0; j < ARRAY_SIZE(du_sizes); j++) {
+		/* A random starting IV. */
+		get_random_bytes(iv, ivsize);
+		err = test_skcipher_multi_du_one(tfm, du_sizes[j], iv);
+		if (err)
+			break;
+		/* And one seeded to carry across a 64-bit limb. */
+		get_random_bytes(iv, ivsize);
+		test_mdu_iv_boundary(iv);
+		err = test_skcipher_multi_du_one(tfm, du_sizes[j], iv);
+		if (err)
+			break;
+		cond_resched();
+	}
+	return err;
+}
+
+/*
+ * Cross-check multi-unit dispatch against a single-unit reference for every
+ * eligible ivsize (16: xts; 32: Adiantum).
+ */
+static int test_skcipher_multi_du(struct crypto_skcipher *tfm)
+{
+	unsigned int ivsize = crypto_skcipher_ivsize(tfm);
+	unsigned int blocksize = crypto_skcipher_blocksize(tfm);
+	u8 keybuf[128];
+	unsigned int keylen;
+	int err;
+
+	if (noslowtests)
+		return 0;
+
+	/* Mirror the API-layer split's eligibility; skip what it rejects. */
+	if (!ivsize || ivsize % sizeof(__le64) || ivsize > TEST_MDU_MAX_IVSIZE)
+		return 0;
+	if (!blocksize || 512 % blocksize)
+		return 0;	/* unit sizes below are 512-multiples */
+	if (crypto_skcipher_alg(tfm)->co.base.cra_flags & CRYPTO_ALG_ASYNC)
+		return 0;	/* the transparent split is sync-only */
+
+	keylen = crypto_skcipher_min_keysize(tfm);
+	if (keylen > sizeof(keybuf))
+		return 0;	/* unusually large key; skip rather than overflow */
+	get_random_bytes(keybuf, keylen);
+	err = crypto_skcipher_setkey(tfm, keybuf, keylen);
+	if (err)
+		return 0;	/* weak/rejected key (e.g. XTS equal halves): skip */
+
+	return test_skcipher_multi_du_sizes(tfm);
+}
+
 static int alg_test_skcipher(const struct alg_test_desc *desc,
 			     const char *driver, u32 type, u32 mask)
 {

@@ -3259,6 +3527,10 @@ static int alg_test_skcipher(const struct alg_test_desc *desc,
 	if (err)
 		goto out;
 
+	err = test_skcipher_multi_du(tfm);
+	if (err)
+		goto out;
+
 	err = test_skcipher_vs_generic_impl(desc->generic_driver, req, tsgls);
 out:
 	free_cipher_test_sglists(tsgls);

index 608b617fb817..ffb66c7c7a65 100644
@@ -115,6 +115,15 @@ struct crypt_iv_operations {
 			 struct dm_crypt_request *dmreq);
 	void (*post)(struct crypt_config *cc, u8 *iv,
 		     struct dm_crypt_request *dmreq);
+
+	/*
+	 * Set for IV modes whose per-sector IV is a little-endian
+	 * data-unit-number counter (IV(s+i) == IV(s)+i) placed in the low
+	 * 64-bit limb, enabling multi-unit batching via the skcipher API-layer
+	 * split.  Clear for non-counter modes (lmk, tcw, ...) and for counter
+	 * modes whose on-disk IV is not that layout (e.g. plain64be).
+	 */
+	bool unit_counter;
 };
 
 struct iv_benbi_private {

@@ -151,6 +160,7 @@ enum cipher_flags {
 	CRYPT_IV_LARGE_SECTORS,		/* Calculate IV from sector_size, not 512B sectors */
 	CRYPT_ENCRYPT_PREPROCESS,	/* Must preprocess data for encryption (elephant) */
 	CRYPT_KEY_MAC_SIZE_SET,		/* The integrity_key_size option was used */
+	CRYPT_MULTI_DATA_UNIT,		/* Batch a bio segment's sectors per crypto request */
 };
 
 /*

@@ -1018,15 +1028,23 @@ static const struct crypt_iv_operations crypt_iv_plain_ops = {
 };
 
 static const struct crypt_iv_operations crypt_iv_plain64_ops = {
-	.generator = crypt_iv_plain64_gen
+	.generator = crypt_iv_plain64_gen,
+	.unit_counter = true,
 };
 
 static const struct crypt_iv_operations crypt_iv_plain64be_ops = {
-	.generator = crypt_iv_plain64be_gen
+	.generator = crypt_iv_plain64be_gen,
+	/*
+	 * No unit_counter: the big-endian, high-limb on-disk layout is not the
+	 * little-endian low-limb counter the API-layer split walks.  Batching
+	 * it needs a template producing this layout; unbatched for now.
+	 */
 };
 
 static const struct crypt_iv_operations crypt_iv_essiv_ops = {
-	.generator = crypt_iv_essiv_gen
+	.generator = crypt_iv_essiv_gen,
+	/* IV input is le64(sector); the salt-encrypt lives in essiv(). */
+	.unit_counter = true,
 };
 
 static const struct crypt_iv_operations crypt_iv_benbi_ops = {

@@ -1349,21 +1367,51 @@ static int crypt_convert_block_aead(struct crypt_config *cc,
 	return r;
 }
 
+/*
+ * Bytes to process in one skcipher request: a whole contiguous segment when
+ * batching (multi-data-unit), else one sector.  0 means an unusable
+ * (sub-sector / misaligned) segment.
+ */
+static unsigned int crypt_skcipher_len(struct crypt_config *cc,
+				       const struct bio_vec *bv_in,
+				       const struct bio_vec *bv_out)
+{
+	const unsigned int sector_size = cc->sector_size;
+
+	if (test_bit(CRYPT_MULTI_DATA_UNIT, &cc->cipher_flags))
+		return round_down(min(bv_in->bv_len, bv_out->bv_len),
+				  sector_size);
+
+	/* Reject unexpected unaligned bio. */
+	if (unlikely(bv_in->bv_len & (sector_size - 1)))
+		return 0;
+	return sector_size;
+}
+
+/*
+ * Encrypt/decrypt one bio segment (one sector, or a whole segment when
+ * batching) and report the bytes done in *out_processed.  The integrity /
+ * preprocess / post handling is inert when batching (crypt_can_batch_units()
+ * excludes those configs).
+ */
 static int crypt_convert_block_skcipher(struct crypt_config *cc,
 					struct convert_context *ctx,
 					struct skcipher_request *req,
-					unsigned int tag_offset)
+					unsigned int tag_offset,
+					unsigned int *out_processed)
 {
 	struct bio_vec bv_in = bio_iter_iovec(ctx->bio_in, ctx->iter_in);
 	struct bio_vec bv_out = bio_iter_iovec(ctx->bio_out, ctx->iter_out);
+	const unsigned int sector_size = cc->sector_size;
 	struct scatterlist *sg_in, *sg_out;
 	struct dm_crypt_request *dmreq;
 	u8 *iv, *org_iv, *tag_iv;
 	__le64 *sector;
+	unsigned int len;
 	int r = 0;
 
-	/* Reject unexpected unaligned bio. */
-	if (unlikely(bv_in.bv_len & (cc->sector_size - 1)))
+	len = crypt_skcipher_len(cc, &bv_in, &bv_out);
+	if (unlikely(!len))
 		return -EIO;
 
 	dmreq = dmreq_of_req(cc, req);

@@ -1386,10 +1434,10 @@ static int crypt_convert_block_skcipher(struct crypt_config *cc,
 	sg_out = &dmreq->sg_out[0];
 
 	sg_init_table(sg_in, 1);
-	sg_set_page(sg_in, bv_in.bv_page, cc->sector_size, bv_in.bv_offset);
+	sg_set_page(sg_in, bv_in.bv_page, len, bv_in.bv_offset);
 
 	sg_init_table(sg_out, 1);
-	sg_set_page(sg_out, bv_out.bv_page, cc->sector_size, bv_out.bv_offset);
+	sg_set_page(sg_out, bv_out.bv_page, len, bv_out.bv_offset);
 
 	if (cc->iv_gen_ops) {
 		/* For READs use IV stored in integrity metadata */

@@ -1410,7 +1458,9 @@ static int crypt_convert_block_skcipher(struct crypt_config *cc,
 		memcpy(iv, org_iv, cc->iv_size);
 	}
 
-	skcipher_request_set_crypt(req, sg_in, sg_out, cc->sector_size, iv);
+	skcipher_request_set_crypt(req, sg_in, sg_out, len, iv);
+	if (test_bit(CRYPT_MULTI_DATA_UNIT, &cc->cipher_flags))
+		skcipher_request_set_unit_size(req, sector_size);
 
 	if (bio_data_dir(ctx->bio_in) == WRITE)
 		r = crypto_skcipher_encrypt(req);

@@ -1420,9 +1470,10 @@ static int crypt_convert_block_skcipher(struct crypt_config *cc,
 	if (!r && cc->iv_gen_ops && cc->iv_gen_ops->post)
 		cc->iv_gen_ops->post(cc, org_iv, dmreq);
 
-	bio_advance_iter(ctx->bio_in, &ctx->iter_in, cc->sector_size);
-	bio_advance_iter(ctx->bio_out, &ctx->iter_out, cc->sector_size);
+	bio_advance_iter(ctx->bio_in, &ctx->iter_in, len);
+	bio_advance_iter(ctx->bio_out, &ctx->iter_out, len);
 
+	*out_processed = len;
 	return r;
 }
 

@@ -1509,13 +1560,25 @@ static void crypt_free_req(struct crypt_config *cc, void *req, struct bio *base_
 		crypt_free_req_skcipher(cc, req, base_bio);
 }
 
+/*
+ * Advance the IV-sector and integrity-tag cursors by @processed bytes; the
+ * bio iterators are advanced by the per-block helpers themselves.
+ */
+static void crypt_convert_advance(struct crypt_config *cc,
+				  struct convert_context *ctx,
+				  unsigned int processed)
+{
+	ctx->cc_sector += processed >> SECTOR_SHIFT;
+	ctx->tag_offset += processed / cc->sector_size;
+}
+
 /*
  * Encrypt / decrypt data from one bio to another one (can be the same one)
  */
 static blk_status_t crypt_convert(struct crypt_config *cc,
 			 struct convert_context *ctx, bool atomic, bool reset_pending)
 {
-	unsigned int sector_step = cc->sector_size >> SECTOR_SHIFT;
+	unsigned int processed;
 	int r;
 
 	/*

@@ -1536,10 +1599,12 @@ static blk_status_t crypt_convert(struct crypt_config *cc,
 
 		atomic_inc(&ctx->cc_pending);
 
+		processed = cc->sector_size;
 		if (crypt_integrity_aead(cc))
 			r = crypt_convert_block_aead(cc, ctx, ctx->r.req_aead, ctx->tag_offset);
 		else
-			r = crypt_convert_block_skcipher(cc, ctx, ctx->r.req, ctx->tag_offset);
+			r = crypt_convert_block_skcipher(cc, ctx, ctx->r.req,
+							 ctx->tag_offset, &processed);
 
 		switch (r) {
 		/*

@@ -1559,8 +1624,7 @@ static blk_status_t crypt_convert(struct crypt_config *cc,
 					 * exit and continue processing in a workqueue
 					 */
 					ctx->r.req = NULL;
-					ctx->tag_offset++;
-					ctx->cc_sector += sector_step;
+					crypt_convert_advance(cc, ctx, processed);
 					return BLK_STS_DEV_RESOURCE;
 				}
 			} else {

@@ -1574,16 +1638,14 @@ static blk_status_t crypt_convert(struct crypt_config *cc,
 		 */
 		case -EINPROGRESS:
 			ctx->r.req = NULL;
-			ctx->tag_offset++;
-			ctx->cc_sector += sector_step;
+			crypt_convert_advance(cc, ctx, processed);
 			continue;
 		/*
 		 * The request was already processed (synchronously).
 		 */
 		case 0:
 			atomic_dec(&ctx->cc_pending);
-			ctx->cc_sector += sector_step;
-			ctx->tag_offset++;
+			crypt_convert_advance(cc, ctx, processed);
 			if (!atomic)
 				cond_resched();
 			continue;

@@ -2345,12 +2407,28 @@ static int crypt_alloc_tfms_aead(struct crypt_config *cc, char *ciphermode)
 	return 0;
 }
 
+/*
+ * Whether multi-unit batching applies: a counter IV mode (unit_counter set),
+ * single-tfm, non-aead, and a per-unit IV step of exactly one (512B sectors
+ * or iv_large_sectors).  The IV must also satisfy the API split's counter
+ * constraints (non-zero multiple of 8, <= 32 bytes).  Integrity is excluded
+ * in crypt_ctr_cipher(), which runs after integrity is configured.
+ */
+static bool crypt_can_batch_units(struct crypt_config *cc)
+{
+	return !crypt_integrity_aead(cc) && cc->tfms_count == 1 &&
+		cc->iv_gen_ops && cc->iv_gen_ops->unit_counter &&
+		cc->iv_size && IS_ALIGNED(cc->iv_size, sizeof(__le64)) &&
+		cc->iv_size <= 32 &&
+		(cc->sector_size == (1 << SECTOR_SHIFT) ||
+		 test_bit(CRYPT_IV_LARGE_SECTORS, &cc->cipher_flags));
+}
+
 static int crypt_alloc_tfms(struct crypt_config *cc, char *ciphermode)
 {
 	if (crypt_integrity_aead(cc))
 		return crypt_alloc_tfms_aead(cc, ciphermode);
-	else
-		return crypt_alloc_tfms_skcipher(cc, ciphermode);
+	return crypt_alloc_tfms_skcipher(cc, ciphermode);
 }
 
 static unsigned int crypt_subkey_size(struct crypt_config *cc)

@@ -2999,7 +3077,6 @@ static int crypt_ctr_cipher_old(struct dm_target *ti, char *cipher_in, char *key
 		goto bad_mem;
 	}
 
-	/* Allocate cipher */
 	ret = crypt_alloc_tfms(cc, cipher_api);
 	if (ret < 0) {
 		ti->error = "Error allocating crypto tfm";

@@ -3063,6 +3140,21 @@ static int crypt_ctr_cipher(struct dm_target *ti, char *cipher_in, char *key)
 		}
 	}
 
+	/*
+	 * Enable multi-unit batching for an eligible config with no integrity
+	 * (integrity is set up after cipher alloc, hence the re-check here).
+	 * The API layer's transparent split is synchronous, so an async cipher
+	 * batches only if it handles multi-unit requests natively.
+	 */
+	if (crypt_can_batch_units(cc) && !cc->integrity_tag_size &&
+	    !cc->integrity_iv_size &&
+	    (crypto_skcipher_alg(any_tfm(cc))->co.base.cra_flags &
+	     (CRYPTO_ALG_ASYNC | CRYPTO_ALG_REQ_SEG)) != CRYPTO_ALG_ASYNC) {
+		set_bit(CRYPT_MULTI_DATA_UNIT, &cc->cipher_flags);
+		DMINFO("Using multi-data-unit crypto offload (du=%u)",
+		       cc->sector_size);
+	}
+
 	/* wipe the kernel key payload copy */
 	if (cc->key_string)
 		memset(cc->key, 0, cc->key_size * sizeof(u8));