Skip to content

Commit 0696727

Browse files
teo-tsirpanisteo-tsirpanis
authored
Efficient RandomAccess async I/O on the thread pool. (#55123)
* Factor cross-platform SafeFileHandle code in a common file. There isn't much actually. * Write a reusable IValueTaskSource to queue async-over-sync RandomAccess I/O on the thread pool. And use it in the RandomAccess.ScheduleSync methods instead of wrapping Task.Factory.StartNew in a ValueTask. * Reduce ThreadPoolValueTaskSource's field count. * Rename SafeFileHandle.ValueTaskSource to OverlappedValueTaskSource. * Address most PR feedback. * Run continuations synchronously and ensure the task's result is set only once. * Address more PR feedback. * Flow the ExecutionContext in ThreadPoolValueTaskSource. * Set the ExecutionContext to null afterwards. * Use separate fields for the two vectored I/O operations, avoiding a cast.
1 parent 607f98c commit 0696727

10 files changed

+279
-65
lines changed

‎src/libraries/System.Private.CoreLib/src/Microsoft/Win32/SafeHandles/SafeFileHandle.ValueTaskSource.Windows.cs renamed to ‎src/libraries/System.Private.CoreLib/src/Microsoft/Win32/SafeHandles/SafeFileHandle.OverlappedValueTaskSource.Windows.cs

Lines changed: 14 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -12,33 +12,34 @@ namespace Microsoft.Win32.SafeHandles
1212
{
1313
public sealed partial class SafeFileHandle : SafeHandleZeroOrMinusOneIsInvalid
1414
{
15-
private ValueTaskSource? _reusableValueTaskSource; // reusable ValueTaskSource that is currently NOT being used
15+
private OverlappedValueTaskSource? _reusableOverlappedValueTaskSource; // reusable OverlappedValueTaskSource that is currently NOT being used
1616

17-
// Rent the reusable ValueTaskSource, or create a new one to use if we couldn't get one (which
18-
// should only happen on first use or if the FileStream is being used concurrently).
19-
internal ValueTaskSource GetValueTaskSource() => Interlocked.Exchange(ref _reusableValueTaskSource, null) ?? new ValueTaskSource(this);
17+
// Rent the reusable OverlappedValueTaskSource, or create a new one to use if we couldn't get one (which
18+
// should only happen on first use or if the SafeFileHandle is being used concurrently).
19+
internal OverlappedValueTaskSource GetOverlappedValueTaskSource() =>
20+
Interlocked.Exchange(ref _reusableOverlappedValueTaskSource, null) ?? new OverlappedValueTaskSource(this);
2021

2122
protected override bool ReleaseHandle()
2223
{
2324
bool result = Interop.Kernel32.CloseHandle(handle);
2425

25-
Interlocked.Exchange(ref _reusableValueTaskSource, null)?.Dispose();
26+
Interlocked.Exchange(ref _reusableOverlappedValueTaskSource, null)?.Dispose();
2627

2728
return result;
2829
}
2930

30-
private void TryToReuse(ValueTaskSource source)
31+
private void TryToReuse(OverlappedValueTaskSource source)
3132
{
3233
source._source.Reset();
3334

34-
if (Interlocked.CompareExchange(ref _reusableValueTaskSource, source, null) is not null)
35+
if (Interlocked.CompareExchange(ref _reusableOverlappedValueTaskSource, source, null) is not null)
3536
{
3637
source._preallocatedOverlapped.Dispose();
3738
}
3839
}
3940

40-
/// <summary>Reusable IValueTaskSource for FileStream ValueTask-returning async operations.</summary>
41-
internal sealed unsafe class ValueTaskSource : IValueTaskSource<int>, IValueTaskSource
41+
/// <summary>Reusable IValueTaskSource for RandomAccess async operations based on Overlapped I/O.</summary>
42+
internal sealed unsafe class OverlappedValueTaskSource : IValueTaskSource<int>, IValueTaskSource
4243
{
4344
internal static readonly IOCompletionCallback s_ioCallback = IOCallback;
4445

@@ -55,7 +56,7 @@ internal sealed unsafe class ValueTaskSource : IValueTaskSource<int>, IValueTask
5556
/// </summary>
5657
internal ulong _result;
5758

58-
internal ValueTaskSource(SafeFileHandle fileHandle)
59+
internal OverlappedValueTaskSource(SafeFileHandle fileHandle)
5960
{
6061
_fileHandle = fileHandle;
6162
_source.RunContinuationsAsynchronously = true;
@@ -112,7 +113,7 @@ internal void RegisterForCancellation(CancellationToken cancellationToken)
112113
{
113114
_cancellationRegistration = cancellationToken.UnsafeRegister(static (s, token) =>
114115
{
115-
ValueTaskSource vts = (ValueTaskSource)s!;
116+
OverlappedValueTaskSource vts = (OverlappedValueTaskSource)s!;
116117
if (!vts._fileHandle.IsInvalid)
117118
{
118119
try
@@ -156,7 +157,7 @@ internal void ReleaseResources()
156157
// done by that cancellation registration, e.g. unregistering. As such, we use _result to both track who's
157158
// responsible for calling Complete and for passing the necessary data between parties.
158159

159-
/// <summary>Invoked when AsyncWindowsFileStreamStrategy has finished scheduling the async operation.</summary>
160+
/// <summary>Invoked when the async operation finished being scheduled.</summary>
160161
internal void FinishedScheduling()
161162
{
162163
// Set the value to 1. If it was already non-0, then the asynchronous operation already completed but
@@ -172,7 +173,7 @@ internal void FinishedScheduling()
172173
/// <summary>Invoked when the asynchronous operation has completed asynchronously.</summary>
173174
private static void IOCallback(uint errorCode, uint numBytes, NativeOverlapped* pOverlapped)
174175
{
175-
ValueTaskSource? vts = (ValueTaskSource?)ThreadPoolBoundHandle.GetNativeOverlappedState(pOverlapped);
176+
OverlappedValueTaskSource? vts = (OverlappedValueTaskSource?)ThreadPoolBoundHandle.GetNativeOverlappedState(pOverlapped);
176177
Debug.Assert(vts is not null);
177178
Debug.Assert(vts._overlapped == pOverlapped, "Overlaps don't match");
178179

‎src/libraries/System.Private.CoreLib/src/Microsoft/Win32/SafeHandles/SafeFileHandle.ThreadPoolValueTaskSource.cs

Lines changed: 220 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,220 @@
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;
5+
using System.Collections.Generic;
6+
using System.Diagnostics;
7+
using System.IO;
8+
using System.Runtime.InteropServices;
9+
using System.Threading;
10+
using System.Threading.Tasks;
11+
using System.Threading.Tasks.Sources;
12+
13+
namespace Microsoft.Win32.SafeHandles
14+
{
15+
public sealed partial class SafeFileHandle : SafeHandleZeroOrMinusOneIsInvalid
16+
{
17+
private ThreadPoolValueTaskSource? _reusableThreadPoolValueTaskSource; // reusable ThreadPoolValueTaskSource that is currently NOT being used
18+
19+
// Rent the reusable ThreadPoolValueTaskSource, or create a new one to use if we couldn't get one (which
20+
// should only happen on first use or if the SafeFileHandle is being used concurrently).
21+
internal ThreadPoolValueTaskSource GetThreadPoolValueTaskSource() =>
22+
Interlocked.Exchange(ref _reusableThreadPoolValueTaskSource, null) ?? new ThreadPoolValueTaskSource(this);
23+
24+
private void TryToReuse(ThreadPoolValueTaskSource source)
25+
{
26+
Interlocked.CompareExchange(ref _reusableThreadPoolValueTaskSource, source, null);
27+
}
28+
29+
/// <summary>
30+
/// A reusable <see cref="IValueTaskSource"/> implementation that
31+
/// queues asynchronous <see cref="RandomAccess"/> operations to
32+
/// be completed synchronously on the thread pool.
33+
/// </summary>
34+
internal sealed class ThreadPoolValueTaskSource : IThreadPoolWorkItem, IValueTaskSource<int>, IValueTaskSource<long>
35+
{
36+
private readonly SafeFileHandle _fileHandle;
37+
private ManualResetValueTaskSourceCore<long> _source;
38+
private Operation _operation = Operation.None;
39+
private ExecutionContext? _context;
40+
41+
// These fields store the parameters for the operation.
42+
// The first two are common for all kinds of operations.
43+
private long _fileOffset;
44+
private CancellationToken _cancellationToken;
45+
// Used by simple reads and writes. Will be unsafely cast to a memory when performing a read.
46+
private ReadOnlyMemory<byte> _singleSegment;
47+
private IReadOnlyList<Memory<byte>>? _readScatterBuffers;
48+
private IReadOnlyList<ReadOnlyMemory<byte>>? _writeGatherBuffers;
49+
50+
internal ThreadPoolValueTaskSource(SafeFileHandle fileHandle)
51+
{
52+
_fileHandle = fileHandle;
53+
}
54+
55+
[Conditional("DEBUG")]
56+
private void ValidateInvariants()
57+
{
58+
Operation op = _operation;
59+
Debug.Assert(op == Operation.None, $"An operation was queued before the previous {op}'s completion.");
60+
}
61+
62+
private long GetResultAndRelease(short token)
63+
{
64+
try
65+
{
66+
return _source.GetResult(token);
67+
}
68+
finally
69+
{
70+
_source.Reset();
71+
_fileHandle.TryToReuse(this);
72+
}
73+
}
74+
75+
public ValueTaskSourceStatus GetStatus(short token) => _source.GetStatus(token);
76+
public void OnCompleted(Action<object?> continuation, object? state, short token, ValueTaskSourceOnCompletedFlags flags) =>
77+
_source.OnCompleted(continuation, state, token, flags);
78+
int IValueTaskSource<int>.GetResult(short token) => (int) GetResultAndRelease(token);
79+
long IValueTaskSource<long>.GetResult(short token) => GetResultAndRelease(token);
80+
81+
private void ExecuteInternal()
82+
{
83+
Debug.Assert(_operation >= Operation.Read && _operation <= Operation.WriteGather);
84+
85+
long result = 0;
86+
Exception? exception = null;
87+
try
88+
{
89+
// This is the operation's last chance to be canceled.
90+
if (_cancellationToken.IsCancellationRequested)
91+
{
92+
exception = new OperationCanceledException(_cancellationToken);
93+
}
94+
else
95+
{
96+
switch (_operation)
97+
{
98+
case Operation.Read:
99+
Memory<byte> writableSingleSegment = MemoryMarshal.AsMemory(_singleSegment);
100+
result = RandomAccess.ReadAtOffset(_fileHandle, writableSingleSegment.Span, _fileOffset);
101+
break;
102+
case Operation.Write:
103+
result = RandomAccess.WriteAtOffset(_fileHandle, _singleSegment.Span, _fileOffset);
104+
break;
105+
case Operation.ReadScatter:
106+
Debug.Assert(_readScatterBuffers != null);
107+
result = RandomAccess.ReadScatterAtOffset(_fileHandle, _readScatterBuffers, _fileOffset);
108+
break;
109+
case Operation.WriteGather:
110+
Debug.Assert(_writeGatherBuffers != null);
111+
result = RandomAccess.WriteGatherAtOffset(_fileHandle, _writeGatherBuffers, _fileOffset);
112+
break;
113+
}
114+
}
115+
}
116+
catch (Exception e)
117+
{
118+
exception = e;
119+
}
120+
finally
121+
{
122+
_operation = Operation.None;
123+
_context = null;
124+
_cancellationToken = default;
125+
_singleSegment = default;
126+
_readScatterBuffers = null;
127+
_writeGatherBuffers = null;
128+
}
129+
130+
if (exception == null)
131+
{
132+
_source.SetResult(result);
133+
}
134+
else
135+
{
136+
_source.SetException(exception);
137+
}
138+
}
139+
140+
void IThreadPoolWorkItem.Execute()
141+
{
142+
if (_context == null || _context.IsDefault)
143+
{
144+
ExecuteInternal();
145+
}
146+
else
147+
{
148+
ExecutionContext.RunForThreadPoolUnsafe(_context, static x => x.ExecuteInternal(), this);
149+
}
150+
}
151+
152+
private void QueueToThreadPool()
153+
{
154+
_context = ExecutionContext.Capture();
155+
ThreadPool.UnsafeQueueUserWorkItem(this, preferLocal: true);
156+
}
157+
158+
public ValueTask<int> QueueRead(Memory<byte> buffer, long fileOffset, CancellationToken cancellationToken)
159+
{
160+
ValidateInvariants();
161+
162+
_operation = Operation.Read;
163+
_singleSegment = buffer;
164+
_fileOffset = fileOffset;
165+
_cancellationToken = cancellationToken;
166+
QueueToThreadPool();
167+
168+
return new ValueTask<int>(this, _source.Version);
169+
}
170+
171+
public ValueTask<int> QueueWrite(ReadOnlyMemory<byte> buffer, long fileOffset, CancellationToken cancellationToken)
172+
{
173+
ValidateInvariants();
174+
175+
_operation = Operation.Write;
176+
_singleSegment = buffer;
177+
_fileOffset = fileOffset;
178+
_cancellationToken = cancellationToken;
179+
QueueToThreadPool();
180+
181+
return new ValueTask<int>(this, _source.Version);
182+
}
183+
184+
public ValueTask<long> QueueReadScatter(IReadOnlyList<Memory<byte>> buffers, long fileOffset, CancellationToken cancellationToken)
185+
{
186+
ValidateInvariants();
187+
188+
_operation = Operation.ReadScatter;
189+
_readScatterBuffers = buffers;
190+
_fileOffset = fileOffset;
191+
_cancellationToken = cancellationToken;
192+
QueueToThreadPool();
193+
194+
return new ValueTask<long>(this, _source.Version);
195+
}
196+
197+
public ValueTask<long> QueueWriteGather(IReadOnlyList<ReadOnlyMemory<byte>> buffers, long fileOffset, CancellationToken cancellationToken)
198+
{
199+
ValidateInvariants();
200+
201+
_operation = Operation.WriteGather;
202+
_writeGatherBuffers = buffers;
203+
_fileOffset = fileOffset;
204+
_cancellationToken = cancellationToken;
205+
QueueToThreadPool();
206+
207+
return new ValueTask<long>(this, _source.Version);
208+
}
209+
210+
private enum Operation : byte
211+
{
212+
None,
213+
Read,
214+
Write,
215+
ReadScatter,
216+
WriteGather
217+
}
218+
}
219+
}
220+
}

‎src/libraries/System.Private.CoreLib/src/Microsoft/Win32/SafeHandles/SafeFileHandle.Unix.cs

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88

99
namespace Microsoft.Win32.SafeHandles
1010
{
11-
public sealed class SafeFileHandle : SafeHandleZeroOrMinusOneIsInvalid
11+
public sealed partial class SafeFileHandle : SafeHandleZeroOrMinusOneIsInvalid
1212
{
1313
// not using bool? as it's not thread safe
1414
private volatile NullableBool _canSeek = NullableBool.Undefined;
@@ -23,11 +23,6 @@ private SafeFileHandle(bool ownsHandle)
2323
SetHandle(new IntPtr(-1));
2424
}
2525

26-
public SafeFileHandle(IntPtr preexistingHandle, bool ownsHandle) : this(ownsHandle)
27-
{
28-
SetHandle(preexistingHandle);
29-
}
30-
3126
public bool IsAsync { get; private set; }
3227

3328
internal bool CanSeek => !IsClosed && GetCanSeek();

‎src/libraries/System.Private.CoreLib/src/Microsoft/Win32/SafeHandles/SafeFileHandle.Windows.cs

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,6 @@ public SafeFileHandle() : base(true)
2020
{
2121
}
2222

23-
public SafeFileHandle(IntPtr preexistingHandle, bool ownsHandle) : base(ownsHandle)
24-
{
25-
SetHandle(preexistingHandle);
26-
}
27-
2823
public bool IsAsync => (GetFileOptions() & FileOptions.Asynchronous) != 0;
2924

3025
internal bool CanSeek => !IsClosed && GetFileType() == Interop.Kernel32.FileTypes.FILE_TYPE_DISK;

‎src/libraries/System.Private.CoreLib/src/Microsoft/Win32/SafeHandles/SafeFileHandle.cs

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
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;
5+
6+
namespace Microsoft.Win32.SafeHandles
7+
{
8+
public sealed partial class SafeFileHandle : SafeHandleZeroOrMinusOneIsInvalid
9+
{
10+
public SafeFileHandle(IntPtr preexistingHandle, bool ownsHandle) : base(ownsHandle)
11+
{
12+
SetHandle(preexistingHandle);
13+
}
14+
}
15+
}

‎src/libraries/System.Private.CoreLib/src/System.Private.CoreLib.Shared.projitems

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,8 @@
6161
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\CriticalHandleZeroOrMinusOneIsInvalid.cs" />
6262
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeHandleMinusOneIsInvalid.cs" />
6363
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeHandleZeroOrMinusOneIsInvalid.cs" />
64+
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeFileHandle.cs" />
65+
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeFileHandle.ThreadPoolValueTaskSource.cs" />
6466
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeWaitHandle.cs" />
6567
<Compile Include="$(MSBuildThisFileDirectory)System\AccessViolationException.cs" />
6668
<Compile Include="$(MSBuildThisFileDirectory)System\Action.cs" />
@@ -1517,7 +1519,7 @@
15171519
<Compile Include="$(CommonPath)Interop\Windows\Kernel32\Interop.GetModuleFileName.cs">
15181520
<Link>Common\Interop\Windows\Kernel32\Interop.GetModuleFileName.cs</Link>
15191521
</Compile>
1520-
<Compile Include="$(CommonPath)Interop\Windows\Kernel32\Interop.GetOverlappedResult.cs">
1522+
<Compile Include="$(CommonPath)Interop\Windows\Kernel32\Interop.GetOverlappedResult.cs">
15211523
<Link>Common\Interop\Windows\Kernel32\Interop.GetOverlappedResult.cs</Link>
15221524
</Compile>
15231525
<Compile Include="$(CommonPath)Interop\Windows\Kernel32\Interop.GetProcessMemoryInfo.cs">
@@ -1775,7 +1777,7 @@
17751777
<Compile Include="$(MSBuildThisFileDirectory)Internal\Console.Windows.cs" />
17761778
<Compile Include="$(MSBuildThisFileDirectory)Internal\Win32\RegistryKey.cs" />
17771779
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeFileHandle.Windows.cs" />
1778-
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeFileHandle.ValueTaskSource.Windows.cs" />
1780+
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeFileHandle.OverlappedValueTaskSource.Windows.cs" />
17791781
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeFindHandle.Windows.cs" />
17801782
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeRegistryHandle.cs" />
17811783
<Compile Include="$(MSBuildThisFileDirectory)Microsoft\Win32\SafeHandles\SafeRegistryHandle.Windows.cs" />

0 commit comments

Comments
 (0)