Skip to content

Commit f3bc27d

Browse files
fjtiradofjtirado
committed
[Fix #1691] MapInMemory approach
Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>
1 parent 037b79a commit f3bc27d

6 files changed

Lines changed: 303 additions & 88 deletions

File tree

‎impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceUtils.java‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
/*
2+
* Copyright 2020-Present The Serverless Workflow Specification Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.serverlessworkflow.impl.persistence;
17+
18+
import java.util.ArrayList;
19+
import java.util.List;
20+
import java.util.Map;
21+
import java.util.Map.Entry;
22+
import java.util.stream.Collectors;
23+
24+
public class PersistenceUtils {
25+
26+
private PersistenceUtils() {}
27+
28+
public static <T> Map<String, List<T>> mapMapToMapList(Map<String, Map<Integer, T>> map) {
29+
return map.entrySet().stream()
30+
.collect(Collectors.toMap(Entry::getKey, e -> PersistenceUtils.mapToList(e.getValue())));
31+
}
32+
33+
public static <T> List<T> mapToList(Map<Integer, T> map) {
34+
List<T> list = new ArrayList<>(map.size());
35+
for (int i = 0; i < map.size(); i++) {
36+
list.add(map.get(i));
37+
}
38+
return list;
39+
}
40+
}

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

Lines changed: 182 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,182 @@
1+
/*
2+
* Copyright 2020-Present The Serverless Workflow Specification Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.serverlessworkflow.impl.persistence.hashing;
17+
18+
import java.util.ArrayList;
19+
import java.util.Arrays;
20+
import java.util.HashMap;
21+
import java.util.List;
22+
import java.util.Map;
23+
import java.util.Optional;
24+
import java.util.concurrent.ConcurrentHashMap;
25+
import java.util.function.Consumer;
26+
import java.util.function.Function;
27+
import java.util.stream.Collectors;
28+
29+
public class HashMappingCoordinator {
30+
31+
public static final HashMappingCoordinator build(
32+
Function<String, Map<String, List<byte[]>>> retriever,
33+
Consumer<Map<String, List<HashMappingInfo>>> writer) {
34+
return new HashMappingCoordinator(retriever, writer);
35+
}
36+
37+
private static class BytesWithFlag {
38+
private final byte[] bytes;
39+
private boolean persisted;
40+
41+
public BytesWithFlag(byte[] bytes) {
42+
this.bytes = bytes;
43+
this.persisted = false;
44+
}
45+
46+
public BytesWithFlag(byte[] bytes, boolean persisted) {
47+
this.bytes = bytes;
48+
this.persisted = persisted;
49+
}
50+
51+
public void persist() {
52+
persisted = true;
53+
}
54+
55+
public boolean isTransient() {
56+
return !persisted;
57+
}
58+
}
59+
60+
private record PendingWrite(String key, int index, BytesWithFlag bytes) {}
61+
62+
private final Function<String, Map<String, List<byte[]>>> retriever;
63+
private final Consumer<Map<String, List<HashMappingInfo>>> writer;
64+
65+
private HashMappingCoordinator(
66+
Function<String, Map<String, List<byte[]>>> retriever,
67+
Consumer<Map<String, List<HashMappingInfo>>> writer) {
68+
this.retriever = retriever;
69+
this.writer = writer;
70+
}
71+
72+
private static Map<String, Map<String, List<BytesWithFlag>>> mappingInfo =
73+
new ConcurrentHashMap<>();
74+
75+
private Map<String, List<PendingWrite>> pending = new HashMap<>();
76+
77+
private HashMappingInfo from(PendingWrite pending) {
78+
return new HashMappingInfo(pending.key, pending.index, pending.bytes.bytes);
79+
}
80+
81+
public int calculateIndex(String instanceId, HashItem item, byte[] bytes) {
82+
Map<String, List<BytesWithFlag>> instanceMap = getInstanceMap(instanceId);
83+
String key = item.key();
84+
synchronized (instanceMap) {
85+
List<BytesWithFlag> list = instanceMap.computeIfAbsent(key, __ -> new ArrayList<>());
86+
if (list.isEmpty()) {
87+
int index = 0;
88+
BytesWithFlag bytesWithFlag = new BytesWithFlag(bytes);
89+
list.add(bytesWithFlag);
90+
addWrite(instanceId, key, index, bytesWithFlag);
91+
return index;
92+
}
93+
for (int i = 0; i < list.size(); i++) {
94+
BytesWithFlag bytesWithFlag = list.get(i);
95+
if (Arrays.equals(bytesWithFlag.bytes, bytes)) {
96+
if (bytesWithFlag.isTransient()) {
97+
addWrite(instanceId, key, i, bytesWithFlag);
98+
}
99+
return i;
100+
}
101+
}
102+
BytesWithFlag bytesWithFlag = new BytesWithFlag(bytes);
103+
list.add(bytesWithFlag);
104+
int index = list.size() - 1;
105+
addWrite(instanceId, key, index, bytesWithFlag);
106+
return index;
107+
}
108+
}
109+
110+
private void addWrite(String instanceId, String key, int index, BytesWithFlag bytes) {
111+
pending
112+
.computeIfAbsent(instanceId, __ -> new ArrayList<>())
113+
.add(new PendingWrite(key, index, bytes));
114+
}
115+
116+
public Optional<byte[]> readBytes(String instanceId, HashItem item, int index) {
117+
Map<String, List<BytesWithFlag>> instanceMap = getInstanceMap(instanceId);
118+
synchronized (instanceMap) {
119+
return Optional.ofNullable(instanceMap.get(item.key()))
120+
.filter(list -> list.size() > index)
121+
.map(list -> list.get(index))
122+
.map(bytes -> bytes.bytes);
123+
}
124+
}
125+
126+
private Map<String, List<BytesWithFlag>> getInstanceMap(String instanceId) {
127+
return mappingInfo.computeIfAbsent(
128+
instanceId,
129+
k ->
130+
retriever.apply(instanceId).entrySet().stream()
131+
.collect(
132+
Collectors.toMap(
133+
Map.Entry::getKey,
134+
e ->
135+
e.getValue().stream()
136+
.map(b -> new BytesWithFlag(b, true))
137+
.collect(Collectors.toCollection(ArrayList::new)))));
138+
}
139+
140+
public void persist() {
141+
Map<String, List<HashMappingInfo>> result = new HashMap<>();
142+
for (Map.Entry<String, List<PendingWrite>> item : pending.entrySet()) {
143+
String instanceId = item.getKey();
144+
List<HashMappingInfo> list = new ArrayList<>();
145+
result.put(instanceId, list);
146+
Map<String, List<BytesWithFlag>> instanceMap = mappingInfo.get(instanceId);
147+
if (instanceMap != null) {
148+
synchronized (instanceMap) {
149+
item.getValue()
150+
.forEach(
151+
v -> {
152+
if (v.bytes.isTransient()) {
153+
list.add(from(v));
154+
}
155+
});
156+
}
157+
}
158+
}
159+
writer.accept(result);
160+
}
161+
162+
public void afterCommit() {
163+
for (Map.Entry<String, List<PendingWrite>> item : pending.entrySet()) {
164+
String instanceId = item.getKey();
165+
Map<String, List<BytesWithFlag>> instanceMap = mappingInfo.get(instanceId);
166+
if (instanceMap != null) {
167+
synchronized (instanceMap) {
168+
item.getValue().forEach(v -> v.bytes.persist());
169+
}
170+
}
171+
}
172+
pending.clear();
173+
}
174+
175+
public void afterRollback() {
176+
pending.clear();
177+
}
178+
179+
public void afterRemove(String instanceId) {
180+
mappingInfo.remove(instanceId);
181+
}
182+
}

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

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
/*
2+
* Copyright 2020-Present The Serverless Workflow Specification Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.serverlessworkflow.impl.persistence.hashing;
17+
18+
public record HashMappingInfo(String key, int index, byte[] bytes) {}

‎impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BytesMapInstanceTransaction.java‎

Lines changed: 42 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -33,20 +33,21 @@
3333
import io.serverlessworkflow.impl.persistence.CompletedTaskInfo;
3434
import io.serverlessworkflow.impl.persistence.PersistenceInstanceInfo;
3535
import io.serverlessworkflow.impl.persistence.PersistenceTaskInfo;
36+
import io.serverlessworkflow.impl.persistence.PersistenceUtils;
3637
import io.serverlessworkflow.impl.persistence.RetriedTaskInfo;
3738
import io.serverlessworkflow.impl.persistence.hashing.HashFactory;
3839
import io.serverlessworkflow.impl.persistence.hashing.HashItem;
40+
import io.serverlessworkflow.impl.persistence.hashing.HashMappingCoordinator;
41+
import io.serverlessworkflow.impl.persistence.hashing.HashMappingInfo;
3942
import java.io.ByteArrayInputStream;
4043
import java.io.ByteArrayOutputStream;
4144
import java.io.IOException;
4245
import java.io.UncheckedIOException;
4346
import java.net.URI;
4447
import java.time.OffsetDateTime;
45-
import java.util.Arrays;
4648
import java.util.HashMap;
49+
import java.util.List;
4750
import java.util.Map;
48-
import java.util.concurrent.Callable;
49-
import java.util.concurrent.locks.Lock;
5051

5152
public abstract class BytesMapInstanceTransaction
5253
extends BigMapInstanceTransaction<byte[], byte[], byte[], byte[], byte[], byte[]> {
@@ -56,16 +57,37 @@ public abstract class BytesMapInstanceTransaction
5657
private static final byte VERSION_2 = 2;
5758
private static final byte VERSION_3 = 3;
5859
private static final byte[] PROCESSED_VALUE = {1};
60+
private static final String SEPARATOR = ":";
5961

6062
private final WorkflowBufferFactory bufferFactory;
6163
private final HashFactory hashFactory;
62-
private final Lock hashLock;
64+
protected final HashMappingCoordinator hashCoordinator;
6365

64-
protected BytesMapInstanceTransaction(
65-
WorkflowBufferFactory factory, HashFactory hashFactory, Lock hashLock) {
66+
protected BytesMapInstanceTransaction(WorkflowBufferFactory factory, HashFactory hashFactory) {
6667
this.bufferFactory = factory;
6768
this.hashFactory = hashFactory;
68-
this.hashLock = hashLock;
69+
this.hashCoordinator =
70+
HashMappingCoordinator.build(this::retrieveBlobData, this::writeBlobData);
71+
}
72+
73+
private Map<String, List<byte[]>> retrieveBlobData(String instanceId) {
74+
Map<String, Map<Integer, byte[]>> result = new HashMap<>();
75+
for (Map.Entry<String, byte[]> entry : blobData(instanceId).entrySet()) {
76+
String[] splitted = entry.getKey().split(SEPARATOR);
77+
result
78+
.computeIfAbsent(splitted[0], __ -> new HashMap<>())
79+
.put(Integer.parseInt(splitted[1]), entry.getValue());
80+
}
81+
return PersistenceUtils.mapMapToMapList(result);
82+
}
83+
84+
private void writeBlobData(Map<String, List<HashMappingInfo>> writeInfo) {
85+
for (Map.Entry<String, List<HashMappingInfo>> entry : writeInfo.entrySet()) {
86+
Map<String, byte[]> blobData = blobData(entry.getKey());
87+
for (HashMappingInfo info : entry.getValue()) {
88+
blobData.put(info.key() + SEPARATOR + info.index(), info.bytes());
89+
}
90+
}
6991
}
7092

7193
protected abstract Map<String, byte[]> blobData(String instanceId);
@@ -75,7 +97,9 @@ protected BytesMapInstanceTransaction(
7597
@Override
7698
public void removeProcessInstance(WorkflowContextData workflowContext) {
7799
super.removeProcessInstance(workflowContext);
78-
removeBlobData(workflowContext.instanceData().id());
100+
String instanceId = workflowContext.instanceData().id();
101+
removeBlobData(instanceId);
102+
hashCoordinator.afterRemove(instanceId);
79103
}
80104

81105
@Override
@@ -254,47 +278,25 @@ private PersistenceTaskInfo readVersion0(WorkflowInputBuffer buffer) {
254278
buffer.readBoolean() ? buffer.readString() : null);
255279
}
256280

257-
protected abstract <T> T executeNewTransaction(Callable<T> runnable);
258-
259281
private void writeLargeObject(
260282
HashItem item, WorkflowInstanceData instanceData, WorkflowOutputBuffer writer, byte[] bytes) {
261-
262-
boolean duplicated;
263-
String key = item.key();
264-
hashLock.lock();
265-
try {
266-
duplicated =
267-
executeNewTransaction(
268-
() -> {
269-
Map<String, byte[]> map = blobData(instanceData.id());
270-
byte[] value = map.get(key);
271-
if (value == null) {
272-
map.put(key, bytes);
273-
return false;
274-
} else {
275-
return !Arrays.equals(value, bytes);
276-
}
277-
});
278-
} finally {
279-
hashLock.unlock();
280-
}
281-
282-
if (duplicated) {
283-
legacyWriteObject(writer, bytes);
284-
} else {
285-
writer.writeByte(item.id());
286-
item.writeKey(writer);
287-
}
283+
int index = hashCoordinator.calculateIndex(instanceData.id(), item, bytes);
284+
writer.writeByte(item.id());
285+
item.writeKey(writer);
286+
writer.writeShort((short) index);
288287
}
289288

290289
private Object readLargeObject(String instanceId, WorkflowInputBuffer buffer) {
291290
return hashFactory
292291
.fromBuffer(buffer.readByte(), buffer)
293292
.map(
294293
item -> {
295-
Map<String, byte[]> map = blobData(instanceId);
296-
ByteArrayInputStream bytes = new ByteArrayInputStream(map.get(item.key()));
297-
try (WorkflowInputBuffer input = bufferFactory.input(bytes)) {
294+
try (WorkflowInputBuffer input =
295+
bufferFactory.input(
296+
new ByteArrayInputStream(
297+
hashCoordinator
298+
.readBytes(instanceId, item, buffer.readShort())
299+
.orElseThrow()))) {
298300
return input.readObject();
299301
}
300302
})

0 commit comments

Comments
 (0)