Make broadcasts thread safe
MODIFY
HiddenWallet.Daemon/DaemonHub.cs
HiddenWallet.Daemon/DaemonHub.cs
@@ -8,22 +8,22 @@ namespace HiddenWallet.Daemon
{
public class DaemonHub : Hub
{
- public void GetTorStatus()
+ public async Task GetTorStatusAsync()
{
string status = Tor.State.ToString();
- NotificationBroadcaster.Instance.BroadcastTorState(status);
+ await NotificationBroadcaster.Instance.BroadcastTorStateAsync(status);
}
public async Task GetHeaderHeightAsync()
{
var (Success, Height) = await Global.WalletWrapper.WalletJob.TryGetHeaderHeightAsync();
- NotificationBroadcaster.Instance.BroadcastHeaderHeight(Height.Value.ToString());
+ await NotificationBroadcaster.Instance.BroadcastHeaderHeightAsync(Height.Value.ToString());
}
- public void TumblerStatusBroadcastRequest()
+ public async Task TumblerStatusBroadcastRequestAsync()
{
TumblerStatusResponse status = Global.WalletWrapper.GetTumblerStatusResponse();
- NotificationBroadcaster.Instance.BroadcastTumblerStatus(status);
+ await NotificationBroadcaster.Instance.BroadcastTumblerStatusAsync(status);
}
}
}
\ No newline at end of file

MODIFY
HiddenWallet.Daemon/NotificationBroadcaster.cs
HiddenWallet.Daemon/NotificationBroadcaster.cs
@@ -3,6 +3,7 @@ using HiddenWallet.Daemon.Models;
using HiddenWallet.Models;
using Microsoft.AspNetCore.SignalR;
using Newtonsoft.Json;
+using Nito.AsyncEx;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
@@ -17,6 +18,8 @@ namespace HiddenWallet.Daemon
private IHubContext<DaemonHub> _context;
+ private static AsyncLock AsyncLock = new AsyncLock();
+
private NotificationBroadcaster() { }
public static NotificationBroadcaster Instance => _instance.Value;
@@ -29,60 +32,93 @@ namespace HiddenWallet.Daemon
}
}
- public void BroadcastMempool(string mempoolCount)
+ public async Task BroadcastMempoolAsync(string mempoolCount)
+ {
+ using (await AsyncLock.LockAsync())
{
+ await DelayUntilContextNullAsync();
IClientProxy proxy = _context.Clients.All;
- proxy.InvokeAsync("mempoolChanged", mempoolCount);
+ await proxy.InvokeAsync("mempoolChanged", mempoolCount);
+ }
}
- public void BroadcastTrackerHeight(string trackerHeight)
+ public async Task BroadcastTrackerHeightAsync(string trackerHeight)
{
+ using (await AsyncLock.LockAsync())
+ {
+ await DelayUntilContextNullAsync();
IClientProxy proxy = _context.Clients.All;
- proxy.InvokeAsync("trackerHeightChanged", trackerHeight);
+ await proxy.InvokeAsync("trackerHeightChanged", trackerHeight);
+ }
}
- public void BroadcastHeaderHeight(string headerHeight)
+ public async Task BroadcastHeaderHeightAsync(string headerHeight)
+ {
+ using (await AsyncLock.LockAsync())
{
+ await DelayUntilContextNullAsync();
IClientProxy proxy = _context.Clients.All;
- proxy.InvokeAsync("headerHeightChanged", headerHeight);
+ await proxy.InvokeAsync("headerHeightChanged", headerHeight);
+ }
}
- public void BroadcastNodeCount(string nodeCount)
+ public async Task BroadcastNodeCountAsync(string nodeCount)
+ {
+ using (await AsyncLock.LockAsync())
{
+ await DelayUntilContextNullAsync();
IClientProxy proxy = _context.Clients.All;
- proxy.InvokeAsync("nodeCountChanged", nodeCount);
+ await proxy.InvokeAsync("nodeCountChanged", nodeCount);
+ }
}
- public void BroadcastWalletState(string state)
+ public async Task BroadcastWalletStateAsync(string state)
{
+ using (await AsyncLock.LockAsync())
+ {
+ await DelayUntilContextNullAsync();
IClientProxy proxy = _context.Clients.All;
- proxy.InvokeAsync("walletStateChanged", state);
+ await proxy.InvokeAsync("walletStateChanged", state);
+ }
}
- public void BroadcastChangeBump()
+ public async Task BroadcastChangeBumpAsync()
{
+ using (await AsyncLock.LockAsync())
+ {
+ await DelayUntilContextNullAsync();
IClientProxy proxy = _context.Clients.All;
- proxy.InvokeAsync("changeBump");
+ await proxy.InvokeAsync("changeBump");
+ }
}
- public void BroadcastTorState(string state)
+ public async Task BroadcastTorStateAsync(string state)
{
- if (_context != null)
+ using (await AsyncLock.LockAsync())
{
+ await DelayUntilContextNullAsync();
IClientProxy proxy = _context.Clients.All;
- proxy.InvokeAsync("torStateChanged", state);
- }
- else
- {
- Console.WriteLine("IHubContext not set yet. Tor Status Broadcast failed.");
+ await proxy.InvokeAsync("torStateChanged", state);
}
}
- public void BroadcastTumblerStatus(TumblerStatusResponse status)
+ public async Task BroadcastTumblerStatusAsync(TumblerStatusResponse status)
+ {
+ using (await AsyncLock.LockAsync())
{
+ await DelayUntilContextNullAsync();
IClientProxy proxy = _context.Clients.All;
string json = JsonConvert.SerializeObject(status);
- proxy.InvokeAsync("mixerStatusChanged", json);
+ await proxy.InvokeAsync("mixerStatusChanged", json);
+ }
+ }
+
+ private async Task DelayUntilContextNullAsync()
+ {
+ while (_context == null)
+ {
+ await Task.Delay(100);
+ }
}
}
}
\ No newline at end of file

