Skip to content

Commit

Permalink
[LFUCache]Add frequency of key and delete the empty cache queue (apac…
Browse files Browse the repository at this point in the history 
…he#7967)

* 1.Change the freqTable from Array to TreeMap
2.Add a timeout field to determine if an empty queue can be deleted

* 1.Change the freqTable from Array to TreeMap
2.Add a timeout field to determine if an empty queue can be deleted
3.Add a method to get frequency of the key

* 1.Change the freqTable from Array to TreeMap
2.Add a timeout field to determine if an empty queue can be deleted
3.Add a method to get frequency of the key

* remove unused imports

Co-authored-by: liwenliang <liwenliang@weidian.com>
  • Loading branch information
@iiweniiang
iiweniiang and liwenliang committed Jun 7, 2021
1 parent bfcf003 commit e8581fd
Show file tree
Hide file tree
Showing 2 changed files with 209 additions and 35 deletions.
214 changes: 179 additions & 35 deletions dubbo-common/src/main/java/org/apache/dubbo/common/utils/LFUCache.java
View file
Expand Up @@ -16,26 +16,35 @@
*/
package org.apache.dubbo.common.utils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class LFUCache<K, V> {

private Map<K, CacheNode<K, V>> map;
private CacheDeque<K, V>[] freqTable;
private Map<Long, CacheDeque<K, V>> freqTable;

private final int capacity;
private int evictionCount;
private int curSize = 0;
private long removeFreqEntryTimeout;

private final ReentrantLock lock = new ReentrantLock();
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
private static final int DEFAULT_INITIAL_CAPACITY = 1000;

private static final float DEFAULT_EVICTION_FACTOR = 0.75f;

private static final long DEFAULT_REMOVE_FREQ_TABLE_TIME_OUT = 1800000L;

public LFUCache() {
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_EVICTION_FACTOR);
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_EVICTION_FACTOR, DEFAULT_REMOVE_FREQ_TABLE_TIME_OUT);
}

