Commit 0a208f3a authored by Honfika's avatar Honfika

Creating less byte arrays. CopyToStream functions simlified. ReadBytesAsync...

Creating less byte arrays. CopyToStream functions simlified. ReadBytesAsync reads the data to the buffer which is passed in the argument and returns immediately when any data read. (old read method can be removed, it is not used)
parent 2a7c1f91
using System.IO;
using System.IO.Compression;
using System.Threading.Tasks;
using Titanium.Web.Proxy.Helpers;
namespace Titanium.Web.Proxy.Decompression
{
......@@ -14,17 +15,23 @@ namespace Titanium.Web.Proxy.Decompression
using (var stream = new MemoryStream(compressedArray))
using (var decompressor = new DeflateStream(stream, CompressionMode.Decompress))
{
var buffer = new byte[bufferSize];
using (var output = new MemoryStream())
var buffer = BufferPool.GetBuffer(bufferSize);
try
{
int read;
while ((read = await decompressor.ReadAsync(buffer, 0, buffer.Length)) > 0)
using (var output = new MemoryStream())
{
output.Write(buffer, 0, read);
}
int read;
while ((read = await decompressor.ReadAsync(buffer, 0, buffer.Length)) > 0)
{
output.Write(buffer, 0, read);
}
return output.ToArray();
return output.ToArray();
}
}
finally
{
BufferPool.ReturnBuffer(buffer);
}
}
}
......
using System.IO;
using System.IO.Compression;
using System.Threading.Tasks;
using Titanium.Web.Proxy.Helpers;
namespace Titanium.Web.Proxy.Decompression
{
......@@ -13,16 +14,23 @@ namespace Titanium.Web.Proxy.Decompression
{
using (var decompressor = new GZipStream(new MemoryStream(compressedArray), CompressionMode.Decompress))
{
var buffer = new byte[bufferSize];
using (var output = new MemoryStream())
var buffer = BufferPool.GetBuffer(bufferSize);
try
{
int read;
while ((read = await decompressor.ReadAsync(buffer, 0, buffer.Length)) > 0)
using (var output = new MemoryStream())
{
output.Write(buffer, 0, read);
}
int read;
while ((read = await decompressor.ReadAsync(buffer, 0, buffer.Length)) > 0)
{
output.Write(buffer, 0, read);
}
return output.ToArray();
return output.ToArray();
}
}
finally
{
BufferPool.ReturnBuffer(buffer);
}
}
}
......
......@@ -117,12 +117,12 @@ namespace Titanium.Web.Proxy.EventArguments
if (WebSession.Request.ContentLength > 0)
{
//If not chunked then its easy just read the amount of bytes mentioned in content length header of response
await ProxyClient.ClientStreamReader.CopyBytesToStream(bufferSize, requestBodyStream,
await ProxyClient.ClientStreamReader.CopyBytesToStream(requestBodyStream,
WebSession.Request.ContentLength);
}
else if (WebSession.Request.HttpVersion.Major == 1 && WebSession.Request.HttpVersion.Minor == 0)
{
await WebSession.ServerConnection.StreamReader.CopyBytesToStream(bufferSize, requestBodyStream, long.MaxValue);
await WebSession.ServerConnection.StreamReader.CopyBytesToStream(requestBodyStream, long.MaxValue);
}
}
WebSession.Request.RequestBody = await GetDecompressedResponseBody(WebSession.Request.ContentEncoding,
......@@ -155,12 +155,12 @@ namespace Titanium.Web.Proxy.EventArguments
if (WebSession.Response.ContentLength > 0)
{
//If not chunked then its easy just read the amount of bytes mentioned in content length header of response
await WebSession.ServerConnection.StreamReader.CopyBytesToStream(bufferSize, responseBodyStream,
await WebSession.ServerConnection.StreamReader.CopyBytesToStream(responseBodyStream,
WebSession.Response.ContentLength);
}
else if (WebSession.Response.HttpVersion.Major == 1 && WebSession.Response.HttpVersion.Minor == 0 || WebSession.Response.ContentLength == -1)
{
await WebSession.ServerConnection.StreamReader.CopyBytesToStream(bufferSize, responseBodyStream, long.MaxValue);
await WebSession.ServerConnection.StreamReader.CopyBytesToStream(responseBodyStream, long.MaxValue);
}
}
......
......@@ -34,34 +34,31 @@ namespace Titanium.Web.Proxy.Extensions
/// copies the specified bytes to the stream from the input stream
/// </summary>
/// <param name="streamReader"></param>
/// <param name="bufferSize"></param>
/// <param name="stream"></param>
/// <param name="totalBytesToRead"></param>
/// <returns></returns>
internal static async Task CopyBytesToStream(this CustomBinaryReader streamReader, int bufferSize, Stream stream, long totalBytesToRead)
internal static async Task CopyBytesToStream(this CustomBinaryReader streamReader, Stream stream, long totalBytesToRead)
{
var totalbytesRead = 0;
long bytesToRead = totalBytesToRead < bufferSize ? totalBytesToRead : bufferSize;
byte[] buffer = streamReader.Buffer;
long remainingBytes = totalBytesToRead;
while (totalbytesRead < totalBytesToRead)
while (remainingBytes > 0)
{
var buffer = await streamReader.ReadBytesAsync(bytesToRead);
if (buffer.Length == 0)
int bytesToRead = buffer.Length;
if (remainingBytes < bytesToRead)
{
break;
bytesToRead = (int)remainingBytes;
}
totalbytesRead += buffer.Length;
var remainingBytes = totalBytesToRead - totalbytesRead;
if (remainingBytes < bytesToRead)
int bytesRead = await streamReader.ReadBytesAsync(buffer, bytesToRead);
if (bytesRead == 0)
{
bytesToRead = remainingBytes;
break;
}
await stream.WriteAsync(buffer, 0, buffer.Length);
remainingBytes -= bytesRead;
await stream.WriteAsync(buffer, 0, bytesRead);
}
}
......@@ -80,8 +77,8 @@ namespace Titanium.Web.Proxy.Extensions
if (chunkSize != 0)
{
var buffer = await clientStreamReader.ReadBytesAsync(chunkSize);
await stream.WriteAsync(buffer, 0, buffer.Length);
await CopyBytesToStream(clientStreamReader, stream, chunkSize);
//chunk trail
await clientStreamReader.ReadLineAsync();
}
......@@ -132,30 +129,7 @@ namespace Titanium.Web.Proxy.Extensions
contentLength = long.MaxValue;
}
int bytesToRead = bufferSize;
if (contentLength < bufferSize)
{
bytesToRead = (int)contentLength;
}
var buffer = new byte[bufferSize];
var bytesRead = 0;
var totalBytesRead = 0;
while ((bytesRead += await inStreamReader.BaseStream.ReadAsync(buffer, 0, bytesToRead)) > 0)
{
await outStream.WriteAsync(buffer, 0, bytesRead);
totalBytesRead += bytesRead;
if (totalBytesRead == contentLength)
break;
bytesRead = 0;
var remainingBytes = contentLength - totalBytesRead;
bytesToRead = remainingBytes > (long)bufferSize ? bufferSize : (int)remainingBytes;
}
await CopyBytesToStream(inStreamReader, outStream, contentLength);
}
else
{
......@@ -178,14 +152,11 @@ namespace Titanium.Web.Proxy.Extensions
if (chunkSize != 0)
{
var buffer = await inStreamReader.ReadBytesAsync(chunkSize);
var chunkHeadBytes = Encoding.ASCII.GetBytes(chunkSize.ToString("x2"));
await outStream.WriteAsync(chunkHeadBytes, 0, chunkHeadBytes.Length);
await outStream.WriteAsync(ProxyConstants.NewLineBytes, 0, ProxyConstants.NewLineBytes.Length);
await outStream.WriteAsync(buffer, 0, chunkSize);
await CopyBytesToStream(inStreamReader, outStream, chunkSize);
await outStream.WriteAsync(ProxyConstants.NewLineBytes, 0, ProxyConstants.NewLineBytes.Length);
await inStreamReader.ReadLineAsync();
......
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace Titanium.Web.Proxy.Helpers
{
internal static class BufferPool
{
private static readonly ConcurrentQueue<byte[]> buffers = new ConcurrentQueue<byte[]>();
internal static byte[] GetBuffer(int bufferSize)
{
byte[] buffer;
if (!buffers.TryDequeue(out buffer) || buffer.Length != bufferSize)
{
buffer = new byte[bufferSize];
}
return buffer;
}
internal static void ReturnBuffer(byte[] buffer)
{
if (buffer != null)
{
buffers.Enqueue(buffer);
}
}
}
}
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading.Tasks;
......@@ -16,30 +14,22 @@ namespace Titanium.Web.Proxy.Helpers
{
private readonly CustomBufferedStream stream;
private readonly int bufferSize;
private readonly byte[] staticBuffer;
private readonly Encoding encoding;
private static readonly ConcurrentQueue<byte[]> buffers
= new ConcurrentQueue<byte[]>();
private volatile bool disposed;
internal byte[] Buffer { get; }
internal CustomBinaryReader(CustomBufferedStream stream, int bufferSize)
{
this.stream = stream;
if (!buffers.TryDequeue(out staticBuffer) || staticBuffer.Length != bufferSize)
{
staticBuffer = new byte[bufferSize];
}
Buffer = BufferPool.GetBuffer(bufferSize);
this.bufferSize = bufferSize;
//default to UTF-8
encoding = Encoding.UTF8;
}
internal Stream BaseStream => stream;
/// <summary>
/// Read a line from the byte stream
/// </summary>
......@@ -51,7 +41,7 @@ namespace Titanium.Web.Proxy.Helpers
int bufferDataLength = 0;
// try to use the thread static buffer, usually it is enough
var buffer = staticBuffer;
var buffer = Buffer;
while (stream.DataAvailable || await stream.FillBufferAsync())
{
......@@ -63,6 +53,7 @@ namespace Titanium.Web.Proxy.Helpers
{
return encoding.GetString(buffer, 0, bufferDataLength - 1);
}
//end of stream
if (newChar == '\0')
{
......@@ -95,6 +86,7 @@ namespace Titanium.Web.Proxy.Helpers
{
requestLines.Add(tmpLine);
}
return requestLines;
}
......@@ -118,7 +110,7 @@ namespace Titanium.Web.Proxy.Helpers
{
int bytesToRead = bufferSize;
var buffer = staticBuffer;
var buffer = Buffer;
if (totalBytesToRead < bufferSize)
{
bytesToRead = (int)totalBytesToRead;
......@@ -148,19 +140,30 @@ namespace Titanium.Web.Proxy.Helpers
{
//Normally this should not happen. Resize the buffer anyway
var newBuffer = new byte[totalBytesRead];
Buffer.BlockCopy(buffer, 0, newBuffer, 0, totalBytesRead);
System.Buffer.BlockCopy(buffer, 0, newBuffer, 0, totalBytesRead);
buffer = newBuffer;
}
return buffer;
}
/// <summary>
/// Read the specified number (or less) of raw bytes from the base stream to the given buffer
/// </summary>
/// <param name="buffer"></param>
/// <param name="bytesToRead"></param>
/// <returns>The number of bytes read</returns>
internal Task<int> ReadBytesAsync(byte[] buffer, int bytesToRead)
{
return stream.ReadAsync(buffer, 0, bytesToRead);
}
public void Dispose()
{
if (!disposed)
{
disposed = true;
buffers.Enqueue(staticBuffer);
BufferPool.ReturnBuffer(Buffer);
}
}
......@@ -172,7 +175,7 @@ namespace Titanium.Web.Proxy.Helpers
private void ResizeBuffer(ref byte[] buffer, long size)
{
var newBuffer = new byte[size];
Buffer.BlockCopy(buffer, 0, newBuffer, 0, buffer.Length);
System.Buffer.BlockCopy(buffer, 0, newBuffer, 0, buffer.Length);
buffer = newBuffer;
}
}
......
......@@ -16,7 +16,7 @@ namespace Titanium.Web.Proxy.Helpers
{
private readonly Stream baseStream;
private readonly byte[] streamBuffer;
private byte[] streamBuffer;
private int bufferLength;
......@@ -30,7 +30,7 @@ namespace Titanium.Web.Proxy.Helpers
public CustomBufferedStream(Stream baseStream, int bufferSize)
{
this.baseStream = baseStream;
streamBuffer = new byte[bufferSize];
streamBuffer = BufferPool.GetBuffer(bufferSize);
}
/// <summary>
......@@ -330,9 +330,10 @@ namespace Titanium.Web.Proxy.Helpers
protected override void Dispose(bool disposing)
{
baseStream.Dispose();
BufferPool.ReturnBuffer(streamBuffer);
streamBuffer = null;
}
/// <summary>
/// When overridden in a derived class, gets a value indicating whether the current stream supports reading.
/// </summary>
......
......@@ -599,7 +599,7 @@ namespace Titanium.Web.Proxy
//send the request body bytes to server
if (args.WebSession.Request.ContentLength > 0)
{
await args.ProxyClient.ClientStreamReader.CopyBytesToStream(BufferSize, postStream, args.WebSession.Request.ContentLength);
await args.ProxyClient.ClientStreamReader.CopyBytesToStream(postStream, args.WebSession.Request.ContentLength);
}
//Need to revist, find any potential bugs
//send the request body bytes to server in chunks
......
......@@ -75,6 +75,7 @@
<Compile Include="Extensions\ByteArrayExtensions.cs" />
<Compile Include="Extensions\FuncExtensions.cs" />
<Compile Include="Extensions\StringExtensions.cs" />
<Compile Include="Helpers\BufferPool.cs" />
<Compile Include="Helpers\CustomBufferedStream.cs" />
<Compile Include="Helpers\Network.cs" />
<Compile Include="Helpers\RunTime.cs" />
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment