Skip to content
This repository was archived by the owner on Dec 18, 2018. It is now read-only.

Commit afe944c

Browse files
benaadamsbenaadams
committed
Simplify SocketInput, remove locks, only use pooled blocks
1 parent 4f8ec86 commit afe944c

File tree

5 files changed

+97
-130
lines changed

5 files changed

+97
-130
lines changed

‎src/Microsoft.AspNet.Server.Kestrel/Filter/SocketInputStream.cs

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -74,11 +74,7 @@ public override void SetLength(long value)
7474

7575
public override void Write(byte[] buffer, int offset, int count)
7676
{
77-
var inputBuffer = _socketInput.IncomingStart(count);
78-
79-
Buffer.BlockCopy(buffer, offset, inputBuffer.Data.Array, inputBuffer.Data.Offset, count);
80-
81-
_socketInput.IncomingComplete(count, error: null);
77+
_socketInput.IncomingData(buffer, offset, count);
8278
}
8379

8480
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken token)
@@ -90,7 +86,7 @@ public override Task WriteAsync(byte[] buffer, int offset, int count, Cancellati
9086
protected override void Dispose(bool disposing)
9187
{
9288
// Close _socketInput with a fake zero-length write that will result in a zero-length read.
93-
_socketInput.IncomingComplete(0, error: null);
89+
_socketInput.IncomingData(null, 0, 0);
9490
base.Dispose(disposing);
9591
}
9692
}

‎src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -140,11 +140,11 @@ private static Libuv.uv_buf_t AllocCallback(UvStreamHandle handle, int suggested
140140

141141
private Libuv.uv_buf_t OnAlloc(UvStreamHandle handle, int suggestedSize)
142142
{
143-
var result = _rawSocketInput.IncomingStart(2048);
143+
var result = _rawSocketInput.IncomingStart();
144144

145145
return handle.Libuv.buf_init(
146-
result.DataPtr,
147-
result.Data.Count);
146+
result.Pin() + result.End,
147+
result.Data.Offset + result.Data.Count - result.End);
148148
}
149149

150150
private static void ReadCallback(UvStreamHandle handle, int status, object state)

‎src/Microsoft.AspNet.Server.Kestrel/Http/SocketInput.cs

Lines changed: 89 additions & 113 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
using System.IO;
66
using System.Runtime.CompilerServices;
77
using System.Threading;
8-
using System.Threading.Tasks;
98
using Microsoft.AspNet.Server.Kestrel.Infrastructure;
109

1110
namespace Microsoft.AspNet.Server.Kestrel.Http
@@ -25,7 +24,6 @@ public class SocketInput : ICriticalNotifyCompletion
2524
private MemoryPoolBlock2 _head;
2625
private MemoryPoolBlock2 _tail;
2726
private MemoryPoolBlock2 _pinned;
28-
private readonly object _sync = new Object();
2927

3028
public SocketInput(MemoryPool2 memory, IThreadPool threadPool)
3129
{
@@ -34,98 +32,104 @@ public SocketInput(MemoryPool2 memory, IThreadPool threadPool)
3432
_awaitableState = _awaitableIsNotCompleted;
3533
}
3634

37-
public ArraySegment<byte> Buffer { get; set; }
38-
3935
public bool RemoteIntakeFin { get; set; }
4036

41-
public bool IsCompleted
37+
public bool IsCompleted => (_awaitableState == _awaitableIsCompleted);
38+
39+
public MemoryPoolBlock2 IncomingStart()
4240
{
43-
get
41+
const int minimumSize = 2048;
42+
43+
if (_tail != null && minimumSize <= _tail.Data.Offset + _tail.Data.Count - _tail.End)
4444
{
45-
return Equals(_awaitableState, _awaitableIsCompleted);
45+
_pinned = _tail;
46+
}
47+
else
48+
{
49+
_pinned = _memory.Lease();
4650
}
47-
}
48-
49-
public void Skip(int count)
50-
{
51-
Buffer = new ArraySegment<byte>(Buffer.Array, Buffer.Offset + count, Buffer.Count - count);
52-
}
5351

54-
public ArraySegment<byte> Take(int count)
55-
{
56-
var taken = new ArraySegment<byte>(Buffer.Array, Buffer.Offset, count);
57-
Skip(count);
58-
return taken;
52+
return _pinned;
5953
}
6054

