Skip to content

Make agility context account for clear() size in quota #4574

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

Open
ohadzeliger wants to merge 1 commit into
base: main
Choose a base branch
from
+154 −14
Open

Make agility context account for clear() size in quota #4574

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions ...yer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/AgileContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,8 @@
*/

package com.apple.foundationdb.record.lucene.directory;

Check notice on line 22 in fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/AgileContext.java

View workflow job for this annotation

GitHub Actions / coverage 

File coverage: 98.0% (149/152 lines) | Changed lines: 100.0% (6/6 lines)
import com.apple.foundationdb.Range;
import com.apple.foundationdb.annotation.API;
import com.apple.foundationdb.record.RecordCoreStorageException;
import com.apple.foundationdb.record.logging.KeyValueLogMessage;
Expand All @@ -40,6 +41,19 @@

/**
* A floating window (agile) context - create sub contexts and commit them as they reach their time/size quota.
*
* <p>The size quota deliberately counts only mutations that the caller issued explicitly, through
* {@link #set}, {@link #clear(byte[])} and {@link #clear(Range)}. It must <b>not</b> be derived from
* {@code FDBTransactionContext.getApproximateTransactionSize()}, even though that value - the summation of
* mutations, read conflict ranges and write conflict ranges - is what FDB's commit size limit actually
* governs. This is done in order to ensure a read-only transaction does not commit due to read-size quota calculation.
* (It may fail with conflicts of ot does).</p>
*
* <p>The accepted consequence is that two kinds of growth are invisible to the size quota: a write issued
* directly on the inner context from inside an {@link #apply} or {@link #accept} lambda, and read conflict
* ranges. A read-heavy transaction can therefore still approach FDB's commit size limit with no size-quota
* protection, leaving the time quota as the only backstop. Callers that read in bulk should bound their own
* work rather than rely on this class to do it. A {@link ReadOnlyNonAgileContext} can also be used.</p>
*/
@API(API.Status.INTERNAL)
public class AgileContext implements AgilityContext {
Expand Down Expand Up @@ -270,6 +284,18 @@
});
}

@Override
public void clear(final byte[] key) {
AgilityContext.super.clear(key);
currentWriteSize += key.length;
}

@Override
public void clear(final Range range) {
AgilityContext.super.clear(range);
currentWriteSize += range.begin.length + range.end.length;
}

private void ensureOpen() {
if (closed) {
throw new RecordCoreStorageException("Agile context is already closed", lastException);
Expand Down
142 changes: 128 additions & 14 deletions ...cene/src/test/java/com/apple/foundationdb/record/lucene/directory/AgilityContextTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2015-2023 Apple Inc. and the FoundationDB project authors
* Copyright 2015-2026 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand All @@ -20,6 +20,7 @@

package com.apple.foundationdb.record.lucene.directory;

import com.apple.foundationdb.Range;
import com.apple.foundationdb.Transaction;
import com.apple.foundationdb.record.RecordCoreException;
import com.apple.foundationdb.record.RecordCoreStorageException;
Expand Down Expand Up @@ -239,9 +240,21 @@ void testAgilityContextConcurrentNonExplicitCommitsExplicitParams(int sizeQuota)
}

private enum Method {
Set,
Apply,
Accept
Set(true, true),
Apply(true, false),
Accept(true, false),
PointClear(false, true),
RangeClear(false, true);

/** Whether the method leaves a readable value behind, which decides how the data is verified. */
private final boolean writesValues;
/** Whether the method's bytes reach {@code currentWriteSize}, and so can drive the size quota. */
private final boolean sizeAccounted;

Method(final boolean writesValues, final boolean sizeAccounted) {
this.writesValues = writesValues;
this.sizeAccounted = sizeAccounted;
}
}

private enum LimitType {
Expand All @@ -262,11 +275,11 @@ private enum LimitType {
static Stream<Arguments> agilityContextLimits() {
return Stream.of(true, false).flatMap(useProp ->
Arrays.stream(LimitType.values()).flatMap(limitType ->
Arrays.stream(Method.values()).filter(method ->
// AgilityContext is only aware of bytes written when set is called
limitType == LimitType.Time || method == Method.Set)
.map(method ->
Arguments.of(useProp, method, limitType))));
Arrays.stream(Method.values())
// A write issued from inside an apply/accept lambda is invisible to the size
// quota so only the methods that account for their own bytes can drive the size limit.
.filter(method -> limitType == LimitType.Time || method.sizeAccounted)
.map(method -> Arguments.of(useProp, method, limitType))));
}

@ParameterizedTest(name = "useProp:{0},{1} by {2}")
Expand All @@ -285,6 +298,20 @@ void testAgilityContextOneLongWrite(boolean useProp, Method method, LimitType li
"And looked down one as far as I could\n" +
"To where it bent in the undergrowth;" ;

if (!method.writesValues) {
// Pre-write data for cases where the Method does not write by itself (so that we can verify that clearing
// actually worked). Use fdb.openContext() rather than openContext(): the latter attaches the shared timer, and these
// sets would then be counted against the quota events this test asserts on.
try (FDBRecordContext context = fdb.openContext()) {
final Subspace subspace = path.toSubspace(context);
for (int i = 0; i < loopCount; i++) {
context.ensureActive().set(subspace.pack(Tuple.from(2023, i)),
Tuple.from(i, RobertFrost, 0).pack());
}
context.commit();
}
}

try (FDBRecordContext context = useProp ? openContext(insertProps) : openContext()) {
final Subspace subspace = path.toSubspace(context);
final AgilityContext agilityContext =
Expand Down Expand Up @@ -316,6 +343,12 @@ void testAgilityContextOneLongWrite(boolean useProp, Method method, LimitType li
innerContext.ensureActive().set(key, val);
});
break;
case PointClear:
agilityContext.clear(key);
break;
case RangeClear:
agilityContext.clear(Range.startsWith(key));
break;
default:
throw new AssertionError("Unexpected enum value " + method);
}
Expand All @@ -332,18 +365,99 @@ void testAgilityContextOneLongWrite(boolean useProp, Method method, LimitType li
for (int i = 0; i < loopCount; i++) {
byte[] key = subspace.pack(Tuple.from(2023, i));
final byte[] bytes = agilityContext.get(key).join();
final Tuple retTuple = Tuple.fromBytes(bytes);
assertEquals(i, retTuple.getLong(0));
assertEquals(RobertFrost, retTuple.getString(1));
if (method.writesValues) {
final Tuple retTuple = Tuple.fromBytes(bytes);
assertEquals(i, retTuple.getLong(0));
assertEquals(RobertFrost, retTuple.getString(1));
} else {
assertNull(bytes);
}
}
}
}

