Support for multi-threaded cache access

Previously, if a `@Cacheable` method was accessed with the same key by
multiple threads, the underlying method was invoked several times instead
of blocking the threads while the value is computed. This scenario
typically affects users that enable caching to avoid calling a costly
method too often. When said method can be invoked by an arbitrary number
of clients on startup, caching has close to no effect.

This commit adds a new method on `Cache` that implements the read-through
pattern:

```
<T> T get(Object key, Callable<T> valueLoader);
```

If an entry for a given key is not found, the specified `Callable` is
invoked to "load" the value and cache it before returning it to the
caller. Because the entire operation is managed by the underlying cache
provider, it is much more easier to guarantee that the loader (e.g. the
annotated method) will be called only once in case of concurrent access.

A new `sync` attribute to the `@Cacheable` annotation has been addded.
When this flag is enabled, the caching abstraction invokes the new
`Cache` method define above. This new mode bring a set of limitations:

* It can't be combined with other cache operations
* Only one `@Cacheable` operation can be specified
* Only one cache is allowed
* `condition` and `unless` attribute are not supported

The rationale behind those limitations is that the underlying Cache is
taking care of the actual caching operation so we can't really apply
any SpEL or multiple caches handling there.

Issue: SPR-9254

spring-context/src/main/java/org/springframework/cache/Cache.java

@@ -16,6 +16,8 @@
 
 package org.springframework.cache;
 
+import java.util.concurrent.Callable;
+
 /**
  * Interface that defines common cache operations.
  *
@@ -73,6 +75,23 @@ public interface Cache {
 	 */
 	<T> T get(Object key, Class<T> type);
 
+	/**
+	 * Return the value to which this cache maps the specified key, obtaining
+	 * that value from {@code valueLoader} if necessary. This method provides
+	 * a simple substitute for the conventional "if cached, return; otherwise
+	 * create, cache and return" pattern.
+	 * <p>If possible, implementations should ensure that the loading operation
+	 * is synchronized so that the specified {@code valueLoader} is only called
+	 * once in case of concurrent access on the same key.
+	 * <p>If the {@code valueLoader} throws an exception, it is wrapped in
+	 * a {@link ValueRetrievalException}
+	 * @param key the key whose associated value is to be returned
+	 * @return the value to which this cache maps the specified key
+	 * @throws ValueRetrievalException if the {@code valueLoader} throws an exception
+	 * @since 4.3
+	 */
+	<T> T get(Object key, Callable<T> valueLoader);
+
 	/**
 	 * Associate the specified value with the specified key in this cache.
 	 * <p>If the cache previously contained a mapping for this key, the old
@@ -133,4 +152,24 @@ public interface Cache {
 		Object get();
 	}
 
+	/**
+	 * TODO
+	 */
+	@SuppressWarnings("serial")
+	class ValueRetrievalException extends RuntimeException {
+
+		private final Object key;
+
+		public ValueRetrievalException(Object key, Callable<?> loader, Throwable ex) {
+			super(String.format("Value for key '%s' could not " +
+					"be loaded using '%s'", key, loader), ex);
+			this.key = key;
+		}
+
+		public Object getKey() {
+			return key;
+		}
+
+	}
+
 }

spring-context/src/main/java/org/springframework/cache/annotation/Cacheable.java

@@ -22,6 +22,7 @@ import java.lang.annotation.Inherited;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
+import java.util.concurrent.Callable;
 
 import org.springframework.core.annotation.AliasFor;
 
@@ -154,4 +155,21 @@ public @interface Cacheable {
 	 */
 	String unless() default "";
 
+	/**
+	 * Synchronize the invocation of the underlying method if several threads are
+	 * attempting to load a value for the same key. The synchronization leads to
+	 * a couple of limitations:
+	 * <ol>
+	 * <li>{@link #unless()} is not supported</li>
+	 * <li>Only one cache may be specified</li>
+	 * <li>No other cache-related operation can be combined</li>
+	 * </ol>
+	 * This is effectively a hint and the actual cache provider that you are
+	 * using may not support it in a synchronized fashion. Check your provider
+	 * documentation for more details on the actual semantics.
+	 * @since 4.3
+	 * @see org.springframework.cache.Cache#get(Object, Callable)
+	 */
+	boolean sync() default false;
+
 }