61-
public IncomingBuffer IncomingStart(int minimumSize)
55+
public void IncomingData(byte[] buffer, int offset, int count)
6256
{
63-
lock (_sync)
57+
if (count > 0)
6458
{
65-
if (_tail != null && minimumSize <= _tail.Data.Offset + _tail.Data.Count - _tail.End)
59+
if (_tail == null)
6660
{
67-
_pinned = _tail;
68-
var data = new ArraySegment<byte>(_pinned.Data.Array, _pinned.End, _pinned.Data.Offset + _pinned.Data.Count - _pinned.End);
69-
var dataPtr = _pinned.Pin() + _pinned.End;
70-
return new IncomingBuffer
71-
{
72-
Data = data,
73-
DataPtr = dataPtr,
74-
};
61+
_tail = _memory.Lease();
62+
}
63+
64+
var iterator = new MemoryPoolIterator2(_tail, _tail.End);
65+
iterator.CopyFrom(buffer, offset, count);
66+
67+
if (_head == null)
68+
{
69+
_head = _tail;
7570
}
76-
}
7771

78-
_pinned = _memory.Lease(minimumSize);
79-
return new IncomingBuffer
72+
_tail = iterator.Block;
73+
}
74+
else
8075
{
81-
Data = _pinned.Data,
82-
DataPtr = _pinned.Pin() + _pinned.End
83-
};
76+
RemoteIntakeFin = true;
77+
}
78+
79+
Complete();
8480
}
8581

8682
public void IncomingComplete(int count, Exception error)
8783
{
88-
Action awaitableState;
89-
90-
lock (_sync)
84+
// Unpin may called without an earlier Pin
85+
if (_pinned != null)
9186
{
92-
// Unpin may called without an earlier Pin
93-
if (_pinned != null)
87+
88+
_pinned.End += count;
89+
90+
if (_head == null)
9491
{
95-
_pinned.Unpin();
96-
97-
_pinned.End += count;
98-
if (_head == null)
99-
{
100-
_head = _tail = _pinned;
101-
}
102-
else if (_tail == _pinned)
103-
{
104-
// NO-OP: this was a read into unoccupied tail-space
105-
}
106-
else
107-
{
108-
_tail.Next = _pinned;
109-
_tail = _pinned;
110-
}
92+
_head = _tail = _pinned;
11193
}
112-
_pinned = null;
113-
114-
if (count == 0)
94+
else if (_tail == _pinned)
11595
{
116-
RemoteIntakeFin = true;
96+
// NO-OP: this was a read into unoccupied tail-space
11797
}
118-
if (error != null)
98+
else
11999
{
120-
_awaitableError = error;
100+
_tail.Next = _pinned;
101+
_tail = _pinned;
121102
}
122103

123-
awaitableState = Interlocked.Exchange(
124-
ref _awaitableState,
125-
_awaitableIsCompleted);
104+
_pinned = null;
105+
}
126106

127-
_manualResetEvent.Set();
107+
if (count == 0)
108+
{
109+
RemoteIntakeFin = true;
128110
}
111+
if (error != null)
112+
{
113+
_awaitableError = error;
114+
}
115+
116+
Complete();
117+
}
118+
119+
public void AbortAwaiting()
120+
{
121+
_awaitableError = new ObjectDisposedException(nameof(SocketInput), "The request was aborted");
122+
123+
Complete();
124+
}
125+
126+
private void Complete()
127+
{
128+
var awaitableState = Interlocked.Exchange(
129+
ref _awaitableState,
130+
_awaitableIsCompleted);
131+
132+
_manualResetEvent.Set();
129133

130134
if (awaitableState != _awaitableIsCompleted &&
131135
awaitableState != _awaitableIsNotCompleted)
@@ -136,10 +140,7 @@ public void IncomingComplete(int count, Exception error)
136140

137141
public MemoryPoolIterator2 ConsumingStart()
138142
{
139-
lock (_sync)
140-
{
141-
return new MemoryPoolIterator2(_head);
142-
}
143+
return new MemoryPoolIterator2(_head);
143144
}
144145

145146
public void ConsumingComplete(
@@ -148,50 +149,31 @@ public void ConsumingComplete(
148149
{
149150
MemoryPoolBlock2 returnStart = null;
150151
MemoryPoolBlock2 returnEnd = null;
151-
lock (_sync)
152+
if (!consumed.IsDefault)
152153
{
153-
if (!consumed.IsDefault)
154-
{
155-
returnStart = _head;
156-
returnEnd = consumed.Block;
157-
_head = consumed.Block;
158-
_head.Start = consumed.Index;
159-
}
160-
if (!examined.IsDefault &&
161-
examined.IsEnd &&
162-
RemoteIntakeFin == false &&
163-
_awaitableError == null)
164-
{
165-
_manualResetEvent.Reset();
154+
returnStart = _head;
155+
returnEnd = consumed.Block;
156+
_head = consumed.Block;
157+
_head.Start = consumed.Index;
158+
}
159+
if (!examined.IsDefault &&
160+
examined.IsEnd &&
161+
RemoteIntakeFin == false &&
162+
_awaitableError == null)
163+
{
164+
_manualResetEvent.Reset();
166165

167-
var awaitableState = Interlocked.CompareExchange(
168-
ref _awaitableState,
169-
_awaitableIsNotCompleted,
170-
_awaitableIsCompleted);
171-
}
166+
var awaitableState = Interlocked.CompareExchange(
167+
ref _awaitableState,
168+
_awaitableIsNotCompleted,
169+
_awaitableIsCompleted);
172170
}
171+
173172
while (returnStart != returnEnd)
174173
{
175174
var returnBlock = returnStart;
176175
returnStart = returnStart.Next;
177-
returnBlock.Pool?.Return(returnBlock);
178-
}
179-
}
180-
181-
public void AbortAwaiting()
182-
{
183-
_awaitableError = new ObjectDisposedException(nameof(SocketInput), "The request was aborted");
184-
185-
var awaitableState = Interlocked.Exchange(
186-
ref _awaitableState,
187-
_awaitableIsCompleted);
188-
189-
_manualResetEvent.Set();
190-
191-
if (awaitableState != _awaitableIsCompleted &&
192-
awaitableState != _awaitableIsNotCompleted)
193-
{
194-
_threadPool.Run(awaitableState);
176+
returnBlock.Pool.Return(returnBlock);
195177
}
196178
}
197179

@@ -247,11 +229,5 @@ public void GetResult()
247229
throw new IOException(error.Message, error);
248230
}
249231
}
250-
251-
public struct IncomingBuffer
252-
{
253-
public ArraySegment<byte> Data;
254-
public IntPtr DataPtr;
255-
}
256232
}
257233
}

