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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
