yandrey321 commented on code in PR #10350:
URL: https://github.com/apache/ozone/pull/10350#discussion_r3762273702
##########
hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChecksumCache.java:
##########
@@ -65,58 +66,102 @@ 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
+ /**
+ * 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.limit();
if (currChunkLength == prevChunkLength) {
LOG.debug("ChunkBuffer data limit 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 limit (" +
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).
+ assert i == checksums.size() - 1 || i == checksums.size();
Review Comment:
replace assert with Preconditions.checkState()?
##########
hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChecksumCache.java:
##########
@@ -65,58 +66,102 @@ 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
+ /**
+ * 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.limit();
Review Comment:
currChunkLength is taken from data.limit(), and ciEnd is derived from it
(L102-103), but the walk below only ever processes sum(src.limit() −
src.position()) bytes over asByteBufferList() — i.e. data.remaining(), not
data.limit(). These are equal only when every underlying buffer starts at
position 0. If a partially-positioned ChunkBuffer is ever passed with
useCache=true, ciEnd will disagree with the number of windows actually produced
and the terminal if (i != ciEnd) check will throw IllegalStateException.
the simplest fix would be:
final int currChunkLength = data.remaining();
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
