diff options
Diffstat (limited to 'lib/Net.Http')
-rw-r--r-- | lib/Net.Http/src/Core/Buffering/ContextLockedBufferManager.cs | 12 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/Buffering/IHttpBufferManager.cs | 10 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/Compression/ManagedHttpCompressor.cs | 21 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/HttpContext.cs | 26 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/HttpServerBase.cs | 32 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/Request/HttpInputStream.cs | 27 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/Response/ChunkDataAccumulator.cs | 52 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/Response/HttpContextResponseWriting.cs | 14 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/Response/HttpResponse.cs | 44 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/Response/HttpStreamResponse.cs | 50 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/Response/HttpstreamResponse.cs | 50 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/Response/ResponsBodyDataState.cs | 100 | ||||
-rw-r--r-- | lib/Net.Http/src/Core/Response/ResponseWriter.cs | 298 |
13 files changed, 382 insertions, 354 deletions
diff --git a/lib/Net.Http/src/Core/Buffering/ContextLockedBufferManager.cs b/lib/Net.Http/src/Core/Buffering/ContextLockedBufferManager.cs index 827af56..90bdd8c 100644 --- a/lib/Net.Http/src/Core/Buffering/ContextLockedBufferManager.cs +++ b/lib/Net.Http/src/Core/Buffering/ContextLockedBufferManager.cs @@ -137,15 +137,13 @@ namespace VNLib.Net.Http.Core.Buffering } ///<inheritdoc/> - public void ZeroAll() + public void FreeAll(bool zero) { - //Zero the buffer completely - MemoryUtil.InitializeBlock(_handle!.Memory); - } + if (zero) + { + MemoryUtil.InitializeBlock(_handle!.Memory); + } - ///<inheritdoc/> - public void FreeAll() - { //Clear buffer memory structs to allow gc _requestHeaderBuffer.FreeBuffer(); _responseHeaderBuffer.FreeBuffer(); diff --git a/lib/Net.Http/src/Core/Buffering/IHttpBufferManager.cs b/lib/Net.Http/src/Core/Buffering/IHttpBufferManager.cs index c7020ff..9dc067a 100644 --- a/lib/Net.Http/src/Core/Buffering/IHttpBufferManager.cs +++ b/lib/Net.Http/src/Core/Buffering/IHttpBufferManager.cs @@ -1,5 +1,5 @@ /* -* Copyright (c) 2023 Vaughn Nugent +* Copyright (c) 2024 Vaughn Nugent * * Library: VNLib * Package: VNLib.Net.Http @@ -79,13 +79,9 @@ namespace VNLib.Net.Http.Core.Buffering void AllocateBuffer(IHttpMemoryPool allocator); /// <summary> - /// Zeros all internal buffers - /// </summary> - void ZeroAll(); - - /// <summary> /// Frees all internal buffers /// </summary> - void FreeAll(); + /// <param name="zeroAll">A value that indicates if the buffer should be zeored before it's returned to the pool</param> + void FreeAll(bool zeroAll); } } diff --git a/lib/Net.Http/src/Core/Compression/ManagedHttpCompressor.cs b/lib/Net.Http/src/Core/Compression/ManagedHttpCompressor.cs index fbb17c2..bfd848a 100644 --- a/lib/Net.Http/src/Core/Compression/ManagedHttpCompressor.cs +++ b/lib/Net.Http/src/Core/Compression/ManagedHttpCompressor.cs @@ -27,17 +27,8 @@ using System.Diagnostics; namespace VNLib.Net.Http.Core.Compression { - internal sealed class ManagedHttpCompressor : IResponseCompressor + internal sealed class ManagedHttpCompressor(IHttpCompressorManager manager) : IResponseCompressor { - //Store the compressor - private readonly IHttpCompressorManager _provider; - - public ManagedHttpCompressor(IHttpCompressorManager provider) - { - Debug.Assert(provider != null, "Expected non-null provider"); - _provider = provider; - } - /* * The compressor alloc is deferd until the first call to Init() * This is because user-code should not be called during the constructor @@ -59,7 +50,7 @@ namespace VNLib.Net.Http.Core.Compression Debug.Assert(!output.IsEmpty, "Expected non-zero output buffer"); //Compress the block - return _provider.CompressBlock(_compressor!, input, output); + return manager.CompressBlock(_compressor!, input, output); } ///<inheritdoc/> @@ -69,19 +60,19 @@ namespace VNLib.Net.Http.Core.Compression Debug.Assert(_compressor != null); Debug.Assert(!output.IsEmpty, "Expected non-zero output buffer"); - return _provider.Flush(_compressor!, output); + return manager.Flush(_compressor!, output); } ///<inheritdoc/> public void Init(CompressionMethod compMethod) { //Defer alloc the compressor - _compressor ??= _provider.AllocCompressor(); + _compressor ??= manager.AllocCompressor(); Debug.Assert(_compressor != null); //Init the compressor and get the block size - BlockSize = _provider.InitCompressor(_compressor, compMethod); + BlockSize = manager.InitCompressor(_compressor, compMethod); initialized = true; } @@ -94,7 +85,7 @@ namespace VNLib.Net.Http.Core.Compression { Debug.Assert(_compressor != null, "Compressor was initialized, exepcted a non null instance"); - _provider.DeinitCompressor(_compressor); + manager.DeinitCompressor(_compressor); initialized = false; } } diff --git a/lib/Net.Http/src/Core/HttpContext.cs b/lib/Net.Http/src/Core/HttpContext.cs index 3216a9b..8cecf30 100644 --- a/lib/Net.Http/src/Core/HttpContext.cs +++ b/lib/Net.Http/src/Core/HttpContext.cs @@ -1,5 +1,5 @@ /* -* Copyright (c) 2023 Vaughn Nugent +* Copyright (c) 2024 Vaughn Nugent * * Library: VNLib * Package: VNLib.Net.Http @@ -25,6 +25,7 @@ using System; using System.IO; using System.Text; +using System.Diagnostics; using System.Threading.Tasks; using VNLib.Utils; @@ -82,7 +83,7 @@ namespace VNLib.Net.Http.Core private readonly ManagedHttpCompressor? _compressor; private ITransportContext? _ctx; - public HttpContext(HttpServer server) + public HttpContext(HttpServer server, CompressionMethod supportedMethods) { ParentServer = server; @@ -91,11 +92,17 @@ namespace VNLib.Net.Http.Core /* * We can alloc a new compressor if the server supports compression. * If no compression is supported, the compressor will never be accessed + * and never needs to be allocated */ - _compressor = server.SupportedCompressionMethods == CompressionMethod.None ? - null : - new ManagedHttpCompressor(server.Config.CompressorManager!); - + if (supportedMethods != CompressionMethod.None) + { + Debug.Assert(server.Config.CompressorManager != null, "Expected non-null provider"); + _compressor = new ManagedHttpCompressor(server.Config.CompressorManager); + } + else + { + _compressor = null; + } //Init buffer manager, if compression is supported, we need to alloc a buffer for the compressor Buffers = new(server.Config.BufferConfig, _compressor != null); @@ -190,12 +197,9 @@ namespace VNLib.Net.Http.Core //Release response/requqests Response.OnRelease(); - - //Zero before returning to pool - Buffers.ZeroAll(); - + //Free buffers - Buffers.FreeAll(); + Buffers.FreeAll(true); return true; } diff --git a/lib/Net.Http/src/Core/HttpServerBase.cs b/lib/Net.Http/src/Core/HttpServerBase.cs index a73867f..f5f3563 100644 --- a/lib/Net.Http/src/Core/HttpServerBase.cs +++ b/lib/Net.Http/src/Core/HttpServerBase.cs @@ -41,7 +41,6 @@ using System.Net.Sockets; using System.Threading.Tasks; using System.Collections.Frozen; using System.Collections.Generic; -using System.Security.Authentication; using VNLib.Utils.Logging; using VNLib.Utils.Memory.Caching; @@ -141,9 +140,7 @@ namespace VNLib.Net.Http //Configure roots and their directories ServerRoots = sites.ToFrozenDictionary(static r => r.Hostname, static tv => tv, StringComparer.OrdinalIgnoreCase); //Compile and store the timeout keepalive header - KeepAliveTimeoutHeaderValue = $"timeout={(int)_config.ConnectionKeepAlive.TotalSeconds}"; - //Create a new context store - ContextStore = ObjectRental.CreateReusable(() => new HttpContext(this)); + KeepAliveTimeoutHeaderValue = $"timeout={(int)_config.ConnectionKeepAlive.TotalSeconds}"; //Setup config copy with the internal http pool Transport = transport; //Cache supported compression methods, or none if compressor is null @@ -151,6 +148,9 @@ namespace VNLib.Net.Http CompressionMethod.None : config.CompressorManager.GetSupportedMethods(); + //Create a new context store + ContextStore = ObjectRental.CreateReusable(() => new HttpContext(this, SupportedCompressionMethods)); + //Cache wildcard root _wildcardRoot = ServerRoots.GetValueOrDefault(WILDCARD_KEY); @@ -270,20 +270,6 @@ namespace VNLib.Net.Http } /* - * An SslStream may throw a win32 exception with HRESULT 0x80090327 - * when processing a client certificate (I believe anyway) only - * an issue on some clients (browsers) - */ - - private const int UKNOWN_CERT_AUTH_HRESULT = unchecked((int)0x80090327); - - /// <summary> - /// An invlaid frame size may happen if data is recieved on an open socket - /// but does not contain valid SSL handshake data - /// </summary> - private const int INVALID_FRAME_HRESULT = unchecked((int)0x80131620); - - /* * A worker task that listens for connections from the transport */ private async Task ListenWorkerDoWork() @@ -301,7 +287,7 @@ namespace VNLib.Net.Http //Listen for new connection ITransportContext ctx = await Transport.AcceptAsync(StopToken!.Token); - //Try to dispatch the recieved event + //Try to dispatch the received event _ = DataReceivedAsync(ctx).ConfigureAwait(false); } catch (OperationCanceledException) @@ -309,14 +295,6 @@ namespace VNLib.Net.Http //Closing, exit loop break; } - catch(AuthenticationException ae) when(ae.HResult == INVALID_FRAME_HRESULT) - { - _config.ServerLog.Debug("A TLS connection attempt was made but an invalid TLS frame was received"); - } - catch (AuthenticationException ae) - { - _config.ServerLog.Error(ae); - } catch (Exception ex) { _config.ServerLog.Error(ex); diff --git a/lib/Net.Http/src/Core/Request/HttpInputStream.cs b/lib/Net.Http/src/Core/Request/HttpInputStream.cs index 7c010a3..e36d1e4 100644 --- a/lib/Net.Http/src/Core/Request/HttpInputStream.cs +++ b/lib/Net.Http/src/Core/Request/HttpInputStream.cs @@ -1,5 +1,5 @@ /* -* Copyright (c) 2023 Vaughn Nugent +* Copyright (c) 2024 Vaughn Nugent * * Library: VNLib * Package: VNLib.Net.Http @@ -38,18 +38,14 @@ namespace VNLib.Net.Http.Core /// <summary> /// Specialized stream to allow reading a request entity body with a fixed content length. /// </summary> - internal sealed class HttpInputStream : Stream + internal sealed class HttpInputStream(IHttpContextInformation ContextInfo) : Stream { - private readonly IHttpContextInformation ContextInfo; - private long ContentLength; private Stream? InputStream; private long _position; - private InitDataBuffer? _initalData; - - public HttpInputStream(IHttpContextInformation contextInfo) => ContextInfo = contextInfo; + private InitDataBuffer? _initalData; private long Remaining => Math.Max(ContentLength - _position, 0); @@ -170,17 +166,16 @@ namespace VNLib.Net.Http.Core //Return number of bytes written to the buffer return writer.Written; - } - - - public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) - { - return ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); - } + } public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) { - //Calculate the amount of data that can be read into the buffer + /* + * Iniitally I'm calculating the amount of data that can be read into + * the buffer, up to the maxium input data size. This value will clamp + * the buffer in the writer below, so it cannot read more than is + * available from the transport. + */ int bytesToRead = (int)Math.Min(buffer.Length, Remaining); if (bytesToRead == 0) @@ -256,7 +251,7 @@ namespace VNLib.Net.Http.Core read = await ReadAsync(HttpServer.WriteOnlyScratchBuffer, CancellationToken.None) .ConfigureAwait(true); - } while (read != 0); + } while (read > 0); } /// <summary> diff --git a/lib/Net.Http/src/Core/Response/ChunkDataAccumulator.cs b/lib/Net.Http/src/Core/Response/ChunkDataAccumulator.cs index cacce4c..31493ba 100644 --- a/lib/Net.Http/src/Core/Response/ChunkDataAccumulator.cs +++ b/lib/Net.Http/src/Core/Response/ChunkDataAccumulator.cs @@ -50,31 +50,14 @@ namespace VNLib.Net.Http.Core.Response * Must always leave enough room for trailing crlf at the end of * the buffer */ - private readonly int TotalMaxBufferSize => Buffer.Size - (int)Context.CrlfSegment.Length; - - /// <summary> - /// Complets and returns the memory segment containing the chunk data to send - /// to the client. This also resets the accumulator. - /// </summary> - /// <returns></returns> - public readonly Memory<byte> GetChunkData(int accumulatedSize) - { - //Update the chunk size - int reservedOffset = UpdateChunkSize(Buffer, Context, accumulatedSize); - int endPtr = GetPointerToEndOfUsedBuffer(accumulatedSize); - - //Write trailing chunk delimiter - endPtr += Context.CrlfSegment.DangerousCopyTo(Buffer, endPtr); - - return Buffer.GetMemory()[reservedOffset..endPtr]; - } + private readonly int TotalMaxBufferSize => Buffer.Size - Context.CrlfSegment.Length; /// <summary> /// Complets and returns the memory segment containing the chunk data to send /// to the client. /// </summary> /// <returns></returns> - public readonly Memory<byte> GetFinalChunkData(int accumulatedSize) + public readonly Memory<byte> GetChunkData(int accumulatedSize, bool isFinalChunk) { //Update the chunk size int reservedOffset = UpdateChunkSize(Buffer, Context, accumulatedSize); @@ -83,8 +66,11 @@ namespace VNLib.Net.Http.Core.Response //Write trailing chunk delimiter endPtr += Context.CrlfSegment.DangerousCopyTo(Buffer, endPtr); - //Write final chunk to the end of the accumulator - endPtr += Context.FinalChunkSegment.DangerousCopyTo(Buffer, endPtr); + if (isFinalChunk) + { + //Write final chunk to the end of the accumulator + endPtr += Context.FinalChunkSegment.DangerousCopyTo(Buffer, endPtr); + } return Buffer.GetMemory()[reservedOffset..endPtr]; } @@ -107,22 +93,6 @@ namespace VNLib.Net.Http.Core.Response => TotalMaxBufferSize - GetPointerToEndOfUsedBuffer(accumulatedSize); - /* - * Completed chunk is the segment of the buffer that contains the size segment - * followed by the accumulated chunk data, and the trailing crlf. - * - * The accumulated data position is the number of chunk bytes accumulated - * in the data segment. This does not include the number of reserved bytes - * are before it. - * - * We can get the value that points to the end of the used buffer - * and use the memory range operator to get the segment from the reserved - * segment, to the actual end of the data segment. - */ - private readonly Memory<byte> GetCompleteChunk(int reservedOffset, int accumulatedSize) - => Buffer.GetMemory()[reservedOffset..accumulatedSize]; - - private static int GetPointerToEndOfUsedBuffer(int accumulatedSize) => accumulatedSize + ReservedSize; /* @@ -181,14 +151,10 @@ namespace VNLib.Net.Http.Core.Response int reservedOffset = ReservedSize - totalChunkBufferBytes; //Copy encoded chunk size to the reserved segment - ref byte reservedSegRef = ref buffer.DangerousGetBinRef(reservedOffset); - ref byte chunkSizeBufRef = ref MemoryMarshal.GetReference(chunkSizeBinBuffer); - - //We know the block is super small MemoryUtil.SmallMemmove( - ref chunkSizeBufRef, + in MemoryMarshal.GetReference(chunkSizeBinBuffer), 0, - ref reservedSegRef, + ref buffer.DangerousGetBinRef(reservedOffset), 0, (ushort)totalChunkBufferBytes ); diff --git a/lib/Net.Http/src/Core/Response/HttpContextResponseWriting.cs b/lib/Net.Http/src/Core/Response/HttpContextResponseWriting.cs index 918ffe1..dcd0553 100644 --- a/lib/Net.Http/src/Core/Response/HttpContextResponseWriting.cs +++ b/lib/Net.Http/src/Core/Response/HttpContextResponseWriting.cs @@ -45,22 +45,14 @@ namespace VNLib.Net.Http.Core ValueTask discardTask = Request.InputStream.DiscardRemainingAsync(); - //See if response data needs to be written + //See if response data needs to be written, if so we can parallel discard and write if (ResponseBody.HasData) { //Parallel the write and discard Task response = WriteResponseInternalAsync(); - if (discardTask.IsCompletedSuccessfully) - { - //If discard is already complete, await the response - await response; - } - else - { - //If discard is not complete, await both, avoid wait-all method because it will allocate - await Task.WhenAll(discardTask.AsTask(), response); - } + //in .NET 8.0 WhenAll is now allocation free, so no biggie + await Task.WhenAll(discardTask.AsTask(), response); } else { diff --git a/lib/Net.Http/src/Core/Response/HttpResponse.cs b/lib/Net.Http/src/Core/Response/HttpResponse.cs index 3f2ae56..ec9879b 100644 --- a/lib/Net.Http/src/Core/Response/HttpResponse.cs +++ b/lib/Net.Http/src/Core/Response/HttpResponse.cs @@ -1,5 +1,5 @@ /* -* Copyright (c) 2023 Vaughn Nugent +* Copyright (c) 2024 Vaughn Nugent * * Library: VNLib * Package: VNLib.Net.Http @@ -39,16 +39,15 @@ using VNLib.Net.Http.Core.Buffering; namespace VNLib.Net.Http.Core.Response { - internal sealed class HttpResponse : IHttpLifeCycle + internal sealed class HttpResponse(IHttpContextInformation ContextInfo, IHttpBufferManager manager) : IHttpLifeCycle #if DEBUG , IStringSerializeable #endif { - private readonly IHttpContextInformation ContextInfo; - private readonly HashSet<HttpCookie> Cookies; - private readonly DirectStream ReusableDirectStream; - private readonly ChunkedStream ReusableChunkedStream; - private readonly HeaderDataAccumulator Writer; + private readonly HashSet<HttpCookie> Cookies = []; + private readonly DirectStream ReusableDirectStream = new(); + private readonly ChunkedStream ReusableChunkedStream = new(manager.ChunkAccumulatorBuffer, ContextInfo); + private readonly HeaderDataAccumulator Writer = new(manager.ResponseHeaderBuffer, ContextInfo); private int _headerWriterPosition; @@ -60,29 +59,13 @@ namespace VNLib.Net.Http.Core.Response /// <summary> /// Response header collection /// </summary> - public VnWebHeaderCollection Headers { get; } + public VnWebHeaderCollection Headers { get; } = []; /// <summary> /// The current http status code value /// </summary> internal HttpStatusCode StatusCode => _code; - public HttpResponse(IHttpContextInformation ctx, IHttpBufferManager manager) - { - ContextInfo = ctx; - - //Initialize a new header collection and a cookie jar - Headers = new(); - Cookies = new(); - - //Init header accumulator - Writer = new(manager.ResponseHeaderBuffer, ContextInfo); - - //Create a new chunked stream - ReusableChunkedStream = new(manager.ChunkAccumulatorBuffer, ContextInfo); - ReusableDirectStream = new(); - } - /// <summary> /// Sets the status code of the response /// </summary> @@ -365,9 +348,9 @@ namespace VNLib.Net.Http.Core.Response /// <summary> /// Writes chunked HTTP message bodies to an underlying streamwriter /// </summary> - private sealed class ChunkedStream : ReusableResponseStream, IResponseDataWriter + private sealed class ChunkedStream(IChunkAccumulatorBuffer buffer, IHttpContextInformation context) : ReusableResponseStream, IResponseDataWriter { - private readonly ChunkDataAccumulator _chunkAccumulator; + private readonly ChunkDataAccumulator _chunkAccumulator = new(buffer, context); /* * Tracks the number of bytes accumulated in the @@ -375,9 +358,6 @@ namespace VNLib.Net.Http.Core.Response */ private int _accumulatedBytes; - public ChunkedStream(IChunkAccumulatorBuffer buffer, IHttpContextInformation context) - => _chunkAccumulator = new(buffer, context); - #region Hooks ///<inheritdoc/> @@ -402,11 +382,9 @@ namespace VNLib.Net.Http.Core.Response * write the final termination sequence to the transport. */ - Memory<byte> chunkData = isFinal ? - _chunkAccumulator.GetFinalChunkData(_accumulatedBytes) : - _chunkAccumulator.GetChunkData(_accumulatedBytes); + Memory<byte> chunkData = _chunkAccumulator.GetChunkData(_accumulatedBytes, isFinal); - //Reset accumulator + //Reset accumulator now that we captured the final chunk _accumulatedBytes = 0; //Write remaining data to stream diff --git a/lib/Net.Http/src/Core/Response/HttpStreamResponse.cs b/lib/Net.Http/src/Core/Response/HttpStreamResponse.cs new file mode 100644 index 0000000..b08d2ab --- /dev/null +++ b/lib/Net.Http/src/Core/Response/HttpStreamResponse.cs @@ -0,0 +1,50 @@ +/* +* Copyright (c) 2024 Vaughn Nugent +* +* Library: VNLib +* Package: VNLib.Net.Http +* File: HttpStreamResponse.cs +* +* HttpStreamResponse.cs is part of VNLib.Net.Http which is part +* of the larger VNLib collection of libraries and utilities. +* +* VNLib.Net.Http is free software: you can redistribute it and/or modify +* it under the terms of the GNU Affero General Public License as +* published by the Free Software Foundation, either version 3 of the +* License, or (at your option) any later version. +* +* VNLib.Net.Http is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU Affero General Public License for more details. +* +* You should have received a copy of the GNU Affero General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ + +/* + * This file handles response entity processing. It handles in-memory response + * processing, as well as stream response processing. It handles constraints + * such as content-range limits. I tried to eliminate or reduce the amount of + * memory copying required to process the response entity. + */ + +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace VNLib.Net.Http.Core.Response +{ + internal readonly struct HttpStreamResponse(Stream stream) : IHttpStreamResponse + { + ///<inheritdoc/> + public readonly void Dispose() => stream.Dispose(); + + ///<inheritdoc/> + public readonly ValueTask DisposeAsync() => stream.DisposeAsync(); + + ///<inheritdoc/> + public readonly ValueTask<int> ReadAsync(Memory<byte> buffer) => stream!.ReadAsync(buffer, CancellationToken.None); + } +}
\ No newline at end of file diff --git a/lib/Net.Http/src/Core/Response/HttpstreamResponse.cs b/lib/Net.Http/src/Core/Response/HttpstreamResponse.cs new file mode 100644 index 0000000..b08d2ab --- /dev/null +++ b/lib/Net.Http/src/Core/Response/HttpstreamResponse.cs @@ -0,0 +1,50 @@ +/* +* Copyright (c) 2024 Vaughn Nugent +* +* Library: VNLib +* Package: VNLib.Net.Http +* File: HttpStreamResponse.cs +* +* HttpStreamResponse.cs is part of VNLib.Net.Http which is part +* of the larger VNLib collection of libraries and utilities. +* +* VNLib.Net.Http is free software: you can redistribute it and/or modify +* it under the terms of the GNU Affero General Public License as +* published by the Free Software Foundation, either version 3 of the +* License, or (at your option) any later version. +* +* VNLib.Net.Http is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU Affero General Public License for more details. +* +* You should have received a copy of the GNU Affero General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ + +/* + * This file handles response entity processing. It handles in-memory response + * processing, as well as stream response processing. It handles constraints + * such as content-range limits. I tried to eliminate or reduce the amount of + * memory copying required to process the response entity. + */ + +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace VNLib.Net.Http.Core.Response +{ + internal readonly struct HttpStreamResponse(Stream stream) : IHttpStreamResponse + { + ///<inheritdoc/> + public readonly void Dispose() => stream.Dispose(); + + ///<inheritdoc/> + public readonly ValueTask DisposeAsync() => stream.DisposeAsync(); + + ///<inheritdoc/> + public readonly ValueTask<int> ReadAsync(Memory<byte> buffer) => stream!.ReadAsync(buffer, CancellationToken.None); + } +}
\ No newline at end of file diff --git a/lib/Net.Http/src/Core/Response/ResponsBodyDataState.cs b/lib/Net.Http/src/Core/Response/ResponsBodyDataState.cs new file mode 100644 index 0000000..797d490 --- /dev/null +++ b/lib/Net.Http/src/Core/Response/ResponsBodyDataState.cs @@ -0,0 +1,100 @@ +/* +* Copyright (c) 2024 Vaughn Nugent +* +* Library: VNLib +* Package: VNLib.Net.Http +* File: ResponseWriter.cs +* +* ResponseWriter.cs is part of VNLib.Net.Http which is part of the larger +* VNLib collection of libraries and utilities. +* +* VNLib.Net.Http is free software: you can redistribute it and/or modify +* it under the terms of the GNU Affero General Public License as +* published by the Free Software Foundation, either version 3 of the +* License, or (at your option) any later version. +* +* VNLib.Net.Http is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU Affero General Public License for more details. +* +* You should have received a copy of the GNU Affero General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ + +/* + * This file handles response entity processing. It handles in-memory response + * processing, as well as stream response processing. It handles constraints + * such as content-range limits. I tried to eliminate or reduce the amount of + * memory copying required to process the response entity. + */ + +using System.IO; + +namespace VNLib.Net.Http.Core.Response +{ + internal readonly struct ResponsBodyDataState + { + /// <summary> + /// A value that inidcates if the response entity has been set + /// </summary> + public readonly bool IsSet; + /// <summary> + /// A value that indicates if the response entity requires buffering + /// </summary> + public readonly bool BufferRequired; + /// <summary> + /// The length (in bytes) of the response entity + /// </summary> + public readonly long Legnth; + + public readonly IHttpStreamResponse? Stream; + public readonly IMemoryResponseReader? MemResponse; + public readonly Stream? RawStream; + + private ResponsBodyDataState(IHttpStreamResponse stream, long length) + { + Legnth = length; + Stream = stream; + MemResponse = null; + RawStream = null; + IsSet = true; + BufferRequired = true; + } + + private ResponsBodyDataState(IMemoryResponseReader reader) + { + Legnth = reader.Remaining; + MemResponse = reader; + Stream = null; + RawStream = null; + IsSet = true; + BufferRequired = false; + } + + private ResponsBodyDataState(Stream stream, long length) + { + Legnth = length; + Stream = null; + MemResponse = null; + RawStream = stream; + IsSet = true; + BufferRequired = true; + } + + internal readonly HttpStreamResponse GetRawStreamResponse() => new(RawStream!); + + internal readonly void Dispose() + { + Stream?.Dispose(); + MemResponse?.Close(); + RawStream?.Dispose(); + } + + public static ResponsBodyDataState FromMemory(IMemoryResponseReader stream) => new(stream); + + public static ResponsBodyDataState FromStream(IHttpStreamResponse stream, long length) => new(stream, length); + + public static ResponsBodyDataState FromRawStream(Stream stream, long length) => new(stream, length); + } +}
\ No newline at end of file diff --git a/lib/Net.Http/src/Core/Response/ResponseWriter.cs b/lib/Net.Http/src/Core/Response/ResponseWriter.cs index b5a167c..b60537d 100644 --- a/lib/Net.Http/src/Core/Response/ResponseWriter.cs +++ b/lib/Net.Http/src/Core/Response/ResponseWriter.cs @@ -32,7 +32,6 @@ using System; using System.IO; using System.Diagnostics; -using System.Threading; using System.Threading.Tasks; using System.Runtime.CompilerServices; @@ -41,15 +40,13 @@ using VNLib.Net.Http.Core.Compression; namespace VNLib.Net.Http.Core.Response { - internal sealed class ResponseWriter : IHttpResponseBody { private ResponsBodyDataState _userState; ///<inheritdoc/> public bool HasData => _userState.IsSet; - - //Buffering is required when a stream is set + ///<inheritdoc/> public bool BufferRequired => _userState.BufferRequired; @@ -72,7 +69,7 @@ namespace VNLib.Net.Http.Core.Response Debug.Assert(response != null, "Stream value is null, illegal operation"); Debug.Assert(length > -1, "explicit length passed a negative value, illegal operation"); - _userState = new(response, length); + _userState = ResponsBodyDataState.FromStream(response, length); return true; } @@ -91,7 +88,7 @@ namespace VNLib.Net.Http.Core.Response Debug.Assert(response != null, "Memory response argument was null and expected a value"); //Assign user-state - _userState = new(response); + _userState = ResponsBodyDataState.FromMemory(response); return true; } @@ -112,147 +109,144 @@ namespace VNLib.Net.Http.Core.Response Debug.Assert(length > -1, "explicit length passed a negative value, illegal operation"); //Assign user-state - _userState = new(rawStream, length); + _userState = ResponsBodyDataState.FromRawStream(rawStream, length); return true; } + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void OnComplete() + { + //Clear response containers + _userState.Dispose(); + _userState = default; + + _readSegment = default; + } + + private ReadOnlyMemory<byte> _readSegment; #pragma warning disable CA2007 // Consider calling ConfigureAwait on the awaited task ///<inheritdoc/> - public async Task WriteEntityAsync(IDirectResponsWriter dest, Memory<byte> buffer) + public Task WriteEntityAsync(IDirectResponsWriter dest, Memory<byte> buffer) => WriteEntityAsync(dest, buffer, 0); + + ///<inheritdoc/> + public async Task WriteEntityAsync<TComp>(TComp compressor, IResponseDataWriter writer, Memory<byte> buffer) + where TComp : IResponseCompressor { - //Write a sliding window response - if (_userState.MemResponse != null) - { - //Write response body from memory - while (_userState.MemResponse.Remaining > 0) - { - //Get remaining segment - _readSegment = _userState.MemResponse.GetMemory(); + //Create a chunked response writer struct to pass to write async function + ChunkedResponseWriter<TComp> output = new(writer, compressor); - //Write segment to output stream - await dest.WriteAsync(_readSegment); + await WriteEntityAsync(output, buffer, compressor.BlockSize); - //Advance by the written amount - _userState.MemResponse.Advance(_readSegment.Length); - } + /* + * Once there is no more response data avialable to compress + * we need to flush the compressor, then flush the writer + * to publish all accumulated data to the client + */ - //Disposing of memory response can be deferred until the end of the request since its always syncrhonous - } - else if(_userState.RawStream != null) + do { - Debug.Assert(!buffer.IsEmpty, "Transfer buffer is required for streaming operations"); + //Flush the compressor output + int written = compressor.Flush(writer.GetMemory()); - await ProcessStreamDataAsync(_userState.GetRawStreamResponse(), dest, buffer); - } - else - { - Debug.Assert(!buffer.IsEmpty, "Transfer buffer is required for streaming operations"); + //No more data to buffer + if (written == 0) + { + //final flush and exit + await writer.FlushAsync(true); + break; + } - Debug.Assert(_userState.Stream != null, "Stream value is null, illegal operation"); + if (writer.Advance(written) == 0) + { + //Flush because accumulator is full + await writer.FlushAsync(false); + } - await ProcessStreamDataAsync(_userState.Stream!, dest, buffer); - } + } while (true); } - ///<inheritdoc/> - public async Task WriteEntityAsync<TComp>(TComp comp, IResponseDataWriter writer, Memory<byte> buffer) where TComp : IResponseCompressor + private async Task WriteEntityAsync<TResWriter>(TResWriter dest, Memory<byte> buffer, int blockSize) + where TResWriter : IDirectResponsWriter { //try to clamp the buffer size to the compressor block size - int maxBufferSize = Math.Min(buffer.Length, comp.BlockSize); - if(maxBufferSize > 0) + if (blockSize > 0) { - buffer = buffer[..maxBufferSize]; + buffer = buffer[..Math.Min(buffer.Length, blockSize)]; } - ChunkedResponseWriter<TComp> output = new(writer, comp); - //Write a sliding window response if (_userState.MemResponse != null) { - while (_userState.MemResponse.Remaining > 0) + if (blockSize > 0) + { + while (_userState.MemResponse.Remaining > 0) + { + //Get next segment clamped to the block size + _readSegment = _userState.MemResponse.GetRemainingConstrained(blockSize); + + //Commit output bytes + await dest.WriteAsync(_readSegment); + + //Advance by the written amount + _userState.MemResponse.Advance(_readSegment.Length); + } + } + else { - //Get next segment - _readSegment = comp.BlockSize > 0 ? - _userState.MemResponse.GetRemainingConstrained(comp.BlockSize) - : _userState.MemResponse.GetMemory(); + //Write response body from memory + while (_userState.MemResponse.Remaining > 0) + { + //Get remaining segment + _readSegment = _userState.MemResponse.GetMemory(); - //Commit output bytes - await output.WriteAsync(_readSegment); + //Write segment to output stream + await dest.WriteAsync(_readSegment); - //Advance by the written amount - _userState.MemResponse.Advance(_readSegment.Length); + //Advance by the written amount + _userState.MemResponse.Advance(_readSegment.Length); + } } //Disposing of memory response can be deferred until the end of the request since its always syncrhonous } - else if(_userState.RawStream != null) + else if (_userState.RawStream != null) { Debug.Assert(!buffer.IsEmpty, "Transfer buffer is required for streaming operations"); - //Configure a raw stream response - await ProcessStreamDataAsync(_userState.GetRawStreamResponse(), output, buffer); + await ProcessStreamDataAsync(_userState.GetRawStreamResponse(), dest, buffer, _userState.Legnth); } else { Debug.Assert(!buffer.IsEmpty, "Transfer buffer is required for streaming operations"); + Debug.Assert(_userState.Stream != null, "Stream value is null, illegal state"); - Debug.Assert(_userState.Stream != null, "Stream value is null, illegal operation"); - await ProcessStreamDataAsync(_userState.Stream, output, buffer); + await ProcessStreamDataAsync(_userState.Stream, dest, buffer, _userState.Legnth); } - - - /* - * Once there is no more response data avialable to compress - * we need to flush the compressor, then flush the writer - * to publish all accumulated data to the client - */ - - do - { - //Flush the compressor output - int written = comp.Flush(writer.GetMemory()); - - //No more data to buffer - if (written == 0) - { - //final flush and exit - await writer.FlushAsync(true); - break; - } - - if (writer.Advance(written) == 0) - { - //Flush because accumulator is full - await writer.FlushAsync(false); - } - - } while (true); } - private async Task ProcessStreamDataAsync<TStream, TWriter>(TStream stream, TWriter dest, Memory<byte> buffer) - where TStream: IHttpStreamResponse - where TWriter: IDirectResponsWriter + private static async Task ProcessStreamDataAsync<TStream, TWriter>(TStream stream, TWriter dest, Memory<byte> buffer, long length) + where TStream : IHttpStreamResponse + where TWriter : IDirectResponsWriter { /* * When streams are used, callers will submit an explict length value * which must be respected. This allows the stream size to differ from * the actual content length. This is useful for when the stream is - * non-seekable, or does not have a known length + * non-seekable, or does not have a known length. Also used for + * content-range responses, that are shorter than the whole stream. */ - long total = 0; - while (total < Length) + long sentBytes = 0; + do { - //get offset wrapper of the total buffer or remaining count - Memory<byte> offset = buffer[..(int)Math.Min(buffer.Length, Length - total)]; + Memory<byte> offset = ClampCopyBuffer(buffer, length, sentBytes); - //read + //read only the amount of data that is required int read = await stream.ReadAsync(offset); - //Guard if (read == 0) { break; @@ -261,110 +255,31 @@ namespace VNLib.Net.Http.Core.Response //write only the data that was read (slice) await dest.WriteAsync(offset[..read]); - //Update total - total += read; - } + sentBytes += read; + + } while (sentBytes < length); //Try to dispose the response stream asyncrhonously since we are done with it await stream.DisposeAsync(); } - - private static bool CompressNextSegment<TComp>(ref ForwardOnlyMemoryReader<byte> reader, TComp comp, IResponseDataWriter writer) - where TComp: IResponseCompressor + + private static Memory<byte> ClampCopyBuffer(Memory<byte> buffer, long contentLength, long sentBytes) { - //Get output buffer - Memory<byte> output = writer.GetMemory(); - - //Compress the trimmed block - CompressionResult res = comp.CompressBlock(reader.Window, output); - ValidateCompressionResult(in res); - - //Commit input bytes - reader.Advance(res.BytesRead); - - return writer.Advance(res.BytesWritten) == 0; + //get offset wrapper of the total buffer or remaining count + int bufferSize = (int)Math.Min(buffer.Length, contentLength - sentBytes); + return buffer[..bufferSize]; } [Conditional("DEBUG")] - private static void ValidateCompressionResult(in CompressionResult result) - { + private static void ValidateCompressionResult(in CompressionResult result, int segmentLen) + { Debug.Assert(result.BytesRead > -1, "Compression result returned a negative bytes read value"); Debug.Assert(result.BytesWritten > -1, "Compression result returned a negative bytes written value"); + Debug.Assert(result.BytesWritten <= segmentLen, "Compression result wrote more bytes than the input segment length"); } #pragma warning restore CA2007 // Consider calling ConfigureAwait on the awaited task - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void OnComplete() - { - //Clear rseponse containers - _userState.Dispose(); - _userState = default; - - _readSegment = default; - } - - - private readonly struct ResponsBodyDataState - { - public readonly bool IsSet; - public readonly bool BufferRequired; - public readonly long Legnth; - public readonly IHttpStreamResponse? Stream; - public readonly IMemoryResponseReader? MemResponse; - public readonly Stream? RawStream; - - public ResponsBodyDataState(IHttpStreamResponse stream, long length) - { - Legnth = length; - Stream = stream; - MemResponse = null; - RawStream = null; - IsSet = true; - BufferRequired = true; - } - - public ResponsBodyDataState(IMemoryResponseReader reader) - { - Legnth = reader.Remaining; - MemResponse = reader; - Stream = null; - RawStream = null; - IsSet = true; - BufferRequired = false; - } - - public ResponsBodyDataState(Stream stream, long length) - { - Legnth = length; - Stream = null; - MemResponse = null; - RawStream = stream; - IsSet = true; - BufferRequired = true; - } - - public readonly HttpstreamResponse GetRawStreamResponse() => new(RawStream!); - - public readonly void Dispose() - { - Stream?.Dispose(); - MemResponse?.Close(); - RawStream?.Dispose(); - } - } - - private readonly struct HttpstreamResponse(Stream stream) : IHttpStreamResponse - { - ///<inheritdoc/> - public readonly void Dispose() => stream.Dispose(); - - ///<inheritdoc/> - public readonly ValueTask DisposeAsync() => stream.DisposeAsync(); - - ///<inheritdoc/> - public readonly ValueTask<int> ReadAsync(Memory<byte> buffer) => stream!.ReadAsync(buffer, CancellationToken.None); - } + private readonly struct ChunkedResponseWriter<TComp>(IResponseDataWriter writer, TComp comp) : IDirectResponsWriter where TComp : IResponseCompressor @@ -378,7 +293,7 @@ namespace VNLib.Net.Http.Core.Response do { //Compress the buffered data and flush if required - if (CompressNextSegment(ref streamReader, comp, writer)) + if (CompressNextSegment(ref streamReader)) { //Time to flush await writer.FlushAsync(false); @@ -386,6 +301,21 @@ namespace VNLib.Net.Http.Core.Response } while (streamReader.WindowSize > 0); } + + private readonly bool CompressNextSegment(ref ForwardOnlyMemoryReader<byte> reader) + { + //Get output buffer + Memory<byte> output = writer.GetMemory(); + + //Compress the trimmed block + CompressionResult res = comp.CompressBlock(reader.Window, output); + ValidateCompressionResult(in res, output.Length); + + //Commit input bytes + reader.Advance(res.BytesRead); + + return writer.Advance(res.BytesWritten) == 0; + } } } }
\ No newline at end of file |