diff --git a/docs/storage-lifecycle.md b/docs/storage-lifecycle.md index f225d3c9c1..74e403caf6 100644 --- a/docs/storage-lifecycle.md +++ b/docs/storage-lifecycle.md @@ -69,6 +69,15 @@ does not guarantee that it will return on Java thread interruption. The call rem registered until its read and iterator cleanup finish; cancellation does not permit another thread to close a native iterator that is still in use. +### Server OLTP traversal batches + +Server OLTP traversal workers close each active batch iterator on completion, +interruption or failure, then release their thread's graph transactions. A capacity +stop also releases producers waiting on a full queue. Iterator cleanup errors remain +visible when the capacity stop ends traversal normally; a business failure retains +its cleanup errors as suppressed exceptions. Caller transactions remain with the +request thread. + ### Callback executor configuration The gRPC application callback queue is always unbounded (`Integer.MAX_VALUE`), matching diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/OltpTraverser.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/OltpTraverser.java index 75e6ca6a03..1cd09cd566 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/OltpTraverser.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/OltpTraverser.java @@ -335,22 +335,32 @@ public EdgesConsumer(Consumer consumer, long capacity) { @Override public void accept(Iterator edgeIter) { - // Bind worker leases before a caller-created lazy query opens a session. - graph().tx().readWrite(); - Iterator ids = prepare(edgeIter); - long counter = 0; - while (ids.hasNext()) { - if (Thread.currentThread().isInterrupted()) { - LOG.warn("Consumer is Interrupted"); - break; + Iterator toClose = edgeIter; + Throwable failure = null; + try { + // Bind worker leases before a caller-created lazy query opens a session. + graph().tx().readWrite(); + Iterator ids = prepare(edgeIter); + toClose = ids != null ? ids : edgeIter; + long counter = 0; + while (ids.hasNext()) { + if (Thread.currentThread().isInterrupted()) { + LOG.warn("Consumer is Interrupted"); + break; + } + counter++; + this.consumer.accept(ids.next()); } - counter++; - this.consumer.accept(ids.next()); - } - long total = edgeIterCounter.addAndGet(counter); - // traverse by batch & improve performance - if (this.capacity != NO_LIMIT && total >= this.capacity) { - throw new Consumers.StopExecution("reach capacity"); + long total = edgeIterCounter.addAndGet(counter); + // traverse by batch & improve performance + if (this.capacity != NO_LIMIT && total >= this.capacity) { + throw new Consumers.StopExecution("reach capacity"); + } + } catch (RuntimeException | Error e) { + failure = e; + throw e; + } finally { + closeIterator(toClose, failure); } } } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java index ea6e77a1ec..914d4161fe 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java @@ -57,6 +57,7 @@ public final class Consumers { private final CountDownLatch latch; private final BlockingQueue queue; private volatile Throwable exception = null; + private volatile boolean stopped; public Consumers(ExecutorService executor, Consumer consumer) { this(executor, consumer, null); @@ -98,6 +99,7 @@ public Consumers(ExecutorService executor, public void start(String name) { this.exception = null; + this.stopped = false; if (this.executor == null) { return; } @@ -117,15 +119,15 @@ public void start(String name) { private Void safeRun(ContextCallable task) { try { return task.call(); // may fail before/after runAndDone() - } catch (Exception e) { + } catch (Throwable e) { // This exception is from ContextCallable wrapper (setContext/resetContext/delegate dispatch), // not from runAndDone() business logic (that one is handled inside runAndDone()). - if (this.exception == null) { - this.exception = e; + if (this.recordException(e)) { LOG.error("Consumer worker failed in ContextCallable wrapper", e); } else { LOG.warn("Additional worker failure in ContextCallable wrapper; first exception already recorded", e); } + this.stop(); this.exceptionHandle(e); } finally { this.latch.countDown(); @@ -136,18 +138,21 @@ private Void safeRun(ContextCallable task) { private Void runAndDone() { try { this.run(); - } catch (Exception e) { + } catch (Throwable e) { if (e instanceof StopExecution) { - this.queue.clear(); - putQueueEnd(); + Throwable[] suppressed = e.getSuppressed(); + // Stop stays control flow, but failed cleanup must reach the caller. + for (Throwable failure : suppressed) { + this.recordException(failure); + } } else { - if (this.exception == null) { - this.exception = e; + if (this.recordException(e)) { LOG.error("Unhandled exception in consumer task", e); } else { LOG.warn("Additional exception in consumer task; first exception already recorded", e); } } + this.stop(); this.exceptionHandle(e); } finally { this.done(); @@ -155,6 +160,11 @@ private Void runAndDone() { return null; } + private void stop() { + this.stopped = true; + this.queue.clear(); + } + private void run() { LOG.debug("Start to work..."); while (this.consume()) { @@ -167,6 +177,9 @@ private void run() { private boolean consume() { V elem = null; while (elem == null) { + if (this.stopped) { + return false; + } try { elem = this.queue.poll(CONSUMER_WAKE_PERIOD, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { @@ -175,6 +188,9 @@ private boolean consume() { } } + if (this.stopped) { + return false; + } if (elem == QUEUE_END) { putQueueEnd(); return false; @@ -192,9 +208,7 @@ private void exceptionHandle(Throwable e) { try { this.exceptionHandle.accept(e); } catch (Throwable ex) { - if (this.exception == null) { - this.exception = ex; - } else { + if (!this.recordException(ex)) { LOG.warn("Error while calling exceptionHandle()", ex); } } @@ -208,19 +222,32 @@ private void done() { try { this.doneHandle.run(); } catch (Throwable e) { - if (this.exception == null) { - this.exception = e; - } else { + if (!this.recordException(e)) { LOG.warn("Error while calling done()", e); } + this.stop(); } } + private synchronized boolean recordException(Throwable failure) { + if (this.exception == null) { + this.exception = failure; + return true; + } + if (this.exception != failure) { + for (Throwable suppressed : this.exception.getSuppressed()) { + if (suppressed == failure) { + return false; + } + } + this.exception.addSuppressed(failure); + } + return false; + } + private Throwable throwException() { assert this.exception != null; - Throwable e = this.exception; - this.exception = null; - return e; + return this.exception; } public void provide(V v) throws Throwable { @@ -232,7 +259,17 @@ public void provide(V v) throws Throwable { throw this.throwException(); } else { try { - this.queue.put(v); + while (!this.stopped && + !this.queue.offer(v, CONSUMER_WAKE_PERIOD, TimeUnit.MILLISECONDS)) { + // A stopped worker must not leave a producer blocked on a full queue. + } + if (this.stopped) { + this.queue.clear(); + if (this.exception != null) { + throw this.throwException(); + } + throw new StopExecution("Consumer stopped"); + } } catch (InterruptedException e) { LOG.warn("Interrupt while queuing QUEUE_END", e); } @@ -242,7 +279,10 @@ public void provide(V v) throws Throwable { private void putQueueEnd() { if (this.executor != null) { try { - this.queue.put(QUEUE_END); + while (!this.stopped && + !this.queue.offer(QUEUE_END, CONSUMER_WAKE_PERIOD, TimeUnit.MILLISECONDS)) { + // Stop wakes consumers through its state, without another END token. + } } catch (InterruptedException e) { LOG.warn("Interrupted while enqueue", e); } @@ -262,7 +302,8 @@ public void await() throws Throwable { for (Future f : this.runningFutures) { f.cancel(true); } - this.exception = new HugeException(error, e); + this.recordException(new HugeException(error, e)); + this.stop(); LOG.warn(error, e); } } diff --git a/hugegraph-server/hugegraph-test/pom.xml b/hugegraph-server/hugegraph-test/pom.xml index 8627174dd9..5e0e51fe8f 100644 --- a/hugegraph-server/hugegraph-test/pom.xml +++ b/hugegraph-server/hugegraph-test/pom.xml @@ -228,11 +228,6 @@ org.jacoco jacoco-maven-plugin - - - org/apache/hugegraph/traversal/algorithm/*.class - - pre-test diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java index 707b586e49..4f3ea5436c 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java @@ -38,6 +38,7 @@ @RunWith(Suite.class) @Suite.SuiteClasses({ BackendLeaseCleanupTest.class, + OltpTraverserCursorTest.class, OltpTraverserRocksDBTest.class, PropertyKeyCoreTest.class, VertexLabelCoreTest.class, diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java new file mode 100644 index 0000000000..3e53074f8e --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java @@ -0,0 +1,457 @@ +/* + * 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.hugegraph.core; + +import java.lang.reflect.Field; +import java.lang.reflect.Modifier; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Iterator; +import java.util.Set; +import java.util.UUID; +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.AtomicReference; + +import org.apache.commons.configuration2.MapConfiguration; +import org.apache.commons.io.FileUtils; +import org.apache.hugegraph.HugeFactory; +import org.apache.hugegraph.HugeGraph; +import org.apache.hugegraph.backend.query.QueryBatch; +import org.apache.hugegraph.backend.query.QueryResults; +import org.apache.hugegraph.backend.store.rocksdb.RocksDBSessions; +import org.apache.hugegraph.backend.tx.GraphTransaction; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.id.Id; +import org.apache.hugegraph.id.IdGenerator; +import org.apache.hugegraph.iterator.MapperIterator; +import org.apache.hugegraph.query.Query; +import org.apache.hugegraph.testutil.Utils; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.traversal.algorithm.OltpTraverser; +import org.apache.hugegraph.type.define.Directions; +import org.apache.hugegraph.util.Consumers; +import org.apache.tinkerpop.gremlin.structure.Edge; +import org.apache.tinkerpop.gremlin.structure.T; +import org.apache.tinkerpop.gremlin.structure.Vertex; +import org.apache.tinkerpop.gremlin.structure.util.CloseableIterator; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.Before; +import org.junit.Test; +import org.rocksdb.RocksIterator; + +public class OltpTraverserCursorTest { + + @Before + public void requireRocksDB() { + Assume.assumeTrue("Native cursor regression requires RocksDB", + "rocksdb".equals(Utils.getConf().getString("backend"))); + CoreTestSuite.initEnv(); + } + + @Test + public void testConsumerFailureClosesNativeCursor() throws Exception { + this.checkNativeCursor("consumer"); + } + + @Test + public void testStopExecutionClosesNativeCursor() throws Exception { + this.checkNativeCursor("stop"); + } + + @Test + public void testInterruptedConsumerClosesNativeCursor() throws Exception { + this.checkNativeCursor("interrupt"); + } + + @Test + public void testPrepareFailureClosesNativeCursor() throws Exception { + this.checkNativeCursor("prepare"); + } + + @Test + public void testCloseFailureIsSuppressedAfterNativeCursorCloses() throws Exception { + this.checkNativeCursor("close"); + } + + @Test + public void testAsyncCapacityStopClosesNativeCursor() throws Exception { + this.checkNativeCursor("async-capacity"); + } + + @Test + public void testAsyncCapacityStopPropagatesCloseFailure() throws Exception { + this.checkNativeCursor("async-capacity-close"); + } + + @Test + public void testAsyncCapacityStopReleasesBlockedProducer() throws Exception { + this.checkNativeCursor("async-capacity-producer"); + } + + @Test + public void testAsyncCapacityCloseFailureReleasesBlockedProducer() throws Exception { + this.checkNativeCursor("async-capacity-producer-close"); + } + + @Test + public void testAsyncConsumerFailureKeepsSuppressedCloseFailure() throws Exception { + this.checkNativeCursor("async-consumer-close"); + } + + @Test + public void testAsyncBusinessFailureReleasesBlockedProducer() throws Exception { + this.checkNativeCursor("async-consumer-producer-close"); + } + + @Test + public void testAsyncFatalErrorReleasesBlockedProducer() throws Exception { + this.checkNativeCursor("async-fatal-producer"); + } + + @Test + public void testCapacityStopEndsActualLazyTraversal() throws Exception { + this.checkNativeCursor("lazy-producer"); + } + + private void checkNativeCursor(String outcome) throws Exception { + Path directory = Files.createTempDirectory("oltp-cursor-"); + HugeConfig config = new HugeConfig(new MapConfiguration(Collections.emptyMap())); + config.setProperty("backend", "rocksdb"); + config.setProperty("serializer", "binary"); + config.setProperty("store", "cursor_" + UUID.randomUUID().toString().replace("-", "")); + config.setProperty("oltp.concurrent_threads", "lazy-producer".equals(outcome) ? 1 : 0); + config.setProperty("rocksdb.data_path", directory.toString()); + config.setProperty("rocksdb.wal_path", directory.toString()); + HugeGraph graph = null; + RocksDBSessions control = null; + Iterator edges = null; + RocksIterator nativeCursor = null; + int producerReads = -1; + try { + graph = HugeFactory.open(config); + graph.initBackend(); + graph.schema().propertyKey("seq").asInt().create(); + graph.schema().vertexLabel("vertex").useCustomizeNumberId().create(); + graph.schema().edgeLabel("linked").sourceLabel("vertex").targetLabel("vertex") + .multiTimes().properties("seq").sortKeys("seq").create(); + Vertex source = graph.addVertex(T.label, "vertex", T.id, 1L); + Vertex target = graph.addVertex(T.label, "vertex", T.id, 2L); + // The backend prefetches 500 edges; record 501 keeps its native cursor open. + for (int i = 0; i < 501; i++) { + source.addEdge("linked", target, "seq", i); + } + graph.tx().commit(); + Id label = graph.schema().getEdgeLabel("linked").id(); + Query query = GraphTransaction.constructEdgesQuery(IdGenerator.of(1L), Directions.OUT, + new Id[]{label}); + edges = graph.edges(query); + Assert.assertTrue(edges.hasNext()); + nativeCursor = nativeCursor(edges, Collections.newSetFromMap(new IdentityHashMap<>())); + Assert.assertNotNull("Expected the real RocksDB scan cursor", nativeCursor); + Assert.assertTrue(nativeCursor.isOwningHandle()); + Object store = graph.storeProvider().loadGraphStore(config); + RocksDBSessions sessions = Whitebox.getInternalState(store, "sessions"); + // Keep the native DB alive during handle observation after caller FINISHED cleanup. + control = sessions.copy(config, "cursor-control", "g"); + control.session(); + if ("lazy-producer".equals(outcome)) { + // This isolated JVM fixture owns the single-worker pool used below. + OltpTraverser.destroy(); + } + Probe traverser = new Probe(graph); + Throwable failure = outcome.contains("fatal") ? new AssertionError("fatal failure") : + "stop".equals(outcome) ? new Consumers.StopExecution("limit") : + new IllegalStateException("business failure"); + RuntimeException closeFailure = new IllegalStateException("close failure"); + if (outcome.endsWith("close")) { + edges = new MapperIterator(edges, edge -> edge) { + @Override + public void close() throws Exception { + super.close(); + throw closeFailure; + } + }; + } + Iterator input = edges; + OltpTraverser.EdgesConsumer consumer; + if ("prepare".equals(outcome)) { + consumer = traverser.new EdgesConsumer(id -> Assert.fail(), OltpTraverser.NO_LIMIT) { + @Override + protected Iterator prepare(Iterator iterator) { + if (failure instanceof Error) { + throw (Error) failure; + } + throw (RuntimeException) failure; + } + }; + } else { + consumer = traverser.new OneStepEdgeIterConsumer(id -> { + if (!outcome.startsWith("async-capacity") && !"lazy-producer".equals(outcome)) { + if (failure instanceof Error) { + throw (Error) failure; + } + throw (RuntimeException) failure; + } + }, outcome.startsWith("async-capacity") || "lazy-producer".equals(outcome) ? + 1L : OltpTraverser.NO_LIMIT); + } + if ("lazy-producer".equals(outcome)) { + producerReads = this.consumeLazyProducer(traverser, consumer, input); + Assert.assertEquals(501L, traverser.edgeIterCounter.get()); + } else if (outcome.startsWith("async-")) { + AtomicReference handled = new AtomicReference<>(); + Throwable thrown = this.consumeAsync(consumer, input, handled, outcome); + if (outcome.startsWith("async-capacity")) { + Assert.assertEquals(501L, traverser.edgeIterCounter.get()); + if (outcome.endsWith("close")) { + Assert.assertNotNull("Close failure must remain visible after capacity Stop", thrown); + Assert.assertFalse(thrown instanceof Consumers.StopExecution); + Assert.assertTrue(handled.get() instanceof Consumers.StopExecution); + Assert.assertSame(handled.get().getSuppressed()[0], thrown); + Assert.assertSame(closeFailure, thrown.getCause()); + } else { + Assert.assertNull(thrown); + Assert.assertTrue(handled.get() instanceof Consumers.StopExecution); + } + } else { + Assert.assertSame(failure, thrown); + Assert.assertSame(failure, handled.get()); + Assert.assertEquals(outcome.endsWith("close") ? 1 : 0, thrown.getSuppressed().length); + if (outcome.endsWith("close")) { + Assert.assertSame(closeFailure, thrown.getSuppressed()[0].getCause()); + } + } + } else if ("interrupt".equals(outcome)) { + Thread.currentThread().interrupt(); + try { + consumer.accept(input); + Assert.assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + } else { + RuntimeException thrown = org.apache.hugegraph.testutil.Assert.assertThrows( + RuntimeException.class, () -> consumer.accept(input)); + Assert.assertSame(failure, thrown); + if ("close".equals(outcome)) { + Assert.assertEquals(1, thrown.getSuppressed().length); + Assert.assertTrue(thrown.getSuppressed()[0].getCause().getMessage().contains("close failure")); + } + } + HugeFactory.closeCurrentThreadTransactions(); + Assert.assertTrue("Control lease must keep the DB alive", control.databaseOpened()); + Assert.assertFalse("Traversal retained its native cursor after " + outcome, + nativeCursor.isOwningHandle()); + if ("lazy-producer".equals(outcome)) { + Assert.assertEquals("Capacity Stop must end the lazy producer", 2, producerReads); + } + } finally { + // These are teardown only; the native handle assertion above must run first. + try { + if (nativeCursor != null && nativeCursor.isOwningHandle()) { + nativeCursor.close(); + } + if (edges != null && !outcome.endsWith("close")) { + CloseableIterator.closeIterator(edges); + } + } finally { + try { + HugeFactory.closeCurrentThreadTransactions(); + if (control != null) { + control.close(); + } + if (graph != null && !graph.closed()) { + graph.drop(); + } + } finally { + if (graph != null) { + HugeFactory.remove(graph); + } + FileUtils.deleteDirectory(directory.toFile()); + } + } + } + } + + private Throwable consumeAsync(OltpTraverser.EdgesConsumer consumer, + Iterator input, AtomicReference handled, + String outcome) throws Exception { + ExecutorService worker = Executors.newFixedThreadPool(1); + ExecutorService waiter = Executors.newSingleThreadExecutor(); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + boolean fullQueue = outcome.startsWith("async-capacity") || outcome.contains("producer"); + try { + Consumers> consumers = new Consumers<>(worker, iterator -> { + entered.countDown(); + try { + if (!release.await(5L, TimeUnit.SECONDS)) { + throw new AssertionError("Consumer was not released"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + consumer.accept(iterator); + }, HugeFactory::closeCurrentThreadTransactions, handled::set, 1); + consumers.start("cursor-close-failure"); + try { + consumers.provide(input); + Assert.assertTrue(entered.await(5L, TimeUnit.SECONDS)); + if (fullQueue) { + // One active input and two queued inputs fill the real queue. + consumers.provide(Collections.emptyIterator()); + consumers.provide(Collections.emptyIterator()); + } + } catch (Throwable failure) { + throw new AssertionError(failure); + } + CountDownLatch calling = new CountDownLatch(1); + Future result = waiter.submit(() -> { + calling.countDown(); + Throwable failure = null; + try { + if (outcome.contains("producer")) { + consumers.provide(Collections.emptyIterator()); + } + } catch (Consumers.StopExecution ignored) { + // Stop ends production successfully, as in Oltp's producer loop. + } catch (Throwable providedFailure) { + failure = providedFailure; + } + try { + consumers.await(); + } catch (Throwable awaitedFailure) { + if (failure == null) { + failure = awaitedFailure; + } + } + return failure; + }); + Assert.assertTrue(calling.await(5L, TimeUnit.SECONDS)); + if (fullQueue) { + // The producer or await END enqueue must wait on the full queue. + org.apache.hugegraph.testutil.Assert.assertThrows( + java.util.concurrent.TimeoutException.class, + () -> result.get(100L, TimeUnit.MILLISECONDS)); + } + release.countDown(); + return result.get(3L, TimeUnit.SECONDS); + } finally { + release.countDown(); + waiter.shutdownNow(); + worker.shutdownNow(); + // The pre-fix producer can swallow one interrupt before entering await(). + for (int i = 0; i < 50 && !waiter.awaitTermination(100L, TimeUnit.MILLISECONDS); i++) { + waiter.shutdownNow(); + } + Assert.assertTrue(waiter.isTerminated()); + Assert.assertTrue(worker.awaitTermination(5L, TimeUnit.SECONDS)); + } + } + + private int consumeLazyProducer(Probe traverser, OltpTraverser.EdgesConsumer consumer, + Iterator input) throws Exception { + Consumers.ExecutorPool pool = Whitebox.getInternalState(OltpTraverser.class, "executors"); + ExecutorService worker = pool.getExecutor(); + pool.returnExecutor(worker); + class LazyInputs implements Iterator>, AutoCloseable { + + private int reads; + private boolean closed; + + @Override + public boolean hasNext() { + if (this.reads == 1) { + try { + // Runs after the real Consumers task and its worker transaction cleanup. + worker.submit(() -> { }).get(5L, TimeUnit.SECONDS); + } catch (Exception e) { + throw new AssertionError(e); + } + } + return this.reads < 100; + } + + @Override + public Iterator next() { + if (this.reads >= 100) { + throw new java.util.NoSuchElementException(); + } + return this.reads++ == 0 ? input : Collections.emptyIterator(); + } + + @Override + public void close() { + this.closed = true; + } + } + LazyInputs source = new LazyInputs(); + try { + long total = traverser.traverseLazy(source, consumer); + Assert.assertEquals((long) source.reads, total); + Assert.assertTrue("Traversal must close its lazy producer", source.closed); + return source.reads; + } finally { + OltpTraverser.destroy(); + Assert.assertTrue(worker.awaitTermination(5L, TimeUnit.SECONDS)); + } + } + + private static RocksIterator nativeCursor(Object value, Set visited) throws Exception { + if (value instanceof RocksIterator) { + return (RocksIterator) value; + } + if (!(value instanceof Iterator || value instanceof QueryResults || value instanceof QueryBatch) || + !visited.add(value)) { + return null; + } + for (Class type = value.getClass(); type != null && + type.getName().startsWith("org.apache.hugegraph."); type = type.getSuperclass()) { + for (Field field : type.getDeclaredFields()) { + if (Modifier.isStatic(field.getModifiers())) { + continue; + } + field.setAccessible(true); + RocksIterator found = nativeCursor(field.get(value), visited); + if (found != null) { + return found; + } + } + } + return null; + } + + private static class Probe extends OltpTraverser { + + Probe(HugeGraph graph) { + super(graph); + } + + long traverseLazy(Iterator> source, java.util.function.Consumer> consumer) { + return this.traverseByOne(source, consumer, "native-lazy-producer-stop"); + } + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java index fb06e21545..b83e083a77 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java @@ -114,6 +114,7 @@ import org.apache.hugegraph.unit.store.RamIntObjectMapTest; import org.apache.hugegraph.unit.traversal.ShortestPathTraverserTest; import org.apache.hugegraph.unit.util.CompressUtilTest; +import org.apache.hugegraph.unit.util.ConsumersTest; import org.apache.hugegraph.unit.util.JsonUtilTest; import org.apache.hugegraph.unit.util.RateLimiterTest; import org.apache.hugegraph.unit.util.StringEncodingTest; @@ -248,6 +249,7 @@ RocksDBTableQueryByIdsTest.class, /* utils */ + ConsumersTest.class, VersionTest.class, JsonUtilTest.class, StringEncodingTest.class, diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java index 1bb152ec75..9db801be19 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java @@ -17,8 +17,13 @@ package org.apache.hugegraph.unit.util; +import java.util.Arrays; +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.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import org.apache.hugegraph.testutil.Assert; @@ -50,33 +55,31 @@ public void testStartProvideAwaitNormal() throws Throwable { } } - /** - * Regression test for deadlock: - * - * ContextCallable fails before entering runAndDone(). - * await() must still return because latch is decremented in safeRun(). - */ @Test(timeout = 1000) - public void testAwaitDoesNotHangWhenContextCallableFails() throws Throwable { + public void testAwaitPreservesFatalConsumerError() throws Throwable { ExecutorService executor = Executors.newFixedThreadPool(1); + CountDownLatch release = new CountDownLatch(1); + AssertionError failure = new AssertionError("fatal consumer failure"); try { - // Use AssertionError to bypass the inner catch(Exception) loop in runAndDone() - // This simulates a scenario where an exception escapes the task logic - // (similar to how a ContextCallable failure would behave from safeRun's perspective) Consumers consumers = new Consumers<>(executor, v -> { - throw new AssertionError("Simulated fatal error (OOM/StackOverflow/etc)"); + try { + if (!release.await(5L, TimeUnit.SECONDS)) { + throw new AssertionError("Consumer was not released"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + throw failure; }); consumers.start("test-fatal-error"); consumers.provide(1); - // Verification: - // Without the fix, the latch would never be decremented (because runAndDone crashes), causing await() to hang. - // With the fix (safeRun wrapper), the finally block ensures latch.countDown() is called. - consumers.await(); - - // Note: consumer.exception will be null because safeRun only catches Exception, not Error. - // This is acceptable behavior for fatal errors, as long as it doesn't deadlock. + release.countDown(); + Assert.assertSame(failure, Assert.assertThrows(AssertionError.class, consumers::await)); } finally { + release.countDown(); executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(5L, TimeUnit.SECONDS)); } } @@ -108,4 +111,267 @@ public void testAwaitThrowsWhenConsumerThrows() throws Throwable { executor.shutdownNow(); } } + + @Test + public void testStopWithFullQueueAllowsAwait() throws Exception { + this.checkStopWithFullQueue(false, false); + } + + @Test + public void testStopWithFullQueueReleasesProducer() throws Exception { + this.checkStopWithFullQueue(true, false); + } + + @Test + public void testStopCleanupFailureWithFullQueueAllowsAwait() throws Exception { + this.checkStopWithFullQueue(false, true); + } + + @Test + public void testStopCleanupFailureWithFullQueueReleasesProducer() throws Exception { + this.checkStopWithFullQueue(true, true); + } + + @Test + public void testBusinessFailureWithFullQueueReleasesProducer() throws Exception { + this.checkTerminalWithFullQueue(true, false, new IllegalStateException("business failure")); + } + + @Test + public void testFatalErrorWithFullQueueReleasesProducer() throws Exception { + this.checkTerminalWithFullQueue(true, false, new AssertionError("fatal failure")); + } + + private void checkStopWithFullQueue(boolean blockedProducer, boolean cleanupFailure) throws Exception { + this.checkTerminalWithFullQueue(blockedProducer, cleanupFailure, null); + } + + private void checkTerminalWithFullQueue(boolean blockedProducer, boolean cleanupFailure, + Throwable terminalFailure) throws Exception { + ExecutorService worker = Executors.newFixedThreadPool(1); + ExecutorService caller = Executors.newSingleThreadExecutor(); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger processed = new AtomicInteger(); + AtomicInteger done = new AtomicInteger(); + CountDownLatch finished = new CountDownLatch(1); + RuntimeException closeFailure = new IllegalStateException("cleanup failure"); + try { + Consumers consumers = new Consumers<>(worker, value -> { + processed.incrementAndGet(); + entered.countDown(); + try { + if (!release.await(5L, TimeUnit.SECONDS)) { + throw new AssertionError("Consumer was not released"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + if (terminalFailure instanceof Error) { + throw (Error) terminalFailure; + } else if (terminalFailure != null) { + throw (RuntimeException) terminalFailure; + } + Consumers.StopExecution stop = new Consumers.StopExecution("capacity"); + if (cleanupFailure) { + stop.addSuppressed(closeFailure); + } + throw stop; + }, () -> { + done.incrementAndGet(); + finished.countDown(); + }, 1); + consumers.start("full-queue-stop"); + try { + consumers.provide(1); + Assert.assertTrue(entered.await(5L, TimeUnit.SECONDS)); + consumers.provide(2); + consumers.provide(3); + } catch (Throwable failure) { + throw new AssertionError(failure); + } + CountDownLatch calling = new CountDownLatch(1); + Future result = caller.submit(() -> { + calling.countDown(); + try { + if (blockedProducer) { + consumers.provide(4); + } + consumers.await(); + return null; + } catch (Throwable failure) { + return failure; + } + }); + Assert.assertTrue(calling.await(5L, TimeUnit.SECONDS)); + Assert.assertThrows(TimeoutException.class, () -> result.get(100L, TimeUnit.MILLISECONDS)); + release.countDown(); + Throwable returned = result.get(3L, TimeUnit.SECONDS); + if (blockedProducer && !cleanupFailure && terminalFailure == null) { + Assert.assertTrue(returned instanceof Consumers.StopExecution); + } else { + Assert.assertSame(cleanupFailure ? closeFailure : terminalFailure, returned); + } + Assert.assertTrue(finished.await(5L, TimeUnit.SECONDS)); + if (blockedProducer && (cleanupFailure || terminalFailure != null)) { + // A failure reported by provide() must not leave await() blocked. + Future joined = caller.submit(() -> { + try { + consumers.await(); + return null; + } catch (Throwable failure) { + return failure; + } + }); + Assert.assertSame(cleanupFailure ? closeFailure : terminalFailure, + joined.get(3L, TimeUnit.SECONDS)); + } + Assert.assertEquals(1, processed.get()); + Assert.assertEquals(1, done.get()); + } finally { + release.countDown(); + caller.shutdownNow(); + worker.shutdownNow(); + for (int i = 0; i < 50 && !caller.awaitTermination(100L, TimeUnit.MILLISECONDS); i++) { + caller.shutdownNow(); + } + Assert.assertTrue(caller.isTerminated()); + Assert.assertTrue(worker.awaitTermination(5L, TimeUnit.SECONDS)); + } + } + @Test + public void testProducerFailureRemainsPrimaryAfterCleanupFailure() throws Exception { + this.checkProducerFailureAfterCleanup(1); + } + + @Test + public void testConcurrentCleanupFailuresRemainSuppressed() throws Exception { + this.checkProducerFailureAfterCleanup(2); + } + + private void checkProducerFailureAfterCleanup(int workers) throws Exception { + ExecutorService worker = Executors.newFixedThreadPool(workers); + ExecutorService caller = Executors.newSingleThreadExecutor(); + CountDownLatch entered = new CountDownLatch(workers); + CountDownLatch releaseFailure = new CountDownLatch(1); + CountDownLatch observed = new CountDownLatch(1); + AtomicInteger done = new AtomicInteger(); + RuntimeException business = new IllegalStateException("business failure"); + RuntimeException[] cleanup = new RuntimeException[workers]; + for (int i = 0; i < workers; i++) { + cleanup[i] = new IllegalStateException("cleanup failure " + i); + } + try { + Consumers consumers = new Consumers<>(worker, value -> { + entered.countDown(); + awaitGate(releaseFailure); + throw business; + }, () -> { + int index = done.getAndIncrement(); + // Cleanup fails strictly after the blocked producer observes business. + awaitGate(observed); + throw cleanup[index]; + }, 1); + consumers.start("primary-cleanup-race"); + try { + for (int i = 0; i < workers; i++) { + consumers.provide(i); + } + Assert.assertTrue(entered.await(5L, TimeUnit.SECONDS)); + for (int i = 0; i < workers + 1; i++) { + consumers.provide(workers + i); + } + } catch (Throwable failure) { + throw new AssertionError(failure); + } + CountDownLatch calling = new CountDownLatch(1); + Future provided = caller.submit(() -> { + calling.countDown(); + try { + consumers.provide(2 * workers + 1); + return null; + } catch (Throwable failure) { + return failure; + } + }); + Assert.assertTrue(calling.await(5L, TimeUnit.SECONDS)); + Assert.assertThrows(TimeoutException.class, () -> provided.get(100L, TimeUnit.MILLISECONDS)); + releaseFailure.countDown(); + Assert.assertSame(business, provided.get(3L, TimeUnit.SECONDS)); + observed.countDown(); + Future awaited = caller.submit(() -> { + try { + consumers.await(); + return null; + } catch (Throwable failure) { + return failure; + } + }); + Assert.assertSame(business, awaited.get(3L, TimeUnit.SECONDS)); + Assert.assertEquals(workers, business.getSuppressed().length); + for (RuntimeException failure : cleanup) { + Assert.assertTrue(Arrays.asList(business.getSuppressed()).contains(failure)); + } + Assert.assertEquals(workers, done.get()); + } finally { + releaseFailure.countDown(); + observed.countDown(); + caller.shutdownNow(); + worker.shutdownNow(); + Assert.assertTrue(caller.awaitTermination(5L, TimeUnit.SECONDS)); + Assert.assertTrue(worker.awaitTermination(5L, TimeUnit.SECONDS)); + } + } + + @Test + public void testCleanStopEndsLazyProducer() throws Exception { + ExecutorService worker = Executors.newFixedThreadPool(1); + CountDownLatch stopped = new CountDownLatch(1); + AtomicInteger reads = new AtomicInteger(); + AtomicInteger processed = new AtomicInteger(); + try { + Consumers consumers = new Consumers<>(worker, value -> { + processed.incrementAndGet(); + throw new Consumers.StopExecution("capacity"); + }, stopped::countDown, 1); + consumers.start("lazy-producer-stop"); + try { + for (int i = 0; i < 100; i++) { + if (i == 1) { + Assert.assertTrue(stopped.await(5L, TimeUnit.SECONDS)); + } + reads.incrementAndGet(); + consumers.provide(i); + } + } catch (Consumers.StopExecution ignored) { + // Same successful producer termination contract as Oltp traversal. + } catch (Throwable failure) { + throw new AssertionError(failure); + } finally { + try { + consumers.await(); + } catch (Throwable failure) { + throw new AssertionError(failure); + } + } + Assert.assertEquals(2, reads.get()); + Assert.assertEquals(1, processed.get()); + } finally { + worker.shutdownNow(); + Assert.assertTrue(worker.awaitTermination(5L, TimeUnit.SECONDS)); + } + } + + private static void awaitGate(CountDownLatch gate) { + try { + if (!gate.await(5L, TimeUnit.SECONDS)) { + throw new AssertionError("Gate was not released"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + }