Parallelize per-member ECIES encryption and commitment computation (#494)

community/base/src/main/scala/com/digitalasset/canton/crypto/InteractiveSubmission.scala

@@ -25,7 +25,7 @@ import com.digitalasset.daml.lf.value.Value.ContractId
 
 import java.util.UUID
 import scala.collection.immutable.{SortedMap, SortedSet}
-import scala.concurrent.ExecutionContext
+import scala.concurrent.{ExecutionContext, Future}
 import scala.util.{Failure, Success, Try}
 
 object InteractiveSubmission {
@@ -219,19 +219,28 @@ object InteractiveSubmission {
             )
         )
 
-        (invalidSignatures, validSignatures) = signatures.map { signature =>
-          signingKeysWithThreshold.keys
-            .find(_.fingerprint == signature.authorizingLongTermKey)
-            .toRight(
-              s"Signing key ${signature.authorizingLongTermKey} is not a valid key for $party"
-            )
-            .flatMap(key =>
-              cryptoPureApi
-                .verifySignature(hash.unwrap, key, signature, SigningKeyUsage.ProtocolOnly)
-                .map(_ => key.fingerprint)
-                .leftMap(_.toString)
-            )
-        }.separate
+        // Parallelize per-signature verification within a party. Each verification is
+        // pure CPU (ECDSA/Ed25519) with no data dependency between signatures.
+        verificationResults <- EitherT.right[String](
+          FutureUnlessShutdown.outcomeF(
+            Future.traverse(signatures.toList) { signature =>
+              Future {
+                signingKeysWithThreshold.keys
+                  .find(_.fingerprint == signature.authorizingLongTermKey)
+                  .toRight(
+                    s"Signing key ${signature.authorizingLongTermKey} is not a valid key for $party"
+                  )
+                  .flatMap(key =>
+                    cryptoPureApi
+                      .verifySignature(hash.unwrap, key, signature, SigningKeyUsage.ProtocolOnly)
+                      .map(_ => key.fingerprint)
+                      .leftMap(_.toString)
+                  )
+              }
+            }
+          )
+        )
+        (invalidSignatures, validSignatures) = verificationResults.separate
         validSignaturesSet = validSignatures.toSet
         _ = {
           // Log invalid signatures at info level because it is unexpected,

community/base/src/main/scala/com/digitalasset/canton/crypto/SyncCryptoApiParticipantProvider.scala

@@ -6,6 +6,7 @@ package com.digitalasset.canton.crypto
 import cats.data.EitherT
 import cats.syntax.either.*
 import cats.syntax.functor.*
+import cats.syntax.parallel.*
 import cats.syntax.traverse.*
 import com.daml.nonempty.NonEmpty
 import com.digitalasset.canton.checked
@@ -620,10 +621,16 @@ class SynchronizerSnapshotSyncCryptoApi(
     EitherT(
       ipsSnapshot
         .encryptionKey(members)
-        .map { keys =>
-          members
-            .traverse(encryptFor(keys))
-            .map(_.toMap)
+        .flatMap { keys =>
+          // Parallelize per-member asymmetric encryption. Each encryption is pure CPU
+          // (ECIES/RSA key wrapping) with no data dependency between members.
+          members.toList
+            .parTraverse { member =>
+              FutureUnlessShutdown.outcomeF(
+                scala.concurrent.Future(encryptFor(keys)(member))
+              )
+            }
+            .map(_.sequence.map(_.toMap))
         }
     )
   }

community/participant/src/main/scala/com/digitalasset/canton/participant/pruning/AcsCommitmentProcessor.scala

@@ -3127,17 +3127,29 @@ object AcsCommitmentProcessor extends HasLoggerName {
   private[pruning] def computeCommitmentsPerParticipant(
       cmts: Map[ParticipantId, Map[SortedSet[InternedPartyId], AcsCommitment.CommitmentType]],
       cachedCommitments: CachedCommitments,
-  ): Map[ParticipantId, AcsCommitment.CommitmentType] =
-    cmts.map { case (p, hashes) =>
-      (
-        p,
-        cachedCommitments
-          .computeCmtFromCached(p, hashes)
-          .getOrElse(
-            commitmentsFromStkhdCmts(hashes.values.toSeq.filter(_ != emptyCommitment))
-          ),
-      )
+  ): Map[ParticipantId, AcsCommitment.CommitmentType] = {
+    // Parallelize across counter-participants. Each participant's commitment hash
+    // is computed independently (LtHash16 XOR of its stakeholder group commitments).
+    // With 50+ counter-participants this removes a sequential bottleneck at each
+    // reconciliation interval.
+    import java.util.concurrent.{ConcurrentHashMap, ForkJoinPool}
+    val result = new ConcurrentHashMap[ParticipantId, AcsCommitment.CommitmentType]()
+    val tasks = cmts.map { case (p, hashes) =>
+      ForkJoinPool.commonPool().submit(new Runnable {
+        def run(): Unit = {
+          val cmt = cachedCommitments
+            .computeCmtFromCached(p, hashes)
+            .getOrElse(
+              commitmentsFromStkhdCmts(hashes.values.toSeq.filter(_ != emptyCommitment))
+            )
+          result.put(p, cmt)
+        }
+      })
     }
+    tasks.foreach(_.get()) // wait for all
+    import scala.jdk.CollectionConverters.*
+    result.asScala.toMap
+  }
 
   @VisibleForTesting
   private[pruning] def commitmentsFromStkhdCmts(

community/participant/src/main/scala/com/digitalasset/canton/participant/pruning/RunningCommitments.scala

@@ -77,33 +77,47 @@ abstract class GenericRunningCommitments[T: Pretty](
   ): Unit =
     lock.exclusive {
       this.rt = rt
-      change.activations.foreach { case (cid, stakeholdersAndReassignmentCounter) =>
-        val sortedStakeholders =
-          SortedSet(stakeholdersAndReassignmentCounter.stakeholders.toSeq*)
-        val h = commitments.getOrElseUpdate(sortedStakeholders, LtHash16())
-        AcsCommitmentProcessor.addContractToCommitmentDigest(
-          h,
-          cid,
-          stakeholdersAndReassignmentCounter.reassignmentCounter,
-        )
-        loggingContext.debug(
-          s"Adding to commitment activation cid $cid reassignmentCounter ${stakeholdersAndReassignmentCounter.reassignmentCounter}"
-        )
-        deltaB += sortedStakeholders -> h
+
+      // Group activations and deactivations by stakeholder set so we can
+      // process independent groups in parallel. Within each group, operations
+      // are serial (same mutable LtHash16 accumulator).
+      val activationsByGroup = change.activations.groupBy { case (_, sr) =>
+        SortedSet(sr.stakeholders.toSeq*)
       }
-      change.deactivations.foreach { case (cid, stakeholdersAndReassignmentCounter) =>
-        val sortedStakeholders =
-          SortedSet(stakeholdersAndReassignmentCounter.stakeholders.toSeq*)
-        val h = commitments.getOrElseUpdate(sortedStakeholders, LtHash16())
-        AcsCommitmentProcessor.removeContractFromCommitmentDigest(
-          h,
-          cid,
-          stakeholdersAndReassignmentCounter.reassignmentCounter,
-        )
-        loggingContext.debug(
-          s"Removing from commitment deactivation cid $cid reassignmentCounter ${stakeholdersAndReassignmentCounter.reassignmentCounter}"
-        )
-        deltaB += sortedStakeholders -> h
+      val deactivationsByGroup = change.deactivations.groupBy { case (_, sr) =>
+        SortedSet(sr.stakeholders.toSeq*)
+      }
+      val allGroups = (activationsByGroup.keySet ++ deactivationsByGroup.keySet).toSeq
+
+      // Ensure all LtHash16 accumulators exist before parallel access
+      allGroups.foreach(stkhs => commitments.getOrElseUpdate(stkhs, LtHash16()))
+
+      // Process groups in parallel — each group has its own LtHash16
+      val pool = java.util.concurrent.ForkJoinPool.commonPool()
+      val tasks = allGroups.map { sortedStakeholders =>
+        pool.submit(new Runnable {
+          def run(): Unit = {
+            val h = commitments(sortedStakeholders)
+            activationsByGroup.getOrElse(sortedStakeholders, Map.empty).foreach {
+              case (cid, sr) =>
+                AcsCommitmentProcessor.addContractToCommitmentDigest(h, cid, sr.reassignmentCounter)
+            }
+            deactivationsByGroup.getOrElse(sortedStakeholders, Map.empty).foreach {
+              case (cid, sr) =>
+                AcsCommitmentProcessor.removeContractFromCommitmentDigest(
+                  h,
+                  cid,
+                  sr.reassignmentCounter,
+                )
+            }
+          }
+        })
+      }
+      tasks.foreach(_.get())
+
+      // Update delta tracking (single-threaded, deltaB is not thread-safe)
+      allGroups.foreach { sortedStakeholders =>
+        deltaB += sortedStakeholders -> commitments(sortedStakeholders)
       }
     }