‎test/Microsoft.AspNet.Server.KestrelTests/FrameTests.cs

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -55,9 +55,7 @@ public void EmptyHeaderValuesCanBeParsed(string rawHeaders, int numHeaders)
5555
var headerCollection = new FrameRequestHeaders();
5656

5757
var headerArray = Encoding.ASCII.GetBytes(rawHeaders);
58-
var inputBuffer = socketInput.IncomingStart(headerArray.Length);
59-
Buffer.BlockCopy(headerArray, 0, inputBuffer.Data.Array, inputBuffer.Data.Offset, headerArray.Length);
60-
socketInput.IncomingComplete(headerArray.Length, null);
58+
socketInput.IncomingData(headerArray, 0, headerArray.Length);
6159

6260
var success = Frame.TakeMessageHeaders(socketInput, headerCollection);
6361

‎test/Microsoft.AspNet.Server.KestrelTests/TestInput.cs

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -29,11 +29,8 @@ public TestInput()
2929

3030
public void Add(string text, bool fin = false)
3131
{
32-
var encoding = System.Text.Encoding.ASCII;
33-
var count = encoding.GetByteCount(text);
34-
var buffer = FrameContext.SocketInput.IncomingStart(text.Length);
35-
count = encoding.GetBytes(text, 0, text.Length, buffer.Data.Array, buffer.Data.Offset);
36-
FrameContext.SocketInput.IncomingComplete(count, null);
32+
var data = System.Text.Encoding.ASCII.GetBytes(text);
33+
FrameContext.SocketInput.IncomingData(data, 0, data.Length);
3734
if (fin)
3835
{
3936
FrameContext.SocketInput.RemoteIntakeFin = true;

0 commit comments

Comments
 (0)