2
2
// The .NET Foundation licenses this file to you under the MIT license.
3
3
4
4
using System . Buffers ;
5
+ using System . Diagnostics ;
5
6
using System . IO . Pipelines ;
6
7
using System . IO . Pipes ;
7
8
using System . Net ;
8
9
using System . Threading . Channels ;
9
10
using Microsoft . AspNetCore . Connections ;
10
11
using Microsoft . Extensions . Logging ;
12
+ using NamedPipeOptions = System . IO . Pipes . PipeOptions ;
11
13
using PipeOptions = System . IO . Pipelines . PipeOptions ;
12
14
13
15
namespace Microsoft . AspNetCore . Server . Kestrel . Transport . NamedPipes . Internal ;
@@ -20,21 +22,24 @@ internal sealed class NamedPipeConnectionListener : IConnectionListener
20
22
private readonly CancellationTokenSource _listeningTokenSource = new CancellationTokenSource ( ) ;
21
23
private readonly CancellationToken _listeningToken ;
22
24
private readonly Channel < ConnectionContext > _acceptedQueue ;
23
- private readonly Task _listeningTask ;
24
25
private readonly MemoryPool < byte > _memoryPool ;
25
26
private readonly PipeOptions _inputOptions ;
26
27
private readonly PipeOptions _outputOptions ;
28
+ private readonly Mutex _mutex ;
29
+ private Task ? _listeningTask ;
27
30
private int _disposed ;
28
31
29
32
public NamedPipeConnectionListener (
30
33
NamedPipeEndPoint endpoint ,
31
34
NamedPipeTransportOptions options ,
32
- ILoggerFactory loggerFactory )
35
+ ILoggerFactory loggerFactory ,
36
+ Mutex mutex )
33
37
{
34
38
_log = loggerFactory . CreateLogger ( "Microsoft.AspNetCore.Server.Kestrel.Transport.NamedPipes" ) ;
35
39
_endpoint = endpoint ;
36
40
_options = options ;
37
- _acceptedQueue = Channel . CreateBounded < ConnectionContext > ( new BoundedChannelOptions ( options . Backlog ) ) ;
41
+ _mutex = mutex ;
42
+ _acceptedQueue = Channel . CreateBounded < ConnectionContext > ( new BoundedChannelOptions ( options . Backlog ) { SingleWriter = true } ) ;
38
43
_memoryPool = options . MemoryPoolFactory ( ) ;
39
44
_listeningToken = _listeningTokenSource . Token ;
40
45
@@ -43,50 +48,57 @@ public NamedPipeConnectionListener(
43
48
44
49
_inputOptions = new PipeOptions ( _memoryPool , PipeScheduler . ThreadPool , PipeScheduler . Inline , maxReadBufferSize , maxReadBufferSize / 2 , useSynchronizationContext : false ) ;
45
50
_outputOptions = new PipeOptions ( _memoryPool , PipeScheduler . Inline , PipeScheduler . ThreadPool , maxWriteBufferSize , maxWriteBufferSize / 2 , useSynchronizationContext : false ) ;
51
+ }
52
+
53
+ public void Start ( )
54
+ {
55
+ Debug . Assert ( _listeningTask == null , "Already started" ) ;
46
56
47
- // Start after all fields are initialized.
48
- _listeningTask = StartAsync ( ) ;
57
+ // Start first stream inline to catch creation errors.
58
+ var initialStream = CreateServerStream ( ) ;
59
+
60
+ _listeningTask = StartAsync ( initialStream ) ;
49
61
}
50
62
51
63
public EndPoint EndPoint => _endpoint ;
52
64
53
- private async Task StartAsync ( )
65
+ private async Task StartAsync ( NamedPipeServerStream nextStream )
54
66
{
55
67
try
56
68
{
57
69
while ( true )
58
70
{
59
- NamedPipeServerStream stream ;
60
-
61
71
try
62
72
{
63
- _listeningToken . ThrowIfCancellationRequested ( ) ;
64
-
65
- stream = NamedPipeServerStreamAcl . Create (
66
- _endpoint . PipeName ,
67
- PipeDirection . InOut ,
68
- NamedPipeServerStream . MaxAllowedServerInstances ,
69
- PipeTransmissionMode . Byte ,
70
- _endpoint . PipeOptions ,
71
- inBufferSize : 0 , // Buffer in System.IO.Pipelines
72
- outBufferSize : 0 , // Buffer in System.IO.Pipelines
73
- _options . PipeSecurity ) ;
73
+ var stream = nextStream ;
74
74
75
75
await stream . WaitForConnectionAsync ( _listeningToken ) ;
76
+
77
+ var connection = new NamedPipeConnection ( stream , _endpoint , _log , _memoryPool , _inputOptions , _outputOptions ) ;
78
+ connection . Start ( ) ;
79
+
80
+ // Create the next stream before writing connected stream to the channel.
81
+ // This ensures there is always a created stream and another process can't
82
+ // create a stream with the same name with different a access policy.
83
+ nextStream = CreateServerStream ( ) ;
84
+
85
+ while ( ! _acceptedQueue . Writer . TryWrite ( connection ) )
86
+ {
87
+ if ( ! await _acceptedQueue . Writer . WaitToWriteAsync ( _listeningToken ) )
88
+ {
89
+ throw new InvalidOperationException ( "Accept queue writer was unexpectedly closed." ) ;
90
+ }
91
+ }
76
92
}
77
93
catch ( OperationCanceledException ex ) when ( _listeningToken . IsCancellationRequested )
78
94
{
79
95
// Cancelled the current token
80
96
NamedPipeLog . ConnectionListenerAborted ( _log , ex ) ;
81
97
break ;
82
98
}
83
-
84
- var connection = new NamedPipeConnection ( stream , _endpoint , _log , _memoryPool , _inputOptions , _outputOptions ) ;
85
- connection . Start ( ) ;
86
-
87
- _acceptedQueue . Writer . TryWrite ( connection ) ;
88
99
}
89
100
101
+ nextStream . Dispose ( ) ;
90
102
_acceptedQueue . Writer . TryComplete ( ) ;
91
103
}
92
104
catch ( Exception ex )
@@ -95,6 +107,41 @@ private async Task StartAsync()
95
107
}
96
108
}
97
109
110
+ private NamedPipeServerStream CreateServerStream ( )
111
+ {
112
+ NamedPipeServerStream stream ;
113
+ var pipeOptions = NamedPipeOptions . Asynchronous | NamedPipeOptions . WriteThrough ;
114
+ if ( _options . CurrentUserOnly )
115
+ {
116
+ pipeOptions |= NamedPipeOptions . CurrentUserOnly ;
117
+ }
118
+
119
+ if ( _options . PipeSecurity != null )
120
+ {
121
+ stream = NamedPipeServerStreamAcl . Create (
122
+ _endpoint . PipeName ,
123
+ PipeDirection . InOut ,
124
+ NamedPipeServerStream . MaxAllowedServerInstances ,
125
+ PipeTransmissionMode . Byte ,
126
+ pipeOptions ,
127
+ inBufferSize : 0 , // Buffer in System.IO.Pipelines
128
+ outBufferSize : 0 , // Buffer in System.IO.Pipelines
129
+ _options . PipeSecurity ) ;
130
+ }
131
+ else
132
+ {
133
+ stream = new NamedPipeServerStream (
134
+ _endpoint . PipeName ,
135
+ PipeDirection . InOut ,
136
+ NamedPipeServerStream . MaxAllowedServerInstances ,
137
+ PipeTransmissionMode . Byte ,
138
+ pipeOptions ,
139
+ inBufferSize : 0 ,
140
+ outBufferSize : 0 ) ;
141
+ }
142
+ return stream ;
143
+ }
144
+
98
145
public async ValueTask < ConnectionContext ? > AcceptAsync ( CancellationToken cancellationToken = default )
99
146
{
100
147
while ( await _acceptedQueue . Reader . WaitToReadAsync ( cancellationToken ) )
@@ -109,6 +156,8 @@ private async Task StartAsync()
109
156
return null ;
110
157
}
111
158
159
+ public ValueTask UnbindAsync ( CancellationToken cancellationToken = default ) => DisposeAsync ( ) ;
160
+
112
161
public async ValueTask DisposeAsync ( )
113
162
{
114
163
// A stream may be waiting on WaitForConnectionAsync when dispose happens.
@@ -119,12 +168,10 @@ public async ValueTask DisposeAsync()
119
168
}
120
169
121
170
_listeningTokenSource . Dispose ( ) ;
122
- await _listeningTask ;
123
- }
124
-
125
- public async ValueTask UnbindAsync ( CancellationToken cancellationToken = default )
126
- {
127
- _listeningTokenSource . Cancel ( ) ;
128
- await _listeningTask ;
171
+ _mutex . Dispose ( ) ;
172
+ if ( _listeningTask != null )
173
+ {
174
+ await _listeningTask ;
175
+ }
129
176
}
130
177
}
0 commit comments