mirror of
https://github.com/Smaug123/unofficial-nunit-runner
synced 2025-10-05 09:28:40 +00:00
Permit async ParallelQueue (#279)
This commit is contained in:
Patrick Stevens
GitHub
@@ -4,16 +4,16 @@ open System
|
||||
open System.Threading
|
||||
open System.Threading.Tasks
|
||||
|
||||
type private ThunkEvaluator<'ret> =
|
||||
abstract Eval<'a> : (unit -> 'a) -> AsyncReplyChannel<Result<'a, exn>> -> 'ret
|
||||
type private AsyncThunkEvaluator<'ret> =
|
||||
abstract Eval<'a> : (unit -> Async<'a>) -> AsyncReplyChannel<Result<'a, exn>> -> 'ret
|
||||
|
||||
type private ThunkCrate =
|
||||
abstract Apply<'ret> : ThunkEvaluator<'ret> -> 'ret
|
||||
type private AsyncThunkCrate =
|
||||
abstract Apply<'ret> : AsyncThunkEvaluator<'ret> -> 'ret
|
||||
|
||||
[<RequireQualifiedAccess>]
|
||||
module private ThunkCrate =
|
||||
let make<'a> (t : unit -> 'a) (rc : AsyncReplyChannel<Result<'a, exn>>) : ThunkCrate =
|
||||
{ new ThunkCrate with
|
||||
module private AsyncThunkCrate =
|
||||
let make<'a> (t : unit -> Async<'a>) (rc : AsyncReplyChannel<Result<'a, exn>>) : AsyncThunkCrate =
|
||||
{ new AsyncThunkCrate with
|
||||
member _.Apply e = e.Eval t rc
|
||||
}
|
||||
|
||||
@@ -41,7 +41,11 @@ type private MailboxMessage =
|
||||
| Quit of AsyncReplyChannel<unit>
|
||||
/// Check current state, see if we need to start more tests, etc.
|
||||
| Reconcile
|
||||
| RunTest of within : TestFixture * Parallelizable<unit> option * test : ThunkCrate * context : ExecutionContext
|
||||
| RunTestAsync of
|
||||
within : TestFixture *
|
||||
Parallelizable<unit> option *
|
||||
test : AsyncThunkCrate *
|
||||
context : ExecutionContext
|
||||
| BeginTestFixture of TestFixture * AsyncReplyChannel<TestFixtureRunningToken>
|
||||
| EndTestFixture of TestFixtureTearDownToken * AsyncReplyChannel<unit>
|
||||
|
||||
@@ -310,26 +314,31 @@ type ParallelQueue
|
||||
rc.Reply ()
|
||||
m.Post MailboxMessage.Reconcile
|
||||
return! processTask (Running state) m
|
||||
| MailboxMessage.RunTest (withinFixture, par, message, capturedContext) ->
|
||||
| MailboxMessage.RunTestAsync (withinFixture, par, message, capturedContext) ->
|
||||
let t () =
|
||||
{ new ThunkEvaluator<_> with
|
||||
member _.Eval<'b> (t : unit -> 'b) rc =
|
||||
{ new AsyncThunkEvaluator<_> with
|
||||
member _.Eval<'b> (t : unit -> Async<'b>) rc =
|
||||
let tcs = TaskCompletionSource TaskCreationOptions.RunContinuationsAsynchronously
|
||||
|
||||
fun () ->
|
||||
ExecutionContext.Run (
|
||||
capturedContext,
|
||||
(fun _ ->
|
||||
let result =
|
||||
try
|
||||
let r = t ()
|
||||
Ok r
|
||||
with e ->
|
||||
Error e
|
||||
async {
|
||||
let! result =
|
||||
async {
|
||||
try
|
||||
let! r = t ()
|
||||
return Ok r
|
||||
with e ->
|
||||
return Error e
|
||||
}
|
||||
|
||||
tcs.SetResult ()
|
||||
m.Post MailboxMessage.Reconcile
|
||||
rc.Reply result
|
||||
tcs.SetResult ()
|
||||
m.Post MailboxMessage.Reconcile
|
||||
rc.Reply result
|
||||
}
|
||||
|> Async.StartImmediate
|
||||
),
|
||||
()
|
||||
)
|
||||
@@ -353,19 +362,19 @@ type ParallelQueue
|
||||
let mb = new MailboxProcessor<_> (processTask MailboxState.Idle)
|
||||
do mb.Start ()
|
||||
|
||||
/// Request to run the given action, freely in parallel with other running tests.
|
||||
/// Request to run the given async action, freely in parallel with other running tests.
|
||||
/// The resulting Task will return when the action has completed.
|
||||
member _.Run<'a>
|
||||
member _.RunAsync<'a>
|
||||
(TestFixtureSetupToken parent)
|
||||
(scope : Parallelizable<unit> option)
|
||||
(action : unit -> 'a)
|
||||
(action : unit -> Async<'a>)
|
||||
: 'a Task
|
||||
=
|
||||
let ec = ExecutionContext.Capture ()
|
||||
|
||||
task {
|
||||
let! result =
|
||||
(fun rc -> MailboxMessage.RunTest (parent, scope, ThunkCrate.make action rc, ec))
|
||||
(fun rc -> MailboxMessage.RunTestAsync (parent, scope, AsyncThunkCrate.make action rc, ec))
|
||||
|> mb.PostAndAsyncReply
|
||||
|> Async.StartAsTask
|
||||
|
||||
@@ -374,6 +383,16 @@ type ParallelQueue
|
||||
| Error e -> return Exception.reraiseWithOriginalStackTrace e
|
||||
}
|
||||
|
||||
/// Request to run the given action, freely in parallel with other running tests.
|
||||
/// The resulting Task will return when the action has completed.
|
||||
member this.Run<'a>
|
||||
(parent : TestFixtureSetupToken)
|
||||
(scope : Parallelizable<unit> option)
|
||||
(action : unit -> 'a)
|
||||
: 'a Task
|
||||
=
|
||||
this.RunAsync parent scope (fun () -> async.Return (action ()))
|
||||
|
||||
/// Declare that we wish to start the given test fixture. The resulting Task will return
|
||||
/// when you are allowed to start running tests from that fixture.
|
||||
/// Once you've finished running tests from that fixture, call EndTestFixture.
|
||||
@@ -396,7 +415,14 @@ type ParallelQueue
|
||||
let ec = ExecutionContext.Capture ()
|
||||
|
||||
let! response =
|
||||
(fun rc -> MailboxMessage.RunTest (parent, par, ThunkCrate.make action rc, ec))
|
||||
(fun rc ->
|
||||
MailboxMessage.RunTestAsync (
|
||||
parent,
|
||||
par,
|
||||
AsyncThunkCrate.make (fun () -> async.Return (action ())) rc,
|
||||
ec
|
||||
)
|
||||
)
|
||||
|> mb.PostAndAsyncReply
|
||||
|
||||
match response with
|
||||
@@ -422,7 +448,14 @@ type ParallelQueue
|
||||
let ec = ExecutionContext.Capture ()
|
||||
|
||||
let! response =
|
||||
(fun rc -> MailboxMessage.RunTest (parent, par, ThunkCrate.make action rc, ec))
|
||||
(fun rc ->
|
||||
MailboxMessage.RunTestAsync (
|
||||
parent,
|
||||
par,
|
||||
AsyncThunkCrate.make (fun () -> async.Return (action ())) rc,
|
||||
ec
|
||||
)
|
||||
)
|
||||
|> mb.PostAndAsyncReply
|
||||
|
||||
match response with
|
||||
|
||||
@@ -256,6 +256,7 @@ WoofWare.NUnitTestRunner.ParallelQueue inherit obj, implements IDisposable
|
||||
WoofWare.NUnitTestRunner.ParallelQueue..ctor [constructor]: (int option, WoofWare.NUnitTestRunner.AssemblyParallelScope WoofWare.NUnitTestRunner.Parallelizable option, System.Threading.CancellationToken option)
|
||||
WoofWare.NUnitTestRunner.ParallelQueue.EndTestFixture [method]: WoofWare.NUnitTestRunner.TestFixtureTearDownToken -> unit System.Threading.Tasks.Task
|
||||
WoofWare.NUnitTestRunner.ParallelQueue.Run [method]: WoofWare.NUnitTestRunner.TestFixtureSetupToken -> unit WoofWare.NUnitTestRunner.Parallelizable option -> (unit -> 'a) -> 'a System.Threading.Tasks.Task
|
||||
WoofWare.NUnitTestRunner.ParallelQueue.RunAsync [method]: WoofWare.NUnitTestRunner.TestFixtureSetupToken -> unit WoofWare.NUnitTestRunner.Parallelizable option -> (unit -> 'a Microsoft.FSharp.Control.FSharpAsync) -> 'a System.Threading.Tasks.Task
|
||||
WoofWare.NUnitTestRunner.ParallelQueue.RunTestSetup [method]: WoofWare.NUnitTestRunner.TestFixtureRunningToken -> (unit -> 'a) -> ('a * WoofWare.NUnitTestRunner.TestFixtureSetupToken) System.Threading.Tasks.Task
|
||||
WoofWare.NUnitTestRunner.ParallelQueue.RunTestTearDown [method]: WoofWare.NUnitTestRunner.TestFixtureSetupToken -> (unit -> 'a) -> ('a * WoofWare.NUnitTestRunner.TestFixtureTearDownToken) System.Threading.Tasks.Task
|
||||
WoofWare.NUnitTestRunner.ParallelQueue.StartTestFixture [method]: WoofWare.NUnitTestRunner.TestFixture -> WoofWare.NUnitTestRunner.TestFixtureRunningToken System.Threading.Tasks.Task
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"version": "0.21",
|
||||
"version": "0.22",
|
||||
"publicReleaseRefSpec": [
|
||||
"^refs/heads/main$"
|
||||
],
|
||||
|
||||
@@ -243,23 +243,18 @@ module TestSynchronizationContext =
|
||||
do! queue.EndTestFixture teardown
|
||||
}
|
||||
|
||||
(*
|
||||
[<Test>]
|
||||
let ``ExecutionContext flows correctly through async operations`` () =
|
||||
task {
|
||||
// Create a test fixture
|
||||
let dummyFixture =
|
||||
TestFixture.Empty
|
||||
typeof<obj>
|
||||
(Some (Parallelizable.Yes ClassParallelScope.All))
|
||||
[]
|
||||
[]
|
||||
TestFixture.Empty typeof<obj> (Some (Parallelizable.Yes ClassParallelScope.All)) [] []
|
||||
|
||||
use contexts = TestContexts.Empty ()
|
||||
use queue = new ParallelQueue(Some 4, None)
|
||||
use queue = new ParallelQueue (Some 4, None)
|
||||
|
||||
// Track which context values we see during execution
|
||||
let contextValues = System.Collections.Concurrent.ConcurrentBag<Guid * Guid>()
|
||||
let contextValues = System.Collections.Concurrent.ConcurrentBag<Guid * Guid> ()
|
||||
|
||||
// Start the fixture
|
||||
let! running = queue.StartTestFixture dummyFixture
|
||||
@@ -267,42 +262,45 @@ module TestSynchronizationContext =
|
||||
|
||||
// Create several async operations with different context values
|
||||
let tasks =
|
||||
[1..10]
|
||||
[ 1..10 ]
|
||||
|> List.map (fun i ->
|
||||
task {
|
||||
// Set a unique context value
|
||||
let expectedId = Guid.NewGuid()
|
||||
let expectedId = Guid.NewGuid ()
|
||||
let outputId = OutputStreamId expectedId
|
||||
contexts.AsyncLocal.Value <- outputId
|
||||
|
||||
// Run an async operation that checks the context at multiple points
|
||||
let! actualId =
|
||||
queue.RunAsync setup None (fun () ->
|
||||
async {
|
||||
// Check context immediately
|
||||
let immediate = contexts.AsyncLocal.Value
|
||||
let (OutputStreamId immediateGuid) = immediate
|
||||
contextValues.Add(expectedId, immediateGuid)
|
||||
queue.RunAsync
|
||||
setup
|
||||
None
|
||||
(fun () ->
|
||||
async {
|
||||
// Check context immediately
|
||||
let immediate = contexts.AsyncLocal.Value
|
||||
let (OutputStreamId immediateGuid) = immediate
|
||||
contextValues.Add (expectedId, immediateGuid)
|
||||
|
||||
// Yield to allow potential context loss
|
||||
do! Async.Sleep 10
|
||||
// Yield to allow potential context loss
|
||||
do! Async.Sleep 10
|
||||
|
||||
// Check context after yield
|
||||
let afterYield = contexts.AsyncLocal.Value
|
||||
let (OutputStreamId afterYieldGuid) = afterYield
|
||||
contextValues.Add(expectedId, afterYieldGuid)
|
||||
// Check context after yield
|
||||
let afterYield = contexts.AsyncLocal.Value
|
||||
let (OutputStreamId afterYieldGuid) = afterYield
|
||||
contextValues.Add (expectedId, afterYieldGuid)
|
||||
|
||||
// Do some actual async work
|
||||
do! Task.Delay(10) |> Async.AwaitTask
|
||||
// Do some actual async work
|
||||
do! Task.Delay (10) |> Async.AwaitTask
|
||||
|
||||
// Check context after task
|
||||
let afterTask = contexts.AsyncLocal.Value
|
||||
let (OutputStreamId afterTaskGuid) = afterTask
|
||||
contextValues.Add(expectedId, afterTaskGuid)
|
||||
// Check context after task
|
||||
let afterTask = contexts.AsyncLocal.Value
|
||||
let (OutputStreamId afterTaskGuid) = afterTask
|
||||
contextValues.Add (expectedId, afterTaskGuid)
|
||||
|
||||
return afterTaskGuid
|
||||
}
|
||||
)
|
||||
return afterTaskGuid
|
||||
}
|
||||
)
|
||||
|
||||
// Verify the returned value matches what we set
|
||||
actualId |> shouldEqual expectedId
|
||||
@@ -310,16 +308,16 @@ module TestSynchronizationContext =
|
||||
)
|
||||
|
||||
// Wait for all tasks
|
||||
let! results = Task.WhenAll(tasks)
|
||||
let! results = Task.WhenAll (tasks)
|
||||
results |> Array.iter id
|
||||
|
||||
// Verify all context values were correct
|
||||
let allValues = contextValues |> Seq.toList
|
||||
allValues |> shouldHaveLength 30 // 3 checks per operation * 10 operations
|
||||
allValues |> shouldHaveLength 30 // 3 checks per operation * 10 operations
|
||||
|
||||
// Every captured value should match its expected value
|
||||
for expected, actual in allValues do
|
||||
actual |> shouldEqual expected
|
||||
allValues
|
||||
|> List.iter (fun (expected, actual) -> actual |> shouldEqual expected)
|
||||
|
||||
// Clean up
|
||||
let! _, teardown = queue.RunTestTearDown setup (fun () -> ())
|
||||
@@ -330,80 +328,69 @@ module TestSynchronizationContext =
|
||||
let ``ExecutionContext isolation between concurrent operations`` () =
|
||||
task {
|
||||
let dummyFixture =
|
||||
TestFixture.Empty
|
||||
typeof<obj>
|
||||
(Some (Parallelizable.Yes ClassParallelScope.All))
|
||||
[]
|
||||
[]
|
||||
TestFixture.Empty typeof<obj> (Some (Parallelizable.Yes ClassParallelScope.All)) [] []
|
||||
|
||||
use contexts = TestContexts.Empty ()
|
||||
use queue = new ParallelQueue(Some 4, None)
|
||||
use queue = new ParallelQueue (Some 4, None)
|
||||
|
||||
let! running = queue.StartTestFixture dummyFixture
|
||||
let! _, setup = queue.RunTestSetup running (fun () -> ())
|
||||
|
||||
// Use a barrier to ensure operations run concurrently
|
||||
let barrier = new Barrier(3)
|
||||
let seenValues = System.Collections.Concurrent.ConcurrentBag<int * Guid option>()
|
||||
let barrier = new Barrier (3)
|
||||
let seenValues = System.Collections.Concurrent.ConcurrentBag<int * Guid> ()
|
||||
|
||||
// Create operations that will definitely run concurrently
|
||||
let tasks =
|
||||
[1..3]
|
||||
[ 1..3 ]
|
||||
|> List.map (fun i ->
|
||||
task {
|
||||
// Each task sets its own context value
|
||||
let myId = Guid.NewGuid()
|
||||
let myId = Guid.NewGuid ()
|
||||
contexts.AsyncLocal.Value <- OutputStreamId myId
|
||||
|
||||
let! result =
|
||||
queue.RunAsync setup (Some (Parallelizable.Yes ())) (fun () ->
|
||||
async {
|
||||
// Wait for all tasks to reach this point
|
||||
barrier.SignalAndWait() |> ignore
|
||||
queue.RunAsync
|
||||
setup
|
||||
(Some (Parallelizable.Yes ()))
|
||||
(fun () ->
|
||||
async {
|
||||
// Wait for all tasks to reach this point
|
||||
barrier.SignalAndWait () |> ignore
|
||||
|
||||
// Now check what value we see
|
||||
let currentValue = contexts.AsyncLocal.Value
|
||||
match currentValue with
|
||||
| OutputStreamId guid -> seenValues.Add(i, Some guid)
|
||||
| _ -> seenValues.Add(i, None)
|
||||
// Now check what value we see
|
||||
let currentValue = contexts.AsyncLocal.Value
|
||||
|
||||
// Do some async work
|
||||
do! Async.Sleep 5
|
||||
match currentValue with
|
||||
| OutputStreamId guid -> seenValues.Add (i, guid)
|
||||
|
||||
// Check again after async work
|
||||
let afterAsync = contexts.AsyncLocal.Value
|
||||
match afterAsync with
|
||||
| OutputStreamId guid ->
|
||||
return guid
|
||||
| _ ->
|
||||
return failwith "Lost context after async"
|
||||
}
|
||||
)
|
||||
// Do some async work
|
||||
do! Async.Sleep 5
|
||||
|
||||
// Check again after async work
|
||||
let afterAsync = contexts.AsyncLocal.Value
|
||||
|
||||
match afterAsync with
|
||||
| OutputStreamId guid -> return guid
|
||||
}
|
||||
)
|
||||
|
||||
// Each task should see its own value
|
||||
result |> shouldEqual myId
|
||||
}
|
||||
)
|
||||
|
||||
let! results = Task.WhenAll(tasks)
|
||||
let! results = Task.WhenAll (tasks)
|
||||
results |> Array.iter id
|
||||
|
||||
// Verify we saw 3 different values (one per task)
|
||||
let values = seenValues |> Seq.toList
|
||||
values |> shouldHaveLength 3
|
||||
|
||||
// Each task should have seen a value
|
||||
for (taskId, value) in values do
|
||||
value |> shouldNotEqual None
|
||||
|
||||
// All seen values should be different (no context bleeding)
|
||||
let uniqueValues =
|
||||
values
|
||||
|> List.choose snd
|
||||
|> List.distinct
|
||||
let uniqueValues = values |> List.map snd |> List.distinct
|
||||
uniqueValues |> shouldHaveLength 3
|
||||
|
||||
let! _, teardown = queue.RunTestTearDown setup (fun () -> ())
|
||||
do! queue.EndTestFixture teardown
|
||||
}
|
||||
*)
|
||||
|
||||
Reference in New Issue
Block a user