Skip to content

Commit dccdd58

Browse files
tobias-tenglertobias-tengler
authored
[Fusion] Add executor warmup (#8750)
1 parent 5f9ea2e commit dccdd58

File tree

10 files changed

+638
-131
lines changed

10 files changed

+638
-131
lines changed

‎src/HotChocolate/Core/src/Execution.Abstractions/Execution/IRequestExecutorWarmupTask.cs‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
namespace HotChocolate.Execution;
2+
3+
/// <summary>
4+
/// Represents a task to be run on a <see cref="IRequestExecutor"/>
5+
/// before it's ready to handle requests.
6+
/// </summary>
7+
public interface IRequestExecutorWarmupTask
8+
{
9+
/// <summary>
10+
/// Warms up the <paramref name="executor"/>.
11+
/// </summary>
12+
Task WarmupAsync(IRequestExecutor executor, CancellationToken cancellationToken);
13+
}

‎src/HotChocolate/Fusion-vnext/src/Fusion.AspNetCore/DependencyInjection/AspNetCoreFusionGatewayBuilderExtensions.HttpRequestInterceptor.cs‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
using HotChocolate.AspNetCore;
2+
using HotChocolate.Fusion.Configuration;
3+
using Microsoft.Extensions.DependencyInjection.Extensions;
4+
5+
namespace Microsoft.Extensions.DependencyInjection;
6+
7+
public static partial class AspNetCoreFusionGatewayBuilderExtensions
8+
{
9+
public static IFusionGatewayBuilder AddHttpRequestInterceptor<T>(
10+
this IFusionGatewayBuilder builder)
11+
where T : IHttpRequestInterceptor, new()
12+
{
13+
ArgumentNullException.ThrowIfNull(builder);
14+
15+
return builder.ConfigureSchemaServices(
16+
(_, s) =>
17+
{
18+
s.RemoveAll<IHttpRequestInterceptor>();
19+
s.AddSingleton<IHttpRequestInterceptor>(new T());
20+
});
21+
}
22+
23+
public static IFusionGatewayBuilder AddHttpRequestInterceptor(
24+
this IFusionGatewayBuilder builder,
25+
Func<IServiceProvider, IHttpRequestInterceptor> factory)
26+
{
27+
ArgumentNullException.ThrowIfNull(builder);
28+
ArgumentNullException.ThrowIfNull(factory);
29+
30+
return builder.ConfigureSchemaServices(
31+
(_, s) =>
32+
{
33+
s.RemoveAll<IHttpRequestInterceptor>();
34+
s.AddSingleton(factory);
35+
});
36+
}
37+
}

‎src/HotChocolate/Fusion-vnext/src/Fusion.AspNetCore/DependencyInjection/FusionServerServiceCollectionExtensions.cs‎

Lines changed: 5 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,8 @@ public static IFusionGatewayBuilder AddGraphQLGatewayServer(
2525

2626
return services
2727
.AddGraphQLGateway(name)
28-
.AddGraphQLGatewayServerCore()
28+
.AddGraphQLGatewayServerCore(maxAllowedRequestSize)
29+
.AddStartupInitialization()
2930
.AddDefaultHttpRequestInterceptor()
3031
.AddSubscriptionServices();
3132
}
@@ -65,33 +66,12 @@ private static IFusionGatewayBuilder AddGraphQLGatewayServerCore(
6566
return builder;
6667
}
6768

68-
public static IFusionGatewayBuilder AddHttpRequestInterceptor<T>(
69+
private static IFusionGatewayBuilder AddStartupInitialization(
6970
this IFusionGatewayBuilder builder)
70-
where T : IHttpRequestInterceptor, new()
7171
{
72-
ArgumentNullException.ThrowIfNull(builder);
72+
builder.Services.AddHostedService<RequestExecutorWarmupService>();
7373

74-
return builder.ConfigureSchemaServices(
75-
(_, s) =>
76-
{
77-
s.RemoveAll<IHttpRequestInterceptor>();
78-
s.AddSingleton<IHttpRequestInterceptor>(new T());
79-
});
80-
}
81-
82-
public static IFusionGatewayBuilder AddHttpRequestInterceptor(
83-
this IFusionGatewayBuilder builder,
84-
Func<IServiceProvider, IHttpRequestInterceptor> factory)
85-
{
86-
ArgumentNullException.ThrowIfNull(builder);
87-
ArgumentNullException.ThrowIfNull(factory);
88-
89-
return builder.ConfigureSchemaServices(
90-
(_, s) =>
91-
{
92-
s.RemoveAll<IHttpRequestInterceptor>();
93-
s.AddSingleton(factory);
94-
});
74+
return builder;
9575
}
9676

9777
private static IFusionGatewayBuilder AddDefaultHttpRequestInterceptor(

‎src/HotChocolate/Fusion-vnext/src/Fusion.AspNetCore/HotChocolate.Fusion.AspNetCore.csproj‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,10 @@
1010
<ProjectReference Include="..\Fusion.Execution\HotChocolate.Fusion.Execution.csproj" />
1111
</ItemGroup>
1212

13+
<ItemGroup>
14+
<FrameworkReference Include="Microsoft.AspNetCore.App" />
15+
</ItemGroup>
16+
1317
<ItemGroup>
1418
<EmbeddedResource Update="Properties\FusionUtilitiesResources.resx">
1519
<Generator>ResXFileCodeGenerator</Generator>

‎src/HotChocolate/Fusion-vnext/src/Fusion.AspNetCore/RequestExecutorWarmupService.cs‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
using HotChocolate.Execution;
2+
using HotChocolate.Fusion.Configuration;
3+
using HotChocolate.Fusion.Execution;
4+
using Microsoft.Extensions.Hosting;
5+
using Microsoft.Extensions.Options;
6+
7+
namespace Microsoft.Extensions.DependencyInjection;
8+
9+
internal sealed class RequestExecutorWarmupService(
10+
IOptionsMonitor<FusionGatewaySetup> optionsMonitor,
11+
IRequestExecutorProvider provider) : IHostedService
12+
{
13+
public async Task StartAsync(CancellationToken cancellationToken)
14+
{
15+
var warmupTasks = new List<Task>();
16+
17+
foreach (var schemaName in provider.SchemaNames)
18+
{
19+
var setup = optionsMonitor.Get(schemaName);
20+
21+
var requestOptions = FusionRequestExecutorManager.CreateRequestOptions(setup);
22+
23+
if (!requestOptions.LazyInitialization)
24+
{
25+
var warmupTask = WarmupAsync(schemaName, cancellationToken);
26+
warmupTasks.Add(warmupTask);
27+
}
28+
}
29+
30+
await Task.WhenAll(warmupTasks).ConfigureAwait(false);
31+
}
32+
33+
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
34+
35+
private async Task WarmupAsync(string schemaName, CancellationToken cancellationToken)
36+
{
37+
await provider.GetExecutorAsync(schemaName, cancellationToken).ConfigureAwait(false);
38+
}
39+
}

‎src/HotChocolate/Fusion-vnext/src/Fusion.Execution/DependencyInjection/CoreFusionGatewayBuilderExtensions.Warmup.cs‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
using System.Diagnostics.CodeAnalysis;
2+
using HotChocolate.Execution;
3+
using HotChocolate.Fusion.Configuration;
4+
5+
namespace Microsoft.Extensions.DependencyInjection;
6+
7+
public static partial class CoreFusionGatewayBuilderExtensions
8+
{
9+
/// <summary>
10+
/// Adds a warmup task that will be executed on each newly created request executor.
11+
/// </summary>
12+
public static IFusionGatewayBuilder AddWarmupTask(
13+
this IFusionGatewayBuilder builder,
14+
Func<IRequestExecutor, CancellationToken, Task> warmupFunc)
15+
{
16+
ArgumentNullException.ThrowIfNull(builder);
17+
ArgumentNullException.ThrowIfNull(warmupFunc);
18+
19+
return builder.AddWarmupTask(new DelegateWarmupTask(warmupFunc));
20+
}
21+
22+
/// <summary>
23+
/// Adds a warmup task that will be executed on each newly created request executor.
24+
/// </summary>
25+
public static IFusionGatewayBuilder AddWarmupTask(
26+
this IFusionGatewayBuilder builder,
27+
IRequestExecutorWarmupTask warmupTask)
28+
{
29+
ArgumentNullException.ThrowIfNull(builder);
30+
ArgumentNullException.ThrowIfNull(warmupTask);
31+
32+
builder.ConfigureSchemaServices((_, sc) => sc.AddSingleton(warmupTask));
33+
34+
return builder;
35+
}
36+
37+
/// <summary>
38+
/// Adds a warmup task that will be executed on each newly created request executor.
39+
/// </summary>
40+
public static IFusionGatewayBuilder AddWarmupTask<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] T>(
41+
this IFusionGatewayBuilder builder)
42+
where T : class, IRequestExecutorWarmupTask
43+
{
44+
ArgumentNullException.ThrowIfNull(builder);
45+
46+
builder.ConfigureSchemaServices(
47+
static (_, sc) => sc.AddSingleton<IRequestExecutorWarmupTask, T>());
48+
49+
return builder;
50+
}
51+
52+
private sealed class DelegateWarmupTask(Func<IRequestExecutor, CancellationToken, Task> warmupFunc)
53+
: IRequestExecutorWarmupTask
54+
{
55+
public Task WarmupAsync(IRequestExecutor requestExecutor, CancellationToken cancellationToken)
56+
{
57+
return warmupFunc.Invoke(requestExecutor, cancellationToken);
58+
}
59+
}
60+
}

‎src/HotChocolate/Fusion-vnext/src/Fusion.Execution/Execution/FusionRequestExecutorManager.cs‎

Lines changed: 62 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ internal sealed class FusionRequestExecutorManager
3535
, IAsyncDisposable
3636
{
3737
private readonly object _lock = new();
38-
private readonly SemaphoreSlim _semaphore = new(1, 1);
38+
private readonly ConcurrentDictionary<string, SemaphoreSlim> _semaphoreBySchema = new();
3939
private readonly ConcurrentDictionary<string, RequestExecutorRegistration> _registry = [];
4040
private readonly IOptionsMonitor<FusionGatewaySetup> _optionsMonitor;
4141
private readonly IServiceProvider _applicationServices;
@@ -90,7 +90,8 @@ private async ValueTask<IRequestExecutor> GetOrCreateRequestExecutorAsync(
9090
string schemaName,
9191
CancellationToken cancellationToken)
9292
{
93-
await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
93+
var semaphore = GetSemaphoreForSchema(schemaName);
94+
await semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
9495

9596
try
9697
{
@@ -106,10 +107,32 @@ private async ValueTask<IRequestExecutor> GetOrCreateRequestExecutorAsync(
106107
}
107108
finally
108109
{
109-
_semaphore.Release();
110+
semaphore.Release();
110111
}
111112
}
112113

114+
private SemaphoreSlim GetSemaphoreForSchema(string schemaName)
115+
=> _semaphoreBySchema.GetOrAdd(schemaName, _ => new SemaphoreSlim(1, 1));
116+
117+
private async ValueTask EvictExecutorAsync(FusionRequestExecutor executor, CancellationToken cancellationToken)
118+
{
119+
await _executorEvents.WriteEvictedAsync(executor, cancellationToken);
120+
121+
EvictRequestExecutorAsync(executor).FireAndForget();
122+
}
123+
124+
private static async Task EvictRequestExecutorAsync(FusionRequestExecutor previousExecutor)
125+
{
126+
var evictionTimeout = previousExecutor.Schema.Features
127+
.GetRequired<FusionRequestOptions>().EvictionTimeout;
128+
129+
// we will give the request executor some grace period to finish all requests
130+
// in the pipeline.
131+
await Task.Delay(evictionTimeout).ConfigureAwait(false);
132+
133+
await previousExecutor.DisposeAsync().ConfigureAwait(false);
134+
}
135+
113136
private async ValueTask<RequestExecutorRegistration> CreateInitialRegistrationAsync(
114137
string schemaName,
115138
CancellationToken cancellationToken)
@@ -119,10 +142,14 @@ private async ValueTask<RequestExecutorRegistration> CreateInitialRegistrationAs
119142
var (configuration, documentProvider) =
120143
await GetSchemaDocumentAsync(setup.DocumentProvider, cancellationToken).ConfigureAwait(false);
121144

145+
var executor = CreateRequestExecutor(schemaName, configuration);
146+
147+
await WarmupExecutorAsync(executor, cancellationToken).ConfigureAwait(false);
148+
122149
return new RequestExecutorRegistration(
123150
this,
124151
documentProvider,
125-
CreateRequestExecutor(schemaName, configuration),
152+
executor,
126153
configuration);
127154
}
128155

@@ -149,6 +176,15 @@ private FusionRequestExecutor CreateRequestExecutor(
149176
return executor;
150177
}
151178

179+
private async Task WarmupExecutorAsync(IRequestExecutor executor, CancellationToken cancellationToken)
180+
{
181+
var warmupTasks = executor.Schema.Services.GetServices<IRequestExecutorWarmupTask>();
182+
foreach (var warmupTask in warmupTasks)
183+
{
184+
await warmupTask.WarmupAsync(executor, cancellationToken).ConfigureAwait(false);
185+
}
186+
}
187+
152188
private async ValueTask<(FusionConfiguration, IFusionConfigurationProvider)> GetSchemaDocumentAsync(
153189
Func<IServiceProvider, IFusionConfigurationProvider>? documentProviderFactory,
154190
CancellationToken cancellationToken)
@@ -165,7 +201,7 @@ private FusionRequestExecutor CreateRequestExecutor(
165201
return (await documentPromise.Task.ConfigureAwait(false), documentProvider);
166202
}
167203

168-
private static FusionRequestOptions CreateRequestOptions(FusionGatewaySetup setup)
204+
internal static FusionRequestOptions CreateRequestOptions(FusionGatewaySetup setup)
169205
{
170206
var options = new FusionRequestOptions();
171207

@@ -174,15 +210,7 @@ private static FusionRequestOptions CreateRequestOptions(FusionGatewaySetup setu
174210
configure.Invoke(options);
175211
}
176212

177-
if (options.OperationExecutionPlanCacheSize < 16)
178-
{
179-
options.OperationExecutionPlanCacheSize = 16;
180-
}
181-
182-
if (options.OperationDocumentCacheSize < 16)
183-
{
184-
options.OperationDocumentCacheSize = 16;
185-
}
213+
options.MakeReadOnly();
186214

187215
return options;
188216
}
@@ -490,6 +518,13 @@ public async ValueTask DisposeAsync()
490518
session.OnCompleted();
491519
}
492520

521+
foreach (var semaphore in _semaphoreBySchema.Values)
522+
{
523+
semaphore.Dispose();
524+
}
525+
526+
_semaphoreBySchema.Clear();
527+
493528
_observers = [];
494529
}
495530

@@ -555,7 +590,7 @@ private async Task WaitForUpdatesAsync()
555590
break;
556591
}
557592

