Skip to content

Commit d007fb5

Browse files
Wraith2Wraith2
committed
Add ValueTask stream overloads on SNI streams
1 parent b5d7bb6 commit d007fb5

8 files changed

Lines changed: 308 additions & 105 deletions

File tree

‎src/Microsoft.Data.SqlClient/netcore/src/Microsoft.Data.SqlClient.csproj‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -279,7 +279,9 @@
279279
<Compile Include="Microsoft\Data\SqlClient\SqlDiagnosticListener.NetStandard.cs" />
280280
<Compile Include="Microsoft\Data\SqlClient\SqlDelegatedTransaction.NetStandard.cs" />
281281
<Compile Include="Microsoft\Data\SqlClient\TdsParser.NetStandard.cs" />
282+
<Compile Include="Microsoft\Data\SqlClient\SNI\ConcurrentQueueSemaphore.NetStandard.cs" />
282283
<Compile Include="Microsoft\Data\SqlClient\SNI\SslOverTdsStream.NetStandard.cs" />
284+
<Compile Include="Microsoft\Data\SqlClient\SNI\SNIStreams.NetStandard.cs" />
283285
</ItemGroup>
284286
<ItemGroup Condition="'$(OSGroup)' != 'AnyOS' AND '$(TargetFramework)' != 'netstandard2.0'">
285287
<Compile Include="..\..\src\Microsoft\Data\SqlClient\AlwaysEncryptedAttestationException.cs">
@@ -309,6 +311,8 @@
309311
<Compile Include="Microsoft\Data\SqlClient\SqlDelegatedTransaction.NetCoreApp.cs" />
310312
<Compile Include="Microsoft\Data\SqlClient\TdsParser.NetCoreApp.cs" />
311313
<Compile Include="Microsoft\Data\SqlClient\SNI\SslOverTdsStream.NetCoreApp.cs" />
314+
<Compile Include="Microsoft\Data\SqlClient\SNI\ConcurrentQueueSemaphore.NetCoreApp.cs" />
315+
<Compile Include="Microsoft\Data\SqlClient\SNI\SNIStreams.NetCoreApp.cs" />
312316
</ItemGroup>
313317
<ItemGroup Condition="'$(OSGroup)' != 'AnyOS' AND '$(TargetGroup)' == 'netcoreapp' AND '$(BuildSimulator)' == 'true'">
314318
<Compile Include="Microsoft\Data\SqlClient\SimulatorEnclaveProvider.NetCoreApp.cs" />
@@ -436,6 +440,7 @@
436440
<Compile Include="$(CommonPath)\CoreLib\System\Threading\Tasks\TaskToApm.cs">
437441
<Link>Common\CoreLib\System\Threading\Tasks\TaskToApm.cs</Link>
438442
</Compile>
443+
<Compile Include="Microsoft\Data\SqlClient\SNI\ConcurrentQueueSemaphore.cs" />
439444
<Compile Include="Microsoft\Data\SqlClient\SNI\SNIError.cs" />
440445
<Compile Include="Microsoft\Data\SqlClient\SNI\SNIHandle.cs" />
441446
<Compile Include="Microsoft\Data\SqlClient\SNI\SNILoadHandle.cs" />
@@ -447,8 +452,8 @@
447452
<Compile Include="Microsoft\Data\SqlClient\SNI\SNIPacketPool.cs" />
448453
<Compile Include="Microsoft\Data\SqlClient\SNI\SNIPhysicalHandle.cs" />
449454
<Compile Include="Microsoft\Data\SqlClient\SNI\SNIProxy.cs" />
450-
<Compile Include="Microsoft\Data\SqlClient\SNI\SNITcpHandle.cs" />
451455
<Compile Include="Microsoft\Data\SqlClient\SNI\SNIStreams.cs" />
456+
<Compile Include="Microsoft\Data\SqlClient\SNI\SNITcpHandle.cs" />
452457
<Compile Include="Microsoft\Data\SqlClient\SNI\SslOverTdsStream.cs" />
453458
<Compile Include="Microsoft\Data\SqlClient\SNI\SNICommon.cs" />
454459
<Compile Include="Microsoft\Data\SqlClient\SNI\SspiClientContextStatus.cs" />
@@ -778,7 +783,7 @@
778783
<PackageReference Include="System.Buffers" Version="$(SystemBuffersVersion)" />
779784
<PackageReference Include="System.Runtime.Caching" Version="$(SystemRuntimeCachingVersion)" />
780785
<PackageReference Include="System.Security.Cryptography.Cng" Version="$(SystemSecurityCryptographyCngVersion)" />
781-
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="$(MicrosoftSourceLinkGitHubVersion)" PrivateAssets="All"/>
786+
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="$(MicrosoftSourceLinkGitHubVersion)" PrivateAssets="All" />
782787
</ItemGroup>
783788
<ItemGroup>
784789
<EmbeddedResource Update="Resources\Strings.resx">

