Schedule instead of direct start
MODIFY
WalletWasabi/WabiSabi/Client/CoinJoinManager.cs
WalletWasabi/WabiSabi/Client/CoinJoinManager.cs
@@ -116,9 +116,10 @@ public class CoinJoinManager : BackgroundService
// This is a shared resource and that's why it is concurrent. Alternatives are locking structures,
// using a single lock around its access or use a channel.
var trackedCoinJoins = new ConcurrentDictionary<string, CoinJoinTracker>();
+ var trackedAutoStarts = new ConcurrentDictionary<Wallet, TrackedAutoStart>();
- var commandsHandlingTask = Task.Run(() => HandleCoinJoinCommandsAsync(trackedCoinJoins, stoppingToken), stoppingToken);
- var monitorCoinJoinTask = Task.Run(() => MonitorAndHandlingCoinJoinFinallizationAsync(trackedCoinJoins, stoppingToken), stoppingToken);
+ var commandsHandlingTask = Task.Run(() => HandleCoinJoinCommandsAsync(trackedCoinJoins, trackedAutoStarts, stoppingToken), stoppingToken);
+ var monitorCoinJoinTask = Task.Run(() => MonitorAndHandlingCoinJoinFinallizationAsync(trackedCoinJoins, trackedAutoStarts, stoppingToken), stoppingToken);
await Task.WhenAny(commandsHandlingTask, monitorCoinJoinTask).ConfigureAwait(false);
@@ -126,10 +127,9 @@ public class CoinJoinManager : BackgroundService
await WaitAndHandleResultOfTasksAsync(nameof(monitorCoinJoinTask), monitorCoinJoinTask).ConfigureAwait(false);
}
- private async Task HandleCoinJoinCommandsAsync(ConcurrentDictionary<string, CoinJoinTracker> trackedCoinJoins, CancellationToken stoppingToken)
+ private async Task HandleCoinJoinCommandsAsync(ConcurrentDictionary<string, CoinJoinTracker> trackedCoinJoins, ConcurrentDictionary<Wallet, TrackedAutoStart> trackedAutoStarts, CancellationToken stoppingToken)
{
var coinJoinTrackerFactory = new CoinJoinTrackerFactory(HttpClientFactory, RoundStatusUpdater, stoppingToken);
- var trackedAutoStarts = new ConcurrentDictionary<Wallet, TrackedAutoStart>();
void StartCoinJoinCommand(StartCoinJoinCommand startCommand)
{
@@ -144,7 +144,7 @@ public class CoinJoinManager : BackgroundService
if (IsUserInSendWorkflow)
{
- ScheduleRestartAutomatically(walletToStart, startCommand.StopWhenAllMixed, startCommand.OverridePlebStop);
+ ScheduleRestartAutomatically(walletToStart, trackedAutoStarts, startCommand.StopWhenAllMixed, startCommand.OverridePlebStop, stoppingToken);
NotifyCoinJoinStartError(walletToStart, CoinjoinError.UserInSendWorkflow);
return;
@@ -153,7 +153,7 @@ public class CoinJoinManager : BackgroundService
if (walletToStart.IsUnderPlebStop && !startCommand.OverridePlebStop)
{
walletToStart.LogDebug("PlebStop preventing coinjoin.");
- ScheduleRestartAutomatically(walletToStart, startCommand.StopWhenAllMixed, startCommand.OverridePlebStop);
+ ScheduleRestartAutomatically(walletToStart, trackedAutoStarts, startCommand.StopWhenAllMixed, startCommand.OverridePlebStop, stoppingToken);
NotifyCoinJoinStartError(walletToStart, CoinjoinError.NotEnoughUnprivateBalance);
return;
@@ -161,7 +161,7 @@ public class CoinJoinManager : BackgroundService
if (WalletManager.Synchronizer?.LastResponse is not { } synchronizerResponse)
{
- ScheduleRestartAutomatically(walletToStart, startCommand.StopWhenAllMixed, startCommand.OverridePlebStop);
+ ScheduleRestartAutomatically(walletToStart, trackedAutoStarts, startCommand.StopWhenAllMixed, startCommand.OverridePlebStop, stoppingToken);
NotifyCoinJoinStartError(walletToStart, CoinjoinError.BackendNotSynchronized);
return;
@@ -172,7 +172,7 @@ public class CoinJoinManager : BackgroundService
walletToStart.LogDebug("All mixed!");
if (!startCommand.StopWhenAllMixed)
{
- ScheduleRestartAutomatically(walletToStart, startCommand.StopWhenAllMixed, startCommand.OverridePlebStop);
+ ScheduleRestartAutomatically(walletToStart, trackedAutoStarts, startCommand.StopWhenAllMixed, startCommand.OverridePlebStop, stoppingToken);
}
NotifyCoinJoinStartError(walletToStart, CoinjoinError.AllCoinsPrivate);
@@ -183,7 +183,7 @@ public class CoinJoinManager : BackgroundService
if (coinCandidates.Length == 0)
{
walletToStart.LogDebug("No candidate coins available to mix.");
- ScheduleRestartAutomatically(walletToStart, startCommand.StopWhenAllMixed, startCommand.OverridePlebStop);
+ ScheduleRestartAutomatically(walletToStart, trackedAutoStarts, startCommand.StopWhenAllMixed, startCommand.OverridePlebStop, stoppingToken);
NotifyCoinJoinStartError(walletToStart, CoinjoinError.NoCoinsToMix);
return;
@@ -226,7 +226,41 @@ public class CoinJoinManager : BackgroundService
}
}
- void ScheduleRestartAutomatically(Wallet walletToStart, bool stopWhenAllMixed, bool overridePlebStop)
+ void TryRemoveTrackedAutoStart(Wallet wallet)
+ {
+ if (trackedAutoStarts.TryRemove(wallet, out var trackedAutoStart))
+ {
+ trackedAutoStart.CancellationTokenSource.Cancel();
+ trackedAutoStart.CancellationTokenSource.Dispose();
+ }
+ }
+
+ while (!stoppingToken.IsCancellationRequested)
+ {
+ var command = await CommandChannel.Reader.ReadAsync(stoppingToken).ConfigureAwait(false);
+
+ switch (command)
+ {
+ case StartCoinJoinCommand startCommand:
+ StartCoinJoinCommand(startCommand);
+ break;
+
+ case StopCoinJoinCommand stopCommand:
+ StopCoinJoinCommand(stopCommand);
+ break;
+ }
+ }
+
+ foreach (var trackedAutoStart in trackedAutoStarts.Values)
+ {
+ trackedAutoStart.CancellationTokenSource.Cancel();
+ trackedAutoStart.CancellationTokenSource.Dispose();
+ }
+
+ await WaitAndHandleResultOfTasksAsync(nameof(trackedAutoStarts), trackedAutoStarts.Values.Select(x => x.Task).ToArray()).ConfigureAwait(false);
+ }
+
+ private void ScheduleRestartAutomatically(Wallet walletToStart, ConcurrentDictionary<Wallet, TrackedAutoStart> trackedAutoStarts, bool stopWhenAllMixed, bool overridePlebStop, CancellationToken stoppingToken)
{
if (trackedAutoStarts.ContainsKey(walletToStart))
{
@@ -269,50 +303,28 @@ public class CoinJoinManager : BackgroundService
}
}
- void TryRemoveTrackedAutoStart(Wallet wallet)
+ private async Task MonitorAndHandlingCoinJoinFinallizationAsync(ConcurrentDictionary<string, CoinJoinTracker> trackedCoinJoins, ConcurrentDictionary<Wallet, TrackedAutoStart> trackedAutoStarts, CancellationToken stoppingToken)
{
- if (trackedAutoStarts.TryRemove(wallet, out var trackedAutoStart))
- {
- trackedAutoStart.CancellationTokenSource.Cancel();
- trackedAutoStart.CancellationTokenSource.Dispose();
- }
- }
-
while (!stoppingToken.IsCancellationRequested)
{
- var command = await CommandChannel.Reader.ReadAsync(stoppingToken).ConfigureAwait(false);
-
- switch (command)
+ // Handles coinjoin finalization and notification.
+ var finishedCoinJoins = trackedCoinJoins.Where(x => x.Value.IsCompleted).Select(x => x.Value).ToImmutableArray();
+ foreach (var finishedCoinJoin in finishedCoinJoins)
{
- case StartCoinJoinCommand startCommand:
- StartCoinJoinCommand(startCommand);
- break;
+ await HandleCoinJoinFinalizationAsync(finishedCoinJoin, trackedCoinJoins, stoppingToken).ConfigureAwait(false);
- case StopCoinJoinCommand stopCommand:
- StopCoinJoinCommand(stopCommand);
- break;
- }
- }
+ NotifyCoinJoinCompletion(finishedCoinJoin);
- foreach (var trackedAutoStart in trackedAutoStarts.Values)
+ if (!finishedCoinJoin.IsStopped && !stoppingToken.IsCancellationRequested)
{
- trackedAutoStart.CancellationTokenSource.Cancel();
- trackedAutoStart.CancellationTokenSource.Dispose();
- }
+ finishedCoinJoin.Wallet.LogInfo($"{nameof(CoinJoinClient)} restart automatically.");
- await WaitAndHandleResultOfTasksAsync(nameof(trackedAutoStarts), trackedAutoStarts.Values.Select(x => x.Task).ToArray()).ConfigureAwait(false);
+ ScheduleRestartAutomatically(finishedCoinJoin.Wallet, trackedAutoStarts, finishedCoinJoin.StopWhenAllMixed, finishedCoinJoin.OverridePlebStop, stoppingToken);
}
-
- private async Task MonitorAndHandlingCoinJoinFinallizationAsync(ConcurrentDictionary<string, CoinJoinTracker> trackedCoinJoins, CancellationToken stoppingToken)
- {
- while (!stoppingToken.IsCancellationRequested)
- {
- // Handles coinjoin finalization and notification.
- var finishedCoinJoins = trackedCoinJoins.Where(x => x.Value.IsCompleted).Select(x => x.Value).ToImmutableArray();
- foreach (var finishedCoinJoin in finishedCoinJoins)
+ else
{
- await HandleCoinJoinFinalizationAsync(finishedCoinJoin, trackedCoinJoins, stoppingToken).ConfigureAwait(false);
- NotifyCoinJoinCompletion(finishedCoinJoin);
+ NotifyWalletStoppedCoinJoin(finishedCoinJoin.Wallet);
+ }
}
// Updates the highest coinjoin client state.
var inProgress = trackedCoinJoins.Values.Where(wtd => !wtd.IsCompleted).ToImmutableArray();
@@ -379,17 +391,6 @@ public class CoinJoinManager : BackgroundService
finishedCoinJoin.WalletCoinJoinProgressChanged -= CoinJoinTracker_WalletCoinJoinProgressChanged;
finishedCoinJoin.Dispose();
}
-
- if (!finishedCoinJoin.IsStopped && !cancellationToken.IsCancellationRequested)
- {
- wallet.LogInfo($"{nameof(CoinJoinClient)} restart automatically.");
-
- await StartAsync(finishedCoinJoin.Wallet, finishedCoinJoin.StopWhenAllMixed, finishedCoinJoin.OverridePlebStop, cancellationToken).ConfigureAwait(false);
- }
- else
- {
- NotifyWalletStoppedCoinJoin(wallet);
- }
}
/// <summary>

Schedule instead of direct start