FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

chore: update ChunkSegmenter to optionally allow a limit on the number of bytes it should consume by BenWhitehead · Pull Request #3279 · googleapis/java-storage · GitHub

/ java-storage Public

chore: update ChunkSegmenter to optionally allow a limit on the number of bytes it should consume #3279

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
BenWhitehead merged 2 commits into from
Sep 9, 2025
Conversation 5 Commits 2 Checks 21 Files changed
Merged

chore: update ChunkSegmenter to optionally allow a limit on the number of bytes it should consume #3279

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8134b4f
chore: update ChunkSegmenter to optionally allow a limit on the numbe…
BenWhitehead Sep 4, 2025
e07f2a4
chore: fix blockSize reset logic
BenWhitehead Sep 5, 2025
File filter

Filter by extension

Filter by extension .java  (4) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
3 changes: 2 additions & 1 deletion ...e/src/main/java/com/google/cloud/storage/BidiAppendableUnbufferedWritableByteChannel.java
Show comments
View file
Open in desktop
This file contains hidden or 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
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,8 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) th
RewindableContent rewindableContent = RewindableContent.of(srcs, srcsOffset, srcsLength);
long totalBufferRemaining = rewindableContent.getLength();

ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength, true);
ChunkSegment[] data =
chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength, true, availableCapacity);
if (data.length == 0) {
return 0;
}
Expand Down
78 changes: 60 additions & 18 deletions google-cloud-storage/src/main/java/com/google/cloud/storage/ChunkSegmenter.java
Show comments
View file
Open in desktop
This file contains hidden or 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
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,10 @@

import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import com.google.common.math.IntMath;
import com.google.common.primitives.Ints;
import com.google.protobuf.ByteString;
import java.math.RoundingMode;
import java.nio.ByteBuffer;
Expand Down Expand Up @@ -97,66 +99,96 @@ ChunkSegment[] segmentBuffers(
// turn this into a single branch, rather than multiple that would need to be checked each
// element of the iteration
if (allowUnalignedBlocks) {
return segmentWithUnaligned(bbs, offset, length);
return segmentWithUnaligned(bbs, offset, length, Long.MAX_VALUE);
} else {
return segmentWithoutUnaligned(bbs, offset, length);
return segmentWithoutUnaligned(bbs, offset, length, Long.MAX_VALUE);
}
}

