You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2026/08/20 18:29:40 UTC
(tinkerpop) 01/09: Extract AbstractLogStorage base from GraphBinaryStorage
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 ce004eec0b48246f02c70dda3c9fd0e5d79dc856
Author: Stephen Mallette <st...@amazon.com>
AuthorDate: Wed Aug 19 14:16:14 2026 +0000
Extract AbstractLogStorage base from GraphBinaryStorage
Move the codec-agnostic log-structured machinery — file layout, VERSION marker,
CRC framing, replay fold, SyncMode durability, and crash-safe + threshold
compaction — into a new abstract AbstractLogStorage. GraphBinaryStorage now
supplies only the element codec through four hooks (encodeCommit, decodeFrame,
writeSnapshot, beginReplay). Pure refactor: on-disk format and behavior are
unchanged, so a new codec can be built by overriding the hooks alone.
Assisted-by: Claude Code:claude-opus-4-8
---
...hBinaryStorage.java => AbstractLogStorage.java} | 442 +++++++-----------
.../structure/storage/GraphBinaryStorage.java | 512 ++-------------------
2 files changed, 202 insertions(+), 752 deletions(-)
diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/AbstractLogStorage.java
similarity index 63%
copy from tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java
copy to tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/AbstractLogStorage.java
index 2aaa7643cc..b2bdf249f5 100644
--- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java
+++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/AbstractLogStorage.java
@@ -19,20 +19,15 @@
package org.apache.tinkerpop.gremlin.tinkergraph.structure.storage;
import org.apache.commons.configuration2.Configuration;
-import org.apache.tinkerpop.gremlin.structure.Edge;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.io.binary.GraphBinaryReader;
-import org.apache.tinkerpop.gremlin.structure.io.binary.GraphBinaryWriter;
-import org.apache.tinkerpop.gremlin.structure.io.binary.TypeSerializerRegistry;
import org.apache.tinkerpop.gremlin.structure.util.Attachable;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.AbstractTinkerGraph;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerEdge;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertex;
+import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
@@ -42,7 +37,6 @@ import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
-import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.nio.channels.FileChannel;
import java.nio.file.AtomicMoveNotSupportedException;
@@ -51,40 +45,24 @@ import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.zip.CRC32;
/**
- * A durable {@link TinkerStorage} engine that persists a {@code TinkerStorageGraph} as an append-only commit log
- * ("write-ahead log") serialized with GraphBinary. On each committed transaction the changeset is appended to
- * {@code log.gbin} as a single record; on open the optional {@code snapshot.gbin} is read followed by the log, with the
- * folded result re-applied to the in-memory graph. {@link #compact(AbstractTinkerGraph)} rewrites the snapshot from the
- * current committed state and truncates the log.
+ * Log-structured durable storage machinery shared by {@link TinkerStorage} engines, independent of how an element is
+ * encoded. It persists a {@code TinkerStorageGraph} as an append-only commit log ({@code log.gbin}) plus an optional
+ * folded {@code snapshot.gbin}; on open the snapshot is read followed by the log, last-write-wins per element id.
* <p/>
- * On commit the appended record is made durable according to the configured {@link SyncMode}: {@link SyncMode#COMMIT}
- * (default) {@code fsync}s so the commit survives an OS crash or power loss, while {@link SyncMode#OS} only flushes to
- * the operating system.
+ * This base owns everything that is not the element codec: the on-disk file layout, the single-source-of-truth
+ * {@code VERSION} marker, length+CRC frame framing (which tells an interrupted trailing append apart from genuine
+ * corruption), the replay fold loop, durability via {@link SyncMode}, and crash-safe atomic compaction with a
+ * size threshold. Concrete engines supply only the codec through {@link #encodeCommit}, {@link #decodeFrame},
+ * {@link #writeSnapshot}, and (optionally) {@link #beginReplay} for per-replay decode state.
* <p/>
- * On-disk compatibility: the store's format version is recorded once in a {@code VERSION} marker file (magic +
- * version), which is the single source of truth even when a store momentarily holds a snapshot and a log written at
- * different times. Individual files carry only the magic for identity/corruption detection. Opening a store whose
- * marker names an unsupported version fails loudly — records are never misread across a format change. The engine
- * keeps this simple by not attempting in-place migration: the supported path across an incompatible format bump is to
- * export via {@code g.io()} before upgrading. Additive changes should prefer new record opcodes; unknown opcodes are
- * a hard error (a durable store must not silently drop records it cannot parse), so a genuinely incompatible change
- * bumps the version.
- * <p/>
- * The in-memory graph remains authoritative (write-through). This engine does not support graphs larger than memory.
- * <p/>
- * Known limitation (write amplification): a commit records each changed element in full — a single property change on
- * a large element rewrites the whole element to the log. Elements are typically small and automatic compaction bounds
- * the resulting log growth, so this is accepted rather than mitigated with per-property deltas, which would complicate
- * the {@link TinkerStorageMutation} contract and the replay fold. The snapshot, by contrast, is streamed one element
- * at a time so compaction never holds a second full copy of the graph in heap.
+ * The in-memory graph remains authoritative (write-through); this machinery does not support graphs larger than memory.
*/
-public final class GraphBinaryStorage implements TinkerStorage {
+public abstract class AbstractLogStorage implements TinkerStorage {
/**
* Magic bytes ("TGSB" — TinkerGraph Storage Binary) at the start of every storage file, so a file can be
@@ -101,8 +79,7 @@ public final class GraphBinaryStorage implements TinkerStorage {
static final byte FORMAT_VERSION = 1;
/**
- * Store-level version marker file, holding {@link #MAGIC} followed by the one-byte {@link #FORMAT_VERSION}. It is
- * the single source of truth for the store's format version.
+ * Store-level version marker file, holding {@link #MAGIC} followed by the one-byte {@link #FORMAT_VERSION}.
*/
static final String VERSION_FILE = "VERSION";
@@ -112,27 +89,19 @@ public final class GraphBinaryStorage implements TinkerStorage {
*/
static final int HEADER_SIZE = MAGIC.length;
- private static final byte OP_PUT_VERTEX = 1;
- private static final byte OP_DEL_VERTEX = 2;
- private static final byte OP_PUT_EDGE = 3;
- private static final byte OP_DEL_EDGE = 4;
-
static final String SNAPSHOT_FILE = "snapshot.gbin";
static final String LOG_FILE = "log.gbin";
- private final GraphBinaryWriter writer = new GraphBinaryWriter(TypeSerializerRegistry.INSTANCE);
- private final GraphBinaryReader reader = new GraphBinaryReader(TypeSerializerRegistry.INSTANCE);
+ /**
+ * Default automatic-compaction threshold: 64 MB of appended log since the last compaction.
+ */
+ static final long DEFAULT_COMPACT_THRESHOLD_BYTES = 64L * 1024 * 1024;
private File directory;
private File snapshotFile;
private File logFile;
private File versionFile;
- /**
- * Default automatic-compaction threshold: 64 MB of appended log since the last compaction.
- */
- static final long DEFAULT_COMPACT_THRESHOLD_BYTES = 64L * 1024 * 1024;
-
private DataOutputStream logOut;
private FileOutputStream logFos;
private SyncMode syncMode = SyncMode.COMMIT;
@@ -140,11 +109,42 @@ public final class GraphBinaryStorage implements TinkerStorage {
private long logBytesSinceCompaction = 0;
private boolean closed = false;
+ // ----------------------------------------------------------------------------------------- codec hooks
+
+ /**
+ * Encode a committing transaction's changeset into a single record payload (the framing is added by the caller).
+ */
+ protected abstract byte[] encodeCommit(long txVersion,
+ Collection<TinkerStorageMutation<TinkerVertex>> changedVertices,
+ Collection<TinkerStorageMutation<TinkerEdge>> changedEdges) throws IOException;
+
+ /**
+ * Decode one record payload, folding its puts and deletes into the supplied maps (last-write-wins per id).
+ */
+ protected abstract void decodeFrame(byte[] record,
+ Map<Object, DetachedVertex> vertices,
+ Map<Object, DetachedEdge> edges) throws IOException;
+
+ /**
+ * Write the entire current committed state of the graph to {@code out} as framed records (via {@link #writeFrame}),
+ * for compaction. The fixed {@link #MAGIC} header has already been written to {@code out}.
+ */
+ protected abstract void writeSnapshot(AbstractTinkerGraph graph, DataOutputStream out) throws IOException;
+
+ /**
+ * Reset any per-replay decode state (e.g. a dictionary) before a fold begins. Default is a no-op.
+ */
+ protected void beginReplay() {
+ // no-op by default
+ }
+
+ // ----------------------------------------------------------------------------------------- lifecycle
+
@Override
public void open(final AbstractTinkerGraph graph, final Configuration config) {
final String location = config.getString(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION, null);
if (null == location)
- throw new IllegalStateException(String.format("%s must be set to use the GraphBinary storage engine",
+ throw new IllegalStateException(String.format("%s must be set to use a durable storage engine",
TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION));
this.directory = new File(location);
this.snapshotFile = new File(directory, SNAPSHOT_FILE);
@@ -159,68 +159,9 @@ public final class GraphBinaryStorage implements TinkerStorage {
establishStoreVersion();
}
- /**
- * Read and validate the store-level version marker, or create it for a new store. This is the single source of
- * truth for the store's format version: a marker naming an unsupported version, or bad magic, fails the open
- * loudly rather than risking a misread. The supported path across an incompatible format bump is to export the
- * graph via {@code g.io()} before upgrading.
- */
- private void establishStoreVersion() {
- // an existing store (has a snapshot or log) written before the marker existed is treated as version 1
- final boolean storeHasData = snapshotFile.exists() || logFile.exists();
- if (!versionFile.exists()) {
- if (storeHasData && FORMAT_VERSION != 1)
- throw new IllegalStateException(String.format(
- "Storage location %s has data but no version marker; cannot confirm it is format version %d",
- directory, FORMAT_VERSION));
- writeStoreVersion();
- return;
- }
- try (final DataInputStream in = new DataInputStream(new java.io.BufferedInputStream(new FileInputStream(versionFile)))) {
- final byte[] magic = new byte[MAGIC.length];
- readFully(in, magic);
- if (!Arrays.equals(magic, MAGIC))
- throw new IOException(String.format("%s is not a TinkerGraph storage version marker (bad magic)", versionFile));
- final byte version = in.readByte();
- if (version != FORMAT_VERSION)
- throw new IOException(String.format(
- "Unsupported storage format version %d at %s (this build writes %d); export via g.io() before upgrading",
- version, directory, FORMAT_VERSION));
- } catch (IOException ex) {
- throw new UncheckedIOException(String.format("Could not read storage version marker %s", versionFile), ex);
- }
- }
-
- /**
- * Write the store-level version marker ({@link #MAGIC} + {@link #FORMAT_VERSION}) durably.
- */
- private void writeStoreVersion() {
- try (final FileOutputStream fos = new FileOutputStream(versionFile);
- final DataOutputStream out = new DataOutputStream(fos)) {
- out.write(MAGIC);
- out.writeByte(FORMAT_VERSION);
- out.flush();
- fos.getFD().sync();
- } catch (IOException ex) {
- throw new UncheckedIOException(String.format("Could not write storage version marker %s", versionFile), ex);
- }
- }
-
- /**
- * Ensure the backing directory exists, creating it if necessary. Called on open and again before writing a
- * snapshot, since {@code close()} may be invoked more than once and the directory may have been removed in between.
- */
- private void ensureDirectory() {
- if (directory.exists()) {
- if (!directory.isDirectory())
- throw new IllegalStateException(String.format("Storage location %s exists but is not a directory", directory));
- } else if (!directory.mkdirs()) {
- throw new IllegalStateException(String.format("Could not create storage directory %s", directory));
- }
- }
-
@Override
public void replay(final AbstractTinkerGraph graph) {
+ beginReplay();
// Fold snapshot then log into final state: last write per id wins, deletes remove.
final Map<Object, DetachedVertex> vertices = new LinkedHashMap<>();
final Map<Object, DetachedEdge> edges = new LinkedHashMap<>();
@@ -242,84 +183,13 @@ public final class GraphBinaryStorage implements TinkerStorage {
graph.tx().commit();
}
- /**
- * Read every record in a file, folding puts and deletes into the supplied maps.
- */
- private void foldRecords(final File file, final Map<Object, DetachedVertex> vertices, final Map<Object, DetachedEdge> edges) {
- final long fileLength = file.length();
- try (final DataInputStream in = new DataInputStream(new java.io.BufferedInputStream(new FileInputStream(file)))) {
- long remaining = readAndVerifyHeader(in, file, fileLength);
- while (true) {
- final byte[] record = readFrame(in, remaining);
- if (record == null)
- break;
- // account for the header (length + crc) and payload just consumed
- remaining -= 2L * Integer.BYTES + record.length;
- applyRecord(record, vertices, edges);
- }
- } catch (IOException ex) {
- throw new UncheckedIOException(String.format("Could not read storage file %s", file), ex);
- }
- }
-
- /**
- * Read and validate the per-file header (magic only), returning the number of record bytes that follow it. An
- * empty file (freshly created, no header yet) is treated as having no records. The store's format version is
- * validated once against the {@link #VERSION_FILE} marker in {@link #open}, not here.
- */
- private long readAndVerifyHeader(final DataInputStream in, final File file, final long fileLength) throws IOException {
- if (fileLength == 0)
- return 0;
- if (fileLength < HEADER_SIZE)
- throw new IOException(String.format("Corrupt storage file %s: shorter than its %d-byte header", file, HEADER_SIZE));
- final byte[] magic = new byte[MAGIC.length];
- readFully(in, magic);
- if (!Arrays.equals(magic, MAGIC))
- throw new IOException(String.format("%s is not a TinkerGraph storage file (bad magic)", file));
- return fileLength - HEADER_SIZE;
- }
-
- private void applyRecord(final byte[] record, final Map<Object, DetachedVertex> vertices, final Map<Object, DetachedEdge> edges) throws IOException {
- // the format version is validated once per file in readAndVerifyHeader, so records no longer repeat it
- final ByteBufferBuffer buffer = new ByteBufferBuffer(record);
- buffer.readLong(); // txVersion, retained for diagnostics/future use
- final int entryCount = buffer.readInt();
- for (int i = 0; i < entryCount; i++) {
- final byte op = buffer.readByte();
- switch (op) {
- case OP_PUT_VERTEX: {
- final Vertex v = reader.read(buffer);
- vertices.put(v.id(), (DetachedVertex) v);
- break;
- }
- case OP_DEL_VERTEX: {
- final Object id = reader.read(buffer);
- vertices.remove(id);
- break;
- }
- case OP_PUT_EDGE: {
- final Edge e = reader.read(buffer);
- edges.put(e.id(), (DetachedEdge) e);
- break;
- }
- case OP_DEL_EDGE: {
- final Object id = reader.read(buffer);
- edges.remove(id);
- break;
- }
- default:
- throw new IOException("Unknown storage op code: " + op);
- }
- }
- }
-
@Override
public void persist(final long txVersion,
final Collection<TinkerStorageMutation<TinkerVertex>> changedVertices,
final Collection<TinkerStorageMutation<TinkerEdge>> changedEdges) {
ensureLogOpen();
try {
- final byte[] frame = encodeRecord(txVersion, changedVertices, changedEdges);
+ final byte[] frame = encodeCommit(txVersion, changedVertices, changedEdges);
writeFrame(logOut, frame);
logBytesSinceCompaction += 2L * Integer.BYTES + frame.length; // length + crc prefixes + payload
} catch (IOException ex) {
@@ -327,39 +197,6 @@ public final class GraphBinaryStorage implements TinkerStorage {
}
}
- /**
- * Serialize a commit record: txVersion, entry count, then each entry as an op byte followed by either the
- * serialized element (put) or the serialized id (delete). The format version lives in the file header, not the
- * record.
- */
- private byte[] encodeRecord(final long txVersion,
- final Collection<TinkerStorageMutation<TinkerVertex>> changedVertices,
- final Collection<TinkerStorageMutation<TinkerEdge>> changedEdges) throws IOException {
- final ByteBufferBuffer buffer = new ByteBufferBuffer();
- buffer.writeLong(txVersion);
- buffer.writeInt(changedVertices.size() + changedEdges.size());
- for (final TinkerStorageMutation<TinkerVertex> m : changedVertices) {
- if (m.isDeleted()) {
- buffer.writeByte(OP_DEL_VERTEX);
- writer.write(m.id(), buffer);
- } else {
- buffer.writeByte(OP_PUT_VERTEX);
- // detach to a stable form independent of the transactional element
- writer.write(DetachedFactory.detach(m.element(), true), buffer);
- }
- }
- for (final TinkerStorageMutation<TinkerEdge> m : changedEdges) {
- if (m.isDeleted()) {
- buffer.writeByte(OP_DEL_EDGE);
- writer.write(m.id(), buffer);
- } else {
- buffer.writeByte(OP_PUT_EDGE);
- writer.write(DetachedFactory.detach(m.element(), true), buffer);
- }
- }
- return buffer.toWrittenArray();
- }
-
@Override
public void flush() {
if (closed)
@@ -428,66 +265,96 @@ public final class GraphBinaryStorage implements TinkerStorage {
compact(graph);
}
+ @Override
+ public void close() {
+ closeLog();
+ closed = true;
+ }
+
+ // ----------------------------------------------------------------------------------------- version marker
+
/**
- * Atomically move {@code source} onto {@code target}, replacing any existing target. Falls back to a non-atomic
- * replacing move on filesystems that do not support atomic moves.
+ * Read and validate the store-level version marker, or create it for a new store. This is the single source of
+ * truth for the store's format version: a marker naming an unsupported version, or bad magic, fails the open
+ * loudly rather than risking a misread.
*/
- private static void atomicMove(final File source, final File target) throws IOException {
- try {
- Files.move(source.toPath(), target.toPath(),
- StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
- } catch (AtomicMoveNotSupportedException anse) {
- Files.move(source.toPath(), target.toPath(), StandardCopyOption.REPLACE_EXISTING);
+ private void establishStoreVersion() {
+ final boolean storeHasData = snapshotFile.exists() || logFile.exists();
+ if (!versionFile.exists()) {
+ if (storeHasData && FORMAT_VERSION != 1)
+ throw new IllegalStateException(String.format(
+ "Storage location %s has data but no version marker; cannot confirm it is format version %d",
+ directory, FORMAT_VERSION));
+ writeStoreVersion();
+ return;
+ }
+ try (final DataInputStream in = new DataInputStream(new BufferedInputStream(new FileInputStream(versionFile)))) {
+ final byte[] magic = new byte[MAGIC.length];
+ readFully(in, magic);
+ if (!Arrays.equals(magic, MAGIC))
+ throw new IOException(String.format("%s is not a TinkerGraph storage version marker (bad magic)", versionFile));
+ final byte version = in.readByte();
+ if (version != FORMAT_VERSION)
+ throw new IOException(String.format(
+ "Unsupported storage format version %d at %s (this build writes %d); export via g.io() before upgrading",
+ version, directory, FORMAT_VERSION));
+ } catch (IOException ex) {
+ throw new UncheckedIOException(String.format("Could not read storage version marker %s", versionFile), ex);
}
}
- /**
- * fsync the storage directory so that recent namespace changes (a rename into place, a file deletion) are durable.
- * A directory fsync is required because those operations only update the directory entry, which the earlier file
- * fsync does not cover.
- */
- private void syncDirectory() {
- try (final FileChannel dirChannel = FileChannel.open(directory.toPath(), StandardOpenOption.READ)) {
- dirChannel.force(true);
+ private void writeStoreVersion() {
+ try (final FileOutputStream fos = new FileOutputStream(versionFile);
+ final DataOutputStream out = new DataOutputStream(fos)) {
+ out.write(MAGIC);
+ out.writeByte(FORMAT_VERSION);
+ out.flush();
+ fos.getFD().sync();
} catch (IOException ex) {
- // some platforms (notably Windows) cannot open a directory as a channel; the atomic rename is the
- // durability guarantee there, so treat inability to sync the directory as non-fatal
+ throw new UncheckedIOException(String.format("Could not write storage version marker %s", versionFile), ex);
}
}
- /**
- * Write the entire current committed state of the graph to {@code out} as a stream of single-element put records,
- * one frame per vertex and per edge. Each frame is an ordinary put record (see {@link #encodeRecord}) with an
- * entry count of one, so {@link #foldRecords} reconstructs the graph from these frames exactly as it would from a
- * commit log. Writing one element at a time keeps peak memory bounded to a single element rather than materializing
- * the whole graph as one byte array, so a snapshot never needs to hold a second full copy of the graph in heap.
- */
- private void writeSnapshot(final AbstractTinkerGraph graph, final DataOutputStream out) throws IOException {
- final Iterator<Vertex> vertexIterator = graph.vertices();
- while (vertexIterator.hasNext())
- writeElementFrame(out, OP_PUT_VERTEX, vertexIterator.next());
- final Iterator<Edge> edgeIterator = graph.edges();
- while (edgeIterator.hasNext())
- writeElementFrame(out, OP_PUT_EDGE, edgeIterator.next());
+ private void ensureDirectory() {
+ if (directory.exists()) {
+ if (!directory.isDirectory())
+ throw new IllegalStateException(String.format("Storage location %s exists but is not a directory", directory));
+ } else if (!directory.mkdirs()) {
+ throw new IllegalStateException(String.format("Could not create storage directory %s", directory));
+ }
}
- /**
- * Encode a single element as a one-entry put record and write it as a framed record to {@code out}. Only one
- * element's bytes are held in memory at a time.
- */
- private void writeElementFrame(final DataOutputStream out, final byte op, final Object element) throws IOException {
- final ByteBufferBuffer buffer = new ByteBufferBuffer();
- buffer.writeLong(0L); // snapshot records have no single tx version
- buffer.writeInt(1);
- buffer.writeByte(op);
- writer.write(DetachedFactory.detach(element, true), buffer);
- writeFrame(out, buffer.toWrittenArray());
+ // ----------------------------------------------------------------------------------------- fold / framing
+
+ private void foldRecords(final File file, final Map<Object, DetachedVertex> vertices, final Map<Object, DetachedEdge> edges) {
+ final long fileLength = file.length();
+ try (final DataInputStream in = new DataInputStream(new BufferedInputStream(new FileInputStream(file)))) {
+ long remaining = readAndVerifyHeader(in, file, fileLength);
+ while (true) {
+ final byte[] record = readFrame(in, remaining);
+ if (record == null)
+ break;
+ remaining -= 2L * Integer.BYTES + record.length;
+ decodeFrame(record, vertices, edges);
+ }
+ } catch (IOException ex) {
+ throw new UncheckedIOException(String.format("Could not read storage file %s", file), ex);
+ }
}
- @Override
- public void close() {
- closeLog();
- closed = true;
+ /**
+ * Read and validate the per-file header (magic only), returning the number of record bytes that follow it.
+ */
+ private long readAndVerifyHeader(final DataInputStream in, final File file, final long fileLength) throws IOException {
+ if (fileLength == 0)
+ return 0;
+ if (fileLength < HEADER_SIZE)
+ throw new IOException(String.format("Corrupt storage file %s: shorter than its %d-byte header", file, HEADER_SIZE));
+ final byte[] magic = new byte[MAGIC.length];
+ readFully(in, magic);
+ if (!Arrays.equals(magic, MAGIC))
+ throw new IOException(String.format("%s is not a TinkerGraph storage file (bad magic)", file));
+ return fileLength - HEADER_SIZE;
}
private void ensureLogOpen() {
@@ -505,14 +372,6 @@ public final class GraphBinaryStorage implements TinkerStorage {
}
}
- /**
- * Write the per-file header ({@link #MAGIC}) at the start of a storage file. The format version is recorded once
- * per store in the {@link #VERSION_FILE} marker, not per file.
- */
- private static void writeHeader(final DataOutputStream out) throws IOException {
- out.write(MAGIC);
- }
-
private void closeLog() {
if (logOut != null) {
try {
@@ -527,12 +386,19 @@ public final class GraphBinaryStorage implements TinkerStorage {
}
}
+ /**
+ * Write the per-file header ({@link #MAGIC}) at the start of a storage file.
+ */
+ private static void writeHeader(final DataOutputStream out) throws IOException {
+ out.write(MAGIC);
+ }
+
/**
* Write a framed record: a 4-byte big-endian payload length, a 4-byte CRC32 of the payload, then the payload.
* The checksum lets a reader tell a bit-flip inside a complete frame (corruption) from a short final frame left
- * by an interrupted append (truncation).
+ * by an interrupted append (truncation). Available to codec subclasses writing per-element snapshot frames.
*/
- private static void writeFrame(final DataOutputStream out, final byte[] payload) throws IOException {
+ protected static void writeFrame(final DataOutputStream out, final byte[] payload) throws IOException {
final CRC32 crc = new CRC32();
crc.update(payload);
out.writeInt(payload.length);
@@ -541,14 +407,9 @@ public final class GraphBinaryStorage implements TinkerStorage {
}
/**
- * Read a framed record, or return {@code null} at end of the readable log. A frame that is only partially present
- * — the file ends inside the header or payload — is treated as an interrupted trailing append (truncation) and
- * ends reading so earlier committed records still load. A frame that is fully present but whose stored CRC does
- * not match its payload is genuine corruption and is raised, rather than silently dropping it and everything
- * after it.
- *
- * @param remaining bytes left in the file at the current position; used to distinguish a short trailing frame
- * (truncation) from a complete frame, and to bound the payload allocation against a garbage length
+ * Read a framed record, or return {@code null} at end of the readable log. A frame only partially present is
+ * treated as an interrupted trailing append (truncation) and ends reading; a fully-present frame whose stored CRC
+ * does not match is genuine corruption and is raised.
*/
private static byte[] readFrame(final DataInputStream in, final long remaining) throws IOException {
if (remaining == 0)
@@ -588,4 +449,29 @@ public final class GraphBinaryStorage implements TinkerStorage {
off += read;
}
}
+
+ /**
+ * Atomically move {@code source} onto {@code target}, replacing any existing target. Falls back to a non-atomic
+ * replacing move on filesystems that do not support atomic moves.
+ */
+ private static void atomicMove(final File source, final File target) throws IOException {
+ try {
+ Files.move(source.toPath(), target.toPath(),
+ StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
+ } catch (AtomicMoveNotSupportedException anse) {
+ Files.move(source.toPath(), target.toPath(), StandardCopyOption.REPLACE_EXISTING);
+ }
+ }
+
+ /**
+ * fsync the storage directory so that recent namespace changes (a rename into place, a file deletion) are durable.
+ */
+ private void syncDirectory() {
+ try (final FileChannel dirChannel = FileChannel.open(directory.toPath(), StandardOpenOption.READ)) {
+ dirChannel.force(true);
+ } catch (IOException ex) {
+ // some platforms (notably Windows) cannot open a directory as a channel; the atomic rename is the
+ // durability guarantee there, so treat inability to sync the directory as non-fatal
+ }
+ }
}
diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java
index 2aaa7643cc..a5f3fd1c5c 100644
--- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java
+++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java
@@ -18,65 +18,29 @@
*/
package org.apache.tinkerpop.gremlin.tinkergraph.structure.storage;
-import org.apache.commons.configuration2.Configuration;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.io.binary.GraphBinaryReader;
import org.apache.tinkerpop.gremlin.structure.io.binary.GraphBinaryWriter;
import org.apache.tinkerpop.gremlin.structure.io.binary.TypeSerializerRegistry;
-import org.apache.tinkerpop.gremlin.structure.util.Attachable;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.AbstractTinkerGraph;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerEdge;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertex;
-import java.io.BufferedOutputStream;
-import java.io.DataInputStream;
import java.io.DataOutputStream;
-import java.io.EOFException;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.io.UncheckedIOException;
-import java.nio.channels.FileChannel;
-import java.nio.file.AtomicMoveNotSupportedException;
-import java.nio.file.Files;
-import java.nio.file.StandardCopyOption;
-import java.nio.file.StandardOpenOption;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
-import java.util.LinkedHashMap;
import java.util.Map;
-import java.util.zip.CRC32;
/**
- * A durable {@link TinkerStorage} engine that persists a {@code TinkerStorageGraph} as an append-only commit log
- * ("write-ahead log") serialized with GraphBinary. On each committed transaction the changeset is appended to
- * {@code log.gbin} as a single record; on open the optional {@code snapshot.gbin} is read followed by the log, with the
- * folded result re-applied to the in-memory graph. {@link #compact(AbstractTinkerGraph)} rewrites the snapshot from the
- * current committed state and truncates the log.
- * <p/>
- * On commit the appended record is made durable according to the configured {@link SyncMode}: {@link SyncMode#COMMIT}
- * (default) {@code fsync}s so the commit survives an OS crash or power loss, while {@link SyncMode#OS} only flushes to
- * the operating system.
- * <p/>
- * On-disk compatibility: the store's format version is recorded once in a {@code VERSION} marker file (magic +
- * version), which is the single source of truth even when a store momentarily holds a snapshot and a log written at
- * different times. Individual files carry only the magic for identity/corruption detection. Opening a store whose
- * marker names an unsupported version fails loudly — records are never misread across a format change. The engine
- * keeps this simple by not attempting in-place migration: the supported path across an incompatible format bump is to
- * export via {@code g.io()} before upgrading. Additive changes should prefer new record opcodes; unknown opcodes are
- * a hard error (a durable store must not silently drop records it cannot parse), so a genuinely incompatible change
- * bumps the version.
- * <p/>
- * The in-memory graph remains authoritative (write-through). This engine does not support graphs larger than memory.
+ * The GraphBinary {@link TinkerStorage} codec on top of {@link AbstractLogStorage}. The base owns the durable
+ * log-structured machinery (file layout, {@code VERSION} marker, CRC framing, replay fold, {@link SyncMode}
+ * durability, crash-safe and threshold compaction); this class supplies only how an element is encoded and decoded,
+ * using the GraphBinary serializers ({@link GraphBinaryWriter}/{@link GraphBinaryReader}).
* <p/>
* Known limitation (write amplification): a commit records each changed element in full — a single property change on
* a large element rewrites the whole element to the log. Elements are typically small and automatic compaction bounds
@@ -84,203 +48,53 @@ import java.util.zip.CRC32;
* the {@link TinkerStorageMutation} contract and the replay fold. The snapshot, by contrast, is streamed one element
* at a time so compaction never holds a second full copy of the graph in heap.
*/
-public final class GraphBinaryStorage implements TinkerStorage {
-
- /**
- * Magic bytes ("TGSB" — TinkerGraph Storage Binary) at the start of every storage file, so a file can be
- * identified as one written by this engine (and a foreign or corrupt file rejected) before any record is read.
- */
- static final byte[] MAGIC = { 'T', 'G', 'S', 'B' };
-
- /**
- * On-disk format version of the store. Recorded once per store in the {@link #VERSION_FILE} marker rather than in
- * every file, so a store has a single unambiguous version even when it momentarily holds a snapshot and a log
- * written at different times. A future format bump is detected against this marker so an older store is rejected
- * (never silently misread); the supported migration path is to export via {@code g.io()} before upgrading.
- */
- static final byte FORMAT_VERSION = 1;
-
- /**
- * Store-level version marker file, holding {@link #MAGIC} followed by the one-byte {@link #FORMAT_VERSION}. It is
- * the single source of truth for the store's format version.
- */
- static final String VERSION_FILE = "VERSION";
-
- /**
- * Bytes of the per-file header: just {@link #MAGIC}. The format version lives in the store-level
- * {@link #VERSION_FILE}, not in each file.
- */
- static final int HEADER_SIZE = MAGIC.length;
+public final class GraphBinaryStorage extends AbstractLogStorage {
private static final byte OP_PUT_VERTEX = 1;
private static final byte OP_DEL_VERTEX = 2;
private static final byte OP_PUT_EDGE = 3;
private static final byte OP_DEL_EDGE = 4;
- static final String SNAPSHOT_FILE = "snapshot.gbin";
- static final String LOG_FILE = "log.gbin";
-
private final GraphBinaryWriter writer = new GraphBinaryWriter(TypeSerializerRegistry.INSTANCE);
private final GraphBinaryReader reader = new GraphBinaryReader(TypeSerializerRegistry.INSTANCE);
- private File directory;
- private File snapshotFile;
- private File logFile;
- private File versionFile;
-
/**
- * Default automatic-compaction threshold: 64 MB of appended log since the last compaction.
+ * Serialize a commit record: txVersion, entry count, then each entry as an op byte followed by either the
+ * serialized element (put) or the serialized id (delete).
*/
- static final long DEFAULT_COMPACT_THRESHOLD_BYTES = 64L * 1024 * 1024;
-
- private DataOutputStream logOut;
- private FileOutputStream logFos;
- private SyncMode syncMode = SyncMode.COMMIT;
- private long compactThresholdBytes = DEFAULT_COMPACT_THRESHOLD_BYTES;
- private long logBytesSinceCompaction = 0;
- private boolean closed = false;
-
@Override
- public void open(final AbstractTinkerGraph graph, final Configuration config) {
- final String location = config.getString(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION, null);
- if (null == location)
- throw new IllegalStateException(String.format("%s must be set to use the GraphBinary storage engine",
- TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION));
- this.directory = new File(location);
- this.snapshotFile = new File(directory, SNAPSHOT_FILE);
- this.logFile = new File(directory, LOG_FILE);
- this.versionFile = new File(directory, VERSION_FILE);
- this.syncMode = SyncMode.fromConfigValue(config.getString(TinkerGraph.GREMLIN_TINKERGRAPH_STORAGE_SYNC, null));
- this.compactThresholdBytes = config.getLong(
- TinkerGraph.GREMLIN_TINKERGRAPH_STORAGE_COMPACT_THRESHOLD, DEFAULT_COMPACT_THRESHOLD_BYTES);
- // seed the counter with any pre-existing log so a graph reopened with a large log still compacts promptly
- this.logBytesSinceCompaction = logFile.exists() ? logFile.length() : 0;
- ensureDirectory();
- establishStoreVersion();
- }
-
- /**
- * Read and validate the store-level version marker, or create it for a new store. This is the single source of
- * truth for the store's format version: a marker naming an unsupported version, or bad magic, fails the open
- * loudly rather than risking a misread. The supported path across an incompatible format bump is to export the
- * graph via {@code g.io()} before upgrading.
- */
- private void establishStoreVersion() {
- // an existing store (has a snapshot or log) written before the marker existed is treated as version 1
- final boolean storeHasData = snapshotFile.exists() || logFile.exists();
- if (!versionFile.exists()) {
- if (storeHasData && FORMAT_VERSION != 1)
- throw new IllegalStateException(String.format(
- "Storage location %s has data but no version marker; cannot confirm it is format version %d",
- directory, FORMAT_VERSION));
- writeStoreVersion();
- return;
- }
- try (final DataInputStream in = new DataInputStream(new java.io.BufferedInputStream(new FileInputStream(versionFile)))) {
- final byte[] magic = new byte[MAGIC.length];
- readFully(in, magic);
- if (!Arrays.equals(magic, MAGIC))
- throw new IOException(String.format("%s is not a TinkerGraph storage version marker (bad magic)", versionFile));
- final byte version = in.readByte();
- if (version != FORMAT_VERSION)
- throw new IOException(String.format(
- "Unsupported storage format version %d at %s (this build writes %d); export via g.io() before upgrading",
- version, directory, FORMAT_VERSION));
- } catch (IOException ex) {
- throw new UncheckedIOException(String.format("Could not read storage version marker %s", versionFile), ex);
- }
- }
-
- /**
- * Write the store-level version marker ({@link #MAGIC} + {@link #FORMAT_VERSION}) durably.
- */
- private void writeStoreVersion() {
- try (final FileOutputStream fos = new FileOutputStream(versionFile);
- final DataOutputStream out = new DataOutputStream(fos)) {
- out.write(MAGIC);
- out.writeByte(FORMAT_VERSION);
- out.flush();
- fos.getFD().sync();
- } catch (IOException ex) {
- throw new UncheckedIOException(String.format("Could not write storage version marker %s", versionFile), ex);
- }
- }
-
- /**
- * Ensure the backing directory exists, creating it if necessary. Called on open and again before writing a
- * snapshot, since {@code close()} may be invoked more than once and the directory may have been removed in between.
- */
- private void ensureDirectory() {
- if (directory.exists()) {
- if (!directory.isDirectory())
- throw new IllegalStateException(String.format("Storage location %s exists but is not a directory", directory));
- } else if (!directory.mkdirs()) {
- throw new IllegalStateException(String.format("Could not create storage directory %s", directory));
+ protected byte[] encodeCommit(final long txVersion,
+ final Collection<TinkerStorageMutation<TinkerVertex>> changedVertices,
+ final Collection<TinkerStorageMutation<TinkerEdge>> changedEdges) throws IOException {
+ final ByteBufferBuffer buffer = new ByteBufferBuffer();
+ buffer.writeLong(txVersion);
+ buffer.writeInt(changedVertices.size() + changedEdges.size());
+ for (final TinkerStorageMutation<TinkerVertex> m : changedVertices) {
+ if (m.isDeleted()) {
+ buffer.writeByte(OP_DEL_VERTEX);
+ writer.write(m.id(), buffer);
+ } else {
+ buffer.writeByte(OP_PUT_VERTEX);
+ // detach to a stable form independent of the transactional element
+ writer.write(DetachedFactory.detach(m.element(), true), buffer);
+ }
}
- }
-
- @Override
- public void replay(final AbstractTinkerGraph graph) {
- // Fold snapshot then log into final state: last write per id wins, deletes remove.
- final Map<Object, DetachedVertex> vertices = new LinkedHashMap<>();
- final Map<Object, DetachedEdge> edges = new LinkedHashMap<>();
-
- if (snapshotFile.exists())
- foldRecords(snapshotFile, vertices, edges);
- if (logFile.exists())
- foldRecords(logFile, vertices, edges);
-
- if (vertices.isEmpty() && edges.isEmpty())
- return;
-
- // Attach vertices first so edges can find their endpoints, then commit once.
- for (final DetachedVertex v : vertices.values())
- v.attach(Attachable.Method.getOrCreate(graph));
- for (final DetachedEdge e : edges.values())
- e.attach(Attachable.Method.getOrCreate(graph));
-
- graph.tx().commit();
- }
-
- /**
- * Read every record in a file, folding puts and deletes into the supplied maps.
- */
- private void foldRecords(final File file, final Map<Object, DetachedVertex> vertices, final Map<Object, DetachedEdge> edges) {
- final long fileLength = file.length();
- try (final DataInputStream in = new DataInputStream(new java.io.BufferedInputStream(new FileInputStream(file)))) {
- long remaining = readAndVerifyHeader(in, file, fileLength);
- while (true) {
- final byte[] record = readFrame(in, remaining);
- if (record == null)
- break;
- // account for the header (length + crc) and payload just consumed
- remaining -= 2L * Integer.BYTES + record.length;
- applyRecord(record, vertices, edges);
+ for (final TinkerStorageMutation<TinkerEdge> m : changedEdges) {
+ if (m.isDeleted()) {
+ buffer.writeByte(OP_DEL_EDGE);
+ writer.write(m.id(), buffer);
+ } else {
+ buffer.writeByte(OP_PUT_EDGE);
+ writer.write(DetachedFactory.detach(m.element(), true), buffer);
}
- } catch (IOException ex) {
- throw new UncheckedIOException(String.format("Could not read storage file %s", file), ex);
}
+ return buffer.toWrittenArray();
}
- /**
- * Read and validate the per-file header (magic only), returning the number of record bytes that follow it. An
- * empty file (freshly created, no header yet) is treated as having no records. The store's format version is
- * validated once against the {@link #VERSION_FILE} marker in {@link #open}, not here.
- */
- private long readAndVerifyHeader(final DataInputStream in, final File file, final long fileLength) throws IOException {
- if (fileLength == 0)
- return 0;
- if (fileLength < HEADER_SIZE)
- throw new IOException(String.format("Corrupt storage file %s: shorter than its %d-byte header", file, HEADER_SIZE));
- final byte[] magic = new byte[MAGIC.length];
- readFully(in, magic);
- if (!Arrays.equals(magic, MAGIC))
- throw new IOException(String.format("%s is not a TinkerGraph storage file (bad magic)", file));
- return fileLength - HEADER_SIZE;
- }
-
- private void applyRecord(final byte[] record, final Map<Object, DetachedVertex> vertices, final Map<Object, DetachedEdge> edges) throws IOException {
- // the format version is validated once per file in readAndVerifyHeader, so records no longer repeat it
+ @Override
+ protected void decodeFrame(final byte[] record,
+ final Map<Object, DetachedVertex> vertices,
+ final Map<Object, DetachedEdge> edges) throws IOException {
final ByteBufferBuffer buffer = new ByteBufferBuffer(record);
buffer.readLong(); // txVersion, retained for diagnostics/future use
final int entryCount = buffer.readInt();
@@ -313,156 +127,12 @@ public final class GraphBinaryStorage implements TinkerStorage {
}
}
- @Override
- public void persist(final long txVersion,
- final Collection<TinkerStorageMutation<TinkerVertex>> changedVertices,
- final Collection<TinkerStorageMutation<TinkerEdge>> changedEdges) {
- ensureLogOpen();
- try {
- final byte[] frame = encodeRecord(txVersion, changedVertices, changedEdges);
- writeFrame(logOut, frame);
- logBytesSinceCompaction += 2L * Integer.BYTES + frame.length; // length + crc prefixes + payload
- } catch (IOException ex) {
- throw new UncheckedIOException("Could not append transaction to storage log", ex);
- }
- }
-
/**
- * Serialize a commit record: txVersion, entry count, then each entry as an op byte followed by either the
- * serialized element (put) or the serialized id (delete). The format version lives in the file header, not the
- * record.
+ * Stream the current committed state as one framed put record per vertex and per edge, so peak memory is bounded
+ * to a single element rather than materializing the whole graph as one byte array.
*/
- private byte[] encodeRecord(final long txVersion,
- final Collection<TinkerStorageMutation<TinkerVertex>> changedVertices,
- final Collection<TinkerStorageMutation<TinkerEdge>> changedEdges) throws IOException {
- final ByteBufferBuffer buffer = new ByteBufferBuffer();
- buffer.writeLong(txVersion);
- buffer.writeInt(changedVertices.size() + changedEdges.size());
- for (final TinkerStorageMutation<TinkerVertex> m : changedVertices) {
- if (m.isDeleted()) {
- buffer.writeByte(OP_DEL_VERTEX);
- writer.write(m.id(), buffer);
- } else {
- buffer.writeByte(OP_PUT_VERTEX);
- // detach to a stable form independent of the transactional element
- writer.write(DetachedFactory.detach(m.element(), true), buffer);
- }
- }
- for (final TinkerStorageMutation<TinkerEdge> m : changedEdges) {
- if (m.isDeleted()) {
- buffer.writeByte(OP_DEL_EDGE);
- writer.write(m.id(), buffer);
- } else {
- buffer.writeByte(OP_PUT_EDGE);
- writer.write(DetachedFactory.detach(m.element(), true), buffer);
- }
- }
- return buffer.toWrittenArray();
- }
-
- @Override
- public void flush() {
- if (closed)
- return;
- if (logOut != null) {
- try {
- // flush the JVM buffer into the OS page cache; durable against a JVM process crash
- logOut.flush();
- // in COMMIT mode also force the OS page cache to the device, so an acknowledged commit is durable
- // against an OS crash or power loss. OS mode stops at the flush above and accepts that weaker guarantee.
- if (syncMode == SyncMode.COMMIT)
- logFos.getFD().sync();
- } catch (IOException ex) {
- throw new UncheckedIOException("Could not flush storage log", ex);
- }
- }
- }
-
- @Override
- public void compact(final AbstractTinkerGraph graph) {
- if (closed)
- return;
- // Write a fresh snapshot of the current committed state, then truncate the log. This must be crash-safe: at
- // no point may a crash leave the store without a readable snapshot-or-log covering the committed state.
- // Ordering is write-tmp -> fsync tmp -> atomically rename tmp over the snapshot -> fsync dir (the rename is
- // now durable) -> delete the log -> fsync dir. The old snapshot is only ever replaced by an atomic rename, so
- // a crash at any step leaves either the old (snapshot + log) or the new (snapshot) intact — never neither.
- closeLog();
- ensureDirectory();
- final File tmp = new File(directory, SNAPSHOT_FILE + ".tmp");
- try (final FileOutputStream fos = new FileOutputStream(tmp);
- final DataOutputStream out = new DataOutputStream(new BufferedOutputStream(fos))) {
- writeHeader(out);
- writeSnapshot(graph, out);
- out.flush();
- // force the snapshot's bytes to the device before it is renamed into place
- fos.getFD().sync();
- } catch (IOException ex) {
- throw new UncheckedIOException("Could not write storage snapshot", ex);
- }
-
- try {
- // atomically replace the snapshot; no delete-then-rename window where the snapshot is briefly absent
- atomicMove(tmp, snapshotFile);
- // fsync the directory so the rename survives a crash before we touch the log
- syncDirectory();
-
- // truncate the log now that the snapshot durably reflects the committed state
- if (logFile.exists() && !logFile.delete())
- throw new IOException("Could not truncate storage log " + logFile);
- // fsync the directory again so the log's removal is durable
- syncDirectory();
- } catch (IOException ex) {
- throw new UncheckedIOException("Could not finalize storage snapshot", ex);
- }
-
- // the log is now empty; the accumulated state lives in the snapshot
- logBytesSinceCompaction = 0;
- }
-
@Override
- public void maybeCompact(final AbstractTinkerGraph graph) {
- if (closed || compactThresholdBytes <= 0)
- return;
- if (logBytesSinceCompaction >= compactThresholdBytes)
- compact(graph);
- }
-
- /**
- * Atomically move {@code source} onto {@code target}, replacing any existing target. Falls back to a non-atomic
- * replacing move on filesystems that do not support atomic moves.
- */
- private static void atomicMove(final File source, final File target) throws IOException {
- try {
- Files.move(source.toPath(), target.toPath(),
- StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
- } catch (AtomicMoveNotSupportedException anse) {
- Files.move(source.toPath(), target.toPath(), StandardCopyOption.REPLACE_EXISTING);
- }
- }
-
- /**
- * fsync the storage directory so that recent namespace changes (a rename into place, a file deletion) are durable.
- * A directory fsync is required because those operations only update the directory entry, which the earlier file
- * fsync does not cover.
- */
- private void syncDirectory() {
- try (final FileChannel dirChannel = FileChannel.open(directory.toPath(), StandardOpenOption.READ)) {
- dirChannel.force(true);
- } catch (IOException ex) {
- // some platforms (notably Windows) cannot open a directory as a channel; the atomic rename is the
- // durability guarantee there, so treat inability to sync the directory as non-fatal
- }
- }
-
- /**
- * Write the entire current committed state of the graph to {@code out} as a stream of single-element put records,
- * one frame per vertex and per edge. Each frame is an ordinary put record (see {@link #encodeRecord}) with an
- * entry count of one, so {@link #foldRecords} reconstructs the graph from these frames exactly as it would from a
- * commit log. Writing one element at a time keeps peak memory bounded to a single element rather than materializing
- * the whole graph as one byte array, so a snapshot never needs to hold a second full copy of the graph in heap.
- */
- private void writeSnapshot(final AbstractTinkerGraph graph, final DataOutputStream out) throws IOException {
+ protected void writeSnapshot(final AbstractTinkerGraph graph, final DataOutputStream out) throws IOException {
final Iterator<Vertex> vertexIterator = graph.vertices();
while (vertexIterator.hasNext())
writeElementFrame(out, OP_PUT_VERTEX, vertexIterator.next());
@@ -472,8 +142,7 @@ public final class GraphBinaryStorage implements TinkerStorage {
}
/**
- * Encode a single element as a one-entry put record and write it as a framed record to {@code out}. Only one
- * element's bytes are held in memory at a time.
+ * Encode a single element as a one-entry put record and write it as a framed record to {@code out}.
*/
private void writeElementFrame(final DataOutputStream out, final byte op, final Object element) throws IOException {
final ByteBufferBuffer buffer = new ByteBufferBuffer();
@@ -483,109 +152,4 @@ public final class GraphBinaryStorage implements TinkerStorage {
writer.write(DetachedFactory.detach(element, true), buffer);
writeFrame(out, buffer.toWrittenArray());
}
-
- @Override
- public void close() {
- closeLog();
- closed = true;
- }
-
- private void ensureLogOpen() {
- if (logOut == null) {
- try {
- final boolean freshFile = !logFile.exists() || logFile.length() == 0;
- // retain the FileOutputStream so flush() can reach its FileDescriptor for fsync
- logFos = new FileOutputStream(logFile, true);
- logOut = new DataOutputStream(new BufferedOutputStream(logFos));
- if (freshFile)
- writeHeader(logOut);
- } catch (IOException ex) {
- throw new UncheckedIOException("Could not open storage log for append", ex);
- }
- }
- }
-
- /**
- * Write the per-file header ({@link #MAGIC}) at the start of a storage file. The format version is recorded once
- * per store in the {@link #VERSION_FILE} marker, not per file.
- */
- private static void writeHeader(final DataOutputStream out) throws IOException {
- out.write(MAGIC);
- }
-
- private void closeLog() {
- if (logOut != null) {
- try {
- logOut.flush();
- logOut.close();
- } catch (IOException ex) {
- throw new UncheckedIOException("Could not close storage log", ex);
- } finally {
- logOut = null;
- logFos = null;
- }
- }
- }
-
- /**
- * Write a framed record: a 4-byte big-endian payload length, a 4-byte CRC32 of the payload, then the payload.
- * The checksum lets a reader tell a bit-flip inside a complete frame (corruption) from a short final frame left
- * by an interrupted append (truncation).
- */
- private static void writeFrame(final DataOutputStream out, final byte[] payload) throws IOException {
- final CRC32 crc = new CRC32();
- crc.update(payload);
- out.writeInt(payload.length);
- out.writeInt((int) crc.getValue());
- out.write(payload);
- }
-
- /**
- * Read a framed record, or return {@code null} at end of the readable log. A frame that is only partially present
- * — the file ends inside the header or payload — is treated as an interrupted trailing append (truncation) and
- * ends reading so earlier committed records still load. A frame that is fully present but whose stored CRC does
- * not match its payload is genuine corruption and is raised, rather than silently dropping it and everything
- * after it.
- *
- * @param remaining bytes left in the file at the current position; used to distinguish a short trailing frame
- * (truncation) from a complete frame, and to bound the payload allocation against a garbage length
- */
- private static byte[] readFrame(final DataInputStream in, final long remaining) throws IOException {
- if (remaining == 0)
- return null; // clean end of file, exactly on a frame boundary
- if (remaining < 2L * Integer.BYTES)
- return null; // not even a full header left — interrupted append
-
- final int length = in.readInt();
- final int storedCrc = in.readInt();
- if (length < 0)
- throw new IOException("Corrupt storage frame: negative payload length " + length);
- if ((long) length > remaining - 2L * Integer.BYTES)
- return null; // frame claims more bytes than remain — truncated trailing append
-
- final byte[] payload = new byte[length];
- try {
- readFully(in, payload);
- } catch (EOFException eof) {
- return null; // partial trailing payload from an interrupted append
- }
-
- final CRC32 crc = new CRC32();
- crc.update(payload);
- if ((int) crc.getValue() != storedCrc)
- throw new IOException(String.format(
- "Corrupt storage frame: CRC mismatch (stored %08x, computed %08x) in a fully-present %d-byte record",
- storedCrc, (int) crc.getValue(), length));
- return payload;
- }
-
- private static void readFully(final InputStream in, final byte[] dst) throws IOException {
- int off = 0;
- while (off < dst.length) {
- final int read = in.read(dst, off, dst.length - off);
- if (read < 0)
- throw new EOFException();
- off += read;
- }
- }
}