-
Notifications
You must be signed in to change notification settings - Fork 1.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement decompression in PipeliningServerHandler (#10155)
* Replace chunked write API This PR replaces the writeChunked and writeFile APIs with a new writeStream API that takes an InputStream. This removes the need for the ChunkedWriteHandler. Chunked writes were used for two purposes: Sending file regions and sending InputStreams. This has always complicated the HTTP pipeline somewhat as the pipeline had to deal with not just HttpContent objects but also ChunkedInput and FileRegion objects. This PR replaces the machinery for InputStream writing with a more straight-forward solution that reads the data on the IO thread and then sends it down the channel. Additionally, the file-specific APIs based on RandomAccessFile are removed. The body writer now just creates an InputStream for the file region in question and sends that. This removes support for zero-copy transfers, however that is a niche feature anyway because it doesn't work with TLS or HTTP/2. If someone wants a performant HTTP server, HTTP/2 takes priority over zero-copy so it makes little sense. This PR may have small conflicts with #10131 as that PR changed the PipeliningServerHandler body handling a little bit. Otherwise this PR should have no visible impact on users. * remove unused class * remove unused class * Fix request backpressure PipeliningServerHandler was supposed to implement backpressure, but it turns out that auto read was still enabled and that the implementation didn't really work. This means that it would keep reading even if that means buffering data when the downstream can't keep up. This PR disables auto read and fixes the read implementation in PipeliningServerHandler. In principle there should be no change to users, this just means that instead of buffering any incoming data internally, backpressure is now applied to the client. This PR is based on #10138 but is separate for easier review. It also has conflicts with #10131. * Implement decompression in PipeliningServerHandler This patch implements the logic of HttpContentDecompressor in PipeliningServerHandler. This allows us to shrink the pipeline a little. The perf impact for uncompressed requests should basically be zero. This builds on the changes in #10142. * address review * revert * add DecompressionSpec
- Loading branch information
Showing
4 changed files
with
386 additions
and
18 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
76 changes: 76 additions & 0 deletions
76
http-server-netty/src/test/groovy/io/micronaut/http/server/netty/DecompressionSpec.groovy
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| package io.micronaut.http.server.netty | ||
|
|
||
| import io.micronaut.context.ApplicationContext | ||
| import io.micronaut.context.annotation.Requires | ||
| import io.micronaut.http.HttpRequest | ||
| import io.micronaut.http.annotation.Body | ||
| import io.micronaut.http.annotation.Controller | ||
| import io.micronaut.http.annotation.Post | ||
| import io.micronaut.http.client.HttpClient | ||
| import io.micronaut.runtime.server.EmbeddedServer | ||
| import io.netty.buffer.ByteBuf | ||
| import io.netty.buffer.ByteBufUtil | ||
| import io.netty.buffer.Unpooled | ||
| import io.netty.channel.ChannelHandler | ||
| import io.netty.channel.embedded.EmbeddedChannel | ||
| import io.netty.handler.codec.compression.SnappyFrameEncoder | ||
| import io.netty.handler.codec.compression.ZlibCodecFactory | ||
| import io.netty.handler.codec.compression.ZlibWrapper | ||
| import io.netty.handler.codec.http.HttpHeaderNames | ||
| import io.netty.handler.codec.http.HttpHeaderValues | ||
| import spock.lang.Specification | ||
|
|
||
| import java.util.concurrent.ThreadLocalRandom | ||
|
|
||
| class DecompressionSpec extends Specification { | ||
| def decompression(ChannelHandler compressor, CharSequence contentEncoding) { | ||
| given: | ||
| def ctx = ApplicationContext.run(['spec.name': 'DecompressionSpec']) | ||
| def server = ctx.getBean(EmbeddedServer).start() | ||
| def client = ctx.createBean(HttpClient, server.URI).toBlocking() | ||
|
|
||
| def compChannel = new EmbeddedChannel(compressor) | ||
| byte[] uncompressed = new byte[1024] | ||
| ThreadLocalRandom.current().nextBytes(uncompressed) | ||
| compChannel.writeOutbound(Unpooled.copiedBuffer(uncompressed)) | ||
| compChannel.finish() | ||
| ByteBuf compressed = Unpooled.buffer() | ||
| while (true) { | ||
| ByteBuf o = compChannel.readOutbound() | ||
| if (o == null) { | ||
| break | ||
| } | ||
| compressed.writeBytes(o) | ||
| o.release() | ||
| } | ||
|
|
||
| when: | ||
| client.exchange(HttpRequest.POST("/decompress", ByteBufUtil.getBytes(compressed)).header(HttpHeaderNames.CONTENT_ENCODING, contentEncoding)) | ||
| then: | ||
| ctx.getBean(Ctrl).data == uncompressed | ||
|
|
||
| cleanup: | ||
| client.close() | ||
| server.stop() | ||
| ctx.close() | ||
|
|
||
| where: | ||
| contentEncoding | compressor | ||
| HttpHeaderValues.GZIP | ZlibCodecFactory.newZlibEncoder(ZlibWrapper.GZIP) | ||
| HttpHeaderValues.X_GZIP | ZlibCodecFactory.newZlibEncoder(ZlibWrapper.GZIP) | ||
| HttpHeaderValues.DEFLATE | ZlibCodecFactory.newZlibEncoder(ZlibWrapper.NONE) | ||
| HttpHeaderValues.X_DEFLATE | ZlibCodecFactory.newZlibEncoder(ZlibWrapper.NONE) | ||
| HttpHeaderValues.SNAPPY | new SnappyFrameEncoder() | ||
| } | ||
|
|
||
| @Requires(property = "spec.name", value = "DecompressionSpec") | ||
| @Controller | ||
| static class Ctrl { | ||
| byte[] data | ||
|
|
||
| @Post("/decompress") | ||
| void receive(@Body byte[] data) { | ||
| this.data = data | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.