This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new ad89a887ad fix(wasm): isolate concurrent executions with instance pool
(#7582)
ad89a887ad is described below
commit ad89a887ad25fbfee6bc050432395324ab3ffd18
Author: eye-gu <[email protected]>
AuthorDate: Thu Oct 15 14:07:11 2026 +0800
fix(wasm): isolate concurrent executions with instance pool (#7582)
* fix(wasm): isolate concurrent executions with instance pool
Fixes #7558
* fix(wasm): isolate concurrent executions with instance pool
* fix(wasm): fix checkstyle, harden nesting, add echo fixture
* fix(wasm): address review: cap default pool size at 8, lease via
BlockingQueue, isolate test state
* fix review
* fix
* fix(wasm): address review: fail fast on nested export calls during
construction, use Math.toIntExact in echo test
* fix(wasm): address review: bounded lease wait with diagnostics, fail-fast
getMemory, suppress release failures
---
.github/scripts/verify-wasm-fixtures.py | 33 +-
.../shenyu/plugin/wasm/api/loader/WasmLoader.java | 271 ++++++++++++++--
.../plugin/wasm/api/WasmLoaderConcurrentTest.java | 341 +++++++++++++++++++++
...sm.api.WasmLoaderConcurrentTest$EchoPlugin.wasm | Bin 0 -> 124 bytes
.../src/test/wasm-concurrent/Makefile | 23 ++
.../src/test/wasm-concurrent/echo.wat | 27 ++
6 files changed, 649 insertions(+), 46 deletions(-)
diff --git a/.github/scripts/verify-wasm-fixtures.py
b/.github/scripts/verify-wasm-fixtures.py
index 9e8cd729a5..7238e0decb 100644
--- a/.github/scripts/verify-wasm-fixtures.py
+++ b/.github/scripts/verify-wasm-fixtures.py
@@ -43,21 +43,26 @@ def verify_wat():
version = subprocess.check_output(["wat2wasm", "--version"],
text=True).strip()
if version != "1.0.42":
raise RuntimeError("WAT verification requires WABT 1.0.42, found " +
version)
+ families = [
+ (BASE, ["wasm-bridge", "wasm-gateway"]),
+ (API, ["wasm-concurrent"]),
+ ]
count = 0
- for family in ["wasm-bridge", "wasm-gateway"]:
- directory = BASE / family
- if not directory.exists():
- continue
- with tempfile.TemporaryDirectory(prefix="shenyu-wat-") as temporary:
- run(["make", "-C", str(directory), "build", "RESOURCES=" +
temporary])
- binaries = list(Path(temporary).glob("*.wasm"))
- if not binaries:
- raise RuntimeError("No fixtures generated for " + family)
- for rebuilt in binaries:
- committed = BASE / "resources" / rebuilt.name
- if rebuilt.read_bytes() != committed.read_bytes():
- raise RuntimeError("WAT fixture differs from committed
binary: " + rebuilt.name)
- count += 1
+ for source_root, names in families:
+ for family in names:
+ directory = source_root / family
+ if not directory.exists():
+ continue
+ with tempfile.TemporaryDirectory(prefix="shenyu-wat-") as
temporary:
+ run(["make", "-C", str(directory), "build", "RESOURCES=" +
temporary])
+ binaries = list(Path(temporary).glob("*.wasm"))
+ if not binaries:
+ raise RuntimeError("No fixtures generated for " + family)
+ for rebuilt in binaries:
+ committed = source_root / "resources" / rebuilt.name
+ if rebuilt.read_bytes() != committed.read_bytes():
+ raise RuntimeError("WAT fixture differs from committed
binary: " + rebuilt.name)
+ count += 1
if not count:
raise RuntimeError("No WAT fixture families found")
print("Verified " + str(count) + " byte-identical WAT fixtures",
flush=True)
diff --git
a/shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java
b/shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java
index 14fcf64b07..2871ba67f8 100644
---
a/shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java
+++
b/shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java
@@ -20,7 +20,9 @@ package org.apache.shenyu.plugin.wasm.api.loader;
import com.dylibso.chicory.compiler.MachineFactoryCompiler;
import com.dylibso.chicory.runtime.ExportFunction;
import com.dylibso.chicory.runtime.HostFunction;
+import com.dylibso.chicory.runtime.ImportValues;
import com.dylibso.chicory.runtime.Instance;
+import com.dylibso.chicory.runtime.Machine;
import com.dylibso.chicory.runtime.Memory;
import com.dylibso.chicory.runtime.Store;
import com.dylibso.chicory.wasm.Parser;
@@ -34,11 +36,17 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.URL;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
+import java.util.function.Function;
/**
* The WasmLoader aims to load wasm file and provide the wasm function to java,
@@ -46,6 +54,12 @@ import java.util.function.Consumer;
*
* <p>Powered by <a href="https://chicory.dev/";>Chicory</a> — a pure-Java
WebAssembly
* runtime with zero native dependencies.
+ *
+ * <p>Concurrency model: a Chicory {@link Instance} is not thread-safe
(interpreter
+ * stacks, guest heap, linear memory), so every exported call leases one
pooled instance
+ * exclusively, waiting at most {@link #leaseTimeoutMillis()} for a free one;
host
+ * callbacks run inside the lease and {@link #getMemory()} resolves to the
instance
+ * executing on the calling thread, failing fast outside a call.
*/
public class WasmLoader implements AutoCloseable {
@@ -53,20 +67,41 @@ public class WasmLoader implements AutoCloseable {
private static final String IMPORT_WASM_MODULE_NAME = "shenyu";
+ // Upper bound on parallel executions; capped so worst-case memory (grown
guest
+ // heaps persist per instance) stays bounded.
+ private static final int DEFAULT_POOL_SIZE =
Math.min(Runtime.getRuntime().availableProcessors(), 8);
+
+ // Bounded wait for a pooled instance: pool exhaustion (or a nested-call
circular
+ // wait) fails the request after this timeout instead of hanging it
forever.
+ private static final long DEFAULT_LEASE_TIMEOUT_MILLIS = 30_000L;
+
+ // Instances currently executing on this thread (a stack: nested calls
push/pop);
+ // getMemory() in host callbacks resolves against the top entry.
+ private final ThreadLocal<ArrayDeque<Instance>> bindings =
ThreadLocal.withInitial(ArrayDeque::new);
+
private final AtomicBoolean closed = new AtomicBoolean(false);
-
+
+ // becomes true after the instance pool is fully populated
+ private volatile boolean ready;
+
private final String wasmName;
private final Store store;
- private Instance instance;
+ // first created instance; probes export existence for getWasmExtern (all
pooled
+ // instances share one module, so any of them answers for all)
+ private final Instance primary;
+
+ private final BlockingQueue<Instance> freeInstances;
+
+ private final int poolSize;
public WasmLoader() {
this(null, null);
}
-
+
/**
- * Constructs a WasmLoader, locates the {@code .wasm} resource, and
instantiates the module.
+ * Constructs a WasmLoader with the default pool size (at most 8
instances).
*
* <p>Initialization order:
* <ol>
@@ -81,6 +116,19 @@ public class WasmLoader implements AutoCloseable {
* host functions and {@link
#initWasmCallJavaFunc(Store)} are skipped
*/
public WasmLoader(final Class<?> wasmClass, final Consumer<Store>
initializer) {
+ this(wasmClass, initializer, DEFAULT_POOL_SIZE);
+ }
+
+ /**
+ * Constructs a WasmLoader with an explicit instance pool size.
+ *
+ * @param wasmClass the class whose name determines the .wasm resource
path; if null, uses
+ * {@code this.getClass()} (inheritance mode)
+ * @param initializer optional consumer to register host functions; when
non-null the built-in
+ * host functions and {@link
#initWasmCallJavaFunc(Store)} are skipped
+ * @param poolSize the number of WASM instances to instantiate (at
least 1)
+ */
+ public WasmLoader(final Class<?> wasmClass, final Consumer<Store>
initializer, final int poolSize) {
final Class<?> clazz = Objects.nonNull(wasmClass) ? wasmClass :
this.getClass();
this.wasmName = clazz.getName() + ".wasm";
try {
@@ -117,44 +165,91 @@ public class WasmLoader implements AutoCloseable {
}
WasmModule module = Parser.parse(wasmBytes);
- // Try runtime compiler (WASM → JVM bytecode) for better execution
speed.
- // Falls back gracefully to interpreter if something goes wrong.
+ final ImportValues importValues = store.toImportValues();
+ // Try the runtime compiler (factory shared by all pooled
instances);
+ // fall back to the interpreter if compilation or first
instantiation fails.
+ Function<Instance, Machine> machineFactory;
+ Instance firstInstance;
try {
- this.instance = Instance.builder(module)
- .withImportValues(store.toImportValues())
-
.withMachineFactory(MachineFactoryCompiler.compile(module))
- .withStart(false)
- .build();
- LOG.debug("Using runtime compiler for {}", wasmName);
+ machineFactory = compileMachineFactory(module);
+ if (Objects.isNull(machineFactory)) {
+ firstInstance = createInstance(module, importValues, null);
+ LOG.debug("Using interpreter for {}", wasmName);
+ } else {
+ firstInstance = createInstance(module, importValues,
machineFactory);
+ LOG.debug("Using runtime compiler for {}", wasmName);
+ }
} catch (LinkageError | RuntimeException compilerError) {
LOG.warn("Runtime compiler unavailable, falling back to
interpreter for {}: {}",
wasmName, compilerError.getMessage());
- this.instance = Instance.builder(module)
- .withImportValues(store.toImportValues())
- .withStart(false)
- .build();
+ machineFactory = null;
+ firstInstance = createInstance(module, importValues, null);
}
- // Call _initialize if present (required by TinyGo -target
wasm-unknown).
- findWasmExtern("_initialize").ifPresent(initFn -> {
- initFn.apply();
- LOG.debug("Called _initialize for {}", wasmName);
- });
-
- if (Objects.isNull(instance.memory())) {
- throw new ShenyuWasmInitException("memory not available in
wasm file: " + wasmName);
+ final int size = Math.max(1, poolSize);
+ final List<Instance> instances = new ArrayList<>(size);
+ instances.add(firstInstance);
+ for (int i = 1; i < size; i++) {
+ instances.add(createInstance(module, importValues,
machineFactory));
+ }
+ this.poolSize = size;
+ this.primary = firstInstance;
+ this.freeInstances = new ArrayBlockingQueue<>(size);
+ // guest initialization runs after the execution mode is fixed,
+ // so its failures are not retried as compiler failures
+ for (Instance instance : instances) {
+ initialize(instance);
+ freeInstances.offer(instance);
}
+ this.ready = true;
} catch (IOException e) {
throw new ShenyuWasmInitException(e);
}
}
+ /**
+ * Compiles the module for JVM-bytecode execution. Overriding to return
null
+ * forces the interpreter machine (used by tests to cover the fallback
path).
+ *
+ * @param module the parsed WASM module
+ * @return the compiled machine factory, or null for the interpreter
+ */
+ protected Function<Instance, Machine> compileMachineFactory(final
WasmModule module) {
+ return MachineFactoryCompiler.compile(module);
+ }
+
+ private Instance createInstance(final WasmModule module, final
ImportValues importValues,
+ final Function<Instance, Machine>
machineFactory) {
+ Instance.Builder builder = Instance.builder(module)
+ .withImportValues(importValues)
+ .withStart(false);
+ if (Objects.nonNull(machineFactory)) {
+ builder = builder.withMachineFactory(machineFactory);
+ }
+ Instance instance = builder.build();
+ if (Objects.isNull(instance.memory())) {
+ throw new ShenyuWasmInitException("memory not available in wasm
file: " + wasmName);
+ }
+ return instance;
+ }
+
+ private void initialize(final Instance instance) {
+ // Call _initialize if present (required by TinyGo -target
wasm-unknown),
+ // bound to this instance so init-time host callbacks see its own
memory.
+ findExport(instance, "_initialize").ifPresent(initFn -> {
+ executeBound(instance, initFn, new long[0]);
+ LOG.debug("Called _initialize for {}", wasmName);
+ });
+ }
+
private void registerWasiStubs(final Store store) {
// Use Chicory's built-in full WASI preview1 implementation for
comprehensive
// WASI function coverage, matching the previous wasmtime-java
behavior.
// Only inherit stdout/stderr so WASM guest output (e.g. Go println,
// Rust eprintln!) is visible for debugging. Avoid inheriting the full
- // system context (env/args/filesystem) for security hardening.
+ // system context (env/args/filesystem): besides security hardening,
+ // preopened files would add mutable state to the WASI object shared
+ // by all pooled instances.
var options = com.dylibso.chicory.wasi.WasiOptions.builder()
.withStdout(System.out)
.withStderr(System.err)
@@ -192,7 +287,7 @@ public class WasmLoader implements AutoCloseable {
(instance, args) -> new long[]{onPutResult(args[0], args[1],
(int) args[2])}
));
}
-
+
/**
* Override in subclasses to register additional host functions exposed to
the
* WASM module. This is called during construction after the built-in host
@@ -266,6 +361,13 @@ public class WasmLoader implements AutoCloseable {
/**
* Get the WASM exported function.
*
+ * <p>The returned function leases a pooled instance for the whole call,
+ * making concurrent invocations safe.
+ *
+ * <p>Reentrancy constraint: exports must not be invoked from a host
callback
+ * that runs during construction (e.g. triggered by {@code _initialize}) —
+ * the pool is not populated yet, so such calls fail fast.
+ *
* @param funcName the name of the WASM exported function
* @return an Optional containing the ExportFunction if found, otherwise
empty
*/
@@ -274,6 +376,14 @@ public class WasmLoader implements AutoCloseable {
}
private Optional<ExportFunction> findWasmExtern(final String funcName) {
+ if (findExport(primary, funcName).isPresent()) {
+ return Optional.of(args -> executeOnPool(funcName, args));
+ }
+ LOG.debug("WASM export function '{}' not found in {}", funcName,
wasmName);
+ return Optional.empty();
+ }
+
+ private static Optional<ExportFunction> findExport(final Instance
instance, final String funcName) {
// Chicory throws InvalidException for an absent export. Check the
names first
// so optional hooks remain optional without hiding invalid export
types.
var exports = instance.module().exportSection();
@@ -282,17 +392,114 @@ public class WasmLoader implements AutoCloseable {
return Optional.of(instance.export(funcName));
}
}
- LOG.debug("WASM export function '{}' not found in {}", funcName,
wasmName);
return Optional.empty();
}
-
+
+ private long[] executeOnPool(final String funcName, final long[] args) {
+ final Instance lease = acquire(funcName);
+ Throwable failure = null;
+ try {
+ return executeBound(lease, lease.export(funcName), args);
+ } catch (final Throwable t) {
+ // remembered so a release failure below cannot mask the original
failure
+ failure = t;
+ throw t;
+ } finally {
+ try {
+ release(lease);
+ } catch (final RuntimeException e) {
+ if (Objects.nonNull(failure)) {
+ failure.addSuppressed(e);
+ } else {
+ throw e;
+ }
+ }
+ }
+ }
+
+ // binds the thread to the instance so guest host callbacks resolve
getMemory() to it;
+ // pops on exit (nested calls each push their own entry)
+ private long[] executeBound(final Instance instance, final ExportFunction
export, final long[] args) {
+ final ArrayDeque<Instance> stack = bindings.get();
+ stack.push(instance);
+ try {
+ return export.apply(args);
+ } finally {
+ stack.pop();
+ }
+ }
+
/**
- * Get the WASM module's linear memory.
+ * Max time to wait for a free pooled instance before failing the call.
+ * Overridable so tests can exercise the timeout path quickly.
*
- * @return the WASM linear memory
+ * @return the lease timeout in milliseconds
+ */
+ protected long leaseTimeoutMillis() {
+ return DEFAULT_LEASE_TIMEOUT_MILLIS;
+ }
+
+ private Instance acquire(final String funcName) {
+ if (!ready) {
+ // reached only from a host callback during construction (e.g. a
nested
+ // export call from _initialize): no initialized instance is free
yet,
+ // so waiting could never succeed
+ throw new IllegalStateException(
+ "Nested export call during construction of " + wasmName
+ + "; the instance pool is not ready yet");
+ }
+ final ArrayDeque<Instance> active = bindings.get();
+ if (active.size() >= poolSize) {
+ // nesting deeper than the pool: this thread already leases every
instance,
+ // so waiting could never succeed
+ throw new IllegalStateException(
+ "Nested export calls exceed the WASM instance pool size
for " + wasmName);
+ }
+ final long timeoutMillis = leaseTimeoutMillis();
+ final Instance lease;
+ try {
+ lease = freeInstances.poll(timeoutMillis, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ShenyuWasmException(
+ "Interrupted while waiting for a pooled WASM instance of "
+ wasmName, e);
+ }
+ if (Objects.isNull(lease)) {
+ final int inUse = poolSize - freeInstances.size();
+ throw new ShenyuWasmException(
+ "Timed out after " + timeoutMillis + "ms waiting for a
pooled WASM instance of "
+ + wasmName + " for export '" + funcName + "' (pool
size " + poolSize
+ + ", " + inUse + " in use, " + active.size() + "
held by this thread)");
+ }
+ return lease;
+ }
+
+ private void release(final Instance lease) {
+ if (!freeInstances.offer(lease)) {
+ throw new IllegalStateException(
+ "Failed to return pooled WASM instance for " + wasmName
+ + "; queue is full, which indicates a
double-release or pool corruption");
+ }
+ }
+
+ /**
+ * Get the WASM linear memory of the instance executing on the current
thread.
+ *
+ * <p>Only valid inside an exported call or a host callback on the calling
thread
+ * (where it resolves to the leased instance): pooled instances each own
their
+ * memory, so there is no safe fallback and unbound callers fail fast.
+ *
+ * @return the WASM linear memory of the currently executing instance
+ * @throws IllegalStateException when no instance is executing on this
thread
*/
public Memory getMemory() {
- return instance.memory();
+ final Instance current = bindings.get().peek();
+ if (Objects.isNull(current)) {
+ throw new IllegalStateException(
+ "getMemory() called outside an exported call or host
callback of " + wasmName
+ + "; pooled instances have per-instance memory");
+ }
+ return current.memory();
}
/**
@@ -303,7 +510,7 @@ public class WasmLoader implements AutoCloseable {
public String getWasmName() {
return wasmName;
}
-
+
/**
* Get the Chicory Store.
*
diff --git
a/shenyu-plugin/shenyu-plugin-wasm-api/src/test/java/org/apache/shenyu/plugin/wasm/api/WasmLoaderConcurrentTest.java
b/shenyu-plugin/shenyu-plugin-wasm-api/src/test/java/org/apache/shenyu/plugin/wasm/api/WasmLoaderConcurrentTest.java
new file mode 100644
index 0000000000..b9fd95cc04
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-wasm-api/src/test/java/org/apache/shenyu/plugin/wasm/api/WasmLoaderConcurrentTest.java
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.wasm.api;
+
+import com.dylibso.chicory.runtime.ExportFunction;
+import com.dylibso.chicory.runtime.Instance;
+import com.dylibso.chicory.runtime.Machine;
+import com.dylibso.chicory.wasm.WasmModule;
+import org.apache.shenyu.plugin.wasm.api.exception.ShenyuWasmException;
+import org.apache.shenyu.plugin.wasm.api.loader.WasmLoader;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
+import java.util.function.LongFunction;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Regression tests for issue #7558: pooled instances must isolate concurrent
calls
+ * while still allowing executions to overlap.
+ */
+public class WasmLoaderConcurrentTest {
+
+ private final String expectedResult = "go result";
+
+ private final Map<Long, String> results = new ConcurrentHashMap<>();
+
+ private final AtomicLong ids = new AtomicLong();
+
+ @Test
+ public void concurrentExportsAreIsolatedTest() throws Exception {
+ assertConcurrentCallsAreIsolated(new TestLoader(4, 0L, null, null),
argId -> expectedResult);
+ }
+
+ @Test
+ public void concurrentExportsAreIsolatedOnInterpreterTest() throws
Exception {
+ assertConcurrentCallsAreIsolated(new InterpretedTestLoader(4), argId
-> expectedResult);
+ }
+
+ @Test
+ public void concurrentEchoPayloadsAreIsolatedTest() throws Exception {
+ // the echo guest returns exactly its own input, so input-side swaps
between calls are visible
+ assertConcurrentCallsAreIsolated(new EchoTestLoader(4), argId ->
"hello from java " + argId);
+ }
+
+ @Test
+ public void pooledInstancesExecuteInParallelTest() throws Exception {
+ final long blockedArgId = -1L;
+ final CountDownLatch blockEntered = new CountDownLatch(1);
+ final CountDownLatch releaseBlock = new CountDownLatch(1);
+ try (WasmLoader loader = new TestLoader(2, blockedArgId, blockEntered,
releaseBlock)) {
+ final ExportFunction execute = loader.getWasmExtern("execute")
+ .orElseThrow(() -> new IllegalStateException("execute
export not found"));
+ final ExecutorService pool = Executors.newFixedThreadPool(2);
+ try {
+ final Future<?> blockedCall = pool.submit(() ->
execute.apply(blockedArgId));
+ assertTrue(blockEntered.await(10, TimeUnit.SECONDS), "first
call never reached its host callback");
+
+ // must complete on a free instance, not queue behind the
parked one
+ final Future<String> secondCall = pool.submit(() -> {
+ final long argumentId = ids.incrementAndGet();
+ execute.apply(argumentId);
+ return results.remove(argumentId);
+ });
+ assertEquals(expectedResult, secondCall.get(10,
TimeUnit.SECONDS),
+ "second call should complete on a free pooled
instance, not queue behind the parked one");
+
+ releaseBlock.countDown();
+ blockedCall.get(10, TimeUnit.SECONDS);
+ assertEquals(expectedResult, results.remove(blockedArgId));
+ assertTrue(results.isEmpty(), "stray results indicate
cross-call payload leakage");
+ } finally {
+ releaseBlock.countDown();
+ pool.shutdownNow();
+ }
+ }
+ }
+
+ @Test
+ public void nestedExportsUseDistinctInstancesTest() throws Exception {
+ try (WasmLoader loader = new RecursiveTestLoader(3, 3)) {
+ final ExportFunction execute = loader.getWasmExtern("execute")
+ .orElseThrow(() -> new IllegalStateException("execute
export not found"));
+ execute.apply(-1L);
+ for (long argId = -1L; argId >= -3L; argId--) {
+ assertEquals(expectedResult, results.remove(argId), "nested
call " + argId + " lost its payload");
+ }
+ assertTrue(results.isEmpty(), "stray results indicate cross-call
payload leakage");
+ }
+ }
+
+ @Test
+ public void nestingBeyondPoolSizeFailsExplicitlyTest() throws Exception {
+ try (WasmLoader loader = new RecursiveTestLoader(2, 3)) {
+ final ExportFunction execute = loader.getWasmExtern("execute")
+ .orElseThrow(() -> new IllegalStateException("execute
export not found"));
+ final IllegalStateException ex =
assertThrows(IllegalStateException.class, () -> execute.apply(-1L));
+ assertTrue(ex.getMessage().contains("Nested export calls exceed
the WASM instance pool size"),
+ "unexpected message: " + ex.getMessage());
+ assertTrue(results.isEmpty(), "failed nesting must not leave stray
payloads");
+ }
+ }
+
+ @Test
+ public void poolExhaustionTimesOutInsteadOfHangingTest() throws Exception {
+ final long blockedArgId = -1L;
+ final CountDownLatch blockEntered = new CountDownLatch(1);
+ final CountDownLatch releaseBlock = new CountDownLatch(1);
+ try (WasmLoader loader = new TimeoutTestLoader(1, blockedArgId,
blockEntered, releaseBlock)) {
+ final ExportFunction execute = loader.getWasmExtern("execute")
+ .orElseThrow(() -> new IllegalStateException("execute
export not found"));
+ final ExecutorService pool = Executors.newFixedThreadPool(1);
+ try {
+ final Future<?> blockedCall = pool.submit(() ->
execute.apply(blockedArgId));
+ assertTrue(blockEntered.await(10, TimeUnit.SECONDS), "first
call never reached its host callback");
+
+ // the only instance is parked in the blocked callback: the
pool must
+ // fail this call after the (shortened) timeout instead of
hanging it
+ final ShenyuWasmException ex =
assertThrows(ShenyuWasmException.class,
+ () -> execute.apply(ids.incrementAndGet()));
+ assertTrue(ex.getMessage().contains("Timed out"), "unexpected
message: " + ex.getMessage());
+ // the parked call deterministically holds the only instance,
so the
+ // diagnostics must report the pool as fully in use
+ assertTrue(ex.getMessage().contains("pool size 1, 1 in use"),
+ "unexpected message: " + ex.getMessage());
+
+ releaseBlock.countDown();
+ blockedCall.get(10, TimeUnit.SECONDS);
+ assertEquals(expectedResult, results.remove(blockedArgId));
+ } finally {
+ releaseBlock.countDown();
+ pool.shutdownNow();
+ }
+ }
+ }
+
+ @Test
+ public void getMemoryOutsideExecutionFailsFastTest() {
+ try (WasmLoader loader = new TestLoader(1, 0L, null, null)) {
+ final IllegalStateException ex =
assertThrows(IllegalStateException.class, loader::getMemory);
+ assertTrue(ex.getMessage().contains("outside an exported call or
host callback"),
+ "unexpected message: " + ex.getMessage());
+ }
+ }
+
+ private void assertConcurrentCallsAreIsolated(final WasmLoader loader,
+ final LongFunction<String>
expectedFor) throws Exception {
+ try (loader) {
+ final ExportFunction execute = loader.getWasmExtern("execute")
+ .orElseThrow(() -> new IllegalStateException("execute
export not found"));
+ final int threads = 8;
+ final int iterations = 100;
+ final ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try {
+ final List<Future<List<String>>> futures = new ArrayList<>();
+ for (int i = 0; i < threads; i++) {
+ futures.add(pool.submit(() -> runIterations(execute,
iterations, expectedFor)));
+ }
+ final List<String> failures = new ArrayList<>();
+ for (Future<List<String>> future : futures) {
+ failures.addAll(future.get());
+ }
+ assertTrue(failures.isEmpty(), () ->
String.join(System.lineSeparator(), failures));
+ assertTrue(results.isEmpty(), "stray results indicate
cross-call payload leakage");
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+ }
+
+ private List<String> runIterations(final ExportFunction execute, final int
iterations,
+ final LongFunction<String> expectedFor)
{
+ final List<String> failures = new ArrayList<>();
+ for (int i = 0; i < iterations; i++) {
+ final long argumentId = ids.incrementAndGet();
+ try {
+ execute.apply(argumentId);
+ final String result = results.remove(argumentId);
+ final String expected = expectedFor.apply(argumentId);
+ if (!expected.equals(result)) {
+ failures.add(String.format("%s argId=%d expected '%s' but
got '%s'",
+ Thread.currentThread().getName(), argumentId,
expected, result));
+ }
+ } catch (RuntimeException e) {
+ failures.add(String.format("%s argId=%d threw %s: %s",
+ Thread.currentThread().getName(), argumentId,
e.getClass().getSimpleName(), e.getMessage()));
+ }
+ }
+ return failures;
+ }
+
+ /**
+ * Mirrors real plugin subclasses: built-in callbacks reach guest memory
through
+ * {@code getMemory()}. A non-zero {@code blockArgId} parks the matching
call
+ * inside its {@code get_args} callback until {@code releaseBlock} fires.
+ */
+ class TestLoader extends WasmLoader {
+
+ private final long blockArgId;
+
+ private final CountDownLatch blockEntered;
+
+ private final CountDownLatch releaseBlock;
+
+ TestLoader(final int poolSize, final long blockArgId,
+ final CountDownLatch blockEntered, final CountDownLatch
releaseBlock) {
+ super(AbstractWasmPluginTest.GoWasmPlugin.class, null, poolSize);
+ this.blockArgId = blockArgId;
+ this.blockEntered = blockEntered;
+ this.releaseBlock = releaseBlock;
+ }
+
+ @Override
+ protected long onGetArgs(final long argId, final long addr, final int
len) {
+ if (blockArgId != 0L && argId == blockArgId) {
+ blockEntered.countDown();
+ try {
+ assertTrue(releaseBlock.await(10, TimeUnit.SECONDS),
"blocked callback was never released");
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ final byte[] data = ("hello from java " +
argId).getBytes(StandardCharsets.UTF_8);
+ final int writeLen = Math.min(len, data.length);
+ getMemory().write(Math.toIntExact(addr), data, 0, writeLen);
+ return writeLen;
+ }
+
+ @Override
+ protected long onPutResult(final long argId, final long addr, final
int len) {
+ results.put(argId, new
String(getMemory().readBytes(Math.toIntExact(addr), len),
StandardCharsets.UTF_8));
+ return 0;
+ }
+ }
+
+ /** Forces the interpreter machine to cover the fallback execution path. */
+ final class InterpretedTestLoader extends TestLoader {
+
+ InterpretedTestLoader(final int poolSize) {
+ super(poolSize, 0L, null, null);
+ }
+
+ @Override
+ protected Function<Instance, Machine> compileMachineFactory(final
WasmModule module) {
+ return null;
+ }
+ }
+
+ /**
+ * Drives nested export calls from the {@code get_args} callback: a
negative
+ * argId {@code -n} recurses to level {@code n + 1} until {@code depth} is
reached.
+ */
+ class RecursiveTestLoader extends TestLoader {
+
+ private final int depth;
+
+ RecursiveTestLoader(final int poolSize, final int depth) {
+ super(poolSize, 0L, null, null);
+ this.depth = depth;
+ }
+
+ @Override
+ protected long onGetArgs(final long argId, final long addr, final int
len) {
+ if (argId < 0L && -argId < depth) {
+ getWasmExtern("execute").orElseThrow().apply(argId - 1L);
+ }
+ return super.onGetArgs(argId, addr, len);
+ }
+ }
+
+ /** Shortens the lease wait so pool-exhaustion timeouts surface in
milliseconds. */
+ final class TimeoutTestLoader extends TestLoader {
+
+ TimeoutTestLoader(final int poolSize, final long blockArgId,
+ final CountDownLatch blockEntered, final
CountDownLatch releaseBlock) {
+ super(poolSize, blockArgId, blockEntered, releaseBlock);
+ }
+
+ @Override
+ protected long leaseTimeoutMillis() {
+ return 200L;
+ }
+ }
+
+ /** Name carrier for the echo fixture built from wasm-concurrent/echo.wat.
*/
+ static class EchoPlugin {
+ }
+
+ /**
+ * Loads the echo guest: {@code put_result} returns exactly the bytes
+ * {@code get_args} delivered, so results must match each call's own input.
+ */
+ class EchoTestLoader extends WasmLoader {
+
+ EchoTestLoader(final int poolSize) {
+ super(EchoPlugin.class, null, poolSize);
+ }
+
+ @Override
+ protected long onGetArgs(final long argId, final long addr, final int
len) {
+ final byte[] data = ("hello from java " +
argId).getBytes(StandardCharsets.UTF_8);
+ final int writeLen = Math.min(len, data.length);
+ getMemory().write(Math.toIntExact(addr), data, 0, writeLen);
+ return writeLen;
+ }
+
+ @Override
+ protected long onPutResult(final long argId, final long addr, final
int len) {
+ results.put(argId, new
String(getMemory().readBytes(Math.toIntExact(addr), len),
StandardCharsets.UTF_8));
+ return 0;
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-wasm-api/src/test/resources/org.apache.shenyu.plugin.wasm.api.WasmLoaderConcurrentTest$EchoPlugin.wasm
b/shenyu-plugin/shenyu-plugin-wasm-api/src/test/resources/org.apache.shenyu.plugin.wasm.api.WasmLoaderConcurrentTest$EchoPlugin.wasm
new file mode 100644
index 0000000000..2e44f49fdd
Binary files /dev/null and
b/shenyu-plugin/shenyu-plugin-wasm-api/src/test/resources/org.apache.shenyu.plugin.wasm.api.WasmLoaderConcurrentTest$EchoPlugin.wasm
differ
diff --git
a/shenyu-plugin/shenyu-plugin-wasm-api/src/test/wasm-concurrent/Makefile
b/shenyu-plugin/shenyu-plugin-wasm-api/src/test/wasm-concurrent/Makefile
new file mode 100644
index 0000000000..cbc36b957d
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-wasm-api/src/test/wasm-concurrent/Makefile
@@ -0,0 +1,23 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+WAT2WASM ?= wat2wasm
+RESOURCES := ../resources
+
+.PHONY: build
+build:
+ $(WAT2WASM) echo.wat -o
'$(RESOURCES)/org.apache.shenyu.plugin.wasm.api.WasmLoaderConcurrentTest$$EchoPlugin.wasm'
diff --git
a/shenyu-plugin/shenyu-plugin-wasm-api/src/test/wasm-concurrent/echo.wat
b/shenyu-plugin/shenyu-plugin-wasm-api/src/test/wasm-concurrent/echo.wat
new file mode 100644
index 0000000000..f62d47eeaf
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-wasm-api/src/test/wasm-concurrent/echo.wat
@@ -0,0 +1,27 @@
+(;
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+;)
+
+;; Pure echo guest: put_result returns exactly the bytes get_args delivered,
+;; so each call's result is bound to its own request payload.
+(module
+ (import "shenyu" "get_args" (func $get_args (param i64 i64 i32) (result
i32)))
+ (import "shenyu" "put_result" (func $put_result (param i64 i64 i32) (result
i32)))
+ (memory (export "memory") 1)
+ (func (export "execute") (param $argId i64)
+ (local $len i32)
+ (local.set $len (call $get_args (local.get $argId) (i64.const 1024)
(i32.const 4096)))
+ (drop (call $put_result (local.get $argId) (i64.const 1024) (local.get
$len)))))
