Implement smarter batching.
MODIFY
WalletWasabi/CoinJoin/Coordinator/Rounds/CoordinatorRound.cs
WalletWasabi/CoinJoin/Coordinator/Rounds/CoordinatorRound.cs
@@ -1186,32 +1186,20 @@ namespace WalletWasabi.CoinJoin.Coordinator.Rounds
}
Logger.LogInfo($"Mempool acceptance is unsuccessful! Number of Alices: {Alices.Count}.");
- var alicesSpent = new HashSet<Alice>();
- var alicesUnconfirmed = new HashSet<Alice>();
// The created tx was not accepted. Let's figure out why. Is it because an Alice doublespent or because of too long mempool chains.
- var checkingTasks = new List<(Alice alice, Task<GetTxOutResponse> task)>();
- var batch = RpcClient.PrepareBatch();
- foreach (Alice alice in Alices)
- {
- foreach (var input in alice.Inputs.Select(x => x.Outpoint))
- {
- checkingTasks.Add((alice, batch.GetTxOutAsync(input.Hash, (int)input.N, includeMempool: true)));
- }
- }
- var waiting = Task.WhenAll(checkingTasks.Select(t => t.task));
- await batch.SendBatchAsync().ConfigureAwait(false);
- await waiting.ConfigureAwait(false);
+ var responses = await GetTxOutForAllInputsAsync().ConfigureAwait(false);
- foreach (var t in checkingTasks)
+ var alicesSpent = new HashSet<Alice>();
+ var alicesUnconfirmed = new HashSet<Alice>();
+ foreach (var (alice, resp) in responses)
{
- var response = await t.task.ConfigureAwait(false);
- if (response is null)
+ if (resp is null)
{
- alicesSpent.Add(t.alice);
+ alicesSpent.Add(alice);
}
- else if (response.Confirmations <= 0)
+ else if (resp.Confirmations <= 0)
{
- alicesUnconfirmed.Add(t.alice);
+ alicesUnconfirmed.Add(alice);
}
}
@@ -1247,6 +1235,44 @@ namespace WalletWasabi.CoinJoin.Coordinator.Rounds
}
}
+ private async Task<List<(Alice alice, GetTxOutResponse resp)>> GetTxOutForAllInputsAsync()
+ {
+ var responses = new List<(Alice alice, GetTxOutResponse resp)>();
+
+ var inputAliceDic = new Dictionary<OutPoint, Alice>();
+ foreach (Alice alice in Alices)
+ {
+ foreach (var input in alice.Inputs.Select(x => x.Outpoint))
+ {
+ inputAliceDic.Add(input, alice);
+ }
+ }
+
+ foreach (var dicBatch in inputAliceDic.Batch(8)) // 8 is default rpcworkqueue/2, so other requests can go.
+ {
+ var checkingTasks = new List<(Alice alice, Task<GetTxOutResponse> task)>();
+ var batch = RpcClient.PrepareBatch();
+
+ foreach (var aliceInput in dicBatch)
+ {
+ var alice = aliceInput.Value;
+ var input = aliceInput.Key;
+ checkingTasks.Add((alice, batch.GetTxOutAsync(input.Hash, (int)input.N, includeMempool: true)));
+ }
+ var waiting = Task.WhenAll(checkingTasks.Select(t => t.task)); // Not sure this waiting is needed, but no test gets here, so oh well.
+ await batch.SendBatchAsync().ConfigureAwait(false);
+ await waiting.ConfigureAwait(false);
+
+ foreach (var t in checkingTasks)
+ {
+ var resp = await t.task.ConfigureAwait(false);
+ responses.Add((t.alice, resp));
+ }
+ }
+
+ return responses;
+ }
+
private async Task<(bool accept, string rejectReason)> TestMempoolAcceptWithTransactionSimulationAsync()
{
// Check if mempool would accept a fake transaction created with all the registered inputs.

MODIFY
WalletWasabi/Extensions/LinqExtensions.cs
WalletWasabi/Extensions/LinqExtensions.cs
@@ -7,6 +7,36 @@ namespace System.Linq
{
public static class LinqExtensions
{
+ public static IEnumerable<IEnumerable<T>> Batch<T>(
+ this IEnumerable<T> source, int size)
+ {
+ T[] bucket = null;
+ var count = 0;
+
+ foreach (var item in source)
+ {
+ if (bucket == null)
+ bucket = new T[size];
+
+ bucket[count++] = item;
+
+ if (count != size)
+ continue;
+
+ yield return bucket.Select(x => x);
+
+ bucket = null;
+ count = 0;
+ }
+
+ // Return the last bucket with all remaining elements
+ if (bucket != null && count > 0)
+ {
+ Array.Resize(ref bucket, count);
+ yield return bucket.Select(x => x);
+ }
+ }
+
public static T RandomElement<T>(this IEnumerable<T> source)
{
T current = default;

Implement smarter batching.