Add CompressibleBytesOutputStream for compression #24927

Tim-Brooks · 2017-05-28T23:01:49Z

This is a follow-up to #23941. Currently there are a number of
complexities related to compression. The raw DeflaterOutputStream must
be closed prior to sending bytes to ensure that EOS bytes are written.
But the underlying ReleasableBytesStreamOutput cannot be closed until
the bytes are sent to ensure that the bytes are not reused.

Right now we have three different stream references hanging around in
TCPTransport to handle this complexity. This commit introduces
CompressibleBytesOutputStream to be one stream implemenation that will
behave properly with or without compression enabled.

Tim-Brooks · 2017-05-28T23:05:57Z

I was particularly interested in feedback on this question:

Should the CompressibleBytesOutputStream class live in:

org.elasticsearch.transport
org.elasticsearch.common.io.stream

s1monw

looks good left some suggestions

s1monw · 2017-05-29T12:24:18Z

core/src/main/java/org/elasticsearch/transport/CompressibleBytesOutputStream.java

+
+import java.io.IOException;
+
+public class CompressibleBytesOutputStream extends StreamOutput implements Releasable {


lets keep this class here and make it pkg private and final. it's very special

s1monw · 2017-05-29T12:26:03Z

core/src/main/java/org/elasticsearch/transport/CompressibleBytesOutputStream.java

+     * @return bytes underlying the stream
+     * @throws IOException if an exception occurs when writing or flushing
+     */
+    public BytesStream finishStream() throws IOException {


maybe just call this public BytesReference materializeAndClose() throws IOException and don't even return the stream.

What about:

public BytesReference materialize() throws IOException

or

public BytesReference materializeBytes() throws IOException

I just kind of feel weird about materializeAndClose() because calling that does not release you from the obligation that you still must call the regular close().

s1monw · 2017-05-29T12:26:37Z

core/src/main/java/org/elasticsearch/transport/CompressibleBytesOutputStream.java

+
+import java.io.IOException;
+
+public class CompressibleBytesOutputStream extends StreamOutput implements Releasable {


can we have a simple unittest for this class?

nik9000

I left some nit picky things, mostly because it is fun to have opinions. Feel free to ignore.

nik9000 · 2017-05-29T13:22:50Z

core/src/main/java/org/elasticsearch/transport/CompressibleBytesOutputStream.java

+        this.bytesStreamOutput = bytesStreamOutput;
+        this.shouldCompress = shouldCompress;
+        if (shouldCompress) {
+            this.stream = CompressorFactory.COMPRESSOR.streamOutput(Streams.flushOnCloseStream(bytesStreamOutput));


Personally I don't like mixing this.stream with stream. It makes me think you are shadowing and need this. sometimes. No big deal, but it slows down reading to have to follow that little mental dead end.

nik9000 · 2017-05-29T13:25:01Z

core/src/main/java/org/elasticsearch/transport/CompressibleBytesOutputStream.java

+
+    @Override
+    public void close() {
+        // If we are not using compression stream == bytesStreamOutput


I wonder if it'd actually be clearer not to have shouldCompress and instead check for reference equality here.

++ also add an assertion that check that shouldCompress has the corresponding value?

nik9000 · 2017-05-29T13:26:17Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

        boolean addedReleaseListener = false;
-        StreamOutput stream = Streams.flushOnCloseStream(bStream);
        try {
            // only compress if asked, and, the request is not bytes, since then only


You probably should move this comment.

jaymode

This is a nice improvement! left a few comments

jaymode · 2017-05-30T13:32:53Z

core/src/main/java/org/elasticsearch/transport/CompressibleBytesOutputStream.java

+
+import java.io.IOException;
+
+final class CompressibleBytesOutputStream extends StreamOutput implements Releasable {


can you add a class level javadoc about the class and why/what it is used for

jaymode · 2017-05-30T13:37:56Z

core/src/main/java/org/elasticsearch/transport/CompressibleBytesOutputStream.java

+
+    @Override
+    public void close() {
+        if (stream == bytesStreamOutput) {


maybe use shouldCompress here to make it a little clearer?

if (shouldCompress) { IOUtils.closeWhileHandlingException(stream, bytesStreamOutput); } else { assert stream == bytesStreamOutput : "the stream variable is not the same instance as bytesStreamOutput"; IOUtils.closeWhileHandlingException(stream); }

jaymode · 2017-05-30T13:38:26Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

+
+        // only compress if asked and the request is not bytes. Otherwise only
+        // the header part is compressed, and the "body" can't be extracted as compressed
+        boolean compressMessage = options.compress() && canCompress(request);


make this final?

jaymode · 2017-05-30T13:39:58Z

core/src/test/java/org/elasticsearch/transport/CompressibleBytesOutputStreamTests.java

+            StreamInput streamInput = CompressorFactory.COMPRESSOR.streamInput(bStream.bytes().streamInput());
+            byte[] actualBytes = new byte[expectedBytes.length];
+            streamInput.readBytes(actualBytes, 0, expectedBytes.length);
+            fail("Expected to receive EOFException");


use expectThrows?

jaymode · 2017-05-30T13:40:50Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

-        // this might be a problem in deflate after all but it's important to close it for now.
-        stream.close();
-        final BytesReference messageBody = writtenBytes.bytes();
+        // we have to call finishStream here before accessing the bytes. A CompressibleBytesOutputStream


s/finishStream/materializeBytes

jaymode · 2017-05-30T13:41:08Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

-        stream.close();
-        final BytesReference messageBody = writtenBytes.bytes();
+        // we have to call finishStream here before accessing the bytes. A CompressibleBytesOutputStream
+        // might be implementing compression. And finish stream ensures that some marker bytes (EOS marker)


s/finish stream/materialize bytes

jaymode

LGTM

Tim-Brooks · 2017-05-31T02:18:52Z

@s1monw Are you good with these changes?

s1monw

left a suggestion otehrwise LGTM nice simplification

s1monw · 2017-05-31T08:41:40Z

core/src/main/java/org/elasticsearch/transport/CompressibleBytesOutputStream.java

+            stream.close();
+        }
+
+        return bytesStreamOutput.bytes();


should we also close the bytesStreamOutput after this? or can we simply call close() before it seems like it's safe for the BytesStreamOutput#close() method?

I don't think so. The bytesStreamOutput is still being used at that point. That's why this is separating calling the compressing stream and the bytes stream. Must close the compressing stream to write EOS bytes. But wait until we are done using the underlying bytes to close the byte stream.

Tim-Brooks · 2017-05-31T15:40:56Z

Alright once ci finishes I will merge this.

I'm not sure if you all noticed, but

I started by checking shouldCompress boolean to determine what stream I should close.
@nik9000 suggested that I check stream identity to determine what stream I should close. I made that change.
@jaymode suggested I check shouldCompress boolean to determine what stream I should close. I made the change back.
@s1monw suggested that I check stream identity to determine what stream I should close. And add an assertion on the boolean.

I left it on Simon's suggestion which kind of seemed like a compromise between the differing suggestions.

This is related to elastic#24927. There was a small possibility that a test was attempting to compress a stream with zero bytes. This was causing a failure. This test now requires at least one byte.

This is related to #24927. There was a small possibility that a test was attempting to compress a stream with zero bytes. This was causing a failure. This test now requires at least one byte.

This is a follow-up to elastic#23941. Currently there are a number of complexities related to compression. The raw DeflaterOutputStream must be closed prior to sending bytes to ensure that EOS bytes are written. But the underlying ReleasableBytesStreamOutput cannot be closed until the bytes are sent to ensure that the bytes are not reused. Right now we have three different stream references hanging around in TCPTransport to handle this complexity. This commit introduces CompressibleBytesOutputStream to be one stream implemenation that will behave properly with or without compression enabled.

This is related to #24927. There was a small possibility that a test was attempting to compress a stream with zero bytes. This was causing a failure. This test now requires at least one byte.

Tim-Brooks added 4 commits May 27, 2017 15:16

Add first stream implementation

3829a84

Use one stream in tcp transport

5e73e97

Slightly rework streams

467585a

Fix checkstyle and compress boolean logic

85d42dc

Tim-Brooks added :Distributed/Network Http and internode communication implementations >non-issue review v6.0.0 labels May 28, 2017

Tim-Brooks requested review from s1monw, jaymode and jasontedor May 28, 2017 23:01

Tim-Brooks added 2 commits May 28, 2017 18:14

Pass stream to constructor

54cee2f

Merge remote-tracking branch 'upstream/master' into rework_stream

d571f7c

s1monw requested changes May 29, 2017

View reviewed changes

nik9000 reviewed May 29, 2017

View reviewed changes

Tim-Brooks added 2 commits May 29, 2017 16:48

Make a number of changes based on review

a84e7e3

Add test

43bd03d

jaymode requested changes May 30, 2017

View reviewed changes

Tim-Brooks added 2 commits May 30, 2017 11:04

Make changes based on review

d121346

Use expectThrows

fdb58f0

jaymode approved these changes May 30, 2017

View reviewed changes

Merge remote-tracking branch 'upstream/master' into rework_stream

79c0f3d

s1monw approved these changes May 31, 2017

View reviewed changes

Add assertions

7f1af8d

Tim-Brooks merged commit 90a5574 into elastic:master May 31, 2017

Tim-Brooks deleted the rework_stream branch May 31, 2017 16:00

Tim-Brooks mentioned this pull request May 31, 2017

Fix broken build from stream with zero bytes #24993

Merged

clintongormley added v6.0.0-alpha2 and removed v6.0.0 labels Jun 6, 2017

jasontedor mentioned this pull request Nov 25, 2017

Circuit breaker grows indefinitely when >2GiB of mget is issued (and possibly at other places as well) #27525

Closed

This was referenced Nov 27, 2017

Add CompressibleBytesOutputStream for compression #27540

Merged

Throw UOE from compressible bytes stream reset #27564

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add CompressibleBytesOutputStream for compression #24927

Add CompressibleBytesOutputStream for compression #24927

Tim-Brooks commented May 28, 2017

Tim-Brooks commented May 28, 2017 •

edited

Loading

s1monw left a comment

s1monw May 29, 2017

s1monw May 29, 2017

Tim-Brooks May 29, 2017 •

edited

Loading

s1monw May 29, 2017

nik9000 left a comment

nik9000 May 29, 2017

nik9000 May 29, 2017

s1monw May 31, 2017

nik9000 May 29, 2017

jaymode left a comment

jaymode May 30, 2017

jaymode May 30, 2017

jaymode May 30, 2017

jaymode May 30, 2017

jaymode May 30, 2017

jaymode May 30, 2017

jaymode left a comment

Tim-Brooks commented May 31, 2017

s1monw left a comment

s1monw May 31, 2017

Tim-Brooks May 31, 2017

Tim-Brooks commented May 31, 2017 •

edited

Loading


		import java.io.IOException;

		public class CompressibleBytesOutputStream extends StreamOutput implements Releasable {


		import java.io.IOException;

		final class CompressibleBytesOutputStream extends StreamOutput implements Releasable {

Add CompressibleBytesOutputStream for compression #24927

Add CompressibleBytesOutputStream for compression #24927

Conversation

Tim-Brooks commented May 28, 2017

Tim-Brooks commented May 28, 2017 • edited Loading

s1monw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Tim-Brooks May 29, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nik9000 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jaymode left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jaymode left a comment

Choose a reason for hiding this comment

Tim-Brooks commented May 31, 2017

s1monw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Tim-Brooks commented May 31, 2017 • edited Loading

Tim-Brooks commented May 28, 2017 •

edited

Loading

Tim-Brooks May 29, 2017 •

edited

Loading

Tim-Brooks commented May 31, 2017 •

edited

Loading