/**
Expand All @@ -60,14 +69,36 @@ public LFUCache(final int maxCapacity, final float evictionFactor) {
this.capacity = maxCapacity;
this.evictionCount = (int) (capacity * evictionFactor);
this.map = new HashMap<>();
this.freqTable = new CacheDeque[capacity + 1];
for (int i = 0; i <= capacity; i++) {
freqTable[i] = new CacheDeque<>();
this.freqTable = new TreeMap<>(Long::compareTo);
freqTable.put(1L, new CacheDeque<>());
}

/**
* Constructs and initializes cache with specified capacity and eviction
* factor. Unacceptable parameter values followed with
* {@link IllegalArgumentException}.
*
* @param maxCapacity cache max capacity
* @param evictionFactor cache proceedEviction factor
* @param removeFreqEntryTimeout cache queue remove timeout
*/
@SuppressWarnings("unchecked")
public LFUCache(final int maxCapacity, final float evictionFactor, final long removeFreqEntryTimeout) {
if (maxCapacity <= 0) {
throw new IllegalArgumentException("Illegal initial capacity: " +
maxCapacity);
}
for (int i = 0; i < capacity; i++) {
freqTable[i].nextDeque = freqTable[i + 1];
boolean factorInRange = evictionFactor <= 1 && evictionFactor > 0;
if (!factorInRange || Float.isNaN(evictionFactor)) {
throw new IllegalArgumentException("Illegal eviction factor value:"
+ evictionFactor);
}
freqTable[capacity].nextDeque = freqTable[capacity];
this.capacity = maxCapacity;
this.evictionCount = (int) (capacity * evictionFactor);
this.removeFreqEntryTimeout = removeFreqEntryTimeout;
this.map = new HashMap<>();
this.freqTable = new TreeMap<>(Long::compareTo);
freqTable.put(1L, new CacheDeque<>());
}

public int getCapacity() {
Expand All @@ -76,31 +107,31 @@ public int getCapacity() {

public V put(final K key, final V value) {
CacheNode<K, V> node;
lock.lock();
lock.writeLock().lock();
try {
node = map.get(key);
if (node != null) {
CacheNode.withdrawNode(node);
node.value = value;
freqTable[0].addLastNode(node);
moveToNextFreqQueue(node.incrFreq(), node);
map.put(key, node);
} else {
node = freqTable[0].addLast(key, value);
map.put(key, node);
curSize++;
if (curSize > capacity) {
if (curSize + 1 > capacity) {
proceedEviction();
}
node = freqTable.get(1L).addLast(key, value);
map.put(key, node);
curSize++;
}
} finally {
lock.unlock();
lock.writeLock().unlock();
}
return node.value;
}

public V remove(final K key) {
CacheNode<K, V> node = null;
lock.lock();
lock.writeLock().lock();
try {
if (map.containsKey(key)) {
node = map.remove(key);
Expand All @@ -110,51 +141,137 @@ public V remove(final K key) {
curSize--;
}
} finally {
lock.unlock();
lock.writeLock().unlock();
}
return (node != null) ? node.value : null;
}

public V get(final K key) {
CacheNode<K, V> node = null;
lock.lock();
lock.writeLock().lock();
try {
if (map.containsKey(key)) {
node = map.get(key);
CacheNode.withdrawNode(node);
node.owner.nextDeque.addLastNode(node);
moveToNextFreqQueue(node.incrFreq(), node);
}
} finally {
lock.unlock();
lock.writeLock().unlock();
}
return (node != null) ? node.value : null;
}

/**
* Returns size of the freq table
*
* @return size
*/
public int getFreqTableSize(){
return freqTable.size();
}

/**
* Returns freq of the element
*
* @return freq
*/
public Long getFreq(final K key) {
CacheNode<K, V> node = null;
lock.readLock().lock();
try {
if (map.containsKey(key)) {
node = map.get(key);
return node.getFreq();
}
} finally {
lock.readLock().unlock();
}
return null;
}

/**
* Returns node list of this frequency
*
* @return node list
*/
private List<CacheNode<K,V>> getFreqList(final Long freq){
if(freq == null){
return null;
}
lock.writeLock().lock();
try {
if (freqTable.containsKey(freq)) {
if(freqTable.get(freq).nodeMap.size() > 0){
return new ArrayList<>(freqTable.get(freq).nodeMap.values());
}
}
} finally {
lock.writeLock().unlock();
}
return null;
}

/**
* Returns node list's size of this frequency
*
* @return node list's size
*/
public int getFreqListSize(final Long freq){
if(freq == null){
return 0;
}
lock.writeLock().lock();
try {
if (freqTable.containsKey(freq)) {
return freqTable.get(freq).size.get();
}
} finally {
lock.writeLock().unlock();
}
return 0;
}

/**
* Evicts less frequently used elements corresponding to eviction factor,
* specified at instantiation step.
*
* @return number of evicted elements
*/
private int proceedEviction() {
int targetSize = capacity - evictionCount;
int targetSize = capacity - evictionCount - 1;
int evictedElements = 0;

FREQ_TABLE_ITER_LOOP:
for (int i = 0; i <= capacity; i++) {
Set<Long> freqKeys = freqTable.keySet();
boolean evictionEnd = false;
for (Long freq : freqKeys) {
CacheDeque<K, V> q = freqTable.get(freq);
CacheNode<K, V> node;
while (!freqTable[i].isEmpty()) {
node = freqTable[i].pollFirst();
remove(node.key);
if (targetSize >= curSize) {
break FREQ_TABLE_ITER_LOOP;
if(!evictionEnd) {
while (!q.isEmpty()) {
node = q.pollFirst();
remove(node.key);
evictedElements++;
if (targetSize >= curSize) {
evictionEnd = true;
break;
}
}
evictedElements++;
}
// If the queue is empty for a long time, delete the queue
if (removeFreqEntryTimeout > 0 && freq > 1 && q.isEmpty() && (System.currentTimeMillis() - q.getLastReqTime()) >= removeFreqEntryTimeout) {
freqTable.remove(freq);
}
}
return evictedElements;
}

/**
* Move the node to the next cache queue
*/
private void moveToNextFreqQueue(long newFreq, CacheNode<K, V> node){
freqTable.putIfAbsent(newFreq, new CacheDeque<>());
freqTable.get(newFreq).addLastNode(node);
}

/**
* Returns cache current size.
*
Expand All @@ -170,6 +287,7 @@ static class CacheNode<K, V> {
CacheNode<K, V> next;
K key;
V value;
volatile AtomicLong freq = new AtomicLong(1);
CacheDeque<K, V> owner;

CacheNode() {
Expand All @@ -180,6 +298,14 @@ static class CacheNode<K, V> {
this.value = value;
}

long incrFreq(){
return freq.incrementAndGet();
}

long getFreq(){
return freq.get();
}

/**
* This method takes specified node and reattaches it neighbors nodes
* links to each other, so specified node will no longer tied with them.
Expand All @@ -196,6 +322,8 @@ static <K, V> CacheNode<K, V> withdrawNode(
node.prev.next = node.next;
if (node.next != null) {
node.next.prev = node.prev;
node.owner.nodeMap.remove(node.key);
node.owner.size.decrementAndGet();
}
}
return node;
Expand All @@ -216,8 +344,9 @@ static class CacheDeque<K, V> {

CacheNode<K, V> last;
CacheNode<K, V> first;
CacheDeque<K, V> nextDeque;

Map<K, CacheNode<K, V>> nodeMap;
long lastReqTime;
volatile AtomicInteger size = new AtomicInteger(0);
/**
* Constructs list and initializes last and first pointers.
*/
Expand All @@ -226,6 +355,7 @@ static class CacheDeque<K, V> {
first = new CacheNode<>();
last.next = first;
first.prev = last;
nodeMap = new HashMap<>();
}

/**
Expand All @@ -243,6 +373,8 @@ CacheNode<K, V> addLast(final K key, final V value) {
node.prev = last;
node.next.prev = node;
last.next = node;
this.setLastReqTime(System.currentTimeMillis());
this.size.incrementAndGet();
return node;
}

Expand All @@ -252,6 +384,9 @@ CacheNode<K, V> addLastNode(final CacheNode<K, V> node) {
node.prev = last;
node.next.prev = node;
last.next = node;
this.setLastReqTime(System.currentTimeMillis());
this.nodeMap.put(node.key, node);
this.size.incrementAndGet();
return node;
}

Expand All @@ -268,6 +403,8 @@ CacheNode<K, V> pollFirst() {
first.prev.next = first;
node.prev = null;
node.next = null;
this.nodeMap.remove(node.key);
this.size.decrementAndGet();
}
return node;
}
Expand All @@ -281,6 +418,13 @@ boolean isEmpty() {
return last.next == first;
}

}
public CacheDeque<K, V> setLastReqTime(long lastReqTime) {
this.lastReqTime = lastReqTime;
return this;
}

public long getLastReqTime() {
return lastReqTime;
}
}
}

0 comments on commit e8581fd

Please sign in to comment.