Merge pull request #386 from LiUSemWeb/385-optimized-persisted-cache-…

…for-cardinality-queries

385 optimized persisted cache for cardinality queries

hefquin-base/src/main/java/se/liu/ida/hefquin/base/datastructures/impl/cache/CacheEntry.java

@@ -9,4 +9,11 @@ public interface CacheEntry<ObjectType>
 	 * Returns the object that is cached via this cache entry.
 	 */
 	ObjectType getObject();
+
+	/**
+	 * Returns the time at which this cache entry was created.
+	 */
+	long createdAt();
 }
+
+

hefquin-base/src/main/java/se/liu/ida/hefquin/base/datastructures/impl/cache/CacheEntryBase.java

@@ -1,17 +1,45 @@
 package se.liu.ida.hefquin.base.datastructures.impl.cache;
 
-public class CacheEntryBase<ObjectType> implements CacheEntry<ObjectType>
+import java.io.Serializable;
+import java.time.Instant;
+
+public class CacheEntryBase<ObjectType> implements CacheEntry<ObjectType>, Serializable
 {
+	private static final long serialVersionUID = 1L;
 	protected final ObjectType obj;
+	protected final long creationTime;
 
 	public CacheEntryBase( final ObjectType obj ) {
+		this( obj, Instant.now().toEpochMilli() );
+	}
+
+	public CacheEntryBase( final ObjectType obj, final long creationTime ) {
 		assert obj != null;
 		this.obj = obj;
+		this.creationTime = creationTime;
 	}
 
 	@Override
 	public ObjectType getObject() {
 		return obj;
 	}
 
+	@Override
+	public long createdAt() {
+		return creationTime;
+	}
+
+	@Override
+	public boolean equals( final Object obj ) {
+		if ( this == obj )
+			return true;
+		if ( obj == null || getClass() != obj.getClass() )
+			return false;
+		return this.getObject().equals(((CacheEntryBase<?>) obj).getObject());
+	}
+
+	@Override
+	public int hashCode() {
+		return getObject().hashCode();
+	}
 }

hefquin-base/src/main/java/se/liu/ida/hefquin/base/datastructures/impl/cache/CacheEntryBaseFactory.java

@@ -7,7 +7,7 @@ public class CacheEntryBaseFactory<ObjectType> implements CacheEntryFactory<Cach
 {
 	@Override
 	public CacheEntryBase<ObjectType> createCacheEntry( final ObjectType obj ) {
-		return new CacheEntryBase<>(obj);
+		return new CacheEntryBase<>( obj );
 	}
 
 }

hefquin-base/src/main/java/se/liu/ida/hefquin/base/datastructures/impl/cache/CacheEntryFactory.java

@@ -8,5 +8,5 @@ public interface CacheEntryFactory<EntryType extends CacheEntry<ObjectType>, Obj
 	/**
 	 * Creates and returns a new EntryType object that wraps the given object.
 	 */
-	EntryType createCacheEntry(ObjectType obj);
+	EntryType createCacheEntry( ObjectType obj );
 }

hefquin-base/src/main/java/se/liu/ida/hefquin/base/datastructures/impl/cache/CacheInvalidationPolicy.java

@@ -11,5 +11,5 @@ public interface CacheInvalidationPolicy<EntryType extends CacheEntry<ObjectType
 	 * Returns <code>true</code> if the given cache entry is still
 	 * valid according to this cache invalidation policy.
 	 */
-	boolean isStillValid(EntryType e);
+	boolean isStillValid( EntryType e );
 }

hefquin-base/src/main/java/se/liu/ida/hefquin/base/datastructures/impl/cache/CacheInvalidationPolicyTimeToLive.java

@@ -0,0 +1,21 @@
+package se.liu.ida.hefquin.base.datastructures.impl.cache;
+
+import java.time.Instant;
+
+public class CacheInvalidationPolicyTimeToLive<EntryType extends CacheEntry<ObjectType>, ObjectType>
+		implements CacheInvalidationPolicy<EntryType, ObjectType>
+{
+	protected final long timeToLive;
+
+	public CacheInvalidationPolicyTimeToLive( final long timeToLive ){
+		this.timeToLive = timeToLive;
+	}
+
+	/**
+	 * Returns <code>true</code> if the given cache entry has not reached
+	 * the time to live considered by this policy.
+	 */
+	public boolean isStillValid( final EntryType e ) {
+		return e.createdAt() + timeToLive > Instant.now().toEpochMilli();
+	}
+}

hefquin-engine/pom.xml

@@ -29,5 +29,10 @@
       <artifactId>hefquin-graphqlconnector</artifactId>
       <version>0.0.4-SNAPSHOT</version>
     </dependency>
