This is an automated email from the ASF dual-hosted git repository.
smengcl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 0c349a3f541 HDDS-15356. Make multi-buffer chunk checksum
allocation-free (#10350)
0c349a3f541 is described below
commit 0c349a3f54166a15a061b92041831194bd89a01d
Author: Siyao Meng <[email protected]>
AuthorDate: Wed Aug 12 18:39:23 2026 -0700
HDDS-15356. Make multi-buffer chunk checksum allocation-free (#10350)
---
.../org/apache/hadoop/ozone/common/Checksum.java | 186 ++++++++++++++-------
.../apache/hadoop/ozone/common/ChecksumCache.java | 117 +++++++++----
.../hadoop/ozone/common/utils/BufferUtils.java | 23 +++
.../apache/hadoop/ozone/common/TestChecksum.java | 17 ++
.../hadoop/ozone/common/TestChecksumCache.java | 73 +++++++-
.../ozone/common/TestChecksumMultiBuffer.java | 97 +++++++++++
.../hadoop/ozone/common/utils/TestBufferUtils.java | 81 +++++++++
.../ozone/container/keyvalue/KeyValueHandler.java | 4 +-
8 files changed, 490 insertions(+), 108 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/Checksum.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/Checksum.java
index 6a530cacc9a..f4010497774 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/Checksum.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/Checksum.java
@@ -24,7 +24,6 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType;
@@ -53,57 +52,108 @@ public class Checksum {
*/
private final ChecksumCache checksumCache;
- private static Function<ByteBuffer, ByteString> newMessageDigestFunction(
- String algorithm) {
- final MessageDigest md;
+ private static MessageDigest newMessageDigest(String algorithm) {
try {
- md = MessageDigest.getInstance(algorithm);
+ return MessageDigest.getInstance(algorithm);
} catch (NoSuchAlgorithmException e) {
throw new IllegalStateException(
"Failed to get MessageDigest for " + algorithm, e);
}
- return data -> {
- md.reset();
- md.update(data);
- return ByteString.copyFrom(md.digest());
- };
}
public static ByteString int2ByteString(int n) {
return UnsafeByteOperations.unsafeWrap(IntegerCodec.get().toByteArray(n));
}
- private static Function<ByteBuffer, ByteString>
newChecksumByteBufferFunction(
- Supplier<ChecksumByteBuffer> constructor) {
- final ChecksumByteBuffer algorithm = constructor.get();
- return data -> {
- algorithm.reset();
- algorithm.update(data);
- return int2ByteString((int)algorithm.getValue());
+ /**
+ * Streaming checksum strategy: feed multiple ByteBuffer slices via
+ * {@link #update}, then read the result via {@link #finish}, then
+ * {@link #reset} to start a new window. Used by both the no-cache and
+ * cache compute paths to avoid the {@code byte[bytesPerChecksum]}
+ * allocation that {@code ChunkBuffer.iterate} performs whenever a
+ * checksum window straddles multiple underlying buffers - the
+ * {@link ChecksumByteBuffer#update(ByteBuffer)} and
+ * {@link MessageDigest#update(ByteBuffer)} contracts both define
+ * incremental updates as byte-equivalent to a single update over the
+ * concatenation.
+ */
+ interface StreamingChecksum {
+ void reset();
+
+ void update(ByteBuffer slice);
+
+ ByteString finish();
+ }
+
+ private static StreamingChecksum streamingCrc(
+ Supplier<ChecksumByteBuffer> ctor) {
+ final ChecksumByteBuffer cb = ctor.get();
+ return new StreamingChecksum() {
+ @Override
+ public void reset() {
+ cb.reset();
+ }
+
+ @Override
+ public void update(ByteBuffer slice) {
+ cb.update(slice);
+ }
+
+ @Override
+ public ByteString finish() {
+ return int2ByteString((int) cb.getValue());
+ }
+ };
+ }
+
+ private static StreamingChecksum streamingDigest(String algorithm) {
+ final MessageDigest md = newMessageDigest(algorithm);
+ return new StreamingChecksum() {
+ @Override
+ public void reset() {
+ md.reset();
+ }
+
+ @Override
+ public void update(ByteBuffer slice) {
+ md.update(slice);
+ }
+
+ @Override
+ public ByteString finish() {
+ // The JCA SPI does not guarantee exclusive ownership of the digest
+ // array, while Protobuf unsafeWrap requires exclusive, immutable
+ // ownership. Copying 16 bytes for MD5 or 32 bytes for SHA-256 is
cheap.
+ return ByteString.copyFrom(md.digest());
+ }
};
}
/** The algorithms for {@link ChecksumType}. */
enum Algorithm {
- NONE(() -> data -> ByteString.EMPTY),
- CRC32(() ->
- newChecksumByteBufferFunction(ChecksumByteBufferFactory::crc32Impl)),
- CRC32C(() ->
- newChecksumByteBufferFunction(ChecksumByteBufferFactory::crc32CImpl)),
- SHA256(() -> newMessageDigestFunction("SHA-256")),
- MD5(() -> newMessageDigestFunction("MD5"));
-
- private final Supplier<Function<ByteBuffer, ByteString>> constructor;
+ // NONE is reachable via Algorithm.valueOf(ChecksumType.NONE) only if
+ // computeChecksum's NONE short-circuit is bypassed; throw to surface
+ // such a misuse rather than silently producing empty checksums.
+ NONE(() -> {
+ throw new UnsupportedOperationException(
+ "ChecksumType.NONE has no StreamingChecksum");
+ }),
+ CRC32(() -> streamingCrc(ChecksumByteBufferFactory::crc32Impl)),
+ CRC32C(() -> streamingCrc(ChecksumByteBufferFactory::crc32CImpl)),
+ SHA256(() -> streamingDigest("SHA-256")),
+ MD5(() -> streamingDigest("MD5"));
+
+ private final Supplier<StreamingChecksum> constructor;
static Algorithm valueOf(ChecksumType type) {
return valueOf(type.name());
}
- Algorithm(Supplier<Function<ByteBuffer, ByteString>> constructor) {
+ Algorithm(Supplier<StreamingChecksum> constructor) {
this.constructor = constructor;
}
- Function<ByteBuffer, ByteString> newChecksumFunction() {
+ StreamingChecksum newStreamingChecksum() {
return constructor.get();
}
}
@@ -218,65 +268,71 @@ public ChecksumData computeChecksum(ChunkBuffer data)
return computeChecksum(data, false);
}
+ /**
+ * This method does not advance the positions of {@code data}'s underlying
+ * buffers. Both the no-cache and cache paths slice via
+ * {@link ByteBuffer#duplicate()}.
+ */
public ChecksumData computeChecksum(ChunkBuffer data, boolean useCache)
throws OzoneChecksumException {
if (checksumType == ChecksumType.NONE) {
- // Since type is set to NONE, we do not need to compute the checksums
return new ChecksumData(checksumType, bytesPerChecksum);
}
- final Function<ByteBuffer, ByteString> function;
+ final StreamingChecksum algo;
try {
- function = Algorithm.valueOf(checksumType).newChecksumFunction();
+ algo = Algorithm.valueOf(checksumType).newStreamingChecksum();
} catch (Exception e) {
- throw new OzoneChecksumException("Failed to get the checksum function
for " + checksumType, e);
+ throw new OzoneChecksumException(
+ "Failed to create streaming checksum for " + checksumType, e);
}
- final List<ByteString> checksumList;
- if (checksumCache == null || !useCache) {
- // When checksumCache is not enabled:
- // Checksum is computed for each bytesPerChecksum number of bytes of data
- // starting at offset 0. The last checksum might be computed for the
- // remaining data with length less than bytesPerChecksum.
- checksumList = new ArrayList<>();
- for (ByteBuffer b : data.iterate(bytesPerChecksum)) {
- checksumList.add(computeChecksum(b, function, bytesPerChecksum)); //
merge this?
- }
- } else {
- // When checksumCache is enabled:
- // We only need to update the last checksum in the cache, then pass it
along.
- checksumList = checksumCache.computeChecksum(data, function);
- }
+ final List<ByteString> checksumList = (checksumCache == null || !useCache)
+ ? computeChecksumDirect(data, algo)
+ : checksumCache.computeChecksum(data, algo, bytesPerChecksum);
return new ChecksumData(checksumType, bytesPerChecksum, checksumList);
}
/**
- * Compute checksum using the algorithm for the data upto the max length.
- * @param data input data
- * @param function the checksum function
- * @param maxLength the max length of data
- * @return computed checksum ByteString
+ * Walk {@code data}'s underlying ByteBuffer list, slicing each window of
+ * {@link #bytesPerChecksum} bytes via {@link ByteBuffer#duplicate()} and
+ * feeding slices to {@code algo}. No linearization byte[] is allocated
+ * when a window straddles multiple buffers.
*/
- protected static ByteString computeChecksum(ByteBuffer data,
- Function<ByteBuffer, ByteString> function, int maxLength) {
- final int limit = data.limit();
- try {
- final int maxIndex = data.position() + maxLength;
- if (limit > maxIndex) {
- data.limit(maxIndex);
+ private List<ByteString> computeChecksumDirect(ChunkBuffer data,
+ StreamingChecksum algo) {
+ final int dataLength = data.remaining();
+ final int checksumCount = dataLength == 0 ? 0 : 1 + (dataLength - 1) /
bytesPerChecksum;
+ final List<ByteString> result = new ArrayList<>(checksumCount);
+ int windowRemaining = bytesPerChecksum;
+ algo.reset();
+
+ for (ByteBuffer src : data.asByteBufferList()) {
+ int srcPos = src.position();
+ final int srcLim = src.limit();
+ while (srcPos < srcLim) {
+ final int n = Math.min(srcLim - srcPos, windowRemaining);
+ algo.update(BufferUtils.slice(src, srcPos, n));
+ srcPos += n;
+ windowRemaining -= n;
+ if (windowRemaining == 0) {
+ result.add(algo.finish());
+ algo.reset();
+ windowRemaining = bytesPerChecksum;
+ }
}
- return function.apply(data);
- } finally {
- data.limit(limit);
}
+ if (windowRemaining < bytesPerChecksum) {
+ // Unaligned trailing window.
+ result.add(algo.finish());
+ }
+ return result;
}
public static void verifySingleChecksum(ByteBuffer buffer, int offset, int
bytesPerChecksum,
ByteString checksum, ChecksumType checksumType) throws
OzoneChecksumException {
- final ByteBuffer duplicated = buffer.duplicate();
- duplicated.position(offset).limit(offset + bytesPerChecksum);
final ChecksumData cd = new ChecksumData(checksumType, bytesPerChecksum,
Collections.singletonList(checksum));
- verifyChecksum(duplicated, cd, 0);
+ verifyChecksum(BufferUtils.slice(buffer, offset, bytesPerChecksum), cd, 0);
}
/**
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChecksumCache.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChecksumCache.java
index 3b96b35eba6..b59fbdb3f04 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChecksumCache.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChecksumCache.java
@@ -17,10 +17,12 @@
package org.apache.hadoop.ozone.common;
+import com.google.common.base.Preconditions;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
-import java.util.function.Function;
+import org.apache.hadoop.ozone.common.Checksum.StreamingChecksum;
+import org.apache.hadoop.ozone.common.utils.BufferUtils;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -65,58 +67,105 @@ public List<ByteString> getChecksums() {
return checksums;
}
- public List<ByteString> computeChecksum(ChunkBuffer data,
Function<ByteBuffer, ByteString> function) {
- // Indicates how much data the current chunk buffer holds
- final int currChunkLength = data.limit();
+ /**
+ * Recompute checksums for the windows that have changed since the last
+ * call: bytes {@code [ciStart * bytesPerChecksum, currChunkLength)} where
+ * {@code ciStart = prevChunkLength / bytesPerChecksum} (the index of the
+ * first window whose result may have changed - either the previously-
+ * partial last window now has more bytes, or new full windows have been
+ * appended).
+ *
+ * <p>Walks {@code data}'s underlying buffer list directly (no
+ * {@code iterate()} byte[] linearization) and feeds slices to {@code algo}
+ * incrementally; the cached prefix is skipped via index arithmetic so
+ * those bytes are never re-fed.
+ */
+ List<ByteString> computeChecksum(ChunkBuffer data,
+ StreamingChecksum algo, int chksumSize) {
+ if (chksumSize != bytesPerChecksum) {
+ throw new IllegalArgumentException("bytesPerChecksum mismatch: cache="
+ + bytesPerChecksum + " call=" + chksumSize);
+ }
+ final int currChunkLength = data.remaining();
if (currChunkLength == prevChunkLength) {
- LOG.debug("ChunkBuffer data limit same as last time ({}). No new
checksums need to be computed", prevChunkLength);
+ LOG.debug("ChunkBuffer data length same as last time ({}). "
+ + "No new checksums need to be computed", prevChunkLength);
return checksums;
}
-
- // Sanity check
if (currChunkLength < prevChunkLength) {
- // If currChunkLength <= lastChunkLength, it indicates a bug that needs
to be addressed.
- // It means BOS has not properly clear()ed the cache when a new chunk is
started in that code path.
- throw new IllegalArgumentException("ChunkBuffer data limit (" +
currChunkLength + ")" +
- " must not be smaller than last time (" + prevChunkLength + ")");
+ // Indicates a bug: BOS did not clear() the cache before starting a new
chunk.
+ throw new IllegalArgumentException("ChunkBuffer data length (" +
currChunkLength + ")"
+ + " must not be smaller than last time (" + prevChunkLength + ")");
}
- // One or more checksums need to be computed
-
- // Start of the checksum index that need to be (re)computed
+ // Index of the first window that needs (re)computing.
final int ciStart = prevChunkLength / bytesPerChecksum;
- final int ciEnd = currChunkLength / bytesPerChecksum + (currChunkLength %
bytesPerChecksum == 0 ? 0 : 1);
- int i = 0;
- for (ByteBuffer b : data.iterate(bytesPerChecksum)) {
- if (i < ciStart) {
- i++;
+ final int ciEnd = currChunkLength / bytesPerChecksum
+ + (currChunkLength % bytesPerChecksum == 0 ? 0 : 1);
+ // Bytes to skip (the cached full windows preceding ciStart).
+ long bytesToSkip = (long) ciStart * bytesPerChecksum;
+
+ int i = ciStart;
+ int windowRemaining = bytesPerChecksum;
+ algo.reset();
+ long position = 0;
+
+ for (ByteBuffer src : data.asByteBufferList()) {
+ int srcPos = src.position();
+ final int srcLim = src.limit();
+ final int srcLen = srcLim - srcPos;
+
+ // Fast-forward through buffers that lie entirely within the cached
+ // prefix.
+ if (position + srcLen <= bytesToSkip) {
+ position += srcLen;
continue;
}
-
- // variable i can either point to:
- // 1. the last element in the list -- in which case the checksum needs
to be updated
- // 2. one after the last element -- in which case a new checksum needs
to be added
- assert i == checksums.size() - 1 || i == checksums.size();
-
- // TODO: Furthermore for CRC32/CRC32C, it can be even more efficient by
updating the last checksum byte-by-byte.
- final ByteString checksum = Checksum.computeChecksum(b, function,
bytesPerChecksum);
- if (i == checksums.size()) {
- checksums.add(checksum);
- } else {
- checksums.set(i, checksum);
+ // First buffer that crosses into the not-yet-cached region: advance
+ // srcPos to the boundary.
+ if (position < bytesToSkip) {
+ srcPos += (int) (bytesToSkip - position);
+ position = bytesToSkip;
}
- i++;
+ while (srcPos < srcLim) {
+ final int n = Math.min(srcLim - srcPos, windowRemaining);
+ algo.update(BufferUtils.slice(src, srcPos, n));
+ srcPos += n;
+ position += n;
+ windowRemaining -= n;
+ if (windowRemaining == 0) {
+ storeChecksum(i++, algo.finish());
+ algo.reset();
+ windowRemaining = bytesPerChecksum;
+ }
+ }
+ }
+ if (windowRemaining < bytesPerChecksum) {
+ // Unaligned trailing window.
+ storeChecksum(i++, algo.finish());
}
- // Sanity check
if (i != ciEnd) {
throw new IllegalStateException("ChecksumCache: Checksum index end does
not match expectation");
}
- // Update last written index
prevChunkLength = currChunkLength;
return checksums;
}
+
+ private void storeChecksum(int i, ByteString cs) {
+ // i can either point to the last cached element (recompute - the
+ // previously-partial trailing window) or one past it (append a new
+ // checksum for newly-arrived bytes).
+ Preconditions.checkState(
+ i == checksums.size() - 1 || i == checksums.size(),
+ "Unexpected checksum index %s for cache size %s", i, checksums.size());
+ if (i == checksums.size()) {
+ checksums.add(cs);
+ } else {
+ checksums.set(i, cs);
+ }
+ }
}
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/utils/BufferUtils.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/utils/BufferUtils.java
index 794504b1ef3..b05d2dd9a45 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/utils/BufferUtils.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/utils/BufferUtils.java
@@ -23,6 +23,7 @@
import java.nio.channels.GatheringByteChannel;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,6 +69,28 @@ public static ByteBuffer[] assignByteBuffers(long totalLen,
return dataBuffers;
}
+ /**
+ * Return a non-copying {@link ByteBuffer#duplicate()} of {@code src} that
+ * covers exactly {@code [position, position + length)}. Read-only-ness
+ * and direct/heap kind are inherited from {@code src}.
+ */
+ public static ByteBuffer slice(ByteBuffer src, int position, int length) {
+ Objects.requireNonNull(src, "src must not be null");
+ Preconditions.checkArgument(position >= 0,
+ "position (%s) must not be negative", position);
+ Preconditions.checkArgument(length >= 0,
+ "length (%s) must not be negative", length);
+ Preconditions.checkArgument(position <= src.limit(),
+ "position (%s) exceeds source limit (%s)", position, src.limit());
+ Preconditions.checkArgument(length <= src.limit() - position,
+ "position (%s) + length (%s) exceeds source limit (%s)",
+ position, length, src.limit());
+ final ByteBuffer slice = src.duplicate();
+ slice.position(position);
+ slice.limit(position + length);
+ return slice;
+ }
+
/**
* Return a read only ByteBuffer list for the input ByteStrings list.
*/
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksum.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksum.java
index eb11fe53d9c..ce597971c04 100644
---
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksum.java
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksum.java
@@ -20,10 +20,14 @@
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import java.nio.ByteBuffer;
+import java.util.Collections;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -100,4 +104,17 @@ public void
testChecksumMismatchForDifferentChecksumTypes(boolean useChecksumCac
// The two checksums should not match as they have different types
assertNotEquals(checksum1, checksum2, "Checksums should not match for
different checksum types");
}
+
+ @Test
+ public void testChecksumFromNonzeroPosition() throws Exception {
+ final ChunkBuffer data = mock(ChunkBuffer.class);
+ when(data.position()).thenReturn(Integer.MAX_VALUE - 1);
+ when(data.limit()).thenReturn(Integer.MAX_VALUE);
+ when(data.remaining()).thenReturn(1);
+ when(data.asByteBufferList()).thenReturn(
+ Collections.singletonList(ByteBuffer.wrap(new byte[] {1})));
+
+ final Checksum checksum = getChecksum(null, false);
+ assertEquals(1, checksum.computeChecksum(data).getChecksums().size());
+ }
}
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksumCache.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksumCache.java
index 0b7e9a7b198..0d0253c2130 100644
---
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksumCache.java
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksumCache.java
@@ -18,10 +18,11 @@
package org.apache.hadoop.ozone.common;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.List;
-import java.util.function.Function;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType;
import org.apache.hadoop.ozone.common.Checksum.Algorithm;
+import org.apache.hadoop.ozone.common.Checksum.StreamingChecksum;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.params.ParameterizedTest;
@@ -33,7 +34,7 @@
class TestChecksumCache {
@ParameterizedTest
- @EnumSource(ChecksumType.class)
+ @EnumSource(value = ChecksumType.class, names = {"CRC32", "CRC32C",
"SHA256", "MD5"})
void testComputeChecksum(ChecksumType checksumType) throws Exception {
final int bytesPerChecksum = 16;
ChecksumCache checksumCache = new ChecksumCache(bytesPerChecksum);
@@ -45,7 +46,7 @@ void testComputeChecksum(ChecksumType checksumType) throws
Exception {
byteArray[i] = (byte) (i % 128);
}
- final Function<ByteBuffer, ByteString> function =
Algorithm.valueOf(checksumType).newChecksumFunction();
+ final StreamingChecksum algo =
Algorithm.valueOf(checksumType).newStreamingChecksum();
int iEnd = size / bytesPerChecksum + (size % bytesPerChecksum == 0 ? 0 :
1);
List<ByteString> lastRes = null;
@@ -54,19 +55,75 @@ void testComputeChecksum(ChecksumType checksumType) throws
Exception {
ByteBuffer byteBuffer = ByteBuffer.wrap(byteArray, 0, byteBufferLength);
try (ChunkBuffer chunkBuffer =
ChunkBuffer.wrap(byteBuffer.asReadOnlyBuffer())) {
- List<ByteString> res = checksumCache.computeChecksum(chunkBuffer,
function);
- System.out.println(res);
- // Verify that every entry in the res list except the last one is the
same as the one in lastRes list
+ List<ByteString> res = checksumCache.computeChecksum(chunkBuffer,
algo, bytesPerChecksum);
+ // Every entry except the last must be unchanged from the prior
iteration
+ // (those windows are fully cached and never re-computed).
if (i > 0) {
for (int j = 0; j < res.size() - 1; j++) {
Assertions.assertEquals(lastRes.get(j), res.get(j));
}
}
- lastRes = res;
+ lastRes = new ArrayList<>(res);
}
}
- // Sanity check
checksumCache.clear();
}
+
+ @ParameterizedTest
+ @EnumSource(value = ChecksumType.class, names = {"CRC32", "CRC32C",
"SHA256", "MD5"})
+ void testGrowingMultiBufferMatchesDirectChecksum(ChecksumType checksumType)
+ throws Exception {
+ final int bytesPerChecksum = 16;
+ final byte[] data = new byte[66];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte) i;
+ }
+
+ final Checksum cached = new Checksum(checksumType, bytesPerChecksum, true);
+ final Checksum direct = new Checksum(checksumType, bytesPerChecksum);
+ final int[] lengths = {1, 7, 15, 16, 17, 22, 31, 32, 33, 47, 48, 49, 65,
66};
+ for (int length : lengths) {
+ final ChecksumData expected = direct.computeChecksum(
+ split(data, length, 7));
+ final ChecksumData actual = cached.computeChecksum(
+ split(data, length, 7), true);
+ Assertions.assertEquals(expected.getChecksums(), actual.getChecksums(),
+ "cached checksums must match direct checksums at length " + length);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = ChecksumType.class, names = {"CRC32", "CRC32C",
"SHA256", "MD5"})
+ void testPartiallyPositionedMultiBufferMatchesDirectChecksum(
+ ChecksumType checksumType) throws Exception {
+ final int bytesPerChecksum = 4;
+ final byte[] data = new byte[11];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte) i;
+ }
+
+ final Checksum direct = new Checksum(checksumType, bytesPerChecksum);
+ final Checksum cached = new Checksum(checksumType, bytesPerChecksum, true);
+ final ChecksumData expected =
direct.computeChecksum(partiallyPositioned(data));
+ final ChecksumData actual =
cached.computeChecksum(partiallyPositioned(data), true);
+
+ Assertions.assertEquals(expected.getChecksums(), actual.getChecksums());
+ }
+
+ private static ChunkBuffer split(byte[] data, int length, int bufferSize) {
+ final List<ByteBuffer> buffers = new ArrayList<>();
+ for (int offset = 0; offset < length; offset += bufferSize) {
+ final int size = Math.min(bufferSize, length - offset);
+ buffers.add(ByteBuffer.wrap(data, offset, size).slice());
+ }
+ return ChunkBuffer.wrap(buffers);
+ }
+
+ private static ChunkBuffer partiallyPositioned(byte[] data) {
+ final List<ByteBuffer> buffers = new ArrayList<>();
+ buffers.add(ByteBuffer.wrap(data, 3, 4));
+ buffers.add(ByteBuffer.wrap(data, 7, 4).slice());
+ return ChunkBuffer.wrap(buffers);
+ }
}
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksumMultiBuffer.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksumMultiBuffer.java
new file mode 100644
index 00000000000..56a66387923
--- /dev/null
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChecksumMultiBuffer.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.common;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+/**
+ * A checksum window that straddles multiple underlying buffers must
+ * produce the same checksum bytes as a single contiguous buffer over the
+ * same data.
+ */
+class TestChecksumMultiBuffer {
+
+ private static final int CHUNK_SIZE = 4 * 1024 * 1024;
+ private static final int BYTES_PER_CHECKSUM = 1024 * 1024;
+
+ @ParameterizedTest
+ @EnumSource(value = ChecksumType.class, names = {"CRC32", "CRC32C",
+ "SHA256", "MD5"})
+ void splitBufferProducesSameChecksumAsSingleBuffer(ChecksumType type)
+ throws Exception {
+ byte[] data = new byte[CHUNK_SIZE];
+ ThreadLocalRandom.current().nextBytes(data);
+
+ Checksum checksum = new Checksum(type, BYTES_PER_CHECKSUM);
+
+ // Single contiguous buffer.
+ ChecksumData single = checksum.computeChecksum(
+ ChunkBuffer.wrap(ByteBuffer.wrap(data.clone())));
+
+ // Split into 16 pieces of 256KB - each 1MB checksum window straddles
+ // exactly 4 underlying buffers.
+ int piece = 256 * 1024;
+ List<ByteBuffer> pieces = new ArrayList<>();
+ for (int off = 0; off < CHUNK_SIZE; off += piece) {
+ pieces.add(ByteBuffer.wrap(data, off, piece).slice());
+ }
+ ChecksumData split = checksum.computeChecksum(ChunkBuffer.wrap(pieces));
+
+ assertEquals(single.getChecksums().size(), split.getChecksums().size(),
+ "checksum count must match");
+ assertEquals(single.getChecksums(), split.getChecksums(),
+ "single-buffer and split-buffer must produce identical checksums "
+ + "for " + type);
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = ChecksumType.class, names = {"CRC32", "CRC32C",
+ "SHA256", "MD5"})
+ void unalignedTrailingWindowIsHandled(ChecksumType type) throws Exception {
+ // Chunk size deliberately not a multiple of bytesPerChecksum, plus
+ // unaligned buffer splits, to exercise the trailing-partial-window
+ // path in computeChecksumDirect.
+ int total = BYTES_PER_CHECKSUM * 3 + 12345;
+ byte[] data = new byte[total];
+ ThreadLocalRandom.current().nextBytes(data);
+
+ Checksum checksum = new Checksum(type, BYTES_PER_CHECKSUM);
+
+ ChecksumData single = checksum.computeChecksum(
+ ChunkBuffer.wrap(ByteBuffer.wrap(data.clone())));
+
+ int piece = 333 * 1024; // intentionally awkward
+ List<ByteBuffer> pieces = new ArrayList<>();
+ for (int off = 0; off < total; off += piece) {
+ int len = Math.min(piece, total - off);
+ pieces.add(ByteBuffer.wrap(data, off, len).slice());
+ }
+ ChecksumData split = checksum.computeChecksum(ChunkBuffer.wrap(pieces));
+
+ assertEquals(single.getChecksums(), split.getChecksums(),
+ "unaligned split must match single-buffer for " + type);
+ }
+}
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/utils/TestBufferUtils.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/utils/TestBufferUtils.java
new file mode 100644
index 00000000000..3d5f091bc11
--- /dev/null
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/utils/TestBufferUtils.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.common.utils;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.nio.ByteBuffer;
+import org.junit.jupiter.api.Test;
+
+class TestBufferUtils {
+
+ @Test
+ void slicePreservesSourceAndUsesRequestedRange() {
+ ByteBuffer source = ByteBuffer.allocate(16);
+ source.position(2);
+ source.limit(12);
+
+ ByteBuffer slice = BufferUtils.slice(source, 4, 5);
+
+ assertEquals(2, source.position());
+ assertEquals(12, source.limit());
+ assertEquals(4, slice.position());
+ assertEquals(9, slice.limit());
+ }
+
+ @Test
+ void sliceRejectsNegativePosition() {
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> BufferUtils.slice(ByteBuffer.allocate(8), -1, 1));
+
+ assertEquals("position (-1) must not be negative", exception.getMessage());
+ }
+
+ @Test
+ void sliceRejectsNegativeLength() {
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> BufferUtils.slice(ByteBuffer.allocate(8), 0, -1));
+
+ assertEquals("length (-1) must not be negative", exception.getMessage());
+ }
+
+ @Test
+ void sliceRejectsPositionBeyondLimit() {
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> BufferUtils.slice(ByteBuffer.allocate(8), 9, 0));
+
+ assertEquals("position (9) exceeds source limit (8)",
+ exception.getMessage());
+ }
+
+ @Test
+ void sliceRejectsRangeBeyondLimitWithoutOverflow() {
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> BufferUtils.slice(ByteBuffer.allocate(8), 1,
+ Integer.MAX_VALUE));
+
+ assertEquals(
+ "position (1) + length (2147483647) exceeds source limit (8)",
+ exception.getMessage());
+ }
+}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
index 1f8af28add1..a32a56a0267 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
@@ -1067,7 +1067,9 @@ private void
validateChunkChecksumData(ChunkBufferToByteString data, ChunkInfo i
final ChunkBuffer b = (ChunkBuffer)data;
Checksum.verifyChecksum(b.duplicate(b.position(), b.limit()),
info.getChecksumData(), 0);
} else {
-
Checksum.verifyChecksum(data.toByteString(byteBufferToByteString).asReadOnlyByteBuffer(),
+ // Skip concatenating into one ByteString - that would materialize
+ // a chunk-sized copy on the hot write path.
+
Checksum.verifyChecksum(data.toByteStringList(byteBufferToByteString),
info.getChecksumData(), 0);
}
} catch (OzoneChecksumException ex) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
