Add ability to pause an actor whilst awaiting an off-actor task.

Jos Hickson · Jos Hickson · commit a7a1cd7200da · 2017-06-02T16:32:37.000+01:00
diff --git a/Winton.Extensions.Threading.Actor.Tests.Unit/ActorTests.cs b/Winton.Extensions.Threading.Actor.Tests.Unit/ActorTests.cs
@@ -875,6 +875,76 @@ public void ShouldStopActorAndNotProcessAnyAlreadyEnqueuedWorkIfStartWorkCancell
             attempts.Should().Be(0);
         }
 
+        [Fact]
+        public async Task ShouldBeAbleToPauseActorUntilResumeFromAwait()
+        {
+            var actor = CreateActor();
+            var numbers = new List<int>();
+
+            await actor.Start();
+
+            var tasks =
+                new[]
+                {
+                    actor.Enqueue(
+                        async () =>
+                        {
+                            numbers.Add(1);
+
+                            await Task.Delay(TimeSpan.FromSeconds(1)).WhileActorPaused();
+
+                            numbers.Add(2);
+                        }),
+                    actor.Enqueue(() => numbers.Add(3)),
+                    actor.Enqueue(() => numbers.Add(4)),
+                    actor.Enqueue(() => numbers.Add(5))
+                };
+
+            await Task.WhenAll(tasks);
+
+            numbers.Should().Equal(1, 2, 3, 4, 5);
+        }
+
+        [Fact]
+        public async Task ShouldBeAbleToPauseActorUntilResumeFromAwaitReturningData()
+        {
+            var actor = CreateActor();
+            var numbers = new List<int>();
+            var promise = new TaskCompletionSource<int>();
+
+            await actor.Start();
+
+            var tasks =
+                new[]
+                {
+                    actor.Enqueue(
+                        async () =>
+                        {
+                            numbers.Add(1);
+
+                            var next = await promise.Task.WhileActorPaused();
+
+                            numbers.Add(next);
+                        }),
+                    actor.Enqueue(() => numbers.Add(3)),
+                    actor.Enqueue(() => numbers.Add(4)),
+                    actor.Enqueue(() => numbers.Add(5))
+                };
+
+            var promiseSetter =
+                Task.Run(
+                    async () =>
+                    {
+                        await Task.Delay(TimeSpan.FromSeconds(1));
+                        promise.SetResult(2);
+                    });
+
+            await Task.WhenAll(tasks);
+            await promiseSetter;
+
+            numbers.Should().Equal(1, 2, 3, 4, 5);
+        }
+
         [Flags]
         private enum ActorCreateOptions
         {
diff --git a/Winton.Extensions.Threading.Actor.Tests.Unit/Winton.Extensions.Threading.Actor.Tests.Unit.csproj b/Winton.Extensions.Threading.Actor.Tests.Unit/Winton.Extensions.Threading.Actor.Tests.Unit.csproj
@@ -11,7 +11,7 @@
   <ItemGroup>
     <PackageReference Include="FluentAssertions" Version="4.19.2" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.0.0" />
-    <PackageReference Include="Moq" Version="4.7.1" />
+    <PackageReference Include="Moq" Version="4.7.12" />
     <PackageReference Include="xunit" Version="2.2.0" />
     <PackageReference Include="xunit.runner.visualstudio" Version="2.2.0" />
   </ItemGroup>
diff --git a/Winton.Extensions.Threading.Actor/Actor.cs b/Winton.Extensions.Threading.Actor/Actor.cs
@@ -46,8 +46,8 @@ private Actor(ActorImpl impl)
         /// </summary>
         public static ActorId CurrentId
         {
-            get { return _currentId; }
-            internal set { _currentId = value; }
+            get => _currentId;
+            internal set => _currentId = value;
         }
 
         /// <inheritdoc cref="IActor.Id"/>
@@ -56,13 +56,13 @@ public static ActorId CurrentId
         /// <inheritdoc cref="IActor.StartWork"/>
         public ActorStartWork StartWork
         {
-            set { _impl.StartWork = value; }
+            set => _impl.StartWork = value;
         }
 
         /// <inheritdoc cref="IActor.StopWork"/>
         public ActorStopWork StopWork
         {
-            set { _impl.StopWork = value; }
+            set => _impl.StopWork = value;
         }
 
         /// <inheritdoc cref="IActor.Start"/>
diff --git a/Winton.Extensions.Threading.Actor/Internal/ActorImpl.cs b/Winton.Extensions.Threading.Actor/Internal/ActorImpl.cs
@@ -16,12 +16,12 @@ public ActorImpl(IActorTaskFactory actorTaskFactory)
 
         public ActorStartWork StartWork
         {
-            set { Context.StartWork = value; }
+            set => Context.StartWork = value;
         }
 
         public ActorStopWork StopWork
         {
-            set { Context.StopWork = value; }
+            set => Context.StopWork = value;
         }
 
         public Task Start()
diff --git a/Winton.Extensions.Threading.Actor/Internal/ActorSynchronizationContext.cs b/Winton.Extensions.Threading.Actor/Internal/ActorSynchronizationContext.cs
@@ -7,16 +7,23 @@ internal sealed class ActorSynchronizationContext : SynchronizationContext
     {
         private readonly ActorTaskScheduler _scheduler;
         private readonly IActorTaskFactory _actorTaskFactory;
+        private readonly ActorTaskKind _actorTaskKind;
 
-        public ActorSynchronizationContext(ActorTaskScheduler scheduler, IActorTaskFactory actorTaskFactory)
+        public ActorSynchronizationContext(ActorTaskScheduler scheduler, IActorTaskFactory actorTaskFactory, ActorTaskKind actorTaskKind)
         {
             _scheduler = scheduler;
             _actorTaskFactory = actorTaskFactory;
+            _actorTaskKind = actorTaskKind;
+        }
+
+        public ActorSynchronizationContext ChangeActorTaskKind(ActorTaskKind actorTaskKind)
+        {
+            return new ActorSynchronizationContext(_scheduler, _actorTaskFactory, actorTaskKind);
         }
 
         public override void Post(SendOrPostCallback callback, object state)
         {
-            _actorTaskFactory.Create(() => callback(state), CancellationToken.None, TaskCreationOptions.HideScheduler).Start(_scheduler);
+            _actorTaskFactory.Create(() => callback(state), CancellationToken.None, TaskCreationOptions.HideScheduler, _actorTaskKind).Start(_scheduler);
         }
     }
 }