+    <dependency>
+      <groupId>net.openhft</groupId>
+      <artifactId>chronicle-map</artifactId>
+      <version>3.27ea0</version>
+    </dependency>
   </dependencies>
 </project>

hefquin-engine/src/main/java/se/liu/ida/hefquin/engine/federation/access/impl/FederationAccessManagerWithChronicleMapCache.java

@@ -0,0 +1,217 @@
+package se.liu.ida.hefquin.engine.federation.access.impl;
+
+import java.io.IOException;
+import java.util.Date;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+import se.liu.ida.hefquin.base.datastructures.impl.cache.CacheEntry;
+import se.liu.ida.hefquin.base.datastructures.impl.cache.CacheEntryFactory;
+import se.liu.ida.hefquin.base.datastructures.impl.cache.CacheInvalidationPolicy;
+import se.liu.ida.hefquin.base.datastructures.impl.cache.CacheInvalidationPolicyTimeToLive;
+import se.liu.ida.hefquin.base.datastructures.impl.cache.CachePolicies;
+import se.liu.ida.hefquin.base.datastructures.impl.cache.CacheReplacementPolicy;
+import se.liu.ida.hefquin.base.datastructures.impl.cache.CacheReplacementPolicyFactory;
+import se.liu.ida.hefquin.base.datastructures.impl.cache.CacheReplacementPolicyLRU;
+import se.liu.ida.hefquin.engine.federation.BRTPFServer;
+import se.liu.ida.hefquin.engine.federation.SPARQLEndpoint;
+import se.liu.ida.hefquin.engine.federation.TPFServer;
+import se.liu.ida.hefquin.engine.federation.access.BRTPFRequest;
+import se.liu.ida.hefquin.engine.federation.access.CardinalityResponse;
+import se.liu.ida.hefquin.engine.federation.access.DataRetrievalResponse;
+import se.liu.ida.hefquin.engine.federation.access.FederationAccessException;
+import se.liu.ida.hefquin.engine.federation.access.FederationAccessManager;
+import se.liu.ida.hefquin.engine.federation.access.SPARQLRequest;
+import se.liu.ida.hefquin.engine.federation.access.TPFRequest;
+import se.liu.ida.hefquin.engine.federation.access.impl.cache.CardinalityCacheEntry;
+import se.liu.ida.hefquin.engine.federation.access.impl.cache.CardinalityCacheEntryFactory;
+import se.liu.ida.hefquin.engine.federation.access.impl.cache.CardinalityCacheKey;
+import se.liu.ida.hefquin.engine.federation.access.impl.cache.ChronicleMapCardinalityCache;
+import se.liu.ida.hefquin.engine.federation.access.impl.response.CachedCardinalityResponseImpl;
+
+/**
+ * A FederationAccessManager implementation that incorporates persistent disk
+ * caching of cardinality requests.
+ */
+public class FederationAccessManagerWithChronicleMapCache extends FederationAccessManagerWithCache
+{
+	protected final ChronicleMapCardinalityCache cardinalityCache;
+	protected final static int defaultCacheCapacity = 1000;
+
+	public FederationAccessManagerWithChronicleMapCache( final FederationAccessManager fedAccMan,
+	                                                     final int cacheCapacity,
+	                                                     final CachePolicies<Key, CompletableFuture<? extends DataRetrievalResponse>, ? extends CacheEntry<CompletableFuture<? extends DataRetrievalResponse>>> cachePolicies,
+	                                                     final CachePolicies<CardinalityCacheKey, Integer, CardinalityCacheEntry> cardinalityCachePolicies )
+		throws IOException
+	{
+		super( fedAccMan, cacheCapacity, cachePolicies );
+		cardinalityCache = new ChronicleMapCardinalityCache( cardinalityCachePolicies, cacheCapacity );
+	}
+
+	public FederationAccessManagerWithChronicleMapCache( final FederationAccessManager fedAccMan,
+	                                                     final int cacheCapacity,
+	                                                     final long timeToLive )
+			throws IOException
+	{
+		this( fedAccMan,
+		      cacheCapacity,
+		      new MyDefaultCachePolicies(),
+		      new MyDefaultCardinalityCachePolicies( timeToLive ) );
+	}
+
+	/**
+	 * Creates a {@link FederationAccessManagerWithChronicleMapCache} with the default configuration.
+	 */
+	public FederationAccessManagerWithChronicleMapCache( final ExecutorService execService ) throws IOException
+	{
+		this( new AsyncFederationAccessManagerImpl( execService ),
+		      defaultCacheCapacity,
+		      new MyDefaultCachePolicies(),
+		      new MyDefaultCardinalityCachePolicies() );
+	}
+
+	@Override
+	public CompletableFuture<CardinalityResponse> issueCardinalityRequest( final SPARQLRequest req,
+	                                                                       final SPARQLEndpoint fm )
+			throws FederationAccessException
+	{
+		final CardinalityCacheKey key = new CardinalityCacheKey( req, fm );
+		final Date requestStartTime = new Date();
+		final CardinalityCacheEntry cachedEntry = cardinalityCache.get( key );
+		final Date requestEndTime = new Date();
+		if ( cachedEntry != null ) {
+			cacheHitsSPARQLCardinality++;
+			final CardinalityResponse cr = new CachedCardinalityResponseImpl( fm,
+			                                                                  req,
+			                                                                  cachedEntry.getObject(),
+			                                                                  requestStartTime,
+			                                                                  requestEndTime );
+			return CompletableFuture.completedFuture( cr );
+		}
+
+		final CompletableFuture<CardinalityResponse> newResponse = fedAccMan.issueCardinalityRequest( req, fm );
+		newResponse.thenAccept( value -> {
+			cardinalityCache.put( key, value.getCardinality() );
+		} );
+		return newResponse;
+	}
+
+	@Override
+	public CompletableFuture<CardinalityResponse> issueCardinalityRequest( final TPFRequest req,
+	                                                                       final TPFServer fm )
+			throws FederationAccessException
+	{
+		final CardinalityCacheKey key = new CardinalityCacheKey( req, fm );
+		final Date requestStartTime = new Date();
+		final CardinalityCacheEntry cachedEntry = cardinalityCache.get( key );
+		final Date requestEndTime = new Date();
+		if ( cachedEntry != null ) {
+			cacheHitsTPFCardinality++;
+			final CardinalityResponse cr = new CachedCardinalityResponseImpl( fm,
+			                                                                  req,
+			                                                                  cachedEntry.getObject(),
+			                                                                  requestStartTime,
+			                                                                  requestEndTime );
+			return CompletableFuture.completedFuture( cr );
+		}
+
+		final CompletableFuture<CardinalityResponse> newResponse = fedAccMan.issueCardinalityRequest( req, fm );
+		newResponse.thenAccept( value -> {
+			cardinalityCache.put( key, value.getCardinality() );
+		} );
+		return newResponse;
+	}
+
+	@Override
+	public CompletableFuture<CardinalityResponse> issueCardinalityRequest( final TPFRequest req,
+	                                                                       final BRTPFServer fm )
+			throws FederationAccessException
+	{
+		final CardinalityCacheKey key = new CardinalityCacheKey( req, fm );
+		final Date requestStartTime = new Date();
+		final CardinalityCacheEntry cachedEntry = cardinalityCache.get( key );
+		final Date requestEndTime = new Date();
+		if ( cachedEntry != null ) {
+			cacheHitsTPFCardinality++;
+			final CardinalityResponse cr = new CachedCardinalityResponseImpl( fm,
+			                                                                  req,
+			                                                                  cachedEntry.getObject(),
+			                                                                  requestStartTime,
+			                                                                  requestEndTime );
+			return CompletableFuture.completedFuture( cr );
+		}
+
+		final CompletableFuture<CardinalityResponse> newResponse = fedAccMan.issueCardinalityRequest( req, fm );
+		newResponse.thenAccept( value -> {
+			cardinalityCache.put( key, value.getCardinality() );
+		} );
+		return newResponse;
+	}
+
+	@Override
+	public CompletableFuture<CardinalityResponse> issueCardinalityRequest( final BRTPFRequest req,
+	                                                                       final BRTPFServer fm )
+			throws FederationAccessException
+	{
+		final CardinalityCacheKey key = new CardinalityCacheKey( req, fm );
+		final Date requestStartTime = new Date();
+		final CardinalityCacheEntry cachedEntry = cardinalityCache.get( key );
+		final Date requestEndTime = new Date();
+		if ( cachedEntry != null ) {
+			cacheHitsTPFCardinality++;
+			final CardinalityResponse cr = new CachedCardinalityResponseImpl( fm,
+			                                                                  req,
+			                                                                  cachedEntry.getObject(),
+			                                                                  requestStartTime,
+			                                                                  requestEndTime );
+			return CompletableFuture.completedFuture( cr );
+		}
+
+		final CompletableFuture<CardinalityResponse> newResponse = fedAccMan.issueCardinalityRequest( req, fm );
+		newResponse.thenAccept( value -> {
+			cardinalityCache.put( key, value.getCardinality() );
+		} );
+		return newResponse;
+	}
+
+	/**
+	 * Clears the persisted cardinality cache map.
+	 */
+	public void clearCardinalityCache() {
+		cardinalityCache.clear();
+	}
+
+	protected static class MyDefaultCardinalityCachePolicies implements CachePolicies<CardinalityCacheKey, Integer, CardinalityCacheEntry>
+	{
+		protected final long timeToLive;
+		protected final static long defaultTimeToLive = 300_000; // 5 minutes 
+
+		public MyDefaultCardinalityCachePolicies() {
+			this( defaultTimeToLive );
+		}
+
+		public MyDefaultCardinalityCachePolicies( final long timeToLive ) {
+			this.timeToLive = timeToLive;
+		}
+
+		@Override
+		public CacheEntryFactory<CardinalityCacheEntry, Integer> getEntryFactory() {
+			return new CardinalityCacheEntryFactory();
+		}
+
+		@Override
+		public CacheReplacementPolicyFactory<CardinalityCacheKey, Integer, CardinalityCacheEntry> getReplacementPolicyFactory() {
+			return new CacheReplacementPolicyFactory<>() {
+				@Override
+				public CacheReplacementPolicy<CardinalityCacheKey, Integer, CardinalityCacheEntry> create() {
+					return new CacheReplacementPolicyLRU<>();
+				}
+			};
+		}
+
+		@Override
+		public CacheInvalidationPolicy<CardinalityCacheEntry, Integer> getInvalidationPolicy() {
+			return new CacheInvalidationPolicyTimeToLive<>( timeToLive );
+		}
+	} // end of MyDefaultCachePolicies
+}

