ComputeIfAbsent 在jdk8下的死锁场景

## 一、ConcurrentHashMap的死锁场景

先看一段代码，这里主要是使用了computeIfAbsent进行更新

```
 		Map<String, Integer> map = new ConcurrentHashMap<>();
        map.computeIfAbsent(
                "AaAa",
                key -> {
                    log.info("update another node. but AaAa.hashcode = BBBB.hashcode");
                    return map.computeIfAbsent(
                            "BBBB",
                            key2 -> 42);
                }
        );
```

这里的期望是： 将map中，BBBB的值更新成42， 随后将AaAa的值更新成BBBB的值

但实际运行就会发现，这段代码产生了死锁。

## 二、 computeIfAbsent干了什么?

第一次更新：将关键的路径摘取如下

```java
	// 不断请求更新table
	for (Node<K,V>[] tab = table;;) {
		// ...省略判断是第一次插入的一些操作,i是要插入的数组下标
		// 节点不存在，插入预占节点
		Node<K,V> r = new ReservationNode<K,V>();
        synchronized (r) {
        	// cas 将 null 更新成预占的 r 节点
            if (casTabAt(tab, i, null, r)) {
                binCount = 1;
                Node<K,V> node = null;
                try {
                	// 执行接口传入的方法获取值，构建node节点
                    if ((val = mappingFunction.apply(key)) != null)
                        node = new Node<K,V>(h, key, val, null);
                } finally {
                	// 无论结果是什么，直接用 node 替换预占的 r 
                    setTabAt(tab, i, node);
                }
            }
        }
        // 更新完成，跳出
        if (binCount != 0)
            break;
	}
```

这里逻辑在插入场景是正常的，但computeIfAbsent的注释里提到了：`Some attempted update operations on this map by other threads may be blocked while computation is in progress, so the computation should be short and simple, and must not attempt to update any other mappings of this map.` 当计算函数被调用时，其他线程对此Map的某些更新操作可能会被阻止，因此计算函数应该简短而简单，并且不得尝试更新此Map的其他key。

为什么会这样呢？在计算函数中更新key具体怎么触发的死锁呢？继续抽象代码（这里的处理，computer、computeIfAbsent、computeIfPresent都是相似的）

```java
	// 不断请求更新table
	for (Node<K,V>[] tab = table;;) {
		// ...省略判断是更新的一些操作，f=待更新节点，fh=f.hash，i 是要插入的数组下标
		// 这里注意，fh不是hashcode，而是构造时的hash值: -1=重定向节点MOVED；-2=树的根节点TREEBIN；-3=临时预占节点RESERVED
		boolean added = false;
        synchronized (f) {
        	// 由于key的hashcode一致，所以更新下标 i 一致，找到了插入场景的预占节点 r
            if (tabAt(tab, i) == f) {
                // 如果是正常的节点，就在这里更新了，但现在是预占节点，hash=-3，无法进入更新break的逻辑
                if (fh >= 0) {
                    binCount = 1;
                    for (Node<K,V> e = f;; ++binCount) {
                        // ...省略校验自己更新自己，若是自更新直接返回并break;
                        // ...省略一些更新操作, 通过接口传入的计算函数得到值，更新并break
                    }
                }
                // 判断是否是TreeBin，显然也不是
                else if (f instanceof TreeBin) {
                    binCount = 2;
                    // ...省略一些更新操作， 更新并break
                }
            }
        }
        // 没有更新成功，那么继续不断重试，死锁启动！
        if (binCount != 0) {
            if (binCount >= TREEIFY_THRESHOLD)
                treeifyBin(tab, i);
            if (!added)
                return val;
            break;
        }
	}
```

## 三、怎么解决死锁？

ConcurrentHashMap作者Doug Lea，在jdk8的computeIfAbsent注释中，提及了计算函数中更新节点有问题，那为什么在JDK8中没有解决呢？又怎么在JDK9中解决的呢？

JDK8 未修复，是因为一开始，作者认为很难解决，无法通用地检测使用者遇到了死锁场景。但作者随后自己收回了这个话，并在JDK9修复了它 https://bugs.openjdk.org/browse/JDK-8062841。

先看JDK9的官方解决方案

```java
		if (fh >= 0) {
           // ...省略原更新逻辑
        }
        else if (f instanceof TreeBin) {
            // ...省略原更新逻辑
        }
        else if (f instanceof ReservationNode)
        	// 新增对于预占节点，直接抛出异常
            throw new IllegalStateException("Recursive update");		
```

一点思考题：为什么不改成`if(fh >= 0 || f instanceof ReservationNode)`以此允许更新操作呢？

>  这里如果将预占节点直接赋值为key=BBBB的有效节点，计算函数返回更新成功的值，这里是正常的。

>  但在第一次插入场景，由于拿到了正常值，程序又无法感知预占节点已经被有效节点替换了，就有再构建一个key=AaAa的node的有效节点，覆盖插入，BBBB的节点信息就丢失，造成线程不安全。

## 四、一个生产事故的样例

正常来说，像样例一的使用方式，我们是较少使用，也很快通过review发现，但是被参杂到Caffeine缓存框架下，就很难检测和避免事故的发生了。

