TorControlClient: Add subscription API.
MODIFY
WalletWasabi.Tests/UnitTests/Tor/Control/TorControlClientTests.cs
WalletWasabi.Tests/UnitTests/Tor/Control/TorControlClientTests.cs
@@ -144,5 +144,85 @@ namespace WalletWasabi.Tests.UnitTests.Tor.Control
// No more async events.
_ = Assert.ThrowsAsync<OperationCanceledException>(async () => await eventsEnumerator.MoveNextAsync());
}
+
+ /// <summary>Verifies behavior of the subscription API and its logical subscription model.</summary>
+ [Fact]
+ public async Task SubscribeAndUnsubscribeAsync()
+ {
+ using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(120));
+
+ Pipe toServer = new();
+ Pipe toClient = new();
+
+ // Set up Tor control client.
+ await using TorControlClient client = new(pipeReader: toClient.Reader, pipeWriter: toServer.Writer);
+
+ Logger.LogTrace("Client: Subscribe 'CIRC' events.");
+ {
+ Task task = client.SubscribeEventsAsync(new string[] { "CIRC" }, timeoutCts.Token);
+
+ Logger.LogTrace("Server: Wait for 'SETEVENTS CIRC' command.");
+ string command = await toServer.Reader.ReadLineAsync(timeoutCts.Token);
+ Assert.Equal("SETEVENTS CIRC", command);
+
+ Logger.LogTrace("Server: Reply with OK code.");
+ await toClient.Writer.WriteAsciiAsync("250 OK\r\n", timeoutCts.Token);
+
+ await task;
+ }
+
+ Logger.LogTrace("Client: Subscribe 'CIRC' (already subscribed) and 'STATUS_CLIENT' (not subscribed) events.");
+ {
+ Task task = client.SubscribeEventsAsync(new string[] { "CIRC", "STATUS_CLIENT" }, timeoutCts.Token);
+
+ // CIRC is already subscribed.
+ Logger.LogTrace("Server: Wait for 'SETEVENTS CIRC,STATUS_CLIENT' command.");
+ string command = await toServer.Reader.ReadLineAsync(timeoutCts.Token);
+
+ // This means that BOTH 'CIRC' and 'STATUS_CLIENT' must be subscribed now.
+ // Note: Given we count logical event subscriptions, 'CIRC' is now (logically) subscribed twice!
+ Assert.Equal("SETEVENTS CIRC,STATUS_CLIENT", command);
+
+ Logger.LogTrace("Server: Reply with OK code.");
+ await toClient.Writer.WriteAsciiAsync("250 OK\r\n", timeoutCts.Token);
+
+ await task;
+ }
+
+ Logger.LogTrace("Client: Unsubscribe 'CIRC' and 'STATUS_CLIENT' events.");
+ {
+ Task task = client.UnsubscribeEventsAsync(new string[] { "CIRC", "STATUS_CLIENT" }, timeoutCts.Token);
+
+ // CIRC is already subscribed.
+ Logger.LogTrace("Server: Wait for 'SETEVENTS CIRC' command.");
+ string command = await toServer.Reader.ReadLineAsync(timeoutCts.Token);
+
+ // This means that CIRC is still subscribed (!). The reason for that is that we count logical subscriptions,
+ // so when two distinct components can work the subscription API and they don't affect each other.
+ Assert.Equal("SETEVENTS CIRC", command);
+
+ Logger.LogTrace("Server: Reply with OK code.");
+ await toClient.Writer.WriteAsciiAsync("250 OK\r\n", timeoutCts.Token);
+
+ await task;
+ }
+
+ Logger.LogTrace("Client: Unsubscribe 'CIRC' events.");
+ {
+ Task task = client.UnsubscribeEventsAsync(new string[] { "CIRC" }, timeoutCts.Token);
+
+ // CIRC is already subscribed.
+ Logger.LogTrace("Server: Wait for 'SETEVENTS' command.");
+ string command = await toServer.Reader.ReadLineAsync(timeoutCts.Token);
+
+ // This means that no events are subscribed.
+ Assert.Equal("SETEVENTS", command);
+
+ Logger.LogTrace("Server: Reply with OK code.");
+ await toClient.Writer.WriteAsciiAsync("250 OK\r\n", timeoutCts.Token);
+
+ await task;
+ }
+ }
}
}