‎src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SNI/ConcurrentQueueSemaphore.NetCoreApp.cs‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
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+
// See the LICENSE file in the project root for more information.
4+
5+
using System.Threading;
6+
using System.Threading.Tasks;
7+
8+
namespace Microsoft.Data.SqlClient.SNI
9+
{
10+
internal sealed partial class ConcurrentQueueSemaphore
11+
{
12+
public ValueTask WaitAsync(CancellationToken cancellationToken)
13+
{
14+
// try sync wait with 0 which will not block to see if we need to do an async wait
15+
if (_semaphore.Wait(0, cancellationToken))
16+
{
17+
return new ValueTask();
18+
}
19+
else
20+
{
21+
var tcs = new TaskCompletionSource<bool>();
22+
_queue.Enqueue(tcs);
23+
_semaphore.WaitAsync().ContinueWith(
24+
continuationAction: s_continuePop,
25+
state: _queue,
26+
cancellationToken: cancellationToken
27+
);
28+
return new ValueTask(tcs.Task);
29+
}
30+
}
31+
}
32+
}

‎src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SNI/ConcurrentQueueSemaphore.NetStandard.cs‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
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+
// See the LICENSE file in the project root for more information.
4+
5+
using System.Threading;
6+
using System.Threading.Tasks;
7+
8+
namespace Microsoft.Data.SqlClient.SNI
9+
{
10+
internal sealed partial class ConcurrentQueueSemaphore
11+
{
12+
public Task WaitAsync(CancellationToken cancellationToken)
13+
{
14+
var tcs = new TaskCompletionSource<bool>();
15+
_queue.Enqueue(tcs);
16+
_semaphore.WaitAsync().ContinueWith(
17+
continuationAction: s_continuePop,
18+
state: _queue,
19+
cancellationToken: cancellationToken
20+
);
21+
return tcs.Task;
22+
}
23+
}
24+
}

‎src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SNI/ConcurrentQueueSemaphore.cs‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
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+
// See the LICENSE file in the project root for more information.
4+
5+
using System;
6+
using System.Collections.Concurrent;
7+
using System.Threading;
8+
using System.Threading.Tasks;
9+
10+
namespace Microsoft.Data.SqlClient.SNI
11+
{
12+
/// <summary>
13+
/// This class implements a FIFO Queue with SemaphoreSlim for ordered execution of parallel tasks.
14+
/// Currently used in Managed SNI (SNISslStream) to override SslStream's WriteAsync implementation.
15+
/// </summary>
16+
internal sealed partial class ConcurrentQueueSemaphore
17+
{
18+
private static readonly Action<Task, object> s_continuePop = ContinuePop;
19+
20+
private readonly SemaphoreSlim _semaphore;
21+
private readonly ConcurrentQueue<TaskCompletionSource<bool>> _queue =
22+
new ConcurrentQueue<TaskCompletionSource<bool>>();
23+
24+
public ConcurrentQueueSemaphore(int initialCount)
25+
{
26+
_semaphore = new SemaphoreSlim(initialCount);
27+
}
28+
29+
public void Release()
30+
{
31+
_semaphore.Release();
32+
}
33+
34+
private static void ContinuePop(Task task, object state)
35+
{
36+
ConcurrentQueue<TaskCompletionSource<bool>> queue = (ConcurrentQueue<TaskCompletionSource<bool>>)state;
37+
if (queue.TryDequeue(out TaskCompletionSource<bool> popped))
38+
{
39+
popped.SetResult(true);
40+
}
41+
}
42+
}
43+
44+
}

