This is an automated email from the ASF dual-hosted git repository. spmallette pushed a commit to branch tinkergraph-storage in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit e17b7025f8156c343b6895a4f386226b24fcc23e Author: Stephen Mallette <[email protected]> AuthorDate: Mon Aug 3 20:53:41 2026 +0000 Serialize concurrent commit writes to TinkerStorageGraph storage TinkerGraph transactions lock only their own changed elements, so commits touching disjoint elements ran their commit paths concurrently and both wrote to the storage engine's single append log at once, interleaving and corrupting its records. A fair per-graph commit-write lock now serializes the engine's persist/flush (and the compact/close paths that rewrite the same files), held only around the durable write so disjoint commits still proceed in parallel up to that point. Assisted-by: Claude Code:claude-opus-4-8 --- .../tinkergraph/structure/AbstractTinkerGraph.java | 23 ++++- .../tinkergraph/structure/TinkerStorageGraph.java | 11 ++- .../tinkergraph/structure/TinkerTransaction.java | 12 ++- .../tinkergraph/structure/storage/SyncMode.java | 9 +- .../AbstractTinkerStorageConformanceTest.java | 53 +++++++++++ .../structure/storage/ConcurrencyProbeStorage.java | 82 ++++++++++++++++ .../storage/StorageCommitSerializationTest.java | 105 +++++++++++++++++++++ 7 files changed, 285 insertions(+), 10 deletions(-) diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/AbstractTinkerGraph.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/AbstractTinkerGraph.java index ac1a3e6474..bb663ad0bd 100644 --- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/AbstractTinkerGraph.java +++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/AbstractTinkerGraph.java @@ -45,6 +45,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; /** * Base class for {@link TinkerMemoryGraph} and {@link TinkerStorageGraph}. @@ -84,6 +85,15 @@ public abstract class AbstractTinkerGraph implements TinkerGraph { */ protected TinkerStorage storage; + /** + * Serializes the durable write of a committing transaction. TinkerGraph transactions lock only their own changed + * elements, so two commits touching disjoint elements run their commit paths concurrently; without this lock they + * would both write to the storage engine's single append log at once and interleave (corrupt) its records. Held + * only around the engine's persist/flush, so commits of disjoint elements still proceed in parallel up to that + * point. Fair, so committers are served in arrival order and none is starved. + */ + protected final ReentrantLock storageCommitLock = new ReentrantLock(true); + /** * Guard set while a graph is replaying its storage log on open. While {@code true}, mutations must not be * re-persisted, otherwise replay would append the loaded data back to the log. @@ -308,9 +318,16 @@ public abstract class AbstractTinkerGraph implements TinkerGraph { @Override public void close() { if (storage != null) { - storage.flush(); - storage.compact(this); - storage.close(); + // serialize against concurrent commit writes: close flushes, compacts, and closes the log, which must not + // interleave with a transaction appending to it. + storageCommitLock.lock(); + try { + storage.flush(); + storage.compact(this); + storage.close(); + } finally { + storageCommitLock.unlock(); + } } serviceRegistry.close(); GqlDeclarativeMatchStrategy.evict(this); diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerStorageGraph.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerStorageGraph.java index 7dd54500a7..409f618f25 100644 --- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerStorageGraph.java +++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerStorageGraph.java @@ -303,8 +303,15 @@ public final class TinkerStorageGraph extends AbstractTinkerGraph { */ public void compact() { if (storage != null) { - storage.flush(); - storage.compact(this); + // hold the same lock as the commit write path: compaction closes the log, rewrites the snapshot, and + // truncates the log, which must not interleave with a concurrent transaction appending to that log. + storageCommitLock.lock(); + try { + storage.flush(); + storage.compact(this); + } finally { + storageCommitLock.unlock(); + } } } diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerTransaction.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerTransaction.java index 7bef8a9264..50c48bde80 100644 --- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerTransaction.java +++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerTransaction.java @@ -179,10 +179,16 @@ final class TinkerTransaction extends AbstractThreadLocalTransaction { // write-ahead: durably persist the changeset before applying the in-memory commit, so a failure here // aborts the commit (via the catch below) and leaves memory and disk consistent. Skipped while the graph - // is replaying its storage log on open. + // is replaying its storage log on open. Serialized by storageCommitLock because commits of disjoint + // elements otherwise reach the engine's single append log concurrently and interleave its records. if (graph.storage != null && !graph.loading) { - graph.storage.persist(txVersion, toVertexMutations(changedVertices), toEdgeMutations(changedEdges)); - graph.storage.flush(); + graph.storageCommitLock.lock(); + try { + graph.storage.persist(txVersion, toVertexMutations(changedVertices), toEdgeMutations(changedEdges)); + graph.storage.flush(); + } finally { + graph.storageCommitLock.unlock(); + } } // commit all changes diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/SyncMode.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/SyncMode.java index de049797fe..d83e2d00a9 100644 --- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/SyncMode.java +++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/SyncMode.java @@ -46,8 +46,13 @@ public enum SyncMode { // TODO: add an INTERVAL mode (group commit) — a peer value on this same key, encoded as "interval:<ms>", that // fsyncs at most once per <ms> window rather than once per commit, bounding the crash-loss window by time while // amortizing fsync cost across commits. It implies fsync (a batched COMMIT), so it slots in as a third - // mutually-exclusive mode without changing the meaning of COMMIT or OS. Deferred until concurrent commits are - // serialized, since group commit's payoff is amortizing one fsync across many concurrent committers. + // mutually-exclusive mode without changing the meaning of COMMIT or OS. + // + // This is coupled to the commit-write lock that serializes concurrent commits (see the persist/flush step in + // TinkerTransaction.doCommit): once that lock exists, COMMIT holds it across the fsync, so every commit serializes + // on disk-sync latency. INTERVAL is the fix — hold the lock only for the buffer append (fast) and fsync one batch + // for many transactions. So INTERVAL should be built on top of that lock, not before it: it is the performance + // pass that makes serialized commits cheap, which is why it is deferred until concurrent commits are serialized. /** * Resolve a configuration value to a {@link SyncMode}, matched case-insensitively, defaulting to {@link #COMMIT} diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/AbstractTinkerStorageConformanceTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/AbstractTinkerStorageConformanceTest.java index 561320bcff..552cd37da4 100644 --- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/AbstractTinkerStorageConformanceTest.java +++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/AbstractTinkerStorageConformanceTest.java @@ -32,7 +32,14 @@ import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; +import java.util.ArrayList; import java.util.Iterator; +import java.util.List; +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 static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -231,6 +238,52 @@ public abstract class AbstractTinkerStorageConformanceTest { graph.close(); } + @Test + public void shouldPersistConcurrentCommitsWithoutLossAcrossReopen() throws Exception { + // End-to-end companion to StorageCommitSerializationTest: many threads commit disjoint vertices at once and, + // on reopen, every record must survive. This exercises the real engine but cannot by itself *prove* the lock + // works — log corruption from interleaving is scheduling-dependent — so the deterministic guarantee is + // asserted separately by StorageCommitSerializationTest via a probe engine. + final int threads = 8; + final int commitsPerThread = 50; + final TinkerStorageGraph writeGraph = open(); + try { + final ExecutorService pool = Executors.newFixedThreadPool(threads); + final CountDownLatch start = new CountDownLatch(1); + final List<Future<?>> futures = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + final int threadId = t; + futures.add(pool.submit(() -> { + start.await(); // release all threads together to maximize contention on the commit path + for (int i = 0; i < commitsPerThread; i++) { + final int id = threadId * commitsPerThread + i; + writeGraph.addVertex(T.id, id, "value", id); + writeGraph.tx().commit(); + } + return null; + })); + } + start.countDown(); + for (final Future<?> f : futures) + f.get(60, TimeUnit.SECONDS); + pool.shutdown(); + assertTrue(pool.awaitTermination(60, TimeUnit.SECONDS)); + } finally { + writeGraph.close(); + } + + // reopen from disk: a corrupt (interleaved) log frame would throw or drop records here + final TinkerStorageGraph reopened = open(); + try { + final int expected = threads * commitsPerThread; + assertEquals(expected, countOf(reopened.vertices())); + for (int id = 0; id < expected; id++) + assertEquals(Integer.valueOf(id), reopened.vertices(id).next().value("value")); + } finally { + reopened.close(); + } + } + private static long countOf(final Iterator<?> it) { long count = 0; while (it.hasNext()) { diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/ConcurrencyProbeStorage.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/ConcurrencyProbeStorage.java new file mode 100644 index 0000000000..21b6c014da --- /dev/null +++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/ConcurrencyProbeStorage.java @@ -0,0 +1,82 @@ +/* + * 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.tinkerpop.gremlin.tinkergraph.structure.storage; + +import org.apache.commons.configuration2.Configuration; +import org.apache.tinkerpop.gremlin.tinkergraph.structure.AbstractTinkerGraph; +import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerEdge; +import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertex; + +import java.util.Collection; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * A {@link TinkerStorage} test double that deterministically detects whether the commit path serializes writes. It + * persists nothing; instead {@link #persist} tracks how many threads are inside it at once and records a violation if + * ever more than one is. It also sleeps briefly while "inside" to widen the window, so an unserialized commit path + * trips the detector reliably rather than racily. Selected by fully-qualified class name via + * {@code gremlin.tinkergraph.storage}. State is static because the engine is instantiated reflectively. + */ +public final class ConcurrencyProbeStorage implements TinkerStorage { + + static final AtomicInteger inFlight = new AtomicInteger(0); + static final AtomicInteger maxObserved = new AtomicInteger(0); + static final AtomicInteger concurrentEntries = new AtomicInteger(0); + + static void reset() { + inFlight.set(0); + maxObserved.set(0); + concurrentEntries.set(0); + } + + @Override + public void open(final AbstractTinkerGraph graph, final Configuration config) { } + + @Override + public void replay(final AbstractTinkerGraph graph) { } + + @Override + public void persist(final long txVersion, + final Collection<TinkerStorageMutation<TinkerVertex>> changedVertices, + final Collection<TinkerStorageMutation<TinkerEdge>> changedEdges) { + final int concurrent = inFlight.incrementAndGet(); + try { + maxObserved.accumulateAndGet(concurrent, Math::max); + if (concurrent > 1) + concurrentEntries.incrementAndGet(); + // widen the window so an unserialized path is caught deterministically, not by luck + try { + Thread.sleep(1); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } + } finally { + inFlight.decrementAndGet(); + } + } + + @Override + public void flush() { } + + @Override + public void compact(final AbstractTinkerGraph graph) { } + + @Override + public void close() { } +} diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/StorageCommitSerializationTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/StorageCommitSerializationTest.java new file mode 100644 index 0000000000..2a0d5dd174 --- /dev/null +++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/StorageCommitSerializationTest.java @@ -0,0 +1,105 @@ +/* + * 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.tinkerpop.gremlin.tinkergraph.structure.storage; + +import org.apache.commons.configuration2.BaseConfiguration; +import org.apache.commons.configuration2.Configuration; +import org.apache.tinkerpop.gremlin.structure.Graph; +import org.apache.tinkerpop.gremlin.structure.T; +import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph; +import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerStorageGraph; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.util.ArrayList; +import java.util.List; +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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** + * Deterministically verifies that {@link TinkerStorageGraph} serializes the durable write of concurrent commits. + * Because TinkerGraph transactions lock only their own changed elements, commits touching disjoint elements run their + * commit paths concurrently; the storage engine's single append log would interleave (corrupt) without the + * commit-write lock. Rather than rely on a race actually corrupting the log, this drives commits through + * {@link ConcurrencyProbeStorage}, which records whether two threads are ever inside {@code persist()} at once. With + * the lock that count is exactly zero; without it the probe trips reliably. + */ +public class StorageCommitSerializationTest { + + @Rule + public TemporaryFolder tempFolder = new TemporaryFolder(); + + private String location; + + @Before + public void setUp() throws Exception { + location = tempFolder.newFolder("storage").getAbsolutePath(); + ConcurrencyProbeStorage.reset(); + } + + private TinkerStorageGraph open() { + final Configuration conf = new BaseConfiguration(); + conf.setProperty(Graph.GRAPH, TinkerStorageGraph.class.getName()); + conf.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_STORAGE, ConcurrencyProbeStorage.class.getName()); + conf.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION, location); + return TinkerStorageGraph.open(conf); + } + + @Test + public void shouldSerializeConcurrentCommitWrites() throws Exception { + final int threads = 8; + final int commitsPerThread = 20; + final TinkerStorageGraph graph = open(); + try { + final ExecutorService pool = Executors.newFixedThreadPool(threads); + final CountDownLatch start = new CountDownLatch(1); + final List<Future<?>> futures = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + final int threadId = t; + futures.add(pool.submit(() -> { + start.await(); + for (int i = 0; i < commitsPerThread; i++) { + graph.addVertex(T.id, threadId * commitsPerThread + i); + graph.tx().commit(); + } + return null; + })); + } + start.countDown(); + for (final Future<?> f : futures) + f.get(60, TimeUnit.SECONDS); + pool.shutdown(); + assertTrue(pool.awaitTermination(60, TimeUnit.SECONDS)); + } finally { + graph.close(); + } + + // the lock must have kept persist() strictly single-threaded + assertEquals("commits entered persist() concurrently", 0, ConcurrencyProbeStorage.concurrentEntries.get()); + assertEquals("more than one thread was inside persist() at once", 1, ConcurrencyProbeStorage.maxObserved.get()); + } +}
