huahaofeng
/
aspnetcore
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Reuse ReadAsync for Consume

This commit is contained in:
Ben Adams 2015-11-15 22:27:39 +00:00
parent 1589b54018
commit ecc439555e
4 changed files with 20 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/Microsoft.AspNet.Server.Kestrel/Filter/LibuvStream.cs
View File

@ -85,7 +85,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Filter
private Task<int> ReadAsync(ArraySegment<byte> buffer) private Task<int> ReadAsync(ArraySegment<byte> buffer)
{ {
return _input.ReadAsync(buffer); return _input.ReadAsync(buffer.Array, buffer.Offset, buffer.Count);
} }
} }
} }

110
src/Microsoft.AspNet.Server.Kestrel/Http/MessageBody.cs
View File

@ -44,7 +44,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
var firstLoop = true; var firstLoop = true;
do do
{ {
result = SkipAsyncImplementation(cancellationToken); result = ReadAsyncImplementation(default(ArraySegment<byte>), cancellationToken);
if (!result.IsCompleted) if (!result.IsCompleted)
{ {
if (firstLoop && Interlocked.Exchange(ref _send100Continue, 0) == 1) if (firstLoop && Interlocked.Exchange(ref _send100Continue, 0) == 1)
@ -68,8 +68,6 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
public abstract Task<int> ReadAsyncImplementation(ArraySegment<byte> buffer, CancellationToken cancellationToken); public abstract Task<int> ReadAsyncImplementation(ArraySegment<byte> buffer, CancellationToken cancellationToken);
public abstract Task<int> SkipAsyncImplementation(CancellationToken cancellationToken);
public static MessageBody For( public static MessageBody For(
string httpVersion, string httpVersion,
IDictionary<string, StringValues> headers, IDictionary<string, StringValues> headers,
@ -138,11 +136,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
public override Task<int> ReadAsyncImplementation(ArraySegment<byte> buffer, CancellationToken cancellationToken) public override Task<int> ReadAsyncImplementation(ArraySegment<byte> buffer, CancellationToken cancellationToken)
{ {
return _context.SocketInput.ReadAsync(buffer); return _context.SocketInput.ReadAsync(buffer.Array, buffer.Offset, buffer.Array == null ? 8192 : buffer.Count);
}
public override Task<int> SkipAsyncImplementation(CancellationToken cancellationToken)
{
return _context.SocketInput.SkipAsync(4096);
} }
} }
@ -163,35 +157,13 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
{ {
var input = _context.SocketInput; var input = _context.SocketInput;
var limit = Math.Min(buffer.Count, _inputLength); var limit = buffer.Array == null ? _inputLength : Math.Min(buffer.Count, _inputLength);
if (limit == 0) if (limit == 0)
{ {
return 0; return 0;
} }
var limitedBuffer = new ArraySegment<byte>(buffer.Array, buffer.Offset, limit); var actual = await _context.SocketInput.ReadAsync(buffer.Array, buffer.Offset, limit);
var actual = await _context.SocketInput.ReadAsync(limitedBuffer);
_inputLength -= actual;
if (actual == 0)
{
throw new InvalidDataException("Unexpected end of request content");
}
return actual;
}
public override async Task<int> SkipAsyncImplementation(CancellationToken cancellationToken)
{
var input = _context.SocketInput;
var limit = Math.Min(4096, _inputLength);
if (limit == 0)
{
return 0;
}
var actual = await _context.SocketInput.SkipAsync(limit);
_inputLength -= actual; _inputLength -= actual;
if (actual == 0) if (actual == 0)
@ -244,7 +216,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
} }
while (_mode == Mode.ChunkData) while (_mode == Mode.ChunkData)
{ {
var limit = Math.Min(buffer.Count, _inputLength); var limit = buffer.Array == null ? _inputLength : Math.Min(buffer.Count, _inputLength);
if (limit != 0) if (limit != 0)
{ {
await input; await input;
@ -291,78 +263,6 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
return 0; return 0;
} }
public override async Task<int> SkipAsyncImplementation(CancellationToken cancellationToken)
{
var input = _context.SocketInput;
while (_mode != Mode.Complete)
{
while (_mode == Mode.ChunkPrefix)
{
var chunkSize = 0;
if (!TakeChunkedLine(input, ref chunkSize))
{
await input;
}
else if (chunkSize == 0)
{
_mode = Mode.Complete;
}
else
{
_mode = Mode.ChunkData;
}
_inputLength = chunkSize;
}
while (_mode == Mode.ChunkData)
{
var limit = Math.Min(4096, _inputLength);
if (limit != 0)
{
await input;
}
var begin = input.ConsumingStart();
int actual;
var end = begin.Skip(limit, out actual);
_inputLength -= actual;
input.ConsumingComplete(end, end);
if (_inputLength == 0)
{
_mode = Mode.ChunkSuffix;
}
if (actual != 0)
{
return actual;
}
}
while (_mode == Mode.ChunkSuffix)
{
var scan = input.ConsumingStart();
var consumed = scan;
var ch1 = scan.Take();
var ch2 = scan.Take();
if (ch1 == -1 || ch2 == -1)
{
input.ConsumingComplete(consumed, scan);
await input;
}
else if (ch1 == '\r' && ch2 == '\n')
{
input.ConsumingComplete(scan, scan);
_mode = Mode.ChunkPrefix;
}
else
{
throw new NotImplementedException("INVALID REQUEST FORMAT");
}
}
}
return 0;
}
private static bool TakeChunkedLine(SocketInput baton, ref int chunkSizeOut) private static bool TakeChunkedLine(SocketInput baton, ref int chunkSizeOut)
{ {
var scan = baton.ConsumingStart(); var scan = baton.ConsumingStart();

26
src/Microsoft.AspNet.Server.Kestrel/Http/SocketInputExtensions.cs
View File

@ -8,7 +8,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
{ {
public static class SocketInputExtensions public static class SocketInputExtensions
{ {
public static async Task<int> ReadAsync(this SocketInput input, ArraySegment<byte> buffer) public static async Task<int> ReadAsync(this SocketInput input, byte[] buffer, int offset, int count)
{ {
while (true) while (true)
{ {
@ -16,29 +16,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
var begin = input.ConsumingStart(); var begin = input.ConsumingStart();
int actual; int actual;
var end = begin.CopyTo(buffer.Array, buffer.Offset, buffer.Count, out actual); var end = begin.CopyTo(buffer, offset, count, out actual);
input.ConsumingComplete(end, end);
if (actual != 0)
{
return actual;
}
if (input.RemoteIntakeFin)
{
return 0;
}
}
}
public static async Task<int> SkipAsync(this SocketInput input, int limit)
{
while (true)
{
await input;
var begin = input.ConsumingStart();
int actual;
var end = begin.Skip(limit, out actual);
input.ConsumingComplete(end, end); input.ConsumingComplete(end, end);
if (actual != 0) if (actual != 0)

47
src/Microsoft.AspNet.Server.Kestrel/Infrastructure/MemoryPoolIterator2.cs
View File

@ -600,18 +600,27 @@ namespace Microsoft.AspNet.Server.Kestrel.Infrastructure
if (remaining <= following) if (remaining <= following)
{ {
actual = count; actual = count;
Buffer.BlockCopy(block.Array, index, array, offset, remaining); if (array != null)
{
Buffer.BlockCopy(block.Array, index, array, offset, remaining);
}
return new MemoryPoolIterator2(block, index + remaining); return new MemoryPoolIterator2(block, index + remaining);
} }
else if (block.Next == null) else if (block.Next == null)
{ {
actual = count - remaining + following; actual = count - remaining + following;
Buffer.BlockCopy(block.Array, index, array, offset, following); if (array != null)
{
Buffer.BlockCopy(block.Array, index, array, offset, following);
}
return new MemoryPoolIterator2(block, index + following); return new MemoryPoolIterator2(block, index + following);
} }
else else
{ {
Buffer.BlockCopy(block.Array, index, array, offset, following); if (array != null)
{
Buffer.BlockCopy(block.Array, index, array, offset, following);
}
offset += following; offset += following;
remaining -= following; remaining -= following;
block = block.Next; block = block.Next;
@ -619,37 +628,5 @@ namespace Microsoft.AspNet.Server.Kestrel.Infrastructure
} }
} }
} }
public MemoryPoolIterator2 Skip(int limit, out int actual)
{
if (IsDefault)
{
actual = 0;
return this;
}
var block = _block;
var index = _index;
var remaining = limit;
while (true)
{
var following = block.End - index;
if (remaining <= following)
{
actual = limit;
return new MemoryPoolIterator2(block, index + remaining);
}
else if (block.Next == null)
{
actual = limit - remaining + following;
return new MemoryPoolIterator2(block, index + following);
}
else
{
remaining -= following;
block = block.Next;
index = block.Start;
}
}
}
} }
} }