Implement receiving async events in TorControlClient.
ADD
WalletWasabi.Tests/UnitTests/Tor/Control/TorControlClientTests.cs
+using System;
+using System.Collections.Generic;
+using System.IO.Pipelines;
+using System.Threading;
+using System.Threading.Tasks;
+using WalletWasabi.Logging;
+using WalletWasabi.Tor.Control;
+using WalletWasabi.Tor.Control.Messages;
+using Xunit;
+
+namespace WalletWasabi.Tests.UnitTests.Tor.Control
+{
+ public class TorControlClientTests
+ {
+ /// <summary>Verifies that client receives correct async events from Tor.</summary>
+ [Fact]
+ public async Task ReceiveTorAsyncEventsUsingForeachAsync()
+ {
+ using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(120));
+
+ // Test parameters.
+ const int ExpectedEventsNo = 5;
+ const string AsyncEventContent = "CIRC 1000 EXTENDED moria1,moria2";
+
+ Pipe toServer = new();
+ Pipe toClient = new();
+
+ // Set up Tor control client.
+ await using TorControlClient client = new(pipeReader: toClient.Reader, pipeWriter: toServer.Writer);
+
+ // Subscribe to Tor events.
+ IAsyncEnumerable<TorControlReply> events = client.ReadEventsAsync(timeoutCts.Token);
+
+ // Send a Tor event to all subscribed clients (only one here).
+ // This must happen after a client is subscribed.
+ Task serverTask = Task.Run(async () =>
+ {
+ for (int i = 0; i < ExpectedEventsNo; i++)
+ {
+ Logger.LogTrace($"Server: Send async Tor event (#{i}): '650 {AsyncEventContent}'.");
+ await toClient.Writer.WriteAsciiAsync($"650 {AsyncEventContent}\r\n", timeoutCts.Token);
+ }
+ });
+
+ // Iterate received events.
+ int counter = 0;
+
+ // Client should get all the events.
+ await foreach (TorControlReply receivedEvent in events)
+ {
+ counter++;
+
+ Logger.LogTrace($"Client: Received event (#{counter}): '{receivedEvent}'.");
+ Assert.Equal(StatusCode.AsynchronousEventNotify, receivedEvent.StatusCode);
+ string line = Assert.Single(receivedEvent.ResponseLines);
+ Assert.Equal(AsyncEventContent, line);
+
+ if (counter == ExpectedEventsNo)
+ {
+ break;
+ }
+ }
+ }
+
+ /// <summary>Verifies a correct result of a mix of sync and async messages from Tor.</summary>
+ [Fact]
+ public async Task ReceivingMixOfSyncAndAsyncMessageFromTorControlAsync()
+ {
+ using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(120));
+
+ // Test parameters.
+ const string AsyncEventContent = "CIRC 1000 EXTENDED moria1,moria2";
+
+ Pipe toServer = new();
+ Pipe toClient = new();
+
+ // Set up Tor control client.
+ await using TorControlClient client = new(pipeReader: toClient.Reader, pipeWriter: toServer.Writer);
+
+ // Subscribe to Tor events.
+ IAsyncEnumerable<TorControlReply> events = client.ReadEventsAsync(timeoutCts.Token);
+ IAsyncEnumerator<TorControlReply> eventsEnumerator = events.GetAsyncEnumerator();
+ ValueTask<bool> firstReplyTask = eventsEnumerator.MoveNextAsync();
+
+ Task serverTask = Task.Run(async () =>
+ {
+ Logger.LogTrace($"Server: Send msg #1 (async) to client: '650 {AsyncEventContent}'.");
+ await toClient.Writer.WriteAsciiAsync($"650 {AsyncEventContent}\r\n", timeoutCts.Token);
+
+ Logger.LogTrace($"Server: Send msg #2 (async) to client: '650 {AsyncEventContent}'.");
+ await toClient.Writer.WriteAsciiAsync($"650 {AsyncEventContent}\r\n", timeoutCts.Token);
+
+ Logger.LogTrace("Server: Wait for TAKEOWNERSHIP command.");
+ string command = await toServer.Reader.ReadLineAsync(timeoutCts.Token);
+ Assert.Equal("TAKEOWNERSHIP", command);
+
+ Logger.LogTrace("Server: Send msg #3 (sync) to client in response to TAKEOWNERSHIP command.");
+ await toClient.Writer.WriteAsciiAsync($"250 OK\r\n", timeoutCts.Token);
+
+ Logger.LogTrace($"Server: Send msg #4 (async) to client: '650 {AsyncEventContent}'.");
+ await toClient.Writer.WriteAsciiAsync($"650 {AsyncEventContent}\r\n", timeoutCts.Token);
+ });
+
+ Logger.LogTrace("Client: Receive msg #1 (async).");
+ {
+ await firstReplyTask.AsTask().WithAwaitCancellationAsync(timeoutCts.Token);
+ TorControlReply receivedEvent1 = eventsEnumerator.Current;
+ Assert.Equal(StatusCode.AsynchronousEventNotify, receivedEvent1.StatusCode);
+ string line = Assert.Single(receivedEvent1.ResponseLines);
+ Assert.Equal(AsyncEventContent, line);
+ }
+
+ // Msg #2 is sent before expected reply (msg #3).
+ TorControlReply takeOwnershipReply = await client.TakeOwnershipAsync(timeoutCts.Token);
+ Assert.True(takeOwnershipReply.Success);
+
+ Logger.LogTrace("Client: Receive msg #2 (async).");
+ {
+ Assert.True(await eventsEnumerator.MoveNextAsync());
+ TorControlReply receivedEvent2 = eventsEnumerator.Current;
+ Assert.Equal(StatusCode.AsynchronousEventNotify, receivedEvent2.StatusCode);
+ string line = Assert.Single(receivedEvent2.ResponseLines);
+ Assert.Equal(AsyncEventContent, line);
+ }
+
+ Logger.LogTrace("Client: Receive msg #4 (async) - i.e. third async event.");
+ {
+ Assert.True(await eventsEnumerator.MoveNextAsync());
+ TorControlReply receivedEvent3 = eventsEnumerator.Current;
+ Assert.Equal(StatusCode.AsynchronousEventNotify, receivedEvent3.StatusCode);
+ string line = Assert.Single(receivedEvent3.ResponseLines);
+ Assert.Equal(AsyncEventContent, line);
+ }
+
+ // Client decided to stop reading Tor async events.
+ timeoutCts.Cancel();
+
+ // No more async events.
+ _ = Assert.ThrowsAsync<OperationCanceledException>(async () => await eventsEnumerator.MoveNextAsync());
+ }
+ }
+}