‎src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SNI/SNIStreams.NetCoreApp.cs‎

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
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+
// See the LICENSE file in the project root for more information.
4+
5+
using System.Net.Security;
6+
using System.IO;
7+
using System.Threading;
8+
using System.Threading.Tasks;
9+
using System.Net.Sockets;
10+
using System;
11+
12+
namespace Microsoft.Data.SqlClient.SNI
13+
{
14+
15+
internal sealed partial class SNISslStream
16+
{
17+
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
18+
{
19+
ValueTask<int> valueTask = ReadAsync(new Memory<byte>(buffer, offset, count), cancellationToken);
20+
if (valueTask.IsCompletedSuccessfully)
21+
{
22+
return Task.FromResult(valueTask.Result);
23+
}
24+
else
25+
{
26+
return valueTask.AsTask();
27+
}
28+
}
29+
30+
public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
31+
{
32+
await _readAsyncSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
33+
try
34+
{
35+
return await base.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
36+
}
37+
finally
38+
{
39+
_readAsyncSemaphore.Release();
40+
}
41+
}
42+
43+
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
44+
{
45+
ValueTask valueTask = WriteAsync(new Memory<byte>(buffer, offset, count), cancellationToken);
46+
if (valueTask.IsCompletedSuccessfully)
47+
{
48+
return Task.CompletedTask;
49+
}
50+
else
51+
{
52+
return valueTask.AsTask();
53+
}
54+
}
55+
56+
public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
57+
{
58+
await _writeAsyncSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
59+
try
60+
{
61+
await base.WriteAsync(buffer, cancellationToken).ConfigureAwait(false);
62+
}
63+
finally
64+
{
65+
_writeAsyncSemaphore.Release();
66+
}
67+
}
68+
}
69+
70+
71+
internal sealed partial class SNINetworkStream
72+
{
73+
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
74+
{
75+
ValueTask<int> valueTask = ReadAsync(new Memory<byte>(buffer, offset, count), cancellationToken);
76+
if (valueTask.IsCompletedSuccessfully)
77+
{
78+
return Task.FromResult(valueTask.Result);
79+
}
80+
else
81+
{
82+
return valueTask.AsTask();
83+
}
84+
}
85+
86+
public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
87+
{
88+
await _readAsyncSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
89+
try
90+
{
91+
return await base.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
92+
}
93+
finally
94+
{
95+
_readAsyncSemaphore.Release();
96+
}
97+
}
98+
99+
// Prevent the WriteAsync collisions by running the task in a Semaphore Slim
100+
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
101+
{
102+
ValueTask valueTask = WriteAsync(new Memory<byte>(buffer, offset, count), cancellationToken);
103+
if (valueTask.IsCompletedSuccessfully)
104+
{
105+
return Task.CompletedTask;
106+
}
107+
else
108+
{
109+
return valueTask.AsTask();
110+
}
111+
}
112+
113+
public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
114+
{
115+
await _writeAsyncSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
116+
try
117+
{
118+
await base.WriteAsync(buffer, cancellationToken).ConfigureAwait(false);
119+
}
120+
finally
121+
{
122+
_writeAsyncSemaphore.Release();
123+
}
124+
}
125+
}
126+
}

‎src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SNI/SNIStreams.NetStandard.cs‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
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+
// See the LICENSE file in the project root for more information.
4+
5+
using System.Net.Security;
6+
using System.IO;
7+
using System.Threading;
8+
using System.Threading.Tasks;
9+
using System.Net.Sockets;
10+
11+
namespace Microsoft.Data.SqlClient.SNI
12+
{
13+
internal sealed partial class SNISslStream
14+
{
15+
public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
16+
{
17+
await _readAsyncSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
18+
try
19+
{
20+
return await base.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
21+
}
22+
finally
23+
{
24+
_readAsyncSemaphore.Release();
25+
}
26+
}
27+
28+
public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
29+
{
30+
await _writeAsyncSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
31+
try
32+
{
33+
await base.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
34+
}
35+
finally
36+
{
37+
_writeAsyncSemaphore.Release();
38+
}
39+
}
40+
}
41+
42+
internal sealed partial class SNINetworkStream
43+
{
44+
45+
public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
46+
{
47+
await _readAsyncSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
48+
try
49+
{
50+
return await base.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
51+
}
52+
finally
53+
{
54+
_readAsyncSemaphore.Release();
55+
}
56+
}
57+
58+
// Prevent the WriteAsync collisions by running the task in a Semaphore Slim
59+
public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
60+
{
61+
await _writeAsyncSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
62+
try
63+
{
64+
await base.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
65+
}
66+
finally
67+
{
68+
_writeAsyncSemaphore.Release();
69+
}
70+
}
71+
}
72+
}

0 commit comments

Comments
 (0)