From 255db1f8b9a3131fae412d6cd867e32bd3af6c6d Mon Sep 17 00:00:00 2001 From: Andrii Chebukin Date: Sun, 24 Mar 2024 03:23:36 +0400 Subject: [PATCH] Refactored `AsyncVal` to use `PooledResizeArray` under the hood --- .../GraphQLWebsocketMiddleware.fs | 2 +- src/FSharp.Data.GraphQL.Server/Execution.fs | 6 +-- .../FSharp.Data.GraphQL.Server.fsproj | 1 + src/FSharp.Data.GraphQL.Shared/AsyncVal.fs | 52 +++++++++++-------- .../FSharp.Data.GraphQL.Shared.fsproj | 5 ++ .../Helpers/PooledResizeArray.fs | 36 +++++++++++++ 6 files changed, 75 insertions(+), 27 deletions(-) create mode 100644 src/FSharp.Data.GraphQL.Shared/Helpers/PooledResizeArray.fs diff --git a/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs b/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs index 7e225cf6a..8e130f075 100644 --- a/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs +++ b/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs @@ -82,7 +82,7 @@ type GraphQLWebSocketMiddleware<'Root> let receiveMessageViaSocket (cancellationToken : CancellationToken) (serializerOptions : JsonSerializerOptions) (socket : WebSocket) = taskResult { let buffer = ArrayPool.Shared.Rent options.ReadBufferSize try - let completeMessage = new PooledList () + use completeMessage = new PooledResizeArray () let mutable segmentResponse : WebSocketReceiveResult = null while (not cancellationToken.IsCancellationRequested) && socket |> isSocketOpen diff --git a/src/FSharp.Data.GraphQL.Server/Execution.fs b/src/FSharp.Data.GraphQL.Server/Execution.fs index a66377d47..cb6b6ee30 100644 --- a/src/FSharp.Data.GraphQL.Server/Execution.fs +++ b/src/FSharp.Data.GraphQL.Server/Execution.fs @@ -300,7 +300,7 @@ let deferResults path (res : ResolverResult) : IObservable Observable.singleton <| DeferredErrors (null, errs, formattedPath) /// Collect together an array of results using the appropriate execution strategy. -let collectFields (strategy : ExecutionStrategy) (rs : AsyncVal>> []) : AsyncVal []>> = asyncVal { +let collectFields (strategy : ExecutionStrategy) (rs : AsyncVal>> seq) : AsyncVal []>> = asyncVal { let! collected = match strategy with | Parallel -> AsyncVal.collectParallel rs @@ -354,8 +354,7 @@ let rec private direct (returnDef : OutputDef) (ctx : ResolveFieldContext) (path | :? System.Collections.IEnumerable as enumerable -> enumerable |> Seq.cast - |> Seq.toArray - |> Array.mapi resolveItem + |> Seq.mapi resolveItem |> collectFields Parallel |> AsyncVal.map(ResolverResult.mapValue(fun items -> KeyValuePair(name, items |> Array.map(fun d -> d.Value) |> box))) | _ -> raise <| GQLMessageException (ErrorMessages.expectedEnumerableValue ctx.ExecutionInfo.Identifier (value.GetType())) @@ -543,7 +542,6 @@ and executeObjectFields (fields : ExecutionInfo list) (objName : string) (objDef let! res = fields |> Seq.map executeField - |> Seq.toArray |> collectFields Parallel match res with | Error errs -> return Error errs diff --git a/src/FSharp.Data.GraphQL.Server/FSharp.Data.GraphQL.Server.fsproj b/src/FSharp.Data.GraphQL.Server/FSharp.Data.GraphQL.Server.fsproj index cca4e1812..ddd43933f 100644 --- a/src/FSharp.Data.GraphQL.Server/FSharp.Data.GraphQL.Server.fsproj +++ b/src/FSharp.Data.GraphQL.Server/FSharp.Data.GraphQL.Server.fsproj @@ -23,6 +23,7 @@ + diff --git a/src/FSharp.Data.GraphQL.Shared/AsyncVal.fs b/src/FSharp.Data.GraphQL.Shared/AsyncVal.fs index 80b29fe05..ec8c4cfef 100644 --- a/src/FSharp.Data.GraphQL.Shared/AsyncVal.fs +++ b/src/FSharp.Data.GraphQL.Shared/AsyncVal.fs @@ -130,13 +130,15 @@ module AsyncVal = /// executed asynchronously, one by one with regard to their order in array. /// 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[]> = - if values.Length = 0 then Value [||] - elif values |> Array.exists isAsync then + let collectSequential (values : AsyncVal<'T> seq) : AsyncVal<'T[]> = + let values = new PooledResizeArray<_> (values) + let length = values.Count + if length = 0 then Value [||] + elif values.Exists isAsync then Async (async { - let results = Array.zeroCreate values.Length - let exceptions = ResizeArray values.Length - for i = 0 to values.Length - 1 do + let results = Array.zeroCreate length + use exceptions = new PooledResizeArray<_> (length) + for i = 0 to length - 1 do let v = values.[i] match v with | Value v -> results.[i] <- v @@ -144,35 +146,39 @@ module AsyncVal = let! r = a results.[i] <- r | Failure f -> exceptions.Add f + values.Dispose() match exceptions.Count with | 0 -> return results | 1 -> return exceptions.First().Reraise () - | _ -> return AggregateException exceptions |> raise + | _ -> return AggregateException (exceptions.AsReadOnly()) |> raise }) else - let exceptions = + use values = values + use exceptions = values - |> Array.choose (function - | Failure f -> Some f - | _ -> None) - match exceptions.Length with - | 0 -> Value (values |> Array.map (fun (Value v) -> v)) + |> PooledResizeArray.vChoose (function + | Failure f -> ValueSome f + | _ -> ValueNone) + match exceptions.Count with + | 0 -> Value (values |> Seq.map (fun (Value v) -> v) |> Seq.toArray) | 1 -> Failure (exceptions.First ()) - | _ -> Failure (AggregateException exceptions) + | _ -> Failure (AggregateException (exceptions.AsReadOnly())) /// Converts array of AsyncVals into AsyncVal with array results. /// In case when are non-immediate values in provided array, they are /// executed all in parallel, in unordered fashion. Order of values /// inside returned array is maintained. /// If the array contains a Failure, then the entire array will not resolve - let collectParallel (values : AsyncVal<'T>[]) : AsyncVal<'T[]> = - if values.Length = 0 then Value [||] + let collectParallel (values : AsyncVal<'T> seq) : AsyncVal<'T[]> = + use values = new PooledResizeArray<_> (values) + let length = values.Count + if length = 0 then Value [||] else - let indexes = List<_> (0) - let continuations = List<_> (0) - let results = Array.zeroCreate values.Length - let exceptions = ResizeArray values.Length - for i = 0 to values.Length - 1 do + let indexes = new PooledResizeArray<_> (length) + let continuations = new PooledResizeArray<_> (length) + let results = Array.zeroCreate length + use exceptions = new PooledResizeArray<_> (length) + for i = 0 to length - 1 do let value = values.[i] match value with | Value v -> results.[i] <- v @@ -182,13 +188,15 @@ module AsyncVal = | Failure f -> exceptions.Add f match exceptions.Count with | 1 -> AsyncVal.Failure (exceptions.First ()) - | count when count > 1 -> AsyncVal.Failure (AggregateException exceptions) + | count when count > 1 -> AsyncVal.Failure (AggregateException (exceptions.AsReadOnly())) | _ -> if indexes.Count = 0 then Value (results) else Async (async { let! vals = continuations |> Async.Parallel for i = 0 to indexes.Count - 1 do results.[indexes.[i]] <- vals.[i] + indexes.Dispose() + continuations.Dispose() return results }) diff --git a/src/FSharp.Data.GraphQL.Shared/FSharp.Data.GraphQL.Shared.fsproj b/src/FSharp.Data.GraphQL.Shared/FSharp.Data.GraphQL.Shared.fsproj index a201c4d9a..1d3614a3b 100644 --- a/src/FSharp.Data.GraphQL.Shared/FSharp.Data.GraphQL.Shared.fsproj +++ b/src/FSharp.Data.GraphQL.Shared/FSharp.Data.GraphQL.Shared.fsproj @@ -17,6 +17,9 @@ <_Parameter1>FSharp.Data.GraphQL.Server + + <_Parameter1>FSharp.Data.GraphQL.Server.AspNetCore + <_Parameter1>FSharp.Data.GraphQL.Client @@ -29,6 +32,7 @@ + @@ -44,6 +48,7 @@ + diff --git a/src/FSharp.Data.GraphQL.Shared/Helpers/PooledResizeArray.fs b/src/FSharp.Data.GraphQL.Shared/Helpers/PooledResizeArray.fs new file mode 100644 index 000000000..a8a6271c0 --- /dev/null +++ b/src/FSharp.Data.GraphQL.Shared/Helpers/PooledResizeArray.fs @@ -0,0 +1,36 @@ +namespace FSharp.Data.GraphQL + +open Collections.Pooled + +type internal PooledResizeArray<'T> = PooledList<'T> + +module internal PooledResizeArray = + + let ofSeqWithCapacity capacity items = + let list = new PooledResizeArray<'T> (capacity = capacity) + list.AddRange (collection = items) + list + + let exists f (list : PooledResizeArray<'T>) = list.Exists (f) + + let length (list : PooledResizeArray<'T>) = list.Count + + let vChoose f (list : PooledResizeArray<'T>) = + let res = new PooledResizeArray<_> (list.Count) + + for i = 0 to length list - 1 do + match f list[i] with + | ValueNone -> () + | ValueSome b -> res.Add (b) + + res + + let choose f (list : PooledResizeArray<'T>) = + let res = new PooledResizeArray<_> (list.Count) + + for i = 0 to length list - 1 do + match f list[i] with + | None -> () + | Some b -> res.Add (b) + + res