AfflictionNetworks/depot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added support for clearing caches.

Browse Source
This commit is contained in:
Michael Bayne
2011-06-27 21:08:01 +00:00
parent d5b06f23e2
commit 3f8f15af9f
7 changed files with 274 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/samskivert/depot/CacheAdapter.java
+8
View File
@@ -58,6 +58,14 @@ public interface CacheAdapter
*/
public <T> Iterable<Serializable> enumerate (String cacheId);
/**
* Clears the cache with the specified name.
*
* @param localOnly if true, only clear the cache on the local node, do not broadcast
* instructions for all distributed nodes to clear this cache as well.
*/
public void clear (String cacheId, boolean localOnly);
/**
* Shut down all operations, e.g. persisting memory contents to disk.
*/
src/main/java/com/samskivert/depot/DepotRepository.java
+1 -44
View File
@@ -240,50 +240,7 @@ public abstract class DepotRepository
Class<T> type, CacheStrategy cache, Iterable<? extends QueryClause> clauses)
throws DatabaseException
{
DepotMarshaller<T> marsh = _ctx.getMarshaller(type);
switch (cache) {
case LONG_KEYS: case SHORT_KEYS: case BEST: case RECORDS:
String reason = null;
if (marsh.getTableName() == null) {
reason = type + " is computed";
} else if (!marsh.hasPrimaryKey()) {
reason = type + " has no primary key";
} else {
for (QueryClause clause : clauses) {
if (clause instanceof FieldOverride) {
reason = "query uses a FieldOverride clause";
break;
}
}
}
if (cache == CacheStrategy.BEST) {
cache = (reason != null) ? CacheStrategy.NONE : CacheStrategy.SHORT_KEYS;
} else if (reason != null) {
// if user explicitly asked for a strategy we can't do, protest
throw new IllegalArgumentException(
"Cannot use " + cache + " strategy because " + reason);
}
break;
case NONE: case CONTENTS:
break; // NONE and CONTENTS can always be used.
}
if (!_ctx.isUsingCache()) {
cache = CacheStrategy.NONE;
}
switch (cache) {
case SHORT_KEYS: case LONG_KEYS: case RECORDS:
return _ctx.invoke(new FindAllQuery.WithCache<T>(_ctx, type, clauses, cache));
default:
return _ctx.invoke(new FindAllQuery.Explicitly<T>(
_ctx, type, clauses, cache == CacheStrategy.CONTENTS));
}
return _ctx.invoke(FindAllQuery.newCachedFullRecordQuery(_ctx, type, cache, clauses));
}
/**
src/main/java/com/samskivert/depot/EHCacheAdapter.java
+9
View File
@@ -173,6 +173,15 @@ public class EHCacheAdapter
return result;
}
// from CacheAdapter
public void clear (String cacheId, boolean localOnly)
{
EHCacheBin<?> bin = _bins.get(cacheId);
if (bin != null) {
bin.getCache().removeAll(localOnly);
}
}
// from CacheAdapter
public void shutdown ()
{
src/main/java/com/samskivert/depot/PersistenceContext.java
+27
View File
@@ -425,6 +425,33 @@ public class PersistenceContext
}
}
/**
* Requests that the cache for the specified persistent record be cleared. Note that this
* clears only the <em>by-primary-key</em> cache for the index in question. It does not clear
* the <em>keyset</em> or <em>contents</em> query caches. Use {@link Query#clearCache} to clear
* these caches.
*
* @param localOnly if true, only the cache in this JVM will be cleared, no broadcast message
* will be sent to instruct all distributed nodes to also clear this cache.
*/
public void cacheClear (Class<? extends PersistentRecord> pClass, boolean localOnly)
{
cacheClear(pClass.getName(), localOnly);
}
/**
* Requests that the cache with the specified id be cleared.
*
* @param localOnly if true, only the cache in this JVM will be cleared, no broadcast message
* will be sent to instruct all distributed nodes to also clear this cache.
*/
public void cacheClear (String cacheId, boolean localOnly)
{
if (_cache != null) {
_cache.clear(cacheId, localOnly);
}
}
/**
* Registers a new cache listener with the cache associated with the given class.
*/
src/main/java/com/samskivert/depot/Query.java
+31
View File
@@ -555,6 +555,37 @@ public class Query<T extends PersistentRecord>
return _repo.deleteAll(_pclass, _where, _limit, invalidator);
}
/**
* Clears the cache for the query configured by this instance. If the query would not result in
* cache usage for whatever reason, this will NOOP.
*
* <p> Note: Depot uses numerous caches, so it is important to know what you are doing. If the
* records fetched by this query have a primary key, Depot performs a two-phase query where the
* primary keys for the records that satisify this query are first fetched and stored in a
* <em>keyset</em> cache, then the actual records needed to satisfy the query are fetched using
* the <em>by-primary-key</em> cache. This method will only clear the <em>keyset</em> cache.
* That may be sufficient for your purposes, or you may need to call {@link
* PersistenceContext#cacheClear(Class,boolean)} to clear the <em>by-primary-key</em> cache as
* well. If the record in question does not define a primary key, Depot will use a
* <em>contents</em> cache to store the entire records fetched as a result of executing this
* query. In that case, calling this method is sufficient to ensure that no cached data will be
* used to fulfill subsequent similarly configured queries. </p>
*
* <p> Finally, note that query configuration other than {@link #cache} (i.e. {@link #limit},
* {@link #groupBy}, {@link #join}, etc.) will have no influence on this operation. </p>
*
* @param localOnly if true, only the cache in this JVM will be cleared, no broadcast message
* will be sent to instruct all distributed nodes to also clear this cache.
*/
public void clearCache (boolean localOnly)
{
String cacheId = FindAllQuery.newCachedFullRecordQuery(
_ctx, _pclass, _cache, getClauses()).getCacheId();
if (cacheId != null) {
_ctx.cacheClear(cacheId, localOnly);
}
}
protected Query (PersistenceContext ctx, DepotRepository repo, Class<T> pclass)
{
_ctx = ctx;
src/main/java/com/samskivert/depot/impl/FindAllQuery.java
+146 -87
View File
@@ -47,24 +47,66 @@ public abstract class FindAllQuery<T extends PersistentRecord,R>
extends Fetcher<List<R>>
{
/**
* A base class for queries that fetch a full record at a time.
* A base class for cached queries that fetch a full record at a time.
*/
public static abstract class FullRecordQuery<T extends PersistentRecord>
extends FindAllQuery<T, T>
public static abstract class CachedFullRecordQuery<T extends PersistentRecord>
extends FullRecordQuery<T>
{
protected FullRecordQuery (PersistenceContext ctx, Class<T> type) {
super(type, ctx.getMarshaller(type), new CloningCloner<T>());
_dmarsh = ctx.getMarshaller(type);
protected CachedFullRecordQuery (PersistenceContext ctx, Class<T> type) {
super(ctx, type);
}
protected DepotMarshaller<T> _dmarsh;
/**
* Returns the id of the cache in which this query's results will be stored, or null if the
* query is not configured to use a cache.
*/
public String getCacheId () {
return (_qkey == null) ? null : _qkey.getCacheId();
}
protected SimpleCacheKey _qkey;
}
/**
* The two-pass collection query implementation. See {@link DepotRepository#findAll} for
* details.
*/
public static class WithCache<T extends PersistentRecord> extends FullRecordQuery<T>
public static class WithKeys<T extends PersistentRecord> extends FullRecordQuery<T>
{
public WithKeys (PersistenceContext ctx, Iterable<Key<T>> keys)
throws DatabaseException
{
super(ctx, keys.iterator().next().getPersistentClass());
_keys = keys;
}
@Override // from Fetcher
public List<T> getCachedResult (PersistenceContext ctx)
{
// look up what we can from the cache
_fetchKeys = loadFromCache(ctx, _keys, _entities);
// if we found everything, we can just return our result straight away, yay!
return _fetchKeys.isEmpty() ? resolve(_keys, _entities) : null;
}
// from Fetcher
public List<T> invoke (PersistenceContext ctx, Connection conn, DatabaseLiaison liaison)
throws SQLException
{
return loadAndResolve(ctx, conn, _keys, _fetchKeys, _entities, null);
}
protected Iterable<Key<T>> _keys;
protected Set<Key<T>> _fetchKeys;
protected Map<Key<T>, T> _entities = Maps.newHashMap();
}
/**
* The two-pass collection query implementation. See {@link DepotRepository#findAll} for
* details.
*/
public static class WithCache<T extends PersistentRecord> extends CachedFullRecordQuery<T>
{
public WithCache (PersistenceContext ctx, Class<T> type,
Iterable<? extends QueryClause> clauses, CacheStrategy strategy)
@@ -94,7 +136,6 @@ public abstract class FindAllQuery<T extends PersistentRecord,R>
default:
throw new IllegalArgumentException("Unexpected cache strategy: " + strategy);
}
}
@Override // from Fetcher
@@ -148,7 +189,6 @@ public abstract class FindAllQuery<T extends PersistentRecord,R>
return loadAndResolve(ctx, conn, _keys, _fetchKeys, _entities, stmtString);
}
protected SimpleCacheKey _qkey;
protected CacheCategory _category;
protected SelectClause _select;
protected KeySet<T> _keys;
@@ -156,60 +196,20 @@ public abstract class FindAllQuery<T extends PersistentRecord,R>
protected Map<Key<T>, T> _entities = Maps.newHashMap();
}
/**
* The two-pass collection query implementation. See {@link DepotRepository#findAll} for
* details.
*/
public static class WithKeys<T extends PersistentRecord> extends FullRecordQuery<T>
{
public WithKeys (PersistenceContext ctx, Iterable<Key<T>> keys)
throws DatabaseException
{
super(ctx, keys.iterator().next().getPersistentClass());
_keys = keys;
}
@Override // from Fetcher
public List<T> getCachedResult (PersistenceContext ctx)
{
// look up what we can from the cache
_fetchKeys = loadFromCache(ctx, _keys, _entities);
// if we found everything, we can just return our result straight away, yay!
return _fetchKeys.isEmpty() ? resolve(_keys, _entities) : null;
}
// from Fetcher
public List<T> invoke (PersistenceContext ctx, Connection conn, DatabaseLiaison liaison)
throws SQLException
{
return loadAndResolve(ctx, conn, _keys, _fetchKeys, _entities, null);
}
protected Iterable<Key<T>> _keys;
protected Set<Key<T>> _fetchKeys;
protected Map<Key<T>, T> _entities = Maps.newHashMap();
}
/**
* The single-pass collection query implementation. See {@link DepotRepository#findAll} for
* details.
*/
public static class Explicitly<T extends PersistentRecord> extends FullRecordQuery<T>
public static class Explicitly<T extends PersistentRecord> extends CachedFullRecordQuery<T>
{
public Explicitly (PersistenceContext ctx, Class<T> type,
Iterable<? extends QueryClause> clauses, boolean cachedContents)
throws DatabaseException
{
super(ctx, type);
_select = new SelectClause(type, _dmarsh.getSelections(), clauses);
if (cachedContents) {
_qkey = new SimpleCacheKey(_dmarsh.getTableName() + "Contents", _select.toString());
} else {
_qkey = null;
}
_qkey = !cachedContents ? null :
new SimpleCacheKey(_dmarsh.getTableName() + "Contents", _select.toString());
}
@Override // from Fetcher
@@ -246,7 +246,6 @@ public abstract class FindAllQuery<T extends PersistentRecord,R>
}
protected SelectClause _select;
protected SimpleCacheKey _qkey;
}
public static class Projection<T extends PersistentRecord,R>
@@ -286,47 +285,51 @@ public abstract class FindAllQuery<T extends PersistentRecord,R>
protected DepotTypes _types;
}
protected static class ProjectionQueryMarshaller<T extends PersistentRecord,R>
implements QueryMarshaller<T,R>
public static <T extends PersistentRecord> CachedFullRecordQuery<T> newCachedFullRecordQuery (
PersistenceContext ctx, Class<T> type, CacheStrategy strategy,
Iterable<? extends QueryClause> clauses)
{
public ProjectionQueryMarshaller (Projector<T,R> cset, DepotTypes types) {
_cset = cset;
_types = types;
}
DepotMarshaller<T> marsh = ctx.getMarshaller(type);
public String getTableName () {
return _types.getTableName(_cset.ptype);
}
switch (strategy) {
case LONG_KEYS: case SHORT_KEYS: case BEST: case RECORDS:
String reason = null;
if (marsh.getTableName() == null) {
reason = type + " is computed";
public SQLExpression<?>[] getSelections () {
return _cset.selexps;
}
} else if (!marsh.hasPrimaryKey()) {
reason = type + " has no primary key";
public Key<T> getPrimaryKey (Object object) {
return _types.getMarshaller(_cset.ptype).getPrimaryKey(object);
}
public R createObject (ResultSet rs) throws SQLException {
Object[] data = new Object[_cset.selexps.length];
for (int ii = 0; ii < data.length; ii++) {
SQLExpression<?> exp = _cset.selexps[ii];
if (exp instanceof ColumnExp<?>) {
ColumnExp<?> col = (ColumnExp<?>)exp;
data[ii] = _types.getMarshaller(col.getPersistentClass()).
getFieldMarshaller(col.name).getFromSet(rs, ii+1);
} else {
// TEMP: in the case of selecting computed expressions, we rely on the types to
// be correct by construction; TODO: this will probably bite us when JDBC
// drivers choose Long instead of Integer or whatnot, so we'll need to set up
// more complex machinery
data[ii] = rs.getObject(ii+1);
} else {
for (QueryClause clause : clauses) {
if (clause instanceof FieldOverride) {
reason = "query uses a FieldOverride clause";
break;
}
}
}
return _cset.createObject(data);
if (strategy == CacheStrategy.BEST) {
strategy = (reason != null) ? CacheStrategy.NONE : CacheStrategy.SHORT_KEYS;
} else if (reason != null) {
// if user explicitly asked for a strategy we can't do, protest
throw new IllegalArgumentException(
"Cannot use " + strategy + " strategy because " + reason);
}
break;
case NONE: case CONTENTS:
break; // NONE and CONTENTS can always be used.
}
protected Projector<T,R> _cset;
protected DepotTypes _types;
// if we're not using a cache, then skip all of the above
strategy = ctx.isUsingCache() ? strategy : CacheStrategy.NONE;
switch (strategy) {
case SHORT_KEYS: case LONG_KEYS: case RECORDS:
return new WithCache<T>(ctx, type, clauses, strategy);
default:
return new Explicitly<T>(ctx, type, clauses, strategy == CacheStrategy.CONTENTS);
}
}
// from Fetcher
@@ -457,6 +460,62 @@ public abstract class FindAllQuery<T extends PersistentRecord,R>
return builder.append(")").toString();
}
/** A base class for queries that fetch a full record at a time. */
protected static abstract class FullRecordQuery<T extends PersistentRecord>
extends FindAllQuery<T, T>
{
protected FullRecordQuery (PersistenceContext ctx, Class<T> type) {
super(type, ctx.getMarshaller(type), new CloningCloner<T>());
_dmarsh = ctx.getMarshaller(type);
}
protected DepotMarshaller<T> _dmarsh;
}
/** Helper for {@link Projection}. */
protected static class ProjectionQueryMarshaller<T extends PersistentRecord,R>
implements QueryMarshaller<T,R>
{
public ProjectionQueryMarshaller (Projector<T,R> cset, DepotTypes types) {
_cset = cset;
_types = types;
}
public String getTableName () {
return _types.getTableName(_cset.ptype);
}
public SQLExpression<?>[] getSelections () {
return _cset.selexps;
}
public Key<T> getPrimaryKey (Object object) {
return _types.getMarshaller(_cset.ptype).getPrimaryKey(object);
}
public R createObject (ResultSet rs) throws SQLException {
Object[] data = new Object[_cset.selexps.length];
for (int ii = 0; ii < data.length; ii++) {
SQLExpression<?> exp = _cset.selexps[ii];
if (exp instanceof ColumnExp<?>) {
ColumnExp<?> col = (ColumnExp<?>)exp;
data[ii] = _types.getMarshaller(col.getPersistentClass()).
getFieldMarshaller(col.name).getFromSet(rs, ii+1);
} else {
// TEMP: in the case of selecting computed expressions, we rely on the types to
// be correct by construction; TODO: this will probably bite us when JDBC
// drivers choose Long instead of Integer or whatnot, so we'll need to set up
// more complex machinery
data[ii] = rs.getObject(ii+1);
}
}
return _cset.createObject(data);
}
protected Projector<T,R> _cset;
protected DepotTypes _types;
}
// we have to factor out cloning data for cache storage so that we can support storing
// immutable data (like Integer and String) which need not and cannot be cloned, versus mutable
// data (like PersistentRecords and Tuples) which must be cloned
src/test/java/com/samskivert/depot/TestCacheAdapter.java
+52 -27
View File
@@ -6,6 +6,7 @@ package com.samskivert.depot;
import java.io.Serializable;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -20,6 +21,57 @@ import com.samskivert.util.Tuple;
*/
public class TestCacheAdapter implements CacheAdapter
{
// from interface CacheAdapter
public <T> CacheAdapter.CachedValue<T> lookup (String cacheId, Serializable key)
{
// System.err.println("GET " + key + ": " + _cache.containsKey(key));
@SuppressWarnings("unchecked")
CachedValue<T> value = (CachedValue<T>) _cache.get(
new Tuple<String, Serializable>(cacheId, key));
return value;
}
// from interface CacheAdapter
public <T> void store (CacheCategory category, String cacheId, Serializable key, T value)
{
// System.err.println("STORE " + key);
_cache.put(new Tuple<String, Serializable>(cacheId, key), new TestCachedValue<T>(value));
}
// from interface CacheAdapter
public void remove (String cacheId, Serializable key)
{
// System.err.println("REMOVE " + key);
_cache.remove(new Tuple<String, Serializable>(cacheId, key));
}
// from interface CacheAdapter
public <T> Iterable<Serializable> enumerate (String cacheId)
{
// in a real implementation this would be a lazily constructed iterable
List<Serializable> result = Lists.newArrayList();
for (Map.Entry<Tuple<String, Serializable>, CachedValue<?>> entry: _cache.entrySet()) {
if (entry.getKey().left.equals(cacheId)) {
result.add(entry.getKey().right);
}
}
return result;
}
// from interface CacheAdapter
public void clear (String cacheId, boolean localOnly)
{
synchronized (_cache) {
for (Iterator<Tuple<String, Serializable>> iter = _cache.keySet().iterator();
iter.hasNext(); ) {
Tuple<String, Serializable> key = iter.next();
if (key.left.equals(cacheId)) {
iter.remove();
}
}
}
}
// from interface CacheAdapter
public void shutdown ()
{
@@ -37,33 +89,6 @@ public class TestCacheAdapter implements CacheAdapter
protected final T _value;
}
public <T> CacheAdapter.CachedValue<T> lookup (String cacheId, Serializable key) {
// System.err.println("GET " + key + ": " + _cache.containsKey(key));
@SuppressWarnings("unchecked")
CachedValue<T> value = (CachedValue<T>) _cache.get(
new Tuple<String, Serializable>(cacheId, key));
return value;
}
public <T> void store (CacheCategory category, String cacheId, Serializable key, T value) {
// System.err.println("STORE " + key);
_cache.put(new Tuple<String, Serializable>(cacheId, key), new TestCachedValue<T>(value));
}
public void remove (String cacheId, Serializable key) {
// System.err.println("REMOVE " + key);
_cache.remove(new Tuple<String, Serializable>(cacheId, key));
}
public <T> Iterable<Serializable> enumerate (String cacheId)
{
// in a real implementation this would be a lazily constructed iterable
List<Serializable> result = Lists.newArrayList();
for (Map.Entry<Tuple<String, Serializable>, CachedValue<?>> entry: _cache.entrySet()) {
if (entry.getKey().left.equals(cacheId)) {
result.add(entry.getKey().right);
}
}
return result;
}
protected Map<Tuple<String, Serializable>, CachedValue<?>> _cache =
Collections.synchronizedMap(
Maps.<Tuple<String, Serializable>, CachedValue<?>>newHashMap());