Copilot commented on code in PR #7582:
URL: https://github.com/apache/shenyu/pull/7582#discussion_r4269112506
##########
shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java:
##########
@@ -282,17 +369,62 @@ private Optional<ExportFunction> findWasmExtern(final
String funcName) {
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();
+ try {
+ return executeBound(lease, lease.export(funcName), args);
+ } finally {
+ release(lease);
+ }
+ }
+
+ // 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();
+ }
+ }
+
+ private Instance acquire() {
+ 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);
+ }
+ try {
+ return freeInstances.take();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(
+ "Interrupted while waiting for a pooled WASM instance of "
+ wasmName, e);
+ }
+ }
+
+ private void release(final Instance lease) {
+ freeInstances.offer(lease);
Review Comment:
`release()` uses `offer()` and ignores the boolean return value. If
`offer()` ever fails (e.g., accidental double-release, release-after-close, or
queue corruption), the instance is silently dropped which can later deadlock
callers in `acquire()`. Use a non-silent operation (e.g., `put()` or `add()`)
or check the return value and fail fast with a clear exception.
##########
shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java:
##########
@@ -117,38 +154,77 @@ public WasmLoader(final Class<?> wasmClass, final
Consumer<Store> initializer) {
}
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();
+ machineFactory = compileMachineFactory(module);
+ firstInstance = createInstance(module, importValues,
machineFactory);
LOG.debug("Using runtime compiler for {}", wasmName);
Review Comment:
`compileMachineFactory()` is documented to allow returning `null` to force
interpreter execution (used by tests), but the constructor always logs `Using
runtime compiler` on the happy path even when `machineFactory` is `null` and
the interpreter is actually used. Adjust the control flow so `null` is treated
as the interpreter path (including logging) to keep runtime behavior and the
method contract consistent.
##########
shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java:
##########
@@ -282,17 +369,62 @@ private Optional<ExportFunction> findWasmExtern(final
String funcName) {
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();
+ try {
+ return executeBound(lease, lease.export(funcName), args);
+ } finally {
+ release(lease);
+ }
+ }
Review Comment:
Each invocation performs `lease.export(funcName)` lookup. Since
`getWasmExtern()` already validates the export exists, consider caching the
`ExportFunction` per instance (or per `(funcName, instance)` pair) when
building the pool to avoid repeated export resolution in hot paths.
##########
shenyu-plugin/shenyu-plugin-wasm-api/src/test/java/org/apache/shenyu/plugin/wasm/api/WasmLoaderConcurrentTest.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.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");
+ }
+ }
+
+ 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((int) 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((int) addr,
len), StandardCharsets.UTF_8));
Review Comment:
The test writes to WASM memory by casting `addr` (a `long`) to `int`. If an
address outside the 32-bit range is ever passed, this truncates silently and
can produce incorrect writes that are hard to diagnose. Prefer a checked
conversion (e.g., `Math.toIntExact(addr)`) in the test harness to fail fast on
unexpected addresses.
##########
shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java:
##########
@@ -282,17 +369,62 @@ private Optional<ExportFunction> findWasmExtern(final
String funcName) {
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();
+ try {
+ return executeBound(lease, lease.export(funcName), args);
+ } finally {
+ release(lease);
+ }
+ }
+
+ // 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();
+ }
+ }
+
+ private Instance acquire() {
+ 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);
+ }
+ try {
+ return freeInstances.take();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(
+ "Interrupted while waiting for a pooled WASM instance of "
+ wasmName, e);
+ }
+ }
Review Comment:
`acquire()` blocks indefinitely on `freeInstances.take()`, but it does not
consult the `closed` flag. If a loader is closed while threads are blocked here
(or if an instance is lost due to an incorrect release path), those threads may
hang permanently. Consider incorporating a closed-aware acquisition strategy
(e.g., check `closed` before/after waiting, or use `poll(timeout)` with a
bounded wait and a clearer failure mode).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
