Skip to content

Commit bd7c3cb

Browse files
bluestreak01bluestreak01
committed
perf(qwp): binary-search sealed segments in SegmentRing.findSegmentContaining0
findSegmentContaining0 walked sealedSegments[sealedHead..size) front to back to find the segment covering an fsn: O(live-sealed) per call, and the live window is unbounded when the producer outpaces the drain. The callers are cursor (re)positioning on reconnect (findSegmentContaining / pinSegmentContaining via CursorSendEngine), so the cost was per reconnect, not per row -- but it scaled with exactly the backlog a reconnect storm tends to coincide with. The sealed list is strictly ascending and contiguous in baseSeq on every mutation path: rotation rebases the promoted spare to previous.baseSeq() + previous.frameCount() before sealing, recovery sorts by unsigned baseSeq and rejects any gap via validateContiguous, and all removals are prefix-only. The only candidate that can contain fsn is therefore the rightmost sealed segment with baseSeq <= fsn. Binary-search for it (unsigned compare, matching sortByBaseSeq's key order) and re-check that single candidate with the original containment predicate; the active-segment fallback is unchanged. O(log live-sealed), zero allocation, same nulls/misses on every input including negative fsn, empty window, and post-trim sealedHead > 0. Adds two SegmentRingTest guards pinning the full boundary matrix (segment bases, interiors, last fsns, below/above range, single-frame segments, post-trim windows); both were run green against the pre-fix linear implementation first, so the rewrite is provably semantics-preserving.
1 parent 8fd842f commit bd7c3cb

2 files changed

Lines changed: 148 additions & 5 deletions

File tree

‎core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentRing.java‎

Lines changed: 24 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1054,9 +1054,10 @@ public synchronized ObjList<MmapSegment> drainTrimmable() {
10541054
* to position the I/O thread's cursor at the first unacked frame for
10551055
* replay.
10561056
* <p>
1057-
* Walks sealed first (oldest → newest) then the active. The sealed list
1058-
* is small enough -- and reconnects are rare enough -- that the linear
1059-
* scan cost doesn't matter.
1057+
* Binary-searches the sealed list, then falls back to the active
1058+
* segment: O(log live-sealed) per call, so repositioning stays cheap
1059+
* even when a producer-outpaces-drain backlog has grown the sealed
1060+
* list without bound.
10601061
*/
10611062
public synchronized MmapSegment findSegmentContaining(long fsn) {
10621063
return findSegmentContaining0(fsn);
@@ -1469,8 +1470,26 @@ public synchronized long totalSegmentBytes() {
14691470
}
14701471

14711472
private MmapSegment findSegmentContaining0(long fsn) {
1472-
for (int i = sealedHead, n = sealedSegments.size(); i < n; i++) {
1473-
MmapSegment segment = sealedSegments.get(i);
1473+
// The sealed list is strictly ascending and contiguous in baseSeq:
1474+
// rotation seals the predecessor exactly where the promoted spare's
1475+
// rebased baseSeq starts, and recovery sorts the chain then rejects
1476+
// any gap via validateContiguous. The only sealed segment that can
1477+
// cover fsn is therefore the rightmost one whose baseSeq is at or
1478+
// below fsn -- binary-search for it (unsigned, matching
1479+
// sortByBaseSeq's key order) instead of walking a list that grows
1480+
// without bound while the producer outpaces the drain.
1481+
int lo = sealedHead;
1482+
int hi = sealedSegments.size() - 1;
1483+
while (lo <= hi) {
1484+
int mid = (lo + hi) >>> 1;
1485+
if (Long.compareUnsigned(sealedSegments.get(mid).baseSeq(), fsn) <= 0) {
1486+
lo = mid + 1;
1487+
} else {
1488+
hi = mid - 1;
1489+
}
1490+
}
1491+
if (hi >= sealedHead) {
1492+
MmapSegment segment = sealedSegments.get(hi);
14741493
long base = segment.baseSeq();
14751494
if (fsn >= base && fsn < base + segment.frameCount()) {
14761495
return segment;

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentRingTest.java‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@
4444
import static org.junit.Assert.assertEquals;
4545
import static org.junit.Assert.assertNotNull;
4646
import static org.junit.Assert.assertNull;
47+
import static org.junit.Assert.assertSame;
4748
import static org.junit.Assert.assertTrue;
4849

4950
public class SegmentRingTest {
@@ -869,6 +870,129 @@ public void testMaxBytesPerSegmentSurvivesOpenExisting() throws Exception {
869870
});
870871
}
871872

873+
/**
874+
* Pins findSegmentContaining across the full boundary matrix so the
875+
* O(log N) lookup rewrite is provably semantics-preserving: empty ring,
876+
* miss below the head segment, exact segment base, mid-segment, last FSN
877+
* in a segment, tail sealed segment, active-segment hits, miss above the
878+
* published range, and misses below a trimmed live head (sealedHead > 0).
879+
* Three frames per segment give every sealed segment a distinct base,
880+
* mid and last FSN.
881+
*/
882+
@Test
883+
public void testFindSegmentContainingBoundaryMatrix() throws Exception {
884+
TestUtils.assertMemoryLeak(() -> {
885+
long segSize = MmapSegment.HEADER_SIZE
886+
+ 3 * (MmapSegment.FRAME_HEADER_SIZE + 8);
887+
long buf = Unsafe.malloc(8, MemoryTag.NATIVE_DEFAULT);
888+
try {
889+
fillPattern(buf, 8, 0);
890+
MmapSegment initial = MmapSegment.createInMemory(0, segSize);
891+
try (SegmentRing ring = new SegmentRing(initial, segSize)) {
892+
// Empty ring: no frame published anywhere, every lookup misses.
893+
assertNull(ring.findSegmentContaining(-1));
894+
assertNull(ring.findSegmentContaining(0));
895+
896+
for (int i = 0; i < 12; i++) {
897+
if (ring.needsHotSpare()) {
898+
ring.installHotSpare(
899+
MmapSegment.createInMemory(ring.nextSeqHint(), segSize));
900+
}
901+
assertEquals(i, ring.appendOrFsn(buf, 8));
902+
}
903+
// Sealed: [0-2], [3-5], [6-8]; active: [9-11].
904+
for (long fsn = 0; fsn < 12; fsn++) {
905+
MmapSegment seg = ring.findSegmentContaining(fsn);
906+
assertNotNull("fsn " + fsn, seg);
907+
assertEquals("fsn " + fsn, fsn / 3 * 3, seg.baseSeq());
908+
}
909+
assertSame(ring.getActive(), ring.findSegmentContaining(9));
910+
assertSame(ring.getActive(), ring.findSegmentContaining(11));
911+
assertNull(ring.findSegmentContaining(-1));
912+
assertNull(ring.findSegmentContaining(12));
913+
assertNull(ring.findSegmentContaining(Long.MIN_VALUE));
914+
assertNull(ring.findSegmentContaining(Long.MAX_VALUE));
915+
916+
// Trim the two fully-ACK'd oldest sealed segments: their
917+
// FSNs must now miss while the surviving window (with a
918+
// non-zero sealedHead) still resolves every live FSN.
919+
ring.acknowledge(5);
920+
MmapSegment trimmable;
921+
int removed = 0;
922+
while ((trimmable = ring.firstTrimmable()) != null) {
923+
assertTrue(ring.removeTrimmable(trimmable));
924+
trimmable.close();
925+
removed++;
926+
}
927+
assertEquals(2, removed);
928+
for (long fsn = 0; fsn < 6; fsn++) {
929+
assertNull("trimmed fsn " + fsn, ring.findSegmentContaining(fsn));
930+
}
931+
for (long fsn = 6; fsn < 12; fsn++) {
932+
MmapSegment seg = ring.findSegmentContaining(fsn);
933+
assertNotNull("fsn " + fsn, seg);
934+
assertEquals("fsn " + fsn, fsn / 3 * 3, seg.baseSeq());
935+
}
936+
}
937+
} finally {
938+
Unsafe.free(buf, 8, MemoryTag.NATIVE_DEFAULT);
939+
}
940+
});
941+
}
942+
943+
/**
944+
* Same lookup pinned over one-frame segments, where every FSN is both a
945+
* segment base and a segment's last FSN: 33 sealed segments (odd live
946+
* count), then a 17-segment trim leaving an even 16-segment window with
947+
* sealedHead well inside the backing list. Complements the three-frame
948+
* matrix by exercising both live-window parities and a larger list.
949+
*/
950+
@Test
951+
public void testFindSegmentContainingSingleFrameSegments() throws Exception {
952+
TestUtils.assertMemoryLeak(() -> {
953+
long segSize = MmapSegment.HEADER_SIZE + MmapSegment.FRAME_HEADER_SIZE + 1;
954+
long buf = Unsafe.malloc(1, MemoryTag.NATIVE_DEFAULT);
955+
try {
956+
MmapSegment initial = MmapSegment.createInMemory(0, segSize);
957+
try (SegmentRing ring = new SegmentRing(initial, segSize)) {
958+
for (int i = 0; i < 34; i++) {
959+
assertEquals(i, ring.appendOrFsn(buf, 1));
960+
ring.installHotSpare(
961+
MmapSegment.createInMemory(ring.nextSeqHint(), segSize));
962+
}
963+
// Sealed: FSNs 0..32, one frame each; active: [33].
964+
for (long fsn = 0; fsn <= 33; fsn++) {
965+
MmapSegment seg = ring.findSegmentContaining(fsn);
966+
assertNotNull("fsn " + fsn, seg);
967+
assertEquals("fsn " + fsn, fsn, seg.baseSeq());
968+
}
969+
assertNull(ring.findSegmentContaining(-1));
970+
assertNull(ring.findSegmentContaining(34));
971+
972+
ring.acknowledge(16);
973+
MmapSegment trimmable;
974+
int removed = 0;
975+
while ((trimmable = ring.firstTrimmable()) != null) {
976+
assertTrue(ring.removeTrimmable(trimmable));
977+
trimmable.close();
978+
removed++;
979+
}
980+
assertEquals(17, removed);
981+
for (long fsn = 0; fsn < 17; fsn++) {
982+
assertNull("trimmed fsn " + fsn, ring.findSegmentContaining(fsn));
983+
}
984+
for (long fsn = 17; fsn <= 33; fsn++) {
985+
MmapSegment seg = ring.findSegmentContaining(fsn);
986+
assertNotNull("fsn " + fsn, seg);
987+
assertEquals("fsn " + fsn, fsn, seg.baseSeq());
988+
}
989+
}
990+
} finally {
991+
Unsafe.free(buf, 1, MemoryTag.NATIVE_DEFAULT);
992+
}
993+
});
994+
}
995+
872996
private static void fillPattern(long addr, int len, int seed) {
873997
for (int i = 0; i < len; i++) {
874998
Unsafe.getUnsafe().putByte(addr + i, (byte) (seed * 31 + i + 17));

0 commit comments

Comments
 (0)