Skip to content

Thread safety around LocalBucketing #63

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
May 18, 2023
Merged

Thread safety around LocalBucketing #63

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4c617ea
adding some thread safety around LocalBucketing
May 18, 2023
b1a2610
removed old, unused getVariable() method
May 18, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 12 additions & 35 deletions src/main/java/com/devcycle/sdk/server/local/bucketing/LocalBucketing.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,7 +203,7 @@ private byte[] readAssemblyScriptUint8Array(int address)
return bufferData;
}

public void storeConfig(String sdkKey, String config) {
public synchronized void storeConfig(String sdkKey, String config) {
unpinAll();
int sdkKeyAddress = getSDKKeyAddress(sdkKey);
int configAddress = newUint8ArrayParameter(config.getBytes(StandardCharsets.UTF_8));
Expand All @@ -213,15 +213,15 @@ public void storeConfig(String sdkKey, String config) {
fn.accept(sdkKeyAddress, configAddress);
}

public void setPlatformData(String platformData) {
public synchronized void setPlatformData(String platformData) {
unpinAll();
int platformDataAddress = newUint8ArrayParameter(platformData.getBytes(StandardCharsets.UTF_8));
Func setPlatformDataPtr = linker.get(store, "", "setPlatformDataUTF8").get().func();
WasmFunctions.Consumer1<Integer> fn = WasmFunctions.consumer(store, setPlatformDataPtr, I32);
fn.accept(platformDataAddress);
}

public void setClientCustomData(String sdkKey, String customData) {
public synchronized void setClientCustomData(String sdkKey, String customData) {
unpinAll();
int sdkKeyAddress = getSDKKeyAddress(sdkKey);
int customDataAddress = newUint8ArrayParameter(customData.getBytes(StandardCharsets.UTF_8));;
Expand All @@ -230,7 +230,7 @@ public void setClientCustomData(String sdkKey, String customData) {
fn.accept(sdkKeyAddress, customDataAddress);
}

public BucketedUserConfig generateBucketedConfig(String sdkKey, User user) throws JsonProcessingException {
public synchronized BucketedUserConfig generateBucketedConfig(String sdkKey, User user) throws JsonProcessingException {
unpinAll();
String userString = OBJECT_MAPPER.writeValueAsString(user);

Expand All @@ -249,30 +249,7 @@ public BucketedUserConfig generateBucketedConfig(String sdkKey, User user) throw
return config;
}

public String getVariable(String sdkKey, User user, String key, Variable.TypeEnum variableTypeEnum, boolean shouldTrackEvent) throws JsonProcessingException {
// need some kind of mutex?
String userString = OBJECT_MAPPER.writeValueAsString(user);

int wasmVariableType = this.variableTypeMap.get(variableTypeEnum);

unpinAll();
int sdkKeyAddress = getSDKKeyAddress(sdkKey);
int userAddress = newWasmString(userString);
int keyAddress = newWasmString(key);

Func getVariablePtr = linker.get(store, "", "variableForUser").get().func();
WasmFunctions.Function5<Integer, Integer, Integer, Integer, Integer, Integer> variableForUser = WasmFunctions.func(
store, getVariablePtr, I32, I32, I32, I32, I32, I32);

int resultAddress = variableForUser.call(sdkKeyAddress, userAddress, keyAddress, wasmVariableType, shouldTrackEvent ? 1 : 0);
if (resultAddress == 0){
return null;
}
String variableString = readWasmString(resultAddress);
return variableString;
}

public byte[] getVariableForUserProtobuf(byte[] serializedParams){
public synchronized byte[] getVariableForUserProtobuf(byte[] serializedParams){
int paramsAddr = newUint8ArrayParameter(serializedParams);

Func getVariablePtr = linker.get(store, "", "variableForUser_PB").get().func();
Expand All @@ -290,7 +267,7 @@ public byte[] getVariableForUserProtobuf(byte[] serializedParams){
return varBytes;
}

public void initEventQueue(String sdkKey, String options) {
public synchronized void initEventQueue(String sdkKey, String options) {
unpinAll();
int sdkKeyAddress = getSDKKeyAddress(sdkKey);
int optionsAddress = newWasmString(options);
Expand All @@ -300,7 +277,7 @@ public void initEventQueue(String sdkKey, String options) {
fn.accept(sdkKeyAddress, optionsAddress);
}

public void queueEvent(String sdkKey, String user, String event) {
public synchronized void queueEvent(String sdkKey, String user, String event) {
unpinAll();
int sdkKeyAddress = newWasmString(sdkKey);
int userAddress = getPinnedParameter(user);
Expand All @@ -311,7 +288,7 @@ public void queueEvent(String sdkKey, String user, String event) {
fn.accept(sdkKeyAddress, userAddress, eventAddress);
}

public void queueAggregateEvent(String sdkKey, String event, String variableVariationMap) {
public synchronized void queueAggregateEvent(String sdkKey, String event, String variableVariationMap) {
unpinAll();
int sdkKeyAddress = getSDKKeyAddress(sdkKey);
int eventAddress = getPinnedParameter(event);
Expand All @@ -322,7 +299,7 @@ public void queueAggregateEvent(String sdkKey, String event, String variableVari
fn.accept(sdkKeyAddress, eventAddress, variableVariationMapAddress);
}

public FlushPayload[] flushEventQueue(String sdkKey) throws JsonProcessingException {
public synchronized FlushPayload[] flushEventQueue(String sdkKey) throws JsonProcessingException {
unpinAll();
int sdkKeyAddress = getSDKKeyAddress(sdkKey);

Expand All @@ -345,7 +322,7 @@ public FlushPayload[] flushEventQueue(String sdkKey) throws JsonProcessingExcept
return payloads;
}

public void onPayloadFailure(String sdkKey, String payloadId, boolean retryable) {
public synchronized void onPayloadFailure(String sdkKey, String payloadId, boolean retryable) {
unpinAll();
int sdkKeyAddress = getSDKKeyAddress(sdkKey);
int payloadIdAddress = newWasmString(payloadId);
Expand All @@ -355,7 +332,7 @@ public void onPayloadFailure(String sdkKey, String payloadId, boolean retryable)
fn.accept(sdkKeyAddress, payloadIdAddress, retryable ? 1 : 0);
}

public void onPayloadSuccess(String sdkKey, String payloadId) {
public synchronized void onPayloadSuccess(String sdkKey, String payloadId) {
unpinAll();
int sdkKeyAddress = getSDKKeyAddress(sdkKey);
int payloadIdAddress = newWasmString(payloadId);
Expand All @@ -365,7 +342,7 @@ public void onPayloadSuccess(String sdkKey, String payloadId) {
fn.accept(sdkKeyAddress, payloadIdAddress);
}

public int getEventQueueSize(String sdkKey) {
public synchronized int getEventQueueSize(String sdkKey) {
unpinAll();
int sdkKeyAddress = getSDKKeyAddress(sdkKey);

Expand Down