youmuliu/stgsharp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HLSF use long as aloloc size

Browse Source
This commit is contained in:
Nitload
2026-02-23 12:32:15 +08:00
parent 4bd88ccbca
commit c155cab510
552 changed files with 154266 additions and 154330 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StgSharp.Common/Threading/AdaptiveBackoffStrategy.cs
+365 -365
View File
@@ -1,366 +1,366 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="AdaptiveBackoffStrategy"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading;
/// <summary>
/// Efficient adaptive backoff strategy that dynamically selects optimal backoff methods based on
/// system load and success rate
/// </summary>
public struct AdaptiveBackoffStrategy
{
private const int EVALUATION_INTERVAL_MS = 2000; // Global re-evaluation interval
private const int EXPLORATION_PROBABILITY = 10; // 10% probability to explore non-optimal strategies
private const int FAST_CONVERGENCE_THRESHOLD = 1000; // Fast convergence threshold
// Configuration constants
private const int INITIAL_SPINWAIT_THRESHOLD = 8; // Cheap spin count for first few SpinWait iterations
private const int MIN_SAMPLES_FOR_EVALUATION = 100; // Minimum sample count before starting evaluation
// Global shared state (all threads share learning results)
private static readonly PerformanceStats[] s_globalStats = new PerformanceStats[3];
private static volatile BackoffMethod s_globalBestMethod;
private static volatile int s_evaluationInProgress;
private static long s_lastGlobalEvaluationTicks;
private BackoffMethod _currentMethod;
private int _consecutiveFailures;
private int _localAttempts;
// Instance state
private SpinWait _spinWait;
private uint _fastRng; // Lightweight pseudo-random number generator
// Static constructor: initialize optimal strategy based on system characteristics
static AdaptiveBackoffStrategy()
{
// Based on your benchmark results:
// - Low core count (<= 8): SpinWait has clear advantage
// - Medium core count (<= 16): Close performance, slightly favor SpinWait
// - High core count (>= 20): Sleep0 has significant advantage under high concurrency
int logicalCores = Environment.ProcessorCount;
if (logicalCores <= 8)
{
s_globalBestMethod = BackoffMethod.SpinWait;
} else if (logicalCores <= 16)
{
s_globalBestMethod = BackoffMethod.SpinWait; // Default conservative choice
} else
{
// High core count systems, based on your 64-thread benchmark results
s_globalBestMethod = BackoffMethod.Sleep0;
}
s_lastGlobalEvaluationTicks = Environment.TickCount64;
// Initialize statistics array
for (int i = 0; i < s_globalStats.Length; i++)
{
s_globalStats[i].Reset();
}
}
/// <summary>
/// Gets the currently used backoff method (for debugging/monitoring)
/// </summary>
public readonly BackoffMethod CurrentMethod => _currentMethod;
/// <summary>
/// Gets the globally considered optimal backoff method
/// </summary>
public static BackoffMethod GlobalBestMethod => s_globalBestMethod;
/// <summary>
/// Creates a new adaptive backoff strategy instance
/// </summary>
public static AdaptiveBackoffStrategy Create()
{
return new AdaptiveBackoffStrategy
{
_spinWait = default,
_currentMethod = s_globalBestMethod,
_consecutiveFailures = 0,
_localAttempts = 0,
_fastRng = (uint)(Environment.TickCount64 ^ Thread.CurrentThread.ManagedThreadId * 0x9E3779B9U)
};
}
/// <summary>
/// Gets global statistics for the specified strategy (for monitoring)
/// </summary>
public static (long attempts, long successes, double successRate) GetGlobalStats(BackoffMethod method)
{
PerformanceStats stats = s_globalStats[(int)method];
return (stats.Attempts, stats.Successes, stats.SuccessRate);
}
/// <summary>
/// Called after CAS failure, performs backoff and may switch strategies
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void OnFailureAndBackoff()
{
_consecutiveFailures++;
// Record failure
s_globalStats[(int)_currentMethod].RecordAttempt(success:false);
// Periodically re-evaluate global optimal strategy
if (ShouldEvaluateGlobalStrategy())
{
EvaluateAndUpdateGlobalStrategy();
}
// Decide whether to switch strategies based on failure count and current strategy
if (_consecutiveFailures >= INITIAL_SPINWAIT_THRESHOLD && ShouldSwitchStrategy())
{
SwitchToOptimalStrategy();
}
// Perform the actual backoff operation
PerformBackoff();
}
/// <summary>
/// Called after CAS success, resets state
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void OnSuccess()
{
// Record success
s_globalStats[(int)_currentMethod].RecordAttempt(success:true);
// Reset state
_spinWait = default;
_consecutiveFailures = 0;
// Fast convergence: tend to keep current strategy when continuously successful
if (_localAttempts < FAST_CONVERGENCE_THRESHOLD)
{
_localAttempts++;
}
}
/// <summary>
/// Evaluates and updates the global optimal strategy
/// </summary>
private static void EvaluateAndUpdateGlobalStrategy()
{
// Use CAS to ensure only one thread performs evaluation
if (Interlocked.CompareExchange(ref s_evaluationInProgress, 1, 0) != 0)
{
return;
}
try
{
long now = Environment.TickCount64;
// Double-check to avoid duplicate evaluation
if (now - Volatile.Read(ref s_lastGlobalEvaluationTicks) <= EVALUATION_INTERVAL_MS)
{
return;
}
BackoffMethod bestMethod = s_globalBestMethod;
double bestSuccessRate = -1.0;
// Find the strategy with the highest success rate
for (int i = 0; i < s_globalStats.Length; i++)
{
PerformanceStats stats = s_globalStats[i];
if (stats.Attempts >= MIN_SAMPLES_FOR_EVALUATION)
{
double successRate = stats.SuccessRate;
if (successRate > bestSuccessRate)
{
bestSuccessRate = successRate;
bestMethod = (BackoffMethod)i;
}
}
}
// Update global optimal strategy
s_globalBestMethod = bestMethod;
Volatile.Write(ref s_lastGlobalEvaluationTicks, now);
// Reset statistics (maintain exploratory behavior)
for (int i = 0; i < s_globalStats.Length; i++)
{
s_globalStats[i].Reset();
}
}
finally
{
Volatile.Write(ref s_evaluationInProgress, 0);
}
}
/// <summary>
/// Fast pseudo-random number generator (LCG)
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private uint NextRandom()
{
_fastRng = _fastRng * 1103515245U + 12345U;
return _fastRng;
}
/// <summary>
/// Performs the actual backoff operation
/// </summary>
[MethodImpl(MethodImplOptions.NoInlining)] // Avoid inlining to reduce hot path code size
private void PerformBackoff()
{
switch (_currentMethod)
{
case BackoffMethod.SpinWait:
if (_consecutiveFailures <= INITIAL_SPINWAIT_THRESHOLD)
{
_spinWait.SpinOnce(sleep1Threshold:-1);
} else if (_spinWait.NextSpinWillYield)
{
int yields = (int)(NextRandom() & 7) + 1; // 1-8 times
for (int i = 0; i < yields; i++) {
_ = Thread.Yield();
}
} else
{
_spinWait.SpinOnce(sleep1Threshold:-1);
}
break;
case BackoffMethod.Yield:
_ = Thread.Yield();
break;
case BackoffMethod.Sleep0:
Thread.Sleep(0);
break;
default:
break;
}
}
/// <summary>
/// Checks whether global strategy evaluation should be performed
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static bool ShouldEvaluateGlobalStrategy()
{
long now = Environment.TickCount64;
return now - Volatile.Read(ref s_lastGlobalEvaluationTicks) > EVALUATION_INTERVAL_MS;
}
/// <summary>
/// Determines whether strategy should be switched
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private bool ShouldSwitchStrategy()
{
// Fast convergence phase: reduce strategy switching
if (_localAttempts < FAST_CONVERGENCE_THRESHOLD && _consecutiveFailures < 32)
{
return false;
}
// Small probability exploration: occasionally try other strategies even when current is optimal
return NextRandom() % 100 < EXPLORATION_PROBABILITY || _currentMethod != s_globalBestMethod;
}
/// <summary>
/// Switches to the currently considered optimal strategy
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void SwitchToOptimalStrategy()
{
BackoffMethod targetMethod;
// Most of the time choose the globally optimal
if (NextRandom() % 100 >= EXPLORATION_PROBABILITY)
{
targetMethod = s_globalBestMethod;
} else
{
// Small probability random exploration
targetMethod = (BackoffMethod)(NextRandom() % 3);
}
if (targetMethod != _currentMethod)
{
_currentMethod = targetMethod;
_spinWait = default; // Reset SpinWait state when switching strategies
_localAttempts = 0;
}
}
public enum BackoffMethod : byte
{
SpinWait = 0,
Yield = 1,
Sleep0 = 2
}
private struct PerformanceStats
{
public long Attempts;
public long LastResetTicks;
public long Successes;
public readonly double SuccessRate
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => Attempts == 0 ? 0.0 : (double)Successes / Attempts;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void RecordAttempt(bool success)
{
_ = Interlocked.Increment(ref Attempts);
if (success) {
_ = Interlocked.Increment(ref Successes);
}
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Reset()
{
Attempts = 0;
Successes = 0;
LastResetTicks = Environment.TickCount64;
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="AdaptiveBackoffStrategy"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading;
/// <summary>
/// Efficient adaptive backoff strategy that dynamically selects optimal backoff methods based on
/// system load and success rate
/// </summary>
public struct AdaptiveBackoffStrategy
{
private const int EVALUATION_INTERVAL_MS = 2000; // Global re-evaluation interval
private const int EXPLORATION_PROBABILITY = 10; // 10% probability to explore non-optimal strategies
private const int FAST_CONVERGENCE_THRESHOLD = 1000; // Fast convergence threshold
// Configuration constants
private const int INITIAL_SPINWAIT_THRESHOLD = 8; // Cheap spin count for first few SpinWait iterations
private const int MIN_SAMPLES_FOR_EVALUATION = 100; // Minimum sample count before starting evaluation
// Global shared state (all threads share learning results)
private static readonly PerformanceStats[] s_globalStats = new PerformanceStats[3];
private static volatile BackoffMethod s_globalBestMethod;
private static volatile int s_evaluationInProgress;
private static long s_lastGlobalEvaluationTicks;
private BackoffMethod _currentMethod;
private int _consecutiveFailures;
private int _localAttempts;
// Instance state
private SpinWait _spinWait;
private uint _fastRng; // Lightweight pseudo-random number generator
// Static constructor: initialize optimal strategy based on system characteristics
static AdaptiveBackoffStrategy()
{
// Based on your benchmark results:
// - Low core count (<= 8): SpinWait has clear advantage
// - Medium core count (<= 16): Close performance, slightly favor SpinWait
// - High core count (>= 20): Sleep0 has significant advantage under high concurrency
int logicalCores = Environment.ProcessorCount;
if (logicalCores <= 8)
{
s_globalBestMethod = BackoffMethod.SpinWait;
} else if (logicalCores <= 16)
{
s_globalBestMethod = BackoffMethod.SpinWait; // Default conservative choice
} else
{
// High core count systems, based on your 64-thread benchmark results
s_globalBestMethod = BackoffMethod.Sleep0;
}
s_lastGlobalEvaluationTicks = Environment.TickCount64;
// Initialize statistics array
for (int i = 0; i < s_globalStats.Length; i++)
{
s_globalStats[i].Reset();
}
}
/// <summary>
/// Gets the currently used backoff method (for debugging/monitoring)
/// </summary>
public readonly BackoffMethod CurrentMethod => _currentMethod;
/// <summary>
/// Gets the globally considered optimal backoff method
/// </summary>
public static BackoffMethod GlobalBestMethod => s_globalBestMethod;
/// <summary>
/// Creates a new adaptive backoff strategy instance
/// </summary>
public static AdaptiveBackoffStrategy Create()
{
return new AdaptiveBackoffStrategy
{
_spinWait = default,
_currentMethod = s_globalBestMethod,
_consecutiveFailures = 0,
_localAttempts = 0,
_fastRng = (uint)(Environment.TickCount64 ^ Thread.CurrentThread.ManagedThreadId * 0x9E3779B9U)
};
}
/// <summary>
/// Gets global statistics for the specified strategy (for monitoring)
/// </summary>
public static (long attempts, long successes, double successRate) GetGlobalStats(BackoffMethod method)
{
PerformanceStats stats = s_globalStats[(int)method];
return (stats.Attempts, stats.Successes, stats.SuccessRate);
}
/// <summary>
/// Called after CAS failure, performs backoff and may switch strategies
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void OnFailureAndBackoff()
{
_consecutiveFailures++;
// Record failure
s_globalStats[(int)_currentMethod].RecordAttempt(success:false);
// Periodically re-evaluate global optimal strategy
if (ShouldEvaluateGlobalStrategy())
{
EvaluateAndUpdateGlobalStrategy();
}
// Decide whether to switch strategies based on failure count and current strategy
if (_consecutiveFailures >= INITIAL_SPINWAIT_THRESHOLD && ShouldSwitchStrategy())
{
SwitchToOptimalStrategy();
}
// Perform the actual backoff operation
PerformBackoff();
}
/// <summary>
/// Called after CAS success, resets state
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void OnSuccess()
{
// Record success
s_globalStats[(int)_currentMethod].RecordAttempt(success:true);
// Reset state
_spinWait = default;
_consecutiveFailures = 0;
// Fast convergence: tend to keep current strategy when continuously successful
if (_localAttempts < FAST_CONVERGENCE_THRESHOLD)
{
_localAttempts++;
}
}
/// <summary>
/// Evaluates and updates the global optimal strategy
/// </summary>
private static void EvaluateAndUpdateGlobalStrategy()
{
// Use CAS to ensure only one thread performs evaluation
if (Interlocked.CompareExchange(ref s_evaluationInProgress, 1, 0) != 0)
{
return;
}
try
{
long now = Environment.TickCount64;
// Double-check to avoid duplicate evaluation
if (now - Volatile.Read(ref s_lastGlobalEvaluationTicks) <= EVALUATION_INTERVAL_MS)
{
return;
}
BackoffMethod bestMethod = s_globalBestMethod;
double bestSuccessRate = -1.0;
// Find the strategy with the highest success rate
for (int i = 0; i < s_globalStats.Length; i++)
{
PerformanceStats stats = s_globalStats[i];
if (stats.Attempts >= MIN_SAMPLES_FOR_EVALUATION)
{
double successRate = stats.SuccessRate;
if (successRate > bestSuccessRate)
{
bestSuccessRate = successRate;
bestMethod = (BackoffMethod)i;
}
}
}
// Update global optimal strategy
s_globalBestMethod = bestMethod;
Volatile.Write(ref s_lastGlobalEvaluationTicks, now);
// Reset statistics (maintain exploratory behavior)
for (int i = 0; i < s_globalStats.Length; i++)
{
s_globalStats[i].Reset();
}
}
finally
{
Volatile.Write(ref s_evaluationInProgress, 0);
}
}
/// <summary>
/// Fast pseudo-random number generator (LCG)
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private uint NextRandom()
{
_fastRng = _fastRng * 1103515245U + 12345U;
return _fastRng;
}
/// <summary>
/// Performs the actual backoff operation
/// </summary>
[MethodImpl(MethodImplOptions.NoInlining)] // Avoid inlining to reduce hot path code size
private void PerformBackoff()
{
switch (_currentMethod)
{
case BackoffMethod.SpinWait:
if (_consecutiveFailures <= INITIAL_SPINWAIT_THRESHOLD)
{
_spinWait.SpinOnce(sleep1Threshold:-1);
} else if (_spinWait.NextSpinWillYield)
{
int yields = (int)(NextRandom() & 7) + 1; // 1-8 times
for (int i = 0; i < yields; i++) {
_ = Thread.Yield();
}
} else
{
_spinWait.SpinOnce(sleep1Threshold:-1);
}
break;
case BackoffMethod.Yield:
_ = Thread.Yield();
break;
case BackoffMethod.Sleep0:
Thread.Sleep(0);
break;
default:
break;
}
}
/// <summary>
/// Checks whether global strategy evaluation should be performed
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static bool ShouldEvaluateGlobalStrategy()
{
long now = Environment.TickCount64;
return now - Volatile.Read(ref s_lastGlobalEvaluationTicks) > EVALUATION_INTERVAL_MS;
}
/// <summary>
/// Determines whether strategy should be switched
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private bool ShouldSwitchStrategy()
{
// Fast convergence phase: reduce strategy switching
if (_localAttempts < FAST_CONVERGENCE_THRESHOLD && _consecutiveFailures < 32)
{
return false;
}
// Small probability exploration: occasionally try other strategies even when current is optimal
return NextRandom() % 100 < EXPLORATION_PROBABILITY || _currentMethod != s_globalBestMethod;
}
/// <summary>
/// Switches to the currently considered optimal strategy
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void SwitchToOptimalStrategy()
{
BackoffMethod targetMethod;
// Most of the time choose the globally optimal
if (NextRandom() % 100 >= EXPLORATION_PROBABILITY)
{
targetMethod = s_globalBestMethod;
} else
{
// Small probability random exploration
targetMethod = (BackoffMethod)(NextRandom() % 3);
}
if (targetMethod != _currentMethod)
{
_currentMethod = targetMethod;
_spinWait = default; // Reset SpinWait state when switching strategies
_localAttempts = 0;
}
}
public enum BackoffMethod : byte
{
SpinWait = 0,
Yield = 1,
Sleep0 = 2
}
private struct PerformanceStats
{
public long Attempts;
public long LastResetTicks;
public long Successes;
public readonly double SuccessRate
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => Attempts == 0 ? 0.0 : (double)Successes / Attempts;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void RecordAttempt(bool success)
{
_ = Interlocked.Increment(ref Attempts);
if (success) {
_ = Interlocked.Increment(ref Successes);
}
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Reset()
{
Attempts = 0;
Successes = 0;
LastResetTicks = Environment.TickCount64;
}
}
}
StgSharp.Common/Threading/BufferExpansionLock.cs
+125 -125
View File
@@ -1,125 +1,125 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="BufferExpansionLock"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.Threading
{
/// <summary>
/// Provide a lock for managing concurrent access to a buffer that can be expanded and use a
/// common two-pase-copy expanding mechanism.
/// </summary>
public class BufferExpansionLock : IDisposable
{
private static readonly ThreadLocal<int> Count = new(() => 0);
private readonly ReaderWriterLockSlim _copyLock;
private readonly ReaderWriterLockSlim _expansionLock;
public BufferExpansionLock()
{
_expansionLock = new ReaderWriterLockSlim(LockRecursionPolicy.NoRecursion);
_copyLock = new ReaderWriterLockSlim(LockRecursionPolicy.NoRecursion);
}
public bool IsThreadReadingMetaData
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => _expansionLock.IsReadLockHeld;
}
public void Dispose()
{
_expansionLock?.Dispose();
_copyLock?.Dispose();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void EnterBufferCopy()
{
_copyLock.EnterWriteLock();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void EnterBufferRead()
{
int c = Count.Value;
if (c == 0) {
_copyLock.EnterReadLock();
}
Count.Value = c + 1;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void EnterExpansionProcess()
{
_expansionLock.EnterWriteLock();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void EnterMetaDataRead()
{
_expansionLock.EnterReadLock();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ExitBufferCopy()
{
_copyLock.ExitWriteLock();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ExitBufferRead()
{
int c = Count.Value - 1;
Count.Value = c;
if (c == 0) {
_copyLock.ExitReadLock();
}
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ExitExpansionProcess()
{
_expansionLock.ExitWriteLock();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ExitMetaDataRead()
{
_expansionLock.ExitReadLock();
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="BufferExpansionLock"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.Threading
{
/// <summary>
/// Provide a lock for managing concurrent access to a buffer that can be expanded and use a
/// common two-pase-copy expanding mechanism.
/// </summary>
public class BufferExpansionLock : IDisposable
{
private static readonly ThreadLocal<int> Count = new(() => 0);
private readonly ReaderWriterLockSlim _copyLock;
private readonly ReaderWriterLockSlim _expansionLock;
public BufferExpansionLock()
{
_expansionLock = new ReaderWriterLockSlim(LockRecursionPolicy.NoRecursion);
_copyLock = new ReaderWriterLockSlim(LockRecursionPolicy.NoRecursion);
}
public bool IsThreadReadingMetaData
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => _expansionLock.IsReadLockHeld;
}
public void Dispose()
{
_expansionLock?.Dispose();
_copyLock?.Dispose();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void EnterBufferCopy()
{
_copyLock.EnterWriteLock();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void EnterBufferRead()
{
int c = Count.Value;
if (c == 0) {
_copyLock.EnterReadLock();
}
Count.Value = c + 1;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void EnterExpansionProcess()
{
_expansionLock.EnterWriteLock();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void EnterMetaDataRead()
{
_expansionLock.EnterReadLock();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ExitBufferCopy()
{
_copyLock.ExitWriteLock();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ExitBufferRead()
{
int c = Count.Value - 1;
Count.Value = c;
if (c == 0) {
_copyLock.ExitReadLock();
}
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ExitExpansionProcess()
{
_expansionLock.ExitWriteLock();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ExitMetaDataRead()
{
_expansionLock.ExitReadLock();
}
}
}
StgSharp.Common/Threading/CancellableTask.cs
+154 -154
View File
@@ -1,154 +1,154 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="CancellableTask"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.Threading
{
public class CancellableTask : IDisposable
{
private bool disposedValue;
private CancellationTokenSource cts;
private protected Action<CancellationToken> _startup;
private protected Task _internalTask;
private protected CancellableTask()
{
cts = new CancellationTokenSource();
}
private protected CancellableTask(Action<CancellationToken> action)
{
_internalTask = new Task(InternalAction);
cts = new CancellationTokenSource();
}
public bool IsCompleted
{
get => _internalTask.IsCompleted;
}
public CancellationTokenSource Cts
{
get => cts;
}
public int Id
{
get => _internalTask.Id;
}
public void Cancel()
{
Cts.Cancel();
}
public void CancelAndWait()
{
Cts.Cancel();
_internalTask.Wait();
}
// ~CancellableTask()
// {
// Dispose(disposing: false);
// }
public void Dispose()
{
Dispose(disposing:true);
GC.SuppressFinalize(this);
}
public static CancellableTask Run(Action<CancellationToken> startup)
{
CancellableTask ret = new CancellableTask
{
_startup = startup
};
ret._internalTask = Task.Run(ret.InternalAction);
return ret;
}
public void Wait()
{
_internalTask.Wait();
}
public static void WaitAll(IEnumerable<CancellableTask> tasks)
{
IEnumerable<Task> convertedTasks = tasks.Select<CancellableTask, Task>(
ct => ct._internalTask);
Task.WaitAll(convertedTasks.ToArray());
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal void InternalAction()
{
_startup(Cts.Token);
}
protected virtual void Dispose(bool disposing)
{
if (!disposedValue)
{
if (disposing)
{
if (IsCompleted)
{
cts.Dispose();
_internalTask.Dispose();
} else
{
throw new InvalidOperationException(
"Cannot dispose a cancellable task before completed");
}
}
disposedValue = true;
}
}
public static implicit operator Task(CancellableTask task)
{
return task._internalTask;
}
~CancellableTask()
{
Cts.Dispose();
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="CancellableTask"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.Threading
{
public class CancellableTask : IDisposable
{
private bool disposedValue;
private CancellationTokenSource cts;
private protected Action<CancellationToken> _startup;
private protected Task _internalTask;
private protected CancellableTask()
{
cts = new CancellationTokenSource();
}
private protected CancellableTask(Action<CancellationToken> action)
{
_internalTask = new Task(InternalAction);
cts = new CancellationTokenSource();
}
public bool IsCompleted
{
get => _internalTask.IsCompleted;
}
public CancellationTokenSource Cts
{
get => cts;
}
public int Id
{
get => _internalTask.Id;
}
public void Cancel()
{
Cts.Cancel();
}
public void CancelAndWait()
{
Cts.Cancel();
_internalTask.Wait();
}
// ~CancellableTask()
// {
// Dispose(disposing: false);
// }
public void Dispose()
{
Dispose(disposing:true);
GC.SuppressFinalize(this);
}
public static CancellableTask Run(Action<CancellationToken> startup)
{
CancellableTask ret = new CancellableTask
{
_startup = startup
};
ret._internalTask = Task.Run(ret.InternalAction);
return ret;
}
public void Wait()
{
_internalTask.Wait();
}
public static void WaitAll(IEnumerable<CancellableTask> tasks)
{
IEnumerable<Task> convertedTasks = tasks.Select<CancellableTask, Task>(
ct => ct._internalTask);
Task.WaitAll(convertedTasks.ToArray());
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal void InternalAction()
{
_startup(Cts.Token);
}
protected virtual void Dispose(bool disposing)
{
if (!disposedValue)
{
if (disposing)
{
if (IsCompleted)
{
cts.Dispose();
_internalTask.Dispose();
} else
{
throw new InvalidOperationException(
"Cannot dispose a cancellable task before completed");
}
}
disposedValue = true;
}
}
public static implicit operator Task(CancellableTask task)
{
return task._internalTask;
}
~CancellableTask()
{
Cts.Dispose();
}
}
}
StgSharp.Common/Threading/EmptyThread.cs
+58 -58
View File
@@ -1,58 +1,58 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="EmptyThread"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.Threading
{
public static partial class ThreadHelper
{
private static readonly Thread _emptyThreadCache = new Thread(() => { });
static ThreadHelper()
{
_emptyThreadCache.Start();
}
/// <summary>
/// An individual thread doing nothing, and its <see cref="Thread.IsAlive" /> <see
/// langword="property" /> always returns false.
/// </summary>
public static Thread EmptyThread
{
get => _emptyThreadCache;
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="EmptyThread"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.Threading
{
public static partial class ThreadHelper
{
private static readonly Thread _emptyThreadCache = new Thread(() => { });
static ThreadHelper()
{
_emptyThreadCache.Start();
}
/// <summary>
/// An individual thread doing nothing, and its <see cref="Thread.IsAlive" /> <see
/// langword="property" /> always returns false.
/// </summary>
public static Thread EmptyThread
{
get => _emptyThreadCache;
}
}
}
StgSharp.Common/Threading/MainThreadAssert.cs
+51 -51
View File
@@ -1,51 +1,51 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="MainThreadAssert"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Diagnostics;
using System.Threading;
namespace StgSharp.Threading
{
public static partial class ThreadHelper
{
private static readonly int mainThreadId = Environment.CurrentManagedThreadId;
private static readonly ThreadLocal<bool> isNotMainThread = new(
() => Environment.CurrentManagedThreadId != mainThreadId);
[Conditional("DEBUG")]
internal static void MainThreadAssert()
{
if (isNotMainThread.Value) {
throw new InvalidOperationException(
$"Attempt to call sensitive operation which can be only called from main thread from thread {Environment.CurrentManagedThreadId}.");
}
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="MainThreadAssert"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Diagnostics;
using System.Threading;
namespace StgSharp.Threading
{
public static partial class ThreadHelper
{
private static readonly int mainThreadId = Environment.CurrentManagedThreadId;
private static readonly ThreadLocal<bool> isNotMainThread = new(
() => Environment.CurrentManagedThreadId != mainThreadId);
[Conditional("DEBUG")]
internal static void MainThreadAssert()
{
if (isNotMainThread.Value) {
throw new InvalidOperationException(
$"Attempt to call sensitive operation which can be only called from main thread from thread {Environment.CurrentManagedThreadId}.");
}
}
}
}
StgSharp.Common/Threading/NumaBuffer.cs
+47 -47
View File
@@ -1,47 +1,47 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="NumaBuffer"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace StgSharp.Common.Threading
{
public unsafe struct NumaBuffer<T> where T : unmanaged
{
public readonly T* Buffers;
public readonly int BufferColumnCapacity;
public readonly int BufferRowCapacity;
public int CurrentColumnCount;
public int CurrentRowCount;
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="NumaBuffer"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace StgSharp.Common.Threading
{
public unsafe struct NumaBuffer<T> where T : unmanaged
{
public readonly T* Buffers;
public readonly int BufferColumnCapacity;
public readonly int BufferRowCapacity;
public int CurrentColumnCount;
public int CurrentRowCount;
}
}
StgSharp.Common/Threading/SingleConsumerBell.cs
+174 -174
View File
@@ -1,175 +1,175 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="SingleConsumerBell"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Threading;
namespace StgSharp.Threading
{
/// <summary>
/// A single-consumer bell/signal primitive.
/// </summary>
/// <remarks>
/// <para> Producers can call <see cref="Set" /> multiple times to publish signals. The bell
/// keeps a pending count of not-yet-consumed signals. The consumer uses <see
/// cref="Wait(object, CancellationToken)" /> or <see cref="TryWait(object)" /> to consume
/// signals.</para> <para> When a producer increments the pending count from 0 to 1, the
/// internal event is set to wake the consumer. Consuming via <see cref="TryWait(object)" />
/// or <see cref="Wait(object, CancellationToken)" /> clears the entire pending count
/// (aggregate consumption). If new signals arrive while resetting, the event is set again to
/// avoid missed signals.</para> <para> Ownership: only the specified <c> owner </c> can call
/// the consumer APIs.</para>
/// </remarks>
public sealed class SingleConsumerBell : IDisposable
{
private volatile bool _disposed;
private int _pending; // not volatile: used with Interlocked/Volatile APIs to avoid CS0420
private readonly ManualResetEventSlim _evt;
private readonly object _owner;
/// <summary>
/// Initializes a new instance of <see cref="SingleConsumerBell" />.
/// </summary>
/// <param name="owner">
/// The unique consumer identity; subsequent calls to Wait/TryWait must pass the same
/// reference.
/// </param>
/// <param name="spinCount">
/// Spin count passed to the underlying <see cref="ManualResetEventSlim" />.
/// </param>
/// <exception cref="ArgumentNullException">
/// Thrown if <paramref name="owner" /> is null.
/// </exception>
public SingleConsumerBell(object owner, int spinCount = 100)
{
_owner = owner ?? throw new ArgumentNullException(nameof(owner));
_evt = new ManualResetEventSlim(false, spinCount);
}
/// <summary>
/// Disposes the bell, sets the event to wake any waiting consumer, and releases
/// resources.
/// </summary>
public void Dispose()
{
_disposed = true;
_evt.Set();
_evt.Dispose();
}
/// <summary>
/// Publishes a signal by incrementing the not-yet-consumed count and returns the current
/// pending count.
/// </summary>
/// <remarks>
/// If the pending count transitions from 0 to 1, the event is set to wake the consumer.
/// </remarks>
/// <returns>
/// The current pending count including this call.
/// </returns>
public int Set()
{
if (_disposed) {
return Volatile.Read(ref _pending);
}
// Increment pending; if we transitioned 0 -> 1, set the event to wake the consumer
int newCount = Interlocked.Increment(ref _pending);
if (newCount == 1) {
_evt.Set();
}
return newCount;
}
/// <summary>
/// Attempts to consume any pending signals without blocking.
/// </summary>
/// <param name="consumer">
/// The consumer object; must be the same reference provided at construction.
/// </param>
/// <returns>
/// <c> true </c> if a signal was available (pending or event set); otherwise <c> false
/// </c>.
/// </returns>
/// <exception cref="InvalidOperationException">
/// Thrown if <paramref name="consumer" /> is not the owner.
/// </exception>
public bool TryWait(object consumer)
{
if (!ReferenceEquals(_owner, consumer)) {
throw new InvalidOperationException("This bell is owned by another consumer.");
}
if (!_evt.IsSet && Volatile.Read(ref _pending) == 0) {
return false;
}
// Aggregate consumption: clear pending and reset event
_ = Interlocked.Exchange(ref _pending, 0);
_evt.Reset();
// If concurrent Set happened during reset, re-set the event to avoid lost signals
if (Volatile.Read(ref _pending) != 0)
{
_evt.Set();
}
return true;
}
/// <summary>
/// Blocks until a signal is available, then consumes all pending signals.
/// </summary>
/// <param name="consumer">
/// The consumer object; must be the same reference provided at construction.
/// </param>
/// <param name="cancellationToken">
/// A token to cancel the wait.
/// </param>
/// <exception cref="InvalidOperationException">
/// Thrown if <paramref name="consumer" /> is not the owner.
/// </exception>
public void Wait(object consumer, CancellationToken cancellationToken = default)
{
if (!ReferenceEquals(_owner, consumer)) {
throw new InvalidOperationException("This bell is owned by another consumer.");
}
_evt.Wait(cancellationToken);
// Consume this round of signals: clear pending and handle reset/concurrency
_ = Interlocked.Exchange(ref _pending, 0);
_evt.Reset();
if (Volatile.Read(ref _pending) != 0) {
_evt.Set();
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="SingleConsumerBell"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Threading;
namespace StgSharp.Threading
{
/// <summary>
/// A single-consumer bell/signal primitive.
/// </summary>
/// <remarks>
/// <para> Producers can call <see cref="Set" /> multiple times to publish signals. The bell
/// keeps a pending count of not-yet-consumed signals. The consumer uses <see
/// cref="Wait(object, CancellationToken)" /> or <see cref="TryWait(object)" /> to consume
/// signals.</para> <para> When a producer increments the pending count from 0 to 1, the
/// internal event is set to wake the consumer. Consuming via <see cref="TryWait(object)" />
/// or <see cref="Wait(object, CancellationToken)" /> clears the entire pending count
/// (aggregate consumption). If new signals arrive while resetting, the event is set again to
/// avoid missed signals.</para> <para> Ownership: only the specified <c> owner </c> can call
/// the consumer APIs.</para>
/// </remarks>
public sealed class SingleConsumerBell : IDisposable
{
private volatile bool _disposed;
private int _pending; // not volatile: used with Interlocked/Volatile APIs to avoid CS0420
private readonly ManualResetEventSlim _evt;
private readonly object _owner;
/// <summary>
/// Initializes a new instance of <see cref="SingleConsumerBell" />.
/// </summary>
/// <param name="owner">
/// The unique consumer identity; subsequent calls to Wait/TryWait must pass the same
/// reference.
/// </param>
/// <param name="spinCount">
/// Spin count passed to the underlying <see cref="ManualResetEventSlim" />.
/// </param>
/// <exception cref="ArgumentNullException">
/// Thrown if <paramref name="owner" /> is null.
/// </exception>
public SingleConsumerBell(object owner, int spinCount = 100)
{
_owner = owner ?? throw new ArgumentNullException(nameof(owner));
_evt = new ManualResetEventSlim(false, spinCount);
}
/// <summary>
/// Disposes the bell, sets the event to wake any waiting consumer, and releases
/// resources.
/// </summary>
public void Dispose()
{
_disposed = true;
_evt.Set();
_evt.Dispose();
}
/// <summary>
/// Publishes a signal by incrementing the not-yet-consumed count and returns the current
/// pending count.
/// </summary>
/// <remarks>
/// If the pending count transitions from 0 to 1, the event is set to wake the consumer.
/// </remarks>
/// <returns>
/// The current pending count including this call.
/// </returns>
public int Set()
{
if (_disposed) {
return Volatile.Read(ref _pending);
}
// Increment pending; if we transitioned 0 -> 1, set the event to wake the consumer
int newCount = Interlocked.Increment(ref _pending);
if (newCount == 1) {
_evt.Set();
}
return newCount;
}
/// <summary>
/// Attempts to consume any pending signals without blocking.
/// </summary>
/// <param name="consumer">
/// The consumer object; must be the same reference provided at construction.
/// </param>
/// <returns>
/// <c> true </c> if a signal was available (pending or event set); otherwise <c> false
/// </c>.
/// </returns>
/// <exception cref="InvalidOperationException">
/// Thrown if <paramref name="consumer" /> is not the owner.
/// </exception>
public bool TryWait(object consumer)
{
if (!ReferenceEquals(_owner, consumer)) {
throw new InvalidOperationException("This bell is owned by another consumer.");
}
if (!_evt.IsSet && Volatile.Read(ref _pending) == 0) {
return false;
}
// Aggregate consumption: clear pending and reset event
_ = Interlocked.Exchange(ref _pending, 0);
_evt.Reset();
// If concurrent Set happened during reset, re-set the event to avoid lost signals
if (Volatile.Read(ref _pending) != 0)
{
_evt.Set();
}
return true;
}
/// <summary>
/// Blocks until a signal is available, then consumes all pending signals.
/// </summary>
/// <param name="consumer">
/// The consumer object; must be the same reference provided at construction.
/// </param>
/// <param name="cancellationToken">
/// A token to cancel the wait.
/// </param>
/// <exception cref="InvalidOperationException">
/// Thrown if <paramref name="consumer" /> is not the owner.
/// </exception>
public void Wait(object consumer, CancellationToken cancellationToken = default)
{
if (!ReferenceEquals(_owner, consumer)) {
throw new InvalidOperationException("This bell is owned by another consumer.");
}
_evt.Wait(cancellationToken);
// Consume this round of signals: clear pending and handle reset/concurrency
_ = Interlocked.Exchange(ref _pending, 0);
_evt.Reset();
if (Volatile.Read(ref _pending) != 0) {
_evt.Set();
}
}
}
}
StgSharp.Common/Threading/VariableCapacitySemaphore.cs
+316 -316
View File
@@ -1,316 +1,316 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="VariableCapacitySemaphore"
// Project: StgSharp.Common
// AuthorGroup: Nitload
// License: MIT
// -----------------------------------------------------------------------
using System;
using System.Runtime.CompilerServices;
using System.Threading;
namespace StgSharp.Threading
{
/// <summary>
/// A counting semaphore with adjustable maximum capacity at runtime. Implemented using one or
/// more ManualResetEventSlim (MRES) gates and atomic counters to avoid kernel semaphore
/// overhead while supporting high-frequency Release/Wait patterns. Notes: - This is a
/// counting primitive (not a barrier). Permits accumulate up to <see cref="MaxPermits" />. -
/// Reducing <see cref="MaxPermits" /> does not revoke already issued permits; it only clamps
/// future releases. - Striped MRES can reduce contention in highly parallel waits.
/// </summary>
public sealed class VariableCapacitySemaphore : IDisposable
{
// MRES stripes (one gate per stripe). All stripes are set when transitioning 0 -> >0 and reset at >0 -> 0.
private readonly ManualResetEventSlim[] _events;
private volatile int _current;
private volatile int _max;
private readonly int _stripeMask; // for power-of-two stripe counts; otherwise -1
/// <summary>
/// Create a new variable-capacity semaphore.
/// </summary>
/// <param name="initialCount">
/// Initial available permits (clamped to [0, maxPermits]).
/// </param>
/// <param name="maxPermits">
/// Logical maximum permits (>= 1).
/// </param>
/// <param name="stripes">
/// Optional stripe count for MRES (<= 0 to auto choose by CPU).
/// </param>
public VariableCapacitySemaphore(int initialCount = 0, int maxPermits = 1, int stripes = 0)
{
ArgumentOutOfRangeException.ThrowIfLessThan(maxPermits, 1);
ArgumentOutOfRangeException.ThrowIfNegative(initialCount);
_max = maxPermits;
_current = Math.Min(initialCount, maxPermits);
int stripeCount = stripes > 0 ? stripes : ClampToPow2(Math.Min(Environment.ProcessorCount, 8));
_events = new ManualResetEventSlim[stripeCount];
bool set = _current > 0;
for (int i = 0; i < stripeCount; i++) {
_events[i] = new ManualResetEventSlim(set, 0);
}
_stripeMask = IsPow2(stripeCount) ? (stripeCount - 1) : -1;
}
/// <summary>
/// Current number of available permits.
/// </summary>
public int CurrentCount => Volatile.Read(ref _current);
/// <summary>
/// Logical maximum number of permits. Can be changed at runtime.
/// </summary>
public int MaxPermits
{
get => Volatile.Read(ref _max);
set
{
ArgumentOutOfRangeException.ThrowIfLessThan(value, 1);
Volatile.Write(ref _max, value);
// Clamp current to not exceed the new max (do not block; let Wait() consume down).
while (true)
{
int cur = Volatile.Read(ref _current);
if (cur <= value)
{
break;
}
if (Interlocked.CompareExchange(ref _current, value, cur) == cur)
{
break;
}
}
// Reflect availability to gates
if (Volatile.Read(ref _current) > 0)
{
SetAll();
} else
{
ResetAll();
}
}
}
public void Dispose()
{
foreach (ManualResetEventSlim e in _events) {
e.Dispose();
}
}
/// <summary>
/// Release a single permit if under <see cref="MaxPermits" />. Returns the previous
/// count.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public int Release()
{
while (true)
{
int cur = Volatile.Read(ref _current);
int max = Volatile.Read(ref _max);
if (cur >= max) {
return cur;
}
if (Interlocked.CompareExchange(ref _current, cur + 1, cur) == cur)
{
if (cur == 0) {
SetAll();
}
return cur;
}
}
}
/// <summary>
/// Release multiple permits (capped by <see cref="MaxPermits" />). Returns the previous
/// count.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public int Release(int permits)
{
ArgumentOutOfRangeException.ThrowIfNegativeOrZero(permits);
while (true)
{
int cur = Volatile.Read(ref _current);
int max = Volatile.Read(ref _max);
if (cur >= max) {
return cur;
}
long target = (long)cur + permits;
if (target > max) {
target = max;
}
if (Interlocked.CompareExchange(ref _current, (int)target, cur) == cur)
{
if (cur == 0) {
SetAll();
}
return cur;
}
}
}
/// <summary>
/// Try to take a permit without blocking. Returns true if acquired.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool TryWait()
{
while (true)
{
int cur = Volatile.Read(ref _current);
if (cur == 0) {
return false;
}
if (Interlocked.CompareExchange(ref _current, cur - 1, cur) == cur)
{
if (cur - 1 == 0) {
ResetAll();
}
return true;
}
}
}
/// <summary>
/// Block until a permit is available, then consume one.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Wait()
{
ManualResetEventSlim gate = GetStripeEvent();
SpinWait spinner = new SpinWait();
while (true)
{
int cur = Volatile.Read(ref _current);
if (cur > 0 && Interlocked.CompareExchange(ref _current, cur - 1, cur) == cur)
{
if (cur - 1 == 0) {
ResetAll();
}
return;
}
if (!spinner.NextSpinWillYield)
{
spinner.SpinOnce();
} else
{
gate.Wait();
}
}
}
/// <summary>
/// Block until a permit is available or canceled, then consume one.
/// </summary>
public bool Wait(int millisecondsTimeout, CancellationToken cancellationToken = default)
{
ManualResetEventSlim gate = GetStripeEvent();
SpinWait spinner = new SpinWait();
int start = Environment.TickCount;
int remaining = millisecondsTimeout;
while (true)
{
cancellationToken.ThrowIfCancellationRequested();
int cur = Volatile.Read(ref _current);
if (cur > 0 && Interlocked.CompareExchange(ref _current, cur - 1, cur) == cur)
{
if (cur - 1 == 0) {
ResetAll();
}
return true;
}
if (!spinner.NextSpinWillYield)
{
spinner.SpinOnce();
continue;
}
if (millisecondsTimeout < 0)
{
gate.Wait(cancellationToken);
continue;
}
if (remaining <= 0) {
return false;
}
if (!gate.Wait(remaining, cancellationToken)) {
return false;
}
remaining = millisecondsTimeout - (Environment.TickCount - start);
}
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static int ClampToPow2(int x)
{
if (x < 1) {
x = 1;
}
if (x > 8) {
x = 8;
}
int p = 1;
while ((p << 1) <= x) {
p <<= 1;
}
return p;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private ManualResetEventSlim GetStripeEvent()
{
int id = Environment.CurrentManagedThreadId & 0x7fffffff;
if (_stripeMask != -1) {
return _events[id & _stripeMask];
}
int idx = id % _events.Length;
return _events[idx];
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static bool IsPow2(int x) => (x & (x - 1)) == 0;
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void ResetAll()
{
foreach (ManualResetEventSlim e in _events) {
e.Reset();
}
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void SetAll()
{
foreach (ManualResetEventSlim e in _events) {
e.Set();
}
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="VariableCapacitySemaphore"
// Project: StgSharp.Common
// AuthorGroup: Nitload
// License: MIT
// -----------------------------------------------------------------------
using System;
using System.Runtime.CompilerServices;
using System.Threading;
namespace StgSharp.Threading
{
/// <summary>
/// A counting semaphore with adjustable maximum capacity at runtime. Implemented using one or
/// more ManualResetEventSlim (MRES) gates and atomic counters to avoid kernel semaphore
/// overhead while supporting high-frequency Release/Wait patterns. Notes: - This is a
/// counting primitive (not a barrier). Permits accumulate up to <see cref="MaxPermits" />. -
/// Reducing <see cref="MaxPermits" /> does not revoke already issued permits; it only clamps
/// future releases. - Striped MRES can reduce contention in highly parallel waits.
/// </summary>
public sealed class VariableCapacitySemaphore : IDisposable
{
// MRES stripes (one gate per stripe). All stripes are set when transitioning 0 -> >0 and reset at >0 -> 0.
private readonly ManualResetEventSlim[] _events;
private volatile int _current;
private volatile int _max;
private readonly int _stripeMask; // for power-of-two stripe counts; otherwise -1
/// <summary>
/// Create a new variable-capacity semaphore.
/// </summary>
/// <param name="initialCount">
/// Initial available permits (clamped to [0, maxPermits]).
/// </param>
/// <param name="maxPermits">
/// Logical maximum permits (>= 1).
/// </param>
/// <param name="stripes">
/// Optional stripe count for MRES (<= 0 to auto choose by CPU).
/// </param>
public VariableCapacitySemaphore(int initialCount = 0, int maxPermits = 1, int stripes = 0)
{
ArgumentOutOfRangeException.ThrowIfLessThan(maxPermits, 1);
ArgumentOutOfRangeException.ThrowIfNegative(initialCount);
_max = maxPermits;
_current = Math.Min(initialCount, maxPermits);
int stripeCount = stripes > 0 ? stripes : ClampToPow2(Math.Min(Environment.ProcessorCount, 8));
_events = new ManualResetEventSlim[stripeCount];
bool set = _current > 0;
for (int i = 0; i < stripeCount; i++) {
_events[i] = new ManualResetEventSlim(set, 0);
}
_stripeMask = IsPow2(stripeCount) ? (stripeCount - 1) : -1;
}
/// <summary>
/// Current number of available permits.
/// </summary>
public int CurrentCount => Volatile.Read(ref _current);
/// <summary>
/// Logical maximum number of permits. Can be changed at runtime.
/// </summary>
public int MaxPermits
{
get => Volatile.Read(ref _max);
set
{
ArgumentOutOfRangeException.ThrowIfLessThan(value, 1);
Volatile.Write(ref _max, value);
// Clamp current to not exceed the new max (do not block; let Wait() consume down).
while (true)
{
int cur = Volatile.Read(ref _current);
if (cur <= value)
{
break;
}
if (Interlocked.CompareExchange(ref _current, value, cur) == cur)
{
break;
}
}
// Reflect availability to gates
if (Volatile.Read(ref _current) > 0)
{
SetAll();
} else
{
ResetAll();
}
}
}
public void Dispose()
{
foreach (ManualResetEventSlim e in _events) {
e.Dispose();
}
}
/// <summary>
/// Release a single permit if under <see cref="MaxPermits" />. Returns the previous
/// count.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public int Release()
{
while (true)
{
int cur = Volatile.Read(ref _current);
int max = Volatile.Read(ref _max);
if (cur >= max) {
return cur;
}
if (Interlocked.CompareExchange(ref _current, cur + 1, cur) == cur)
{
if (cur == 0) {
SetAll();
}
return cur;
}
}
}
/// <summary>
/// Release multiple permits (capped by <see cref="MaxPermits" />). Returns the previous
/// count.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public int Release(int permits)
{
ArgumentOutOfRangeException.ThrowIfNegativeOrZero(permits);
while (true)
{
int cur = Volatile.Read(ref _current);
int max = Volatile.Read(ref _max);
if (cur >= max) {
return cur;
}
long target = (long)cur + permits;
if (target > max) {
target = max;
}
if (Interlocked.CompareExchange(ref _current, (int)target, cur) == cur)
{
if (cur == 0) {
SetAll();
}
return cur;
}
}
}
/// <summary>
/// Try to take a permit without blocking. Returns true if acquired.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool TryWait()
{
while (true)
{
int cur = Volatile.Read(ref _current);
if (cur == 0) {
return false;
}
if (Interlocked.CompareExchange(ref _current, cur - 1, cur) == cur)
{
if (cur - 1 == 0) {
ResetAll();
}
return true;
}
}
}
/// <summary>
/// Block until a permit is available, then consume one.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Wait()
{
ManualResetEventSlim gate = GetStripeEvent();
SpinWait spinner = new SpinWait();
while (true)
{
int cur = Volatile.Read(ref _current);
if (cur > 0 && Interlocked.CompareExchange(ref _current, cur - 1, cur) == cur)
{
if (cur - 1 == 0) {
ResetAll();
}
return;
}
if (!spinner.NextSpinWillYield)
{
spinner.SpinOnce();
} else
{
gate.Wait();
}
}
}
/// <summary>
/// Block until a permit is available or canceled, then consume one.
/// </summary>
public bool Wait(int millisecondsTimeout, CancellationToken cancellationToken = default)
{
ManualResetEventSlim gate = GetStripeEvent();
SpinWait spinner = new SpinWait();
int start = Environment.TickCount;
int remaining = millisecondsTimeout;
while (true)
{
cancellationToken.ThrowIfCancellationRequested();
int cur = Volatile.Read(ref _current);
if (cur > 0 && Interlocked.CompareExchange(ref _current, cur - 1, cur) == cur)
{
if (cur - 1 == 0) {
ResetAll();
}
return true;
}
if (!spinner.NextSpinWillYield)
{
spinner.SpinOnce();
continue;
}
if (millisecondsTimeout < 0)
{
gate.Wait(cancellationToken);
continue;
}
if (remaining <= 0) {
return false;
}
if (!gate.Wait(remaining, cancellationToken)) {
return false;
}
remaining = millisecondsTimeout - (Environment.TickCount - start);
}
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static int ClampToPow2(int x)
{
if (x < 1) {
x = 1;
}
if (x > 8) {
x = 8;
}
int p = 1;
while ((p << 1) <= x) {
p <<= 1;
}
return p;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private ManualResetEventSlim GetStripeEvent()
{
int id = Environment.CurrentManagedThreadId & 0x7fffffff;
if (_stripeMask != -1) {
return _events[id & _stripeMask];
}
int idx = id % _events.Length;
return _events[idx];
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static bool IsPow2(int x) => (x & (x - 1)) == 0;
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void ResetAll()
{
foreach (ManualResetEventSlim e in _events) {
e.Reset();
}
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void SetAll()
{
foreach (ManualResetEventSlim e in _events) {
e.Set();
}
}
}
}