Merge pull request #2904 from Bond-009/buffer

Use System.Buffers in RangeRequestWriter
This commit is contained in:
Anthony Lavado 2020-06-23 12:50:24 -04:00 committed by GitHub
commit 92091f13d3
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 103 additions and 120 deletions

View file

@ -585,7 +585,7 @@ namespace Emby.Server.Implementations.HttpServer
if (!string.IsNullOrWhiteSpace(rangeHeader) && totalContentLength.HasValue) if (!string.IsNullOrWhiteSpace(rangeHeader) && totalContentLength.HasValue)
{ {
var hasHeaders = new RangeRequestWriter(rangeHeader, totalContentLength.Value, stream, contentType, isHeadRequest, _logger) var hasHeaders = new RangeRequestWriter(rangeHeader, totalContentLength.Value, stream, contentType, isHeadRequest)
{ {
OnComplete = options.OnComplete OnComplete = options.OnComplete
}; };
@ -622,8 +622,11 @@ namespace Emby.Server.Implementations.HttpServer
/// <summary> /// <summary>
/// Adds the caching responseHeaders. /// Adds the caching responseHeaders.
/// </summary> /// </summary>
private void AddCachingHeaders(IDictionary<string, string> responseHeaders, TimeSpan? cacheDuration, private void AddCachingHeaders(
bool noCache, DateTime? lastModifiedDate) IDictionary<string, string> responseHeaders,
TimeSpan? cacheDuration,
bool noCache,
DateTime? lastModifiedDate)
{ {
if (noCache) if (noCache)
{ {

View file

@ -1,6 +1,7 @@
#pragma warning disable CS1591 #pragma warning disable CS1591
using System; using System;
using System.Buffers;
using System.Collections.Generic; using System.Collections.Generic;
using System.Globalization; using System.Globalization;
using System.IO; using System.IO;
@ -8,52 +9,17 @@ using System.Net;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MediaBrowser.Model.Services; using MediaBrowser.Model.Services;
using Microsoft.Extensions.Logging;
using Microsoft.Net.Http.Headers; using Microsoft.Net.Http.Headers;
namespace Emby.Server.Implementations.HttpServer namespace Emby.Server.Implementations.HttpServer
{ {
public class RangeRequestWriter : IAsyncStreamWriter, IHttpResult public class RangeRequestWriter : IAsyncStreamWriter, IHttpResult
{ {
/// <summary>
/// Gets or sets the source stream.
/// </summary>
/// <value>The source stream.</value>
private Stream SourceStream { get; set; }
private string RangeHeader { get; set; }
private bool IsHeadRequest { get; set; }
private long RangeStart { get; set; }
private long RangeEnd { get; set; }
private long RangeLength { get; set; }
private long TotalContentLength { get; set; }
public Action OnComplete { get; set; }
private readonly ILogger _logger;
private const int BufferSize = 81920; private const int BufferSize = 81920;
/// <summary>
/// The _options.
/// </summary>
private readonly Dictionary<string, string> _options = new Dictionary<string, string>(); private readonly Dictionary<string, string> _options = new Dictionary<string, string>();
/// <summary> private List<KeyValuePair<long, long?>> _requestedRanges;
/// The us culture.
/// </summary>
private static readonly CultureInfo UsCulture = new CultureInfo("en-US");
/// <summary>
/// Additional HTTP Headers.
/// </summary>
/// <value>The headers.</value>
public IDictionary<string, string> Headers => _options;
/// <summary> /// <summary>
/// Initializes a new instance of the <see cref="RangeRequestWriter" /> class. /// Initializes a new instance of the <see cref="RangeRequestWriter" /> class.
@ -63,8 +29,7 @@ namespace Emby.Server.Implementations.HttpServer
/// <param name="source">The source.</param> /// <param name="source">The source.</param>
/// <param name="contentType">Type of the content.</param> /// <param name="contentType">Type of the content.</param>
/// <param name="isHeadRequest">if set to <c>true</c> [is head request].</param> /// <param name="isHeadRequest">if set to <c>true</c> [is head request].</param>
/// <param name="logger">The logger instance.</param> public RangeRequestWriter(string rangeHeader, long contentLength, Stream source, string contentType, bool isHeadRequest)
public RangeRequestWriter(string rangeHeader, long contentLength, Stream source, string contentType, bool isHeadRequest, ILogger logger)
{ {
if (string.IsNullOrEmpty(contentType)) if (string.IsNullOrEmpty(contentType))
{ {
@ -74,7 +39,6 @@ namespace Emby.Server.Implementations.HttpServer
RangeHeader = rangeHeader; RangeHeader = rangeHeader;
SourceStream = source; SourceStream = source;
IsHeadRequest = isHeadRequest; IsHeadRequest = isHeadRequest;
this._logger = logger;
ContentType = contentType; ContentType = contentType;
Headers[HeaderNames.ContentType] = contentType; Headers[HeaderNames.ContentType] = contentType;
@ -84,6 +48,81 @@ namespace Emby.Server.Implementations.HttpServer
SetRangeValues(contentLength); SetRangeValues(contentLength);
} }
/// <summary>
/// Gets or sets the source stream.
/// </summary>
/// <value>The source stream.</value>
private Stream SourceStream { get; set; }
private string RangeHeader { get; set; }
private bool IsHeadRequest { get; set; }
private long RangeStart { get; set; }
private long RangeEnd { get; set; }
private long RangeLength { get; set; }
private long TotalContentLength { get; set; }
public Action OnComplete { get; set; }
/// <summary>
/// Additional HTTP Headers
/// </summary>
/// <value>The headers.</value>
public IDictionary<string, string> Headers => _options;
/// <summary>
/// Gets the requested ranges.
/// </summary>
/// <value>The requested ranges.</value>
protected List<KeyValuePair<long, long?>> RequestedRanges
{
get
{
if (_requestedRanges == null)
{
_requestedRanges = new List<KeyValuePair<long, long?>>();
// Example: bytes=0-,32-63
var ranges = RangeHeader.Split('=')[1].Split(',');
foreach (var range in ranges)
{
var vals = range.Split('-');
long start = 0;
long? end = null;
if (!string.IsNullOrEmpty(vals[0]))
{
start = long.Parse(vals[0], CultureInfo.InvariantCulture);
}
if (!string.IsNullOrEmpty(vals[1]))
{
end = long.Parse(vals[1], CultureInfo.InvariantCulture);
}
_requestedRanges.Add(new KeyValuePair<long, long?>(start, end));
}
}
return _requestedRanges;
}
}
public string ContentType { get; set; }
public IRequest RequestContext { get; set; }
public object Response { get; set; }
public int Status { get; set; }
public HttpStatusCode StatusCode
{
get => (HttpStatusCode)Status;
set => Status = (int)value;
}
/// <summary> /// <summary>
/// Sets the range values. /// Sets the range values.
/// </summary> /// </summary>
@ -115,50 +154,6 @@ namespace Emby.Server.Implementations.HttpServer
} }
} }
/// <summary>
/// The _requested ranges.
/// </summary>
private List<KeyValuePair<long, long?>> _requestedRanges;
/// <summary>
/// Gets the requested ranges.
/// </summary>
/// <value>The requested ranges.</value>
protected List<KeyValuePair<long, long?>> RequestedRanges
{
get
{
if (_requestedRanges == null)
{
_requestedRanges = new List<KeyValuePair<long, long?>>();
// Example: bytes=0-,32-63
var ranges = RangeHeader.Split('=')[1].Split(',');
foreach (var range in ranges)
{
var vals = range.Split('-');
long start = 0;
long? end = null;
if (!string.IsNullOrEmpty(vals[0]))
{
start = long.Parse(vals[0], UsCulture);
}
if (!string.IsNullOrEmpty(vals[1]))
{
end = long.Parse(vals[1], UsCulture);
}
_requestedRanges.Add(new KeyValuePair<long, long?>(start, end));
}
}
return _requestedRanges;
}
}
public async Task WriteToAsync(Stream responseStream, CancellationToken cancellationToken) public async Task WriteToAsync(Stream responseStream, CancellationToken cancellationToken)
{ {
try try
@ -174,59 +169,44 @@ namespace Emby.Server.Implementations.HttpServer
// If the requested range is "0-", we can optimize by just doing a stream copy // If the requested range is "0-", we can optimize by just doing a stream copy
if (RangeEnd >= TotalContentLength - 1) if (RangeEnd >= TotalContentLength - 1)
{ {
await source.CopyToAsync(responseStream, BufferSize).ConfigureAwait(false); await source.CopyToAsync(responseStream, BufferSize, cancellationToken).ConfigureAwait(false);
} }
else else
{ {
await CopyToInternalAsync(source, responseStream, RangeLength).ConfigureAwait(false); await CopyToInternalAsync(source, responseStream, RangeLength, cancellationToken).ConfigureAwait(false);
} }
} }
} }
finally finally
{ {
if (OnComplete != null) OnComplete?.Invoke();
{
OnComplete();
}
} }
} }
private static async Task CopyToInternalAsync(Stream source, Stream destination, long copyLength) private static async Task CopyToInternalAsync(Stream source, Stream destination, long copyLength, CancellationToken cancellationToken)
{ {
var array = new byte[BufferSize]; var array = ArrayPool<byte>.Shared.Rent(BufferSize);
int bytesRead; try
while ((bytesRead = await source.ReadAsync(array, 0, array.Length).ConfigureAwait(false)) != 0)
{ {
if (bytesRead == 0) int bytesRead;
while ((bytesRead = await source.ReadAsync(array, 0, array.Length, cancellationToken).ConfigureAwait(false)) != 0)
{ {
break; var bytesToCopy = Math.Min(bytesRead, copyLength);
}
var bytesToCopy = Math.Min(bytesRead, copyLength); await destination.WriteAsync(array, 0, Convert.ToInt32(bytesToCopy), cancellationToken).ConfigureAwait(false);
await destination.WriteAsync(array, 0, Convert.ToInt32(bytesToCopy)).ConfigureAwait(false); copyLength -= bytesToCopy;
copyLength -= bytesToCopy; if (copyLength <= 0)
{
if (copyLength <= 0) break;
{ }
break;
} }
} }
} finally
{
public string ContentType { get; set; } ArrayPool<byte>.Shared.Return(array);
}
public IRequest RequestContext { get; set; }
public object Response { get; set; }
public int Status { get; set; }
public HttpStatusCode StatusCode
{
get => (HttpStatusCode)Status;
set => Status = (int)value;
} }
} }
} }