来看一下生产环境Spring+Caffeine是怎么应用到这个bug的

1. 按SpringCache规范提供一个Cache的实例并注册成bean

```java
@Bean
public CaffeineCache userCache(CacheLoader<Object,Object> hexinCacheLoader){
    Cache<Object, Object> cache = Caffeine.newBuilder()
             .maximumSize(800)
             .refreshAfterWrite(10, TimeUnit.SECONDS)
             .expireAfterWrite(20, TimeUnit.SECONDS)
             .recordStats()
             .build(hexinCacheLoader);
    return new CaffeineCache("user", cache);
}
```

这个实际上是一个有最大值的有界缓存，通过工厂类 LocalCacheFactory::newBoundedLocalCache，构造得到 BoundedLocalCache的实例 SSSMSWR

```java
// 源码中层层嵌套继承，这里提取关键处理步骤并铺平
final class SSSMSWR<K, V>  extends  BoundedLocalCache<K, V> {
	// 核心数据结构
	final ConcurrentHashMap<Object, Node<K, V>> data;
    
	// refreshload 处理刷新的地方
  	@Nullable final CacheLoader<K, V> cacheLoader;
  	
    // ...省略一些 代理data进行增删查改的处理
}
```

2. SpringCache收到这个实例后，在CacheAspectSupport处理缓存注解的逻辑如下

```java
private Object execute(final CacheOperationInvoker invoker, Method method, CacheOperationContexts contexts)	
    // ...省略一些对其他场景的处理，比如synchronized、CacheEvict等等
    
	// 默认直接从缓存中取，但这里caffeine继承了LoadingCache，会先执行一遍函数取值和compute
	Cache.ValueWrapper cacheHit = findCachedItem(contexts.get(CacheableOperation.class));

// 未命中时，构造put请求，注意此时put请求还没有执行
	List<CachePutRequest> cachePutRequests = new LinkedList<>();
	if (cacheHit == null) {
		collectPutRequests(contexts.get(CacheableOperation.class),
					CacheOperationExpressionEvaluator.NO_RESULT, cachePutRequests);
	}
	
	if (cacheHit != null && !hasCachePut(contexts)) {
		// 命中且不需要put，直接返回
		cacheValue = cacheHit.get();
		returnValue = wrapCacheValue(method, cacheValue);
	} else {
		// 其余情况，执行函数拿到返回值
		returnValue = invokeOperation(invoker);
		cacheValue = unwrapReturnValue(returnValue);
	}

// 执行put请求，更新缓存
	for (CachePutRequest cachePutRequest : cachePutRequests) {
		cachePutRequest.apply(cacheValue);
	}
	return returnValue;
}
```

3. 注意！这里的坑出现了！

如果配置中hexinCacheLoader同样继承自LoadingCache，那么在findCachedItem操作时，会优先使用LoadingCache::get。

这里的区别在于，默认是直接从缓存中查询是否命中，而LoadingCache是有副作用的，实际是SSSMSWR实例的computeIfAbsent。而不配置走默认的话，是不会死锁的

```java
public class CaffeineCache extends AbstractValueAdaptingCache {
    // 我们配置的cache
	private final com.github.benmanes.caffeine.cache.Cache<Object, Object> cache;
    
    @Override
	@Nullable
	public ValueWrapper get(Object key) {
		if (this.cache instanceof LoadingCache) {
            // SSSMSWR对应LocalLoadingCache
			Object value = ((LoadingCache<Object, Object>) this.cache).get(key);
			return toValueWrapper(value);
		}
        // 默认是直接取现在的缓存
		return super.get(key);
	}
    
}

interface LocalLoadingCache<K, V> extends LocalManualCache<K, V>, LoadingCache<K, V> {
	@Override
  	default @Nullable V get(K key) {
        // 你以为是单纯get？并不，他还把值给更新了
    	return cache().computeIfAbsent(key, mappingFunction());
  	}
}
```

4. 死锁demo

```
@RestController
public class CacheController {
    @Resource
    private CacheService cacheService;

@GetMapping("/")
    @Cacheable(cacheNames = "user", key = "'AaAa'")
    public String hello() {
        return cacheService.getTime();
    }
}

@Component
public class CacheService {

@Cacheable(cacheNames = "user", key = "'BBBB'")
    public String getTime() {
        return "42";
    }
}
```

如果是SpringCache默认逻辑，那么是先BBBB的节点 put更新缓存，然后再给AaAa的节点put更新缓存，不会死锁，ConcurrentHashMap会处理hash冲突。

如果是Caffeine 指定LoadingCache的逻辑，那么是 computer更新AaAa节点时，嵌套更新BBBB节点，就是文章开头遇到的hashcode一致时的死锁问题。

5. 死锁场景的触发和解决

死锁的触发需要三个条件： a：Caffeine 指定了LoadingCache；b：请求链路嵌套调用同一个Caffeine实例；c：请求链路上的缓存key出现hash冲突

在实际使用上，我们会使用自定义的keyGenerator，将spring代理类 + 方法 + 参数，共同生成一个hashcode，所以每次启动 这个hashcode都不一样，冲突概率很低，很难发现。