diff --git a/Winton.Extensions.Threading.Actor/Internal/ActorTask.cs b/Winton.Extensions.Threading.Actor/Internal/ActorTask.cs
@@ -1,3 +1,4 @@
+using System;
 using System.Threading.Tasks;
 
 namespace Winton.Extensions.Threading.Actor.Internal
@@ -9,6 +10,11 @@ public static bool IsActorTask(this Task self)
             return self.AsyncState is ActorTaskContext;
         }
 
+        public static ActorTaskContext GetActorTaskContext(this Task self)
+        {
+            return self.AsyncState as ActorTaskContext ?? throw new InvalidOperationException("Task is not an actor task.");
+        }
+
         public static void Cancel(this Task self)
         {
             var context = self.AsyncState as ActorTaskContext;
diff --git a/Winton.Extensions.Threading.Actor/Internal/ActorTaskContext.cs b/Winton.Extensions.Threading.Actor/Internal/ActorTaskContext.cs
@@ -4,11 +4,14 @@ namespace Winton.Extensions.Threading.Actor.Internal
 {
     internal sealed class ActorTaskContext
     {
-        public ActorTaskContext(CancellationToken cancellationToken)
+        public ActorTaskContext(CancellationToken cancellationToken, ActorTaskKind kind)
         {
+            Kind = kind;
             Canceller = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
         }
 
         public CancellationTokenSource Canceller { get; }
+
+        public ActorTaskKind Kind { get; }
     }
 }
diff --git a/Winton.Extensions.Threading.Actor/Internal/ActorTaskFactoryExtensions.cs b/Winton.Extensions.Threading.Actor/Internal/ActorTaskFactoryExtensions.cs
@@ -8,9 +8,9 @@ internal static class ActorTaskFactoryExtensions
     {
         private const TaskCreationOptions BaseCreationOptions = TaskCreationOptions.HideScheduler;
 
-        public static Task Create(this IActorTaskFactory self, Action work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions)
+        public static Task Create(this IActorTaskFactory self, Action work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions, ActorTaskKind kind = ActorTaskKind.Standard)
         {
-            var actorTaskState = new ActorTaskContext(cancellationToken);
+            var actorTaskState = new ActorTaskContext(cancellationToken, kind);
             return self.Create(state =>
                                {
                                    try
@@ -25,9 +25,9 @@ public static Task Create(this IActorTaskFactory self, Action work, Cancellation
                                actorTaskState.Canceller.Token, taskCreationOptions | BaseCreationOptions, actorTaskState);
         }
 
-        public static Task<T> Create<T>(this IActorTaskFactory self, Func<T> work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions)
+        public static Task<T> Create<T>(this IActorTaskFactory self, Func<T> work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions, ActorTaskKind kind = ActorTaskKind.Standard)
         {
-            var actorTaskState = new ActorTaskContext(cancellationToken);
+            var actorTaskState = new ActorTaskContext(cancellationToken, kind);
             return self.Create(state =>
                                {
                                    try
diff --git a/Winton.Extensions.Threading.Actor/Internal/ActorTaskKind.cs b/Winton.Extensions.Threading.Actor/Internal/ActorTaskKind.cs
@@ -0,0 +1,8 @@
+﻿namespace Winton.Extensions.Threading.Actor.Internal
+{
+    internal enum ActorTaskKind
+    {
+        Standard,
+        Resumer
+    }
+}
diff --git a/Winton.Extensions.Threading.Actor/Internal/ActorTaskScheduler.cs b/Winton.Extensions.Threading.Actor/Internal/ActorTaskScheduler.cs
diff --git a/Winton.Extensions.Threading.Actor/Internal/ActorTaskSchedulerStatus.cs b/Winton.Extensions.Threading.Actor/Internal/ActorTaskSchedulerStatus.cs
diff --git a/Winton.Extensions.Threading.Actor/Internal/ActorTaskSchedulerStatusManager.cs b/Winton.Extensions.Threading.Actor/Internal/ActorTaskSchedulerStatusManager.cs
diff --git a/Winton.Extensions.Threading.Actor/TaskExtensions.cs b/Winton.Extensions.Threading.Actor/TaskExtensions.cs

Original file line number	Diff line number	Diff line change
`@@ -46,8 +46,8 @@ private Actor(ActorImpl impl)`
`46`	`46`	`/// </summary>`
`47`	`47`	`public static ActorId CurrentId`
`48`	`48`	`{`
`49`		`- get { return _currentId; }`
`50`		`- internal set { _currentId = value; }`
	`49`	`+ get => _currentId;`
	`50`	`+ internal set => _currentId = value;`
`51`	`51`	`}`
`52`	`52`
`53`	`53`	`/// <inheritdoc cref="IActor.Id"/>`
`@@ -56,13 +56,13 @@ public static ActorId CurrentId`
`56`	`56`	`/// <inheritdoc cref="IActor.StartWork"/>`
`57`	`57`	`public ActorStartWork StartWork`
`58`	`58`	`{`
`59`		`- set { _impl.StartWork = value; }`
	`59`	`+ set => _impl.StartWork = value;`
`60`	`60`	`}`
`61`	`61`
`62`	`62`	`/// <inheritdoc cref="IActor.StopWork"/>`
`63`	`63`	`public ActorStopWork StopWork`
`64`	`64`	`{`
`65`		`- set { _impl.StopWork = value; }`
	`65`	`+ set => _impl.StopWork = value;`
`66`	`66`	`}`
`67`	`67`
`68`	`68`	`/// <inheritdoc cref="IActor.Start"/>`
Original file line number	Diff line number	Diff line change
`@@ -16,12 +16,12 @@ public ActorImpl(IActorTaskFactory actorTaskFactory)`
`16`	`16`
`17`	`17`	`public ActorStartWork StartWork`
`18`	`18`	`{`
`19`		`- set { Context.StartWork = value; }`
	`19`	`+ set => Context.StartWork = value;`
`20`	`20`	`}`
`21`	`21`
`22`	`22`	`public ActorStopWork StopWork`
`23`	`23`	`{`
`24`		`- set { Context.StopWork = value; }`
	`24`	`+ set => Context.StopWork = value;`
`25`	`25`	`}`
`26`	`26`
`27`	`27`	`public Task Start()`
Original file line number	Diff line number	Diff line change
`@@ -7,16 +7,23 @@ internal sealed class ActorSynchronizationContext : SynchronizationContext`
`7`	`7`	`{`
`8`	`8`	`private readonly ActorTaskScheduler _scheduler;`
`9`	`9`	`private readonly IActorTaskFactory _actorTaskFactory;`
	`10`	`+ private readonly ActorTaskKind _actorTaskKind;`
`10`	`11`
`11`		`- public ActorSynchronizationContext(ActorTaskScheduler scheduler, IActorTaskFactory actorTaskFactory)`
	`12`	`+ public ActorSynchronizationContext(ActorTaskScheduler scheduler, IActorTaskFactory actorTaskFactory, ActorTaskKind actorTaskKind)`
`12`	`13`	`{`
`13`	`14`	`_scheduler = scheduler;`
`14`	`15`	`_actorTaskFactory = actorTaskFactory;`
	`16`	`+ _actorTaskKind = actorTaskKind;`
	`17`	`+ }`
	`18`	`+`
	`19`	`+ public ActorSynchronizationContext ChangeActorTaskKind(ActorTaskKind actorTaskKind)`
	`20`	`+ {`
	`21`	`+ return new ActorSynchronizationContext(_scheduler, _actorTaskFactory, actorTaskKind);`
`15`	`22`	`}`
`16`	`23`
`17`	`24`	`public override void Post(SendOrPostCallback callback, object state)`
`18`	`25`	`{`
`19`		`- _actorTaskFactory.Create(() => callback(state), CancellationToken.None, TaskCreationOptions.HideScheduler).Start(_scheduler);`
	`26`	`+ _actorTaskFactory.Create(() => callback(state), CancellationToken.None, TaskCreationOptions.HideScheduler, _actorTaskKind).Start(_scheduler);`
`20`	`27`	`}`
`21`	`28`	`}`
`22`	`29`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,4 @@`
	`1`	`+using System;`
`1`	`2`	`using System.Threading.Tasks;`
`2`	`3`
`3`	`4`	`namespace Winton.Extensions.Threading.Actor.Internal`
`@@ -9,6 +10,11 @@ public static bool IsActorTask(this Task self)`
`9`	`10`	`return self.AsyncState is ActorTaskContext;`
`10`	`11`	`}`
`11`	`12`
	`13`	`+ public static ActorTaskContext GetActorTaskContext(this Task self)`
	`14`	`+ {`
	`15`	`+ return self.AsyncState as ActorTaskContext ?? throw new InvalidOperationException("Task is not an actor task.");`
	`16`	`+ }`
	`17`	`+`
`12`	`18`	`public static void Cancel(this Task self)`
`13`	`19`	`{`
`14`	`20`	`var context = self.AsyncState as ActorTaskContext;`
Original file line number	Diff line number	Diff line change
`@@ -4,11 +4,14 @@ namespace Winton.Extensions.Threading.Actor.Internal`
`4`	`4`	`{`
`5`	`5`	`internal sealed class ActorTaskContext`
`6`	`6`	`{`
`7`		`- public ActorTaskContext(CancellationToken cancellationToken)`
	`7`	`+ public ActorTaskContext(CancellationToken cancellationToken, ActorTaskKind kind)`
`8`	`8`	`{`
	`9`	`+ Kind = kind;`
`9`	`10`	`Canceller = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);`
`10`	`11`	`}`
`11`	`12`
`12`	`13`	`public CancellationTokenSource Canceller { get; }`
	`14`	`+`
	`15`	`+ public ActorTaskKind Kind { get; }`
`13`	`16`	`}`
`14`	`17`	`}`
Original file line number	Diff line number	Diff line change
`@@ -8,9 +8,9 @@ internal static class ActorTaskFactoryExtensions`
`8`	`8`	`{`
`9`	`9`	`private const TaskCreationOptions BaseCreationOptions = TaskCreationOptions.HideScheduler;`
`10`	`10`
`11`		`- public static Task Create(this IActorTaskFactory self, Action work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions)`
	`11`	`+ public static Task Create(this IActorTaskFactory self, Action work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions, ActorTaskKind kind = ActorTaskKind.Standard)`
`12`	`12`	`{`
`13`		`- var actorTaskState = new ActorTaskContext(cancellationToken);`
	`13`	`+ var actorTaskState = new ActorTaskContext(cancellationToken, kind);`
`14`	`14`	`return self.Create(state =>`
`15`	`15`	`{`
`16`	`16`	`try`
`@@ -25,9 +25,9 @@ public static Task Create(this IActorTaskFactory self, Action work, Cancellation`
`25`	`25`	`actorTaskState.Canceller.Token, taskCreationOptions \| BaseCreationOptions, actorTaskState);`
`26`	`26`	`}`
`27`	`27`
`28`		`- public static Task<T> Create<T>(this IActorTaskFactory self, Func<T> work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions)`
	`28`	`+ public static Task<T> Create<T>(this IActorTaskFactory self, Func<T> work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions, ActorTaskKind kind = ActorTaskKind.Standard)`
`29`	`29`	`{`
`30`		`- var actorTaskState = new ActorTaskContext(cancellationToken);`
	`30`	`+ var actorTaskState = new ActorTaskContext(cancellationToken, kind);`
`31`	`31`	`return self.Create(state =>`
`32`	`32`	`{`
`33`	`33`	`try`