Skip to content

Commit 22ef5a4

Browse files
mparrymparry
authored
Merge pull request #35 from rars/fix/issue-34
fix: do not leave completed ambient transaction if transaction scope is altered in enqueued or startup work
2 parents 62230a4 + 176c5e4 commit 22ef5a4

6 files changed

Lines changed: 177 additions & 12 deletions

File tree

‎Winton.Extensions.Threading.Actor.Tests.Unit/ActorTests.cs‎

Lines changed: 104 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
using System.Linq;
77
using System.Threading;
88
using System.Threading.Tasks;
9+
using System.Transactions;
910
using FluentAssertions;
1011
using Winton.Extensions.Threading.Actor.Tests.Utilities;
1112
using Xunit;
@@ -731,6 +732,108 @@ public void ShouldOnlyBeAbleToSpecifyStartWorkOnce()
731732
action.Should().Throw<InvalidOperationException>().WithMessage("Start work already specified.");
732733
}
733734

735+
[Fact]
736+
public async Task ShouldHaveNoAmbientTransactionInStartUpWork()
737+
{
738+
var attempts = 0;
739+
740+
var actor = CreateActor(ActorCreateOptions.None)
741+
.WithStartWork(async () =>
742+
{
743+
using (var scope = new TransactionScope(TransactionScopeOption.Suppress, TransactionScopeAsyncFlowOption.Enabled))
744+
{
745+
await Task.Delay(200);
746+
} // Restores ambient transaction scope which should be null
747+
748+
if (Transaction.Current == null)
749+
{
750+
Interlocked.Increment(ref attempts);
751+
}
752+
});
753+
754+
await actor.Start();
755+
756+
attempts.Should().Be(1);
757+
}
758+
759+
[Fact]
760+
public async Task ShouldHaveNoAmbientTransactionInEnqueuedWork()
761+
{
762+
var attempts = 0;
763+
764+
var actor = CreateActor(ActorCreateOptions.None);
765+
await actor.Start();
766+
767+
await actor.Enqueue(async () =>
768+
{
769+
using (var scope = new TransactionScope(TransactionScopeOption.Suppress, TransactionScopeAsyncFlowOption.Enabled))
770+
{
771+
await Task.Delay(200);
772+
} // Restores ambient transaction scope which should be null
773+
774+
if (Transaction.Current == null)
775+
{
776+
Interlocked.Increment(ref attempts);
777+
}
778+
});
779+
780+
attempts.Should().Be(1);
781+
}
782+
783+
[Fact]
784+
public async Task ShouldHaveNoAmbientTransactionInEnqueuedWorkThatReturnsData()
785+
{
786+
var attempts = 0;
787+
788+
var actor = CreateActor(ActorCreateOptions.None);
789+
await actor.Start();
790+
791+
var result = await actor.Enqueue(async () =>
792+
{
793+
using (var scope = new TransactionScope(TransactionScopeOption.Suppress, TransactionScopeAsyncFlowOption.Enabled))
794+
{
795+
await Task.Delay(200);
796+
} // Restores ambient transaction scope which should be null
797+
798+
if (Transaction.Current == null)
799+
{
800+
Interlocked.Increment(ref attempts);
801+
}
802+
803+
return attempts;
804+
});
805+
806+
attempts.Should().Be(1);
807+
result.Should().Be(1);
808+
}
809+
810+
[Fact]
811+
public async Task ShouldHaveNoAmbientTransactionInEnqueuedWorkThatIsSynchronousAndReturnsData()
812+
{
813+
var attempts = 0;
814+
815+
var actor = CreateActor(ActorCreateOptions.None);
816+
await actor.Start();
817+
818+
var result = await actor.Enqueue(() =>
819+
{
820+
using (var scope = new TransactionScope(TransactionScopeOption.Suppress))
821+
{
822+
Thread.Sleep(200);
823+
} // Restores ambient transaction scope which should be null
824+
825+
if (Transaction.Current == null)
826+
{
827+
Interlocked.Increment(ref attempts);
828+
}
829+
830+
return attempts;
831+
});
832+
833+
attempts.Should().Be(1);
834+
result.Should().Be(1);
835+
}
836+
734837
[Fact]
735838
public void ShouldOnlyBeAbleToSpecifyStopWorkOnce()
736839
{
@@ -971,7 +1074,7 @@ public async Task ShouldBeAbleToPauseActorUntilResumeFromAwaitReturningData(Acto
9711074

9721075
var tasks = new Task[4];
9731076

974-
tasks[0] =
1077+
tasks[0] =
9751078
actor.Enqueue(
9761079
async () =>
9771080
{

‎Winton.Extensions.Threading.Actor.Tests.Unit/Internal/ActorTaskSchedulerTests.cs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ internal async Task ShouldBeAbleToResumeInitiallyPausedScheduler(ActorTaskTraits
3939
var count = 0;
4040
var task1 = _actorTaskFactory.StartNew(() => ++count, CancellationToken.None, TaskCreationOptions.None, ActorTaskTraits.None);
4141
var task2 = _actorTaskFactory.StartNew(() => ++count, CancellationToken.None, TaskCreationOptions.None, ActorTaskTraits.None);
42-
42+
4343
Task.WhenAny(task1, task2).Wait(TimeSpan.FromSeconds(1)).Should().BeFalse("tasks should not have been executed if the scheduler is paused");
4444

4545
var resumer = _actorTaskFactory.StartNew(() => ++count, CancellationToken.None, TaskCreationOptions.None, resumingTaskTraits);
@@ -54,14 +54,14 @@ public void FailureOfCriticalTaskShouldTerminateSchedulerWithSubsequentTasksBein
5454
{
5555
UnpauseScheduler();
5656

57-
var task1 = _actorTaskFactory.StartNew(() => throw new Exception("oh dear"), CancellationToken.None, TaskCreationOptions.None, ActorTaskTraits.None);
57+
var task1 = _actorTaskFactory.StartNew(new Action(() => throw new Exception("oh dear")), CancellationToken.None, TaskCreationOptions.None, ActorTaskTraits.None);
5858

5959
task1.Awaiting(x => x).Should().Throw<Exception>().WithMessage("oh dear");
6060

6161
_scheduler.TerminatedTask.Wait(TimeSpan.FromMilliseconds(250)).Should().BeFalse();
6262

63-
var task2 = _actorTaskFactory.StartNew(() => throw new Exception("no!!!"), CancellationToken.None, TaskCreationOptions.None, ActorTaskTraits.Critical);
64-
var task3 = _actorTaskFactory.StartNew(() => throw new Exception("shouldn't hit this"), CancellationToken.None, TaskCreationOptions.None, ActorTaskTraits.None);
63+
var task2 = _actorTaskFactory.StartNew(new Action(() => throw new Exception("no!!!")), CancellationToken.None, TaskCreationOptions.None, ActorTaskTraits.Critical);
64+
var task3 = _actorTaskFactory.StartNew(new Action(() => throw new Exception("shouldn't hit this")), CancellationToken.None, TaskCreationOptions.None, ActorTaskTraits.None);
6565

6666
task2.Awaiting(x => x).Should().Throw<Exception>().WithMessage("no!!!");
6767

@@ -97,7 +97,7 @@ void TerminalWork()
9797

9898
var task1 = _actorTaskFactory.StartNew(TerminalWork, CancellationToken.None, TaskCreationOptions.None, ActorTaskTraits.Terminal);
9999
var task2 = _actorTaskFactory.StartNew(() => throw new Exception("shouldn't hit this"), CancellationToken.None, TaskCreationOptions.None, ActorTaskTraits.None);
100-
100+
101101
switch (terminalWorkOutcomeType)
102102
{
103103
case TaskStatus.Canceled:
@@ -269,7 +269,7 @@ public void ForNonLongRunningTasksShouldUseActiveThreadIfNotFromThreadPool()
269269
task3ThreadId = Thread.CurrentThread.ManagedThreadId;
270270
ActorThreadAssertions.CurrentThreadShouldNotBeThreadPoolThread();
271271
}, CancellationToken.None, TaskCreationOptions.None);
272-
272+
273273
barrier.SetResult(true);
274274

275275
ThrowIfWaitTimesOut(task1);

‎Winton.Extensions.Threading.Actor.Tests.Unit/Winton.Extensions.Threading.Actor.Tests.Unit.csproj‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
<Project Sdk="Microsoft.NET.Sdk">
22

3-
<PropertyGroup>
3+
<PropertyGroup>
44
<TargetFrameworks>net5.0</TargetFrameworks>
5-
<SignAssembly>True</SignAssembly>
6-
<DelaySign>False</DelaySign>
5+
<SignAssembly>True</SignAssembly>
6+
<DelaySign>False</DelaySign>
77
<AssemblyOriginatorKeyFile>../keys/Winton.Extensions.Threading.Actor.snk</AssemblyOriginatorKeyFile>
88
<PublicSign Condition=" '$(OS)' != 'Windows_NT' ">true</PublicSign>
99
<LangVersion>latest</LangVersion>

‎Winton.Extensions.Threading.Actor/Actor.cs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -212,13 +212,15 @@ public Task Stop()
212212
private async Task Enqueue(Func<Task> asyncAction, CancellationToken cancellationToken, ActorEnqueueOptions options, ActorTaskTraits taskTraits)
213213
{
214214
var task = _actorTaskFactory.StartNew(asyncAction, cancellationToken, ActorTaskOptions.GetTaskCreationOptions(options), taskTraits);
215-
await await task.ConfigureAwait(false);
215+
var nestedTask = await task.ConfigureAwait(false);
216+
await nestedTask.ConfigureAwait(false);
216217
}
217218

218219
private async Task<T> Enqueue<T>(Func<Task<T>> asyncFunction, CancellationToken cancellationToken, ActorEnqueueOptions options, ActorTaskTraits taskTraits)
219220
{
220221
var task = _actorTaskFactory.StartNew(asyncFunction, cancellationToken, ActorTaskOptions.GetTaskCreationOptions(options), taskTraits);
221-
return await await task.ConfigureAwait(false);
222+
var nestedTask = await task.ConfigureAwait(false);
223+
return await nestedTask.ConfigureAwait(false);
222224
}
223225

224226
private enum StateName

‎Winton.Extensions.Threading.Actor/ActorExtensions.cs‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -186,6 +186,46 @@ public static CancellationToken StoppedToken(this IActor self)
186186
return cancellationTokenSource.Token;
187187
}
188188

189+
internal static Func<Task<T>> SuppressTransactionScopeWrapper<T>(Func<Task<T>> function)
190+
{
191+
return async () =>
192+
{
193+
using (var scope = new TransactionScope(
194+
TransactionScopeOption.Suppress,
195+
TransactionScopeAsyncFlowOption.Enabled))
196+
{
197+
try
198+
{
199+
return await function().ConfigureAwait(false);
200+
}
201+
finally
202+
{
203+
scope.Complete();
204+
}
205+
}
206+
};
207+
}
208+
209+
internal static Func<Task> SuppressTransactionScopeWrapper(Func<Task> function)
210+
{
211+
return async () =>
212+
{
213+
using (var scope = new TransactionScope(
214+
TransactionScopeOption.Suppress,
215+
TransactionScopeAsyncFlowOption.Enabled))
216+
{
217+
try
218+
{
219+
await function().ConfigureAwait(false);
220+
}
221+
finally
222+
{
223+
scope.Complete();
224+
}
225+
}
226+
};
227+
}
228+
189229
internal static Func<T> SuppressTransactionScopeWrapper<T>(Func<T> function)
190230
{
191231
return () =>

‎Winton.Extensions.Threading.Actor/Internal/ActorTaskFactory.cs‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,26 @@ public Task StartNew(Action action, CancellationToken cancellationToken, TaskCre
2020
return Task.Factory.StartNew(ActorExtensions.SuppressTransactionScopeWrapper(action), cancellationTokenSource.Token, taskCreationOptions | TaskCreationOptions.HideScheduler, _scheduler);
2121
}
2222

23+
public Task<Task> StartNew(Func<Task> asyncFunction, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions, ActorTaskTraits traits = ActorTaskTraits.None)
24+
{
25+
var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
26+
27+
ActorTask.CurrentActorTaskTraits = traits;
28+
ActorTask.CurrentCanceller = cancellationTokenSource;
29+
30+
return Task.Factory.StartNew(ActorExtensions.SuppressTransactionScopeWrapper(asyncFunction), cancellationTokenSource.Token, taskCreationOptions | TaskCreationOptions.HideScheduler, _scheduler);
31+
}
32+
33+
public Task<Task<T>> StartNew<T>(Func<Task<T>> asyncFunction, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions, ActorTaskTraits traits = ActorTaskTraits.None)
34+
{
35+
var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
36+
37+
ActorTask.CurrentActorTaskTraits = traits;
38+
ActorTask.CurrentCanceller = cancellationTokenSource;
39+
40+
return Task.Factory.StartNew(ActorExtensions.SuppressTransactionScopeWrapper(asyncFunction), cancellationTokenSource.Token, taskCreationOptions | TaskCreationOptions.HideScheduler, _scheduler);
41+
}
42+
2343
public Task<T> StartNew<T>(Func<T> function, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions, ActorTaskTraits traits = ActorTaskTraits.None)
2444
{
2545
var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);

0 commit comments

Comments
 (0)