Skip to content

Commit 646e5f5

Browse files
fjtiradofjtirado
authored
[Fix #1723] Adding LRU Cache (#1724)
* [Fix #1723] Adding LRU Cache Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com> * [Fix #1723] Adding pin mechanism to LRU cache Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com> * [Fix #1723] Human to the rescue Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com> * [Fix #1723] Additional review comments Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com> * [Fix #7123] Optimizing key generation for MD5HashItem Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com> * [Fix #1723] New LRUCache implementation based on LinkedHashMap Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com> * [Fix #1723] Review comments Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com> --------- Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>
1 parent a269eee commit 646e5f5

10 files changed

Lines changed: 2390 additions & 36 deletions

File tree

‎impl/persistence/api/pom.xml‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,5 +12,16 @@
1212
<groupId>io.serverlessworkflow</groupId>
1313
<artifactId>serverlessworkflow-impl-core</artifactId>
1414
</dependency>
15+
<dependency>
16+
<groupId>org.junit.jupiter</groupId>
17+
<artifactId>junit-jupiter-engine</artifactId>
18+
<scope>test</scope>
19+
</dependency>
20+
<dependency>
21+
<groupId>org.assertj</groupId>
22+
<artifactId>assertj-core</artifactId>
23+
<scope>test</scope>
24+
</dependency>
1525
</dependencies>
26+
1627
</project>

‎impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashFactory.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,12 +18,27 @@
1818
import com.github.f4b6a3.ulid.Ulid;
1919
import com.github.f4b6a3.ulid.UlidFactory;
2020
import io.serverlessworkflow.impl.marshaller.WorkflowInputBuffer;
21+
import io.serverlessworkflow.impl.persistence.hashing.HashMappingCoordinator.BytesWithFlag;
22+
import java.util.List;
23+
import java.util.Map;
2124
import java.util.Optional;
25+
import java.util.function.Consumer;
26+
import java.util.function.Function;
2227

2328
public class DefaultHashFactory implements HashFactory {
2429

2530
private final UlidFactory idFactory = UlidFactory.newMonotonicInstance();
2631

32+
private final LRUCache<String, Map<String, Map<HashIndex, BytesWithFlag>>> instancesMap;
33+
34+
public DefaultHashFactory() {
35+
this(100);
36+
}
37+
38+
public DefaultHashFactory(int maxEntries) {
39+
this.instancesMap = new LRUCache<>(maxEntries);
40+
}
41+
2742
@Override
2843
public Optional<HashItem> fromBuffer(byte id, WorkflowInputBuffer buffer) {
2944
return Optional.ofNullable(
@@ -68,4 +83,11 @@ public HashIndex indexFromString(String str) {
6883
public HashIndex newIndex() {
6984
return new DefaultHashIndex(idFactory.create());
7085
}
86+
87+
@Override
88+
public HashMappingCoordinator mapCoordinator(
89+
Function<String, Map<String, Map<HashIndex, byte[]>>> retriever,
90+
Consumer<Map<String, List<HashMappingInfo>>> writer) {
91+
return new HashMappingCoordinator(instancesMap, this::newIndex, retriever, writer);
92+
}
7193
}

‎impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashFactory.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,18 @@
1616
package io.serverlessworkflow.impl.persistence.hashing;
1717

1818
import io.serverlessworkflow.impl.marshaller.WorkflowInputBuffer;
19+
import java.util.List;
20+
import java.util.Map;
1921
import java.util.Optional;
22+
import java.util.function.Consumer;
23+
import java.util.function.Function;
2024

2125
public interface HashFactory {
2226

27+
HashMappingCoordinator mapCoordinator(
28+
Function<String, Map<String, Map<HashIndex, byte[]>>> retriever,
29+
Consumer<Map<String, List<HashMappingInfo>>> writer);
30+
2331
Optional<HashItem> fromData(byte[] data);
2432

2533
Optional<HashItem> fromBuffer(byte id, WorkflowInputBuffer buffer);

‎impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java‎

Lines changed: 35 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -22,24 +22,14 @@
2222
import java.util.Map;
2323
import java.util.Map.Entry;
2424
import java.util.Optional;
25-
import java.util.concurrent.ConcurrentHashMap;
2625
import java.util.function.Consumer;
2726
import java.util.function.Function;
27+
import java.util.function.Supplier;
2828
import java.util.stream.Collectors;
2929

3030
public class HashMappingCoordinator {
3131

32-
public static final HashMappingCoordinator build(
33-
HashFactory hashFactory,
34-
Function<String, Map<String, Map<HashIndex, byte[]>>> retriever,
35-
Consumer<Map<String, List<HashMappingInfo>>> writer) {
36-
return new HashMappingCoordinator(hashFactory, retriever, writer);
37-
}
38-
39-
private static Map<String, Map<String, Map<HashIndex, BytesWithFlag>>> mappingInfo =
40-
new ConcurrentHashMap<>();
41-
42-
private static class BytesWithFlag {
32+
static class BytesWithFlag {
4333
private final byte[] bytes;
4434
private boolean persisted;
4535

@@ -64,16 +54,22 @@ public boolean isTransient() {
6454

6555
private record PendingWrite(String key, HashIndex index, BytesWithFlag bytes) {}
6656

67-
private final HashFactory hashFactory;
57+
private Map<String, Map<String, Map<HashIndex, BytesWithFlag>>> instanceMaps = new HashMap<>();
58+
59+
private final Supplier<HashIndex> hashIndexSupplier;
60+
private final LRUCache<String, Map<String, Map<HashIndex, BytesWithFlag>>> mappingInfo;
61+
6862
private final Function<String, Map<String, Map<HashIndex, byte[]>>> retriever;
6963
private final Consumer<Map<String, List<HashMappingInfo>>> writer;
7064

71-
private HashMappingCoordinator(
72-
HashFactory hashFactory,
65+
protected HashMappingCoordinator(
66+
LRUCache<String, Map<String, Map<HashIndex, BytesWithFlag>>> mappingInfo,
67+
Supplier<HashIndex> hashIndexSupplier,
7368
Function<String, Map<String, Map<HashIndex, byte[]>>> retriever,
7469
Consumer<Map<String, List<HashMappingInfo>>> writer) {
75-
this.hashFactory = hashFactory;
7670
this.retriever = retriever;
71+
this.mappingInfo = mappingInfo;
72+
this.hashIndexSupplier = hashIndexSupplier;
7773
this.writer = writer;
7874
}
7975

@@ -84,7 +80,7 @@ private HashMappingInfo from(PendingWrite pending) {
8480
}
8581

8682
public HashIndex calculateIndex(String instanceId, HashItem item, byte[] bytes) {
87-
Map<String, Map<HashIndex, BytesWithFlag>> instanceMap = getInstanceMap(instanceId);
83+
Map<String, Map<HashIndex, BytesWithFlag>> instanceMap = getCachedInstanceMap(instanceId);
8884
String key = item.key();
8985
synchronized (instanceMap) {
9086
Map<HashIndex, BytesWithFlag> duplicateMap =
@@ -98,7 +94,7 @@ public HashIndex calculateIndex(String instanceId, HashItem item, byte[] bytes)
9894
return entry.getKey();
9995
}
10096
}
101-
HashIndex index = hashFactory.newIndex();
97+
HashIndex index = hashIndexSupplier.get();
10298
BytesWithFlag bytesWithFlag = new BytesWithFlag(bytes);
10399
duplicateMap.put(index, bytesWithFlag);
104100
addWrite(instanceId, key, index, bytesWithFlag);
@@ -114,13 +110,21 @@ private void addWrite(String instanceId, String key, HashIndex index, BytesWithF
114110

115111
public Optional<byte[]> readBytes(String instanceId, HashItem item, HashIndex index) {
116112
Map<String, Map<HashIndex, BytesWithFlag>> instanceMap = getInstanceMap(instanceId);
117-
synchronized (instanceMap) {
118-
return Optional.ofNullable(instanceMap.get(item.key()))
119-
.map(m -> m.get(index))
120-
.map(bytes -> bytes.bytes);
113+
try {
114+
synchronized (instanceMap) {
115+
return Optional.ofNullable(instanceMap.get(item.key()))
116+
.map(m -> m.get(index))
117+
.map(bytes -> bytes.bytes);
118+
}
119+
} finally {
120+
mappingInfo.unpin(instanceId);
121121
}
122122
}
123123

124+
private Map<String, Map<HashIndex, BytesWithFlag>> getCachedInstanceMap(String instanceId) {
125+
return instanceMaps.computeIfAbsent(instanceId, k -> getInstanceMap(instanceId));
126+
}
127+
124128
private Map<String, Map<HashIndex, BytesWithFlag>> getInstanceMap(String instanceId) {
125129
return mappingInfo.computeIfAbsent(
126130
instanceId,
@@ -169,18 +173,22 @@ public void afterCommit() {
169173
}
170174
}
171175
}
172-
pending.clear();
176+
cleanUp();
173177
}
174178

175179
public void afterRollback() {
180+
cleanUp();
181+
}
182+
183+
private void cleanUp() {
184+
instanceMaps.keySet().forEach(mappingInfo::unpin);
185+
instanceMaps.clear();
176186
pending.clear();
177187
}
178188

179189
public void afterRemove(String instanceId) {
190+
pending.remove(instanceId);
191+
instanceMaps.remove(instanceId);
180192
mappingInfo.remove(instanceId);
181193
}
182-
183-
public static void clearAll() {
184-
mappingInfo.clear();
185-
}
186194
}

0 commit comments

Comments
 (0)