alex-plekhanov commented on code in PR #13329:
URL: https://github.com/apache/ignite/pull/13329#discussion_r3572520464
##########
modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java:
##########
@@ -503,24 +522,49 @@ private <T> T
readCompressedMessageAndDeserialize(DirectByteBufferStream stream,
if (msg0.dataSize() == 0)
return null;
- byte[] uncompressed = msg0.uncompressed();
+ // Reuse the temp reader across fields/messages instead of allocating
a fresh state stack each time.
+ if (tmpReader == null)
+ tmpReader = new DirectMessageReader(msgFactory, cacheObjProc);
+ else
+ tmpReader.reset();
- ByteBuffer tmpBuf = ByteBuffer.allocateDirect(uncompressed.length);
+ tmpReader.setBuffer(ByteBuffer.wrap(msg0.uncompressed()));
- tmpBuf.put(uncompressed);
- tmpBuf.flip();
+ T res;
- DirectMessageReader tmpReader = new DirectMessageReader(msgFactory,
cacheObjProc);
+ boolean ok = false;
- tmpReader.setBuffer(tmpBuf);
+ try {
+ res = fun.apply(tmpReader);
- T res = fun.apply(tmpReader);
+ // The payload buffer is always complete, so a partial read means
a corrupted stream, not a lack of data.
+ if (!tmpReader.state.item().stream.lastFinished())
+ throw new IgniteException("Failed to deserialize compressed
payload: uncompressed data ended " +
+ "unexpectedly [dataSize=" + msg0.dataSize() + ']');
Review Comment:
Braces required here
##########
modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java:
##########
@@ -469,12 +484,16 @@ public ByteBuffer getBuffer() {
@Override public void beforeNestedRead() {
state.forward();
- state.item().stream.setBuffer(buf);
+ curStream = state.item().stream;
+
+ curStream.setBuffer(buf);
}
/** {@inheritDoc} */
@Override public void afterNestedRead(boolean finished) {
state.backward(finished);
+
+ curStream = state.item().stream;
Review Comment:
Do we need additional `curStream.setBuffer(buf);` call here?
Looks like currently buffer instance can't be changed during message
deserialization (`setBuffer` called with the same `buf` instance on each part
of message). At least for communication messages , but I don't analize other
code paths. But if we change buffer when deserializing nested message there can
be an issue when previous item holds reference to the old buffer. Maybe it's
safer to call `curStream.setBuffer(buf);` here too?
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/CompressedMessage.java:
##########
@@ -90,99 +86,106 @@ public byte[] uncompressed() {
/** @return Next chunk of data or null. */
public byte[] nextChunk() {
- return chunkedReader.nextChunk();
+ return chunkIdx < chunks.size() ? chunks.get(chunkIdx++) : null;
}
- /**
- * @param buf Buffer.
- * @return Compressed data.
- */
- private byte[] compress(ByteBuffer buf) {
- byte[] data = new byte[dataSize];
+ /** @param buf Buffer. */
+ private void compress(ByteBuffer buf) {
+ Deflater deflater = new Deflater(compressionLvl, true);
- buf.get(data);
+ try {
+ deflater.setInput(buf);
+ deflater.finish();
- ByteArrayOutputStream baos = new ByteArrayOutputStream(data.length);
- Deflater deflater = new Deflater(compressionLvl, true);
+ chunks = new ArrayList<>(dataSize / CHUNK_SIZE + 1);
- try (DeflaterOutputStream dos = new DeflaterOutputStream(baos,
deflater)) {
- dos.write(data);
- dos.finish();
- }
- catch (IOException ex) {
- throw new IgniteException(ex);
+ // Incompressible data may expand: raw-deflate worst case below
CHUNK_SIZE is one stored block (~11 bytes overhead).
+ byte[] chunk0 = new byte[dataSize >= CHUNK_SIZE ? CHUNK_SIZE :
dataSize + 16];
+
+ int len = 0;
+
+ while (!deflater.finished()) {
+ len += deflater.deflate(chunk0, len, chunk0.length - len);
+
+ if (len == chunk0.length && !deflater.finished()) {
+ chunks.add(chunk0);
+
+ chunk0 = new byte[CHUNK_SIZE];
+ len = 0;
+ }
+ }
+
+ if (len > 0)
+ chunks.add(len == chunk0.length ? chunk0 :
Arrays.copyOf(chunk0, len));
}
finally {
deflater.end();
}
-
- return baos.toByteArray();
}
/** @return Uncompressed data. */
private byte[] uncompress() {
- if (tmpBuf == null)
- return null;
+ if (chunks == null)
+ throw new IgniteException("Compressed stream is truncated
[expected=" + dataSize + ", inflated=0]");
+
+ long compressedTotal = 0;
- byte[] uncompressedData;
+ for (int i = 0; i < chunks.size(); i++)
+ compressedTotal += chunks.get(i).length;
+
+ // Raw deflate cannot expand beyond ~1032:1, a larger claim means a
corrupted size header; also bounds the
+ // upfront allocation by ~1032x of the actually received bytes.
+ if (dataSize > compressedTotal * 1032L + 64)
Review Comment:
Introduce constant?
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/CompressedMessageSerializer.java:
##########
@@ -110,23 +110,18 @@ public class CompressedMessageSerializer implements
MessageSerializer<Compressed
if (!reader.isLastRead())
return false;
- if (msg.chunk != null) {
- if (msg.tmpBuf.remaining() <= CHUNK_SIZE) {
- ByteBuffer newTmpBuf =
ByteBuffer.allocateDirect(msg.tmpBuf.capacity() * 2);
-
- msg.tmpBuf.flip();
+ if (msg.chunk == null)
+ throw new IgniteException("Failed to read compressed
message: unexpected null chunk " +
+ "(stream is corrupted or the sender is
incompatible).");
Review Comment:
Braces required
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/CompressedMessage.java:
##########
@@ -90,99 +86,106 @@ public byte[] uncompressed() {
/** @return Next chunk of data or null. */
public byte[] nextChunk() {
- return chunkedReader.nextChunk();
+ return chunkIdx < chunks.size() ? chunks.get(chunkIdx++) : null;
}
- /**
- * @param buf Buffer.
- * @return Compressed data.
- */
- private byte[] compress(ByteBuffer buf) {
- byte[] data = new byte[dataSize];
+ /** @param buf Buffer. */
+ private void compress(ByteBuffer buf) {
+ Deflater deflater = new Deflater(compressionLvl, true);
- buf.get(data);
+ try {
+ deflater.setInput(buf);
+ deflater.finish();
- ByteArrayOutputStream baos = new ByteArrayOutputStream(data.length);
- Deflater deflater = new Deflater(compressionLvl, true);
+ chunks = new ArrayList<>(dataSize / CHUNK_SIZE + 1);
- try (DeflaterOutputStream dos = new DeflaterOutputStream(baos,
deflater)) {
- dos.write(data);
- dos.finish();
- }
- catch (IOException ex) {
- throw new IgniteException(ex);
+ // Incompressible data may expand: raw-deflate worst case below
CHUNK_SIZE is one stored block (~11 bytes overhead).
+ byte[] chunk0 = new byte[dataSize >= CHUNK_SIZE ? CHUNK_SIZE :
dataSize + 16];
+
+ int len = 0;
+
+ while (!deflater.finished()) {
+ len += deflater.deflate(chunk0, len, chunk0.length - len);
+
+ if (len == chunk0.length && !deflater.finished()) {
+ chunks.add(chunk0);
+
+ chunk0 = new byte[CHUNK_SIZE];
+ len = 0;
+ }
+ }
+
+ if (len > 0)
+ chunks.add(len == chunk0.length ? chunk0 :
Arrays.copyOf(chunk0, len));
}
finally {
deflater.end();
}
-
- return baos.toByteArray();
}
/** @return Uncompressed data. */
private byte[] uncompress() {
- if (tmpBuf == null)
- return null;
+ if (chunks == null)
+ throw new IgniteException("Compressed stream is truncated
[expected=" + dataSize + ", inflated=0]");
+
+ long compressedTotal = 0;
- byte[] uncompressedData;
+ for (int i = 0; i < chunks.size(); i++)
+ compressedTotal += chunks.get(i).length;
+
+ // Raw deflate cannot expand beyond ~1032:1, a larger claim means a
corrupted size header; also bounds the
+ // upfront allocation by ~1032x of the actually received bytes.
+ if (dataSize > compressedTotal * 1032L + 64)
+ throw new IgniteException("Invalid compressed message data size
[dataSize=" + dataSize +
+ ", compressedBytes=" + compressedTotal + ']');
Review Comment:
Braces required
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
