Skip to content

Commit dcb34de

Browse files
authored
[wasm-mt] A version of LowLevelLifoSemaphore that uses callbacks on the browser (#84491)
This is part of #84489 - landing support for async JS interop on threadpool threads in multi-threaded WebAssembly. This PR adds two things: 1. A new unmanaged `LifoSemaphoreAsyncWait` semaphore that can use Emscripten's ability to push C calls from one thread to another in order to implement a callback-based semaphore - when a thread wants to wait, it sets up a success callback and a timeout callback, and then can return to the JS event loop. When the semaphore is released, Emscripten will trigger the callback to run on the waiting thread. If the wait times out, the timeout callback will run. 2. A new managed `LowLevelLifoAsyncWaitSemaphore` that doesn't have the normal `Wait()` function, and instead needs to use the callback-based `PrepareAsyncWait()` function. Also refactored `LowLevelLifoSemaphore` to pull out a common `LowLevelLifoSemaphoreBase` class to share with the async wait version. * [wasm-mt][mono] Add new LifoSemaphoreAsyncWait C primitive Add a new kind of LifoSemaphore that has a callback-based wait function, instead of a blocking wait using Emscripten's ability to send work from one webworker to another in C. This will allow us to wait for a semaphore from the JS event loop in a web worker. * [wasm-mt][mono] split LowLevelLifoSemaphore into two kinds A normal LowLevelLifoSemaphore that can do a synchronous wait and another that can do a callback-based wait from the JS event loop * Add LowLevelLifoSemaphoreBase Move the counts to the base class Move Release to the base class, make ReleaseCore abstract * make a new LowLevelLifoAsyncWaitSemaphore for wasm-mt * Revert unintentional package-lock.json changes * fix possible null dereference * use a separate icall for async wait InitInternal instead of magic constants that are otherwise not needed in managed * remove dead code; fixup comments * LowLevelLifoSemaphore: decrement timeoutMs if we lost InterlockedCompareExchange When a thread wakes after waiting for a semaphore to be released, if it raced with another thread that is also trying to update the semaphore counts and loses, it has to go back to waiting again. In that case, decrement the remaining timeout by the elapsed wait time so that the next wait is shorter. * better timeout decrement code * move timeoutMs == 0 check to PrepareAsyncWaitCore make PrepareAsyncWaitCore static and remove a redundant argument
1 parent f107b4b commit dcb34de

File tree

14 files changed

+878
-219
lines changed

14 files changed

+878
-219
lines changed

src/coreclr/System.Private.CoreLib/src/System/Threading/LowLevelLifoSemaphore.Unix.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ namespace System.Threading
1111
/// <summary>
1212
/// A LIFO semaphore implemented using the PAL's semaphore with uninterruptible waits.
1313
/// </summary>
14-
internal sealed partial class LowLevelLifoSemaphore : IDisposable
14+
internal sealed partial class LowLevelLifoSemaphore : LowLevelLifoSemaphoreBase, IDisposable
1515
{
1616
private Semaphore? _semaphore;
1717

@@ -34,7 +34,7 @@ public bool WaitCore(int timeoutMs)
3434
[LibraryImport(RuntimeHelpers.QCall, EntryPoint = "WaitHandle_CorWaitOnePrioritizedNative")]
3535
private static partial int WaitNative(SafeWaitHandle handle, int timeoutMs);
3636

37-
public void ReleaseCore(int count)
37+
protected override void ReleaseCore(int count)
3838
{
3939
Debug.Assert(_semaphore != null);
4040
Debug.Assert(count > 0);

src/coreclr/nativeaot/System.Private.CoreLib/src/System/Threading/LowLevelLifoSemaphore.Unix.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ namespace System.Threading
99
/// A LIFO semaphore.
1010
/// Waits on this semaphore are uninterruptible.
1111
/// </summary>
12-
internal sealed partial class LowLevelLifoSemaphore : IDisposable
12+
internal sealed partial class LowLevelLifoSemaphore : LowLevelLifoSemaphoreBase, IDisposable
1313
{
1414
private WaitSubsystem.WaitableObject _semaphore;
1515

@@ -27,7 +27,7 @@ private bool WaitCore(int timeoutMs)
2727
return WaitSubsystem.Wait(_semaphore, timeoutMs, false, true) == WaitHandle.WaitSuccess;
2828
}
2929

30-
private void ReleaseCore(int count)
30+
protected override void ReleaseCore(int count)
3131
{
3232
WaitSubsystem.ReleaseSemaphore(_semaphore, count);
3333
}

src/libraries/System.Private.CoreLib/src/System.Private.CoreLib.Shared.projitems

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2531,6 +2531,7 @@
25312531
<Compile Include="$(MSBuildThisFileDirectory)System\Threading\PortableThreadPool.Windows.cs" Condition="'$(TargetsWindows)' == 'true'" />
25322532
<Compile Include="$(MSBuildThisFileDirectory)System\Threading\LowLevelLifoSemaphore.cs" />
25332533
<Compile Include="$(MSBuildThisFileDirectory)System\Threading\LowLevelLifoSemaphore.Windows.cs" Condition="'$(TargetsWindows)' == 'true'" />
2534+
<Compile Include="$(MSBuildThisFileDirectory)System\Threading\LowLevelLifoSemaphoreBase.cs" />
25342535
<Compile Include="$(MSBuildThisFileDirectory)System\Threading\PreAllocatedOverlapped.cs" Condition="('$(TargetsBrowser)' != 'true' and '$(TargetsWasi)' != 'true') or '$(FeatureWasmThreads)' == 'true'" />
25352536
<Compile Include="$(MSBuildThisFileDirectory)System\Threading\RegisteredWaitHandle.Portable.cs" />
25362537
<Compile Include="$(MSBuildThisFileDirectory)System\Threading\ThreadPoolBoundHandle.cs" />

src/libraries/System.Private.CoreLib/src/System/Threading/LowLevelLifoSemaphore.Windows.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ public bool WaitCore(int timeoutMs)
5050
return success;
5151
}
5252

53-
public void ReleaseCore(int count)
53+
protected override void ReleaseCore(int count)
5454
{
5555
Debug.Assert(count > 0);
5656

src/libraries/System.Private.CoreLib/src/System/Threading/LowLevelLifoSemaphore.cs

Lines changed: 12 additions & 192 deletions
Original file line numberDiff line numberDiff line change
@@ -11,29 +11,13 @@ namespace System.Threading
1111
/// A LIFO semaphore.
1212
/// Waits on this semaphore are uninterruptible.
1313
/// </summary>
14-
internal sealed partial class LowLevelLifoSemaphore : IDisposable
14+
internal sealed partial class LowLevelLifoSemaphore : LowLevelLifoSemaphoreBase, IDisposable
1515
{
16-
private CacheLineSeparatedCounts _separated;
17-
18-
private readonly int _maximumSignalCount;
19-
private readonly int _spinCount;
20-
private readonly Action _onWait;
21-
2216
private const int SpinSleep0Threshold = 10;
2317

2418
public LowLevelLifoSemaphore(int initialSignalCount, int maximumSignalCount, int spinCount, Action onWait)
19+
: base (initialSignalCount, maximumSignalCount, spinCount, onWait)
2520
{
26-
Debug.Assert(initialSignalCount >= 0);
27-
Debug.Assert(initialSignalCount <= maximumSignalCount);
28-
Debug.Assert(maximumSignalCount > 0);
29-
Debug.Assert(spinCount >= 0);
30-
31-
_separated = default;
32-
_separated._counts.SignalCount = (uint)initialSignalCount;
33-
_maximumSignalCount = maximumSignalCount;
34-
_spinCount = spinCount;
35-
_onWait = onWait;
36-
3721
Create(maximumSignalCount);
3822
}
3923

@@ -144,55 +128,6 @@ public bool Wait(int timeoutMs, bool spinWait)
144128
}
145129
}
146130

147-
public void Release(int releaseCount)
148-
{
149-
Debug.Assert(releaseCount > 0);
150-
Debug.Assert(releaseCount <= _maximumSignalCount);
151-
152-
int countOfWaitersToWake;
153-
Counts counts = _separated._counts;
154-
while (true)
155-
{
156-
Counts newCounts = counts;
157-
158-
// Increase the signal count. The addition doesn't overflow because of the limit on the max signal count in constructor.
159-
newCounts.AddSignalCount((uint)releaseCount);
160-
161-
// Determine how many waiters to wake, taking into account how many spinners and waiters there are and how many waiters
162-
// have previously been signaled to wake but have not yet woken
163-
countOfWaitersToWake =
164-
(int)Math.Min(newCounts.SignalCount, (uint)counts.WaiterCount + counts.SpinnerCount) -
165-
counts.SpinnerCount -
166-
counts.CountOfWaitersSignaledToWake;
167-
if (countOfWaitersToWake > 0)
168-
{
169-
// Ideally, limiting to a maximum of releaseCount would not be necessary and could be an assert instead, but since
170-
// WaitForSignal() does not have enough information to tell whether a woken thread was signaled, and due to the cap
171-
// below, it's possible for countOfWaitersSignaledToWake to be less than the number of threads that have actually
172-
// been signaled to wake.
173-
if (countOfWaitersToWake > releaseCount)
174-
{
175-
countOfWaitersToWake = releaseCount;
176-
}
177-
178-
// Cap countOfWaitersSignaledToWake to its max value. It's ok to ignore some woken threads in this count, it just
179-
// means some more threads will be woken next time. Typically, it won't reach the max anyway.
180-
newCounts.AddUpToMaxCountOfWaitersSignaledToWake((uint)countOfWaitersToWake);
181-
}
182-
183-
Counts countsBeforeUpdate = _separated._counts.InterlockedCompareExchange(newCounts, counts);
184-
if (countsBeforeUpdate == counts)
185-
{
186-
Debug.Assert(releaseCount <= _maximumSignalCount - counts.SignalCount);
187-
if (countOfWaitersToWake > 0)
188-
ReleaseCore(countOfWaitersToWake);
189-
return;
190-
}
191-
192-
counts = countsBeforeUpdate;
193-
}
194-
}
195-
196131
private bool WaitForSignal(int timeoutMs)
197132
{
198133
Debug.Assert(timeoutMs > 0 || timeoutMs == -1);
@@ -201,13 +136,15 @@ private bool WaitForSignal(int timeoutMs)
201136

202137
while (true)
203138
{
204-
if (!WaitCore(timeoutMs))
139+
int startWaitTicks = timeoutMs != -1 ? Environment.TickCount : 0;
140+
if (timeoutMs == 0 || !WaitCore(timeoutMs))
205141
{
206142
// Unregister the waiter. The wait subsystem used above guarantees that a thread that wakes due to a timeout does
207143
// not observe a signal to the object being waited upon.
208144
_separated._counts.InterlockedDecrementWaiterCount();
209145
return false;
210146
}
147+
int endWaitTicks = timeoutMs != -1 ? Environment.TickCount : 0;
211148

212149
// Unregister the waiter if this thread will not be waiting anymore, and try to acquire the semaphore
213150
Counts counts = _separated._counts;
@@ -238,132 +175,15 @@ private bool WaitForSignal(int timeoutMs)
238175
}
239176

240177
counts = countsBeforeUpdate;
178+
if (timeoutMs != -1) {
179+
int waitMs = endWaitTicks - startWaitTicks;
180+
if (waitMs >= 0 && waitMs < timeoutMs)
181+
timeoutMs -= waitMs;
182+
else
183+
timeoutMs = 0;
184+
}
241185
}
242186
}
243187
}
244-
245-
private struct Counts : IEquatable<Counts>
246-
{
247-
private const byte SignalCountShift = 0;
248-
private const byte WaiterCountShift = 32;
249-
private const byte SpinnerCountShift = 48;
250-
private const byte CountOfWaitersSignaledToWakeShift = 56;
251-
252-
private ulong _data;
253-
254-
private Counts(ulong data) => _data = data;
255-
256-
private uint GetUInt32Value(byte shift) => (uint)(_data >> shift);
257-
private void SetUInt32Value(uint value, byte shift) =>
258-
_data = (_data & ~((ulong)uint.MaxValue << shift)) | ((ulong)value << shift);
259-
private ushort GetUInt16Value(byte shift) => (ushort)(_data >> shift);
260-
private void SetUInt16Value(ushort value, byte shift) =>
261-
_data = (_data & ~((ulong)ushort.MaxValue << shift)) | ((ulong)value << shift);
262-
private byte GetByteValue(byte shift) => (byte)(_data >> shift);
263-
private void SetByteValue(byte value, byte shift) =>
264-
_data = (_data & ~((ulong)byte.MaxValue << shift)) | ((ulong)value << shift);
265-
266-
public uint SignalCount
267-
{
268-
get => GetUInt32Value(SignalCountShift);
269-
set => SetUInt32Value(value, SignalCountShift);
270-
}
271-
272-
public void AddSignalCount(uint value)
273-
{
274-
Debug.Assert(value <= uint.MaxValue - SignalCount);
275-
_data += (ulong)value << SignalCountShift;
276-
}
277-
278-
public void IncrementSignalCount() => AddSignalCount(1);
279-
280-
public void DecrementSignalCount()
281-
{
282-
Debug.Assert(SignalCount != 0);
283-
_data -= (ulong)1 << SignalCountShift;
284-
}
285-
286-
public ushort WaiterCount
287-
{
288-
get => GetUInt16Value(WaiterCountShift);
289-
set => SetUInt16Value(value, WaiterCountShift);
290-
}
291-
292-
public void IncrementWaiterCount()
293-
{
294-
Debug.Assert(WaiterCount < ushort.MaxValue);
295-
_data += (ulong)1 << WaiterCountShift;
296-
}
297-
298-
public void DecrementWaiterCount()
299-
{
300-
Debug.Assert(WaiterCount != 0);
301-
_data -= (ulong)1 << WaiterCountShift;
302-
}
303-
304-
public void InterlockedDecrementWaiterCount()
305-
{
306-
var countsAfterUpdate = new Counts(Interlocked.Add(ref _data, unchecked((ulong)-1) << WaiterCountShift));
307-
Debug.Assert(countsAfterUpdate.WaiterCount != ushort.MaxValue); // underflow check
308-
}
309-
310-
public byte SpinnerCount
311-
{
312-
get => GetByteValue(SpinnerCountShift);
313-
set => SetByteValue(value, SpinnerCountShift);
314-
}
315-
316-
public void IncrementSpinnerCount()
317-
{
318-
Debug.Assert(SpinnerCount < byte.MaxValue);
319-
_data += (ulong)1 << SpinnerCountShift;
320-
}
321-
322-
public void DecrementSpinnerCount()
323-
{
324-
Debug.Assert(SpinnerCount != 0);
325-
_data -= (ulong)1 << SpinnerCountShift;
326-
}
327-
328-
public byte CountOfWaitersSignaledToWake
329-
{
330-
get => GetByteValue(CountOfWaitersSignaledToWakeShift);
331-
set => SetByteValue(value, CountOfWaitersSignaledToWakeShift);
332-
}
333-
334-
public void AddUpToMaxCountOfWaitersSignaledToWake(uint value)
335-
{
336-
uint availableCount = (uint)(byte.MaxValue - CountOfWaitersSignaledToWake);
337-
if (value > availableCount)
338-
{
339-
value = availableCount;
340-
}
341-
_data += (ulong)value << CountOfWaitersSignaledToWakeShift;
342-
}
343-
344-
public void DecrementCountOfWaitersSignaledToWake()
345-
{
346-
Debug.Assert(CountOfWaitersSignaledToWake != 0);
347-
_data -= (ulong)1 << CountOfWaitersSignaledToWakeShift;
348-
}
349-
350-
public Counts InterlockedCompareExchange(Counts newCounts, Counts oldCounts) =>
351-
new Counts(Interlocked.CompareExchange(ref _data, newCounts._data, oldCounts._data));
352-
353-
public static bool operator ==(Counts lhs, Counts rhs) => lhs.Equals(rhs);
354-
public static bool operator !=(Counts lhs, Counts rhs) => !lhs.Equals(rhs);
355-
356-
public override bool Equals([NotNullWhen(true)] object? obj) => obj is Counts other && Equals(other);
357-
public bool Equals(Counts other) => _data == other._data;
358-
public override int GetHashCode() => (int)_data + (int)(_data >> 32);
359-
}
360-
361-
[StructLayout(LayoutKind.Sequential)]
362-
private struct CacheLineSeparatedCounts
363-
{
364-
private readonly Internal.PaddingFor32 _pad1;
365-
public Counts _counts;
366-
private readonly Internal.PaddingFor32 _pad2;
367-
}
368188
}
369189
}

0 commit comments

Comments
 (0)