MODIFY
HiddenWallet.Daemon/Tor.cs
HiddenWallet.Daemon/Tor.cs
@@ -32,7 +32,9 @@ namespace HiddenWallet.Daemon
_state = value;
try
{
- NotificationBroadcaster.Instance.BroadcastTorState(State.ToString());
+#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+ NotificationBroadcaster.Instance.BroadcastTorStateAsync(State.ToString());
+#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
}
catch (Exception ex)
{

MODIFY
HiddenWallet.Daemon/Wrappers/WalletWrapper.cs
HiddenWallet.Daemon/Wrappers/WalletWrapper.cs
@@ -90,12 +90,12 @@ namespace HiddenWallet.Daemon.Wrappers
WalletJob = new WalletJob();
await WalletJob.InitializeAsync(Tor.SocksPortHandler, Tor.ControlPortClient, safe, trackDefaultSafe: false, accountsToTrack: new SafeAccount[] { AliceAccount, BobAccount });
- WalletJob.StateChanged += WalletJob_StateChanged;
- WalletJob.MemPoolJob.NewTransaction += MemPoolJob_NewTransaction;
- WalletJob.HeaderHeightChanged += WalletJob_HeaderHeightChanged;
- (await WalletJob.GetTrackerAsync()).BestHeightChanged += WalletWrapper_BestHeightChanged;
- WalletJob.ConnectedNodeCountChanged += WalletJob_ConnectedNodeCountChanged;
- WalletJob.CoinJoinService.PeerCountChanged += CoinJoinService_PeerCountChanged;
+ WalletJob.StateChanged += WalletJob_StateChangedAsync;
+ WalletJob.MemPoolJob.NewTransaction += MemPoolJob_NewTransactionAsync;
+ WalletJob.HeaderHeightChanged += WalletJob_HeaderHeightChangedAsync;
+ (await WalletJob.GetTrackerAsync()).BestHeightChanged += WalletWrapper_BestHeightChangedAsync;
+ WalletJob.ConnectedNodeCountChanged += WalletJob_ConnectedNodeCountChangedAsync;
+ WalletJob.CoinJoinService.PeerCountChanged += CoinJoinService_PeerCountChangedAsync;
(await WalletJob.GetTrackerAsync()).TrackedTransactions.CollectionChanged += TrackedTransactions_CollectionChangedAsync;
@@ -120,36 +120,36 @@ namespace HiddenWallet.Daemon.Wrappers
}
}
- private void WalletWrapper_BestHeightChanged(object sender, Height height)
+ private async void WalletWrapper_BestHeightChangedAsync(object sender, Height height)
{
- NotificationBroadcaster.Instance.BroadcastTrackerHeight(height.Value.ToString());
+ await NotificationBroadcaster.Instance.BroadcastTrackerHeightAsync(height.Value.ToString());
}
- private void WalletJob_StateChanged(object sender, EventArgs e)
+ private async void WalletJob_StateChangedAsync(object sender, EventArgs e)
{
_walletState = WalletJob.State.ToString();
- NotificationBroadcaster.Instance.BroadcastWalletState(_walletState);
+ await NotificationBroadcaster.Instance.BroadcastWalletStateAsync(_walletState);
}
- private void WalletJob_ConnectedNodeCountChanged(object sender, int count)
+ private async void WalletJob_ConnectedNodeCountChangedAsync(object sender, int count)
{
- NotificationBroadcaster.Instance.BroadcastNodeCount(count.ToString());
+ await NotificationBroadcaster.Instance.BroadcastNodeCountAsync(count.ToString());
}
- private void WalletJob_HeaderHeightChanged(object sender, Height headerHeight)
+ private async void WalletJob_HeaderHeightChangedAsync(object sender, Height headerHeight)
{
- NotificationBroadcaster.Instance.BroadcastHeaderHeight(headerHeight.Value.ToString());
+ await NotificationBroadcaster.Instance.BroadcastHeaderHeightAsync(headerHeight.Value.ToString());
}
- private void MemPoolJob_NewTransaction(object sender, NewTransactionEventArgs e)
+ private async void MemPoolJob_NewTransactionAsync(object sender, NewTransactionEventArgs e)
{
- NotificationBroadcaster.Instance.BroadcastMempool(e.MempoolTxCount.ToString());
+ await NotificationBroadcaster.Instance.BroadcastMempoolAsync(e.MempoolTxCount.ToString());
}
- private void CoinJoinService_PeerCountChanged(object sender, int peerCount)
+ private async void CoinJoinService_PeerCountChangedAsync(object sender, int peerCount)
{
TumblerStatusResponse status = GetTumblerStatusResponse();
- NotificationBroadcaster.Instance.BroadcastTumblerStatus(status);
+ await NotificationBroadcaster.Instance.BroadcastTumblerStatusAsync(status);
}
public async Task RecoverAsync(string password, string mnemonic, string creationTime)
@@ -178,7 +178,7 @@ namespace HiddenWallet.Daemon.Wrappers
{
_changeBump++;
}
- NotificationBroadcaster.Instance.BroadcastChangeBump();
+ await NotificationBroadcaster.Instance.BroadcastChangeBumpAsync();
}
private async Task UpdateHistoryRelatedMembersAsync()
@@ -265,12 +265,12 @@ namespace HiddenWallet.Daemon.Wrappers
Console.WriteLine("Gracefully shutting down...");
if (WalletJob != null)
{
- WalletJob.StateChanged -= WalletJob_StateChanged;
- WalletJob.MemPoolJob.NewTransaction -= MemPoolJob_NewTransaction;
- WalletJob.HeaderHeightChanged -= WalletJob_HeaderHeightChanged;
- (await WalletJob.GetTrackerAsync()).BestHeightChanged -= WalletWrapper_BestHeightChanged;
- WalletJob.ConnectedNodeCountChanged -= WalletJob_ConnectedNodeCountChanged;
- WalletJob.CoinJoinService.PeerCountChanged -= CoinJoinService_PeerCountChanged;
+ WalletJob.StateChanged -= WalletJob_StateChangedAsync;
+ WalletJob.MemPoolJob.NewTransaction -= MemPoolJob_NewTransactionAsync;
+ WalletJob.HeaderHeightChanged -= WalletJob_HeaderHeightChangedAsync;
+ (await WalletJob.GetTrackerAsync()).BestHeightChanged -= WalletWrapper_BestHeightChangedAsync;
+ WalletJob.ConnectedNodeCountChanged -= WalletJob_ConnectedNodeCountChangedAsync;
+ WalletJob.CoinJoinService.PeerCountChanged -= CoinJoinService_PeerCountChangedAsync;
(await WalletJob.GetTrackerAsync()).TrackedTransactions.CollectionChanged -= TrackedTransactions_CollectionChangedAsync;
}

