BLS: speedup parallel batch verify with Miller loops on local threads

constantine/math/pairings/miller_accumulators.nim

@@ -102,6 +102,18 @@ func update*[FF1, FF2, FpK](ctx: var MillerAccumulator[FF1, FF2, FpK], P: ECP_Sh
   ctx.len += 1
   return true
 
+func handover*(ctx: var MillerAccumulator) {.inline.} =
+  ## Prepare accumulator for cheaper merging.
+  ##
+  ## In a multi-threaded context, multiple accumulators can be created and process subsets of the batch in parallel.
+  ## Accumulators can then be merged:
+  ##    merger_accumulator += mergee_accumulator
+  ## Merging will involve an expensive reduction operation when an accumulation threshold of 8 is reached.
+  ## However merging two reduced accumulators is 136x cheaper.
+  ##
+  ## `Handover` forces this reduction on local threads to limit the burden on the merger thread.
+  ctx.consumeBuffers()
+
 func merge*(ctxDst: var MillerAccumulator, ctxSrc: MillerAccumulator) =
   ## Merge ctxDst <- ctxDst + ctxSrc
   var sCur = 0'u

constantine/signatures/bls_signatures.nim

@@ -477,6 +477,18 @@ func update*[Pubkey, Sig: ECP_ShortW_Aff](
        signature: Sig): bool {.inline.} =
   ctx.update(pubkey, message, signature)
 
+func handover*(ctx: var BLSBatchSigAccumulator) {.inline.} =
+  ## Prepare accumulator for cheaper merging.
+  ##
+  ## In a multi-threaded context, multiple accumulators can be created and process subsets of the batch in parallel.
+  ## Accumulators can then be merged:
+  ##    merger_accumulator += mergee_accumulator
+  ## Merging will involve an expensive reduction operation when an accumulation threshold of 8 is reached.
+  ## However merging two reduced accumulators is 136x cheaper.
+  ##
+  ## `Handover` forces this reduction on local threads to limit the burden on the merger thread.
+  ctx.millerAccum.handover()
+
 func merge*(ctxDst: var BLSBatchSigAccumulator, ctxSrc: BLSBatchSigAccumulator): bool =
   ## Merge 2 BLS signature accumulators: ctxDst <- ctxDst + ctxSrc
   ##

constantine/signatures/bls_signatures_parallel.nim

@@ -154,6 +154,7 @@ proc batchVerify_parallel*[Msg, Pubkey, Sig](
       if not ctx[].update(pubkeys[i], messages[i], signatures[i]):
         return false
 
+    ctx[].handover()
     return true
 
   let partialStates = allocStackArray(Flowvar[bool], numAccums)
@@ -173,7 +174,7 @@ proc batchVerify_parallel*[Msg, Pubkey, Sig](
 
   # Stage 2: Reduce partial pairings
   # --------------------------------
-  if true: # numAccums < 4: # Linear merge
+  if numAccums < 4: # Linear merge
     result = sync partialStates[0]
     for i in 1 ..< numAccums:
       result = result and sync partialStates[i]