558-
var documentHash = XxHash64.HashToUInt64(Encoding.UTF8.GetBytes(configuration.ToString()));
593+
var documentHash = XxHash64.HashToUInt64(Encoding.UTF8.GetBytes(configuration.Schema.ToString()));
559594
var settingsHash = XxHash64.HashToUInt64(GetRawUtf8Value(configuration.Settings.Document.RootElement));
560595

561596
if (documentHash == _documentHash && settingsHash == _settingsHash)
@@ -569,12 +604,12 @@ private async Task WaitForUpdatesAsync()
569604
var previousExecutor = Executor;
570605
var nextExecutor = _manager.CreateRequestExecutor(Executor.Schema.Name, configuration);
571606

572-
// TODO : should we have the warmup tasks here?
607+
await _manager.WarmupExecutorAsync(nextExecutor, _cancellationToken).ConfigureAwait(false);
573608

574609
Executor = nextExecutor;
575610

576-
// we need to free the resources of the current schema as well as for the configuration object.
577-
await previousExecutor.DisposeAsync().ConfigureAwait(false);
611+
await _manager.EvictExecutorAsync(previousExecutor, _cancellationToken);
612+
578613
configuration.Dispose();
579614
}
580615
}
@@ -718,4 +753,13 @@ public static async ValueTask WriteCreatedAsync(
718753
var eventArgs = RequestExecutorEvent.Created(executor);
719754
await executorEvents.Writer.WriteAsync(eventArgs, cancellationToken).ConfigureAwait(false);
720755
}
756+
757+
public static async ValueTask WriteEvictedAsync(
758+
this Channel<RequestExecutorEvent> executorEvents,
759+
FusionRequestExecutor executor,
760+
CancellationToken cancellationToken)
761+
{
762+
var eventArgs = RequestExecutorEvent.Evicted(executor);
763+
await executorEvents.Writer.WriteAsync(eventArgs, cancellationToken).ConfigureAwait(false);
764+
}
721765
}

0 commit comments

Comments
 (0)