hefquin-engine/src/main/java/se/liu/ida/hefquin/engine/federation/access/impl/FederationAccessManagerWithPersistedDiskCache.java

@@ -15,6 +15,8 @@
 import se.liu.ida.hefquin.engine.federation.access.FederationAccessManager;
 import se.liu.ida.hefquin.engine.federation.access.SPARQLRequest;
 import se.liu.ida.hefquin.engine.federation.access.TPFRequest;
+import se.liu.ida.hefquin.engine.federation.access.impl.cache.CardinalityCacheKey;
+import se.liu.ida.hefquin.engine.federation.access.impl.cache.PersistableCardinalityCacheImpl;
 
 /**
  * A FederationAccessManager implementation that incorporates persistent disk
@@ -122,7 +124,7 @@ public CompletableFuture<CardinalityResponse> issueCardinalityRequest( final BRT
 			cacheHitsBRTPFCardinality++;
 			return cachedResponse;
 		}
-		final CompletableFuture<CardinalityResponse> newResponse = fedAccMan.issueCardinalityRequest( req, fm) ;
+		final CompletableFuture<CardinalityResponse> newResponse = fedAccMan.issueCardinalityRequest( req, fm );
 		cardinalityCache.put( key, newResponse );
 		newResponse.thenRun( () -> cardinalityCache.save() );
 		return newResponse;

hefquin-engine/src/main/java/se/liu/ida/hefquin/engine/federation/access/impl/cache/CardinalityCacheEntry.java

@@ -0,0 +1,20 @@
+package se.liu.ida.hefquin.engine.federation.access.impl.cache;
+
+import se.liu.ida.hefquin.base.datastructures.impl.cache.CacheEntryBase;
+
+/**
+ * An entry used when caching cardinality requests.
+ */
+public class CardinalityCacheEntry extends CacheEntryBase<Integer>
+{
+	private static final long serialVersionUID = 1L;
+
+	public CardinalityCacheEntry( final Integer cardinality, final long entryCreatedAt ) {
+		super( cardinality, entryCreatedAt );
+	}
+
+	@Override
+	public String toString() {
+		return "CardinalityCacheEntry{cardinality='" + getObject() + "'}";
+	}
+}

hefquin-engine/src/main/java/se/liu/ida/hefquin/engine/federation/access/impl/cache/CardinalityCacheEntryFactory.java

@@ -0,0 +1,16 @@
+package se.liu.ida.hefquin.engine.federation.access.impl.cache;
+
+import java.time.Instant;
+
+import se.liu.ida.hefquin.base.datastructures.impl.cache.CacheEntryFactory;
+
+/**
+ * An entry used when caching cardinality requests.
+ */
+public class CardinalityCacheEntryFactory implements CacheEntryFactory<CardinalityCacheEntry, Integer>
+{
+	@Override
+	public CardinalityCacheEntry createCacheEntry( final Integer cardinality ) {
+		return new CardinalityCacheEntry( cardinality, Instant.now().toEpochMilli() );
+	}
+}