private static final int LARGE_OP_COUNT = 10_000;
/** Padding that makes each key about 2KB, so that 10,000 operations is roughly 20MB of key bytes. */
private static final int PADDED_KEY_LENGTH = 2_000;

private enum BulkOperation { Set, PointClear, RangeClear }

/**
* Assert that the AgilityContext accounts for mutation size correctly.
*/
@ParameterizedTest(name = "operation:{0}")
@EnumSource(BulkOperation.class)
void testAgilityContextBulkOperationStaysUnderCommitLimit(BulkOperation operation) {
final long sizeQuota = 900_000L;
final long timeQuota = 100_000L;
final String padding = "x".repeat(PADDED_KEY_LENGTH);
try (FDBRecordContext context = openContext()) {
final Subspace subspace = path.toSubspace(context);
final AgilityContext agilityContext = AgilityContext.agile(context, timeQuota, sizeQuota);
for (int i = 0; i < LARGE_OP_COUNT; i++) {
final byte[] key = subspace.pack(Tuple.from(operation.name(), i, padding));
switch (operation) {
case Set:
agilityContext.set(key, Tuple.from(i).pack());
break;
case PointClear:
agilityContext.clear(key);
break;
case RangeClear:
agilityContext.clear(Range.startsWith(key));
break;
default:
throw new AssertionError("Unexpected enum value " + operation);
}
}
agilityContext.flush();
context.commit();
}
// Around 20MB of keys against a 900KB quota is on the order of 20 commits, and twice that for a range
// clear, which charges both bounds. The bound is loose on purpose: the point is that the work was split
// at all, not how finely.
assertThat(timer.getCount(LuceneEvents.Counts.LUCENE_AGILE_COMMITS_SIZE_QUOTA), Matchers.greaterThan(10));
}

/**
* A read must never drive the size quota. This class decides when to commit and the caller cannot opt out
* of one, so making a read trigger a commit would expose a caller that only read to {@code NOT_COMMITTED}
* conflicts on its read conflict ranges. Deriving the quota from
* {@code getApproximateTransactionSize()} does exactly that, and was reverted for it; this test is what
* should fail if it is reattempted.
* <p>
* The read count is deliberately small. A large one would accumulate enough conflict range to exceed
* FDB's commit size limit at flush, which is the accepted limitation the class comment records rather
* than something this test contradicts.
* </p>
*/
@Test
void testAgilityContextReadsDoNotTriggerSizeQuota() {
final long sizeQuota = 1L;
final long timeQuota = 100_000L;
final String padding = "x".repeat(PADDED_KEY_LENGTH);
try (FDBRecordContext context = openContext()) {
final Subspace subspace = path.toSubspace(context);
final AgilityContext agilityContext = AgilityContext.agile(context, timeQuota, sizeQuota);
for (int i = 0; i < loopCount; i++) {
agilityContext.get(subspace.pack(Tuple.from("read", i, padding))).join();
}
agilityContext.flush();
context.commit();
}
// A size quota of 1 would commit on any observed byte, so zero here means reads were not observed at
// all. The large time quota rules out the other trigger.
assertThat(timer.getCount(LuceneEvents.Counts.LUCENE_AGILE_COMMITS_SIZE_QUOTA), Matchers.equalTo(0));
assertThat(timer.getCount(LuceneEvents.Counts.LUCENE_AGILE_COMMITS_TIME_QUOTA), Matchers.equalTo(0));
}

static Stream<Arguments> agilityContextLimitsNotSet() {
return Stream.of(true, false).flatMap(useProp ->
Arrays.stream(LimitType.values()).flatMap(limitType ->
Arrays.stream(Method.values())
.filter(method -> method != Method.Set) // There is no good way for us to inject failure
// Failure can only be injected through a lambda, so Set and the clear methods
// are excluded.
.filter(method -> method == Method.Apply || method == Method.Accept)
.map(method -> Arguments.of(useProp, method, limitType))));
}

Expand Down Expand Up @@ -459,7 +573,7 @@ void napTime(int napTimeMilliseconds) {
@ParameterizedTest
@EnumSource
void testAgilityContextAtomicAttribute(AgilityContextType contextType) {
// assert that commits doesn't happen in he middle of an accept or apply call
// assert that commits doesn't happen in the middle of an accept or apply call
for (int sizeQuota : new int[] {1, 21, 100, 10000}) {
final RecordLayerPropertyStorage.Builder insertProps = RecordLayerPropertyStorage.newBuilder()
.addProp(LuceneRecordContextProperties.LUCENE_AGILE_COMMIT_SIZE_QUOTA, sizeQuota);
Expand Down
Loading