spring-context/src/main/java/org/springframework/cache/annotation/SpringCacheAnnotationParser.java

@@ -107,6 +107,7 @@ public class SpringCacheAnnotationParser implements CacheAnnotationParser, Seria
 		op.setKeyGenerator(cacheable.keyGenerator());
 		op.setCacheManager(cacheable.cacheManager());
 		op.setCacheResolver(cacheable.cacheResolver());
+		op.setSync(cacheable.sync());
 		op.setName(ae.toString());
 
 		defaultConfig.applyDefault(op);

spring-context/src/main/java/org/springframework/cache/concurrent/ConcurrentMapCache.java

@@ -16,6 +16,7 @@
 
 package org.springframework.cache.concurrent;
 
+import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -96,6 +97,30 @@ public class ConcurrentMapCache extends AbstractValueAdaptingCache {
 		return this.store.get(key);
 	}
 
+	@SuppressWarnings("unchecked")
+	@Override
+	public <T> T get(Object key, Callable<T> valueLoader) {
+		if (this.store.containsKey(key)) {
+			return (T) get(key).get();
+		}
+		else {
+			synchronized (this.store) {
+				if (this.store.containsKey(key)) {
+					return (T) get(key).get();
+				}
+				T value;
+				try {
+					value = valueLoader.call();
+				}
+				catch (Exception ex) {
+					throw new ValueRetrievalException(key, valueLoader, ex);
+				}
+				put(key, value);
+				return value;
+			}
+		}
+	}
+
 	@Override
 	public void put(Object key, Object value) {
 		this.store.put(key, toStoreValue(value));

spring-context/src/main/java/org/springframework/cache/config/CacheAdviceParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -109,6 +109,7 @@ class CacheAdviceParser extends AbstractSingleBeanDefinitionParser {
 			nameHolder.setSource(parserContext.extractSource(opElement));
 			CacheableOperation op = prop.merge(opElement, parserContext.getReaderContext(), new CacheableOperation());
 			op.setUnless(getAttributeValue(opElement, "unless", ""));
+			op.setSync(Boolean.valueOf(getAttributeValue(opElement, "sync", "false")));
 
 			Collection<CacheOperation> col = cacheOpMap.get(nameHolder);
 			if (col == null) {

spring-context/src/main/java/org/springframework/cache/interceptor/CacheAspectSupport.java

@@ -23,6 +23,7 @@ import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.commons.logging.Log;
@@ -328,7 +329,34 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
 		return targetClass;
 	}
 
-	private Object execute(CacheOperationInvoker invoker, CacheOperationContexts contexts) {
+	private Object execute(final CacheOperationInvoker invoker, CacheOperationContexts contexts) {
+		// Special handling of synchronized invocation
+		if (contexts.isSynchronized()) {
+			CacheOperationContext context = contexts.get(CacheableOperation.class).iterator().next();
+			if (isConditionPassing(context, ExpressionEvaluator.NO_RESULT)) {
+				Object key = generateKey(context, ExpressionEvaluator.NO_RESULT);
+				Cache cache = context.getCaches().iterator().next();
+				try {
+					return cache.get(key, new Callable<Object>() {
+						@Override
+						public Object call() throws Exception {
+							return invokeOperation(invoker);
+						}
+					});
+				}
+				catch (Cache.ValueRetrievalException ex) {
+					// The invoker wraps any Throwable in a ThrowableWrapper instance so we
+					// can just make sure that one bubbles up the stack.
+					throw (CacheOperationInvoker.ThrowableWrapper) ex.getCause();
+				}
+			}
+			else {
+				// No caching required, only call the underlying method
+				return invokeOperation(invoker);
+			}
+		}
+
+
 		// Process any early evictions
 		processCacheEvicts(contexts.get(CacheEvictOperation.class), true, ExpressionEvaluator.NO_RESULT);
 
@@ -374,7 +402,7 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
 		for (CacheOperationContext context : cachePutContexts) {
 			try {
 				if (!context.isConditionPassing(ExpressionEvaluator.RESULT_UNAVAILABLE)) {
-	                excluded.add(context);
+					excluded.add(context);
 				}
 			}
 			catch (VariableNotAvailableException e) {
@@ -385,7 +413,6 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
 		// check if  all puts have been excluded by condition
 		return cachePutContexts.size() != excluded.size();
 
-
 	}
 
 	private void processCacheEvicts(Collection<CacheOperationContext> contexts, boolean beforeInvocation, Object result) {
@@ -504,18 +531,57 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
 		private final MultiValueMap<Class<? extends CacheOperation>, CacheOperationContext> contexts =
 				new LinkedMultiValueMap<Class<? extends CacheOperation>, CacheOperationContext>();
 
+		private final boolean sync;
+
 		public CacheOperationContexts(Collection<? extends CacheOperation> operations, Method method,
 				Object[] args, Object target, Class<?> targetClass) {
 
 			for (CacheOperation operation : operations) {
 				this.contexts.add(operation.getClass(), getOperationContext(operation, method, args, target, targetClass));
 			}
+			this.sync = determineSyncFlag(method);
 		}
 
 		public Collection<CacheOperationContext> get(Class<? extends CacheOperation> operationClass) {
 			Collection<CacheOperationContext> result = this.contexts.get(operationClass);
 			return (result != null ? result : Collections.<CacheOperationContext>emptyList());
 		}
+
+		public boolean isSynchronized() {
+			return this.sync;
+		}
+
+		private boolean determineSyncFlag(Method method) {
+			List<CacheOperationContext> cacheOperationContexts = this.contexts.get(CacheableOperation.class);
+			if (cacheOperationContexts == null) { // No @Cacheable operation
+				return false;
+			}
+			boolean syncEnabled = false;
+			for (CacheOperationContext cacheOperationContext : cacheOperationContexts) {
+				if (((CacheableOperation) cacheOperationContext.getOperation()).isSync()) {
+					syncEnabled = true;
+					break;
+				}
+			}
+			if (syncEnabled) {
+				if (this.contexts.size() > 1) {
+					throw new IllegalStateException("@Cacheable(sync = true) cannot be combined with other cache operations on '" + method + "'");
+				}
+				if (cacheOperationContexts.size() > 1) {
+					throw new IllegalStateException("Only one @Cacheable(sync = true) entry is allowed on '" + method + "'");
+				}
+				CacheOperationContext cacheOperationContext = cacheOperationContexts.iterator().next();
+				CacheableOperation operation = (CacheableOperation) cacheOperationContext.getOperation();
+				if (cacheOperationContext.getCaches().size() > 1) {
+					throw new IllegalStateException("@Cacheable(sync = true) only allows a single cache on '" + operation + "'");
+				}
+				if (StringUtils.hasText(operation.getUnless())) {
+					throw new IllegalStateException("@Cacheable(sync = true) does not support unless attribute on '" + operation + "'");
+				}
+				return true;
+			}
+			return false;
+		}
 	}
 
 

spring-context/src/main/java/org/springframework/cache/interceptor/CacheOperationInvoker.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -42,7 +42,7 @@ public interface CacheOperationInvoker {
 	 * Wrap any exception thrown while invoking {@link #invoke()}
 	 */
 	@SuppressWarnings("serial")
-	public static class ThrowableWrapper extends RuntimeException {
+	class ThrowableWrapper extends RuntimeException {
 
 		private final Throwable original;
 

spring-context/src/main/java/org/springframework/cache/interceptor/CacheableOperation.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,6 +27,8 @@ public class CacheableOperation extends CacheOperation {
 
 	private String unless;
 
+	private boolean sync;
+
 
 	public String getUnless() {
 		return unless;
@@ -36,12 +38,23 @@ public class CacheableOperation extends CacheOperation {
 		this.unless = unless;
 	}
 
+	public boolean isSync() {
+		return sync;
+	}
+
+	public void setSync(boolean sync) {
+		this.sync = sync;
+	}
+
 	@Override
 	protected StringBuilder getOperationDescription() {
 		StringBuilder sb = super.getOperationDescription();
 		sb.append(" | unless='");
 		sb.append(this.unless);
 		sb.append("'");
+		sb.append(" | sync='");
+		sb.append(this.sync);
+		sb.append("'");
 		return sb;
 	}
 }

spring-context/src/main/java/org/springframework/cache/support/NoOpCacheManager.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,6 +20,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.Set;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -99,6 +100,11 @@ public class NoOpCacheManager implements CacheManager {
 			return null;
 		}
 
+		@Override
+		public <T> T get(Object key, Callable<T> valueLoader) {
+			return null;
+		}
+
 		@Override
 		public String getName() {
 			return this.name;