Add comments to AsyncMutex.
MODIFY
WalletWasabi/Nito/AsyncEx/AsyncMutex.cs
WalletWasabi/Nito/AsyncEx/AsyncMutex.cs
@@ -10,15 +10,33 @@ namespace Nito.AsyncEx
{
public class AsyncMutex
{
+ /// <summary>
+ /// Short name of the mutex. This string added to the end of the mutex name.
+ /// </summary>
private string ShortName { get; set; }
+
+ /// <summary>
+ /// The full name of the named mutex. Global\ included in the string.
+ /// </summary>
private string FullName { get; set; }
+ /// <summary>
+ /// Mutex for interprocess synchronization.
+ /// </summary>
private Mutex Mutex { get; set; }
+
+ /// <summary>
+ /// AsyncLock for local thread synchronization.
+ /// </summary>
private AsyncLock AsyncLock { get; set; }
+
+ /// <summary>
+ /// Separate thread for the mutex where it is created and released.
+ /// </summary>
private Thread MutexThread { get; set; }
/// <summary>
- /// string: mutex name
+ /// Static storage for local mutexes. It can be used to get an already existing AsyncLock by name of the mutex.
/// </summary>
private static Dictionary<string, AsyncMutex> AsyncMutexes { get; } = new Dictionary<string, AsyncMutex>();
@@ -45,10 +63,12 @@ namespace Nito.AsyncEx
{
if (AsyncMutexes.TryGetValue(FullName, out AsyncMutex asyncMutex))
{
+ // Use the already existing lock.
AsyncLock = asyncMutex.AsyncLock;
}
else
{
+ // Create a new lock.
AsyncLock = new AsyncLock();
AsyncMutexes.Add(FullName, this);
@@ -63,6 +83,16 @@ namespace Nito.AsyncEx
private Exception LatestHoldLockException { get; set; }
private object LatestHoldLockExceptionLock { get; } = new object();
+ /// <summary>
+ /// The thread of the Mutex. This whole complex logic is needed because we have serious problems
+ /// with releasing the mutex from a separate thread. Mutex.Dispose() silently fails when you try to
+ /// release the mutex so you have to use Mutex.ReleaseMutex(). It is not possible to release the mutex
+ /// from another thread and this was something hard to achieve in an async/await environment with
+ /// multi-platform compatibility. For example Semaphore is not supporting linux.
+ /// So to ensure that the mutex is created and released on the same thread we have created
+ /// a separate thread and control it from outside.
+ /// </summary>
+ /// <param name="cancellationTokenObj"></param>
private void HoldLock(object cancellationTokenObj)
{
CancellationToken ct = cancellationTokenObj is CancellationToken ?
@@ -73,10 +103,14 @@ namespace Nito.AsyncEx
{
try
{
+ // Wait for until we have something to do. The procedure is to set _command variable then
+ // signal with ToDo then wait until the Done is set.
ToDo.WaitOne();
if (Interlocked.CompareExchange(ref _command, 0, 1) == 1)
{
+ // Create the mutex and acquire it. InitiallyOwned means that if the mutex is not
+ // exists then create it and immediately acquire it.
Mutex = new Mutex(initiallyOwned: true, FullName, out bool createdNew);
if (createdNew)
{
@@ -84,11 +118,15 @@ namespace Nito.AsyncEx
}
else
{
+ // The mutex already exists so we will try to acquire it.
DateTime start = DateTime.Now;
bool acquired = false;
+ // Timeout logic.
while (DateTime.Now - start > TimeSpan.FromSeconds(90))
{
+ // Block for n ms and try to acquire the mutex. Blocking is not a problem
+ // we are on our own thread.
acquired = Mutex.WaitOne(1000);
if (acquired)
@@ -107,22 +145,26 @@ namespace Nito.AsyncEx
if (acquired)
{
+ // Go to finally.
continue;
}
+ // Let it go and throw the exception...
}
}
else if (Interlocked.CompareExchange(ref _command, 0, 2) == 2)
{
+ // Command 2 is releasing the mutex.
Mutex?.ReleaseMutex();
Mutex?.Dispose();
Mutex = null;
- return;
+ return; // End of the Thread.
}
throw new NotImplementedException();
}
catch (Exception ex)
{
+ // We had an exception so store it and jump to finally.
lock (LatestHoldLockExceptionLock)
{
LatestHoldLockException = ex;
@@ -131,11 +173,18 @@ namespace Nito.AsyncEx
finally
{
ToDo.Reset();
- Done.Set();
+ Done.Set(); // Indicate that we are ready with the current command.
}
}
}
+ /// <summary>
+ /// Standard procedure to send command to the mutex thread.
+ /// </summary>
+ /// <param name="command"></param>
+ /// <param name="cancellationToken"></param>
+ /// <param name="pollInterval"></param>
+ /// <returns></returns>
private async Task SetCommandAsync(int command, CancellationToken cancellationToken, int pollInterval)
{
if (!(MutexThread?.IsAlive == true))
@@ -143,19 +192,21 @@ namespace Nito.AsyncEx
throw new InvalidOperationException($"Thread should be alive.");
}
- Interlocked.Exchange(ref _command, command);
+ Interlocked.Exchange(ref _command, command); // Set the command.
lock (LatestHoldLockExceptionLock)
{
LatestHoldLockException = null;
}
- Done.Reset();
- ToDo.Set();
+ Done.Reset(); // Reset the Done.
+ ToDo.Set(); // Indicate that there is a new command.
while (!Done.WaitOne(1))
{
+ // Waiting for Done asynchronously.
await Task.Delay(pollInterval, cancellationToken);
}
lock (LatestHoldLockExceptionLock)
{
+ // If we had an exception then throw it.
if (LatestHoldLockException != null)
{
throw LatestHoldLockException;
@@ -169,12 +220,20 @@ namespace Nito.AsyncEx
ToDo.Set();
}
+ /// <summary>
+ /// The Lock mechanism designed for standard using blocks. This lock is thread and interprocess safe.
+ /// You can create and use it from anywhere.
+ /// </summary>
+ /// <param name="cancellationToken"></param>
+ /// <param name="pollInterval">The frequency of polling the termination of the mutex-thread.</param>
+ /// <returns>The IDisposable await-able Task.</returns>
public async Task<IDisposable> LockAsync(CancellationToken cancellationToken = default, int pollInterval = 100)
{
Exception inner = null;
try
{
+ // Local lock for thread safety.
await AsyncLock.LockAsync(cancellationToken);
if (MutexThread?.IsAlive == true)
@@ -187,6 +246,8 @@ namespace Nito.AsyncEx
MutexThread.Name = $"MutexThread";
MutexThread.Start(cancellationToken);
+
+ // Create the mutex and acquire it.
await SetCommandAsync(1, cancellationToken, pollInterval);
return new Key(this);
@@ -209,6 +270,7 @@ namespace Nito.AsyncEx
// Let it go.
}
+ // If something failed then release all.
ReleaseLock();
throw new IOException($"Couldn't acquire system wide mutex on {ShortName}", inner);
@@ -223,11 +285,13 @@ namespace Nito.AsyncEx
throw new InvalidOperationException($"Thread should be alive.");
}
}
-
+ // Send release command to the mutex-thread.
SetCommand(2);
+ // Wait for it.
MutexThread?.Join();
+ // Release the local lock.
AsyncLock?.ReleaseLock();
}

Add comments to AsyncMutex.