Add parrallelization and timing to Output registration
MODIFY
WalletWasabi/Extensions/TimeSpanExtensions.cs
WalletWasabi/Extensions/TimeSpanExtensions.cs
@@ -2,20 +2,28 @@ using System.Collections.Immutable;
using System.Linq;
using WalletWasabi.Crypto.Randomness;
+#pragma warning disable IDE0130 // Namespace does not match folder structure
+
namespace System;
+#pragma warning restore IDE0130 // Namespace does not match folder structure
public static class TimeSpanExtensions
{
public static ImmutableList<DateTimeOffset> SamplePoisson(this TimeSpan timeFrame, int numberOfEvents)
{
var startTime = DateTimeOffset.UtcNow;
+ return timeFrame.SamplePoissonDelays(numberOfEvents).Select(delay => startTime + delay).ToImmutableList();
+ }
+
+ public static ImmutableList<TimeSpan> SamplePoissonDelays(this TimeSpan timeFrame, int numberOfEvents)
+ {
using var random = new SecureRandom();
TimeSpan Sample(int milliseconds) =>
milliseconds <= 0 ? TimeSpan.Zero : TimeSpan.FromMilliseconds(random.GetInt(0, milliseconds));
return Enumerable
.Range(0, numberOfEvents)
- .Select(_ => startTime + (0.8 * Sample((int)timeFrame.TotalMilliseconds)))
+ .Select(_ => 0.8 * Sample((int)timeFrame.TotalMilliseconds))
.OrderBy(t => t)
.ToImmutableList();
}

MODIFY
WalletWasabi/WabiSabi/Client/CoinJoinClient.cs
WalletWasabi/WabiSabi/Client/CoinJoinClient.cs
@@ -182,9 +182,11 @@ public class CoinJoinClient
// Output registration.
roundState = await RoundStatusUpdater.CreateRoundAwaiter(roundState.Id, Phase.OutputRegistration, cancellationToken).ConfigureAwait(false);
- Logger.LogDebug($"Round ({roundState.Id}): Output registration phase started.");
- await scheduler.StartOutputRegistrationsAsync(outputTxOuts, bobClient, KeyChain, cancellationToken).ConfigureAwait(false);
+ var outputRegistrationEndTime = DateTimeOffset.UtcNow + roundState.OutputRegistrationTimeout - RoundStatusUpdater.Period;
+ Logger.LogDebug($"Round ({roundState.Id}): Output registration phase started - estimated end time: {outputRegistrationEndTime:hh\\:mm\\:ss}");
+
+ await scheduler.StartOutputRegistrationsAsync(outputTxOuts, bobClient, KeyChain, outputRegistrationEndTime, cancellationToken).ConfigureAwait(false);
Logger.LogDebug($"Round ({roundState.Id}): Outputs({outputTxOuts.Count()}) successfully registered.");
// ReadyToSign.

MODIFY
WalletWasabi/WabiSabi/Client/DependencyGraphTaskScheduler.cs
WalletWasabi/WabiSabi/Client/DependencyGraphTaskScheduler.cs
using NBitcoin;
using System.Collections.Generic;
+using System.Collections.Immutable;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
@@ -141,14 +142,12 @@ public class DependencyGraphTaskScheduler
}
}
- public async Task StartOutputRegistrationsAsync(IEnumerable<TxOut> txOuts, BobClient bobClient, IKeyChain keyChain, CancellationToken cancellationToken)
+ public async Task StartOutputRegistrationsAsync(IEnumerable<TxOut> txOuts, BobClient bobClient, IKeyChain keyChain, DateTimeOffset outputRegistrationEndTime, CancellationToken cancellationToken)
{
- List<Task> outputTasks = new();
-
using CancellationTokenSource ctsOnError = new();
using CancellationTokenSource linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, ctsOnError.Token);
- foreach (var (node, txOut) in Enumerable.Zip(Graph.Outputs, txOuts))
+ var nodes = Graph.Outputs.Select(node =>
{
var amountCredsToPresentTasks = Graph.InEdges(node, CredentialType.Amount).Select(edge => DependencyTasks[edge].Task);
var vsizeCredsToPresentTasks = Graph.InEdges(node, CredentialType.Vsize).Select(edge => DependencyTasks[edge].Task);
@@ -158,15 +157,26 @@ public class DependencyGraphTaskScheduler
vsizeCredsToPresentTasks,
Array.Empty<TaskCompletionSource<Credential>>(),
Array.Empty<TaskCompletionSource<Credential>>());
+ return smartRequestNode;
+ });
- var task = smartRequestNode
- .StartOutputRegistrationAsync(bobClient, txOut.ScriptPubKey, cancellationToken)
- .ContinueWith((t) =>
+ var remainingTime = outputRegistrationEndTime - DateTimeOffset.UtcNow;
+ if (remainingTime < TimeSpan.FromSeconds(5))
{
- if (t.IsFaulted && t.Exception is { } exception)
+ throw new InvalidOperationException("No time to register the outputs, aborting.");
+ }
+
+ var delays = remainingTime.SamplePoissonDelays(txOuts.Count());
+
+ var tasks = txOuts.Zip(nodes, delays,
+ async (txOut, smartRequestNode, delay) =>
{
- if (exception.InnerExceptions.Select(e => e.InnerException).Any(x => x is WabiSabiProtocolException wabisabiexc
- && wabisabiexc.ErrorCode == WabiSabiProtocolErrorCode.AlreadyRegisteredScript))
+ try
+ {
+ await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
+ await smartRequestNode.StartOutputRegistrationAsync(bobClient, txOut.ScriptPubKey, cancellationToken).ConfigureAwait(false);
+ }
+ catch (WabiSabiProtocolException ex) when (ex.ErrorCode == WabiSabiProtocolErrorCode.AlreadyRegisteredScript)
{
if (keyChain is KeyChain { KeyManager: var keyManager }
&& keyManager.TryGetKeyForScriptPubKey(txOut.ScriptPubKey, out var hdPubKey))
@@ -174,13 +184,10 @@ public class DependencyGraphTaskScheduler
hdPubKey.SetKeyState(KeyState.Used);
}
}
-
- throw exception;
}
- }, linkedCts.Token);
+ ).ToImmutableArray();
- await task.ConfigureAwait(false);
- }
+ await Task.WhenAll(tasks).ConfigureAwait(false);
}
private IEnumerable<(AliceClient AliceClient, InputNode Node)> PairAliceClientAndRequestNodes(IEnumerable<AliceClient> aliceClients, DependencyGraph graph)

Add parrallelization and timing to Output registration