Skip to content

Fixing streaming of sequences #210

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ivelten wants to merge 8 commits into
base: dev
Choose a base branch
from
Open

Fixing streaming of sequences #210

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
b0f80b8
Adjusting tests for machines with less processor cores
ivelten Feb 1, 2019
c1cf5a4
Removing test from other branch
ivelten Feb 1, 2019
f8c0959
Merge from test-fixes-2
ivelten Feb 1, 2019
fd65327
Increasing tolerance in WaitCompleted
ivelten Feb 1, 2019
81507c3
Merge branch 'test-fixes-2' into defer-adjusts-4
ivelten Feb 1, 2019
83cef85
Disposing test observer after usage
ivelten Feb 4, 2019
c6a6108
Changing the build script to show result of each ran unit test
ivelten Feb 4, 2019
636e4ae
Merge branch 'test-fixes-2' into defer-adjusts-4
ivelten Feb 4, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion .paket/Paket.Restore.targets
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,9 @@
<!-- see https://github.com/fsharp/fslang-design/blob/master/RFCs/FS-1032-fsharp-in-dotnet-sdk.md -->
<DisableImplicitFSharpCoreReference>true</DisableImplicitFSharpCoreReference>
<DisableImplicitSystemValueTupleReference>true</DisableImplicitSystemValueTupleReference>

<!-- Disable Paket restore under NCrunch build -->
<PaketRestoreDisabled Condition="'$(NCrunch)' == '1'">True</PaketRestoreDisabled>
</PropertyGroup>

<Target Name="PaketBootstrapping" Condition="Exists('$(PaketToolsPath)paket.bootstrapper.proj')">
Expand Down Expand Up @@ -102,7 +105,11 @@
<PaketRestoreRequired Condition=" '$(PaketRestoreLockFileHash)' == '' ">true</PaketRestoreRequired>
</PropertyGroup>

<PropertyGroup Condition="'$(PaketPropsVersion)' != '5.174.2' ">
<!--
This value should match the version in the props generated by paket
If they differ, this means we need to do a restore in order to ensure correct dependencies
-->
<PropertyGroup Condition="'$(PaketPropsVersion)' != '5.185.3' ">
<PaketRestoreRequired>true</PaketRestoreRequired>
</PropertyGroup>

Expand Down Expand Up @@ -183,6 +190,7 @@
<ExcludeAssets Condition=" '%(PaketReferencesFileLinesInfo.Splits)' == '6' And %(PaketReferencesFileLinesInfo.CopyLocal) == 'false'">runtime</ExcludeAssets>
<ExcludeAssets Condition=" '%(PaketReferencesFileLinesInfo.Splits)' != '6' And %(PaketReferencesFileLinesInfo.AllPrivateAssets) == 'exclude'">runtime</ExcludeAssets>
<Publish Condition=" '$(PackAsTool)' == 'true' ">true</Publish>
<AllowExplicitVersion>true</AllowExplicitVersion>
</PackageReference>
</ItemGroup>

Expand Down
2 changes: 2 additions & 0 deletions build.fsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,10 +205,12 @@ Target "Build" (fun _ ->
)

Target "RunTests" (fun _ ->
Console.WriteLine("Number of processors: {0}", Environment.ProcessorCount)
DotNetCli.Test (fun p -> { p with
#if MONO
Framework = "netcoreapp2.0"
#endif
AdditionalArgs = [ "--no-build"; "-v=normal" ]
Configuration = "Release"
Project = "tests/FSharp.Data.GraphQL.Tests/FSharp.Data.GraphQL.Tests.fsproj" })
)
Expand Down
49 changes: 23 additions & 26 deletions src/FSharp.Data.GraphQL.Server/Execution.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -259,7 +259,7 @@ type ResolverTree =
| ResolverListNode l -> l.Value
and ResolverLeaf = { Name: string; Value: obj option }
and ResolverError = { Name: string; Message: string; PathToOrigin: obj list }
and ResolverNode = { Name: string; Value: obj option; Children: AsyncVal<ResolverTree> [] }
and ResolverNode = { Name: string; Value: obj option; Children: AsyncVal<ResolverTree> seq }

