-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
✨ Creates REstate.Concurrency.Primitives
- Adds the new project - Updates the Sempahore example to use the library instead of manually crafting a machine
- Loading branch information
Ovan Crone
authored and
Ovan Crone
committed
Feb 2, 2018
1 parent
89f0a0e
commit ef97767
Showing
12 changed files
with
255 additions
and
57 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
137 changes: 137 additions & 0 deletions
137
src/REstate.Concurrency.Primitives/ConcurrencyPrimitiveExtensions.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,137 @@ | ||
| using System; | ||
| using System.Linq; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using REstate.Concurrency.Primitives; | ||
| using REstate.Schematics; | ||
|
|
||
| // ReSharper disable once CheckNamespace | ||
| namespace REstate | ||
| { | ||
| public static class ConcurrencyPrimitiveExtensions | ||
| { | ||
| /// <summary> | ||
| /// Creates a new mutex (lock) using Machine state as the synchronization root. | ||
| /// </summary> | ||
| /// <param name="restateAgent">The agent to use to build the mutex</param> | ||
| /// <param name="retryLimit">The number of attempts to try obtaining a lock; if null then no limit</param> | ||
| /// <param name="cancellationToken">The cancellation token for building the Machine</param> | ||
| /// <returns>A new mutex</returns> | ||
| public static Task<IREstateMutex> CreateLockAsync( | ||
| this IAgent restateAgent, | ||
| int? retryLimit = null, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| return CreateMutexAsync(restateAgent, retryLimit, cancellationToken); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Creates a new mutex (lock) using Machine state as the synchronization root. | ||
| /// </summary> | ||
| /// <param name="restateAgent">The agent to use to build the mutex</param> | ||
| /// <param name="retryLimit">The number of attempts to try obtaining a lock; if null then no limit</param> | ||
| /// <param name="cancellationToken">The cancellation token for building the Machine</param> | ||
| /// <returns>A new mutex</returns> | ||
| public static async Task<IREstateMutex> CreateMutexAsync( | ||
| this IAgent restateAgent, | ||
| int? retryLimit = null, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| var mutexSchematic = CreateSemaphoreSchematic(restateAgent, 1, retryLimit); | ||
|
|
||
| var mutexMachine = await restateAgent.GetStateEngine<int, int>() | ||
| .CreateMachineAsync(mutexSchematic, null, cancellationToken); | ||
|
|
||
| return new REstateSemaphore(mutexMachine); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Creates a new semaphoroe using Machine state as the synchronization root. | ||
| /// </summary> | ||
| /// <param name="restateAgent">The agent to use to build the semaphore</param> | ||
| /// <param name="slots">The number or available slots in the sempahore</param> | ||
| /// <param name="retryLimit">The number of attempts to try obtaining a slot; if null then no limit</param> | ||
| /// <param name="cancellationToken">The cancellation token for building the Machine</param> | ||
| /// <returns>A new semaphore</returns> | ||
| public static async Task<IREstateSemaphore> CreateSemaphoreAsync( | ||
| this IAgent restateAgent, | ||
| int slots, | ||
| int? retryLimit = null, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| if (slots < 1) throw new ArgumentOutOfRangeException(nameof(slots), "Sempahores must have at least one slot."); | ||
|
|
||
| var sempahoreSchematic = CreateSemaphoreSchematic(restateAgent, slots, retryLimit); | ||
|
|
||
| var sempahoreMachine = await restateAgent.GetStateEngine<int, int>() | ||
| .CreateMachineAsync(sempahoreSchematic, null, cancellationToken); | ||
|
|
||
| return new REstateSemaphore(sempahoreMachine); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Creates a Schematic that represents a semaphore with n slots | ||
| /// </summary> | ||
| /// <remarks> | ||
| /// The following is the Schematic in DOT Graph | ||
| /// <![CDATA[ | ||
| /// digraph { | ||
| /// rankdir="LR" | ||
| /// "0" -> "1" [label= " 1 "]; | ||
| /// "1" -> "0" [label=" -1 "]; | ||
| /// "1" -> "2" [label= " 1 "]; | ||
| /// "2" -> "1" [label=" -1 "]; | ||
| /// "2" -> "3" [label= " 1 "]; | ||
| /// "3" -> "2" [label=" -1 "]; | ||
| /// } | ||
| /// ]]> | ||
| /// <image url="$(SolutionDir)\src\REstate.Concurrency.Primitives\diagram_white.png" /> | ||
| /// </remarks> | ||
| private static Schematic<int, int> CreateSemaphoreSchematic(IAgent restateAgent, int slots, int? retryLimit) | ||
| { | ||
| if (slots < 1) throw new ArgumentOutOfRangeException(nameof(slots), "Sempahores must have at least one slot."); | ||
|
|
||
| var mutexBuilder = restateAgent | ||
| .CreateSchematic<int, int>("REstateMutex"); | ||
|
|
||
| if (retryLimit == null) | ||
| { | ||
| mutexBuilder.WithStateConflictRetries(); | ||
| } | ||
| else if (retryLimit.Value > 0) | ||
| { | ||
| mutexBuilder.WithStateConflictRetries(retryLimit.Value); | ||
| } | ||
| else | ||
| { | ||
| // No retries, so just don't set it. | ||
| } | ||
|
|
||
| mutexBuilder | ||
| .WithState(0, state => state | ||
| .AsInitialState() | ||
| .WithReentrance(-1)) | ||
| .WithState(1, state => state | ||
| .WithTransitionFrom(0, 1) | ||
| .WithTransitionTo(0, -1)); | ||
|
|
||
| if (slots == 1) | ||
| return mutexBuilder.Build(); | ||
|
|
||
| var sempahoreBuilder = mutexBuilder; | ||
|
|
||
| foreach (var slot in Enumerable.Range(start: 2, count: slots - 1)) | ||
| { | ||
| sempahoreBuilder.WithState(slot, state => state | ||
| .WithTransitionFrom(slot - 1, 1) | ||
| .WithTransitionTo(slot - 1, -1)); | ||
| } | ||
|
|
||
| var semaphore = sempahoreBuilder.Build(); | ||
|
|
||
| semaphore.SchematicName = $"REstateSemaphoreOf{slots}"; | ||
|
|
||
| return semaphore; | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| using System; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace REstate.Concurrency.Primitives | ||
| { | ||
| public interface IREstateMutex | ||
| { | ||
| Task<IDisposable> EnterAsync(CancellationToken cancellationToken = default); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| namespace REstate.Concurrency.Primitives | ||
| { | ||
| public interface IREstateSemaphore | ||
| : IREstateMutex | ||
| { | ||
| } | ||
| } |
24 changes: 24 additions & 0 deletions
24
src/REstate.Concurrency.Primitives/REstate.Concurrency.Primitives.csproj
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <TargetFrameworks>netstandard1.3;net45</TargetFrameworks> | ||
| <Authors>Ovan Crone</Authors> | ||
| <Company>Psibernetic Solutions</Company> | ||
| <Description>Concurrency primitives such as Semaphores and mutexes build on REstate Machines.</Description> | ||
| <Copyright>Ovan Crone 2016</Copyright> | ||
| <PackageLicenseUrl>https://opensource.org/licenses/MIT</PackageLicenseUrl> | ||
| <Version>3.0.0</Version> | ||
| <AssemblyVersion>3.0.0.0</AssemblyVersion> | ||
| <FileVersion>3.0.0.0</FileVersion> | ||
| <LangVersion>latest</LangVersion> | ||
| <PackageProjectUrl>https://github.com/psibr/REstate.Engine</PackageProjectUrl> | ||
| <RepositoryUrl>https://github.com/psibr/REstate.Engine</RepositoryUrl> | ||
| <RepositoryType>git</RepositoryType> | ||
| <PackageTags>REstate</PackageTags> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\REstate\REstate.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| using System; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using REstate.Engine; | ||
|
|
||
| namespace REstate.Concurrency.Primitives | ||
| { | ||
| public class REstateSemaphore | ||
| : IREstateSemaphore | ||
| { | ||
| public REstateSemaphore(IStateMachine<int, int> semaphoreMachine) | ||
| { | ||
| SemaphoreMachine = semaphoreMachine; | ||
| } | ||
|
|
||
| private IStateMachine<int, int> SemaphoreMachine { get; } | ||
|
|
||
| public async Task<IDisposable> EnterAsync(CancellationToken cancellationToken = default) | ||
| { | ||
| var criticalSection = new CriticalSection(SemaphoreMachine, cancellationToken); | ||
|
|
||
| await criticalSection.EnterAsync(); | ||
|
|
||
| return criticalSection; | ||
| } | ||
|
|
||
| private class CriticalSection | ||
| : IDisposable | ||
| { | ||
| public CriticalSection( | ||
| IStateMachine<int, int> sempahoreStateMachine, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| SempahoreStateMachine = sempahoreStateMachine; | ||
| CancellationToken = cancellationToken; | ||
| } | ||
|
|
||
| public async Task EnterAsync() | ||
| { | ||
| await SempahoreStateMachine.SendAsync(1, CancellationToken); | ||
| } | ||
|
|
||
| private IStateMachine<int, int> SempahoreStateMachine { get; } | ||
| private CancellationToken CancellationToken { get; } | ||
|
|
||
| public void Dispose() | ||
| { | ||
| var mre = new ManualResetEventSlim(); | ||
|
|
||
| SempahoreStateMachine.SendAsync(-1, CancellationToken).GetAwaiter().OnCompleted(() => mre.Set()); | ||
|
|
||
| mre.Wait(CancellationToken.None); | ||
| } | ||
| } | ||
| } | ||
| } |
File renamed without changes
File renamed without changes
File renamed without changes
File renamed without changes