private ChunkSegment[] segmentWithUnaligned(ByteBuffer[] bbs, int offset, int length) {
ChunkSegment[] segmentBuffers(
ByteBuffer[] bbs,
int offset,
int length,
boolean allowUnalignedBlocks,
long maxBytesToConsume) {
// turn this into a single branch, rather than multiple that would need to be checked each
// element of the iteration
if (allowUnalignedBlocks) {
return segmentWithUnaligned(bbs, offset, length, maxBytesToConsume);
} else {
long misaligned = maxBytesToConsume % blockSize;
long alignedMaxBytesToConsume = maxBytesToConsume - misaligned;
return segmentWithoutUnaligned(bbs, offset, length, alignedMaxBytesToConsume);
}
}

private ChunkSegment[] segmentWithUnaligned(
ByteBuffer[] bbs, int offset, int length, long maxBytesToConsume) {
Deque<ChunkSegment> data = new ArrayDeque<>();

long consumed = 0;
for (int i = offset; i < length; i++) {
ByteBuffer buffer = bbs[i];
int remaining;
while ((remaining = buffer.remaining()) > 0) {
consumeBytes(data, remaining, buffer);
while ((remaining = buffer.remaining()) > 0 && consumed < maxBytesToConsume) {
long remainingConsumable = maxBytesToConsume - consumed;
int toConsume = remaining;
if (remainingConsumable < remaining) {
toConsume = Math.toIntExact(remainingConsumable);
}
long consumeBytes = consumeBytes(data, toConsume, buffer);
consumed += consumeBytes;
}
}

return data.toArray(new ChunkSegment[0]);
}

private ChunkSegment[] segmentWithoutUnaligned(ByteBuffer[] bbs, int offset, int length) {
private ChunkSegment[] segmentWithoutUnaligned(
ByteBuffer[] bbs, int offset, int length, long maxBytesToConsume) {
Deque<ChunkSegment> data = new ArrayDeque<>();

final long totalRemaining = Buffers.totalRemaining(bbs, offset, length);
long buffersTotalRemaining = Buffers.totalRemaining(bbs, offset, length);
final long totalRemaining = Math.min(maxBytesToConsume, buffersTotalRemaining);
long consumedSoFar = 0;

int currentBlockPending = blockSize;

outerloop:
for (int i = offset; i < length; i++) {
ByteBuffer buffer = bbs[i];
int remaining;
while ((remaining = buffer.remaining()) > 0) {
long overallRemaining = totalRemaining - consumedSoFar;
if (overallRemaining < blockSize && currentBlockPending == blockSize) {
break;
break outerloop;

BrandonY Sep 4, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

ooo, label break, those're rare!
}

int numBytesConsumable;
if (remaining >= blockSize) {
if (remaining >= blockSize && currentBlockPending == blockSize) {
int blockCount = IntMath.divide(remaining, blockSize, RoundingMode.DOWN);
numBytesConsumable = blockCount * blockSize;
} else if (currentBlockPending < blockSize) {
numBytesConsumable = currentBlockPending;
currentBlockPending = blockSize;
} else {
numBytesConsumable = remaining;
currentBlockPending = currentBlockPending - remaining;
numBytesConsumable = Math.min(remaining, currentBlockPending);
}
if (numBytesConsumable <= 0) {
continue;
break outerloop;

BrandonY Sep 4, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

K...now I'm starting to worry a bit about cyclomatic complexity

BenWhitehead Sep 5, 2025

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

Prior to the appendable upload stuff we generally had a pretty firm bulkhead on the number of buffers that would be passed into these methods outside of tests. With the appendable addition we place less emphasis on early buffering in favor of passing things through wherever possible, so if multiple buffers are passed in here, logically the conditions would prevent consuming any bytes once a break from the while takes place, but by breaking the for as well we avoid the cycles performing work that isn't productive.

And, refactoring everything to nested method calls to allow early returns instead of break to label didn't seem worth it to me.

danielduhh Sep 9, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

What's the follow up here?

BenWhitehead Sep 9, 2025

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

Nothing from my perspective. There isn't a functional difference here between the label break and a separate method with early returns.
}

consumedSoFar += consumeBytes(data, numBytesConsumable, buffer);
int consumed = consumeBytes(data, numBytesConsumable, buffer);
int currentBlockPendingLessConsumed = currentBlockPending - consumed;
currentBlockPending = currentBlockPendingLessConsumed % blockSize;
if (currentBlockPending == 0) {
currentBlockPending = blockSize;
}
consumedSoFar += consumed;
}
}

return data.toArray(new ChunkSegment[0]);
}

private long consumeBytes(Deque<ChunkSegment> data, int numBytesConsumable, ByteBuffer buffer) {
private int consumeBytes(Deque<ChunkSegment> data, int numBytesConsumable, ByteBuffer buffer) {
// either no chunk or most recent chunk is full, start a new one
ChunkSegment peekLast = data.peekLast();
if (peekLast == null || peekLast.b.size() == maxSegmentSize) {
Expand All @@ -167,7 +199,8 @@ private long consumeBytes(Deque<ChunkSegment> data, int numBytesConsumable, Byte
} else {
ChunkSegment chunkSoFar = data.pollLast();
//noinspection ConstantConditions -- covered by peekLast check above
int limit = Math.min(numBytesConsumable, maxSegmentSize - chunkSoFar.b.size());
int limit =
Ints.min(buffer.remaining(), numBytesConsumable, maxSegmentSize - chunkSoFar.b.size());
ChunkSegment datum = newSegment(buffer, limit);
ChunkSegment plus = chunkSoFar.concat(datum);
data.addLast(plus);
Expand Down Expand Up @@ -218,5 +251,14 @@ public Crc32cLengthKnown getCrc32c() {
public boolean isOnlyFullBlocks() {
return onlyFullBlocks;
}

@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("crc32c", crc32c)
.add("onlyFullBlocks", onlyFullBlocks)
.add("b", b)
.toString();
}
}
}
73 changes: 73 additions & 0 deletions google-cloud-storage/src/test/java/com/google/cloud/storage/ChunkSegmenterTest.java
Show comments
View file
Open in desktop
This file contains hidden or 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
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
import com.google.cloud.storage.it.ChecksummedTestContent;
import com.google.common.collect.ImmutableList;
import com.google.common.hash.HashCode;
import com.google.common.hash.Hashing;
Expand Down Expand Up @@ -172,6 +173,78 @@ void allowUnalignedBlocks_false_3() throws Exception {
() -> assertThat(actual).isEqualTo(expected));
}

@Example
void maxBytesToConsume_unaligned() throws Exception {

ChecksummedTestContent ctc = ChecksummedTestContent.gen(64);

ChunkSegmenter segmenter = new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.noCopy(), 6, 3);

List<ChecksummedTestContent> chunks = ctc.chunkup(4);
ByteBuffer[] buffers =
chunks.stream().map(ChecksummedTestContent::asByteBuffer).toArray(ByteBuffer[]::new);
buffers[1].position(1);

ChecksummedTestContent slice = ctc.slice(5, 37);
List<ByteString> expected =
slice.chunkup(6).stream()
.map(ChecksummedTestContent::asByteBuffer)
.map(ByteStringStrategy.noCopy())
.collect(Collectors.toList());

ChunkSegment[] segments = segmenter.segmentBuffers(buffers, 1, buffers.length - 2, true, 37);
List<ByteString> actual =
Arrays.stream(segments).map(ChunkSegment::getB).collect(Collectors.toList());
assertThat(actual).isEqualTo(expected);
}

@Example
void maxBytesToConsume_aligned() throws Exception {

ChecksummedTestContent ctc = ChecksummedTestContent.gen(64);

ChunkSegmenter segmenter = new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.noCopy(), 6, 3);

List<ChecksummedTestContent> chunks = ctc.chunkup(4);
ByteBuffer[] buffers =
chunks.stream().map(ChecksummedTestContent::asByteBuffer).toArray(ByteBuffer[]::new);
buffers[1].position(1);

ChecksummedTestContent slice = ctc.slice(5, 36);
List<ByteString> expected =
slice.chunkup(6).stream()
.map(ChecksummedTestContent::asByteBuffer)
.map(ByteStringStrategy.noCopy())
.collect(Collectors.toList());

ChunkSegment[] segments = segmenter.segmentBuffers(buffers, 1, buffers.length - 2, false, 37);
List<ByteString> actual =
Arrays.stream(segments).map(ChunkSegment::getB).collect(Collectors.toList());
assertThat(actual).isEqualTo(expected);
}

@Example
void alignedConsumeForLargeBuffersOnlyConsumesAligned() throws Exception {

ChecksummedTestContent ctc = ChecksummedTestContent.gen(2048 + 13);

ChunkSegmenter segmenter =
new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.noCopy(), 2048, 256);

ChecksummedTestContent slice = ctc.slice(0, 2048);
List<ByteString> expected =
slice.chunkup(2048).stream()
.map(ChecksummedTestContent::asByteBuffer)
.map(ByteStringStrategy.noCopy())
.collect(Collectors.toList());

ByteBuffer buf = ctc.asByteBuffer();
ChunkSegment[] segments = segmenter.segmentBuffers(new ByteBuffer[] {buf}, 0, 1, false);
List<ByteString> actual =
Arrays.stream(segments).map(ChunkSegment::getB).collect(Collectors.toList());
assertThat(actual).isEqualTo(expected);
}

@Provide("TestData")
static Arbitrary<TestData> arbitraryTestData() {
return Arbitraries.lazyOf(
Expand Down
3 changes: 2 additions & 1 deletion google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java
Show comments
View file
Open in desktop
This file contains hidden or 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
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -263,7 +263,8 @@ public void takeoverJustToFinalizeWorks() throws Exception {
}

private void checkTestbenchIssue733() {
if (p.uploadConfig.getCloseAction() == CloseAction.FINALIZE_WHEN_CLOSING) {
if (backend == Backend.TEST_BENCH
&& p.uploadConfig.getCloseAction() == CloseAction.FINALIZE_WHEN_CLOSING) {
int estimatedMessageCount = 0;
FlushPolicy flushPolicy = p.uploadConfig.getFlushPolicy();
if (flushPolicy instanceof MinFlushSizeFlushPolicy) {
Expand Down
Loading
Back | FazBrowse Home | New Git URL