tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

added locking and logic to avoid iteration order issue in map

pull/12266/head
IrynaMatveieva 2 years ago
parent
befd4cc9c6
commit
e787b805da
8 changed files with 87 additions and 70 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 68
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java
  2. 5
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java
  3. 49
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
  4. 8
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
  5. 4
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java
  6. 14
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
  7. 8
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java
  8. 1
      common/proto/src/main/proto/queue.proto

68
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java
View File

@ -92,6 +92,7 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@ -119,6 +120,8 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
private ListeningExecutorService calculatedFieldExecutor;
private ListeningExecutorService calculatedFieldCallbackExecutor;
private final ConcurrentMap<EntityId, ReentrantLock> entityLocks = new ConcurrentHashMap<>();
private final ConcurrentMap<CalculatedFieldEntityCtxId, CalculatedFieldEntityCtx> states = new ConcurrentHashMap<>();
private static final int MAX_LAST_RECORDS_VALUE = 1024;
@ -536,40 +539,47 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
CalculatedFieldId cfId = calculatedFieldCtx.getCfId();
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, cfId);
if (tpi.isMyPartition()) {
CalculatedFieldEntityCtxId entityCtxId = new CalculatedFieldEntityCtxId(cfId.getId(), entityId.getId());
CalculatedFieldEntityCtx calculatedFieldEntityCtx = states.computeIfAbsent(entityCtxId, ctxId -> fetchCalculatedFieldEntityState(ctxId, calculatedFieldCtx.getCfType()));
Consumer<CalculatedFieldState> performUpdateState = (state) -> {
if (state.updateState(argumentValues)) {
calculatedFieldEntityCtx.setState(state);
states.put(entityCtxId, calculatedFieldEntityCtx);
rocksDBService.put(JacksonUtil.writeValueAsString(entityCtxId), JacksonUtil.writeValueAsString(calculatedFieldEntityCtx));
Map<String, ArgumentEntry> arguments = state.getArguments();
boolean allArgsPresent = arguments.keySet().containsAll(calculatedFieldCtx.getArguments().keySet()) &&
!arguments.containsValue(SingleValueArgumentEntry.EMPTY) && !arguments.containsValue(TsRollingArgumentEntry.EMPTY);
if (allArgsPresent) {
performCalculation(calculatedFieldCtx, state, entityId, calculatedFieldIds);
ReentrantLock lock = entityLocks.computeIfAbsent(entityId, id -> new ReentrantLock());
lock.lock();
try {
CalculatedFieldEntityCtxId entityCtxId = new CalculatedFieldEntityCtxId(cfId.getId(), entityId.getId());
CalculatedFieldEntityCtx calculatedFieldEntityCtx = states.computeIfAbsent(entityCtxId, ctxId -> fetchCalculatedFieldEntityState(ctxId, calculatedFieldCtx.getCfType()));
Consumer<CalculatedFieldState> performUpdateState = (state) -> {
if (state.updateState(argumentValues)) {
calculatedFieldEntityCtx.setState(state);
states.put(entityCtxId, calculatedFieldEntityCtx);
rocksDBService.put(JacksonUtil.writeValueAsString(entityCtxId), JacksonUtil.writeValueAsString(calculatedFieldEntityCtx));
Map<String, ArgumentEntry> arguments = state.getArguments();
boolean allArgsPresent = arguments.keySet().containsAll(calculatedFieldCtx.getArguments().keySet()) &&
!arguments.containsValue(SingleValueArgumentEntry.EMPTY) && !arguments.containsValue(TsRollingArgumentEntry.EMPTY);
if (allArgsPresent) {
performCalculation(calculatedFieldCtx, state, entityId, calculatedFieldIds);
}
}
}
};
};
CalculatedFieldState state = calculatedFieldEntityCtx.getState();
CalculatedFieldState state = calculatedFieldEntityCtx.getState();
boolean allKeysPresent = argumentValues.keySet().containsAll(calculatedFieldCtx.getArguments().keySet());
boolean requiresTsRollingUpdate = calculatedFieldCtx.getArguments().values().stream()
.anyMatch(argument -> "TS_ROLLING".equals(argument.getType()) && state.getArguments().get(argument.getKey()) == null);
boolean allKeysPresent = argumentValues.keySet().containsAll(calculatedFieldCtx.getArguments().keySet());
boolean requiresTsRollingUpdate = calculatedFieldCtx.getArguments().values().stream()
.anyMatch(argument -> "TS_ROLLING".equals(argument.getType()) && state.getArguments().get(argument.getKey()) == null);
if (!allKeysPresent || requiresTsRollingUpdate) {
if (!allKeysPresent || requiresTsRollingUpdate) {
Map<String, Argument> missingArguments = calculatedFieldCtx.getArguments().entrySet().stream()
.filter(entry -> !argumentValues.containsKey(entry.getKey()) || ("TS_ROLLING".equals(entry.getValue().getType()) && state.getArguments().get(entry.getKey()) == null))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
Map<String, Argument> missingArguments = calculatedFieldCtx.getArguments().entrySet().stream()
.filter(entry -> !argumentValues.containsKey(entry.getKey()) || ("TS_ROLLING".equals(entry.getValue().getType()) && state.getArguments().get(entry.getKey()) == null))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
fetchArguments(calculatedFieldCtx.getTenantId(), entityId, missingArguments, argumentValues::putAll)
.addListener(() -> performUpdateState.accept(state),
calculatedFieldCallbackExecutor);
} else {
performUpdateState.accept(state);
fetchArguments(calculatedFieldCtx.getTenantId(), entityId, missingArguments, argumentValues::putAll)
.addListener(() -> performUpdateState.accept(state),
calculatedFieldCallbackExecutor);
} else {
performUpdateState.accept(state);
}
} finally {
lock.unlock();
}
} else {
sendUpdateCalculatedFieldStateMsg(tenantId, cfId, entityId, calculatedFieldIds, argumentValues);
@ -747,7 +757,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
return tsRollingArgumentEntry;
} else if (entryProto.hasSingleValue()) {
TransportProtos.SingleValueProto singleValueProto = entryProto.getSingleValue();
return new SingleValueArgumentEntry(singleValueProto.getTs(), fromObjectProto(singleValueProto.getValue()));
return new SingleValueArgumentEntry(singleValueProto.getTs(), fromObjectProto(singleValueProto.getValue()), singleValueProto.getVersion());
} else {
throw new IllegalArgumentException("Unsupported ArgumentEntryProto type");
}

5
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java
View File

@ -22,8 +22,6 @@ import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import java.util.List;
import java.util.TreeMap;
import java.util.stream.Collectors;
@JsonTypeInfo(
use = JsonTypeInfo.Id.NAME,
@ -48,8 +46,7 @@ public interface ArgumentEntry {
}
static ArgumentEntry createTsRollingArgument(List<TsKvEntry> kvEntries) {
return new TsRollingArgumentEntry(kvEntries.stream().
collect(Collectors.toMap(TsKvEntry::getTs, TsKvEntry::getValue, (oldValue, newValue) -> newValue, TreeMap::new)));
return new TsRollingArgumentEntry(kvEntries);
}
@JsonIgnore

49
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
View File

@ -17,7 +17,6 @@ package org.thingsboard.server.service.cf.ctx.state;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
@ -37,34 +36,30 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
if (arguments == null) {
arguments = new HashMap<>();
}
AtomicBoolean stateUpdated = new AtomicBoolean(false);
argumentValues.forEach((key, argumentEntry) -> {
ArgumentEntry existingArgumentEntry = arguments.get(key);
if (existingArgumentEntry != null) {
if (existingArgumentEntry instanceof SingleValueArgumentEntry) {
if (existingArgumentEntry.hasUpdatedValue(argumentEntry)) {
arguments.put(key, argumentEntry.copy());
stateUpdated.set(true);
}
} else if (existingArgumentEntry instanceof TsRollingArgumentEntry existingTsRollingArgumentEntry) {
if (argumentEntry instanceof TsRollingArgumentEntry tsRollingArgumentEntry) {
if (existingArgumentEntry.hasUpdatedValue(argumentEntry)) {
existingTsRollingArgumentEntry.addAllTsRecords(tsRollingArgumentEntry.getTsRecords());
stateUpdated.set(true);
}
} else if (argumentEntry instanceof SingleValueArgumentEntry singleValueArgumentEntry) {
if (existingArgumentEntry.hasUpdatedValue(argumentEntry)) {
existingTsRollingArgumentEntry.addTsRecord(singleValueArgumentEntry.getTs(), singleValueArgumentEntry.getValue());
stateUpdated.set(true);
}
}
boolean stateUpdated = false;
for (Map.Entry<String, ArgumentEntry> entry : argumentValues.entrySet()) {
String key = entry.getKey();
ArgumentEntry newEntry = entry.getValue();
ArgumentEntry existingEntry = arguments.get(key);
if (existingEntry == null || existingEntry.hasUpdatedValue(newEntry)) {
if (existingEntry instanceof TsRollingArgumentEntry existingTsRollingEntry && newEntry instanceof TsRollingArgumentEntry newTsRollingEntry) {
existingTsRollingEntry.addAllTsRecords(newTsRollingEntry.getTsRecords());
} else if (existingEntry instanceof TsRollingArgumentEntry existingTsRollingEntry && newEntry instanceof SingleValueArgumentEntry singleValueEntry) {
existingTsRollingEntry.addTsRecord(singleValueEntry.getTs(), singleValueEntry.getValue());
} else if (existingEntry instanceof SingleValueArgumentEntry existingSingleValueEntry && newEntry instanceof SingleValueArgumentEntry singleValueEntry
&& singleValueEntry.getVersion() > existingSingleValueEntry.getVersion()) {
arguments.put(key, newEntry.copy());
} else {
arguments.put(key, newEntry.copy());
}
} else {
arguments.put(key, argumentEntry.copy());
stateUpdated.set(true);
stateUpdated = true;
}
});
return stateUpdated.get();
}
return stateUpdated;
}
}

8
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
View File

@ -26,6 +26,8 @@ import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@Data
@ -36,6 +38,7 @@ public class CalculatedFieldCtx {
private EntityId entityId;
private CalculatedFieldType cfType;
private final Map<String, Argument> arguments;
private final List<String> argKeys;
private Output output;
private String expression;
private TbelInvokeService tbelInvokeService;
@ -48,10 +51,11 @@ public class CalculatedFieldCtx {
this.cfType = calculatedField.getType();
CalculatedFieldConfiguration configuration = calculatedField.getConfiguration();
this.arguments = configuration.getArguments();
this.argKeys = new ArrayList<>(arguments.keySet());
this.output = configuration.getOutput();
this.expression = configuration.getExpression();
this.tbelInvokeService = tbelInvokeService;
if (!CalculatedFieldType.SIMPLE.equals(calculatedField.getType())) {
if (CalculatedFieldType.SCRIPT.equals(calculatedField.getType())) {
this.calculatedFieldScriptEngine = initEngine(tenantId, expression, tbelInvokeService);
}
}
@ -65,7 +69,7 @@ public class CalculatedFieldCtx {
tenantId,
tbelInvokeService,
expression,
arguments.keySet().toArray(new String[0])
argKeys.toArray(String[]::new)
);
}

4
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java
View File

@ -49,7 +49,9 @@ public class ScriptCalculatedFieldState extends BaseCalculatedFieldState {
tsRecords.entrySet().removeIf(tsRecord -> tsRecord.getKey() < System.currentTimeMillis() - argument.getTimeWindow());
}
});
Object[] args = arguments.values().stream().map(ArgumentEntry::getValue).toArray();
Object[] args = ctx.getArgKeys().stream()
.map(key -> arguments.get(key).getValue())
.toArray();
ListenableFuture<Map<String, Object>> resultFuture = ctx.getCalculatedFieldScriptEngine().executeToMapAsync(args);
Output output = ctx.getOutput();
return Futures.transform(resultFuture,

14
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
View File

@ -32,11 +32,15 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
private long ts;
private Object value;
private long version;
public SingleValueArgumentEntry(KvEntry entry) {
if (entry instanceof TsKvEntry) {
this.ts = ((TsKvEntry) entry).getTs();
} else if (entry instanceof AttributeKvEntry) {
this.ts = ((AttributeKvEntry) entry).getLastUpdateTs();
if (entry instanceof TsKvEntry tsKvEntry) {
this.ts = tsKvEntry.getTs();
this.version = tsKvEntry.getVersion();
} else if (entry instanceof AttributeKvEntry attributeKvEntry) {
this.ts = attributeKvEntry.getLastUpdateTs();
this.version = attributeKvEntry.getVersion();
}
this.value = entry.getValue();
}
@ -66,7 +70,7 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
@Override
public ArgumentEntry copy() {
return new SingleValueArgumentEntry(this.ts, this.value);
return new SingleValueArgumentEntry(this.ts, this.value, this.version);
}
}

8
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java
View File

@ -16,16 +16,20 @@
package org.thingsboard.server.service.cf.ctx.state;
import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.math.NumberUtils;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
@Data
@NoArgsConstructor
@AllArgsConstructor
@Slf4j
public class TsRollingArgumentEntry implements ArgumentEntry {
@ -35,8 +39,8 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
private TreeMap<Long, Object> tsRecords = new TreeMap<>();
public TsRollingArgumentEntry(TreeMap<Long, Object> tsRecords) {
addAllTsRecords(tsRecords);
public TsRollingArgumentEntry(List<TsKvEntry> kvEntries) {
kvEntries.forEach(tsKvEntry -> addTsRecord(tsKvEntry.getTs(), tsKvEntry.getValue()));
}
/**

1
common/proto/src/main/proto/queue.proto
View File

@ -841,6 +841,7 @@ message TsRollingProto {
message SingleValueProto {
int64 ts = 1;
ObjectProto value = 2;
int64 version = 3;
}
message ObjectProto {

Write Preview
Loading…
Cancel
Save