MODIFY
HiddenWallet.Gui/app/js/status-client.js
HiddenWallet.Gui/app/js/status-client.js
@@ -74,10 +74,10 @@ function signalrStatusUpdate() {
connection.start().then(function () {
console.log("Connected to SignalR on Daemon");
- connection.invoke('GetTorStatus');
+ connection.invoke('GetTorStatusAsync');
}).then(function () {
console.log("Tumbler Status Request");
- connection.invoke('TumblerStatusBroadcastRequest'); //Request that the status is broadcast
+ connection.invoke('TumblerStatusBroadcastRequestAsync'); //Request that the status is broadcast
}).catch(error => {
console.error(error.message);
});
@@ -195,7 +195,7 @@ function getHeaderHeight() {
}
function tumblerStatusBroadcastRequest() {
- connection.invoke('TumblerStatusBroadcastRequest');
+ connection.invoke('TumblerStatusBroadcastRequestAsync');
}
function updateMixerTab() {

MODIFY
HiddenWallet.Tests/FullSpvWalletTests.cs
HiddenWallet.Tests/FullSpvWalletTests.cs
@@ -98,16 +98,16 @@ namespace HiddenWallet.Tests
_syncedOnce = walletJob.State == WalletState.Synced;
}
- private void WalletJob_ConnectedNodeCountChanged(object sender, EventArgs e)
+ private void WalletJob_ConnectedNodeCountChanged(object sender, int nodeCount)
{
var walletJob = sender as WalletJob;
- if (walletJob.MaxConnectedNodeCount == walletJob.ConnectedNodeCount)
+ if (walletJob.MaxConnectedNodeCount == nodeCount)
{
_fullyConnected = true;
Debug.WriteLine(
$"{nameof(walletJob.MaxConnectedNodeCount)} reached: {walletJob.MaxConnectedNodeCount}");
}
- else Debug.WriteLine($"{nameof(walletJob.ConnectedNodeCount)}: {walletJob.ConnectedNodeCount}");
+ else Debug.WriteLine($"{nameof(nodeCount)}: {nodeCount}");
}
// test with a long time used testnet wallet, with exotic, including tumblebit transactions

MODIFY
HiddenWallet.Tests/SendTests.cs
HiddenWallet.Tests/SendTests.cs
@@ -400,15 +400,15 @@ namespace HiddenWallet.Tests
Debug.WriteLine($"{nameof(walletJob.State)}: {walletJob.State}");
}
- private void WalletJob_ConnectedNodeCountChanged(object sender, EventArgs e)
+ private void WalletJob_ConnectedNodeCountChanged(object sender, int nodeCount)
{
var walletJob = sender as WalletJob;
- if (walletJob.MaxConnectedNodeCount == walletJob.ConnectedNodeCount)
+ if (walletJob.MaxConnectedNodeCount == nodeCount)
{
Debug.WriteLine(
$"{nameof(WalletJob.MaxConnectedNodeCount)} reached: {walletJob.MaxConnectedNodeCount}");
}
- else Debug.WriteLine($"{nameof(WalletJob.ConnectedNodeCount)}: {walletJob.ConnectedNodeCount}");
+ else Debug.WriteLine($"{nameof(nodeCount)}: {nodeCount}");
}
bool _txProbArrived;

Make broadcasts thread safe