MODIFY
WalletWasabi.Tests/UnitTests/Tor/Control/Utils/AsyncEventParserTests.cs
WalletWasabi.Tests/UnitTests/Tor/Control/Utils/AsyncEventParserTests.cs
@@ -16,7 +16,7 @@ namespace WalletWasabi.Tests.UnitTests.Tor.Control.Utils
[Fact]
public async Task ParseUnknownEventAsync()
{
- /// "XXX" is not a valid event name.
+ // "XXX" is not a valid event name.
string data = "650 XXX NOTICE\r\n";
TorControlReply rawReply = await TorControlReplyReaderTest.ParseAsync(data);

MODIFY
WalletWasabi/Tor/Control/TorControlClient.cs
WalletWasabi/Tor/Control/TorControlClient.cs
@@ -2,6 +2,7 @@ using Nito.AsyncEx;
using System;
using System.Collections.Generic;
using System.IO.Pipelines;
+using System.Linq;
using System.Net.Sockets;
using System.Runtime.CompilerServices;
using System.Text;
@@ -9,6 +10,7 @@ using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using WalletWasabi.Logging;
+using WalletWasabi.Tor.Control.Exceptions;
using WalletWasabi.Tor.Control.Messages;
namespace WalletWasabi.Tor.Control
@@ -23,6 +25,9 @@ namespace WalletWasabi.Tor.Control
SingleWriter = true
};
+ /// <remarks>This helps with graceful stopping of the reader loop.</remarks>
+ private volatile bool _readLastSyncReply;
+
public TorControlClient(TcpClient tcpClient) :
this(PipeReader.Create(tcpClient.GetStream()), PipeWriter.Create(tcpClient.GetStream()))
{
@@ -67,6 +72,13 @@ namespace WalletWasabi.Tor.Control
/// <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>Key represents an event name and value represents a subscription counter.</summary>
+ private SortedDictionary<string, int> SubscribedEvents { get; } = new();
+
+ /// <summary>Lock to guard all access to <see cref="SubscribedEvents"/>.</summary>
+ /// <remarks><see cref="MessageLock"/> must be locked first if it is needed too.</remarks>
+ private object SubscriptionEventsLock { get; } = new();
+
/// <summary>Number of subscribers that currently listen to Tor's async events using <see cref="ReadEventsAsync"/>.</summary>
/// <remarks>Mainly for tests.</remarks>
internal int SubscriberCount
@@ -116,8 +128,22 @@ namespace WalletWasabi.Tor.Control
/// </summary>
/// <remarks>If server is a client (OP), exit immediately. If it's a relay (OR), close listeners and exit after <c>ShutdownWaitLength</c> seconds.</remarks>
/// <seealso href="https://gitweb.torproject.org/torspec.git/tree/control-spec.txt">See 3.7. SIGNAL.</seealso>
- public Task<TorControlReply> SignalShutdownAsync(CancellationToken cancellationToken = default)
- => SendCommandAsync("SIGNAL SHUTDOWN\r\n", cancellationToken);
+ public async Task<TorControlReply> SignalShutdownAsync(CancellationToken cancellationToken = default)
+ {
+ using IDisposable _ = await MessageLock.LockAsync(cancellationToken).ConfigureAwait(false);
+
+ // This assignment must be done after MessageLock is acquired to prevent the race in calling SendCommand method by someone else.
+ _readLastSyncReply = true;
+
+ TorControlReply reply = await SendCommandNoLockAsync("SIGNAL SHUTDOWN\r\n", cancellationToken).ConfigureAwait(false);
+
+ if (reply.Success)
+ {
+ ReaderCts.Cancel();
+ }
+
+ return reply;
+ }
/// <summary>
/// Causes Tor to stop polling for the existence of a process with its owning controller's PID.
@@ -127,20 +153,27 @@ namespace WalletWasabi.Tor.Control
public Task<TorControlReply> ResetOwningControllerProcessConfAsync(CancellationToken cancellationToken = default)
=> SendCommandAsync("RESETCONF __OwningControllerProcess\r\n", cancellationToken);
- /// <summary>
- /// Sends a command to Tor.
- /// </summary>
+ /// <summary>Sends a command to Tor.</summary>
/// <remarks>This is meant as a low-level API method, if needed for some reason.</remarks>
/// <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);
+ using IDisposable _ = await MessageLock.LockAsync(cancellationToken).ConfigureAwait(false);
+ return await SendCommandNoLockAsync(command, cancellationToken).ConfigureAwait(false);
+ }
+
+ /// <remarks>Lock <see cref="MessageLock"/> must be acquired by the caller.</remarks>
+ private async Task<TorControlReply> SendCommandNoLockAsync(string command, CancellationToken cancellationToken = default)
+ {
+ using CancellationTokenSource linkedCts = CancellationTokenSource.CreateLinkedTokenSource(ReaderCts.Token, cancellationToken);
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);
+ _ = await PipeWriter.WriteAsync(new ReadOnlyMemory<byte>(Encoding.ASCII.GetBytes(command)), linkedCts.Token).ConfigureAwait(false);
+ _ = await PipeWriter.FlushAsync(linkedCts.Token).ConfigureAwait(false);
+
+ TorControlReply reply = await SyncChannel.Reader.ReadAsync(linkedCts.Token).ConfigureAwait(false);
+ Logger.LogTrace($"Client: Reply: '{reply}'");
- TorControlReply reply = await SyncChannel.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
return reply;
}
@@ -182,13 +215,144 @@ namespace WalletWasabi.Tor.Control
lock (AsyncChannelsLock)
{
newList = new(AsyncChannels);
- newList.Remove(channel);
+ _ = newList.Remove(channel);
AsyncChannels = newList;
}
}
}
+ /// <returns>List of event names like <c>CIRC</c>, <c>STATUS_CLIENT</c>, etc.</returns>
+ public List<string> GetSubscribedEvents()
+ {
+ lock (SubscriptionEventsLock)
+ {
+ // Return a copy to avoid multi-threading issues.
+ return SubscribedEvents.Keys.ToList();
+ }
+ }
+
+ /// <summary>Subscribes Tor control events by their names.</summary>
+ /// <remarks>If an event stream is already subscribed, no command is sent to Tor control.</remarks>
+ /// <param name="cancellationToken">
+ /// Useful when the whole application stops. Otherwise, the internal state of this object may get corrupted.
+ /// </param>
+ public async Task SubscribeEventsAsync(string[] names, CancellationToken cancellationToken = default)
+ {
+ using IDisposable _ = await MessageLock.LockAsync(cancellationToken).ConfigureAwait(false);
+
+ bool sendCommand = false;
+ string subscribedEventNames;
+
+ lock (SubscriptionEventsLock)
+ {
+ foreach (string eventName in names)
+ {
+ if (SubscribedEvents.TryGetValue(eventName, out int counter))
+ {
+ SubscribedEvents[eventName] = counter + 1;
+ }
+ else
+ {
+ sendCommand = true;
+ SubscribedEvents.Add(eventName, 1);
+ }
+ }
+
+ // Get all event names that must be subscribed.
+ subscribedEventNames = string.Join(',', SubscribedEvents.Keys);
+ }
+
+ if (sendCommand)
+ {
+ TorControlReply reply = await SendCommandNoLockAsync($"SETEVENTS {subscribedEventNames}\r\n", cancellationToken).ConfigureAwait(false);
+
+ if (!reply.Success)
+ {
+ // This should never happen.
+ throw new TorControlException("Failed to subscribe events.");
+ }
+ }
+ }
+
+ /// <summary>Unsubscribes Tor control events by their names.</summary>
+ /// <remarks>If the event listener counter gets to zero, the event stream is actually truly unsubscribed.</remarks>
+ /// <param name="cancellationToken">
+ /// Useful when the whole application stops. Otherwise, the internal state of this object may get corrupted.
+ /// </param>
+ public async Task UnsubscribeEventsAsync(string[] names, CancellationToken cancellationToken = default)
+ {
+ using IDisposable _ = await MessageLock.LockAsync(cancellationToken).ConfigureAwait(false);
+
+ bool sendCommand = false;
+ string subscribedEventNames;
+
+ lock (SubscriptionEventsLock)
+ {
+ foreach (string eventName in names)
+ {
+ if (SubscribedEvents.TryGetValue(eventName, out int counter))
+ {
+ counter--;
+
+ if (counter > 0)
+ {
+ SubscribedEvents[eventName] = counter;
+ }
+ else
+ {
+ SubscribedEvents.Remove(eventName);
+ sendCommand = true;
+ }
+ }
+ }
+
+ // Get all event names that remained.
+ subscribedEventNames = string.Join(',', SubscribedEvents.Keys);
+ }
+
+ if (sendCommand)
+ {
+ string command = subscribedEventNames == "" ? "SETEVENTS\r\n" : $"SETEVENTS {subscribedEventNames}\r\n";
+ TorControlReply reply = await SendCommandNoLockAsync(command, cancellationToken).ConfigureAwait(false);
+
+ if (!reply.Success)
+ {
+ // This should never happen.
+ throw new TorControlException("Failed to unsubscribe events.");
+ }
+ }
+ }
+
+ /// <summary>Unsubscribes all Tor control events.</summary>
+ public async Task<bool> UnsubscribeAllEventsAsync()
+ {
+ // Sanity timeout.
+ using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
+ using IDisposable _ = await MessageLock.LockAsync(cts.Token).ConfigureAwait(false);
+
+ int count;
+
+ lock (SubscriptionEventsLock)
+ {
+ count = SubscribedEvents.Keys.Count;
+ SubscribedEvents.Clear();
+ }
+
+ if (count > 0)
+ {
+ TorControlReply reply = await SendCommandNoLockAsync($"SETEVENTS\r\n", cts.Token).ConfigureAwait(false);
+
+ if (!reply.Success)
+ {
+ // This should never happen.
+ return false;
+ }
+ }
+
+ return true;
+ }
+
/// <summary>
/// Loop that continues reading received messages from Tor control TCP connection.
/// </summary>
@@ -219,6 +383,12 @@ namespace WalletWasabi.Tor.Control
{
// Propagate a response back to the requester.
await SyncChannel.Writer.WriteAsync(reply, ReaderCts.Token).ConfigureAwait(false);
+
+ if (_readLastSyncReply)
+ {
+ Logger.LogTrace("Request to read last message was issued. No more message reading.");
+ break;
+ }
}
}
}
@@ -236,6 +406,13 @@ namespace WalletWasabi.Tor.Control
public async ValueTask DisposeAsync()
{
+ bool isOk = await UnsubscribeAllEventsAsync().ConfigureAwait(false);
+
+ if (!isOk)
+ {
+ Logger.LogWarning("Failed to unsubscribe all Tor control events.");
+ }
+
// Stop reader loop.
ReaderCts.Cancel();

TorControlClient: Add subscription API.