diff --git a/actuator/src/main/java/org/tron/core/vm/PrecompiledContracts.java b/actuator/src/main/java/org/tron/core/vm/PrecompiledContracts.java index c9968661929..211b8f34db6 100644 --- a/actuator/src/main/java/org/tron/core/vm/PrecompiledContracts.java +++ b/actuator/src/main/java/org/tron/core/vm/PrecompiledContracts.java @@ -30,11 +30,13 @@ import java.util.HashSet; import java.util.List; import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import lombok.AllArgsConstructor; import lombok.Getter; import lombok.Setter; @@ -481,7 +483,11 @@ public void setResult(ProgramResult result) { } protected long getCPUTimeLeftInNanoSecond() { - long left = getVmShouldEndInUs() * VMConstant.ONE_THOUSAND - System.nanoTime(); + return getCPUTimeLeftInNanoSecond(getVmShouldEndInUs()); + } + + private long getCPUTimeLeftInNanoSecond(long deadlineUs) { + long left = deadlineUs * VMConstant.ONE_THOUSAND - System.nanoTime(); if (left <= 0) { throw Program.Exception.notEnoughTime("call"); } else { @@ -489,6 +495,109 @@ protected long getCPUTimeLeftInNanoSecond() { } } + /** + * Returns whether to collect results under the precompile's historical timeout policy. + * DROP rejects any timed-out wait. MATCH_CPU_CHECK still accepts one when + * {@link #allowTimedOutResult(long)} is true. A budget exhausted before waiting is + * rejected either way. Cancellation belongs to {@link #collectParallel}. + */ + boolean awaitParallel(CountDownLatch latch, ParallelTimeout timeout) + throws InterruptedException { + return awaitParallel(latch, timeout, getVmShouldEndInUs()); + } + + private boolean awaitParallel(CountDownLatch latch, ParallelTimeout timeout, long deadlineUs) + throws InterruptedException { + boolean withNoTimeout; + try { + withNoTimeout = latch.await(getCPUTimeLeftInNanoSecond(deadlineUs), TimeUnit.NANOSECONDS); + } catch (OutOfTimeException expired) { + // Budget was already gone, so the historical path does not read results. + return false; + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw interrupted; + } + return withNoTimeout + || (timeout == ParallelTimeout.MATCH_CPU_CHECK && allowTimedOutResult(deadlineUs)); + } + + /** + * Returns null on timeout or cancellation; propagates interruption and worker failures. + * Capture the invocation deadline once for the latch wait and all result reads. + * The latch opens before FutureTask publishes its result, so a bounded read uses the + * same instant as {@link Program#checkCPUTimeLimit}. Values are copied into the + * returned list before {@code finally} runs. Cancel is a no-op on a completed future + * and does not discard those values. + */ + List collectParallel(CountDownLatch latch, List> futures, + ParallelTimeout timeout) throws InterruptedException, ExecutionException { + final long deadlineUs = getVmShouldEndInUs(); + try { + if (!awaitParallel(latch, timeout, deadlineUs)) { + return null; + } + List results = new ArrayList<>(futures.size()); + for (Future future : futures) { + results.add(readFuture(future, deadlineUs)); + } + return results; + } catch (TimeoutException | CancellationException ignored) { + return null; + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw interrupted; + } finally { + cancelAll(futures); + } + } + + private T readFuture(Future future, long deadlineUs) + throws InterruptedException, ExecutionException, TimeoutException { + if (!boundsCollection()) { + return future.get(); + } + // CPU checks reject only when floor(nowNs / 1000) is greater than deadlineUs. + long remainNs = (deadlineUs + 1) * VMConstant.ONE_THOUSAND - System.nanoTime(); + if (remainNs <= 0) { + throw new TimeoutException(); + } + return future.get(remainNs, TimeUnit.NANOSECONDS); + } + + /** Debug, solidity, and pre-energy-adjustment nodes keep the historical untimed read. */ + private static boolean boundsCollection() { + CommonParameter parameter = CommonParameter.getInstance(); + if (parameter.isDebug() || parameter.isSolidityNode()) { + return false; + } + return VMConfig.allowEnergyAdjustment(); + } + + /** Completed futures are unaffected; cancellation of running work is best effort. */ + static void cancelAll(List> futures) { + for (Future future : futures) { + future.cancel(true); + } + } + + /** + * A timed-out latch may still be read when the VM will not turn this call into + * an out-of-time transaction. Debug and solidity nodes skip + * {@link Program#checkCPUTimeLimit}. Without energy adjustment there is no + * end-of-transaction check, so a last opcode can still succeed. Otherwise the + * same microsecond boundary as that opcode check applies. + */ + private boolean allowTimedOutResult(long deadlineUs) { + return !boundsCollection() + || System.nanoTime() / VMConstant.ONE_THOUSAND <= deadlineUs; + } + + enum ParallelTimeout { + DROP, + MATCH_CPU_CHECK + } + protected byte[] dataOne() { byte[] ret = new byte[WORD_SIZE]; ret[31] = 1; @@ -1206,26 +1315,27 @@ private Pair doExecute(byte[] data) // add check CountDownLatch countDownLatch = new CountDownLatch(cnt); List> futures = new ArrayList<>(cnt); - - for (int i = 0; i < cnt; i++) { - Future future = workers - .submit(new RecoverAddrTask(countDownLatch, hash, signatures[i], i)); - futures.add(future); - } - boolean withNoTimeout = countDownLatch - .await(getCPUTimeLeftInNanoSecond(), TimeUnit.NANOSECONDS); - - if (!withNoTimeout) { - logger.info("BatchValidateSign timeout"); - throw Program.Exception.notEnoughTime("call BatchValidateSign precompile method"); - } - - for (Future future : futures) { - RecoverAddrResult result = future.get(); - int index = result.nonce; - if (DataWord.equalAddressByteArray(result.addr, addresses[index])) { - res[index] = 1; + try { + for (int i = 0; i < cnt; i++) { + Future future = workers + .submit(new RecoverAddrTask(countDownLatch, hash, signatures[i], i)); + futures.add(future); + } + List results = collectParallel(countDownLatch, futures, + ParallelTimeout.DROP); + if (results == null) { + logger.info("BatchValidateSign timeout"); + throw Program.Exception.notEnoughTime("call BatchValidateSign precompile method"); + } + for (RecoverAddrResult result : results) { + int index = result.nonce; + if (DataWord.equalAddressByteArray(result.addr, addresses[index])) { + res[index] = 1; + } } + } finally { + // Also covers failure partway through submission. + cancelAll(futures); } } return Pair.of(true, res); @@ -1580,42 +1690,49 @@ public Pair execute(byte[] data) { workers = workersInNonConstantCall; } - // submit check spend task - for (int i = 0; i < spendCount; i++) { - Future futureCheckSpend = workers - .submit(new SaplingCheckSpendTask(countDownLatch, spendCv[i], anchor[i], - nullifier[i], rk[i], spendProof[i], spendAuthSig[i], signHash)); - futures.add(futureCheckSpend); - } - //submit check output task - for (int i = 0; i < receiveCount; i++) { - Future futureCheckOutput = workers - .submit(new SaplingCheckOutputTask(countDownLatch, receiveCv[i], receiveCm[i], - receiveEpk[i], receiveProof[i])); - futures.add(futureCheckOutput); - } - // submit check binding signature - Future futureCheckBindingSig = workers - .submit(new SaplingCheckBingdingSig(countDownLatch, value, bindingSig, - signHash, spendCvs, spendCount * 32, receiveCvs, receiveCount * 32)); - futures.add(futureCheckBindingSig); - - boolean withNoTimeout = countDownLatch.await(getCPUTimeLeftInNanoSecond(), - TimeUnit.NANOSECONDS); - boolean checkResult = true; - for (Future future : futures) { - boolean eachTaskResult = future.get(); - checkResult = checkResult && eachTaskResult; - } - if (checkResult) { - return insertLeaves(frontier, leafCount, receiveCm); - } else { - return Pair.of(true, DataWord.ZERO().getData()); + try { + // submit check spend task + for (int i = 0; i < spendCount; i++) { + Future futureCheckSpend = workers + .submit(new SaplingCheckSpendTask(countDownLatch, spendCv[i], anchor[i], + nullifier[i], rk[i], spendProof[i], spendAuthSig[i], signHash)); + futures.add(futureCheckSpend); + } + //submit check output task + for (int i = 0; i < receiveCount; i++) { + Future futureCheckOutput = workers + .submit(new SaplingCheckOutputTask(countDownLatch, receiveCv[i], receiveCm[i], + receiveEpk[i], receiveProof[i])); + futures.add(futureCheckOutput); + } + // submit check binding signature + Future futureCheckBindingSig = workers + .submit(new SaplingCheckBingdingSig(countDownLatch, value, bindingSig, + signHash, spendCvs, spendCount * 32, receiveCvs, receiveCount * 32)); + futures.add(futureCheckBindingSig); + + List results = collectParallel(countDownLatch, futures, + ParallelTimeout.MATCH_CPU_CHECK); + if (results == null) { + return Pair.of(true, DataWord.ZERO().getData()); + } + boolean checkResult = true; + for (Boolean eachTaskResult : results) { + checkResult = checkResult && eachTaskResult; + } + if (checkResult) { + return insertLeaves(frontier, leafCount, receiveCm); + } else { + return Pair.of(true, DataWord.ZERO().getData()); + } + } finally { + // Also covers failure partway through submission. + cancelAll(futures); } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - logger.info("VerifyTransferProof exception: " + e.getMessage()); } catch (Throwable any) { + if (any instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } String errorMsg = any.getMessage(); if (errorMsg == null && any.getCause() != null) { errorMsg = any.getCause().getMessage(); diff --git a/actuator/src/test/java/org/tron/core/vm/PrecompiledParallelAwaitTest.java b/actuator/src/test/java/org/tron/core/vm/PrecompiledParallelAwaitTest.java new file mode 100644 index 00000000000..c800391f9a6 --- /dev/null +++ b/actuator/src/test/java/org/tron/core/vm/PrecompiledParallelAwaitTest.java @@ -0,0 +1,441 @@ +package org.tron.core.vm; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Test; +import org.tron.common.parameter.CommonParameter; +import org.tron.core.vm.PrecompiledContracts.PrecompiledContract; +import org.tron.core.vm.PrecompiledContracts.PrecompiledContract.ParallelTimeout; +import org.tron.core.vm.config.VMConfig; + +public class PrecompiledParallelAwaitTest { + + @Test + public void testCollectionCapturesDeadlineOnce() throws Exception { + AtomicInteger reads = new AtomicInteger(); + long deadlineUs = System.nanoTime() / 1_000 + 5_000_000; + PrecompiledContract contract = new PrecompiledContracts.Identity() { + @Override + public long getVmShouldEndInUs() { + reads.incrementAndGet(); + return deadlineUs; + } + }; + FutureTask completed = new FutureTask<>(() -> Boolean.TRUE); + completed.run(); + List> futures = new ArrayList<>(); + futures.add(completed); + + withNode(true, false, false, () -> { + assertEquals(Boolean.TRUE, + contract.collectParallel(new CountDownLatch(0), futures, ParallelTimeout.MATCH_CPU_CHECK) + .get(0)); + assertEquals(1, reads.get()); + }); + } + + @Test + public void testCompletedWaitCollectsResults() throws Exception { + ExecutorService pool = Executors.newFixedThreadPool(3); + try { + CountDownLatch latch = new CountDownLatch(3); + AtomicInteger started = new AtomicInteger(); + List> futures = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + futures.add(pool.submit(() -> { + started.incrementAndGet(); + latch.countDown(); + return Boolean.TRUE; + })); + } + boolean collect = await(latch, 5_000, ParallelTimeout.DROP); + assertTrue(collect); + assertEquals(3, started.get()); + for (Future future : futures) { + assertFalse(future.isCancelled()); + assertEquals(Boolean.TRUE, future.get()); + } + } finally { + pool.shutdownNow(); + } + } + + @Test + public void testExpiredDeadlineCancelsWithoutReading() throws Exception { + withNode(true, true, true, () -> { + try (HeldWorkers workers = new HeldWorkers()) { + workers.startOne(); + workers.queueTwo(); + List results = collect(workers.idleLatch(), workers.futures, -5, + ParallelTimeout.MATCH_CPU_CHECK); + assertNull(results); + assertEquals(1, workers.started.get()); + assertTrue(workers.futures.get(1).isCancelled()); + assertTrue(workers.futures.get(2).isCancelled()); + } + }); + } + + @Test + public void testDropPolicyCancelsTimedOutWaitOnSolidity() throws Exception { + withNode(true, false, true, () -> { + try (HeldWorkers workers = new HeldWorkers()) { + workers.startOne(); + workers.queueTwo(); + List results = collect(workers.idleLatch(), workers.futures, 40, + ParallelTimeout.DROP); + assertNull(results); + assertEquals(1, workers.started.get()); + assertTrue(workers.futures.get(1).isCancelled()); + assertTrue(workers.futures.get(2).isCancelled()); + } + }); + } + + @Test + public void testSolidityKeepsResultAfterTimedOutWait() throws Exception { + assertTimedOutWaitIsKept(false, true); + } + + @Test + public void testDebugKeepsResultAfterTimedOutWait() throws Exception { + assertTimedOutWaitIsKept(true, false); + } + + @Test + public void testEnergyAdjustmentOffKeepsResultAfterTimedOutWait() throws Exception { + withNode(false, false, false, () -> { + try (HeldWorkers workers = new HeldWorkers()) { + workers.startOne(); + boolean collect = await(workers.idleLatch(), 40, ParallelTimeout.MATCH_CPU_CHECK); + assertTrue(collect); + assertFalse(workers.futures.get(0).isCancelled()); + workers.release(); + assertEquals(Boolean.TRUE, workers.futures.get(0).get(5, TimeUnit.SECONDS)); + } + }); + } + + @Test + public void testEnergyAdjustmentOnDropsTimedOutWait() throws Exception { + withNode(true, false, false, () -> { + try (HeldWorkers workers = new HeldWorkers()) { + workers.startOne(); + workers.queueTwo(); + List results = collect(workers.idleLatch(), workers.futures, 40, + ParallelTimeout.MATCH_CPU_CHECK); + assertNull(results); + assertEquals(1, workers.started.get()); + assertTrue(workers.futures.get(1).isCancelled()); + assertTrue(workers.futures.get(2).isCancelled()); + } + }); + } + + @Test + public void testInterruptCancelsAndRestoresStatus() throws Exception { + try (HeldWorkers workers = new HeldWorkers()) { + workers.startOne(); + Thread worker = Thread.currentThread(); + Thread interrupter = new Thread(() -> { + try { + Thread.sleep(30); + } catch (InterruptedException ignored) { + Thread.currentThread().interrupt(); + return; + } + worker.interrupt(); + }); + interrupter.start(); + try { + collect(workers.idleLatch(), workers.futures, 5_000, ParallelTimeout.MATCH_CPU_CHECK); + fail("wait should have been interrupted"); + } catch (InterruptedException expected) { + assertTrue(Thread.currentThread().isInterrupted()); + assertTrue(workers.futures.get(0).isCancelled()); + } finally { + Thread.interrupted(); + interrupter.join(1_000); + } + } + } + + @Test(timeout = 3_000) + public void testCollectStopsWhenPublicationMissesOuterCheck() throws Exception { + withNode(true, false, false, () -> { + AtomicInteger started = new AtomicInteger(); + try (LatePublication publication = new LatePublication()) { + publication.queueAnother(started); + List results = collect(publication.latch, publication.futures, 40, + ParallelTimeout.MATCH_CPU_CHECK); + assertNull(results); + assertEquals(0, started.get()); + assertTrue(publication.futures.get(1).isCancelled()); + } + }); + } + + @Test(timeout = 3_000) + public void testSolidityReadsResultPublishedAfterDeadline() throws Exception { + assertLatePublicationIsRead(true, false, true); + } + + @Test(timeout = 3_000) + public void testEnergyAdjustmentOffReadsResultPublishedAfterDeadline() throws Exception { + assertLatePublicationIsRead(false, false, false); + } + + @Test(timeout = 3_000) + public void testInterruptDuringGetCancelsAndRestoresStatus() throws Exception { + withNode(true, false, false, () -> { + try (LatePublication publication = new LatePublication()) { + Thread reader = Thread.currentThread(); + Thread interrupter = new Thread(() -> { + try { + Thread.sleep(30); + } catch (InterruptedException ignored) { + Thread.currentThread().interrupt(); + return; + } + reader.interrupt(); + }); + interrupter.start(); + try { + collect(publication.latch, publication.futures, 5_000, ParallelTimeout.MATCH_CPU_CHECK); + fail("get should have been interrupted"); + } catch (InterruptedException expected) { + assertTrue(Thread.currentThread().isInterrupted()); + assertTrue(publication.future.isCancelled()); + } finally { + Thread.interrupted(); + interrupter.join(1_000); + } + } + }); + } + + @Test + public void testWorkerFailurePropagatesAndCancelsRemainingTasks() throws Exception { + IllegalStateException failure = new IllegalStateException("worker failed"); + FutureTask failed = new FutureTask<>(() -> { + throw failure; + }); + failed.run(); + FutureTask pending = new FutureTask<>(() -> Boolean.TRUE); + List> futures = new ArrayList<>(); + futures.add(failed); + futures.add(pending); + + try { + collect(new CountDownLatch(0), futures, 5_000, ParallelTimeout.MATCH_CPU_CHECK); + fail("worker failure should propagate"); + } catch (ExecutionException expected) { + assertSame(failure, expected.getCause()); + assertTrue(pending.isCancelled()); + assertFalse(failed.isCancelled()); + } + } + + @Test + public void testCancelledResultDropsAndCancelsRemainingTasks() throws Exception { + FutureTask cancelled = new FutureTask<>(() -> Boolean.TRUE); + cancelled.cancel(true); + FutureTask pending = new FutureTask<>(() -> Boolean.TRUE); + List> futures = new ArrayList<>(); + futures.add(cancelled); + futures.add(pending); + + assertNull(collect(new CountDownLatch(0), futures, 5_000, ParallelTimeout.MATCH_CPU_CHECK)); + assertTrue(pending.isCancelled()); + } + + private static void assertLatePublicationIsRead(boolean energy, boolean debug, boolean solidity) + throws Exception { + withNode(energy, debug, solidity, () -> { + ExecutorService pool = Executors.newSingleThreadExecutor(); + try { + CountDownLatch entered = new CountDownLatch(1); + Future future = pool.submit(() -> { + entered.countDown(); + Thread.sleep(200); + return Boolean.TRUE; + }); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + List> futures = new ArrayList<>(); + futures.add(future); + List results = collect(new CountDownLatch(0), futures, 40, + ParallelTimeout.MATCH_CPU_CHECK); + assertEquals(Boolean.TRUE, results.get(0)); + assertFalse(future.isCancelled()); + } finally { + pool.shutdownNow(); + } + }); + } + + private static void assertTimedOutWaitIsKept(boolean debug, boolean solidity) throws Exception { + withNode(true, debug, solidity, () -> { + try (HeldWorkers workers = new HeldWorkers()) { + workers.startOne(); + boolean collect = await(workers.idleLatch(), 40, ParallelTimeout.MATCH_CPU_CHECK); + assertTrue(collect); + assertFalse(workers.futures.get(0).isCancelled()); + workers.release(); + assertEquals(Boolean.TRUE, workers.futures.get(0).get(5, TimeUnit.SECONDS)); + } + }); + } + + private static boolean await(CountDownLatch latch, long delayMs, ParallelTimeout timeout) + throws InterruptedException { + return contractAt(delayMs).awaitParallel(latch, timeout); + } + + private static List collect(CountDownLatch latch, List> futures, + long delayMs, ParallelTimeout timeout) throws Exception { + return contractAt(delayMs).collectParallel(latch, futures, timeout); + } + + private static PrecompiledContract contractAt(long delayMs) { + PrecompiledContract contract = new PrecompiledContracts.Identity(); + contract.setVmShouldEndInUs(System.nanoTime() / 1_000 + delayMs * 1_000); + return contract; + } + + private static void withNode(boolean energyAdjustment, boolean debug, boolean solidity, + ThrowingRunnable body) throws Exception { + CommonParameter parameter = CommonParameter.getInstance(); + boolean previousDebug = parameter.isDebug(); + boolean previousSolidity = parameter.isSolidityNode(); + boolean previousEnergy = VMConfig.allowEnergyAdjustment(); + parameter.setDebug(debug); + parameter.setSolidityNode(solidity); + VMConfig.initAllowEnergyAdjustment(energyAdjustment ? 1 : 0); + try { + body.run(); + } finally { + parameter.setDebug(previousDebug); + parameter.setSolidityNode(previousSolidity); + VMConfig.initAllowEnergyAdjustment(previousEnergy ? 1 : 0); + } + } + + private interface ThrowingRunnable { + void run() throws Exception; + } + + /** + * The task counts the latch down, then blocks before {@code call()} returns. That is the + * window after {@code countDown()} and before {@code FutureTask} publishes the result. + * A second task stays queued on the same single thread. + */ + private static final class LatePublication implements AutoCloseable { + private final ExecutorService pool = Executors.newFixedThreadPool(1); + private final CountDownLatch release = new CountDownLatch(1); + private final CountDownLatch latch = new CountDownLatch(0); + private final List> futures = new ArrayList<>(); + private final Future future; + + private LatePublication() throws InterruptedException { + CountDownLatch entered = new CountDownLatch(1); + future = pool.submit(() -> { + entered.countDown(); + while (release.getCount() > 0) { + try { + release.await(50, TimeUnit.MILLISECONDS); + } catch (InterruptedException ignored) { + // Hold this worker so a queued task cannot start during cancel. + } + } + return Boolean.TRUE; + }); + futures.add(future); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + } + + private void queueAnother(AtomicInteger started) { + futures.add(pool.submit(() -> { + started.incrementAndGet(); + return Boolean.TRUE; + })); + } + + @Override + public void close() { + release.countDown(); + pool.shutdownNow(); + } + } + + /** + * One worker holds the only pool thread until {@link #release()}. Further tasks stay queued, + * so cancellation of those tasks is observable. + */ + private static final class HeldWorkers implements AutoCloseable { + private final ExecutorService pool = Executors.newFixedThreadPool(1); + private final Object lock = new Object(); + private final AtomicInteger started = new AtomicInteger(); + private final List> futures = new ArrayList<>(); + private boolean hold = true; + + private void startOne() throws InterruptedException { + CountDownLatch entered = new CountDownLatch(1); + futures.add(pool.submit(() -> holdUntilReleased(entered))); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + } + + private void queueTwo() { + futures.add(pool.submit(() -> holdUntilReleased(null))); + futures.add(pool.submit(() -> holdUntilReleased(null))); + } + + private Boolean holdUntilReleased(CountDownLatch entered) { + started.incrementAndGet(); + if (entered != null) { + entered.countDown(); + } + synchronized (lock) { + while (hold) { + try { + lock.wait(50); + } catch (InterruptedException ignored) { + // Keep this worker so a queued task cannot start while cancel runs. + } + } + } + return Boolean.TRUE; + } + + private CountDownLatch idleLatch() { + return new CountDownLatch(futures.size() + 1); + } + + private void release() { + synchronized (lock) { + hold = false; + lock.notifyAll(); + } + } + + @Override + public void close() { + release(); + pool.shutdownNow(); + } + } +} diff --git a/framework/src/test/java/org/tron/core/vm/PrecompiledParallelVmTest.java b/framework/src/test/java/org/tron/core/vm/PrecompiledParallelVmTest.java new file mode 100644 index 00000000000..583b3822451 --- /dev/null +++ b/framework/src/test/java/org/tron/core/vm/PrecompiledParallelVmTest.java @@ -0,0 +1,408 @@ +package org.tron.core.vm; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.withSettings; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; +import org.tron.common.crypto.ECKey; +import org.tron.common.parameter.CommonParameter; +import org.tron.common.runtime.InternalTransaction; +import org.tron.common.runtime.InternalTransaction.TrxType; +import org.tron.common.runtime.ProgramResult; +import org.tron.common.runtime.RuntimeImpl; +import org.tron.common.runtime.vm.DataWord; +import org.tron.common.zksnark.JLibrustzcash; +import org.tron.core.actuator.VMActuator; +import org.tron.core.capsule.TransactionCapsule; +import org.tron.core.db.TransactionContext; +import org.tron.core.vm.PrecompiledContracts.BatchValidateSign; +import org.tron.core.vm.PrecompiledContracts.PrecompiledContract; +import org.tron.core.vm.PrecompiledContracts.VerifyTransferProof; +import org.tron.core.vm.config.VMConfig; +import org.tron.core.vm.program.Program; +import org.tron.core.vm.program.invoke.ProgramInvoke; +import org.tron.core.vm.repository.Repository; +import org.tron.protos.Protocol.Transaction; +import org.tron.protos.Protocol.Transaction.Contract.ContractType; +import org.tron.protos.Protocol.Transaction.Result.contractResult; + +/** + * Execute a terminal CALL through the real VM, actuator and runtime result mapping. Transaction + * validation/storage and cryptographic workers are isolated; CPU checks and energy accounting are + * not mocked. With no following opcode, a timeout must be caught by the transaction-end CPU check. + */ +@RunWith(Parameterized.class) +public class PrecompiledParallelVmTest { + + private static final long ENERGY_LIMIT = 1_000_000; + private static final int CALL_ENERGY = 500_000; + private static final int OUTPUT_OFFSET = 4096; + private static final long ON_TIME_BUDGET_US = TimeUnit.SECONDS.toMicros(5); + private static final long TIMEOUT_BUDGET_US = TimeUnit.MILLISECONDS.toMicros(150); + private final boolean transferProof; + private final Completion completion; + private final NodeMode mode; + + private enum Completion { + ON_TIME, LATCH_TIMEOUT, PUBLICATION_TIMEOUT + } + + private enum NodeMode { + ENERGY_ADJUSTMENT, ENERGY_ADJUSTMENT_OFF, DEBUG, SOLIDITY + } + + @BeforeClass + public static void warmUp() { + OperationRegistry.init(); + // Initialize VerifyProof.UNCOMMITTED with the real Merkle primitive before mocking Sapling. + PrecompiledContracts.getContractForAddress(DataWord.ZERO()); + } + + @Parameterized.Parameters(name = "{index}: transferProof={0}, {1}, {2}") + public static Collection cases() { + List cases = new ArrayList<>(); + for (boolean transferProof : new boolean[]{false, true}) { + for (Completion completion : Completion.values()) { + for (NodeMode mode : NodeMode.values()) { + cases.add(new Object[]{transferProof, completion, mode}); + } + } + } + return cases; + } + + public PrecompiledParallelVmTest(boolean transferProof, Completion completion, NodeMode mode) { + this.transferProof = transferProof; + this.completion = completion; + this.mode = mode; + } + + @Test(timeout = 15_000) + public void terminalCallPreservesTransactionResultAndEnergy() throws Exception { + CommonParameter parameter = CommonParameter.getInstance(); + boolean previousDebug = parameter.isDebug(); + boolean previousSolidity = parameter.isSolidityNode(); + // A local snapshot isolates this test's flags without changing the HEAD configuration. + VMConfig.Snapshot snapshot = new VMConfig.Snapshot(); + snapshot.allowEnergyAdjustment = mode != NodeMode.ENERGY_ADJUSTMENT_OFF; + snapshot.allowTvmOsaka = true; + snapshot.allowTvmSelfdestructRestriction = true; + snapshot.allowTvmSolidity059 = true; + snapshot.allowShieldedTRC20Transaction = true; + VMConfig.setLocalSnapshot(snapshot); + parameter.setDebug(mode == NodeMode.DEBUG); + parameter.setSolidityNode(mode == NodeMode.SOLIDITY); + try (MockedStatic sapling = mockStatic(JLibrustzcash.class)) { + sapling.when(() -> JLibrustzcash.librustzcashSaplingCheckSpendNew(any())).thenReturn(true); + sapling.when(() -> JLibrustzcash.librustzcashSaplingCheckOutputNew(any())).thenReturn(true); + sapling.when(() -> JLibrustzcash.librustzcashSaplingFinalCheckNew(any())).thenReturn(true); + // Allow cold mocks and scheduling pauses in successful calls without slowing timeout cases. + long budgetUs = completion == Completion.ON_TIME ? ON_TIME_BUDGET_US : TIMEOUT_BUDGET_US; + Deadline deadline = new Deadline(budgetUs); + ControlledTasks tasks = new ControlledTasks(completion, deadline); + PrecompiledContract contract = transferProof + ? new VerifyTransferProof() : new BatchValidateSign(); + byte[] input = transferProof ? transferInput() : signatureInput(); + // Initialize mocks before the first CPU deadline is read. + Repository repository = mock(Repository.class); + when(repository.newRepositoryChild()).thenReturn(repository); + ProgramInvoke invoke = mock(ProgramInvoke.class); + when(invoke.getContractAddress()).thenReturn(new DataWord(123)); + when(invoke.getCallerAddress()).thenReturn(new DataWord(123)); + when(invoke.getDeposit()).thenReturn(repository); + when(invoke.getEnergyLimit()).thenReturn(ENERGY_LIMIT); + when(invoke.getVmShouldEndInUs()).thenAnswer(ignored -> deadline.get()); + Program program = new Program(new byte[]{(byte) Op.CALL}, new byte[20], invoke, + new InternalTransaction(Transaction.getDefaultInstance(), TrxType.TRX_UNKNOWN_TYPE)); + program.setRootTransactionId(new byte[32]); + program.memorySave(0, input); + // CALL arguments are supplied directly so the only executed opcode is the precompile CALL. + program.stackPush(new DataWord(128)); + program.stackPush(new DataWord(OUTPUT_OFFSET)); + program.stackPush(new DataWord(input.length)); + program.stackPush(DataWord.ZERO()); + program.stackPush(DataWord.ZERO()); + DataWord address = new DataWord(transferProof ? 0x1000002 : 9); + program.stackPush(address); + program.stackPush(new DataWord(CALL_ENERGY)); + // CALL costs 40 plus memory expansion from 11/65 to 132 words (397/227), + // Then the precompile costs 1500/200000. Pin the totals independently of the implementation. + long normalEnergy = transferProof ? 200_267 : 1_937; + Transaction transaction = Transaction.newBuilder().setRawData(Transaction.raw.newBuilder() + .addContract(Transaction.Contract.newBuilder() + .setType(ContractType.TriggerSmartContract))) + .build(); + TransactionContext context = new TransactionContext(null, + new TransactionCapsule(transaction), null, false, false); + + try (MockedStatic precompiles = + mockStatic(PrecompiledContracts.class, CALLS_REAL_METHODS); + MockedConstruction actuators = mockConstruction(VMActuator.class, + withSettings().defaultAnswer(CALLS_REAL_METHODS), (actuator, construction) -> { + // Bypass DB-backed validation only; execute() and its real VM are retained. + doNothing().when(actuator).validate(any()); + setField(actuator, "program", program); + setField(actuator, "rootRepository", repository); + actuator.setTrxType(TrxType.TRX_CONTRACT_CALL_TYPE); + }); + MockedConstruction submissions = mockConstruction(FutureTask.class, + (future, construction) -> controlSubmission(future, construction, tasks))) { + precompiles.when(() -> PrecompiledContracts.getContractForAddress(address)) + .thenReturn(contract); + RuntimeImpl runtime = new RuntimeImpl(); + runtime.execute(context); + assertEquals(1, actuators.constructed().size()); + assertEquals(1, program.getPC()); + assertTrue(program.isStopped()); + assertEquals(Op.CALL, program.getLastOp() & 0xff); + // These checks prevent a test passing through malformed-input or insufficient-energy exits. + assertEquals(transferProof ? 3 : 1, tasks.futures.size()); + assertEquals(tasks.futures.size(), submissions.constructed().size()); + assertEquals(DataWord.ONE(), program.getStack().peek()); + ProgramResult result = runtime.getResult(); + boolean outOfTime = completion != Completion.ON_TIME && mode == NodeMode.ENERGY_ADJUSTMENT; + if (outOfTime) { + assertTrue(result.getException() instanceof Program.OutOfTimeException); + assertEquals(contractResult.OUT_OF_TIME, result.getResultCode()); + assertEquals(ENERGY_LIMIT, result.getEnergyUsed()); + assertTrue(result.getRuntimeError().contains("TX_LAST_OP")); + assertTrue(tasks.futures.stream().allMatch(Future::isCancelled)); + } else { + assertNull(result.getException()); + assertEquals(contractResult.SUCCESS, result.getResultCode()); + assertEquals(normalEnergy, result.getEnergyUsed()); + boolean dropped = !transferProof && completion == Completion.LATCH_TIMEOUT; + byte[] expected = new byte[transferProof ? 96 : 32]; + if (!dropped) { + expected[transferProof ? 31 : 0] = 1; + } + byte[] actual = program.getReturnDataBufferData(DataWord.ZERO(), + program.getReturnDataBufferSize()); + assertArrayEquals(expected, actual); + assertEquals(dropped, tasks.futures.get(0).isCancelled()); + } + if (completion == Completion.PUBLICATION_TIMEOUT && mode == NodeMode.ENERGY_ADJUSTMENT) { + assertEquals(1, tasks.timedReads); + assertEquals(0, tasks.unboundedReads); + } + } + } finally { + parameter.setDebug(previousDebug); + parameter.setSolidityNode(previousSolidity); + VMConfig.clearLocalSnapshot(); + } + } + + /** + * Intercept the default ThreadPoolExecutor's FutureTask construction on the submitting thread. + * Its mocked run() is a no-op on pool threads; result reads use the controlled task below. + */ + private static void controlSubmission(FutureTask future, + MockedConstruction.Context construction, ControlledTasks tasks) throws Exception { + Future result = tasks.submit((Callable) construction.arguments().get(0)); + doAnswer(ignored -> result.get()).when(future).get(); + doAnswer(call -> result.get(call.getArgument(0), call.getArgument(1))) + .when(future).get(anyLong(), any(TimeUnit.class)); + when(future.cancel(anyBoolean())).thenAnswer(call -> result.cancel(call.getArgument(0))); + when(future.isCancelled()).thenAnswer(ignored -> result.isCancelled()); + when(future.isDone()).thenAnswer(ignored -> result.isDone()); + } + + private static void setField(Object object, String name, Object value) throws Exception { + Field field = VMActuator.class.getDeclaredField(name); + field.setAccessible(true); + field.set(object, value); + } + + private static byte[] signatureInput() { + byte[] data = new byte[352]; + byte[] hash = new byte[32]; + hash[0] = 42; + ECKey key = ECKey.fromPrivate(new byte[]{1}); + byte[] signature = key.sign(hash).toByteArray(); + System.arraycopy(hash, 0, data, 0, 32); + putWord(data, 32, 96); + putWord(data, 64, 288); + putWord(data, 96, 1); + putWord(data, 128, 32); + putWord(data, 160, signature.length); + System.arraycopy(signature, 0, data, 192, signature.length); + putWord(data, 288, 1); + System.arraycopy(new DataWord(key.getAddress()).getData(), 0, data, 320, 32); + return data; + } + + private static byte[] transferInput() { + byte[] data = new byte[2080]; + putWord(data, 0, 1312); + putWord(data, 32, 1664); + putWord(data, 64, 1760); + putWord(data, 1312, 1); + putWord(data, 1664, 1); + putWord(data, 1760, 1); + return data; + } + + private static void putWord(byte[] data, int offset, int value) { + System.arraycopy(new DataWord(value).getData(), 0, data, offset, 32); + } + + /** Start the fixed invocation deadline on its first read, after setup and class loading. */ + private static final class Deadline { + private final long budgetUs; + private long deadlineUs; + + private Deadline(long budgetUs) { + this.budgetUs = budgetUs; + } + + private long get() { + if (deadlineUs == 0) { + deadlineUs = System.nanoTime() / 1_000 + budgetUs; + } + return deadlineUs; + } + + private void awaitExpiry() throws InterruptedException { + long remaining = (get() + 1) * 1_000 - System.nanoTime(); + while (remaining > 0) { + new CountDownLatch(1).await(remaining, TimeUnit.NANOSECONDS); + remaining = (get() + 1) * 1_000 - System.nanoTime(); + } + } + } + + /** + * Controlled worker scheduling: defer work past the latch deadline, or finish its callable (and + * count down the real latch) but hold Future publication. No pool replacement or fixed sleeps. + * Callables run on the test thread so Sapling primitives can be isolated from native proof setup. + */ + private static final class ControlledTasks { + private final Completion completion; + private final Deadline deadline; + private final List> futures = new ArrayList<>(); + private int timedReads; + private int unboundedReads; + + private ControlledTasks(Completion completion, Deadline deadline) { + this.completion = completion; + this.deadline = deadline; + } + + private Future submit(Callable task) { + ControlledFuture future = new ControlledFuture<>(task); + if (completion != Completion.LATCH_TIMEOUT) { + future.finishCallable(); + } + futures.add(future); + return future; + } + + private final class ControlledFuture implements Future { + private final Callable task; + private T value; + private Exception failure; + private boolean finished; + private boolean published; + private boolean cancelled; + + private ControlledFuture(Callable task) { + this.task = task; + } + + private void finishCallable() { + if (!finished) { + try { + value = task.call(); + } catch (Exception e) { + failure = e; + } + finished = true; + published = completion == Completion.ON_TIME; + } + } + + @Override + public T get() throws InterruptedException, ExecutionException { + unboundedReads++; + if (completion != Completion.ON_TIME) { + deadline.awaitExpiry(); + } + return publish(); + } + + @Override + public T get(long timeout, TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { + timedReads++; + if (completion != Completion.ON_TIME) { + new CountDownLatch(1).await(timeout, unit); + throw new TimeoutException("worker result held past the invocation deadline"); + } + return publish(); + } + + private T publish() throws ExecutionException { + if (cancelled) { + throw new CancellationException(); + } + finishCallable(); + if (failure != null) { + throw new ExecutionException(failure); + } + published = true; + return value; + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + if (isDone()) { + return false; + } + cancelled = true; + return true; + } + + @Override + public boolean isCancelled() { + return cancelled; + } + + @Override + public boolean isDone() { + return cancelled || published; + } + } + + } +}