Copilot commented on code in PR #7582:
URL: https://github.com/apache/shenyu/pull/7582#discussion_r4272523560
##########
shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java:
##########
@@ -117,38 +154,82 @@ 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();
- 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);
Review Comment:
During construction, `freeInstances` is empty while `initialize(instance)`
runs. If `_initialize` (or an init-time host callback) triggers a nested export
call via `getWasmExtern(...)`, `acquire()` will block forever on
`freeInstances.take()`. Populate the pool before running per-instance
initialization (e.g., pre-offer all instances and temporarily remove the one
being initialized, or offer all *other* instances prior to calling
`initialize(...)`).
##########
shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java:
##########
@@ -282,17 +374,66 @@ 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();
Review Comment:
After `stack.pop()`, the `ThreadLocal` still retains an (empty) `ArrayDeque`
for that thread. On long-lived thread pools, this can prolong retention of
thread-local structures and any stale references if future changes accidentally
leave entries behind. Consider calling `bindings.remove()` when the stack
becomes empty to minimize per-thread retention.
##########
shenyu-plugin/shenyu-plugin-wasm-api/src/main/java/org/apache/shenyu/plugin/wasm/api/loader/WasmLoader.java:
##########
@@ -282,17 +374,66 @@ 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:
Using `freeInstances.take()` can block indefinitely when the pool is
exhausted by other threads, which can hang request threads under load
(especially if nested exports happen inside host callbacks). Consider using a
bounded wait (`poll` with a timeout) and failing fast with a clear exception
message (or a configurable timeout) to avoid unbounded thread parking.
##########
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(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);
+ }
+ }
+
+ /** 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((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:
Casting `addr` from `long` to `int` can truncate silently if an unexpected
value is passed (and it’s inconsistent with `TestLoader` using
`Math.toIntExact(addr)`). Prefer `Math.toIntExact(addr)` here as well so
invalid addresses fail loudly instead of producing corrupted reads/writes.
##########
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(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);
+ }
+ }
+
+ /** 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((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:
Casting `addr` from `long` to `int` can truncate silently if an unexpected
value is passed (and it’s inconsistent with `TestLoader` using
`Math.toIntExact(addr)`). Prefer `Math.toIntExact(addr)` here as well so
invalid addresses fail loudly instead of producing corrupted reads/writes.
--
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]