MODIFY
WalletWasabi/Tor/Control/ReplyCode.cs
WalletWasabi/Tor/Control/ReplyCode.cs
@@ -55,6 +55,7 @@ namespace WalletWasabi.Tor.Control
UnmanagedEntity = 555,
/// <summary>A notification sent following an asynchronous operation.</summary>
+ /// <seealso href="https://gitweb.torproject.org/torspec.git/tree/control-spec.txt">4.1. Asynchronous events</seealso>
AsynchronousEventNotify = 650,
}
}

MODIFY
WalletWasabi/Tor/Control/TorControlClient.cs
WalletWasabi/Tor/Control/TorControlClient.cs
+using Nito.AsyncEx;
using System;
+using System.Collections.Generic;
using System.IO.Pipelines;
using System.Net.Sockets;
+using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
+using System.Threading.Channels;
using System.Threading.Tasks;
using WalletWasabi.Logging;
using WalletWasabi.Tor.Control.Messages;
@@ -14,12 +18,15 @@ namespace WalletWasabi.Tor.Control
/// </summary>
public class TorControlClient : IAsyncDisposable
{
- public TorControlClient(TcpClient tcpClient)
+ private static readonly UnboundedChannelOptions Options = new()
{
- TcpClient = tcpClient;
+ SingleWriter = true
+ };
- PipeReader = PipeReader.Create(tcpClient.GetStream());
- PipeWriter = PipeWriter.Create(tcpClient.GetStream());
+ public TorControlClient(TcpClient tcpClient) :
+ this(PipeReader.Create(tcpClient.GetStream()), PipeWriter.Create(tcpClient.GetStream()))
+ {
+ TcpClient = tcpClient;
}
/// <summary>For testing.</summary>
@@ -28,11 +35,37 @@ namespace WalletWasabi.Tor.Control
TcpClient = null;
PipeReader = pipeReader;
PipeWriter = pipeWriter;
+ ReaderCts = new();
+
+ SyncChannel = Channel.CreateUnbounded<TorControlReply>(Options);
+ AsyncChannels = new List<Channel<TorControlReply>>();
+ ReaderLoopTask = Task.Run(ReaderLoopAsync);
+ MessageLock = new();
}
private TcpClient? TcpClient { get; }
private PipeReader PipeReader { get; }
private PipeWriter PipeWriter { get; }
+ private CancellationTokenSource ReaderCts { get; }
+ private Task ReaderLoopTask { get; }
+
+ /// <summary>Channel only for synchronous replies from Tor control.</summary>
+ /// <remarks>Typically, there is at most one message in the channel at a time.</remarks>
+ private Channel<TorControlReply> SyncChannel { get; }
+
+ /// <summary>Guards <see cref="AsyncChannels"/>.</summary>
+ private object AsyncChannelsLock { get; } = new();
+
+ /// <summary>Channel only for <see cref="StatusCode.AsynchronousEventNotify"/> events.</summary>
+ /// <remarks>
+ /// Guarded by <see cref="AsyncChannelsLock"/>.
+ /// <para>This list should be used only in a copy-on-write way to avoid iterating a modified list.</para>
+ /// </remarks>
+ private List<Channel<TorControlReply>> AsyncChannels { get; set; }
+
+ /// <summary>Lock to when sending a request to Tor control and waiting for a reply.</summary>
+ /// <remarks>Tor control protocol does not provide a foolproof way to recognize that a response belongs to a request.</remarks>
+ private AsyncLock MessageLock { get; }
/// <summary>
/// Gets protocol info (for version 1).
@@ -88,20 +121,115 @@ namespace WalletWasabi.Tor.Control
/// <param name="command">A Tor control command which must end with <c>\r\n</c>.</param>
public async Task<TorControlReply> SendCommandAsync(string command, CancellationToken cancellationToken = default)
{
+ using var _ = await MessageLock.LockAsync(cancellationToken).ConfigureAwait(false);
+
Logger.LogTrace($"Client: About to send command: '{command.TrimEnd()}'");
await PipeWriter.WriteAsync(new ReadOnlyMemory<byte>(Encoding.ASCII.GetBytes(command)), cancellationToken).ConfigureAwait(false);
await PipeWriter.FlushAsync(cancellationToken).ConfigureAwait(false);
- TorControlReply reply = await TorControlReplyReader.ReadReplyAsync(PipeReader, cancellationToken).ConfigureAwait(false);
-
+ TorControlReply reply = await SyncChannel.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
return reply;
}
- public ValueTask DisposeAsync()
+ /// <summary>Allows the caller to read Tor events using <c>await foreach</c>.</summary>
+ /// <remarks>Processing of replies does not block other readers.</remarks>
+ /// <example>
+ /// <code>
+ /// await foreach (TorControlReply reply in TorControlClient.ReadEventsAsync())
+ /// {
+ /// Console.WriteLine(reply.ToString());
+ /// }
+ /// </code>
+ /// </example>
+ public async IAsyncEnumerable<TorControlReply> ReadEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
{
- TcpClient?.Dispose();
+ Channel<TorControlReply> channel = Channel.CreateUnbounded<TorControlReply>(Options);
+ List<Channel<TorControlReply>> newList;
+
+ try
+ {
+ lock (AsyncChannelsLock)
+ {
+ newList = new(AsyncChannels);
+ newList.Add(channel);
- return ValueTask.CompletedTask;
+ AsyncChannels = newList;
+ }
+
+ Logger.LogTrace($"ReadEventsAsync: subscribers: {newList.Count}.");
+
+ await foreach (TorControlReply item in channel.Reader.ReadAllAsync(cancellationToken))
+ {
+ yield return item;
+ }
+ }
+ finally
+ {
+ Logger.LogTrace("ReadEventsAsync: About to unsubscribe.");
+ lock (AsyncChannelsLock)
+ {
+ newList = new(AsyncChannels);
+ newList.Remove(channel);
+
+ AsyncChannels = newList;
+ }
+ }
+ }
+
+ /// <summary>
+ /// Loop that continues reading received messages from Tor control TCP connection.
+ /// </summary>
+ private async Task ReaderLoopAsync()
+ {
+ try
+ {
+ while (!ReaderCts.IsCancellationRequested)
+ {
+ TorControlReply reply = await TorControlReplyReader.ReadReplyAsync(PipeReader, ReaderCts.Token).ConfigureAwait(false);
+
+ if (reply.StatusCode == StatusCode.AsynchronousEventNotify)
+ {
+ List<Channel<TorControlReply>> list;
+
+ lock (AsyncChannelsLock)
+ {
+ list = AsyncChannels;
+ }
+
+ // Notify every "subscriber" who reads all Tor events.
+ foreach (Channel<TorControlReply> channel in list)
+ {
+ await channel.Writer.WriteAsync(reply, ReaderCts.Token).ConfigureAwait(false);
+ }
+ }
+ else
+ {
+ // Propagate a response back to the requester.
+ await SyncChannel.Writer.WriteAsync(reply, ReaderCts.Token).ConfigureAwait(false);
+ }
+ }
+ }
+ catch (OperationCanceledException)
+ {
+ Logger.LogTrace("Reader loop was stopped.");
+ }
+ catch (Exception e)
+ {
+ // This is an unrecoverable issue.
+ Logger.LogError($"Exception occurred in the reader loop: {e}.");
+ throw;
+ }
+ }
+
+ public async ValueTask DisposeAsync()
+ {
+ // Stop reader loop.
+ ReaderCts.Cancel();
+
+ // Wait until the reader loop stops.
+ await ReaderLoopTask.ConfigureAwait(false);
+
+ TcpClient?.Dispose();
}
}
}

MODIFY
WalletWasabi/Tor/Control/TorControlClientFactory.cs
WalletWasabi/Tor/Control/TorControlClientFactory.cs
@@ -31,7 +31,7 @@ namespace WalletWasabi.Tor.Control
Random = random ?? new InsecureRandom();
}
- /// <summary>Helps generate nonces for auth challenges.</summary>
+ /// <summary>Helps generate nonces for AUTH challenges.</summary>
private IRandom Random { get; }
/// <summary>Connects to Tor Control endpoint and authenticates using safe-cookie mechanism.</summary>
@@ -55,7 +55,9 @@ namespace WalletWasabi.Tor.Control
}
finally
{
- if (clientToDispose is not null)
+ // `!=` instead `is not null` to avoid CA2000. The analyzer should be fixed over time.
+ // https://github.com/dotnet/roslyn-analyzers/issues/4981
+ if (clientToDispose != null)
{
await clientToDispose.DisposeAsync().ConfigureAwait(false);
}

Implement receiving async events in TorControlClient.