You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-dev@jackrabbit.apache.org by "smiroslav (via GitHub)" <gi...@apache.org> on 2025/09/16 16:35:44 UTC
[PR] Segment prefetching for CachingSegmentReader [jackrabbit-oak]
smiroslav opened a new pull request, #2513:
URL: https://github.com/apache/jackrabbit-oak/pull/2513
https://issues.apache.org/jira/browse/OAK-11932
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11932: Segment prefetching for CachingSegmentReader [jackrabbit-oak]
Posted by "jsedding (via GitHub)" <gi...@apache.org>.
jsedding commented on PR #2513:
URL: https://github.com/apache/jackrabbit-oak/pull/2513#issuecomment-3386202347
As indicated in #2569, this approach is limited in that a `CachingSegmentArchiveReader` has only got access to a single archive, and thus cannot read segments from other archives.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11932: Segment prefetching for CachingSegmentReader [jackrabbit-oak]
Posted by "jsedding (via GitHub)" <gi...@apache.org>.
jsedding commented on code in PR #2513:
URL: https://github.com/apache/jackrabbit-oak/pull/2513#discussion_r2359952723
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/CachingSegmentArchiveReader.java:
##########
@@ -19,34 +19,123 @@
package org.apache.jackrabbit.oak.segment.spi.persistence.persistentcache;
import org.apache.jackrabbit.oak.commons.Buffer;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.segment.data.SegmentData;
import org.apache.jackrabbit.oak.segment.file.tar.SegmentGraph;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveEntry;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveReader;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
public class CachingSegmentArchiveReader implements SegmentArchiveReader {
+ private static final Logger LOG = LoggerFactory.getLogger(CachingSegmentArchiveReader.class);
+
+
@NotNull
private final PersistentCache persistentCache;
@NotNull
private final SegmentArchiveReader delegate;
+ private final ExecutorService prefetchExecutor;
+ private final Set<UUID> inFlightPrefetch =
+ Collections.newSetFromMap(
+ new ConcurrentHashMap<>());
+ private final boolean prefetchEnabled;
+ private final int prefetchMaxRefs;
+
+
+
public CachingSegmentArchiveReader(
@NotNull PersistentCache persistentCache,
@NotNull SegmentArchiveReader delegate) {
this.persistentCache = persistentCache;
this.delegate = delegate;
+ int threads = Integer.getInteger("oak.segment.cache.threads", 10);
+ this.prefetchEnabled = Boolean.getBoolean("oak.segment.cache.prefetch.enabled");
+ this.prefetchMaxRefs = Integer.getInteger("oak.segment.cache.prefetch.maxRefs", 20);
+ this.prefetchExecutor = Executors.newFixedThreadPool(threads);
}
@Override
@Nullable
public Buffer readSegment(long msb, long lsb) throws IOException {
- return persistentCache.readSegment(msb, lsb, () -> delegate.readSegment(msb, lsb));
+ Buffer buf = persistentCache.readSegment(msb, lsb, () -> delegate.readSegment(msb, lsb));
+ if (buf != null && prefetchEnabled) {
+ schedulePrefetch(msb, lsb, buf);
+ }
+ return buf;
+ }
+
+ private List<UUID> extractReferences(Buffer buffer) {
+ var data = SegmentData.newSegmentData(buffer);
+ int refs = data.getSegmentReferencesCount();
+ ArrayList<UUID> out = new ArrayList<>(refs);
+ for (int i = 0; i < refs; i++) {
+ out.add(new UUID(data.getSegmentReferenceMsb(i), data.getSegmentReferenceLsb(i)));
+ }
+ return out;
+ }
+
+ private void schedulePrefetch(long msb, long lsb, Buffer buffer) {
+ try {
+ List<UUID> refs = extractReferences(buffer);
+ int limit = Math.min(refs.size(), prefetchMaxRefs);
+ for (int i = 0; i < limit; i++) {
+ final UUID ref = refs.get(i);
+ final long rMsb = ref.getMostSignificantBits();
+ final long rLsb = ref.getLeastSignificantBits();
+
+ // Skip if already present in cache
+ if (persistentCache.containsSegment(rMsb, rLsb)) {
+ continue;
+ }
+
+ // Drop prefetch if already in progress for this segment
+ boolean registered = inFlightPrefetch.add(ref);
Review Comment:
Nice idea! I missed this trick so far in #2519 🙂
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/CachingSegmentArchiveReader.java:
##########
@@ -19,34 +19,123 @@
package org.apache.jackrabbit.oak.segment.spi.persistence.persistentcache;
import org.apache.jackrabbit.oak.commons.Buffer;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.segment.data.SegmentData;
import org.apache.jackrabbit.oak.segment.file.tar.SegmentGraph;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveEntry;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveReader;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
public class CachingSegmentArchiveReader implements SegmentArchiveReader {
+ private static final Logger LOG = LoggerFactory.getLogger(CachingSegmentArchiveReader.class);
+
+
@NotNull
private final PersistentCache persistentCache;
@NotNull
private final SegmentArchiveReader delegate;
+ private final ExecutorService prefetchExecutor;
+ private final Set<UUID> inFlightPrefetch =
+ Collections.newSetFromMap(
+ new ConcurrentHashMap<>());
+ private final boolean prefetchEnabled;
+ private final int prefetchMaxRefs;
+
+
+
public CachingSegmentArchiveReader(
@NotNull PersistentCache persistentCache,
@NotNull SegmentArchiveReader delegate) {
this.persistentCache = persistentCache;
this.delegate = delegate;
+ int threads = Integer.getInteger("oak.segment.cache.threads", 10);
+ this.prefetchEnabled = Boolean.getBoolean("oak.segment.cache.prefetch.enabled");
+ this.prefetchMaxRefs = Integer.getInteger("oak.segment.cache.prefetch.maxRefs", 20);
+ this.prefetchExecutor = Executors.newFixedThreadPool(threads);
}
@Override
@Nullable
public Buffer readSegment(long msb, long lsb) throws IOException {
- return persistentCache.readSegment(msb, lsb, () -> delegate.readSegment(msb, lsb));
+ Buffer buf = persistentCache.readSegment(msb, lsb, () -> delegate.readSegment(msb, lsb));
+ if (buf != null && prefetchEnabled) {
+ schedulePrefetch(msb, lsb, buf);
+ }
+ return buf;
+ }
+
+ private List<UUID> extractReferences(Buffer buffer) {
+ var data = SegmentData.newSegmentData(buffer);
+ int refs = data.getSegmentReferencesCount();
+ ArrayList<UUID> out = new ArrayList<>(refs);
+ for (int i = 0; i < refs; i++) {
+ out.add(new UUID(data.getSegmentReferenceMsb(i), data.getSegmentReferenceLsb(i)));
+ }
+ return out;
+ }
+
+ private void schedulePrefetch(long msb, long lsb, Buffer buffer) {
Review Comment:
Indeed, my tests with #2519, which implements a similar mechanism, but on a different level of abstraction, showed that dispatching from the caller thread hurts performance.
I even opted to trigger preloading only from within the load-callback, so preloading is completely off the critical path for cache hits.
Of course that's a trade-off, as it requires the caller thread to load one segment before any preloading happens.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11932: Segment prefetching for CachingSegmentReader [jackrabbit-oak]
Posted by "nfsantos (via GitHub)" <gi...@apache.org>.
nfsantos commented on code in PR #2513:
URL: https://github.com/apache/jackrabbit-oak/pull/2513#discussion_r2363106809
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/CachingSegmentArchiveReader.java:
##########
@@ -19,34 +19,123 @@
package org.apache.jackrabbit.oak.segment.spi.persistence.persistentcache;
import org.apache.jackrabbit.oak.commons.Buffer;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.segment.data.SegmentData;
import org.apache.jackrabbit.oak.segment.file.tar.SegmentGraph;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveEntry;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveReader;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
public class CachingSegmentArchiveReader implements SegmentArchiveReader {
+ private static final Logger LOG = LoggerFactory.getLogger(CachingSegmentArchiveReader.class);
+
+
@NotNull
private final PersistentCache persistentCache;
@NotNull
private final SegmentArchiveReader delegate;
+ private final ExecutorService prefetchExecutor;
+ private final Set<UUID> inFlightPrefetch =
+ Collections.newSetFromMap(
+ new ConcurrentHashMap<>());
+ private final boolean prefetchEnabled;
+ private final int prefetchMaxRefs;
+
+
+
public CachingSegmentArchiveReader(
@NotNull PersistentCache persistentCache,
@NotNull SegmentArchiveReader delegate) {
this.persistentCache = persistentCache;
this.delegate = delegate;
+ int threads = Integer.getInteger("oak.segment.cache.threads", 10);
+ this.prefetchEnabled = Boolean.getBoolean("oak.segment.cache.prefetch.enabled");
+ this.prefetchMaxRefs = Integer.getInteger("oak.segment.cache.prefetch.maxRefs", 20);
+ this.prefetchExecutor = Executors.newFixedThreadPool(threads);
}
@Override
@Nullable
public Buffer readSegment(long msb, long lsb) throws IOException {
- return persistentCache.readSegment(msb, lsb, () -> delegate.readSegment(msb, lsb));
+ Buffer buf = persistentCache.readSegment(msb, lsb, () -> delegate.readSegment(msb, lsb));
+ if (buf != null && prefetchEnabled) {
+ schedulePrefetch(msb, lsb, buf);
+ }
+ return buf;
+ }
+
+ private List<UUID> extractReferences(Buffer buffer) {
+ var data = SegmentData.newSegmentData(buffer);
+ int refs = data.getSegmentReferencesCount();
+ ArrayList<UUID> out = new ArrayList<>(refs);
+ for (int i = 0; i < refs; i++) {
+ out.add(new UUID(data.getSegmentReferenceMsb(i), data.getSegmentReferenceLsb(i)));
+ }
+ return out;
+ }
+
+ private void schedulePrefetch(long msb, long lsb, Buffer buffer) {
+ try {
+ List<UUID> refs = extractReferences(buffer);
+ int limit = Math.min(refs.size(), prefetchMaxRefs);
+ for (int i = 0; i < limit; i++) {
Review Comment:
You are getting a list with all the references but then potentially iterate over only a subset of them. You could save some work by extracting only the references that will be prefetched, this could also avoid allocating an array with all the refs. Using streams with `limit()`may be an easy way of implementing this optimization, as streams are computed lazily.
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/CachingSegmentArchiveReader.java:
##########
@@ -19,34 +19,123 @@
package org.apache.jackrabbit.oak.segment.spi.persistence.persistentcache;
import org.apache.jackrabbit.oak.commons.Buffer;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.segment.data.SegmentData;
import org.apache.jackrabbit.oak.segment.file.tar.SegmentGraph;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveEntry;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveReader;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
public class CachingSegmentArchiveReader implements SegmentArchiveReader {
+ private static final Logger LOG = LoggerFactory.getLogger(CachingSegmentArchiveReader.class);
+
+
@NotNull
private final PersistentCache persistentCache;
@NotNull
private final SegmentArchiveReader delegate;
+ private final ExecutorService prefetchExecutor;
+ private final Set<UUID> inFlightPrefetch =
+ Collections.newSetFromMap(
+ new ConcurrentHashMap<>());
+ private final boolean prefetchEnabled;
+ private final int prefetchMaxRefs;
+
+
+
public CachingSegmentArchiveReader(
@NotNull PersistentCache persistentCache,
@NotNull SegmentArchiveReader delegate) {
this.persistentCache = persistentCache;
this.delegate = delegate;
+ int threads = Integer.getInteger("oak.segment.cache.threads", 10);
+ this.prefetchEnabled = Boolean.getBoolean("oak.segment.cache.prefetch.enabled");
+ this.prefetchMaxRefs = Integer.getInteger("oak.segment.cache.prefetch.maxRefs", 20);
+ this.prefetchExecutor = Executors.newFixedThreadPool(threads);
}
@Override
@Nullable
public Buffer readSegment(long msb, long lsb) throws IOException {
- return persistentCache.readSegment(msb, lsb, () -> delegate.readSegment(msb, lsb));
+ Buffer buf = persistentCache.readSegment(msb, lsb, () -> delegate.readSegment(msb, lsb));
+ if (buf != null && prefetchEnabled) {
+ schedulePrefetch(msb, lsb, buf);
+ }
+ return buf;
+ }
+
+ private List<UUID> extractReferences(Buffer buffer) {
+ var data = SegmentData.newSegmentData(buffer);
+ int refs = data.getSegmentReferencesCount();
+ ArrayList<UUID> out = new ArrayList<>(refs);
+ for (int i = 0; i < refs; i++) {
+ out.add(new UUID(data.getSegmentReferenceMsb(i), data.getSegmentReferenceLsb(i)));
+ }
+ return out;
+ }
+
+ private void schedulePrefetch(long msb, long lsb, Buffer buffer) {
+ try {
+ List<UUID> refs = extractReferences(buffer);
+ int limit = Math.min(refs.size(), prefetchMaxRefs);
+ for (int i = 0; i < limit; i++) {
+ final UUID ref = refs.get(i);
+ final long rMsb = ref.getMostSignificantBits();
+ final long rLsb = ref.getLeastSignificantBits();
+
+ // Skip if already present in cache
+ if (persistentCache.containsSegment(rMsb, rLsb)) {
+ continue;
+ }
Review Comment:
Maybe it's better to do this check in the worker thread, just before the segment is downloaded? The task that is scheduled to download the segment might not execute for a while, if the worker pool is busy, so from this point until the actual download, the segment might have been added to the cache. Or we can leave the check here and do another one before trying to download.
It would also be better to have a mechanism similar to the Guava loading cache, where thread requesting a segment that is not in the cache but is being downloaded will block waiting for the first download to complete, instead of starting a new download. This would avoid duplicate downloads.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11932: Segment prefetching for CachingSegmentReader [jackrabbit-oak]
Posted by "smiroslav (via GitHub)" <gi...@apache.org>.
smiroslav commented on code in PR #2513:
URL: https://github.com/apache/jackrabbit-oak/pull/2513#discussion_r2358290211
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/CachingSegmentArchiveReader.java:
##########
@@ -19,34 +19,123 @@
package org.apache.jackrabbit.oak.segment.spi.persistence.persistentcache;
import org.apache.jackrabbit.oak.commons.Buffer;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.segment.data.SegmentData;
import org.apache.jackrabbit.oak.segment.file.tar.SegmentGraph;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveEntry;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveReader;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
public class CachingSegmentArchiveReader implements SegmentArchiveReader {
+ private static final Logger LOG = LoggerFactory.getLogger(CachingSegmentArchiveReader.class);
+
+
@NotNull
private final PersistentCache persistentCache;
@NotNull
private final SegmentArchiveReader delegate;
+ private final ExecutorService prefetchExecutor;
+ private final Set<UUID> inFlightPrefetch =
+ Collections.newSetFromMap(
+ new ConcurrentHashMap<>());
+ private final boolean prefetchEnabled;
+ private final int prefetchMaxRefs;
+
+
+
public CachingSegmentArchiveReader(
@NotNull PersistentCache persistentCache,
@NotNull SegmentArchiveReader delegate) {
this.persistentCache = persistentCache;
this.delegate = delegate;
+ int threads = Integer.getInteger("oak.segment.cache.threads", 10);
+ this.prefetchEnabled = Boolean.getBoolean("oak.segment.cache.prefetch.enabled");
+ this.prefetchMaxRefs = Integer.getInteger("oak.segment.cache.prefetch.maxRefs", 20);
+ this.prefetchExecutor = Executors.newFixedThreadPool(threads);
}
@Override
@Nullable
public Buffer readSegment(long msb, long lsb) throws IOException {
- return persistentCache.readSegment(msb, lsb, () -> delegate.readSegment(msb, lsb));
+ Buffer buf = persistentCache.readSegment(msb, lsb, () -> delegate.readSegment(msb, lsb));
+ if (buf != null && prefetchEnabled) {
+ schedulePrefetch(msb, lsb, buf);
+ }
+ return buf;
+ }
+
+ private List<UUID> extractReferences(Buffer buffer) {
+ var data = SegmentData.newSegmentData(buffer);
+ int refs = data.getSegmentReferencesCount();
+ ArrayList<UUID> out = new ArrayList<>(refs);
+ for (int i = 0; i < refs; i++) {
+ out.add(new UUID(data.getSegmentReferenceMsb(i), data.getSegmentReferenceLsb(i)));
+ }
+ return out;
+ }
+
+ private void schedulePrefetch(long msb, long lsb, Buffer buffer) {
Review Comment:
@nfsantos made a good remark that this method, from the start, should be executed in a separate thread context, so it does not impact execution time for the thread invoking CachingSegmentArchiveReader#readSegment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org