Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
32 changes: 16 additions & 16 deletions core/src/main/java/io/questdb/client/Sender.java
Original file line number Diff line number Diff line change
Expand Up @@ -1123,7 +1123,7 @@ public int getConnectTimeout() {
// implemented; FLUSH and APPEND are deferred follow-ups (cursor needs
// to learn fsync first).
private SfDurability sfDurability = SfDurability.MEMORY;
private long sfMaxBytes = PARAMETER_NOT_SET_EXPLICITLY;
private long sfMaxSegmentBytes = PARAMETER_NOT_SET_EXPLICITLY;
private long sfMaxTotalBytes = PARAMETER_NOT_SET_EXPLICITLY;
private boolean shouldDestroyPrivKey;
private boolean tlsEnabled;
Expand Down Expand Up @@ -1446,17 +1446,17 @@ public Sender build() {
// (same lock-free architecture, no disk involvement). The
// sf_durability != memory rejection lives in validateParameters
// so it is reached by build() and by no-connect validation alike.
long actualSfMaxBytes = sfMaxBytes == PARAMETER_NOT_SET_EXPLICITLY
long actualSfMaxSegmentBytes = sfMaxSegmentBytes == PARAMETER_NOT_SET_EXPLICITLY
? DEFAULT_SEGMENT_BYTES
: sfMaxBytes;
: sfMaxSegmentBytes;
// Default cap depends on backing: RAM (memory mode) is tight
// by default; disk (SF mode) is cheap so the default is
// generous enough that normal traffic never hits it.
long defaultMaxTotal = sfDir == null
? DEFAULT_MAX_BYTES_MEMORY
: DEFAULT_MAX_BYTES_SF;
long actualSfMaxTotalBytes = sfMaxTotalBytes == PARAMETER_NOT_SET_EXPLICITLY
? Math.max(defaultMaxTotal, actualSfMaxBytes * 2)
? Math.max(defaultMaxTotal, actualSfMaxSegmentBytes * 2)
: sfMaxTotalBytes;
long actualCloseFlushTimeoutMillis = closeFlushTimeoutMillis == CLOSE_FLUSH_TIMEOUT_NOT_SET
? DEFAULT_CLOSE_FLUSH_TIMEOUT_MILLIS
Expand Down Expand Up @@ -1541,7 +1541,7 @@ public Sender build() {
? CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS
: sfAppendDeadlineMillis * 1_000_000L;
CursorSendEngine cursorEngine = new CursorSendEngine(
slotPath, actualSfMaxBytes,
slotPath, actualSfMaxSegmentBytes,
actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY
? errorInboxCapacity
Expand Down Expand Up @@ -1622,7 +1622,7 @@ public Sender build() {
connected.startOrphanDrainers(
orphans,
maxBackgroundDrainers,
actualSfMaxBytes,
actualSfMaxSegmentBytes,
actualSfMaxTotalBytes);
}
}
Expand Down Expand Up @@ -2757,14 +2757,14 @@ public LineSenderBuilder storeAndForwardDurability(SfDurability durability) {
* (4 MiB). Smaller segments mean faster trim of acked data; larger
* segments mean fewer rotations.
*/
public LineSenderBuilder storeAndForwardMaxBytes(long maxBytes) {
public LineSenderBuilder storeAndForwardMaxSegmentBytes(long maxBytes) {
if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
throw new LineSenderException("store_and_forward is only supported for WebSocket transport");
}
if (maxBytes <= 0) {
throw new LineSenderException("sf_max_bytes must be positive: ").put(maxBytes);
throw new LineSenderException("sf_max_segment_bytes must be positive: ").put(maxBytes);
}
this.sfMaxBytes = maxBytes;
this.sfMaxSegmentBytes = maxBytes;
return this;
}

Expand Down Expand Up @@ -3352,12 +3352,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
}
pos = getValue(configurationString, pos, sink, "sender_id");
senderId(sink.toString());
} else if (Chars.equals("sf_max_bytes", sink)) {
} else if (Chars.equals("sf_max_segment_bytes", sink)) {
if (protocol != PROTOCOL_WEBSOCKET) {
throw new LineSenderException("sf_max_bytes is only supported for WebSocket transport");
throw new LineSenderException("sf_max_segment_bytes is only supported for WebSocket transport");
}
pos = getValue(configurationString, pos, sink, "sf_max_bytes");
storeAndForwardMaxBytes(parseSizeValue(sink, "sf_max_bytes"));
pos = getValue(configurationString, pos, sink, "sf_max_segment_bytes");
storeAndForwardMaxSegmentBytes(parseSizeValue(sink, "sf_max_segment_bytes"));
} else if (Chars.equals("sf_max_total_bytes", sink)) {
if (protocol != PROTOCOL_WEBSOCKET) {
throw new LineSenderException("sf_max_total_bytes is only supported for WebSocket transport");
Expand Down Expand Up @@ -3689,8 +3689,8 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString)
if (view.has("sf_append_deadline_millis")) {
sfAppendDeadlineMillis(wsLong(view, v, "sf_append_deadline_millis"));
}
if (view.has("sf_max_bytes")) {
storeAndForwardMaxBytes(wsSize(view, v, "sf_max_bytes"));
if (view.has("sf_max_segment_bytes")) {
storeAndForwardMaxSegmentBytes(wsSize(view, v, "sf_max_segment_bytes"));
}
if (view.has("sf_max_total_bytes")) {
storeAndForwardMaxTotalBytes(wsSize(view, v, "sf_max_total_bytes"));
Expand Down Expand Up @@ -3848,7 +3848,7 @@ public java.util.Map<String, Object> wsConfigSnapshotForTest() {
m.put("request_durable_ack", requestDurableAck);
m.put("sender_id", senderId);
m.put("sf_dir", sfDir);
m.put("sf_max_bytes", sfMaxBytes);
m.put("sf_max_segment_bytes", sfMaxSegmentBytes);
m.put("sf_max_total_bytes", sfMaxTotalBytes);
m.put("sf_durability", sfDurability == null ? null : sfDurability.name());
m.put("sf_append_deadline_millis", sfAppendDeadlineMillis);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
// drained from the head every time a STATUS_DURABLE_ACK frame advances
// any watermark; an entry pops when every (name, seqTxn) it carries is
// covered by durableTableWatermarks. Bounded in practice by the SF on-disk
// cap: once the producer hits sf_max_bytes it blocks, which caps how far
// cap: once the producer hits sf_max_segment_bytes it blocks, which caps how far
// the durable watermark can lag behind the OK watermark.
private final ArrayDeque<PendingDurableEntry> pendingDurable = new ArrayDeque<>();
private final ArrayDeque<PendingDurableEntry> pendingDurablePool = new ArrayDeque<>();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,7 @@ public static MmapSegment create(FilesFacade ff, long pathPtr, String displayPat
// the JVM).
if (!ff.allocate(fd, sizeBytes)) {
ff.close(fd);
// Unlink the partially-created file so a sf_max_bytes-sized
// Unlink the partially-created file so a sf_max_segment_bytes-sized
// empty file does not survive the failure. Under sustained
// disk-full pressure with the manager polling, hundreds would
// otherwise accumulate.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -255,7 +255,7 @@ public static SegmentRing openExisting(String sfDir, long maxBytesPerSegment) {
return null;
}
// Sort by baseSeq ascending. Worst-case segment count is
// sf_max_total_bytes / sf_max_bytes -- at the documented ceiling
// sf_max_total_bytes / sf_max_segment_bytes -- at the documented ceiling
// (1 TiB / 64 MiB) that is ~16K entries, where an O(N²) sort spends
// multiple seconds in compares + shifts before the I/O thread can
// start. In-place quicksort with median-of-three pivot keeps the
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ public final class ConfigSchema {
str("sf_append_deadline_millis", Side.INGRESS);
str("sf_dir", Side.INGRESS);
str("sf_durability", Side.INGRESS);
str("sf_max_bytes", Side.INGRESS);
str("sf_max_segment_bytes", Side.INGRESS);
str("sf_max_total_bytes", Side.INGRESS);
str("transaction", Side.INGRESS);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -720,7 +720,7 @@ public void testIngressOnlyKeysSilentlyAcceptedOnEgress() {
"sf_append_deadline_millis=30000",
"sf_dir=/var/lib/qdb-sf",
"sf_durability=memory",
"sf_max_bytes=4m",
"sf_max_segment_bytes=4m",
"sf_max_total_bytes=10g",
"transaction=on",
};
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ public void testRestartReplaysSealedSegmentsAgainstFreshServer() throws Exceptio
String pad = repeat("x", 64);
String cfg1 = "ws::addr=localhost:" + port1
+ ";sf_dir=" + sfDir
+ ";sf_max_bytes=4096"
+ ";sf_max_segment_bytes=4096"
+ ";close_flush_timeout_millis=0;";
try (Sender s1 = Sender.fromConfig(cfg1)) {
for (int i = 0; i < 50; i++) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ public void testSfDirOnTcpRejected() throws Exception {
}

@Test
public void testSfMaxBytesParsing() throws Exception {
public void testSfMaxSegmentBytesParsing() throws Exception {
TestUtils.assertMemoryLeak(() -> {
AckHandler handler = new AckHandler();
try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
Expand All @@ -101,7 +101,7 @@ public void testSfMaxBytesParsing() throws Exception {

int port = server.getPort();
String config = "ws::addr=localhost:" + port
+ ";sf_dir=" + sfDir + ";sf_max_bytes=131072;";
+ ";sf_dir=" + sfDir + ";sf_max_segment_bytes=131072;";
try (Sender sender = Sender.fromConfig(config)) {
// Write enough data that segments rotate at ~128 KiB boundary.
for (int i = 0; i < 50; i++) {
Expand Down Expand Up @@ -139,7 +139,7 @@ public void testNoSfDirMeansNoSf() throws Exception {
}

/**
* Regression test for the connect-string {@code sf_max_bytes} /
* Regression test for the connect-string {@code sf_max_segment_bytes} /
* {@code sf_max_total_bytes} parser accepting values larger than
* {@code Integer.MAX_VALUE}. The pre-cursor parser used parseInt which
* artificially capped the SF size from the connect string at ~2 GiB.
Expand Down Expand Up @@ -222,7 +222,7 @@ public void testSfDurabilityOnTcpRejected() throws Exception {
}

@Test
public void testSfMaxBytesAcceptsSizeSuffixes() throws Exception {
public void testSfMaxSegmentBytesAcceptsSizeSuffixes() throws Exception {
TestUtils.assertMemoryLeak(() -> {
AckHandler handler = new AckHandler();
try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
Expand All @@ -233,7 +233,7 @@ public void testSfMaxBytesAcceptsSizeSuffixes() throws Exception {
// 64m / 4g should parse identically to their byte-count equivalents.
String config = "ws::addr=localhost:" + port
+ ";sf_dir=" + sfDir
+ ";sf_max_bytes=64m"
+ ";sf_max_segment_bytes=64m"
+ ";sf_max_total_bytes=4g;";
try (Sender sender = Sender.fromConfig(config)) {
sender.table("foo").longColumn("v", 1L).atNow();
Expand Down Expand Up @@ -346,14 +346,14 @@ public void testSenderIdInvalidCharRejected() throws Exception {
}

@Test
public void testSfMaxBytesInvalidSizeSuffixRejected() throws Exception {
public void testSfMaxSegmentBytesInvalidSizeSuffixRejected() throws Exception {
TestUtils.assertMemoryLeak(() -> {
String config = "ws::addr=localhost:1;sf_dir=" + sfDir + ";sf_max_bytes=64x;";
String config = "ws::addr=localhost:1;sf_dir=" + sfDir + ";sf_max_segment_bytes=64x;";
try (Sender ignored = Sender.fromConfig(config)) {
Assert.fail("expected rejection of unknown unit suffix");
} catch (LineSenderException expected) {
Assert.assertTrue(expected.getMessage(),
expected.getMessage().contains("invalid sf_max_bytes"));
expected.getMessage().contains("invalid sf_max_segment_bytes"));
}
});
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -530,7 +530,7 @@ public void testNextSealedAfterStillReturnsCorrectlyWhenCursorWasTrimmed() throw

/**
* Open-time sort regression: at the documented {@code sf_max_total_bytes
* / sf_max_bytes} ceiling (~16K segments) an O(N²) sort over the
* / sf_max_segment_bytes} ceiling (~16K segments) an O(N²) sort over the
* recovered segments burns multi-second wall time before the I/O thread
* can start. The previous selection-sort implementation regressed an
* earlier perf fix on the legacy {@code SegmentLog} path; this test
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ public void testEveryIngressKeyIsHonored() {
assertHonored("request_durable_ack=on", "request_durable_ack", true);
assertHonored("sender_id=probe-1", "sender_id", "probe-1");
assertHonored("sf_dir=/var/probe", "sf_dir", "/var/probe");
assertHonored("sf_max_bytes=4096", "sf_max_bytes", 4096L);
assertHonored("sf_max_segment_bytes=4096", "sf_max_segment_bytes", 4096L);
assertHonored("sf_max_total_bytes=8192", "sf_max_total_bytes", 8192L);
assertHonored("sf_durability=flush", "sf_durability", "FLUSH");
assertHonored("sf_append_deadline_millis=1500", "sf_append_deadline_millis", 1500L);
Expand Down
Loading