Integration robustness contracts - #710
Conversation
Introduce an opt-in control surface a host can install on long-running
jvector operations, without touching any existing call site:
- WorkStage / WorkLimiter / ProgressTracker: the admission + reporting
primitives (throttle down, progress up).
- ProgressLimiter: the single combined facet a host implements.
- LeakyBucketLimiter: a default rate-limited WorkLimiter.
Purely additive. Nothing in the library calls these yet; they establish
the seam that host-driven operations (e.g. compaction) can later be
rewritten against.
Add a host-supplied destination abstraction so compaction output can be
written through a caller-owned channel instead of always allocating its
own file:
- SeekableSink / FileChannelSeekableSink: a minimal seekable byte sink
and its file-backed implementation.
- CompactionDestination / FileCompactionDestination: the compaction
output target, resolvable to a SeekableSink.
Purely additive interface types; no existing code is wired to them on
this branch.
Add ParallelExecutor, a minimal parallel-for abstraction that lets an embedding host supply its own execution strategy -- its own pool, or caller-runs on the calling thread -- instead of jvector reaching for ForkJoinPool.commonPool(): - forkJoin(pool): run on a caller-supplied ForkJoinPool - callerRuns(): run inline, no work escaping to a shared pool - forEachInt / forEach: the parallel-for entry points Additive: a standalone interface with no consumers on this branch. It establishes the execution seam that build, quantization, and compaction paths can be rewritten against later.
Add RuntimeMode, a process-level switch (jvector.mode) separating production from diagnostic runs. Unset is production; diagnostic-only work (e.g. verification walks) sits behind this gate and must be opted into explicitly. Additive: a standalone type with no call sites on this branch. It establishes the gate that diagnostic code paths can consult once wired.
Harden the read path for embedding hosts that manage their own mmap
lifecycle, where a stale offset or a close racing an in-flight read
faults the JVM (SIGSEGV) instead of throwing:
- OnDiskGraphIndex: bound record reads. A node id outside the graph
would become a wild offset into the mapped file (garbage, or a
fault); requireValidNode() rejects it up front, and a stale/corrupt
neighbor block now fails with IllegalStateException instead of
consuming garbage ints.
- ReaderSupplier / SimpleMappedReader: document the close() contract --
the raw-release (immediate unmap -> SIGSEGV) vs coordinated
(liveness handshake -> IllegalStateException) families -- so a host
knows a supplier must not be closed until every vended reader is
quiescent.
Non-breaking: internal bounds checks and documentation only, no
signature or call-path changes. The compactor-side memory safety
(drain-on-unwind, truncate-reused-outputs) stays with the compaction
work, where it lives.
|
Before you submit for review:
If you did not complete any of these, then please explain below. |
There was a problem hiding this comment.
This PR adds in some helpful comments and checks (in OnDiskGraphIndex, ReaderSupplier, SimpleMappedReader) while adding some new interfaces. IMO those are different concerns: it might be better to split the comments and checks off into a separate PR while leaving the new interfaces here.
| * hands one out is {@code io.github.jbellis.jvector.graph.disk.CompactionDestination}. | ||
| */ | ||
| @Experimental | ||
| public interface SeekableSink extends AutoCloseable { |
There was a problem hiding this comment.
This interface appears to serve the same purpose as RandomAccessReader/ReaderSupplier + RandomAccessWriter/IndexWriter. Is this intention to completely replace those existing interfaces with this one? If so, it would be helpful to jot down the advantages of this approach over what we already have.
| if (path == null) { | ||
| throw new NullPointerException("path"); | ||
| } |
There was a problem hiding this comment.
| if (path == null) { | |
| throw new NullPointerException("path"); | |
| } | |
| Objects.requireNonNull(path); |
| Target open() throws IOException; | ||
|
|
||
| /** One compaction's output region plus its commit/abort lifecycle. */ | ||
| interface Target extends AutoCloseable { |
There was a problem hiding this comment.
Instead of passing a CompactionDestination on which open needs to be called to obtain a CompactionDestination.Target, couldn't a client simply pass in a CompactionTarget class directly?
| * e.g. one thread per compaction — instead of a jvector-owned all-core pool. It is the build/finalize | ||
| * counterpart to the caller-runs executor injection already available on the compaction merge path. | ||
| */ | ||
| public interface ParallelExecutor { |
There was a problem hiding this comment.
This interface takes care of the forEach operations for generic Streams and IntStreams, but there are plenty of other terminal operations that are not expressed, for example collect and reduce for generic streams, count, min, max, toArray for primitive streams etc. There are whole other stream types like ByteStream which are not covered. Even if we don't use those terminal operations now, this interface locks us out of using any of those operations in the future.
In theory you could add all those operations to the ParallelExecutor, but that just leaves us with a huge amount of boilerplate to maintain.
The solution used elsewhere in JV is to accept a ForkJoinPool directly, run all parallel stream operations within that pool. This approach has certain downsides (for example, if you want to run a parallel task single-threaded, it'll probably run on the ForkJoin worker thread while the current thread is blocked and idle), but is rather flexible in terms of the operations that are supported.
Are the trade-offs involved in using this interface really worth it?
| * non-decreasing within a stage | ||
| * @param total total work for this stage, or {@code -1} if not yet known | ||
| */ | ||
| void onProgress(WorkStage stage, long completed, long total); |
There was a problem hiding this comment.
Perhaps the onProgress method should be defined for each PhaseScope rather than for each ProgressTracker? That way, implementers of this method don't need to worry about managing multiple progress bars, think about handling errors in case a WorkStage is submitted before it's PhaseScope is started etc.
2 participants
Synopsis
This is a set of changes to make the embedding layer between JVector an hosting runtimes more robust, visible, and efficient. The aspects addressed here are those which we haven't been prescriptive enough about for embedding systems.
Most essentially, these contracts and supporting types and wrappers are used by embeding systems to instruct JVector where and how to do certain things, like writing index outputs, or using thread pools.
The aspects covered in this set include execution controls, rate-limiting, output vectoring (virtualization), runtime-safety checks on non-runtime debugging features, and memory safety.
Why
Jvector today assumes it owns its execution, its output files, and its process. When it runs embedded in a host that manages its own thread pools, IO budget, mmap lifecycle, and cancellation — e.g. Cassandra's SAI vector indexing driving compaction — those assumptions leak:
ForkJoinPool.commonPool()/PhysicalCoreExecutor.pool(), outside the host's budget;This PR adds the extension points a host needs to close those gaps. It is purely additive and opt-in — no existing behavior changes, and nothing in jvector consumes these seams yet. It's the foundation an embedding/compaction effort builds on; a follow-up PR illustrates how the compactor's calling conventions convert onto these seams.
What's in it, in detail
Five contracts, one per commit:
util/work/) —ProgressLimiter=ProgressTracker(progress up) +WorkLimiter(throttle down), withWorkStagefor phase scoping andLeakyBucketLimiteras a default rate-limited limiter. A host installs one to observe and rate-limit a long jvector operation, and to checkpoint cancellation at phase boundaries.disk/) —SeekableSink+CompactionDestination(with file-backed defaults) let a host redirect compaction output through a caller-owned channel — e.g. a slot inside a larger container after a reserved header — instead of jvector always allocating its own file.Targethas an explicitopen → commit → closelifecycle (nocommit()⇒ aborted, partial output discarded).ParallelExecutor(graph/) — a minimal parallel-for abstraction so a host supplies its own pool (forkJoin(pool)) or runs inline (callerRuns()), instead of jvector reaching for the common pool.RuntimeMode(util/) — a process-leveljvector.modegate. Unset is production; diagnostic-only work is opt-in behind it.disk/) — bounded record reads inOnDiskGraphIndex(an out-of-range node id fails withIllegalArgumentExceptioninstead of becoming a wild offset into the mapped file), plus documenting theclose()contract onReaderSupplier/SimpleMappedReader(the raw-release-immediate-unmap vs. coordinated-liveness-handshake families) so a host knows not to close a supplier while any vended reader is live.Design principles
--release 11. Everything lands injvector-base(bytecode v55), so it runs on every supported JDK (11–25). NoMETA-INF/versions/*overrides, no incubator Vector API use — no per-runtime divergence.Deliberately not here
No changes to the compactor or its algorithm, no new call sites, no host-specific code. Adoption is the follow-up PR.
Testing
jvector-basecompiles at--release 11. New unit tests coverProgressLimiter,SeekableSink, andRuntimeMode. The only edits to existing files are non-breaking (Javadoc onReaderSupplier/SimpleMappedReader, an internal bounds check inOnDiskGraphIndex).Stats: 17 files, +1346/−3 (the 3 deletions are lines replaced by the
OnDiskGraphIndexbounds check).Commits
util/work: cooperative work-limiting and progress interfacedisk: pluggable output sink for compaction writesgraph: ParallelExecutor abstraction for host-provided executionutil: RuntimeMode gate for opt-in diagnostic workdisk: memory-safety guards for host-managed reads