Skip to content

Commit ff51fd7

Browse files
HaoKHaoK
authored
Add ISocketConnectionContextFactory (#34769)
1 parent c7fa6c1 commit ff51fd7

File tree

9 files changed

+231
-103
lines changed

9 files changed

+231
-103
lines changed

‎src/Servers/Kestrel/Transport.Sockets/src/Internal/SocketConnection.cs‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,10 @@
11
// Licensed to the .NET Foundation under one or more agreements.
22
// The .NET Foundation licenses this file to you under the MIT license.
33

4-
using System;
54
using System.Buffers;
65
using System.Diagnostics;
76
using System.IO.Pipelines;
87
using System.Net.Sockets;
9-
using System.Threading;
10-
using System.Threading.Tasks;
118
using Microsoft.AspNetCore.Connections;
129
using Microsoft.Extensions.Logging;
1310

‎src/Servers/Kestrel/Transport.Sockets/src/PublicAPI.Unshipped.txt‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,26 @@
33
*REMOVED*Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportFactory.SocketTransportFactory(Microsoft.Extensions.Options.IOptions<Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportOptions> options, Microsoft.Extensions.Logging.ILoggerFactory loggerFactory) -> void
44
*REMOVED*static Microsoft.AspNetCore.Hosting.WebHostBuilderSocketExtensions.UseSockets(this Microsoft.AspNetCore.Hosting.IWebHostBuilder hostBuilder) -> Microsoft.AspNetCore.Hosting.IWebHostBuilder
55
*REMOVED*static Microsoft.AspNetCore.Hosting.WebHostBuilderSocketExtensions.UseSockets(this Microsoft.AspNetCore.Hosting.IWebHostBuilder hostBuilder, System.Action<Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportOptions> configureOptions) -> Microsoft.AspNetCore.Hosting.IWebHostBuilder
6+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionContextFactory
7+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionContextFactory.Create(System.Net.Sockets.Socket! socket) -> Microsoft.AspNetCore.Connections.ConnectionContext!
8+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionContextFactory.Dispose() -> void
9+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionContextFactory.SocketConnectionContextFactory(Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions! options, Microsoft.Extensions.Logging.ILogger! logger) -> void
10+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions
11+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.IOQueueCount.get -> int
12+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.IOQueueCount.set -> void
13+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.MaxReadBufferSize.get -> long?
14+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.MaxReadBufferSize.set -> void
15+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.MaxWriteBufferSize.get -> long?
16+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.MaxWriteBufferSize.set -> void
17+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.SocketConnectionFactoryOptions() -> void
18+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.UnsafePreferInlineScheduling.get -> bool
19+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.UnsafePreferInlineScheduling.set -> void
20+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.WaitForDataBeforeAllocatingBuffer.get -> bool
21+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactoryOptions.WaitForDataBeforeAllocatingBuffer.set -> void
622
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportFactory.BindAsync(System.Net.EndPoint! endpoint, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask<Microsoft.AspNetCore.Connections.IConnectionListener!>
723
~Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportFactory.SocketTransportFactory(Microsoft.Extensions.Options.IOptions<Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportOptions!>! options, Microsoft.Extensions.Logging.ILoggerFactory! loggerFactory) -> void
824
static Microsoft.AspNetCore.Hosting.WebHostBuilderSocketExtensions.UseSockets(this Microsoft.AspNetCore.Hosting.IWebHostBuilder! hostBuilder) -> Microsoft.AspNetCore.Hosting.IWebHostBuilder!
925
static Microsoft.AspNetCore.Hosting.WebHostBuilderSocketExtensions.UseSockets(this Microsoft.AspNetCore.Hosting.IWebHostBuilder! hostBuilder, System.Action<Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportOptions!>! configureOptions) -> Microsoft.AspNetCore.Hosting.IWebHostBuilder!
1026
static Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportOptions.CreateDefaultBoundListenSocket(System.Net.EndPoint! endpoint) -> System.Net.Sockets.Socket!
1127
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportOptions.CreateBoundListenSocket.get -> System.Func<System.Net.EndPoint!, System.Net.Sockets.Socket!>!
12-
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportOptions.CreateBoundListenSocket.set -> void
28+
Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketTransportOptions.CreateBoundListenSocket.set -> void

‎src/Servers/Kestrel/Transport.Sockets/src/SocketConnectionContextFactory.cs‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
// Licensed to the .NET Foundation under one or more agreements.
2+
// The .NET Foundation licenses this file to you under the MIT license.
3+
4+
using System.Buffers;
5+
using System.IO.Pipelines;
6+
using System.Net.Sockets;
7+
using Microsoft.AspNetCore.Connections;
8+
using Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal;
9+
using Microsoft.Extensions.Logging;
10+
11+
namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
12+
{
13+
/// <summary>
14+
/// A factory for socket based connections contexts.
15+
/// </summary>
16+
public sealed class SocketConnectionContextFactory : IDisposable
17+
{
18+
private readonly MemoryPool<byte> _memoryPool;
19+
private readonly SocketConnectionFactoryOptions _options;
20+
private readonly ISocketsTrace _trace;
21+
private readonly int _settingsCount;
22+
private readonly QueueSettings[] _settings;
23+
private int _settingsIndex;
24+
25+
/// <summary>
26+
/// Creates the <see cref="SocketConnectionContextFactory"/>.
27+
/// </summary>
28+
/// <param name="options">The options.</param>
29+
/// <param name="logger">The logger.</param>
30+
public SocketConnectionContextFactory(SocketConnectionFactoryOptions options, ILogger logger)
31+
{
32+
if (options == null)
33+
{
34+
throw new ArgumentNullException(nameof(options));
35+
}
36+
37+
if (logger == null)
38+
{
39+
throw new ArgumentNullException(nameof(logger));
40+
}
41+
42+
_options = options;
43+
_trace = new SocketsTrace(logger);
44+
_memoryPool = _options.MemoryPoolFactory();
45+
_settingsCount = _options.IOQueueCount;
46+
47+
var maxReadBufferSize = _options.MaxReadBufferSize ?? 0;
48+
var maxWriteBufferSize = _options.MaxWriteBufferSize ?? 0;
49+
var applicationScheduler = options.UnsafePreferInlineScheduling ? PipeScheduler.Inline : PipeScheduler.ThreadPool;
50+
51+
if (_settingsCount > 0)
52+
{
53+
_settings = new QueueSettings[_settingsCount];
54+
55+
for (var i = 0; i < _settingsCount; i++)
56+
{
57+
var transportScheduler = options.UnsafePreferInlineScheduling ? PipeScheduler.Inline : new IOQueue();
58+
// https://github.com/aspnet/KestrelHttpServer/issues/2573
59+
var awaiterScheduler = OperatingSystem.IsWindows() ? transportScheduler : PipeScheduler.Inline;
60+
61+
_settings[i] = new QueueSettings()
62+
{
63+
Scheduler = transportScheduler,
64+
InputOptions = new PipeOptions(_memoryPool, applicationScheduler, transportScheduler, maxReadBufferSize, maxReadBufferSize / 2, useSynchronizationContext: false),
65+
OutputOptions = new PipeOptions(_memoryPool, transportScheduler, applicationScheduler, maxWriteBufferSize, maxWriteBufferSize / 2, useSynchronizationContext: false),
66+
SocketSenderPool = new SocketSenderPool(awaiterScheduler)
67+
};
68+
}
69+
}
70+
else
71+
{
72+
var transportScheduler = options.UnsafePreferInlineScheduling ? PipeScheduler.Inline : PipeScheduler.ThreadPool;
73+
// https://github.com/aspnet/KestrelHttpServer/issues/2573
74+
var awaiterScheduler = OperatingSystem.IsWindows() ? transportScheduler : PipeScheduler.Inline;
75+
_settings = new QueueSettings[]
76+
{
77+
new QueueSettings()
78+
{
79+
Scheduler = transportScheduler,
80+
InputOptions = new PipeOptions(_memoryPool, applicationScheduler, transportScheduler, maxReadBufferSize, maxReadBufferSize / 2, useSynchronizationContext: false),
81+
OutputOptions = new PipeOptions(_memoryPool, transportScheduler, applicationScheduler, maxWriteBufferSize, maxWriteBufferSize / 2, useSynchronizationContext: false),
82+
SocketSenderPool = new SocketSenderPool(awaiterScheduler)
83+
}
84+
};
85+
_settingsCount = 1;
86+
}
87+
}
88+
89+
/// <summary>
90+
/// Create a <see cref="ConnectionContext"/> for a socket.
91+
/// </summary>
92+
/// <param name="socket">The socket for the connection.</param>
93+
/// <returns></returns>
94+
public ConnectionContext Create(Socket socket)
95+
{
96+
var setting = _settings[Interlocked.Increment(ref _settingsIndex) % _settingsCount];
97+
98+
var connection = new SocketConnection(socket,
99+
_memoryPool,
100+
setting.Scheduler,
101+
_trace,
102+
setting.SocketSenderPool,
103+
setting.InputOptions,
104+
setting.OutputOptions,
105+
waitForData: _options.WaitForDataBeforeAllocatingBuffer);
106+
107+
connection.Start();
108+
return connection;
109+
}
110+
111+
/// <inheritdoc />
112+
public void Dispose()
113+
{
114+
// Dispose the memory pool
115+
_memoryPool.Dispose();
116+
117+
// Dispose any pooled senders
118+
foreach (var setting in _settings)
119+
{
120+
setting.SocketSenderPool.Dispose();
121+
}
122+
}
123+
124+
private class QueueSettings
125+
{
126+
public PipeScheduler Scheduler { get; init; } = default!;
127+
public PipeOptions InputOptions { get; init; } = default!;
128+
public PipeOptions OutputOptions { get; init; } = default!;
129+
public SocketSenderPool SocketSenderPool { get; init; } = default!;
130+
}
131+
}
132+
}

‎src/Servers/Kestrel/Transport.Sockets/src/SocketConnectionFactoryOptions.cs‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
// Licensed to the .NET Foundation under one or more agreements.
2+
// The .NET Foundation licenses this file to you under the MIT license.
3+
4+
using System.Buffers;
5+
6+
namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
7+
{
8+
/// <summary>
9+
/// Options for <see cref="SocketConnectionContextFactory"/>.
10+
/// </summary>
11+
public class SocketConnectionFactoryOptions
12+
{
13+
/// <summary>
14+
/// Create a new instance.
15+
/// </summary>
16+
public SocketConnectionFactoryOptions() { }
17+
18+
internal SocketConnectionFactoryOptions(SocketTransportOptions transportOptions)
19+
{
20+
IOQueueCount = transportOptions.IOQueueCount;
21+
WaitForDataBeforeAllocatingBuffer = transportOptions.WaitForDataBeforeAllocatingBuffer;
22+
MaxReadBufferSize = transportOptions.MaxReadBufferSize;
23+
MaxWriteBufferSize = transportOptions.MaxWriteBufferSize;
24+
UnsafePreferInlineScheduling = transportOptions.UnsafePreferInlineScheduling;
25+
MemoryPoolFactory = transportOptions.MemoryPoolFactory;
26+
}
27+
28+
/// <summary>
29+
/// The number of I/O queues used to process requests. Set to 0 to directly schedule I/O to the ThreadPool.
30+
/// </summary>
31+
/// <remarks>
32+
/// Defaults to <see cref="Environment.ProcessorCount" /> rounded down and clamped between 1 and 16.
33+
/// </remarks>
34+
public int IOQueueCount { get; set; } = Math.Min(Environment.ProcessorCount, 16);
35+
36+
/// <summary>
37+
/// Wait until there is data available to allocate a buffer. Setting this to false can increase throughput at the cost of increased memory usage.
38+
/// </summary>
39+
/// <remarks>
40+
/// Defaults to true.
41+
/// </remarks>
42+
public bool WaitForDataBeforeAllocatingBuffer { get; set; } = true;
43+
44+
/// <summary>
45+
/// Gets or sets the maximum unconsumed incoming bytes the transport will buffer.
46+
/// </summary>
47+
public long? MaxReadBufferSize { get; set; } = 1024 * 1024;
48+
49+
/// <summary>
50+
/// Gets or sets the maximum outgoing bytes the transport will buffer before applying write backpressure.
51+
/// </summary>
52+
public long? MaxWriteBufferSize { get; set; } = 64 * 1024;
53+
54+
/// <summary>
55+
/// Inline application and transport continuations instead of dispatching to the threadpool.
56+
/// </summary>
57+
/// <remarks>
58+
/// This will run application code on the IO thread which is why this is unsafe.
59+
/// It is recommended to set the DOTNET_SYSTEM_NET_SOCKETS_INLINE_COMPLETIONS environment variable to '1' when using this setting to also inline the completions
60+
/// at the runtime layer as well.
61+
/// This setting can make performance worse if there is expensive work that will end up holding onto the IO thread for longer than needed.
62+
/// Test to make sure this setting helps performance.
63+
/// </remarks>
64+
public bool UnsafePreferInlineScheduling { get; set; }
65+
66+
internal Func<MemoryPool<byte>> MemoryPoolFactory { get; set; } = PinnedBlockMemoryPoolFactory.Create;
67+
}
68+
}

‎src/Servers/Kestrel/Transport.Sockets/src/SocketConnectionListener.cs‎

Lines changed: 8 additions & 91 deletions
Original file line numberDiff line numberDiff line change
@@ -1,87 +1,34 @@
11
// Licensed to the .NET Foundation under one or more agreements.
22
// The .NET Foundation licenses this file to you under the MIT license.
33

4-
using System;
5-
using System.Buffers;
6-
using System.ComponentModel;
74
using System.Diagnostics;
8-
using System.IO.Pipelines;
95
using System.Net;
106
using System.Net.Sockets;
11-
using System.Threading;
12-
using System.Threading.Tasks;
137
using Microsoft.AspNetCore.Connections;
148
using Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal;
9+
using Microsoft.Extensions.Logging;
1510

1611
namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
1712
{
1813
internal sealed class SocketConnectionListener : IConnectionListener
1914
{
20-
private readonly MemoryPool<byte> _memoryPool;
21-
private readonly int _settingsCount;
22-
private readonly Settings[] _settings;
15+
private readonly SocketConnectionContextFactory _factory;
2316
private readonly ISocketsTrace _trace;
2417
private Socket? _listenSocket;
25-
private int _settingsIndex;
2618
private readonly SocketTransportOptions _options;
2719

2820
public EndPoint EndPoint { get; private set; }
2921

3022
internal SocketConnectionListener(
3123
EndPoint endpoint,
3224
SocketTransportOptions options,
33-
ISocketsTrace trace)
25+
ILoggerFactory loggerFactory)
3426
{
3527
EndPoint = endpoint;
36-
_trace = trace;
3728
_options = options;
38-
_memoryPool = _options.MemoryPoolFactory();
39-
var ioQueueCount = options.IOQueueCount;
40-
41-
var maxReadBufferSize = _options.MaxReadBufferSize ?? 0;
42-
var maxWriteBufferSize = _options.MaxWriteBufferSize ?? 0;
43-
var applicationScheduler = options.UnsafePreferInlineScheduling ? PipeScheduler.Inline : PipeScheduler.ThreadPool;
44-
45-
if (ioQueueCount > 0)
46-
{
47-
_settingsCount = ioQueueCount;
48-
_settings = new Settings[_settingsCount];
49-
50-
for (var i = 0; i < _settingsCount; i++)
51-
{
52-
var transportScheduler = options.UnsafePreferInlineScheduling ? PipeScheduler.Inline : new IOQueue();
53-
// https://github.com/aspnet/KestrelHttpServer/issues/2573
54-
var awaiterScheduler = OperatingSystem.IsWindows() ? transportScheduler : PipeScheduler.Inline;
55-
56-
_settings[i] = new Settings
57-
{
58-
Scheduler = transportScheduler,
59-
InputOptions = new PipeOptions(_memoryPool, applicationScheduler, transportScheduler, maxReadBufferSize, maxReadBufferSize / 2, useSynchronizationContext: false),
60-
OutputOptions = new PipeOptions(_memoryPool, transportScheduler, applicationScheduler, maxWriteBufferSize, maxWriteBufferSize / 2, useSynchronizationContext: false),
61-
SocketSenderPool = new SocketSenderPool(awaiterScheduler)
62-
};
63-
}
64-
}
65-
else
66-
{
67-
var transportScheduler = options.UnsafePreferInlineScheduling ? PipeScheduler.Inline : PipeScheduler.ThreadPool;
68-
// https://github.com/aspnet/KestrelHttpServer/issues/2573
69-
var awaiterScheduler = OperatingSystem.IsWindows() ? transportScheduler : PipeScheduler.Inline;
70-
71-
var directScheduler = new Settings[]
72-
{
73-
new Settings
74-
{
75-
Scheduler = transportScheduler,
76-
InputOptions = new PipeOptions(_memoryPool, applicationScheduler, transportScheduler, maxReadBufferSize, maxReadBufferSize / 2, useSynchronizationContext: false),
77-
OutputOptions = new PipeOptions(_memoryPool, transportScheduler, applicationScheduler, maxWriteBufferSize, maxWriteBufferSize / 2, useSynchronizationContext: false),
78-
SocketSenderPool = new SocketSenderPool(awaiterScheduler)
79-
}
80-
};
81-
82-
_settingsCount = directScheduler.Length;
83-
_settings = directScheduler;
84-
}
29+
var logger = loggerFactory.CreateLogger("Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets");
30+
_trace = new SocketsTrace(logger);
31+
_factory = new SocketConnectionContextFactory(new SocketConnectionFactoryOptions(options), logger);
8532
}
8633

8734
internal void Bind()
@@ -125,22 +72,7 @@ internal void Bind()
12572
acceptSocket.NoDelay = _options.NoDelay;
12673
}
12774

128-
var setting = _settings[_settingsIndex];
129-
130-
var connection = new SocketConnection(acceptSocket,
131-
_memoryPool,
132-
setting.Scheduler,
133-
_trace,
134-
setting.SocketSenderPool,
135-
setting.InputOptions,
136-
setting.OutputOptions,
137-
waitForData: _options.WaitForDataBeforeAllocatingBuffer);
138-
139-
connection.Start();
140-
141-
_settingsIndex = (_settingsIndex + 1) % _settingsCount;
142-
143-
return connection;
75+
return _factory.Create(acceptSocket);
14476
}
14577
catch (ObjectDisposedException)
14678
{
@@ -170,24 +102,9 @@ public ValueTask DisposeAsync()
170102
{
171103
_listenSocket?.Dispose();
172104

173-
// Dispose the memory pool
174-
_memoryPool.Dispose();
175-
176-
// Dispose any pooled senders
177-
foreach (var setting in _settings)
178-
{
179-
setting.SocketSenderPool.Dispose();
180-
}
105+
_factory.Dispose();
181106

182107
return default;
183108
}
184-
185-
private class Settings
186-
{
187-
public PipeScheduler Scheduler { get; init; } = default!;
188-
public PipeOptions InputOptions { get; init; } = default!;
189-
public PipeOptions OutputOptions { get; init; } = default!;
190-
public SocketSenderPool SocketSenderPool { get; init; } = default!;
191-
}
192109
}
193110
}

0 commit comments

Comments
 (0)