module ResolverTree =
let rec pathFold leafOp errorOp nodeOp listOp =
Expand All @@ -279,21 +279,21 @@ module ResolverTree =
let! c' = c
return helper path' c'
}
let ts = node.Children |> Array.map mapper
let ts = node.Children |> Seq.map mapper
nodeOp path' node.Name node.Value ts
| ResolverListNode node ->
let path' = (node.Name :> obj)::path
let mapper (i : int) (c : AsyncVal<ResolverTree>) = asyncVal {
let! c' = c
return helper ((box i)::path') c'
}
let ts = node.Children |> Array.mapi mapper
let ts = node.Children |> Seq.mapi mapper
listOp path' node.Name node.Value ts
helper []

let private foldChildren (children : AsyncVal<AsyncVal<KeyValuePair<string, obj> * Error list>> []) =
let private foldChildren (children : AsyncVal<AsyncVal<KeyValuePair<string, obj> * Error list>> seq) =
children
|> Array.fold (fun (kvpsErrs : AsyncVal<KeyValuePair<string, obj> list * Error list>) child -> asyncVal {
|> Seq.fold (fun (kvpsErrs : AsyncVal<KeyValuePair<string, obj> list * Error list>) child -> asyncVal {
let! kvps, errs = kvpsErrs
let! c = child
let! c, e = c
Expand Down Expand Up @@ -345,7 +345,7 @@ let private treeToStream (streamOptions : BufferedStreamOptions) tree =
function
| ResolverListNode list ->
list.Children
|> Array.mapi (fun i x ->
|> Seq.mapi (fun i x ->
asyncVal {
let! x' = x
return i, x'
Expand Down Expand Up @@ -404,7 +404,7 @@ let rec private buildResolverTree (returnDef: OutputDef) (ctx: ResolveFieldConte
| Some v -> buildObjectFields fields objdef ctx fieldExecuteMap name v
| None ->
if ctx.ExecutionInfo.IsNullable
then asyncVal { return ResolverObjectNode { Name = name; Value = None; Children = [| |] } }
then asyncVal { return ResolverObjectNode { Name = name; Value = None; Children = [||] } }
else nullResolverError name
| kind -> failwithf "Unexpected value of ctx.ExecutionPlan.Kind: %A" kind
| Scalar scalardef ->
Expand All @@ -424,28 +424,25 @@ let rec private buildResolverTree (returnDef: OutputDef) (ctx: ResolveFieldConte
match kind with
| ResolveCollection innerPlan -> { ctx with ExecutionInfo = innerPlan }
| kind -> failwithf "Unexpected value of ctx.ExecutionPlan.Kind: %A" kind
let rec build acc (items: obj list) =
match items with
| value::xs ->
if not innerCtx.ExecutionInfo.IsNullable && isNull value
then nullResolverError innerCtx.ExecutionInfo.Identifier
let build (items : seq<obj>) =
let children = seq {
for item in items do
if not innerCtx.ExecutionInfo.IsNullable && isNull item
then yield nullResolverError innerCtx.ExecutionInfo.Identifier
else
let t = asyncVal {
let! res = buildResolverTree innerdef innerCtx fieldExecuteMap (toOption value)
yield asyncVal {
let! res = buildResolverTree innerdef innerCtx fieldExecuteMap (toOption item)
match res with
| ResolverError e when not innerCtx.ExecutionInfo.IsNullable -> return! propagateError name e
| _ -> return res
}
build (t::acc) xs
| [] -> asyncVal { return ResolverListNode { Name = name; Value = value; Children = acc |> List.map (AsyncVal.map (fun x -> x)) |> List.rev |> List.toArray } }
| _ -> return res } }
asyncVal { return ResolverListNode { Name = name; Value = value; Children = children } }
match value with
| None when not ctx.ExecutionInfo.IsNullable -> nullResolverError name
| None -> asyncVal { return ResolverListNode { Name = name; Value = None; Children = [| |] } }
| None -> asyncVal { return ResolverListNode { Name = name; Value = None; Children = [||] } }
| ObjectOption (:? System.Collections.IEnumerable as enumerable) ->
enumerable
|> Seq.cast<obj>
|> Seq.toList
|> build []
|> build
| _ -> raise <| GraphQLException (sprintf "Expected to have enumerable value in field '%s' but got '%O'" ctx.ExecutionInfo.Identifier (value.GetType()))
| Nullable (Output innerdef) ->
// Stop propagation of null values
Expand All @@ -465,7 +462,7 @@ let rec private buildResolverTree (returnDef: OutputDef) (ctx: ResolveFieldConte
| None -> asyncVal { return ResolverError { Name = name; Message = ctx.Schema.ParseError (GraphQLException (sprintf "GraphQL Interface '%s' is not implemented by the type '%s'" idef.Name resolvedDef.Name)); PathToOrigin = [] } }
| None ->
if ctx.ExecutionInfo.IsNullable
then asyncVal { return ResolverObjectNode { Name = name; Value = None; Children = [| |] } }
then asyncVal { return ResolverObjectNode { Name = name; Value = None; Children = [||] } }
else nullResolverError name
| Union udef ->
let possibleTypesFn = ctx.Schema.GetPossibleTypes
Expand All @@ -488,13 +485,13 @@ let rec private buildResolverTree (returnDef: OutputDef) (ctx: ResolveFieldConte
| None -> asyncVal { return ResolverError { Name = name; Message = ctx.Schema.ParseError (GraphQLException (sprintf "GraphQL Union '%s' is not implemented by the type '%s'" udef.Name resolvedDef.Name)); PathToOrigin = [] } }
| None ->
if ctx.ExecutionInfo.IsNullable
then asyncVal { return ResolverObjectNode { Name = name; Value = None; Children = [| |] } }
then asyncVal { return ResolverObjectNode { Name = name; Value = None; Children = [||] } }
else nullResolverError name
| _ -> failwithf "Unexpected value of returnDef: %O" returnDef
match ctx.ExecutionInfo.Kind, returnDef, ctx.ExecutionInfo.IsDeferred with
| ResolveDeferred _, (Scalar _ | Enum _ | Nullable _), false -> asyncVal { return ResolverLeaf { Name = name; Value = None } }
| ResolveDeferred _, (Object _ | Interface _ | Union _), false -> asyncVal { return ResolverObjectNode { Name = name; Value = None; Children = [| |] } }
| ResolveDeferred _, List _, false -> asyncVal { return ResolverListNode { Name = name; Value = Some (upcast [ ]); Children = [| |] } }
| ResolveDeferred _, (Object _ | Interface _ | Union _), false -> asyncVal { return ResolverObjectNode { Name = name; Value = None; Children = [||] } }
| ResolveDeferred _, List _, false -> asyncVal { return ResolverListNode { Name = name; Value = Some (upcast [ ]); Children = [||] } }
| _ -> resolveDirect ctx.ExecutionInfo.Kind

and buildObjectFields (fields: ExecutionInfo list) (objdef: ObjectDef) (ctx: ResolveFieldContext) (fieldExecuteMap: FieldExecuteMap) (name: string) (value: obj): AsyncVal<ResolverTree> =
Expand Down Expand Up @@ -680,7 +677,7 @@ let private executeQueryOrMutation (resultSet: (string * ExecutionInfo) []) (ctx

let mapLive (tree : ResolverTree) (path : obj list) (d : DeferredExecutionInfo) (fieldCtx : ResolveFieldContext) = asyncVal {
let getFieldName (node : ResolverNode) = asyncVal {
match node.Children |> Array.tryHead with
match node.Children |> Seq.tryHead with
| Some c ->
let! res = c
return res.Name
Expand Down
12 changes: 5 additions & 7 deletions src/FSharp.Data.GraphQL.Shared/AsyncVal.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,10 @@ module AsyncVal =
/// Returned array maintain order of values.
/// If the array contains a Failure, then the entire array will not resolve
let collectSequential (values: AsyncVal<'T> []) : AsyncVal<'T []> =
let mapper =
function
| Value v -> v
| other -> failwithf "Expected a synchronous value, but got %O" other
if values.Length = 0 then Value [||]
elif values |> Array.exists isAsync then
Async(async {
Expand All @@ -129,7 +133,7 @@ module AsyncVal =
| Failure f ->
results.[i] <- raise f
return results })
else Value (values |> Array.map (fun (Value v) -> v))
else Value (values |> Array.map mapper)



Expand Down Expand Up @@ -197,12 +201,6 @@ module AsyncExtensions =

/// Computation expression for working on AsyncVals.
let asyncVal = AsyncValBuilder ()

/// Active pattern used for checking if AsyncVal contains immediate value.
let (|Immediate|_|) (x: AsyncVal<'T>) = match x with | Value v -> Some v | _ -> None

/// Active patter used for checking if AsyncVal wraps an Async computation.
let (|Async|_|) (x: AsyncVal<'T>) = match x with | Async a -> Some a | _ -> None

type Microsoft.FSharp.Control.AsyncBuilder with

Expand Down
70 changes: 39 additions & 31 deletions tests/FSharp.Data.GraphQL.Tests/DeferredTests.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,22 @@ open FSharp.Data.GraphQL
open FSharp.Data.GraphQL.Parser
open FSharp.Data.GraphQL.Execution
open System.Threading
open System.Collections.Generic
open System.Collections.Concurrent
open FSharp.Data.GraphQL.Types

#nowarn "40"

let ms x =
let factor =
match Environment.ProcessorCount with
| x when x >= 8 -> 1
| x when x >= 4 -> 4
| _ -> 6
x * factor

let delay time x = async {
do! Async.Sleep(ms time)
return x }

type TestSubject = {
id: string
a: string
Expand Down Expand Up @@ -157,11 +167,6 @@ let DataType =
Define.Field("bufferedList", ListOf AsyncDataType, (fun _ d -> d.bufferedList))
])

let delay ms x = async {
do! Async.Sleep(ms)
return x
}

let data = {
id = "1"
a = "Apple"
Expand Down Expand Up @@ -629,9 +634,9 @@ let ``Each live result should be sent as soon as it is computed`` () =
// The second result is a delayed async field, which is set to compute the value for 5 seconds.
// The first result should come as soon as the live value is updated, which sould be almost instantly.
// Therefore, let's assume that if it does not come in at least 3 seconds, test has failed.
if TimeSpan.FromSeconds(float 3) |> mre1.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 3)) |> mre1.WaitOne |> not
then fail "Timeout while waiting for first deferred result"
if TimeSpan.FromSeconds(float 30) |> mre2.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 10)) |> mre2.WaitOne |> not
then fail "Timeout while waiting for second deferred result"
sub.Received
|> Seq.cast<NameValueLookup>
Expand Down Expand Up @@ -1282,13 +1287,16 @@ let ``Should buffer stream list correctly by timing information``() =
]
"path", upcast [box "testData"; upcast "bufferedList"; upcast [0]]
]
let query = parse """{
testData {
bufferedList @stream(interval : 3000) {
value
let query =
ms 3000
|> sprintf """{
testData {
bufferedList @stream(interval : %i) {
value
}
}
}
}"""
}"""
|> parse
use mre1 = new ManualResetEvent(false)
use mre2 = new ManualResetEvent(false)
let result = query |> executor.AsyncExecute |> sync
Expand All @@ -1306,11 +1314,11 @@ let ``Should buffer stream list correctly by timing information``() =
// to buffer results 3 and 2 (in this order), as together they take less than 3 seconds to compute,
// and send them together on the first batch.
// First result should come in a second batch, as it takes 5 seconds to compute, more than the time limit of the buffer.
if TimeSpan.FromSeconds(float 4) |> mre1.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 4)) |> mre1.WaitOne |> not
then fail "Timeout while waiting for first Deferred GQLResponse"
if TimeSpan.FromSeconds(float 30) |> mre2.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 10)) |> mre2.WaitOne |> not
then fail "Timeout while waiting for second Deferred GQLResponse"
sub.WaitCompleted()
sub.WaitCompleted(timeout = ms 10)
sub.Received
|> Seq.cast<NameValueLookup>
|> itemEquals 0 expectedDeferred1
Expand Down Expand Up @@ -1372,11 +1380,11 @@ let ``Should buffer stream list correctly by count information``() =
// and send them together on the first batch.
// First result should come in a second batch, as it takes 5 seconds to compute, which should be enough
// to put the two other results in a batch with the preferred size.
if TimeSpan.FromSeconds(float 4) |> mre1.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 4)) |> mre1.WaitOne |> not
then fail "Timeout while waiting for first Deferred GQLResponse"
if TimeSpan.FromSeconds(float 30) |> mre2.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 10)) |> mre2.WaitOne |> not
then fail "Timeout while waiting for second Deferred GQLResponse"
sub.WaitCompleted()
sub.WaitCompleted(timeout = ms 10)
sub.Received
|> Seq.cast<NameValueLookup>
|> itemEquals 0 expectedDeferred1
Expand Down Expand Up @@ -1469,11 +1477,11 @@ let ``Each deferred result should be sent as soon as it is computed``() =
// The second result is a delayed async field, which is set to compute the value for 5 seconds.
// The first result should come almost instantly, as it is not a delayed computed field.
// Therefore, let's assume that if it does not come in at least 3 seconds, test has failed.
if TimeSpan.FromSeconds(float 3) |> mre1.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 3)) |> mre1.WaitOne |> not
then fail "Timeout while waiting for first deferred result"
if TimeSpan.FromSeconds(float 30) |> mre2.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 10)) |> mre2.WaitOne |> not
then fail "Timeout while waiting for second deferred result"
sub.WaitCompleted()
sub.WaitCompleted(timeout = ms 10)
sub.Received
|> Seq.cast<NameValueLookup>
|> itemEquals 0 expectedDeferred1
Expand Down Expand Up @@ -1526,11 +1534,11 @@ let ``Each deferred result of a list should be sent as soon as it is computed``
// The first result is a delayed async field, which is set to compute the value for 5 seconds.
// The second result should come first, almost instantly, as it is not a delayed computed field.
// Therefore, let's assume that if it does not come in at least 4 seconds, test has failed.
if TimeSpan.FromSeconds(float 4) |> mre1.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 4)) |> mre1.WaitOne |> not
then fail "Timeout while waiting for first deferred result"
if TimeSpan.FromSeconds(float 30) |> mre2.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 10)) |> mre2.WaitOne |> not
then fail "Timeout while waiting for second deferred result"
sub.WaitCompleted()
sub.WaitCompleted(timeout = ms 10)
sub.Received
|> Seq.cast<NameValueLookup>
|> itemEquals 0 expectedDeferred1
Expand All @@ -1539,7 +1547,7 @@ let ``Each deferred result of a list should be sent as soon as it is computed``
| _ -> fail "Expected Deferred GQLRespnse"

[<Fact>]
let ``Each streamed result should be sent as soon as it is computed``() =
let ``Each streamed result should be sent as soon as it is computed - async seq``() =
let expectedDirect =
NameValueLookup.ofList [
"testData", upcast NameValueLookup.ofList [
Expand Down Expand Up @@ -1584,11 +1592,11 @@ let ``Each streamed result should be sent as soon as it is computed``() =
// The first result is a delayed async field, which is set to compute the value for 5 seconds.
// The second result should come first, almost instantly, as it is not a delayed computed field.
// Therefore, let's assume that if it does not come in at least 4 seconds, test has failed.
if TimeSpan.FromSeconds(float 4) |> mre1.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 4)) |> mre1.WaitOne |> not
then fail "Timeout while waiting for first deferred result"
if TimeSpan.FromSeconds(float 30) |> mre2.WaitOne |> not
if TimeSpan.FromSeconds(float (ms 10)) |> mre2.WaitOne |> not
then fail "Timeout while waiting for second deferred result"
sub.WaitCompleted()
sub.WaitCompleted(timeout = ms 10)
sub.Received
|> Seq.cast<NameValueLookup>
|> itemEquals 0 expectedDeferred1
Expand Down
Loading