You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by if...@apache.org on 2026/08/04 05:20:53 UTC
(cassandra-simulator) 03/22: UNREVIEWED: Add journal module with corrected name
This is an automated email from the ASF dual-hosted git repository.
ifesdjeen pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/cassandra-simulator.git
commit 49725775774ea67f79092a6cc037635c7127b0ac
Author: Alex Petrov <ol...@gmail.com>
AuthorDate: Mon Jul 13 18:48:22 2026 +0000
UNREVIEWED: Add journal module with corrected name
---
journal/build.gradle | 38 +
journal/gradle/wrapper/gradle-wrapper.jar | Bin 0 -> 61574 bytes
journal/gradle/wrapper/gradle-wrapper.properties | 5 +
journal/gradlew | 244 +++++
journal/settings.gradle | 1 +
journal/src/main/java/accord/utils/Invariants.java | 37 +
.../org/apache/cassandra/io/util/FileUtils.java | 72 ++
.../apache/cassandra/journal/ActiveSegment.java | 543 +++++++++++
.../org/apache/cassandra/journal/Compactor.java | 116 +++
.../org/apache/cassandra/journal/Component.java | 47 +
.../org/apache/cassandra/journal/Descriptor.java | 205 +++++
.../apache/cassandra/journal/EntrySerializer.java | 200 +++++
.../java/org/apache/cassandra/journal/Flusher.java | 263 ++++++
.../apache/cassandra/journal/InMemoryIndex.java | 162 ++++
.../java/org/apache/cassandra/journal/Index.java | 111 +++
.../java/org/apache/cassandra/journal/Journal.java | 996 +++++++++++++++++++++
.../apache/cassandra/journal/JournalReadError.java | 38 +
.../cassandra/journal/JournalWriteError.java | 38 +
.../org/apache/cassandra/journal/KeySupport.java | 49 +
.../org/apache/cassandra/journal/Metadata.java | 147 +++
.../java/org/apache/cassandra/journal/Metrics.java | 10 +
.../org/apache/cassandra/journal/OnDiskIndex.java | 350 ++++++++
.../java/org/apache/cassandra/journal/Params.java | 62 ++
.../apache/cassandra/journal/RecordConsumer.java | 26 +
.../java/org/apache/cassandra/journal/Segment.java | 148 +++
.../apache/cassandra/journal/SegmentCompactor.java | 34 +
.../org/apache/cassandra/journal/Segments.java | 246 +++++
.../apache/cassandra/journal/StaticSegment.java | 450 ++++++++++
.../apache/cassandra/journal/ValueSerializer.java | 34 +
.../org/apache/cassandra/journal/package-info.java | 22 +
.../apache/cassandra/journal/DescriptorTest.java | 113 +++
.../org/apache/cassandra/journal/IndexTest.java | 307 +++++++
.../org/apache/cassandra/journal/JournalTest.java | 296 ++++++
.../org/apache/cassandra/journal/MetadataTest.java | 67 ++
.../org/apache/cassandra/journal/SegmentTest.java | 191 ++++
.../org/apache/cassandra/journal/TestParams.java | 75 ++
.../cassandra/journal/TimeUUIDKeySupport.java | 100 +++
.../java/org/apache/cassandra/utils/TimeUUID.java | 64 ++
settings.gradle | 1 +
39 files changed, 5908 insertions(+)
diff --git a/journal/build.gradle b/journal/build.gradle
new file mode 100644
index 0000000..c9a9a7a
--- /dev/null
+++ b/journal/build.gradle
@@ -0,0 +1,38 @@
+plugins {
+ id 'java'
+}
+
+repositories {
+ mavenCentral()
+}
+
+configurations.configureEach {
+ exclude group: 'ch.qos.logback'
+}
+
+java {
+ toolchain {
+ languageVersion = JavaLanguageVersion.of(21)
+ }
+}
+
+dependencies {
+ implementation fileTree(dir: '../build/lib/jars', exclude: ['logback-*.jar', 'slf4j-api-*.jar'])
+ implementation files('../build/classes/main', '../modules/accord/accord-core/build/classes/java/main')
+
+ testImplementation files('../build/test/classes', '../modules/accord/accord-core/build/classes/java/test')
+ implementation 'org.slf4j:slf4j-api:1.7.36'
+
+ testImplementation 'junit:junit:4.13.2'
+ testImplementation 'org.assertj:assertj-core:3.26.3'
+ testImplementation 'org.quicktheories:quicktheories:0.26'
+ runtimeOnly 'org.slf4j:slf4j-nop:1.7.36'
+}
+
+tasks.withType(Test).configureEach {
+ useJUnit()
+ jvmArgs '--add-exports=java.base/sun.nio.ch=ALL-UNNAMED',
+ '--add-exports=java.base/jdk.internal.ref=ALL-UNNAMED',
+ '--add-opens=java.base/java.io=ALL-UNNAMED',
+ '--add-opens=java.base/sun.nio.ch=ALL-UNNAMED'
+}
diff --git a/journal/gradle/wrapper/gradle-wrapper.jar b/journal/gradle/wrapper/gradle-wrapper.jar
new file mode 100644
index 0000000..943f0cb
Binary files /dev/null and b/journal/gradle/wrapper/gradle-wrapper.jar differ
diff --git a/journal/gradle/wrapper/gradle-wrapper.properties b/journal/gradle/wrapper/gradle-wrapper.properties
new file mode 100644
index 0000000..1e2fbf0
--- /dev/null
+++ b/journal/gradle/wrapper/gradle-wrapper.properties
@@ -0,0 +1,5 @@
+distributionBase=GRADLE_USER_HOME
+distributionPath=wrapper/dists
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.10.2-bin.zip
+zipStoreBase=GRADLE_USER_HOME
+zipStorePath=wrapper/dists
diff --git a/journal/gradlew b/journal/gradlew
new file mode 100755
index 0000000..65dcd68
--- /dev/null
+++ b/journal/gradlew
@@ -0,0 +1,244 @@
+#!/bin/sh
+
+#
+# Copyright © 2015-2021 the original authors.
+#
+# Licensed 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
+#
+# https://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.
+#
+
+##############################################################################
+#
+# Gradle start up script for POSIX generated by Gradle.
+#
+# Important for running:
+#
+# (1) You need a POSIX-compliant shell to run this script. If your /bin/sh is
+# noncompliant, but you have some other compliant shell such as ksh or
+# bash, then to run this script, type that shell name before the whole
+# command line, like:
+#
+# ksh Gradle
+#
+# Busybox and similar reduced shells will NOT work, because this script
+# requires all of these POSIX shell features:
+# * functions;
+# * expansions «$var», «${var}», «${var:-default}», «${var+SET}»,
+# «${var#prefix}», «${var%suffix}», and «$( cmd )»;
+# * compound commands having a testable exit status, especially «case»;
+# * various built-in commands including «command», «set», and «ulimit».
+#
+# Important for patching:
+#
+# (2) This script targets any POSIX shell, so it avoids extensions provided
+# by Bash, Ksh, etc; in particular arrays are avoided.
+#
+# The "traditional" practice of packing multiple parameters into a
+# space-separated string is a well documented source of bugs and security
+# problems, so this is (mostly) avoided, by progressively accumulating
+# options in "$@", and eventually passing that to Java.
+#
+# Where the inherited environment variables (DEFAULT_JVM_OPTS, JAVA_OPTS,
+# and GRADLE_OPTS) rely on word-splitting, this is performed explicitly;
+# see the in-line comments for details.
+#
+# There are tweaks for specific operating systems such as AIX, CygWin,
+# Darwin, MinGW, and NonStop.
+#
+# (3) This script is generated from the Groovy template
+# https://github.com/gradle/gradle/blob/HEAD/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+# within the Gradle project.
+#
+# You can find Gradle at https://github.com/gradle/gradle/.
+#
+##############################################################################
+
+# Attempt to set APP_HOME
+
+# Resolve links: $0 may be a link
+app_path=$0
+
+# Need this for daisy-chained symlinks.
+while
+ APP_HOME=${app_path%"${app_path##*/}"} # leaves a trailing /; empty if no leading path
+ [ -h "$app_path" ]
+do
+ ls=$( ls -ld "$app_path" )
+ link=${ls#*' -> '}
+ case $link in #(
+ /*) app_path=$link ;; #(
+ *) app_path=$APP_HOME$link ;;
+ esac
+done
+
+# This is normally unused
+# shellcheck disable=SC2034
+APP_BASE_NAME=${0##*/}
+APP_HOME=$( cd "${APP_HOME:-./}" && pwd -P ) || exit
+
+# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
+
+# Use the maximum available, or set MAX_FD != -1 to use that value.
+MAX_FD=maximum
+
+warn () {
+ echo "$*"
+} >&2
+
+die () {
+ echo
+ echo "$*"
+ echo
+ exit 1
+} >&2
+
+# OS specific support (must be 'true' or 'false').
+cygwin=false
+msys=false
+darwin=false
+nonstop=false
+case "$( uname )" in #(
+ CYGWIN* ) cygwin=true ;; #(
+ Darwin* ) darwin=true ;; #(
+ MSYS* | MINGW* ) msys=true ;; #(
+ NONSTOP* ) nonstop=true ;;
+esac
+
+CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
+
+
+# Determine the Java command to use to start the JVM.
+if [ -n "$JAVA_HOME" ] ; then
+ if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+ # IBM's JDK on AIX uses strange locations for the executables
+ JAVACMD=$JAVA_HOME/jre/sh/java
+ else
+ JAVACMD=$JAVA_HOME/bin/java
+ fi
+ if [ ! -x "$JAVACMD" ] ; then
+ die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+ fi
+else
+ JAVACMD=java
+ which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+fi
+
+# Increase the maximum file descriptors if we can.
+if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
+ case $MAX_FD in #(
+ max*)
+ # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked.
+ # shellcheck disable=SC3045
+ MAX_FD=$( ulimit -H -n ) ||
+ warn "Could not query maximum file descriptor limit"
+ esac
+ case $MAX_FD in #(
+ '' | soft) :;; #(
+ *)
+ # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked.
+ # shellcheck disable=SC3045
+ ulimit -n "$MAX_FD" ||
+ warn "Could not set maximum file descriptor limit to $MAX_FD"
+ esac
+fi
+
+# Collect all arguments for the java command, stacking in reverse order:
+# * args from the command line
+# * the main class name
+# * -classpath
+# * -D...appname settings
+# * --module-path (only if needed)
+# * DEFAULT_JVM_OPTS, JAVA_OPTS, and GRADLE_OPTS environment variables.
+
+# For Cygwin or MSYS, switch paths to Windows format before running java
+if "$cygwin" || "$msys" ; then
+ APP_HOME=$( cygpath --path --mixed "$APP_HOME" )
+ CLASSPATH=$( cygpath --path --mixed "$CLASSPATH" )
+
+ JAVACMD=$( cygpath --unix "$JAVACMD" )
+
+ # Now convert the arguments - kludge to limit ourselves to /bin/sh
+ for arg do
+ if
+ case $arg in #(
+ -*) false ;; # don't mess with options #(
+ /?*) t=${arg#/} t=/${t%%/*} # looks like a POSIX filepath
+ [ -e "$t" ] ;; #(
+ *) false ;;
+ esac
+ then
+ arg=$( cygpath --path --ignore --mixed "$arg" )
+ fi
+ # Roll the args list around exactly as many times as the number of
+ # args, so each arg winds up back in the position where it started, but
+ # possibly modified.
+ #
+ # NB: a `for` loop captures its iteration list before it begins, so
+ # changing the positional parameters here affects neither the number of
+ # iterations, nor the values presented in `arg`.
+ shift # remove old arg
+ set -- "$@" "$arg" # push replacement arg
+ done
+fi
+
+# Collect all arguments for the java command;
+# * $DEFAULT_JVM_OPTS, $JAVA_OPTS, and $GRADLE_OPTS can contain fragments of
+# shell script including quotes and variable substitutions, so put them in
+# double quotes to make sure that they get re-expanded; and
+# * put everything else in single quotes, so that it's not re-expanded.
+
+set -- \
+ "-Dorg.gradle.appname=$APP_BASE_NAME" \
+ -classpath "$CLASSPATH" \
+ org.gradle.wrapper.GradleWrapperMain \
+ "$@"
+
+# Stop when "xargs" is not available.
+if ! command -v xargs >/dev/null 2>&1
+then
+ die "xargs is not available"
+fi
+
+# Use "xargs" to parse quoted args.
+#
+# With -n1 it outputs one arg per line, with the quotes and backslashes removed.
+#
+# In Bash we could simply go:
+#
+# readarray ARGS < <( xargs -n1 <<<"$var" ) &&
+# set -- "${ARGS[@]}" "$@"
+#
+# but POSIX shell has neither arrays nor command substitution, so instead we
+# post-process each arg (as a line of input to sed) to backslash-escape any
+# character that might be a shell metacharacter, then use eval to reverse
+# that process (while maintaining the separation between arguments), and wrap
+# the whole thing up as a single "set" statement.
+#
+# This will of course break if any of these variables contains a newline or
+# an unmatched quote.
+#
+
+eval "set -- $(
+ printf '%s\n' "$DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS" |
+ xargs -n1 |
+ sed ' s~[^-[:alnum:]+,./:=@_]~\\&~g; ' |
+ tr '\n' ' '
+ )" '"$@"'
+
+exec "$JAVACMD" "$@"
diff --git a/journal/settings.gradle b/journal/settings.gradle
new file mode 100644
index 0000000..6f839b0
--- /dev/null
+++ b/journal/settings.gradle
@@ -0,0 +1 @@
+rootProject.name = 'journal'
diff --git a/journal/src/main/java/accord/utils/Invariants.java b/journal/src/main/java/accord/utils/Invariants.java
new file mode 100644
index 0000000..f912b1a
--- /dev/null
+++ b/journal/src/main/java/accord/utils/Invariants.java
@@ -0,0 +1,37 @@
+package accord.utils;
+
+import java.util.function.Supplier;
+
+/**
+ * Minimal invariant helpers used by the journal code.
+ */
+public final class Invariants
+{
+ private Invariants()
+ {
+ }
+
+ public static void require(boolean condition)
+ {
+ if (!condition)
+ throw new IllegalStateException();
+ }
+
+ public static void require(boolean condition, String message, Object... args)
+ {
+ if (!condition)
+ throw new IllegalStateException(String.format(message, args));
+ }
+
+ public static void require(boolean condition, Supplier<String> messageSupplier)
+ {
+ if (!condition)
+ throw new IllegalStateException(messageSupplier.get());
+ }
+
+ public static void requireArgument(boolean condition)
+ {
+ if (!condition)
+ throw new IllegalArgumentException();
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/io/util/FileUtils.java b/journal/src/main/java/org/apache/cassandra/io/util/FileUtils.java
new file mode 100644
index 0000000..99baa13
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/io/util/FileUtils.java
@@ -0,0 +1,72 @@
+package org.apache.cassandra.io.util;
+
+import java.io.Closeable;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+
+public final class FileUtils
+{
+ private static final Object UNSAFE;
+ private static final Method INVOKE_CLEANER;
+
+ static
+ {
+ Object unsafe = null;
+ Method invokeCleaner = null;
+ try
+ {
+ Class<?> unsafeClass = Class.forName("sun.misc.Unsafe");
+ Field field = unsafeClass.getDeclaredField("theUnsafe");
+ field.setAccessible(true);
+ unsafe = field.get(null);
+ invokeCleaner = unsafeClass.getMethod("invokeCleaner", ByteBuffer.class);
+ }
+ catch (Throwable ignore)
+ {
+ }
+ UNSAFE = unsafe;
+ INVOKE_CLEANER = invokeCleaner;
+ }
+
+ private FileUtils() {}
+
+ public static void clean(ByteBuffer buffer)
+ {
+ if (buffer == null || !buffer.isDirect() || UNSAFE == null || INVOKE_CLEANER == null)
+ return;
+ try
+ {
+ INVOKE_CLEANER.invoke(UNSAFE, buffer);
+ }
+ catch (Throwable ignore)
+ {
+ }
+ }
+
+ public static void closeQuietly(Closeable closeable)
+ {
+ if (closeable == null)
+ return;
+ try
+ {
+ closeable.close();
+ }
+ catch (Exception ignore)
+ {
+ }
+ }
+
+ public static void closeQuietly(AutoCloseable closeable)
+ {
+ if (closeable == null)
+ return;
+ try
+ {
+ closeable.close();
+ }
+ catch (Exception ignore)
+ {
+ }
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/ActiveSegment.java b/journal/src/main/java/org/apache/cassandra/journal/ActiveSegment.java
new file mode 100644
index 0000000..51788e1
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/ActiveSegment.java
@@ -0,0 +1,543 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.StandardOpenOption;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import java.util.concurrent.locks.LockSupport;
+
+import org.apache.cassandra.db.TypeSizes;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.utils.Clock;
+import org.apache.cassandra.utils.Simulate;
+import org.apache.cassandra.utils.SyncUtil;
+import org.apache.cassandra.utils.concurrent.AsyncPromise;
+import org.apache.cassandra.utils.concurrent.OpOrder;
+import org.apache.cassandra.utils.concurrent.Ref;
+
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+@Simulate(with=MONITORS)
+public final class ActiveSegment<K, V> extends Segment<K, V>
+{
+ final FileChannel channel;
+
+ // OpOrder used to order appends wrt flush
+ private final OpOrder appendOrder = new OpOrder();
+
+ // position in the buffer we are allocating from
+ private volatile long allocateOffset = 0;
+ private static final AtomicLongFieldUpdater<ActiveSegment> allocateOffsetUpdater = AtomicLongFieldUpdater.newUpdater(ActiveSegment.class, "allocateOffset");
+
+ /*
+ * Allocation started at fsyncedTo or any earlier offset are guaranteed to be both written and flushed.
+ */
+ private volatile int fsyncedTo = 0;
+ @SuppressWarnings("rawtypes")
+ private static final AtomicIntegerFieldUpdater<ActiveSegment> fsyncedToUpdater = AtomicIntegerFieldUpdater.newUpdater(ActiveSegment.class, "fsyncedTo");
+
+ /*
+ * End position of the buffer; initially set to its capacity and
+ * updated to point to the last written position as the segment is being closed
+ * no need to be volatile as writes are protected by appendOrder barrier.
+ */
+ private int endOfBuffer;
+
+ private final Ref<Segment<K, V>> selfRef;
+
+ private final InMemoryIndex<K> index;
+ private Params.FlushMode flushMode;
+ private ActiveSegment(Descriptor descriptor, Params params, InMemoryIndex<K> index, Metadata metadata, KeySupport<K> keySupport)
+ {
+ super(descriptor, metadata, keySupport);
+ this.index = index;
+ this.flushMode = params.flushMode();
+
+ try
+ {
+ channel = FileChannel.open(file.toPath(), StandardOpenOption.WRITE, StandardOpenOption.READ, StandardOpenOption.CREATE);
+ buffer = channel.map(FileChannel.MapMode.READ_WRITE, 0, params.segmentSize());
+ endOfBuffer = buffer.capacity();
+ selfRef = new Ref<>(this, new Tidier(descriptor, channel, buffer));
+ }
+ catch (IOException e)
+ {
+ throw new JournalWriteError(descriptor, file, e);
+ }
+ }
+
+ static <K, V> ActiveSegment<K, V> create(Descriptor descriptor, Params params, KeySupport<K> keySupport)
+ {
+ InMemoryIndex<K> index = InMemoryIndex.create(keySupport);
+ Metadata metadata = Metadata.create();
+ return new ActiveSegment<>(descriptor, params, index, metadata, keySupport);
+ }
+
+ @Override
+ public InMemoryIndex<K> index()
+ {
+ return index;
+ }
+
+ boolean isEmpty()
+ {
+ return allocateOffset == 0;
+ }
+
+ @Override
+ boolean isActive()
+ {
+ return true;
+ }
+
+ @Override
+ ActiveSegment<K, V> asActive()
+ {
+ return this;
+ }
+
+ @Override
+ StaticSegment<K, V> asStatic()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Read the entry and specified offset into the entry holder.
+ * Expects the caller to acquire the ref to the segment and the record to exist.
+ */
+ @Override
+ boolean read(int offset, int size, EntrySerializer.EntryHolder<K> into)
+ {
+ ByteBuffer duplicate = buffer.duplicate().position(offset).limit(offset + size);
+ try
+ {
+ EntrySerializer.read(into, keySupport, duplicate, descriptor.userVersion);
+ }
+ catch (IOException e)
+ {
+ throw new JournalReadError(descriptor, file, e);
+ }
+ return true;
+ }
+
+ /**
+ * Stop writing to this file, flush and close it. Does nothing if the file is already closed.
+ */
+ public synchronized void close(Journal<K, V> journal)
+ {
+ close(journal, true);
+ }
+
+ /**
+ * @return true if the closed segment was definitely empty, false otherwise
+ */
+ private synchronized boolean close(Journal<K, V> journal, boolean persistComponents)
+ {
+ boolean isEmpty = discardUnusedTail();
+ if (!isEmpty)
+ {
+ fsync();
+ if (persistComponents) persistComponents();
+ }
+ release(journal);
+ return isEmpty;
+ }
+
+ /**
+ * Close and discard a pre-allocated, available segment, that's never been exposed
+ */
+ void closeAndDiscard(Journal<K, V> journal)
+ {
+ boolean isEmpty = close(journal, false);
+ if (!isEmpty) throw new IllegalStateException();
+ discard();
+ }
+
+ void closeAndIfEmptyDiscard(Journal<K, V> journal)
+ {
+ boolean isEmpty = close(journal, true);
+ if (isEmpty) discard();
+ }
+
+ void persistComponents()
+ {
+ index.persist(descriptor);
+ metadata.persist(descriptor);
+ SyncUtil.trySyncDir(descriptor.directory);
+ }
+
+ private void discard()
+ {
+ selfRef.ensureReleased();
+
+ descriptor.fileFor(Component.DATA).deleteIfExists();
+ descriptor.fileFor(Component.INDEX).deleteIfExists();
+ descriptor.fileFor(Component.METADATA).deleteIfExists();
+ }
+
+ @Override
+ public Ref<Segment<K, V>> tryRef()
+ {
+ return selfRef.tryRef();
+ }
+
+ @Override
+ public Ref<Segment<K, V>> ref()
+ {
+ return selfRef.ref();
+ }
+
+ @Override
+ public Ref<Segment<K, V>> selfRef()
+ {
+ return selfRef;
+ }
+
+ private static final class Tidier extends Segment.Tidier implements Tidy
+ {
+ private final Descriptor descriptor;
+ private final FileChannel channel;
+ private final ByteBuffer buffer;
+
+ Tidier(Descriptor descriptor, FileChannel channel, ByteBuffer buffer)
+ {
+ this.descriptor = descriptor;
+ this.channel = channel;
+ this.buffer = buffer;
+ }
+
+ @Override
+ void onUnreferenced()
+ {
+ FileUtils.clean(buffer);
+ try
+ {
+ channel.close();
+ }
+ catch (IOException e)
+ {
+ throw new JournalWriteError(descriptor, Component.DATA, e);
+ }
+ }
+
+ @Override
+ public String name()
+ {
+ return descriptor.toString();
+ }
+ }
+
+ public boolean fullyFlushed()
+ {
+ return fsyncedTo == allocateOffset;
+ }
+
+ /**
+ * We are _always_ flushing upon full write, which is why we are tracking only lower allocation bounds.
+ */
+ void fsync(int upTo)
+ {
+ fsyncInternal();
+ fsyncedToUpdater.accumulateAndGet(this, upTo, Math::max);
+ }
+
+ void fsync()
+ {
+ if (fullyFlushed())
+ return;
+
+ waitForModifications();
+ fsyncInternal();
+ }
+
+ /**
+ * Wait for any appends or discardUnusedTail() operations started before this method was called
+ */
+ private void waitForModifications()
+ {
+ // issue a barrier and wait for it
+ appendOrder.awaitNewBarrier();
+ }
+
+ private void fsyncInternal()
+ {
+ try
+ {
+ SyncUtil.force((MappedByteBuffer) buffer);
+ }
+ catch (Exception e) // MappedByteBuffer.force() does not declare IOException but can actually throw it
+ {
+ throw new JournalWriteError(descriptor, file, e);
+ }
+ }
+
+ /**
+ * Ensures no more of this segment is writeable, by allocating any unused section at the end
+ * and marking it discarded void discartUnusedTail()
+ *
+ * @return true if the segment was empty, false otherwise
+ */
+ boolean discardUnusedTail()
+ {
+ try (OpOrder.Group ignored = appendOrder.start())
+ {
+ while (true)
+ {
+ long prev = completeInProgress();
+ int next = endOfBuffer + 1;
+
+ if ((int)prev >= next)
+ {
+ // already stopped allocating, might also be closed
+ assert buffer == null || prev == buffer.capacity() + 1;
+ return false;
+ }
+
+ if (allocateOffsetUpdater.compareAndSet(this, prev, next))
+ {
+ // stopped allocating now; can only succeed once, no further allocation or discardUnusedTail can succeed
+ endOfBuffer = (int)prev;
+ assert buffer != null && next == buffer.capacity() + 1;
+ return prev == 0;
+ }
+ LockSupport.parkNanos(1);
+ }
+ }
+ }
+
+ /*
+ * Entry/bytes allocation logic
+ */
+
+ @SuppressWarnings({ "resource", "RedundantSuppression" }) // op group will be closed by Allocation#write()
+ Allocation allocate(int entrySize)
+ {
+ int totalSize = totalEntrySize(entrySize);
+ OpOrder.Group opGroup = appendOrder.start();
+ try
+ {
+ int position = allocateBytes(totalSize);
+ if (position < 0)
+ {
+ opGroup.close();
+ return null;
+ }
+ return new Allocation(opGroup, buffer.duplicate().position(position).limit(position + totalSize), totalSize);
+ }
+ catch (Throwable t)
+ {
+ opGroup.close();
+ throw t;
+ }
+ }
+
+ private int totalEntrySize(int recordSize)
+ {
+ return EntrySerializer.headerSize(keySupport, descriptor.userVersion)
+ + recordSize
+ + TypeSizes.INT_SIZE // CRC
+ ;
+ }
+
+ // allocate bytes in the segment, or return -1 if not enough space
+ private int allocateBytes(int size)
+ {
+ while (true)
+ {
+ long prev = maybeCompleteInProgress();
+ if (prev < 0)
+ {
+ LockSupport.parkNanos(1); // ConstantBackoffCAS Algorithm from https://arxiv.org/pdf/1305.5800.pdf
+ continue;
+ }
+
+ long next = prev + size;
+ if (next >= endOfBuffer)
+ return -1;
+
+ // TODO (expected): if we write a "safe shutdown" marker we don't need this,
+ // but this provides safe restart in the event the process terminates abruptly but the host remains stable
+ long inProgress = prev | (next << 32);
+ if (!allocateOffsetUpdater.compareAndSet(this, prev, inProgress))
+ {
+ LockSupport.parkNanos(1); // ConstantBackoffCAS Algorithm from https://arxiv.org/pdf/1305.5800.pdf
+ continue;
+ }
+
+ assert buffer != null;
+ buffer.putInt((int)prev, (int)next);
+ allocateOffsetUpdater.compareAndSet(this, inProgress, next);
+ return (int) prev;
+ }
+ }
+
+ public interface RecordPointer
+ {
+ long segment();
+ int start();
+ void awaitUninterruptibly();
+ void onFlush(OnFlush onFlush);
+ }
+
+ public interface OnFlush
+ {
+ void success();
+ void failure(Throwable t);
+ }
+
+ final class Allocation implements RecordPointer, Comparable<Allocation>
+ {
+ private final OpOrder.Group appendOp;
+ private final ByteBuffer buffer;
+ private final int start;
+ private final int length;
+ private final AsyncPromise<Void> future;
+
+ public final long writtenAtNanos; // only set for periodic mode
+
+ Allocation(OpOrder.Group appendOp, ByteBuffer buffer, int length)
+ {
+ this.appendOp = appendOp;
+ this.buffer = buffer;
+ this.start = buffer.position();
+ this.length = length;
+ this.future = new AsyncPromise<>();
+ this.writtenAtNanos = Clock.Global.nanoTime();
+ }
+
+ void flushed()
+ {
+ // in PERIODIC, we notify about persistence after successful write to memory buffer, which means we may call this method twice
+ if (flushMode == Params.FlushMode.PERIODIC)
+ future.trySuccess(null);
+ else
+ future.setSuccess(null);
+ }
+
+ void flushFailed(Throwable t)
+ {
+ if (flushMode == Params.FlushMode.PERIODIC)
+ future.tryFailure(t);
+ else
+ future.setFailure(t);
+ }
+
+ ActiveSegment<?, ?> holder()
+ {
+ return ActiveSegment.this;
+ }
+
+ void write(K id, ByteBuffer record)
+ {
+ try
+ {
+ EntrySerializer.write(id, record, keySupport, buffer, descriptor.userVersion);
+ metadata.update();
+ index.update(id, start, length);
+ }
+ catch (IOException e)
+ {
+ throw new JournalWriteError(descriptor, file, e);
+ }
+ finally
+ {
+ appendOp.close();
+ }
+ }
+
+ @Override
+ public int start()
+ {
+ return start;
+ }
+
+ @Override
+ public void awaitUninterruptibly()
+ {
+ future.awaitUninterruptibly();
+ }
+
+ @Override
+ public void onFlush(OnFlush onFlush)
+ {
+ future.addCallback((ignore_) -> onFlush.success(), onFlush::failure);
+ }
+
+ int end()
+ {
+ return start + length;
+ }
+
+ public long segment()
+ {
+ return descriptor.timestamp;
+ }
+
+ @Override
+ public int compareTo(Allocation o)
+ {
+ int cmp = Long.compare(segment(), o.segment());
+ if (cmp != 0)
+ return cmp;
+ return Integer.compare(start, o.start);
+ }
+
+ public String toString()
+ {
+ return "Allocation{" +
+ "segment=" + segment() +
+ ", start=" + start +
+ ", end=" + end() +
+ '}';
+ }
+ }
+
+ private int maybeCompleteInProgress()
+ {
+ long cur = allocateOffset;
+ int inProgress = (int) (cur >>> 32);
+ if (inProgress == 0) return (int) cur;
+ // finish up the in-progress allocation
+ buffer.putInt((int)cur, inProgress);
+ if (!allocateOffsetUpdater.compareAndSet(this, cur, inProgress))
+ return -1;
+
+ return inProgress;
+ }
+
+ private int completeInProgress()
+ {
+ int result = maybeCompleteInProgress();
+ while (result < 0)
+ result = maybeCompleteInProgress();
+ return result;
+ }
+
+ public String toString()
+ {
+ return "ActiveSegment{" +
+ "descriptor=" + descriptor.timestamp +
+ '}';
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Compactor.java b/journal/src/main/java/org/apache/cassandra/journal/Compactor.java
new file mode 100644
index 0000000..6525df5
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Compactor.java
@@ -0,0 +1,116 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
+import org.apache.cassandra.concurrent.Shutdownable;
+
+import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
+
+public final class Compactor<K, V> implements Runnable, Shutdownable
+{
+ private final Journal<K, V> journal;
+ private final SegmentCompactor<K, V> segmentCompactor;
+ private final ScheduledExecutorPlus executor;
+ private Future<?> scheduled;
+
+ Compactor(Journal<K, V> journal, SegmentCompactor<K, V> segmentCompactor)
+ {
+ this.executor = executorFactory().scheduled(false, journal.name + "-compactor");
+ this.journal = journal;
+ this.segmentCompactor = segmentCompactor;
+ }
+
+ synchronized void start()
+ {
+ if (journal.params.enableCompaction())
+ schedule(journal.params.compactionPeriod(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
+ }
+
+ private synchronized void schedule(long period, TimeUnit units)
+ {
+ scheduled = executor.scheduleWithFixedDelay(this, period, period, units);
+ }
+
+ public synchronized void updateCompactionPeriod(int period, TimeUnit units)
+ {
+ if (!journal.params.enableCompaction())
+ return;
+
+ if (scheduled != null)
+ scheduled.cancel(false);
+
+ schedule(period, units);
+ }
+
+ @Override
+ public void run()
+ {
+ Set<StaticSegment<K, V>> toCompact = new HashSet<>();
+ journal.segments().selectStatic(toCompact);
+ if (toCompact.size() < 2)
+ return;
+
+ try
+ {
+ Collection<StaticSegment<K, V>> newSegments = segmentCompactor.compact(toCompact);
+
+ for (StaticSegment<K, V> segment : newSegments)
+ toCompact.remove(segment);
+
+ journal.replaceCompactedSegments(toCompact, newSegments);
+ for (StaticSegment<K, V> segment : toCompact)
+ segment.discard(journal);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException("Could not compact segments: " + toCompact);
+ }
+ }
+
+ @Override
+ public boolean isTerminated()
+ {
+ return executor.isTerminated();
+ }
+
+ @Override
+ public void shutdown()
+ {
+ executor.shutdown();
+ }
+
+ @Override
+ public Object shutdownNow()
+ {
+ return executor.shutdownNow();
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException
+ {
+ return executor.awaitTermination(timeout, units);
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Component.java b/journal/src/main/java/org/apache/cassandra/journal/Component.java
new file mode 100644
index 0000000..9da332d
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Component.java
@@ -0,0 +1,47 @@
+/*
+ * 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.cassandra.journal;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+enum Component
+{
+ DATA ("data"),
+ INDEX ("indx"),
+ METADATA ("meta");
+ //OFFSET_MAP (".offs"),
+ //INVLALIDATIONS (".invl");
+
+ public static final List<Component> VALUES = Collections.unmodifiableList(Arrays.asList(values()));
+ final String extension;
+
+ Component(String extension)
+ {
+ this.extension = extension;
+ }
+
+ /**
+ * @return if this component for the provided descrtiptor exists on disk
+ */
+ boolean existsFor(Descriptor descriptor)
+ {
+ return descriptor.fileFor(this).exists();
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Descriptor.java b/journal/src/main/java/org/apache/cassandra/journal/Descriptor.java
new file mode 100644
index 0000000..cea68c3
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Descriptor.java
@@ -0,0 +1,205 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.cassandra.io.util.File;
+
+import static java.lang.String.format;
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Timestamp and version encoded in the file name, e.g.
+ * log-1637159888484-2-1-1.data
+ * log-1637159888484-2-1-1.indx
+ * log-1637159888484-2-1-1.meta
+ * log-1637159888484-2-1-1.sync
+ */
+public final class Descriptor implements Comparable<Descriptor>
+{
+ private static final String SEPARATOR = "-";
+ private static final String PREFIX = "log" + SEPARATOR;
+ private static final String TMP_SUFFIX = "tmp";
+
+ private static final Pattern DATA_FILE_PATTERN =
+ Pattern.compile( PREFIX + "(\\d+)" // timestamp
+ + SEPARATOR + "(\\d+)" // generation
+ + SEPARATOR + "(\\d+)" // journal version
+ + SEPARATOR + "(\\d+)" // user version
+ + "\\." + Component.DATA.extension);
+
+ private static final Pattern TMP_FILE_PATTERN =
+ Pattern.compile( PREFIX + "\\d+" // timestamp
+ + SEPARATOR + "\\d+" // generation
+ + SEPARATOR + "\\d+" // journal version
+ + SEPARATOR + "\\d+" // user version
+ + "\\." + "[a-z]+" // component extension
+ + "\\." + TMP_SUFFIX);
+
+
+ /*
+ * NOTE: If and when another journal version is introduced, have implementations
+ * expose the version used via yaml. This way operators can force previous journal
+ * version on upgrade, temporarily, to allow easier downgrades if something goes wrong.
+ */
+ static final int JOURNAL_VERSION_1 = 1;
+ static final int CURRENT_JOURNAL_VERSION = JOURNAL_VERSION_1;
+
+ final File directory;
+ public final long timestamp;
+ public final int generation;
+
+ /**
+ * Serialization version for journal components; bumped as journal
+ * implementation evolves over time.
+ */
+ public final int journalVersion;
+
+ /**
+ * Serialization version for user content - specifically journal keys
+ * and journal values; bumped when user logic evolves.
+ */
+ public final int userVersion;
+
+ Descriptor(File directory, long timestamp, int generation, int journalVersion, int userVersion)
+ {
+ this.directory = directory;
+ this.timestamp = timestamp;
+ this.generation = generation;
+ this.journalVersion = journalVersion;
+ this.userVersion = userVersion;
+ }
+
+ static Descriptor create(File directory, long timestamp, int userVersion)
+ {
+ return new Descriptor(directory, timestamp, 1, CURRENT_JOURNAL_VERSION, userVersion);
+ }
+
+ static Descriptor fromName(File directory, String name)
+ {
+ Matcher matcher = DATA_FILE_PATTERN.matcher(name);
+ if (!matcher.matches())
+ throw new IllegalArgumentException("Provided filename " + new File(directory, name) + " is not valid for a data segment file");
+
+ long timestamp = Long.parseLong(matcher.group(1));
+ int generation = Integer.parseInt(matcher.group(2));
+ int journalVersion = Integer.parseInt(matcher.group(3));
+ int userVersion = Integer.parseInt(matcher.group(4));
+
+ return new Descriptor(directory, timestamp, generation, journalVersion, userVersion);
+ }
+
+ static Descriptor fromFile(File file)
+ {
+ return fromName(file.parent(), file.name());
+ }
+
+ File fileFor(Component component)
+ {
+ return new File(directory, formatFileName(component));
+ }
+
+ File tmpFileFor(Component component)
+ {
+ return new File(directory, formatFileName(component) + '.' + TMP_SUFFIX);
+ }
+
+ static boolean isTmpFile(File file)
+ {
+ return TMP_FILE_PATTERN.matcher(file.name()).matches();
+ }
+
+ private String formatFileName(Component component)
+ {
+ return format("%s%d%s%d%s%d%s%d.%s",
+ PREFIX, timestamp,
+ SEPARATOR, generation,
+ SEPARATOR, journalVersion,
+ SEPARATOR, userVersion,
+ component.extension);
+ }
+
+ static List<Descriptor> list(File directory)
+ {
+ try
+ {
+ return Arrays.stream(directory.listNames((file, name) -> DATA_FILE_PATTERN.matcher(name).matches()))
+ .map(name -> fromName(directory, name))
+ .collect(toList());
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public int compareTo(Descriptor other)
+ {
+ assert this.directory.equals(other.directory)
+ : format("Descriptors have mismatching directories: %s and %s", this.directory, other.directory);
+
+ int cmp = Long.compare(this.timestamp, other.timestamp);
+ if (cmp == 0) cmp = Integer.compare(this.generation, other.generation);
+ if (cmp == 0) cmp = Integer.compare(this.journalVersion, other.journalVersion);
+ if (cmp == 0) cmp = Integer.compare(this.userVersion, other.userVersion);
+ return cmp;
+ }
+
+ @Override
+ public boolean equals(Object other)
+ {
+ if (this == other)
+ return true;
+ return (other instanceof Descriptor) && equals((Descriptor) other);
+ }
+
+ boolean equals(Descriptor other)
+ {
+ assert this.directory.equals(other.directory)
+ : format("Descriptors have mismatching directories: %s and %s", this.directory, other.directory);
+
+ return this.timestamp == other.timestamp
+ && this.generation == other.generation
+ && this.journalVersion == other.journalVersion
+ && this.userVersion == other.userVersion;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ int result = directory.hashCode();
+ result = 31 * result + Long.hashCode(timestamp);
+ result = 31 * result + generation;
+ result = 31 * result + journalVersion;
+ result = 31 * result + userVersion;
+ return result;
+ }
+
+ @Override
+ public String toString()
+ {
+ return format("dir: %s, ts: %d, gen: %d, journal ver: %d, user ver: %d",
+ directory, timestamp, generation, journalVersion, userVersion);
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/EntrySerializer.java b/journal/src/main/java/org/apache/cassandra/journal/EntrySerializer.java
new file mode 100644
index 0000000..bf6464b
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/EntrySerializer.java
@@ -0,0 +1,200 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.zip.CRC32;
+
+import accord.utils.Invariants;
+import org.apache.cassandra.db.TypeSizes;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.Crc;
+
+import static org.apache.cassandra.journal.Journal.validateCRC;
+
+public final class EntrySerializer
+{
+ static <K> void write(K key,
+ ByteBuffer record,
+ KeySupport<K> keySupport,
+ ByteBuffer out,
+ int userVersion)
+ throws IOException
+ {
+ int start = out.position();
+ int totalSize = out.getInt() - start;
+ Invariants.require(totalSize == TypeSizes.INT_SIZE + out.remaining());
+ Invariants.require(totalSize == headerSize(keySupport, userVersion) + record.remaining() + TypeSizes.INT_SIZE);
+
+ keySupport.serialize(key, out, userVersion);
+
+ int headerCrcPosition = out.position();
+ out.position(headerCrcPosition + TypeSizes.INT_SIZE);
+
+ int recordSize = record.remaining();
+ int recordEnd = out.position() + recordSize;
+ Invariants.require(out.limit() == recordEnd + TypeSizes.INT_SIZE);
+ ByteBufferUtil.copyBytes(record, record.position(), out, out.position(), recordSize);
+
+ // update and write crcs
+ CRC32 crc = Crc.crc32();
+ out.position(start);
+ out.limit(headerCrcPosition);
+ crc.update(out);
+ out.limit(recordEnd);
+ out.putInt((int) crc.getValue());
+ crc.update(out);
+ out.limit(recordEnd + 4);
+ out.putInt((int) crc.getValue());
+ }
+
+ // we reuse record as the value we return
+ static <K> void read(EntryHolder<K> into,
+ KeySupport<K> keySupport,
+ ByteBuffer from,
+ int userVersion)
+ throws IOException
+ {
+ into.clear();
+
+ int start = from.position();
+ {
+ int totalSize = from.getInt(start) - start;
+ Invariants.require(totalSize == from.remaining());
+
+ CRC32 crc = Crc.crc32();
+ int headerSize = EntrySerializer.headerSize(keySupport, userVersion);
+ int headerCrc = readAndUpdateHeaderCrc(crc, from, headerSize);
+ validateCRC(crc, headerCrc);
+
+ int recordCrc = readAndUpdateRecordCrc(crc, from, start + totalSize);
+ validateCRC(crc, recordCrc);
+ }
+
+ readValidated(into, from, start, keySupport, userVersion);
+ }
+
+ // slices the provided buffer to assign to into.value
+ static <K> int tryRead(EntryHolder<K> into,
+ KeySupport<K> keySupport,
+ ByteBuffer from,
+ int syncedOffset,
+ int userVersion)
+ throws IOException
+ {
+ CRC32 crc = Crc.crc32();
+ into.clear();
+
+ int start = from.position();
+ if (from.remaining() < TypeSizes.INT_SIZE)
+ return -1;
+
+ int totalSize = from.getInt(start) - start;
+ if (totalSize == 0)
+ return -1;
+
+ if (from.remaining() < totalSize)
+ return handleReadException(new EOFException(), from.limit(), syncedOffset);
+
+ {
+ int headerSize = EntrySerializer.headerSize(keySupport, userVersion);
+ int headerCrc = readAndUpdateHeaderCrc(crc, from, headerSize);
+ try
+ {
+ validateCRC(crc, headerCrc);
+ }
+ catch (IOException e)
+ {
+ return handleReadException(e, from.position() + headerSize, syncedOffset);
+ }
+
+ int recordCrc = readAndUpdateRecordCrc(crc, from, start + totalSize);
+ try
+ {
+ validateCRC(crc, recordCrc);
+ }
+ catch (IOException e)
+ {
+ return handleReadException(e, from.position(), syncedOffset);
+ }
+ }
+
+ readValidated(into, from, start, keySupport, userVersion);
+ return totalSize;
+ }
+
+ private static <K> void readValidated(EntryHolder<K> into, ByteBuffer from, int start, KeySupport<K> keySupport, int userVersion)
+ {
+ from.position(start + TypeSizes.INT_SIZE);
+ into.key = keySupport.deserialize(from, userVersion);
+ from.position(from.position() + 4);
+ into.value = from;
+ into.userVersion = userVersion;
+ }
+
+ private static int readAndUpdateHeaderCrc(CRC32 crc, ByteBuffer from, int headerSize)
+ {
+ int headerEnd = from.position() + headerSize - TypeSizes.INT_SIZE;
+ int headerCrc = from.getInt(headerEnd);
+ from.limit(headerEnd);
+ crc.update(from);
+ return headerCrc;
+ }
+
+ private static int readAndUpdateRecordCrc(CRC32 crc, ByteBuffer from, int limit)
+ {
+ int recordEnd = limit - TypeSizes.INT_SIZE;
+ from.limit(limit);
+ int recordCrc = from.getInt(recordEnd);
+ from.position(from.position() + 4);
+ from.limit(recordEnd);
+ crc.update(from);
+ return recordCrc;
+ }
+
+ private static int handleReadException(IOException e, int bufferPosition, int fsyncedLimit) throws IOException
+ {
+ if (bufferPosition <= fsyncedLimit)
+ throw e;
+ else
+ return -1;
+ }
+
+ static <K> int headerSize(KeySupport<K> keySupport, int userVersion)
+ {
+ return TypeSizes.INT_SIZE // pointer to next entry
+ + keySupport.serializedSize(userVersion) // key/id
+ + TypeSizes.INT_SIZE; // CRC
+ }
+
+ public static final class EntryHolder<K>
+ {
+ public K key;
+ public ByteBuffer value;
+
+ public int userVersion;
+
+ public void clear()
+ {
+ key = null;
+ value = null;
+ }
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Flusher.java b/journal/src/main/java/org/apache/cassandra/journal/Flusher.java
new file mode 100644
index 0000000..952170a
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Flusher.java
@@ -0,0 +1,263 @@
+/*
+ * 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.cassandra.journal;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.utils.UnhandledEnum;
+import org.apache.cassandra.concurrent.Interruptible;
+import org.apache.cassandra.concurrent.Interruptible.State;
+import org.apache.cassandra.concurrent.Shutdownable;
+import org.apache.cassandra.journal.ActiveSegment.Allocation;
+import org.apache.cassandra.journal.Params.FlushMode;
+import org.apache.cassandra.utils.Clock.Global;
+import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.concurrent.Semaphore;
+import org.jctools.queues.MpscUnboundedArrayQueue;
+
+import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
+import static org.apache.cassandra.concurrent.ExecutorFactory.SystemThreadTag.NON_DAEMON;
+import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.Interrupts.SYNCHRONIZED;
+import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.SimulatorSafe.SAFE;
+
+/**
+ * Flusher is responsible for calling fsync on a corresponding file channel according to requested mode semantics.
+ *
+ * Flusher is notified about outstanding allocations for which writes have been completed. Flusher orders them by
+ * position, and, when possible, will pick the highest outstanding offset of the flush, and notify all pending
+ * allocations about successful fsync.
+ *
+ * Flusher relies on journal active segment to know when there will be no more writes done to the segment. In other
+ * words, once flusher sees the active segment has been switched, it waits for all writes to this segment to finish
+ * before calling fsync and starting flushes for the next segment. Main reason for this is preserving history: no
+ * allocation can be considered done until each and every allocation preceeding it is.
+ */
+@SuppressWarnings("rawtypes")
+public class Flusher implements Shutdownable
+{
+ private static final Logger logger = LoggerFactory.getLogger(Flusher.class);
+
+ private final MpscUnboundedArrayQueue<Allocation> queue;
+ private final FlushMode mode;
+ private final Semaphore semaphore;
+ private final Interruptible executor;
+
+ // counts of total pending write and written entries
+ private final AtomicLong pending = new AtomicLong(0);
+ private final AtomicLong written = new AtomicLong(0);
+
+ private long lastFlushNanos;
+
+ private final long flushPeriodNanos;
+ private final long periodicFlushLagBlockNanos;
+
+ private final Journal<?, ?> journal;
+
+ public Flusher(String name, Params params, Journal<?, ?> journal)
+ {
+ this.semaphore = Semaphore.newSemaphore(1);
+ this.queue = new MpscUnboundedArrayQueue<>(1024);
+ this.mode = params.flushMode();
+ String flushExecutorName = String.format("%s-flusher-%s", name, mode.toString().toLowerCase());
+ this.executor = executorFactory().infiniteLoop(flushExecutorName, this::run, SAFE, NON_DAEMON, SYNCHRONIZED);
+
+ this.flushPeriodNanos = mode == FlushMode.BATCH ? -1 : params.flushPeriod(TimeUnit.NANOSECONDS);
+ this.periodicFlushLagBlockNanos = mode == FlushMode.PERIODIC ? params.periodicBlockPeriod(TimeUnit.NANOSECONDS) : -1;
+ this.journal = journal;
+ }
+
+ public void flush(Allocation allocation)
+ {
+ switch (mode)
+ {
+ // A write is successful only after flushing to disk. Mutations form a group (hence the name) that waits for the same sync that happens every flushPeriod
+ case GROUP:
+ pending.incrementAndGet();
+ queue.add(allocation);
+ break;
+ // A write is successful after writing to a buffer in memory. Sync to disk happens every flushPeriod or after reaching the segment size limit.
+ // If flush is lagging by more than periodicFlushLagBlock, start blocking until flushed.
+ case PERIODIC:
+ queue.add(allocation);
+ if (Global.nanoTime() <= allocation.writtenAtNanos + periodicFlushLagBlockNanos)
+ allocation.flushed();
+ break;
+ // A write is successful only after flushing to disk. Every mutation invokes fsync.
+ case BATCH:
+ queue.add(allocation);
+ semaphore.release(1);
+ break;
+ }
+ }
+
+ public void requestExtraFlush()
+ {
+ semaphore.release(1);
+ }
+
+ private List<Allocation> ordered = new ArrayList<>();
+ private ActiveSegment flushingSegment = null;
+
+ @SuppressWarnings("unchecked")
+ private void run(State state)
+ {
+ try
+ {
+ if (state == State.NORMAL)
+ {
+ switch (mode)
+ {
+ default: throw new UnhandledEnum(mode);
+ case BATCH:
+ semaphore.acquire(1);
+ break;
+ case GROUP:
+ long now = Global.nanoTime();
+ if (lastFlushNanos != -1 && lastFlushNanos + flushPeriodNanos < now)
+ semaphore.tryAcquire(1, lastFlushNanos + flushPeriodNanos - now, TimeUnit.NANOSECONDS);
+ break;
+ case PERIODIC:
+ semaphore.tryAcquire(1, flushPeriodNanos, TimeUnit.NANOSECONDS);
+ break;
+ }
+ }
+
+ ActiveSegment activeSegment = journal.currentActiveSegment();
+ if (flushingSegment == null)
+ {
+ flushingSegment = activeSegment;
+ }
+ else if (flushingSegment != activeSegment && flushingSegment.descriptor.timestamp + 1 == activeSegment.descriptor.timestamp)
+ {
+ if (flushingSegment.fullyFlushed())
+ {
+ ActiveSegment fullyFlushed = flushingSegment;
+ flushingSegment = activeSegment;
+ journal.closeActiveSegmentAndOpenAsStatic(fullyFlushed);
+ }
+ else
+ {
+ // Work through allocations that got propagated out-of-order before we switch the segment
+ semaphore.release(1);
+ }
+ }
+
+ queue.drain(ordered::add);
+ if (ordered.isEmpty())
+ return;
+ ordered.sort(Allocation::compareTo);
+
+ int entriesToFlush = 0;
+ Allocation last = null;
+ for (int i = 0; i < ordered.size(); i++)
+ {
+ Allocation current = ordered.get(i);
+
+ // Include all consecutive entries
+ if (last != null && last.end() != current.start())
+ break;
+
+ entriesToFlush++;
+ last = current;
+ }
+
+ if (entriesToFlush > 0)
+ {
+ Throwable t = null;
+ try
+ {
+ last.holder().fsync(last.end());
+ }
+ catch (Throwable e)
+ {
+ t = e;
+ }
+ pending.addAndGet(-entriesToFlush);
+ written.addAndGet(entriesToFlush);
+ List<Allocation> next = new ArrayList<>(Math.max(ordered.size() - entriesToFlush, 2));
+ for (int i = 0; i < ordered.size(); i++)
+ {
+ Allocation allocation = ordered.get(i);
+ if (i < entriesToFlush)
+ {
+ if (t != null)
+ allocation.flushFailed(t);
+ else
+ allocation.flushed();
+ }
+ else
+ {
+ next.add(allocation);
+ }
+ }
+ ordered = next;
+ }
+ lastFlushNanos = Global.nanoTime();
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.error("Caught an exception while flushing", t);
+ List<Allocation> tmp = ordered;
+ ordered = null;
+ for (Allocation allocation : tmp)
+ allocation.flushFailed(t);
+ }
+ }
+
+ long pendingEntries()
+ {
+ return pending.get();
+ }
+
+ long writtenEntries()
+ {
+ return written.get();
+ }
+
+ @Override
+ public boolean isTerminated()
+ {
+ return executor.isTerminated();
+ }
+
+ @Override
+ public void shutdown()
+ {
+ executor.shutdown();
+ }
+
+ @Override
+ public Object shutdownNow()
+ {
+ return executor.shutdownNow();
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException
+ {
+ return executor.awaitTermination(timeout, unit);
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/InMemoryIndex.java b/journal/src/main/java/org/apache/cassandra/journal/InMemoryIndex.java
new file mode 100644
index 0000000..49fe4d1
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/InMemoryIndex.java
@@ -0,0 +1,162 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.FileOutputStreamPlus;
+import org.apache.cassandra.journal.StaticSegment.SequentialReader;
+
+/**
+ * An index for a segment that's still being updated by journal writers concurrently.
+ */
+public final class InMemoryIndex<K> extends Index<K>
+{
+ private static final long[] EMPTY = new long[0];
+
+ private final NavigableMap<K, long[]> index;
+
+ // CSLM#lastKey() can be costly, so track lastId separately;
+ // TODO: this could easily be premature and misguided;
+ // benchmark to ensure it's not acitevly harmful
+ private final AtomicReference<K> lastId;
+
+ static <K> InMemoryIndex<K> create(KeySupport<K> keySupport)
+ {
+ return new InMemoryIndex<>(keySupport, new ConcurrentSkipListMap<>(keySupport));
+ }
+
+ private InMemoryIndex(KeySupport<K> keySupport, NavigableMap<K, long[]> index)
+ {
+ super(keySupport);
+ this.index = index;
+ this.lastId = new AtomicReference<>();
+ }
+
+ public int size()
+ {
+ return index.size();
+ }
+
+ public void update(K id, int offset, int size)
+ {
+ long currentOffsetAndSize = composeOffsetAndSize(offset, size);
+ index.merge(id, new long[] { currentOffsetAndSize },
+ (current, value) ->
+ {
+ long inserting = value[0];
+ int idx = 0;
+ while (idx < current.length)
+ {
+ long cur = current[idx];
+ if (cur <= inserting)
+ {
+ if (cur == inserting)
+ return current; // TODO (expected): throw exception?
+ break;
+ }
+ ++idx;
+ }
+
+ long[] merged = new long[current.length + 1];
+ System.arraycopy(current, 0, merged, 0, idx);
+ merged[idx] = inserting;
+ System.arraycopy(current, idx, merged, idx + 1, current.length - idx);
+ return merged;
+ });
+
+ lastId.accumulateAndGet(id, (current, update) -> (null == current || keySupport.compare(current, update) < 0) ? update : current);
+ }
+
+ @Override
+ @Nullable
+ public K firstId()
+ {
+ return index.isEmpty() ? null : index.firstKey();
+ }
+
+ @Override
+ @Nullable
+ public K lastId()
+ {
+ return lastId.get();
+ }
+
+ @Override
+ public long[] lookUp(K id)
+ {
+ K lastId = lastId();
+ if (lastId == null || keySupport.compare(id, lastId) <= 0)
+ return index.getOrDefault(id, EMPTY);
+ return EMPTY;
+ }
+
+ @Override
+ public long lookUpLast(K id)
+ {
+ long[] offsets = lookUp(id);
+ return offsets.length == 0 ? -1 : offsets[0];
+ }
+
+ @Override
+ long[] lookUpAll(K id)
+ {
+ return lookUp(id);
+ }
+
+ public void persist(Descriptor descriptor)
+ {
+ File tmpFile = descriptor.tmpFileFor(Component.INDEX);
+ try (FileOutputStreamPlus out = new FileOutputStreamPlus(tmpFile))
+ {
+ OnDiskIndex.write(index, keySupport, out, descriptor.userVersion);
+
+ out.flush();
+ out.sync();
+ }
+ catch (IOException e)
+ {
+ throw new JournalWriteError(descriptor, tmpFile, e);
+ }
+ tmpFile.move(descriptor.fileFor(Component.INDEX));
+ }
+
+ static <K> InMemoryIndex<K> rebuild(Descriptor descriptor, KeySupport<K> keySupport, int fsyncedLimit)
+ {
+ InMemoryIndex<K> index = new InMemoryIndex<>(keySupport, new TreeMap<>(keySupport));
+
+ try (SequentialReader<K> reader = StaticSegment.sequentialReader(descriptor, keySupport, fsyncedLimit))
+ {
+ while (reader.advance())
+ index.update(reader.key(), reader.offset, reader.buffer.position() - reader.offset);
+ }
+ return index;
+ }
+
+ @Override
+ public void close()
+ {
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Index.java b/journal/src/main/java/org/apache/cassandra/journal/Index.java
new file mode 100644
index 0000000..fba9b99
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Index.java
@@ -0,0 +1,111 @@
+/*
+ * 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.cassandra.journal;
+
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.utils.Closeable;
+
+/**
+ * Mapping of client supplied ids to in-segment offsets
+ */
+abstract class Index<K> implements Closeable
+{
+ final KeySupport<K> keySupport;
+
+ Index(KeySupport<K> keySupport)
+ {
+ this.keySupport = keySupport;
+ }
+
+ /**
+ * Look up offsets by id. It's possible, due to retries, for a segment
+ * to contain the same record with the same id more than once, at
+ * different offsets.
+ *
+ * @return the found offsets into the segment, if any; can be empty
+ */
+ abstract long[] lookUp(K id);
+
+ /**
+ * Look up offsets by id. It's possible, due to retries, for a segment
+ * to contain the same record with the same id more than once, at
+ * different offsets. Return the first offset for provided record id, or -1 if none.
+ *
+ * @return the first offset into the segment, or -1 is none were found
+ */
+ abstract long lookUpLast(K id);
+
+ abstract long[] lookUpAll(K id);
+
+ /**
+ * @return the first (smallest) id in the index
+ */
+ @Nullable
+ abstract K firstId();
+
+ /**
+ * @return the last (largest) id in the index
+ */
+ @Nullable
+ abstract K lastId();
+
+ /**
+ * @return whether the id falls within lower/upper bounds of the index
+ */
+ boolean mayContainId(K id)
+ {
+ K firstId = firstId();
+ K lastId = lastId();
+
+ return null != firstId && keySupport.compare(id, firstId) >= 0 && (null == lastId || keySupport.compare(id, lastId) <= 0);
+ }
+
+ /**
+ * Helper methods
+ */
+
+ public static int readOffset(long record)
+ {
+ return (int) (0xffffffffL & (record >> 32));
+ }
+
+ public static long writeOffset(long record, int offset)
+ {
+ record &= 0x00000000ffffffffL; //unset all higher bits
+ record |= ((long) offset) << 32;
+ return record;
+ }
+
+ public static int readSize(long record)
+ {
+ return (int) (0xffffffffL & record);
+ }
+
+ public static long writeSize(long record, int size)
+ {
+ record &= 0xffffffff00000000L; // unset all lower bits
+ record |= size;
+ return record;
+ }
+
+ public static long composeOffsetAndSize(int offset, int size)
+ {
+ return writeSize(writeOffset(0, offset), size);
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Journal.java b/journal/src/main/java/org/apache/cassandra/journal/Journal.java
new file mode 100644
index 0000000..feef2db
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Journal.java
@@ -0,0 +1,996 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.channels.ClosedByInterruptException;
+import java.nio.file.FileStore;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
+import java.util.function.Function;
+import java.util.function.LongConsumer;
+import java.util.function.Predicate;
+import java.util.zip.CRC32;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.utils.Invariants;
+import org.apache.cassandra.concurrent.Interruptible.TerminateException;
+import org.apache.cassandra.concurrent.SequentialExecutorPlus;
+import org.apache.cassandra.concurrent.Shutdownable;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.PathUtils;
+import org.apache.cassandra.journal.Segments.ReferencedSegments;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.CloseableIterator;
+import org.apache.cassandra.utils.Crc;
+import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.LazyToString;
+import org.apache.cassandra.utils.MergeIterator;
+import org.apache.cassandra.utils.Simulate;
+import org.apache.cassandra.utils.concurrent.AsyncPromise;
+import org.apache.cassandra.utils.concurrent.OpOrder;
+import org.apache.cassandra.utils.concurrent.Promise;
+import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
+
+import static java.lang.String.format;
+import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
+import static org.apache.cassandra.journal.ActiveSegment.RecordPointer;
+import static org.apache.cassandra.journal.ActiveSegment.create;
+import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+/**
+ * A generic append-only journal with some special features:
+ * <p><ul>
+ * <li>Records can be looked up by key
+ * <li>Invalidated records get purged during segment compaction
+ * </ul><p>
+ *
+ * Segment lifecycle:
+ *
+ * ... --+------------------------/--------------------/------ ...
+ * | fsynced allocation / current allocation /
+ * ... --+--------------------/--------------------/---------- ...
+ * |
+ * |
+ * +-----------+-----------+
+ * | WriteAllocation | <-- Owned by writer thread
+ * +-----------+-----------+
+ * |
+ * | <-- When fully written, notify Flusher about this allocation
+ * |
+ * +-----------+-----------+
+ * | Flusher | <-- Collects consecutive write allocations and fsyncs file channel
+ * +-----------+-----------+
+ * |
+ * | <-- When fsynced, notify writer thread about persistence
+ * |
+ * +-----------+-----------+
+ * | Writer Thread | <-- Executes write callbacks
+ * +-----------+-----------+
+ *
+ * Type parameters:
+ * @param <V> the type of records stored in the journal
+ * @param <K> the type of keys used to address the records;
+ must be fixed-size and byte-order comparable
+ */
+@Simulate(with=MONITORS)
+public class Journal<K, V> implements Shutdownable
+{
+ private static final Logger logger = LoggerFactory.getLogger(Journal.class);
+
+ final String name;
+ final File directory;
+ final Params params;
+
+ final KeySupport<K> keySupport;
+ final ValueSerializer<K, V> valueSerializer;
+
+ final Metrics<K, V> metrics;
+
+ final Compactor<K, V> compactor;
+
+ final AtomicLong nextSegmentId = new AtomicLong();
+
+ private volatile ActiveSegment<K, V> currentSegment = null;
+ private volatile Promise<ActiveSegment<K, V>> availableSegment = null;
+
+ // segment that is ready to be used; allocator thread fills this and blocks until consumed
+ private final Allocator allocator = new Allocator();
+
+ private final AtomicReference<Segments<K, V>> segments = new AtomicReference<>();
+
+ final AtomicReference<State> state = new AtomicReference<>(State.UNINITIALIZED);
+
+ private final Flusher flusher;
+ final SequentialExecutorPlus maintenance;
+ final SequentialExecutorPlus tidy;
+
+ final OpOrder readOrder = new OpOrder();
+
+ public Journal(String name,
+ File directory,
+ Params params,
+ KeySupport<K> keySupport,
+ ValueSerializer<K, V> valueSerializer,
+ SegmentCompactor<K, V> segmentCompactor)
+ {
+ this.name = name;
+ this.directory = directory;
+ this.params = params;
+
+ this.keySupport = keySupport;
+ this.valueSerializer = valueSerializer;
+
+ this.metrics = new Metrics<>(name);
+ this.compactor = new Compactor<>(this, segmentCompactor);
+ this.maintenance = executorFactory().sequential(name + "-maintenance");
+ this.tidy = executorFactory().sequential(name + "-tidy");
+ this.flusher = new Flusher(name, params, this);
+ }
+
+ public void start()
+ {
+ Invariants.require(state.compareAndSet(State.UNINITIALIZED, State.INITIALIZING),
+ "Unexpected journal state during initialization", state);
+ metrics.register(flusher);
+
+ deleteTmpFiles();
+
+ List<Descriptor> descriptors = Descriptor.list(directory);
+ // find the largest existing timestamp
+ descriptors.sort(null);
+ long maxTimestamp = descriptors.isEmpty()
+ ? Long.MIN_VALUE
+ : descriptors.get(descriptors.size() - 1).timestamp;
+ nextSegmentId.set(Math.max(currentTimeMillis(), maxTimestamp + 1));
+ segments.set(Segments.of(StaticSegment.open(descriptors, keySupport)));
+
+ requestSegmentAllocation();
+ advanceSegment(null);
+ Invariants.require(state.compareAndSet(State.INITIALIZING, State.NORMAL),
+ "Unexpected journal state after initialization", state);
+ compactor.start();
+ }
+
+ @VisibleForTesting
+ public void runCompactorForTesting()
+ {
+ compactor.run();
+ }
+
+ public Compactor<K, V> compactor()
+ {
+ return compactor;
+ }
+
+ /**
+ * Cleans up unfinished component files from previous run (metadata and index)
+ */
+ private void deleteTmpFiles()
+ {
+ for (File tmpFile : directory.listUnchecked(Descriptor::isTmpFile))
+ tmpFile.delete();
+ }
+
+ @Override
+ public boolean isTerminated()
+ {
+ return state.get() == State.TERMINATED;
+ }
+
+ public void shutdown()
+ {
+ Invariants.require(state.compareAndSet(State.NORMAL, State.SHUTDOWN),
+ "Unexpected journal state while trying to shut down", state);
+ compactor.shutdown();
+ flusher.shutdown();
+ metrics.deregister();
+ maintenance.shutdown();
+ allocator.shutdown();
+ Invariants.require(state.compareAndSet(State.SHUTDOWN, State.TERMINATED),
+ "Unexpected journal state while trying to shut down", state);
+ }
+
+ @Override
+ public Object shutdownNow()
+ {
+ shutdown();
+ return null;
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException
+ {
+ boolean r = true;
+ r &= compactor.awaitTermination(1, TimeUnit.MINUTES);
+ r &= flusher.awaitTermination(timeout, units);
+ flusher.requestExtraFlush();
+ r &= maintenance.awaitTermination(timeout, units);
+
+ // We can only close segments after we have fully shut down maintenance thread, and we can not shut down tidy earlier since closing requires tidy pool
+ closeAllSegments();
+ tidy.shutdown();
+ r &= tidy.awaitTermination(timeout, units);
+ return r;
+ }
+
+ /**
+ * Looks up a record by the provided id.
+ * <p/>
+ * Looking up an invalidated record may or may not return a record, depending on
+ * compaction progress.
+ * <p/>
+ * In case multiple copies of the record exist in the log (e.g. because of user retries),
+ * the first one found will be returned.
+ *
+ * @param id user-provided record id, expected to roughly correlate with time and go up
+ * @return deserialized record if found, null otherwise
+ */
+ @SuppressWarnings("unused")
+ public V readLast(K id)
+ {
+ EntrySerializer.EntryHolder<K> holder = new EntrySerializer.EntryHolder<>();
+
+ try (OpOrder.Group group = readOrder.start())
+ {
+ for (Segment<K, V> segment : segments.get().allSorted(true))
+ {
+ if (segment.readLast(id, holder))
+ {
+ try (DataInputBuffer in = new DataInputBuffer(holder.value, false))
+ {
+ return valueSerializer.deserialize(holder.key, in, holder.userVersion);
+ }
+ catch (IOException e)
+ {
+ // can only throw if serializer is buggy
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ }
+ return null;
+ }
+
+ public void readAll(K id, RecordConsumer<K> consumer)
+ {
+ EntrySerializer.EntryHolder<K> holder = new EntrySerializer.EntryHolder<>();
+ try (OpOrder.Group group = readOrder.start())
+ {
+ for (Segment<K, V> segment : segments.get().allSorted(false))
+ {
+ segment.readAll(id, holder, consumer);
+ }
+ }
+ }
+
+ @SuppressWarnings("unused")
+ public List<V> readAll(K id)
+ {
+ List<V> res = new ArrayList<>(2);
+ readAll(id, (segment, position, key, buffer, userVersion) -> {
+ try (DataInputBuffer in = new DataInputBuffer(buffer, false))
+ {
+ res.add(valueSerializer.deserialize(key, in, userVersion));
+ }
+ catch (IOException e)
+ {
+ // can only throw if serializer is buggy
+ throw new RuntimeException(e);
+ }
+ });
+ return res;
+ }
+
+ /**
+ * Looks up a record by the provided id, if the value satisfies the provided condition.
+ * <p/>
+ * Looking up an invalidated record may or may not return a record, depending on
+ * compaction progress.
+ * <p/>
+ * In case multiple copies of the record exist in the log (e.g. because of user retries),
+ * and more than one of them satisfy the provided condition, the first one found will be returned.
+ *
+ * @param id user-provided record id, expected to roughly correlate with time and go up
+ * @param condition predicate to test the record against
+ * @return deserialized record if found, null otherwise
+ */
+ @SuppressWarnings("unused")
+ public V readFirstMatching(K id, Predicate<V> condition)
+ {
+ EntrySerializer.EntryHolder<K> holder = new EntrySerializer.EntryHolder<>();
+
+ try (OpOrder.Group group = readOrder.start())
+ {
+ for (Segment<K, V> segment : segments.get().all())
+ {
+ long[] offsets = segment.index().lookUp(id);
+ for (long offsetAndSize : offsets)
+ {
+ int offset = Index.readOffset(offsetAndSize);
+ int size = Index.readSize(offsetAndSize);
+ holder.clear();
+ if (segment.read(offset, size, holder))
+ {
+ try (DataInputBuffer in = new DataInputBuffer(holder.value, false))
+ {
+ V record = valueSerializer.deserialize(holder.key, in, segment.descriptor.userVersion);
+ if (condition.test(record))
+ return record;
+ }
+ catch (IOException e)
+ {
+ // can only throw if serializer is buggy
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Looks up a record by the provided id.
+ * <p/>
+ * Looking up an invalidated record may or may not return a record, depending on
+ * compaction progress.
+ * <p/>
+ * In case multiple copies of the record exist in the log (e.g. because of user retries),
+ * only the first found record will be consumed.
+ *
+ * @param id user-provided record id, expected to roughly correlate with time and go up
+ * @param consumer function to consume the raw record (bytes and invalidation set) if found
+ * @return true if the record was found, false otherwise
+ */
+ @SuppressWarnings("unused")
+ public boolean readLast(K id, RecordConsumer<K> consumer)
+ {
+ try (OpOrder.Group group = readOrder.start())
+ {
+ for (Segment<K, V> segment : segments.get().allSorted(false))
+ {
+ if (!segment.index().mayContainId(id))
+ continue;
+
+ if (segment.readLast(id, consumer))
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Synchronously write a record to the journal.
+ * <p/>
+ * Blocks until the record has been deemed durable according to the journal flush mode.
+ *
+ * @param id user-provided record id, expected to roughly correlate with time and go up
+ * @param record the record to store
+ */
+ public void blockingWrite(K id, V record)
+ {
+ asyncWrite(id, record).awaitUninterruptibly();
+ }
+
+ /**
+ * Asynchronously write a record to the journal. Writes to the journal in the calling thread,
+ * but doesn't wait for flush.
+ * <p/>
+ * Executes the supplied callback on the executor provided once the record has been durably written to disk
+ *
+ * @param id user-provided record id, expected to roughly correlate with time and go up
+ * @param record the record to store
+ */
+ public RecordPointer asyncWrite(K id, V record)
+ {
+ return asyncWrite(id, (out, userVersion) -> valueSerializer.serialize(id, record, out, userVersion));
+ }
+
+ public RecordPointer asyncWrite(K id, Writer writer)
+ {
+ return writeInternal(id, writer);
+ }
+
+ private ActiveSegment<K, V>.Allocation writeInternal(K id, Writer writer)
+ {
+ try (DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get())
+ {
+ writer.write(dob, params.userVersion());
+ ActiveSegment<K, V>.Allocation alloc = allocate(dob.getLength());
+ alloc.write(id, dob.unsafeGetBufferAndFlip());
+ flusher.flush(alloc);
+ return alloc;
+ }
+ catch (IOException e)
+ {
+ // exception during record serialization into the scratch buffer
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private ActiveSegment<K, V>.Allocation allocate(int entrySize)
+ {
+ ActiveSegment<K, V> segment = currentSegment;
+
+ ActiveSegment<K, V>.Allocation alloc;
+ while (null == (alloc = segment.allocate(entrySize)))
+ {
+ if (entrySize >= (params.segmentSize() * 3) / 4)
+ throw new IllegalStateException("entrySize " + entrySize + " too large for a segmentSize of " + params.segmentSize());
+ // failed to allocate; move to a new segment with enough room
+ advanceSegment(segment);
+ segment = currentSegment;
+ }
+ return alloc;
+ }
+
+ /*
+ * Segment allocation logic.
+ */
+
+ private void requestSegmentAllocation()
+ {
+ Promise<ActiveSegment<K, V>> promise = new AsyncPromise<>();
+ synchronized (this)
+ {
+ if (availableSegment == null)
+ {
+ availableSegment = promise;
+ promise = null;
+ }
+ }
+ if (promise == null)
+ maintenance.execute(allocator);
+ else
+ promise.cancel(true);
+ }
+
+ private void advanceSegment(ActiveSegment<K, V> oldSegment)
+ {
+ while (true)
+ {
+ Promise<ActiveSegment<K, V>> waitOn = null;
+ synchronized (this)
+ {
+ // do this in a critical section, so we can maintain the order of
+ // segment construction when moving to allocatingFrom/activeSegments
+ if (currentSegment != oldSegment)
+ return;
+
+ // if a segment is ready, take it now, otherwise wait for the allocator thread to construct it
+ if (availableSegment != null)
+ {
+ if (availableSegment.isDone())
+ {
+ Invariants.require(availableSegment.isSuccess());
+ // success - first add the next segment to segments, and only then make it available for writes
+ ActiveSegment<K, V> nextActive = availableSegment.getNow();
+ availableSegment = null;
+ addNewActiveSegment(nextActive);
+ currentSegment = nextActive;
+ break;
+ }
+ else
+ {
+ waitOn = availableSegment;
+ }
+ }
+ }
+
+ if (waitOn != null)
+ waitOn.awaitThrowUncheckedOnInterrupt();
+ }
+
+ // signal the allocator thread to prepare a new segment
+ requestSegmentAllocation();
+
+ // request that the journal be flushed out-of-band, as we've finished a segment
+ flusher.requestExtraFlush();
+ }
+
+ private void discardAvailableSegment()
+ {
+ Promise<ActiveSegment<K, V>> next;
+ synchronized (this)
+ {
+ next = availableSegment;
+ availableSegment = null;
+ }
+
+ if (next != null)
+ {
+ ActiveSegment<K, V> segment = next.awaitThrowUncheckedOnInterrupt().getNow();
+ if (segment != null) // can be null during shutdown
+ segment.closeAndDiscard(this);
+ }
+ }
+
+ private class Allocator implements Runnable
+ {
+ @Override
+ public void run()
+ {
+ try
+ {
+ State state = Journal.this.state.get();
+ switch (state)
+ {
+ case UNINITIALIZED:
+ throw new IllegalStateException("Should not be in uninitialized state");
+ case INITIALIZING:
+ case NORMAL:
+ runNormal();
+ break;
+ case SHUTDOWN:
+ case TERMINATED:
+ shutdown();
+ break;
+ }
+ }
+ catch (InterruptedException e)
+ {
+ discardAvailableSegment();
+ throw new UncheckedInterruptedException();
+ }
+ }
+
+ private void runNormal() throws InterruptedException
+ {
+ boolean interrupted = false;
+ try
+ {
+ Invariants.require(availableSegment != null);
+ Invariants.require(!availableSegment.isDone());
+
+ // synchronized to prevent thread interrupts while performing IO operations and also
+ // clear interrupted status to prevent ClosedByInterruptException in createSegment()
+ synchronized (this)
+ {
+ interrupted = Thread.interrupted();
+ availableSegment.setSuccess(createSegment());
+ }
+
+ while (availableSegment != null && Thread.interrupted())
+ LockSupport.parkNanos(100);
+ }
+ catch (JournalWriteError e)
+ {
+ if (!(e.getCause() instanceof ClosedByInterruptException))
+ throw e;
+ }
+ catch (Throwable t)
+ {
+ if (!handleError("Failed allocating a journal segment", t))
+ {
+ discardAvailableSegment();
+ throw new TerminateException();
+ }
+ TimeUnit.SECONDS.sleep(1L); // sleep for a second to avoid log spam
+ }
+
+ if (interrupted || Thread.interrupted())
+ {
+ discardAvailableSegment();
+ throw new InterruptedException();
+ }
+ }
+
+ private void shutdown()
+ {
+ Promise<ActiveSegment<K, V>> next;
+ synchronized (this)
+ {
+ next = availableSegment;
+ availableSegment = null;
+ }
+
+ if (next != null)
+ {
+ if (next.isDone())
+ next.getNow().closeAndDiscard(Journal.this);
+ else
+ next.cancel(true);
+ }
+ }
+ }
+
+ private ActiveSegment<K, V> createSegment()
+ {
+ Descriptor descriptor = Descriptor.create(directory, nextSegmentId.getAndIncrement(), params.userVersion());
+ return create(descriptor, params, keySupport);
+ }
+
+ private void closeAllSegments()
+ {
+ Segments<K, V> segments = swapSegments(ignore -> Segments.none());
+ for (Segment<K, V> segment : segments.all())
+ {
+ if (segment.isActive())
+ ((ActiveSegment<K, V>) segment).closeAndIfEmptyDiscard(this);
+ else
+ segment.close(this);
+ }
+ }
+
+ @SuppressWarnings("unused")
+ ReferencedSegments<K, V> selectAndReference(Predicate<Segment<K,V>> selector)
+ {
+ while (true)
+ {
+ ReferencedSegments<K, V> referenced = segments().selectAndReference(selector);
+ if (null != referenced)
+ return referenced;
+ }
+ }
+
+ Segments<K, V> segments()
+ {
+ return segments.get();
+ }
+
+ private Segments<K, V> swapSegments(Function<Segments<K, V>, Segments<K, V>> transformation)
+ {
+ Segments<K, V> currentSegments, newSegments;
+ do
+ {
+ currentSegments = segments();
+ newSegments = transformation.apply(currentSegments);
+ }
+ while (!segments.compareAndSet(currentSegments, newSegments));
+ return currentSegments;
+ }
+
+ private void addNewActiveSegment(ActiveSegment<K, V> activeSegment)
+ {
+ swapSegments(current -> current.withNewActiveSegment(activeSegment));
+ }
+
+ private void removeEmptySegment(ActiveSegment<K, V> activeSegment)
+ {
+ swapSegments(current -> current.withoutEmptySegment(activeSegment));
+ }
+
+ private void replaceCompletedSegment(ActiveSegment<K, V> activeSegment, StaticSegment<K, V> staticSegment)
+ {
+ swapSegments(current -> current.withCompletedSegment(activeSegment, staticSegment));
+ }
+
+ void replaceCompactedSegments(Collection<StaticSegment<K, V>> oldSegments, Collection<StaticSegment<K, V>> compactedSegments)
+ {
+ swapSegments(current -> current.withCompactedSegments(oldSegments, compactedSegments));
+ }
+
+ public ActiveSegment<K, V> currentActiveSegment()
+ {
+ return currentSegment;
+ }
+
+ /**
+ * Take care of a finished active segment:
+ * 1. discard tail
+ * 2. flush to disk
+ * 3. persist index and metadata
+ * 4. open the segment as static
+ * 5. replace the finished active segment with the opened static one in Segments view
+ * 6. release the Ref so the active segment will be cleaned up by its Tidy instance
+ */
+ private class CloseActiveSegmentRunnable implements Runnable
+ {
+ private final ActiveSegment<K, V> completedSegment;
+
+ CloseActiveSegmentRunnable(ActiveSegment<K, V> activeSegment)
+ {
+ this.completedSegment = activeSegment;
+ }
+
+ @Override
+ public void run()
+ {
+ completedSegment.discardUnusedTail();
+ completedSegment.fsync();
+ completedSegment.persistComponents();
+ replaceCompletedSegment(completedSegment, StaticSegment.open(completedSegment.descriptor, keySupport));
+ completedSegment.release(Journal.this);
+ }
+ }
+
+ void closeActiveSegmentAndOpenAsStatic(ActiveSegment<K, V> activeSegment)
+ {
+ if (activeSegment.isEmpty())
+ {
+ removeEmptySegment(activeSegment);
+ activeSegment.closeAndDiscard(this);
+ return;
+ }
+
+ maintenance.execute(new CloseActiveSegmentRunnable(activeSegment));
+ }
+
+ @VisibleForTesting
+ public void closeCurrentSegmentForTestingIfNonEmpty()
+ {
+ ActiveSegment<K, V> segment = currentSegment;
+ if (segment.isEmpty())
+ return;
+ advanceSegment(segment);
+ while (!segments().isSwitched(segment))
+ {
+ LockSupport.parkNanos(1000);
+ }
+ }
+
+ /*
+ * Static helper methods used by journal components
+ */
+
+ static void validateCRC(CRC32 crc, int readCRC) throws Crc.InvalidCrc
+ {
+ if (readCRC != (int) crc.getValue())
+ throw new Crc.InvalidCrc(readCRC, (int) crc.getValue());
+ }
+
+ /*
+ * Error handling
+ */
+
+ /**
+ * @return true if the invoking thread should continue, or false if it should terminate itself
+ */
+ boolean handleError(String message, Throwable t)
+ {
+ Params.FailurePolicy policy = params.failurePolicy();
+ JVMStabilityInspector.inspectJournalThrowable(t, name, policy);
+
+ switch (policy)
+ {
+ default:
+ throw new AssertionError(policy);
+ case DIE:
+ case STOP:
+ StorageService.instance.stopTransports();
+ //$FALL-THROUGH$
+ case STOP_JOURNAL:
+ message = format("%s. Journal %s failure policy is %s; terminating thread.", message, name, policy);
+ logger.error(maybeAddDiskSpaceContext(message), t);
+ return false;
+ case IGNORE:
+ message = format("%s. Journal %s failure policy is %s; ignoring excepton.", message, name, policy);
+ logger.error(maybeAddDiskSpaceContext(message), t);
+ return true;
+ }
+ }
+
+ /**
+ * Add additional information to the error message if the journal directory does not have enough free space.
+ *
+ * @param message the original error message
+ * @return the message with additional information if possible
+ */
+ private String maybeAddDiskSpaceContext(String message)
+ {
+ long availableDiskSpace = PathUtils.tryGetSpace(directory.toPath(), FileStore::getTotalSpace);
+ int segmentSize = params.segmentSize();
+
+ if (availableDiskSpace >= segmentSize)
+ return message;
+
+ return format("%s. %d bytes required for next journal segment but only %d bytes available. " +
+ "Check %s to see if not enough free space is the reason for this error.",
+ message, segmentSize, availableDiskSpace, directory);
+ }
+
+ @VisibleForTesting
+ public void truncateForTesting()
+ {
+ ActiveSegment<?, ?> discarding = currentSegment;
+ if (!discarding.isEmpty()) // if there is no data in the segement then ignore it
+ {
+ closeCurrentSegmentForTestingIfNonEmpty();
+ //TODO (desired): wait for the ActiveSegment to get released, else can see weird race conditions;
+ // this thread will see the static segmenet and will release it (which will delete the file),
+ // and the sync thread will then try to release and will fail as the file no longer exists...
+ while (discarding.selfRef().globalCount() > 0) {}
+ }
+
+ Segments<K, V> statics = swapSegments(s -> s.select(Segment::isActive)).select(Segment::isStatic);
+ for (Segment<K, V> segment : statics.all())
+ ((StaticSegment) segment).discard(this);
+ }
+
+ public interface Writer
+ {
+ void write(DataOutputPlus out, int userVersion) throws IOException;
+ }
+
+ /**
+ * Static segment iterator iterates all keys in _static_ segments in order.
+ */
+ public StaticSegmentKeyIterator staticSegmentKeyIterator()
+ {
+ return new StaticSegmentKeyIterator();
+ }
+
+ /**
+ * List of key and a list of segment descriptors referencing this key
+ */
+ public static class KeyRefs<K>
+ {
+ long segments[];
+ K key;
+ int size;
+
+ public KeyRefs(K key)
+ {
+ this.key = key;
+ }
+
+ private KeyRefs(int maxSize)
+ {
+ this.segments = new long[maxSize];
+ }
+
+ public void segments(LongConsumer consumer)
+ {
+ for (int i = 0; i < size; i++)
+ consumer.accept(segments[i]);
+ }
+
+ public K key()
+ {
+ return key;
+ }
+
+ private void add(K key, long segment)
+ {
+ this.key = key;
+ if (size == 0 || segments[size - 1] < segment)
+ segments[size++] = segment;
+ else
+ Invariants.require(segments[size - 1] == segment,
+ "Tried to add an out-of-order segment: %d, %s", segment,
+ LazyToString.lazy(() -> Arrays.toString(Arrays.copyOf(segments, size))));
+ }
+
+ private void reset()
+ {
+ key = null;
+ size = 0;
+ Arrays.fill(segments, 0);
+ }
+ }
+
+ public class StaticSegmentKeyIterator implements CloseableIterator<KeyRefs<K>>
+ {
+ private final ReferencedSegments<K, V> segments;
+ private final MergeIterator<Head, KeyRefs<K>> iterator;
+
+ public StaticSegmentKeyIterator()
+ {
+ this.segments = selectAndReference(Segment::isStatic);
+ List<Iterator<Head>> iterators = new ArrayList<>(segments.count());
+
+ for (Segment<K, V> segment : segments.allSorted(true))
+ {
+ StaticSegment<K, V> staticSegment = (StaticSegment<K, V>) segment;
+ Iterator<K> iter = staticSegment.index().reader();
+ Head head = new Head(staticSegment.descriptor.timestamp);
+ iterators.add(new Iterator<>()
+ {
+ public boolean hasNext()
+ {
+ return iter.hasNext();
+ }
+
+ public Head next()
+ {
+ head.key = iter.next();
+ return head;
+ }
+ });
+ }
+
+ this.iterator = MergeIterator.get(iterators,
+ (r1, r2) -> {
+ int keyCmp = keySupport.compare(r1.key, r2.key);
+ if (keyCmp != 0)
+ return keyCmp;
+ return Long.compare(r1.segment, r2.segment);
+ },
+ new MergeIterator.Reducer<Head, KeyRefs<K>>()
+ {
+ final KeyRefs<K> ret = new KeyRefs<>(segments.count());
+
+ @Override
+ public void reduce(int idx, Head head)
+ {
+ ret.add(head.key, head.segment);
+ }
+
+ @Override
+ protected KeyRefs<K> getReduced()
+ {
+ return ret;
+ }
+
+ @Override
+ protected void onKeyChange()
+ {
+ ret.reset();
+ super.onKeyChange();
+ }
+ });
+ }
+
+ @Override
+ public void close()
+ {
+ segments.close();
+ }
+
+ public KeyRefs<K> peek()
+ {
+ if (iterator.hasNext())
+ return iterator.peek();
+ return null;
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public KeyRefs<K> next()
+ {
+ return iterator.next();
+ }
+
+ class Head
+ {
+ final long segment;
+ K key;
+ Head(long segment) { this.segment = segment; }
+ }
+ }
+
+ enum State
+ {
+ UNINITIALIZED,
+ INITIALIZING,
+ NORMAL,
+ SHUTDOWN,
+ TERMINATED
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/JournalReadError.java b/journal/src/main/java/org/apache/cassandra/journal/JournalReadError.java
new file mode 100644
index 0000000..87366c8
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/JournalReadError.java
@@ -0,0 +1,38 @@
+/*
+ * 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.cassandra.journal;
+
+import org.apache.cassandra.io.FSReadError;
+import org.apache.cassandra.io.util.File;
+
+public class JournalReadError extends FSReadError
+{
+ public final Descriptor descriptor;
+
+ JournalReadError(Descriptor descriptor, File file, Throwable throwable)
+ {
+ super(throwable, file);
+ this.descriptor = descriptor;
+ }
+
+ JournalReadError(Descriptor descriptor, Component component, Throwable throwable)
+ {
+ super(throwable, descriptor.fileFor(component));
+ this.descriptor = descriptor;
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/JournalWriteError.java b/journal/src/main/java/org/apache/cassandra/journal/JournalWriteError.java
new file mode 100644
index 0000000..03193af
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/JournalWriteError.java
@@ -0,0 +1,38 @@
+/*
+ * 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.cassandra.journal;
+
+import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.io.util.File;
+
+public class JournalWriteError extends FSWriteError
+{
+ public final Descriptor descriptor;
+
+ JournalWriteError(Descriptor descriptor, File file, Throwable throwable)
+ {
+ super(throwable, file);
+ this.descriptor = descriptor;
+ }
+
+ JournalWriteError(Descriptor descriptor, Component component, Throwable throwable)
+ {
+ super(throwable, descriptor.fileFor(component));
+ this.descriptor = descriptor;
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/KeySupport.java b/journal/src/main/java/org/apache/cassandra/journal/KeySupport.java
new file mode 100644
index 0000000..efc41aa
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/KeySupport.java
@@ -0,0 +1,49 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Comparator;
+import java.util.zip.Checksum;
+
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+
+/**
+ * Record keys must satisfy two properties:
+ * <p>
+ * 1. Must have a fixed serialized size
+ * 2. Must be byte-order comparable
+ */
+public interface KeySupport<K> extends Comparator<K>
+{
+ int serializedSize(int userVersion);
+
+ void serialize(K key, DataOutputPlus out, int userVersion) throws IOException;
+ void serialize(K key, ByteBuffer out, int userVersion) throws IOException;
+
+ K deserialize(DataInputPlus in, int userVersion) throws IOException;
+
+ K deserialize(ByteBuffer buffer, int position, int userVersion);
+ K deserialize(ByteBuffer buffer, int userVersion);
+
+ void updateChecksum(Checksum crc, K key, int userVersion);
+
+ int compareWithKeyAt(K key, ByteBuffer buffer, int position, int userVersion);
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Metadata.java b/journal/src/main/java/org/apache/cassandra/journal/Metadata.java
new file mode 100644
index 0000000..0549ef0
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Metadata.java
@@ -0,0 +1,147 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.zip.CRC32;
+
+import org.apache.cassandra.io.util.*;
+import org.apache.cassandra.utils.Crc;
+
+import static org.apache.cassandra.journal.Journal.validateCRC;
+import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
+
+/**
+ * Tracks and serializes the following information:
+ * - total count of records in this segment file
+ * used for compaction prioritisation
+ */
+final class Metadata
+{
+ // TODO: why is this unused?
+ private int fsyncLimit;
+
+ private volatile int recordsCount;
+ private static final AtomicIntegerFieldUpdater<Metadata> recordsCountUpdater =
+ AtomicIntegerFieldUpdater.newUpdater(Metadata.class, "recordsCount");
+
+ static Metadata create()
+ {
+ return new Metadata(0);
+ }
+
+ private Metadata(int recordsCount)
+ {
+ this.recordsCount = recordsCount;
+ }
+
+ void update()
+ {
+ incrementRecordsCount();
+ }
+
+ int fsyncLimit()
+ {
+ return fsyncLimit;
+ }
+
+ private void incrementRecordsCount()
+ {
+ recordsCountUpdater.incrementAndGet(this);
+ }
+
+ int totalCount()
+ {
+ return recordsCount;
+ }
+
+ void write(DataOutputPlus out) throws IOException
+ {
+ CRC32 crc = Crc.crc32();
+ out.writeInt(recordsCount);
+ updateChecksumInt(crc, recordsCount);
+ out.writeInt((int) crc.getValue());
+ }
+
+ static Metadata read(DataInputPlus in) throws IOException
+ {
+ CRC32 crc = Crc.crc32();
+ int recordsCount = in.readInt();
+ updateChecksumInt(crc, recordsCount);
+ validateCRC(crc, in.readInt());
+ return new Metadata(recordsCount);
+ }
+
+ void persist(Descriptor descriptor)
+ {
+ File tmpFile = descriptor.tmpFileFor(Component.METADATA);
+ try (FileOutputStreamPlus out = new FileOutputStreamPlus(tmpFile))
+ {
+ write(out);
+
+ out.flush();
+ out.sync();
+ }
+ catch (IOException e)
+ {
+ throw new JournalWriteError(descriptor, tmpFile, e);
+ }
+ tmpFile.move(descriptor.fileFor(Component.METADATA));
+ }
+
+ static Metadata load(Descriptor descriptor)
+ {
+ File file = descriptor.fileFor(Component.METADATA);
+ try (FileInputStreamPlus in = new FileInputStreamPlus(file))
+ {
+ return read(in);
+ }
+ catch (IOException e)
+ {
+ throw new JournalReadError(descriptor, file, e);
+ }
+ }
+
+ static <K> Metadata rebuild(Descriptor descriptor, KeySupport<K> keySupport)
+ {
+ int recordsCount = 0;
+
+ try (StaticSegment.SequentialReader<K> reader = StaticSegment.sequentialReader(descriptor, keySupport, Integer.MAX_VALUE))
+ {
+ while (reader.advance())
+ ++recordsCount;
+ }
+ catch (JournalReadError e)
+ {
+ // we expect EOF when rebuilding
+ if (!(e.getCause() instanceof EOFException))
+ throw e;
+ }
+
+ return new Metadata(recordsCount);
+ }
+
+ static <K> Metadata rebuildAndPersist(Descriptor descriptor, KeySupport<K> keySupport)
+ {
+ Metadata metadata = rebuild(descriptor, keySupport);
+ metadata.persist(descriptor);
+ return metadata;
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Metrics.java b/journal/src/main/java/org/apache/cassandra/journal/Metrics.java
new file mode 100644
index 0000000..92c6c19
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Metrics.java
@@ -0,0 +1,10 @@
+package org.apache.cassandra.journal;
+
+final class Metrics<K, V>
+{
+ Metrics(String name) {}
+
+ void register(Flusher flusher) {}
+
+ void deregister() {}
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/OnDiskIndex.java b/journal/src/main/java/org/apache/cassandra/journal/OnDiskIndex.java
new file mode 100644
index 0000000..8c662d8
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/OnDiskIndex.java
@@ -0,0 +1,350 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.StandardOpenOption;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.zip.CRC32;
+import javax.annotation.Nullable;
+
+import accord.utils.Invariants;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.utils.AbstractIterator;
+import org.apache.cassandra.utils.Crc;
+
+import static org.apache.cassandra.journal.Journal.validateCRC;
+import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
+import static org.apache.cassandra.utils.FBUtilities.updateChecksumLong;
+
+/**
+ * An on-disk (memory-mapped) index for a completed flushed segment.
+ * <p/>
+ * TODO (expected): block-level CRC
+ */
+final class OnDiskIndex<K> extends Index<K>
+{
+ private static final long[] EMPTY = new long[0];
+
+ private static final int FILE_PREFIX_SIZE = 4 + 4; // count of entries, CRC
+ private static final int VALUE_SIZE = Long.BYTES; // int offset + int size
+
+ private final int KEY_SIZE;
+ private final int ENTRY_SIZE;
+
+ private final Descriptor descriptor;
+
+ private final FileChannel channel;
+ private volatile MappedByteBuffer buffer;
+ private final int entryCount;
+
+ private volatile K firstId, lastId;
+
+ private OnDiskIndex(
+ Descriptor descriptor, KeySupport<K> keySupport, FileChannel channel, MappedByteBuffer buffer, int entryCount)
+ {
+ super(keySupport);
+
+ this.descriptor = descriptor;
+ this.channel = channel;
+ this.buffer = buffer;
+ this.entryCount = entryCount;
+
+ KEY_SIZE = keySupport.serializedSize(descriptor.userVersion);
+ ENTRY_SIZE = KEY_SIZE + VALUE_SIZE;
+ }
+
+ /**
+ * Open the index for reading, validate CRC
+ */
+ @SuppressWarnings({ "resource", "RedundantSuppression" })
+ static <K> OnDiskIndex<K> open(Descriptor descriptor, KeySupport<K> keySupport)
+ {
+ File file = descriptor.fileFor(Component.INDEX);
+ FileChannel channel = null;
+ MappedByteBuffer buffer = null;
+ try
+ {
+ channel = FileChannel.open(file.toPath(), StandardOpenOption.READ);
+ buffer = channel.map(FileChannel.MapMode.READ_ONLY, 0, channel.size());
+
+ int entryCount = buffer.getInt(0);
+ OnDiskIndex<K> index = new OnDiskIndex<>(descriptor, keySupport, channel, buffer, entryCount);
+ index.validate();
+ index.init();
+ return index;
+ }
+ catch (Throwable e)
+ {
+ FileUtils.clean(buffer);
+ FileUtils.closeQuietly(channel);
+ throw new JournalReadError(descriptor, file, e);
+ }
+ }
+
+ private void init()
+ {
+ if (entryCount > 0)
+ {
+ firstId = keyAtIndex(0);
+ lastId = keyAtIndex(entryCount - 1);
+ }
+ }
+
+ @Override
+ public void close()
+ {
+ try
+ {
+ FileUtils.clean(buffer);
+ buffer = null;
+ channel.close();
+ }
+ catch (IOException e)
+ {
+ throw new JournalWriteError(descriptor, Component.INDEX, e);
+ }
+ }
+
+ void validate() throws IOException
+ {
+ CRC32 crc = Crc.crc32();
+
+ try (DataInputBuffer in = new DataInputBuffer(buffer, true))
+ {
+ int entryCount = in.readInt();
+ updateChecksumInt(crc, entryCount);
+ validateCRC(crc, in.readInt());
+
+ Crc.updateCrc32(crc, buffer, FILE_PREFIX_SIZE, FILE_PREFIX_SIZE + entryCount * ENTRY_SIZE);
+ in.skipBytesFully(entryCount * ENTRY_SIZE);
+ validateCRC(crc, in.readInt());
+
+ if (in.available() != 0)
+ throw new IOException("Trailing data encountered in segment index " + descriptor.fileFor(Component.INDEX));
+ }
+ }
+
+ static <K> void write(
+ NavigableMap<K, long[]> entries, KeySupport<K> keySupport, DataOutputPlus out, int userVersion) throws IOException
+ {
+ CRC32 crc = Crc.crc32();
+
+ int size = entries.values()
+ .stream()
+ .mapToInt(offsets -> offsets.length)
+ .sum();
+ out.writeInt(size);
+ updateChecksumInt(crc, size);
+ out.writeInt((int) crc.getValue());
+
+ for (Map.Entry<K, long[]> entry : entries.entrySet())
+ {
+ long prev = -1;
+ for (long offsetAndSize : entry.getValue())
+ {
+ K key = entry.getKey();
+ keySupport.serialize(key, out, userVersion);
+ keySupport.updateChecksum(crc, key, userVersion);
+
+ if (prev != -1)
+ {
+ long tmp = prev;
+ Invariants.require(readOffset(offsetAndSize) < readOffset(prev),
+ () -> String.format("Offsets should be strictly reverse monotonic, but found %d following %d",
+ readOffset(offsetAndSize), readOffset(tmp)));
+ }
+ out.writeLong(offsetAndSize);
+ updateChecksumLong(crc, offsetAndSize);
+ prev = offsetAndSize;
+ }
+ }
+
+ out.writeInt((int) crc.getValue());
+ }
+
+ @Override
+ @Nullable
+ public K firstId()
+ {
+ return firstId;
+ }
+
+ @Override
+ @Nullable
+ public K lastId()
+ {
+ return lastId;
+ }
+
+ @Override
+ public long[] lookUp(K id)
+ {
+ return lookUpAll(id);
+ }
+
+ @Override
+ public long lookUpLast(K id)
+ {
+ if (!mayContainId(id))
+ return -1L;
+
+ int keyIndex = binarySearch(id);
+ return keyIndex < 0 ? -1 : recordAtIndex(keyIndex);
+ }
+
+ @Override
+ public long[] lookUpAll(K id)
+ {
+ if (!mayContainId(id))
+ return EMPTY;
+
+ int someIndex = binarySearch(id);
+ if (someIndex < 0)
+ return EMPTY;
+
+ int firstKeyIndex = someIndex;
+ while (firstKeyIndex > 0 && id.equals(keyAtIndex(firstKeyIndex - 1)))
+ --firstKeyIndex;
+
+ int lastKeyIndex = someIndex;
+ while (lastKeyIndex + 1 < entryCount && id.equals(keyAtIndex(lastKeyIndex + 1)))
+ ++lastKeyIndex;
+
+ long[] all = new long[lastKeyIndex - firstKeyIndex + 1];
+ int idx = firstKeyIndex;
+ for (int i = 0; i < all.length; i++)
+ all[i] = recordAtIndex(idx++);
+ return all;
+ }
+
+ IndexReader reader()
+ {
+ return new IndexReader();
+ }
+
+ public class IndexReader extends AbstractIterator<K>
+ {
+ int idx;
+ K key;
+ int offset;
+ int size;
+
+ IndexReader()
+ {
+ idx = -1;
+ }
+
+ protected K computeNext()
+ {
+ if (advance())
+ return key;
+ else
+ return endOfData();
+ }
+
+ public int offset()
+ {
+ ensureAdvanced();
+ return offset;
+ }
+
+ public int recordSize()
+ {
+ ensureAdvanced();
+ return size;
+ }
+
+ public boolean advance()
+ {
+ if (idx >= entryCount - 1)
+ return false;
+
+ idx++;
+ key = keyAtIndex(idx);
+ long record = recordAtIndex(idx);
+ offset = Index.readOffset(record);
+ size = Index.readSize(record);
+ return true;
+ }
+
+ private void ensureAdvanced()
+ {
+ if (idx < 0)
+ throw new IllegalStateException("Must call advance() before accessing entry content");
+ }
+ }
+
+ private K keyAtIndex(int index)
+ {
+ return keySupport.deserialize(buffer, FILE_PREFIX_SIZE + index * ENTRY_SIZE, descriptor.userVersion);
+ }
+
+ private long recordAtIndex(int index)
+ {
+ return buffer.getLong(FILE_PREFIX_SIZE + index * ENTRY_SIZE + KEY_SIZE);
+ }
+
+ /*
+ * This has been lifted from {@see IndexSummary}'s implementation,
+ * which itself was lifted from Harmony's Collections implementation.
+ */
+ private int binarySearch(K key)
+ {
+ int low = 0, mid = entryCount, high = mid - 1, result = -1;
+ while (low <= high)
+ {
+ mid = (low + high) >>> 1;
+ result = compareWithKeyAt(key, mid);
+ if (result > 0)
+ {
+ low = mid + 1;
+ }
+ else if (result == 0)
+ {
+ return mid;
+ }
+ else
+ {
+ high = mid - 1;
+ }
+ }
+ return -mid - (result < 0 ? 1 : 2);
+ }
+
+ private int compareWithKeyAt(K key, int keyIndex)
+ {
+ int offset = FILE_PREFIX_SIZE + ENTRY_SIZE * keyIndex;
+ return keySupport.compareWithKeyAt(key, buffer, offset, descriptor.userVersion);
+ }
+
+ static <K> OnDiskIndex<K> rebuildAndPersist(Descriptor descriptor, KeySupport<K> keySupport, int fsyncedLimit)
+ {
+ try (InMemoryIndex<K> index = InMemoryIndex.rebuild(descriptor, keySupport, fsyncedLimit))
+ {
+ index.persist(descriptor);
+ }
+ return open(descriptor, keySupport);
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Params.java b/journal/src/main/java/org/apache/cassandra/journal/Params.java
new file mode 100644
index 0000000..452f024
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Params.java
@@ -0,0 +1,62 @@
+/*
+ * 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.cassandra.journal;
+
+import java.util.concurrent.TimeUnit;
+
+public interface Params
+{
+ enum FlushMode { BATCH, GROUP, PERIODIC }
+
+ enum FailurePolicy { STOP, STOP_JOURNAL, IGNORE, DIE }
+
+ /**
+ * @return maximum segment size
+ */
+ int segmentSize();
+
+ /**
+ * @return this journal's {@link FailurePolicy}
+ */
+ FailurePolicy failurePolicy();
+
+ /**
+ * @return journal flush (sync) mode
+ */
+ FlushMode flushMode();
+
+ boolean enableCompaction();
+
+ long compactionPeriod(TimeUnit units);
+
+ /**
+ * @return milliseconds between journal flushes
+ */
+ long flushPeriod(TimeUnit units);
+
+ /**
+ * @return to block writes for while waiting for a slow fsync to complete
+ * when in {@link FlushMode#PERIODIC} mode
+ */
+ long periodicBlockPeriod(TimeUnit units);
+
+ /**
+ * @return user provided version to use for key and value serialization
+ */
+ int userVersion();
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/RecordConsumer.java b/journal/src/main/java/org/apache/cassandra/journal/RecordConsumer.java
new file mode 100644
index 0000000..22d2bc4
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/RecordConsumer.java
@@ -0,0 +1,26 @@
+/*
+ * 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.cassandra.journal;
+
+import java.nio.ByteBuffer;
+
+@FunctionalInterface
+public interface RecordConsumer<K>
+{
+ void accept(long segment, int position, K key, ByteBuffer buffer, int userVersion);
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Segment.java b/journal/src/main/java/org/apache/cassandra/journal/Segment.java
new file mode 100644
index 0000000..607dc38
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Segment.java
@@ -0,0 +1,148 @@
+/*
+ * 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.cassandra.journal;
+
+import java.nio.ByteBuffer;
+
+import accord.utils.Invariants;
+import org.apache.cassandra.concurrent.ExecutorPlus;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.utils.concurrent.OpOrder;
+import org.apache.cassandra.utils.concurrent.Ref;
+import org.apache.cassandra.utils.concurrent.SelfRefCounted;
+
+public abstract class Segment<K, V> implements SelfRefCounted<Segment<K, V>>, Comparable<Segment<K, V>>
+{
+ protected abstract static class Tidier implements Tidy, Runnable
+ {
+ OpOrder.Barrier await;
+ ExecutorPlus executor;
+
+ abstract void onUnreferenced();
+
+ public final void run()
+ {
+ await.await();
+ onUnreferenced();
+ }
+
+ public final void tidy()
+ {
+ executor.execute(this);
+ }
+ }
+
+ final File file;
+ final Descriptor descriptor;
+ final Metadata metadata;
+ final KeySupport<K> keySupport;
+
+ ByteBuffer buffer;
+
+ Segment(Descriptor descriptor, Metadata metadata, KeySupport<K> keySupport)
+ {
+ this.file = descriptor.fileFor(Component.DATA);
+ this.descriptor = descriptor;
+ this.metadata = metadata;
+ this.keySupport = keySupport;
+ }
+
+ abstract Index<K> index();
+
+ abstract boolean isActive();
+
+ boolean isStatic() { return !isActive(); }
+
+ abstract ActiveSegment<K, V> asActive();
+ abstract StaticSegment<K, V> asStatic();
+
+ public long id()
+ {
+ return descriptor.timestamp;
+ }
+
+ /*
+ * Reading entries (by id, by offset, iterate)
+ */
+
+ boolean readLast(K id, RecordConsumer<K> consumer)
+ {
+ long offsetAndSize = index().lookUpLast(id);
+ if (offsetAndSize == -1)
+ return false;
+
+ EntrySerializer.EntryHolder<K> into = new EntrySerializer.EntryHolder<>();
+ int offset = Index.readOffset(offsetAndSize);
+ int size = Index.readSize(offsetAndSize);
+ if (read(offset, size, into))
+ {
+ Invariants.require(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key);
+ consumer.accept(descriptor.timestamp, offset, id, into.value, descriptor.userVersion);
+ return true;
+ }
+ return false;
+ }
+
+ boolean readLast(K id, EntrySerializer.EntryHolder<K> into)
+ {
+ long offsetAndSize = index().lookUpLast(id);
+ if (offsetAndSize == -1 || !read(Index.readOffset(offsetAndSize), Index.readSize(offsetAndSize), into))
+ return false;
+ Invariants.require(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key);
+ return true;
+ }
+
+ void readAll(K id, EntrySerializer.EntryHolder<K> into, RecordConsumer<K> onEntry)
+ {
+ long[] all = index().lookUpAll(id);
+ int prevOffset = Integer.MAX_VALUE;
+ for (int i = 0; i < all.length; i++)
+ {
+ int offset = Index.readOffset(all[i]);
+ int size = Index.readSize(all[i]);
+ Invariants.require(offset < prevOffset);
+ Invariants.require(read(offset, size, into), "Read should always return true");
+ Invariants.require(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key);
+ onEntry.accept(descriptor.timestamp, offset, into.key, into.value, into.userVersion);
+ }
+ }
+
+ @Override
+ public int compareTo(Segment<K, V> that)
+ {
+ return this.descriptor.compareTo(that.descriptor);
+ }
+
+ abstract boolean read(int offset, int size, EntrySerializer.EntryHolder<K> into);
+
+ abstract void close(Journal<K, V> journal);
+
+ void release(Journal<K, V> journal)
+ {
+ Ref<Segment<K, V>> selfRef = selfRef();
+ Tidier tidier = (Tidier) selfRef.tidier();
+ if (journal != null)
+ {
+ // permitted to be null ONLY for tests
+ tidier.await = journal.readOrder.newBarrier();
+ tidier.await.issue();
+ tidier.executor = journal.tidy;
+ }
+ selfRef.release();
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/SegmentCompactor.java b/journal/src/main/java/org/apache/cassandra/journal/SegmentCompactor.java
new file mode 100644
index 0000000..7b84ea8
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/SegmentCompactor.java
@@ -0,0 +1,34 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.util.Collection;
+
+public interface SegmentCompactor<K, V>
+{
+ SegmentCompactor<?, ?> NOOP = (SegmentCompactor<Object, Object>) (segments) -> segments;
+
+ static <K, V> SegmentCompactor<K, V> noop()
+ {
+ //noinspection unchecked
+ return (SegmentCompactor<K, V>) NOOP;
+ }
+
+ Collection<StaticSegment<K, V>> compact(Collection<StaticSegment<K, V>> segments) throws IOException;
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/Segments.java b/journal/src/main/java/org/apache/cassandra/journal/Segments.java
new file mode 100644
index 0000000..636389e
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/Segments.java
@@ -0,0 +1,246 @@
+/*
+ * 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.cassandra.journal;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
+
+import accord.utils.Invariants;
+import org.apache.cassandra.utils.concurrent.Refs;
+
+/**
+ * Consistent, immutable view of active + static segments
+ *
+ * TODO (performance, expected): an interval/range structure for StaticSegment lookup based on min/max key bounds
+ */
+class Segments<K, V>
+{
+ private final Map<Long, Segment<K, V>> segments;
+ private List<Segment<K, V>> sorted;
+
+ Segments(Map<Long, Segment<K, V>> segments)
+ {
+ this.segments = segments;
+ }
+
+ static <K, V> Segments<K, V> of(Collection<Segment<K, V>> segments)
+ {
+ Map<Long, Segment<K, V>> newSegments = newMap(segments.size());
+ for (Segment<K, V> segment : segments)
+ newSegments.put(segment.descriptor.timestamp, segment);
+ return new Segments<>(newSegments);
+ }
+
+ static <K, V> Segments<K, V> none()
+ {
+ return new Segments<>(emptyMap());
+ }
+
+ Segments<K, V> withNewActiveSegment(ActiveSegment<K, V> activeSegment)
+ {
+ Map<Long, Segment<K, V>> newSegments = new HashMap<>(segments);
+ Segment<K, V> oldValue = newSegments.put(activeSegment.descriptor.timestamp, activeSegment);
+ Invariants.require(oldValue == null);
+ return new Segments<>(newSegments);
+ }
+
+ Segments<K, V> withoutEmptySegment(ActiveSegment<K, V> activeSegment)
+ {
+ Map<Long, Segment<K, V>> newSegments = new HashMap<>(segments);
+ Segment<K, V> oldValue = newSegments.remove(activeSegment.descriptor.timestamp);
+ Invariants.require(oldValue.asActive().isEmpty());
+ return new Segments<>(newSegments);
+ }
+
+ Segments<K, V> withCompletedSegment(ActiveSegment<K, V> activeSegment, StaticSegment<K, V> staticSegment)
+ {
+ Invariants.requireArgument(activeSegment.descriptor.equals(staticSegment.descriptor));
+ Map<Long, Segment<K, V>> newSegments = new HashMap<>(segments);
+ Segment<K, V> oldValue = newSegments.put(staticSegment.descriptor.timestamp, staticSegment);
+ Invariants.require(oldValue == activeSegment, () -> String.format("old value %s != new %s", oldValue, activeSegment));
+ return new Segments<>(newSegments);
+ }
+
+ Segments<K, V> withCompactedSegments(Collection<StaticSegment<K, V>> oldSegments, Collection<StaticSegment<K, V>> compactedSegments)
+ {
+ Map<Long, Segment<K, V>> newSegments = new HashMap<>(segments);
+ for (StaticSegment<K, V> oldSegment : oldSegments)
+ {
+ Segment<K, V> oldValue = newSegments.remove(oldSegment.descriptor.timestamp);
+ Invariants.require(oldValue == oldSegment);
+ }
+
+ for (StaticSegment<K, V> compactedSegment : compactedSegments)
+ {
+ Segment<K, V> oldValue = newSegments.put(compactedSegment.descriptor.timestamp, compactedSegment);
+ Invariants.require(oldValue == null);
+ }
+
+ return new Segments<>(newSegments);
+ }
+
+ Iterable<Segment<K, V>> all()
+ {
+ return this.segments.values();
+ }
+
+ /**
+ * Returns segments in timestamp order. Will allocate and sort the segment collection.
+ */
+ List<Segment<K, V>> allSorted(boolean asc)
+ {
+ if (sorted == null)
+ {
+ sorted = new ArrayList<>(segments.values());
+ Collections.sort(sorted);
+ }
+
+ if (asc)
+ return sorted;
+
+ List<Segment<K, V>> reversed = new ArrayList<>(sorted);
+ Collections.reverse(reversed);
+ return reversed;
+ }
+
+ void selectActive(long maxTimestamp, Collection<ActiveSegment<K, V>> into)
+ {
+ for (Segment<K, V> segment : segments.values())
+ if (segment.isActive() && segment.descriptor.timestamp <= maxTimestamp)
+ into.add(segment.asActive());
+ }
+
+ boolean isSwitched(ActiveSegment<K, V> active)
+ {
+ for (Segment<K, V> segment : segments.values())
+ if (segment.isStatic() && active.descriptor.equals(segment.descriptor))
+ return true;
+
+ return false;
+ }
+
+ ActiveSegment<K, V> oldestActive()
+ {
+ List<Segment<K, V>> sorted = allSorted(true);
+ for (int i = 0 ; i < sorted.size() ; ++i)
+ {
+ Segment<K, V> segment = sorted.get(i);
+ if (segment.isActive())
+ return segment.asActive();
+ }
+ return null;
+ }
+
+ Segment<K, V> get(long timestamp)
+ {
+ return segments.get(timestamp);
+ }
+
+ void selectStatic(Collection<StaticSegment<K, V>> into)
+ {
+ for (Segment<K, V> segment : segments.values())
+ if (segment.isStatic())
+ into.add(segment.asStatic());
+ }
+
+ /**
+ * Select segments that could potentially have an entry with the specified ids and
+ * attempt to grab references to them all.
+ *
+ * @return a subset of segments with references to them, or {@code null} if failed to grab the refs
+ */
+ ReferencedSegments<K, V> selectAndReference(Predicate<Segment<K, V>> test)
+ {
+ Map<Long, Segment<K, V>> selectedSegments = select(test).segments;
+ Refs<Segment<K, V>> refs = null;
+ if (!selectedSegments.isEmpty())
+ {
+ refs = Refs.tryRef(selectedSegments.values());
+ if (null == refs)
+ return null;
+ }
+ return new ReferencedSegments<>(selectedSegments, refs);
+ }
+
+ /**
+ * Select segments that could potentially have an entry with the specified ids and
+ * attempt to grab references to them all.
+ *
+ * @return a subset of segments with references to them, or {@code null} if failed to grab the refs
+ */
+ Segments<K, V> select(Predicate<Segment<K, V>> test)
+ {
+ Map<Long, Segment<K, V>> selectedSegments = null;
+ for (Segment<K, V> segment : segments.values())
+ {
+ if (test.test(segment))
+ {
+ if (null == selectedSegments)
+ selectedSegments = newMap(10);
+ selectedSegments.put(segment.descriptor.timestamp, segment);
+ }
+ }
+
+ if (null == selectedSegments)
+ selectedSegments = emptyMap();
+
+ return new Segments<>(selectedSegments);
+ }
+
+ static class ReferencedSegments<K, V> extends Segments<K, V> implements AutoCloseable
+ {
+ private final Refs<Segment<K, V>> refs;
+
+ ReferencedSegments(Map<Long, Segment<K, V>> segments, Refs<Segment<K, V>> refs)
+ {
+ super(segments);
+ this.refs = refs;
+ }
+
+ public int count()
+ {
+ if (refs == null) return 0;
+ else return refs.size();
+ }
+
+ @Override
+ public void close()
+ {
+ if (null != refs)
+ refs.release();
+ }
+ }
+
+ private static final Map<?, ?> EMPTY_MAP = Collections.emptyMap();
+
+ @SuppressWarnings("unchecked")
+ private static <K, V> Map<K, V> emptyMap()
+ {
+ return (Map<K, V>) EMPTY_MAP;
+ }
+
+ private static <K, V> Map<K, V> newMap(int expectedSize)
+ {
+ return new HashMap<>(Math.max(16, expectedSize));
+ }
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/StaticSegment.java b/journal/src/main/java/org/apache/cassandra/journal/StaticSegment.java
new file mode 100644
index 0000000..cf4e22f
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/StaticSegment.java
@@ -0,0 +1,450 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.utils.Closeable;
+import org.apache.cassandra.utils.Throwables;
+import org.apache.cassandra.utils.concurrent.Ref;
+
+/**
+ * An immutable data segment that is no longer written to.
+ * <p>
+ * Can be compacted with input from {@code PersistedInvalidations} into a new smaller segment,
+ * with invalidated entries removed.
+ */
+public final class StaticSegment<K, V> extends Segment<K, V>
+{
+ final FileChannel channel;
+ final int fsyncLimit;
+
+ private final Ref<Segment<K, V>> selfRef;
+
+ private final OnDiskIndex<K> index;
+
+ private StaticSegment(Descriptor descriptor,
+ FileChannel channel,
+ MappedByteBuffer buffer,
+ OnDiskIndex<K> index,
+ Metadata metadata,
+ KeySupport<K> keySupport)
+ {
+ super(descriptor, metadata, keySupport);
+ this.index = index;
+
+ this.channel = channel;
+ this.fsyncLimit = metadata.fsyncLimit();
+ this.buffer = buffer;
+
+ selfRef = new Ref<>(this, new Tidier<>(descriptor, channel, buffer, index));
+ }
+
+ /**
+ * Loads all segments matching the supplied desctiptors
+ *
+ * @param descriptors descriptors of the segments to load
+ * @return list of the loaded segments
+ */
+ static <K, V> List<Segment<K, V>> open(Collection<Descriptor> descriptors, KeySupport<K> keySupport)
+ {
+ List<Segment<K, V>> segments = new ArrayList<>(descriptors.size());
+ for (Descriptor descriptor : descriptors)
+ segments.add(open(descriptor, keySupport));
+ return segments;
+ }
+
+ /**
+ * Load the segment corresponding to the provided desrciptor
+ *
+ * @param descriptor descriptor of the segment to load
+ * @return the loaded segment
+ */
+ @SuppressWarnings({ "resource", "RedundantSuppression" })
+ static <K, V> StaticSegment<K, V> open(Descriptor descriptor, KeySupport<K> keySupport)
+ {
+ if (!Component.DATA.existsFor(descriptor))
+ throw new IllegalArgumentException("Data file for segment " + descriptor + " doesn't exist");
+
+ Metadata metadata = Component.METADATA.existsFor(descriptor)
+ ? Metadata.load(descriptor)
+ : Metadata.rebuildAndPersist(descriptor, keySupport);
+
+ OnDiskIndex<K> index = Component.INDEX.existsFor(descriptor)
+ ? OnDiskIndex.open(descriptor, keySupport)
+ : OnDiskIndex.rebuildAndPersist(descriptor, keySupport, metadata.fsyncLimit());
+
+ try
+ {
+ return internalOpen(descriptor, index, metadata, keySupport);
+ }
+ catch (IOException e)
+ {
+ throw new JournalReadError(descriptor, Component.DATA, e);
+ }
+ }
+
+ private static <K, V> StaticSegment<K, V> internalOpen(
+ Descriptor descriptor, OnDiskIndex<K> index, Metadata metadata, KeySupport<K> keySupport)
+ throws IOException
+ {
+ File file = descriptor.fileFor(Component.DATA);
+ FileChannel channel = FileChannel.open(file.toPath(), StandardOpenOption.READ);
+ MappedByteBuffer buffer = channel.map(FileChannel.MapMode.READ_ONLY, 0, channel.size());
+ return new StaticSegment<>(descriptor, channel, buffer, index, metadata, keySupport);
+ }
+
+ public void close(Journal<K, V> journal)
+ {
+ release(journal);
+ }
+
+ /**
+ * Waits until this segment is unreferenced, closes it, and deltes all files associated with it.
+ */
+ void discard(Journal<K, V> journal)
+ {
+ ((Tidier)selfRef.tidier()).discard = true;
+ close(journal);
+ }
+
+ @Override
+ public Ref<Segment<K, V>> tryRef()
+ {
+ return selfRef.tryRef();
+ }
+
+ @Override
+ public Ref<Segment<K, V>> ref()
+ {
+ return selfRef.ref();
+ }
+
+ @Override
+ public String toString()
+ {
+ return "StaticSegment{" + descriptor + '}';
+ }
+
+ @Override
+ public Ref<Segment<K, V>> selfRef()
+ {
+ return selfRef;
+ }
+
+ private static final class Tidier<K> extends Segment.Tidier implements Tidy
+ {
+ private final Descriptor descriptor;
+ private final FileChannel channel;
+ private final ByteBuffer buffer;
+ private final Index<K> index;
+ boolean discard;
+
+ Tidier(Descriptor descriptor, FileChannel channel, ByteBuffer buffer, Index<K> index)
+ {
+ this.descriptor = descriptor;
+ this.channel = channel;
+ this.buffer = buffer;
+ this.index = index;
+ }
+
+ @Override
+ void onUnreferenced()
+ {
+ FileUtils.clean(buffer);
+ FileUtils.closeQuietly(channel);
+ index.close();
+ if (discard)
+ {
+ Throwable fail = null;
+ for (Component component : Component.VALUES)
+ {
+ try { descriptor.fileFor(component).deleteIfExists(); }
+ catch (Throwable t) { fail = Throwables.merge(fail, t); }
+ }
+ Throwables.maybeFail(fail);
+ }
+ }
+
+ @Override
+ public String name()
+ {
+ return descriptor.toString();
+ }
+ }
+
+ @Override
+ OnDiskIndex<K> index()
+ {
+ return index;
+ }
+
+ @Override
+ boolean isActive()
+ {
+ return false;
+ }
+
+ @Override
+ ActiveSegment<K, V> asActive()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ StaticSegment<K, V> asStatic()
+ {
+ return this;
+ }
+
+ /**
+ * Read the entry and specified offset into the entry holder.
+ * Expects the record to have been written at this offset, but potentially not flushed and lost.
+ */
+ @Override
+ boolean read(int offset, int size, EntrySerializer.EntryHolder<K> into)
+ {
+ ByteBuffer duplicate = buffer.duplicate().position(offset).limit(offset + size);
+ try
+ {
+ return 0 <= EntrySerializer.tryRead(into, keySupport, duplicate, fsyncLimit, descriptor.userVersion);
+ }
+ catch (IOException e)
+ {
+ throw new JournalReadError(descriptor, file, e);
+ }
+ }
+
+ /**
+ * Iterate over and invoke the supplied callback on every record.
+ */
+ void forEachRecord(RecordConsumer<K> consumer)
+ {
+ try (SequentialReader<K> reader = sequentialReader(descriptor, keySupport, fsyncLimit))
+ {
+ while (reader.advance())
+ {
+ consumer.accept(descriptor.timestamp, reader.offset(), reader.key(), reader.record(), descriptor.userVersion);
+ }
+ }
+ }
+
+ /*
+ * Sequential and in-key order reading (replay and components rebuild)
+ */
+
+ static abstract class Reader<K> implements Closeable
+ {
+ enum State { RESET, ADVANCED, EOF }
+
+ public final Descriptor descriptor;
+ protected final KeySupport<K> keySupport;
+
+ protected final File file;
+ protected final FileChannel channel;
+ protected final MappedByteBuffer buffer;
+
+ protected final EntrySerializer.EntryHolder<K> holder = new EntrySerializer.EntryHolder<>();
+ protected int offset = -1;
+ protected State state = State.RESET;
+
+ Reader(Descriptor descriptor, KeySupport<K> keySupport)
+ {
+ this.descriptor = descriptor;
+ this.keySupport = keySupport;
+
+ file = descriptor.fileFor(Component.DATA);
+ try
+ {
+ channel = file.newReadChannel();
+ buffer = channel.map(FileChannel.MapMode.READ_ONLY, 0, channel.size());
+ }
+ catch (NoSuchFileException e)
+ {
+ throw new IllegalArgumentException("Data file for segment " + descriptor + " doesn't exist");
+ }
+ catch (IOException e)
+ {
+ throw new JournalReadError(descriptor, file, e);
+ }
+ }
+
+ @Override
+ public void close()
+ {
+ FileUtils.closeQuietly(channel);
+ FileUtils.clean(buffer);
+ }
+
+ public abstract boolean advance();
+
+ public int offset()
+ {
+ ensureHasAdvanced();
+ return offset;
+ }
+
+ public K key()
+ {
+ ensureHasAdvanced();
+ return holder.key;
+ }
+
+ public ByteBuffer record()
+ {
+ ensureHasAdvanced();
+ return holder.value;
+ }
+
+ protected void ensureHasAdvanced()
+ {
+ if (state != State.ADVANCED)
+ throw new IllegalStateException("Must call advance() before accessing entry content");
+ }
+
+ protected boolean eof()
+ {
+ state = State.EOF;
+ return false;
+ }
+ }
+
+ static <K> SequentialReader<K> sequentialReader(Descriptor descriptor, KeySupport<K> keySupport, int fsyncedLimit)
+ {
+ return new SequentialReader<>(descriptor, keySupport, fsyncedLimit);
+ }
+
+ /**
+ * A sequential data segment reader to use for journal replay and rebuilding
+ * missing auxilirary components (index and metadata).
+ * </p>
+ * Unexpected EOF and CRC mismatches in synced portions of segments are treated
+ * strictly, throwing {@link JournalReadError}. Errors encountered in unsynced portions
+ * of segments are treated as segment EOF.
+ */
+ static final class SequentialReader<K> extends Reader<K>
+ {
+ private final int fsyncedLimit; // exclusive
+
+ SequentialReader(Descriptor descriptor, KeySupport<K> keySupport, int fsyncedLimit)
+ {
+ super(descriptor, keySupport);
+ this.fsyncedLimit = fsyncedLimit;
+ }
+
+ @Override
+ public boolean advance()
+ {
+ if (state == State.EOF)
+ return false;
+
+ reset();
+ return buffer.hasRemaining() ? doAdvance() : eof();
+ }
+
+ private boolean doAdvance()
+ {
+ offset = buffer.position();
+ try
+ {
+ int length = EntrySerializer.tryRead(holder, keySupport, buffer.duplicate(), fsyncedLimit, descriptor.userVersion);
+ if (length < 0)
+ return eof();
+ buffer.position(offset + length);
+ }
+ catch (IOException e)
+ {
+ throw new JournalReadError(descriptor, file, e);
+ }
+
+ state = State.ADVANCED;
+ return true;
+ }
+
+ private void reset()
+ {
+ offset = -1;
+ holder.clear();
+ state = State.RESET;
+ }
+ }
+
+ public StaticSegment.KeyOrderReader<K> keyOrderReader()
+ {
+ return new StaticSegment.KeyOrderReader<>(descriptor, keySupport, index.reader());
+ }
+
+ public static final class KeyOrderReader<K> extends Reader<K> implements Comparable<KeyOrderReader<K>>
+ {
+ private final OnDiskIndex<K>.IndexReader indexReader;
+
+ KeyOrderReader(Descriptor descriptor, KeySupport<K> keySupport, OnDiskIndex<K>.IndexReader indexReader)
+ {
+ super(descriptor, keySupport);
+ this.indexReader = indexReader;
+ }
+
+ @Override
+ public boolean advance()
+ {
+ if (!indexReader.advance())
+ return eof();
+
+ offset = indexReader.offset();
+
+ buffer.limit(offset + indexReader.recordSize())
+ .position(offset);
+ try
+ {
+ EntrySerializer.read(holder, keySupport, buffer, descriptor.userVersion);
+ }
+ catch (IOException e)
+ {
+ throw new JournalReadError(descriptor, file, e);
+ }
+
+ state = State.ADVANCED;
+ return true;
+ }
+
+ @Override
+ public int compareTo(KeyOrderReader<K> that)
+ {
+ this.ensureHasAdvanced();
+ that.ensureHasAdvanced();
+
+ int cmp = keySupport.compare(this.key(), that.key());
+ if (cmp != 0)
+ return cmp;
+ cmp = Long.compare(that.descriptor.timestamp, this.descriptor.timestamp);
+ if (cmp != 0)
+ return cmp;
+ return Integer.compare(that.offset, this.offset);
+ }
+ }
+}
\ No newline at end of file
diff --git a/journal/src/main/java/org/apache/cassandra/journal/ValueSerializer.java b/journal/src/main/java/org/apache/cassandra/journal/ValueSerializer.java
new file mode 100644
index 0000000..69690d3
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/ValueSerializer.java
@@ -0,0 +1,34 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+
+public interface ValueSerializer<K, V>
+{
+ void serialize(K key, V value, DataOutputPlus out, int userVersion) throws IOException;
+
+ /**
+ * Deserialize the value given the key is known. Allows to avoid serializing
+ * redundant information in values, if it can be derived from keys.
+ */
+ V deserialize(K key, DataInputPlus in, int userVersion) throws IOException;
+}
diff --git a/journal/src/main/java/org/apache/cassandra/journal/package-info.java b/journal/src/main/java/org/apache/cassandra/journal/package-info.java
new file mode 100644
index 0000000..5ae20b9
--- /dev/null
+++ b/journal/src/main/java/org/apache/cassandra/journal/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * TODO
+ */
+package org.apache.cassandra.journal;
diff --git a/journal/src/test/java/org/apache/cassandra/journal/DescriptorTest.java b/journal/src/test/java/org/apache/cassandra/journal/DescriptorTest.java
new file mode 100644
index 0000000..2c644cf
--- /dev/null
+++ b/journal/src/test/java/org/apache/cassandra/journal/DescriptorTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import org.junit.Test;
+
+import org.apache.cassandra.io.util.File;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class DescriptorTest
+{
+ @Test
+ public void serde() throws IOException
+ {
+ for (Descriptor descriptor : descriptors())
+ assertThat(Descriptor.fromFile(descriptor.fileFor(Component.DATA))).isEqualTo(descriptor);
+ }
+
+ @Test
+ public void isTmp() throws IOException
+ {
+ for (Descriptor descriptor : descriptors())
+ {
+ for (Component comp : Component.values())
+ {
+ assertThat(Descriptor.isTmpFile(descriptor.tmpFileFor(comp))).isTrue();
+ assertThat(Descriptor.isTmpFile(descriptor.fileFor(comp))).isFalse();
+ }
+ }
+ }
+
+ @Test
+ public void list() throws IOException
+ {
+ for (int i = 0; i < 32; i++)
+ {
+ File dir = tempDir();
+ List<Descriptor> expected = descriptors(dir, i % 8 + 1);
+ for (Descriptor descriptor : expected)
+ descriptor.fileFor(Component.DATA).createFileIfNotExists();
+ assertThat(Descriptor.list(dir)).containsExactlyInAnyOrderElementsOf(expected);
+ }
+ }
+
+ @Test
+ public void order() throws IOException
+ {
+ List<Descriptor> descriptors = descriptors();
+ Collections.shuffle(descriptors, new Random(1));
+ Collections.sort(descriptors);
+
+ Descriptor last = descriptors.get(0);
+ for (int i = 1; i < descriptors.size(); i++)
+ {
+ Descriptor current = descriptors.get(i);
+ assertThat(current.directory).isEqualTo(last.directory);
+ assertThat(current.timestamp).isGreaterThanOrEqualTo(last.timestamp);
+ if (current.timestamp == last.timestamp)
+ assertThat(current.generation).isGreaterThanOrEqualTo(last.generation);
+ if (current.timestamp == last.timestamp && current.generation == last.generation)
+ assertThat(current.journalVersion).isGreaterThanOrEqualTo(last.journalVersion);
+ if (current.timestamp == last.timestamp
+ && current.generation == last.generation
+ && current.journalVersion == last.journalVersion)
+ assertThat(current.userVersion).isGreaterThanOrEqualTo(last.userVersion);
+ last = current;
+ }
+ }
+
+ private static List<Descriptor> descriptors() throws IOException
+ {
+ return descriptors(tempDir(), 64);
+ }
+
+ private static List<Descriptor> descriptors(File dir, int count)
+ {
+ Random random = new Random(1);
+ List<Descriptor> descriptors = new ArrayList<>(count);
+ for (int i = 0; i < count; i++)
+ descriptors.add(new Descriptor(dir, random.nextInt(8), random.nextInt(8), random.nextInt(8), random.nextInt(8)));
+ return descriptors;
+ }
+
+ private static File tempDir() throws IOException
+ {
+ File dir = new File(Files.createTempDirectory("descriptor-test"));
+ dir.deleteRecursiveOnExit();
+ return dir;
+ }
+}
diff --git a/journal/src/test/java/org/apache/cassandra/journal/IndexTest.java b/journal/src/test/java/org/apache/cassandra/journal/IndexTest.java
new file mode 100644
index 0000000..0c665f2
--- /dev/null
+++ b/journal/src/test/java/org/apache/cassandra/journal/IndexTest.java
@@ -0,0 +1,307 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.Test;
+
+import accord.utils.Invariants;
+import org.agrona.collections.IntHashSet;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.TimeUUID;
+import org.quicktheories.core.Gen;
+import org.quicktheories.impl.Constraint;
+
+import static org.apache.cassandra.journal.Index.composeOffsetAndSize;
+import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.quicktheories.QuickTheory.qt;
+
+public class IndexTest
+{
+ private static final long[] EMPTY = new long[0];
+
+ @Test
+ public void testInMemoryIndexBasics()
+ {
+ InMemoryIndex<TimeUUID> index = InMemoryIndex.create(TimeUUIDKeySupport.INSTANCE);
+
+ TimeUUID key0 = nextTimeUUID();
+ TimeUUID key1 = nextTimeUUID();
+ TimeUUID key2 = nextTimeUUID();
+ TimeUUID key3 = nextTimeUUID();
+ TimeUUID key4 = nextTimeUUID();
+
+ assertArrayEquals(EMPTY, index.lookUp(key0));
+ assertArrayEquals(EMPTY, index.lookUp(key1));
+ assertArrayEquals(EMPTY, index.lookUp(key2));
+ assertArrayEquals(EMPTY, index.lookUp(key3));
+ assertArrayEquals(EMPTY, index.lookUp(key4));
+
+ int val11 = 1100;
+ int val21 = 2100;
+ int val22 = 2200;
+ int val31 = 3100;
+ int val32 = 3200;
+ int val33 = 3300;
+
+ index.update(key1, val11, 1);
+ index.update(key2, val21, 2);
+ index.update(key2, val22, 3);
+ index.update(key3, val31, 4);
+ index.update(key3, val32, 5);
+ index.update(key3, val33, 6);
+
+ assertArrayEquals(EMPTY, index.lookUp(key0));
+
+ assertArrayEquals(new long[] { composeOffsetAndSize(val11, 1) }, index.lookUp(key1));
+ assertArrayEquals(new long[] { composeOffsetAndSize(val22, 3), composeOffsetAndSize(val21, 2) }, index.lookUp(key2));
+ assertArrayEquals(new long[] { composeOffsetAndSize(val33, 6), composeOffsetAndSize(val32, 5), composeOffsetAndSize(val31, 4) }, index.lookUp(key3));
+ assertArrayEquals(EMPTY, index.lookUp(key4));
+
+ assertEquals(key1, index.firstId());
+ assertEquals(key3, index.lastId());
+
+ assertFalse(index.mayContainId(key0));
+ assertTrue(index.mayContainId(key1));
+ assertTrue(index.mayContainId(key2));
+ assertTrue(index.mayContainId(key3));
+ assertFalse(index.mayContainId(key4));
+ }
+
+ @Test
+ public void testInMemoryIndexPersists() throws IOException
+ {
+ InMemoryIndex<TimeUUID> inMemory = InMemoryIndex.create(TimeUUIDKeySupport.INSTANCE);
+
+ TimeUUID key0 = nextTimeUUID();
+ TimeUUID key1 = nextTimeUUID();
+ TimeUUID key2 = nextTimeUUID();
+ TimeUUID key3 = nextTimeUUID();
+ TimeUUID key4 = nextTimeUUID();
+
+ int val11 = 1100;
+ int val21 = 2100;
+ int val22 = 2200;
+ int val31 = 3100;
+ int val32 = 3200;
+ int val33 = 3300;
+
+ inMemory.update(key1, val11, 1);
+ inMemory.update(key2, val21, 2);
+ inMemory.update(key2, val22, 3);
+ inMemory.update(key3, val31, 4);
+ inMemory.update(key3, val32, 5);
+ inMemory.update(key3, val33, 6);
+
+ File directory = new File(Files.createTempDirectory(null));
+ directory.deleteOnExit();
+ Descriptor descriptor = Descriptor.create(directory, System.currentTimeMillis(), 1);
+ inMemory.persist(descriptor);
+
+ try (OnDiskIndex<TimeUUID> onDisk = OnDiskIndex.open(descriptor, TimeUUIDKeySupport.INSTANCE))
+ {
+ assertArrayEquals(EMPTY, onDisk.lookUp(key0));
+ assertArrayEquals(new long[] { composeOffsetAndSize(val11, 1) }, onDisk.lookUp(key1));
+ assertArrayEquals(new long[] { composeOffsetAndSize(val22, 3), composeOffsetAndSize(val21, 2) }, onDisk.lookUp(key2));
+ assertArrayEquals(new long[] { composeOffsetAndSize(val33, 6), composeOffsetAndSize(val32, 5), composeOffsetAndSize(val31, 4) }, onDisk.lookUp(key3));
+ assertArrayEquals(EMPTY, onDisk.lookUp(key4));
+
+ assertEquals(key1, onDisk.firstId());
+ assertEquals(key3, onDisk.lastId());
+
+ assertFalse(onDisk.mayContainId(key0));
+ assertTrue(onDisk.mayContainId(key1));
+ assertTrue(onDisk.mayContainId(key2));
+ assertTrue(onDisk.mayContainId(key3));
+ assertFalse(onDisk.mayContainId(key4));
+ }
+ }
+
+ @Test
+ public void prop() throws IOException
+ {
+ Constraint sizeConstraint = Constraint.between(1, 10);
+ Constraint valueSizeConstraint = Constraint.between(0, 10);
+ Constraint positionConstraint = Constraint.between(0, Integer.MAX_VALUE);
+ Gen<TimeUUID> keyGen = rs -> nextTimeUUID();
+ Gen<long[]> valueGen = rs -> {
+ long[] array = new long[(int) rs.next(valueSizeConstraint)];
+ IntHashSet uniq = new IntHashSet();
+ for (int i = 0 ; i < array.length ; ++i)
+ {
+ int offset = (int) rs.next(positionConstraint);
+ while (!uniq.add(offset))
+ offset = (int) rs.next(positionConstraint);
+ array[i] = Index.composeOffsetAndSize(offset, (int) rs.next(positionConstraint));
+ }
+
+ Arrays.sort(array);
+ for (int i = 0 ; i < array.length / 2 ; ++i)
+ {
+ int back = array.length - (1 + i);
+ long v = array[i];
+ array[i] = array[back];
+ array[back] = v;
+ }
+
+ return array;
+ };
+ Gen<Map<TimeUUID, long[]>> gen = rs -> {
+ int size = (int) rs.next(sizeConstraint);
+ Map<TimeUUID, long[]> map = Maps.newHashMapWithExpectedSize(size);
+ for (int i = 0; i < size; i++)
+ {
+ TimeUUID key = keyGen.generate(rs);
+ while (map.containsKey(key))
+ key = keyGen.generate(rs);
+ long[] value = valueGen.generate(rs);
+ map.put(key, value);
+ }
+ return map;
+ };
+ gen = gen.describedAs(map -> {
+ StringBuilder sb = new StringBuilder();
+ for (Map.Entry<TimeUUID, long[]> entry : map.entrySet())
+ sb.append('\n').append(entry.getKey()).append('\t').append(Arrays.toString(entry.getValue()));
+ return sb.toString();
+ });
+ File directory = new File(Files.createTempDirectory(null));
+ directory.deleteOnExit();
+ qt().withFixedSeed(185124544959375L).forAll(gen).checkAssert(map -> test(directory, map));
+ }
+
+ private static void test(File directory, Map<TimeUUID, long[]> map)
+ {
+ InMemoryIndex<TimeUUID> inMemory = InMemoryIndex.create(TimeUUIDKeySupport.INSTANCE);
+ for (Map.Entry<TimeUUID, long[]> e : map.entrySet())
+ {
+ TimeUUID key = e.getKey();
+ assertThat(inMemory.lookUp(key)).isEmpty();
+
+ long[] value = e.getValue();
+ if (value.length == 0)
+ continue;
+ for (long i : value)
+ inMemory.update(key, Index.readOffset(i), Index.readSize(i));
+ for (int i = 1 ; i < value.length ; ++i)
+ Invariants.require(value[i - 1] > value[i]);
+ }
+ assertIndex(map, inMemory);
+
+ Descriptor descriptor = Descriptor.create(directory, System.nanoTime(), 1);
+ inMemory.persist(descriptor);
+
+ try (OnDiskIndex<TimeUUID> onDisk = OnDiskIndex.open(descriptor, TimeUUIDKeySupport.INSTANCE))
+ {
+ assertIndex(map, onDisk);
+
+ List<Pair<TimeUUID, Long>> sortedEntries = new ArrayList<>();
+ for (Map.Entry<TimeUUID, long[]> entry : new TreeMap<>(map).entrySet())
+ {
+ for (long l : entry.getValue())
+ sortedEntries.add(Pair.create(entry.getKey(), l));
+ }
+
+ OnDiskIndex<TimeUUID>.IndexReader iter = onDisk.reader();
+ Iterator<Pair<TimeUUID, Long>> expectedIter = sortedEntries.iterator();
+ while (iter.hasNext())
+ {
+ Pair<TimeUUID, Long> expected = expectedIter.next();
+ Assert.assertEquals(iter.next(), expected.left);
+ Assert.assertEquals(iter.recordSize(), Index.readSize(expected.right));
+ Assert.assertEquals(iter.offset(), Index.readOffset(expected.right));
+ }
+ }
+ }
+
+ private static void assertIndex(Map<TimeUUID, long[]> expected, Index<TimeUUID> actual)
+ {
+ List<TimeUUID> keys = expected.entrySet()
+ .stream()
+ .filter(e -> e.getValue().length != 0)
+ .map(Map.Entry::getKey)
+ .sorted()
+ .collect(Collectors.toList());
+
+ if (keys.isEmpty())
+ {
+ assertThat(actual.firstId()).describedAs("Index %s had wrong firstId", actual).isNull();
+ assertThat(actual.lastId()).describedAs("Index %s had wrong lastId", actual).isNull();
+ }
+ else
+ {
+ assertThat(actual.firstId()).describedAs("Index %s had wrong firstId", actual).isEqualTo(keys.get(0));
+ assertThat(actual.lastId()).describedAs("Index %s had wrong lastId", actual).isEqualTo(keys.get(keys.size() - 1));
+ }
+
+ for (Map.Entry<TimeUUID, long[]> e : expected.entrySet())
+ {
+ TimeUUID key = e.getKey();
+ long[] value = e.getValue();
+ long[] read = actual.lookUp(key);
+
+ if (!Arrays.equals(value, read))
+ actual.lookUp(key);
+
+ if (value.length == 0)
+ {
+ assertThat(read).describedAs("Index %s returned wrong values for %s", actual, key).isEmpty();
+ }
+ else
+ {
+ assertThat(read).describedAs("Index %s returned wrong values for %s", actual, key).isEqualTo(value);
+ assertThat(actual.mayContainId(key)).describedAs("Index %s expected %s to exist", key, actual).isTrue();
+ }
+ }
+ }
+
+ @Test
+ public void testHelperMethods()
+ {
+ Random r = new Random();
+ for (int i = 0; i < 1000000; i++)
+ {
+ long record = 0;
+ int size = Math.abs(r.nextInt());
+ record = Index.writeSize(record, size);
+ int offset = Math.abs(r.nextInt());
+ record = Index.writeOffset(record, offset);
+ assertEquals(size, Index.readSize(record));
+ assertEquals(offset, Index.readOffset(record));
+ assertEquals(record, composeOffsetAndSize(offset, size));
+ }
+ }
+}
\ No newline at end of file
diff --git a/journal/src/test/java/org/apache/cassandra/journal/JournalTest.java b/journal/src/test/java/org/apache/cassandra/journal/JournalTest.java
new file mode 100644
index 0000000..6502634
--- /dev/null
+++ b/journal/src/test/java/org/apache/cassandra/journal/JournalTest.java
@@ -0,0 +1,296 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.UnaryOperator;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.cassandra.harry.checker.TestHelper;
+import org.apache.cassandra.harry.gen.EntropySource;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.utils.TimeUUID;
+import org.apache.cassandra.utils.concurrent.CountDownLatch;
+
+import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
+import static org.junit.Assert.assertEquals;
+
+public class JournalTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(JournalTest.class);
+
+ @Test
+ public void testSimpleReadWrite() throws Throwable
+ {
+ File directory = new File(Files.createTempDirectory("JournalTest"));
+ directory.deleteRecursiveOnExit();
+
+ Journal<TimeUUID, Long> journal =
+ new Journal<>("TestJournal", directory, TestParams.INSTANCE, TimeUUIDKeySupport.INSTANCE, LongSerializer.INSTANCE, SegmentCompactor.noop());
+
+ journal.start();
+
+ TimeUUID id1 = nextTimeUUID();
+ TimeUUID id2 = nextTimeUUID();
+ TimeUUID id3 = nextTimeUUID();
+ TimeUUID id4 = nextTimeUUID();
+
+ journal.blockingWrite(id1, 1L);
+ journal.blockingWrite(id2, 2L);
+ journal.blockingWrite(id3, 3L);
+ journal.blockingWrite(id4, 4L);
+
+ assertEquals(1L, (long) journal.readLast(id1));
+ assertEquals(2L, (long) journal.readLast(id2));
+ assertEquals(3L, (long) journal.readLast(id3));
+ assertEquals(4L, (long) journal.readLast(id4));
+
+ journal.shutdown();
+ journal.awaitTermination(10, TimeUnit.SECONDS);
+ journal = new Journal<>("TestJournal", directory, TestParams.INSTANCE, TimeUUIDKeySupport.INSTANCE, LongSerializer.INSTANCE, SegmentCompactor.noop());
+ journal.start();
+
+ assertEquals(1L, (long) journal.readLast(id1));
+ assertEquals(2L, (long) journal.readLast(id2));
+ assertEquals(3L, (long) journal.readLast(id3));
+ assertEquals(4L, (long) journal.readLast(id4));
+
+ journal.shutdown();
+ }
+
+ @Test
+ public void sequentialReadWriteTest() throws Throwable
+ {
+ File directory = new File(Files.createTempDirectory("JournalTest"));
+ directory.deleteRecursiveOnExit();
+
+ final Journal<TimeUUID, Long> journal = new Journal<>("TestJournal", directory,
+ new TestParams() {
+ @Override
+ public int segmentSize()
+ {
+ return 1024;
+ }
+ },
+ TimeUUIDKeySupport.INSTANCE,
+ LongSerializer.INSTANCE,
+ SegmentCompactor.noop());
+
+ journal.start();
+
+
+ int cycles = 10_000;
+ TestHelper.withRandom(rng -> {
+ Map<TimeUUID, Long> written = new ConcurrentHashMap<>();
+ for (int j = 0; j < cycles; j++)
+ {
+ if (j > 0 && j % 1000 == 0)
+ logger.info("Progress: {}/{}", j, cycles);
+ TimeUUID uuid = nextTimeUUID();
+ long v = rng.next();
+ journal.blockingWrite(uuid, v);
+ written.put(uuid, v);
+ }
+
+ int i = 0;
+ for (Map.Entry<TimeUUID, Long> e : written.entrySet())
+ {
+ long expected = e.getValue();
+ Long actual = journal.readLast(e.getKey());
+ assertEquals(expected, (long) actual);
+ }
+ });
+
+ journal.shutdown();
+ }
+
+ @Test
+ public void concurrentReadWriteTest() throws Throwable
+ {
+ for (int segmentSize : new int[] {1024, 1024 * 1024})
+ {
+ for (Params.FlushMode mode : Params.FlushMode.values())
+ {
+ logger.info("Starting a test run for {} with {}b segment size", mode, segmentSize);
+ concurrentReadWriteTest(segmentSize, mode);
+ }
+ }
+ }
+
+ public void concurrentReadWriteTest(int segmentSize, Params.FlushMode mode) throws Throwable
+ {
+ File directory = new File(Files.createTempDirectory("JournalTest"));
+ directory.deleteRecursiveOnExit();
+
+ final Journal<TimeUUID, Long> journal =
+ new Journal<>("TestJournal", directory, new TestParams() {
+ @Override
+ public int segmentSize()
+ {
+ return segmentSize;
+ }
+
+ @Override
+ public FlushMode flushMode()
+ {
+ return mode;
+ }
+ }, TimeUUIDKeySupport.INSTANCE, LongSerializer.INSTANCE, SegmentCompactor.noop());
+
+ journal.start();
+
+ AtomicLong completedWrites = new AtomicLong();
+ AtomicLong completedReads = new AtomicLong();
+
+ int cycles = 10_000;
+ TestHelper.withRandom(rng -> {
+ List<Thread> threads = new ArrayList<>();
+
+ Map<TimeUUID, Long> written = new ConcurrentHashMap<>();
+ Queue<TimeUUID> recentlyWritten = new ConcurrentLinkedQueue<>();
+ int writers = 10;
+ CountDownLatch writersDone = CountDownLatch.newCountDownLatch(writers);
+ AtomicReference<Throwable> accumulate = new AtomicReference<>();
+ for (int i = 0; i < writers; i++)
+ {
+ EntropySource perThread = rng.derive();
+ Thread writer = new Thread(run(() -> {
+ for (int j = 0; j < cycles; j++)
+ {
+ if (j > 0 && j % 500 == 0)
+ logger.info("Progress: {}/{}", j, cycles);
+
+ TimeUUID uuid = nextTimeUUID();
+ long v = perThread.next();
+ journal.blockingWrite(uuid, v);
+ written.put(uuid, v);
+ recentlyWritten.add(uuid);
+ completedWrites.incrementAndGet();
+ if (accumulate.get() != null)
+ return;
+ }
+ writersDone.decrement();
+ }, accumulate));
+ threads.add(writer);
+ if (i % 2 == 0)
+ {
+ Thread recentReader = new Thread(run(() -> {
+ while (accumulate.get() == null && (writersDone.count() > 0 || !recentlyWritten.isEmpty()))
+ {
+ TimeUUID uuid = recentlyWritten.poll();
+ if (uuid != null)
+ {
+ long expected = written.get(uuid);
+ Long actual = journal.readLast(uuid);
+ assertEquals(expected, (long) actual);
+ }
+ completedReads.incrementAndGet();
+ }
+ }, accumulate));
+ threads.add(recentReader);
+ }
+ else
+ {
+ Thread allReader = new Thread(run(() -> {
+ while (accumulate.get() == null && writersDone.count() > 0)
+ {
+ for (Map.Entry<TimeUUID, Long> e : written.entrySet())
+ {
+ long expected = e.getValue();
+ Long actual = journal.readLast(e.getKey());
+ assertEquals(expected, (long) actual);
+ completedReads.incrementAndGet();
+ }
+ }
+ }, accumulate));
+ threads.add(allReader);
+ }
+ }
+ for (Thread thread : threads)
+ thread.start();
+ for (Thread thread : threads)
+ thread.join();
+ if (accumulate.get() != null)
+ throw accumulate.get();
+ });
+
+ System.out.println(String.format("Finished %d reads, %d writes", completedReads.get(), completedWrites.get()));
+ journal.shutdown();
+ journal.awaitTermination(30, TimeUnit.SECONDS);
+ System.gc();
+ System.gc();
+ }
+
+ // TODO: wasteful long serializer
+ private Runnable run(Runnable r, AtomicReference<Throwable> report)
+ {
+ return () -> {
+ try
+ {
+ r.run();
+ }
+ catch (Throwable t)
+ {
+ report.updateAndGet(new UnaryOperator<Throwable>()
+ {
+ public Throwable apply(Throwable throwable)
+ {
+ if (throwable == null)
+ return t;
+ throwable.addSuppressed(t);
+ return t;
+ }
+ });
+ }
+ };
+ }
+
+ static class LongSerializer implements ValueSerializer<TimeUUID, Long>
+ {
+ static final LongSerializer INSTANCE = new LongSerializer();
+
+ public int serializedSize(TimeUUID key, Long value, int userVersion)
+ {
+ return Long.BYTES;
+ }
+
+ public void serialize(TimeUUID key, Long value, DataOutputPlus out, int userVersion) throws IOException
+ {
+ out.writeLong(value);
+ }
+
+ public Long deserialize(TimeUUID key, DataInputPlus in, int userVersion) throws IOException
+ {
+ return in.readLong();
+ }
+ }
+}
diff --git a/journal/src/test/java/org/apache/cassandra/journal/MetadataTest.java b/journal/src/test/java/org/apache/cassandra/journal/MetadataTest.java
new file mode 100644
index 0000000..2ad9c14
--- /dev/null
+++ b/journal/src/test/java/org/apache/cassandra/journal/MetadataTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.junit.Test;
+
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+
+import static org.junit.Assert.assertEquals;
+
+public class MetadataTest
+{
+ @Test
+ public void testUpdate()
+ {
+ Metadata metadata = Metadata.create();
+
+ metadata.update();
+ metadata.update();
+ metadata.update();
+ metadata.update();
+
+ assertEquals(4, metadata.totalCount());
+ }
+
+ @Test
+ public void testWriteRead() throws IOException
+ {
+ Metadata metadata = Metadata.create();
+
+ metadata.update();
+ metadata.update();
+ metadata.update();
+ metadata.update();
+
+ try (DataOutputBuffer out = DataOutputBuffer.scratchBuffer.get())
+ {
+ metadata.write(out);
+ ByteBuffer serialized = out.buffer();
+
+ try (DataInputBuffer in = new DataInputBuffer(serialized, false))
+ {
+ Metadata deserialized = Metadata.read(in);
+ assertEquals(4, deserialized.totalCount());
+ }
+ }
+ }
+}
diff --git a/journal/src/test/java/org/apache/cassandra/journal/SegmentTest.java b/journal/src/test/java/org/apache/cassandra/journal/SegmentTest.java
new file mode 100644
index 0000000..649b45f
--- /dev/null
+++ b/journal/src/test/java/org/apache/cassandra/journal/SegmentTest.java
@@ -0,0 +1,191 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+
+import org.junit.Test;
+
+import org.apache.cassandra.concurrent.ImmediateExecutor;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.TimeUUID;
+import org.apache.cassandra.utils.concurrent.OpOrder;
+
+import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class SegmentTest
+{
+ @Test
+ public void testWriteReadActiveSegment() throws IOException
+ {
+ TimeUUID id1 = nextTimeUUID();
+ TimeUUID id2 = nextTimeUUID();
+ TimeUUID id3 = nextTimeUUID();
+ TimeUUID id4 = nextTimeUUID();
+
+ ByteBuffer record1 = ByteBufferUtil.bytes("sample record 1");
+ ByteBuffer record2 = ByteBufferUtil.bytes("sample record 2");
+ ByteBuffer record3 = ByteBufferUtil.bytes("sample record 3");
+ ByteBuffer record4 = ByteBufferUtil.bytes("sample record 4");
+
+ File directory = new File(Files.createTempDirectory(null));
+ directory.deleteRecursiveOnExit();
+
+ Descriptor descriptor = Descriptor.create(directory, System.currentTimeMillis(), 1);
+
+ ActiveSegment<TimeUUID, ByteBuffer> segment = ActiveSegment.create(descriptor, params(), TimeUUIDKeySupport.INSTANCE);
+
+ segment.allocate(record1.remaining()).write(id1, record1);
+ segment.allocate(record2.remaining()).write(id2, record2);
+ segment.allocate(record3.remaining()).write(id3, record3);
+ segment.allocate(record4.remaining()).write(id4, record4);
+
+ // read all 4 entries by id and compare with originals
+ EntrySerializer.EntryHolder<TimeUUID> holder = new EntrySerializer.EntryHolder<>();
+
+ segment.readLast(id1, holder);
+ assertEquals(id1, holder.key);
+ assertEquals(record1, holder.value);
+
+ segment.readLast(id2, holder);
+ assertEquals(id2, holder.key);
+ assertEquals(record2, holder.value);
+
+ segment.readLast(id3, holder);
+ assertEquals(id3, holder.key);
+ assertEquals(record3, holder.value);
+
+ segment.readLast(id4, holder);
+ assertEquals(id4, holder.key);
+ assertEquals(record4, holder.value);
+ }
+
+ @Test
+ public void testReadClosedSegmentByID() throws IOException
+ {
+ TimeUUID id1 = nextTimeUUID();
+ TimeUUID id2 = nextTimeUUID();
+ TimeUUID id3 = nextTimeUUID();
+ TimeUUID id4 = nextTimeUUID();
+
+ ByteBuffer record1 = ByteBufferUtil.bytes("sample record 1");
+ ByteBuffer record2 = ByteBufferUtil.bytes("sample record 2");
+ ByteBuffer record3 = ByteBufferUtil.bytes("sample record 3");
+ ByteBuffer record4 = ByteBufferUtil.bytes("sample record 4");
+
+ File directory = new File(Files.createTempDirectory(null));
+ directory.deleteRecursiveOnExit();
+
+ Descriptor descriptor = Descriptor.create(directory, System.currentTimeMillis(), 1);
+
+ ActiveSegment<TimeUUID, ByteBuffer> activeSegment = ActiveSegment.create(descriptor, params(), TimeUUIDKeySupport.INSTANCE);
+
+ activeSegment.allocate(record1.remaining()).write(id1, record1);
+ activeSegment.allocate(record2.remaining()).write(id2, record2);
+ activeSegment.allocate(record3.remaining()).write(id3, record3);
+ activeSegment.allocate(record4.remaining()).write(id4, record4);
+
+ activeSegment.close(null);
+
+ StaticSegment<TimeUUID, ByteBuffer> staticSegment = StaticSegment.open(descriptor, TimeUUIDKeySupport.INSTANCE);
+
+ // read all 4 entries by id and compare with originals
+ EntrySerializer.EntryHolder<TimeUUID> holder = new EntrySerializer.EntryHolder<>();
+
+ staticSegment.readLast(id1, holder);
+ assertEquals(id1, holder.key);
+ assertEquals(record1, holder.value);
+
+ staticSegment.readLast(id2, holder);
+ assertEquals(id2, holder.key);
+ assertEquals(record2, holder.value);
+
+ staticSegment.readLast(id3, holder);
+ assertEquals(id3, holder.key);
+ assertEquals(record3, holder.value);
+
+ staticSegment.readLast(id4, holder);
+ assertEquals(id4, holder.key);
+ assertEquals(record4, holder.value);
+ }
+
+ @Test
+ public void testReadClosedSegmentSequentially() throws IOException
+ {
+ TimeUUID id1 = nextTimeUUID();
+ TimeUUID id2 = nextTimeUUID();
+ TimeUUID id3 = nextTimeUUID();
+ TimeUUID id4 = nextTimeUUID();
+
+ ByteBuffer record1 = ByteBufferUtil.bytes("sample record 1");
+ ByteBuffer record2 = ByteBufferUtil.bytes("sample record 2");
+ ByteBuffer record3 = ByteBufferUtil.bytes("sample record 3");
+ ByteBuffer record4 = ByteBufferUtil.bytes("sample record 4");
+
+ File directory = new File(Files.createTempDirectory(null));
+ directory.deleteRecursiveOnExit();
+
+ Descriptor descriptor = Descriptor.create(directory, System.currentTimeMillis(), 1);
+
+ ActiveSegment<TimeUUID, ByteBuffer> activeSegment = ActiveSegment.create(descriptor, params(), TimeUUIDKeySupport.INSTANCE);
+
+ activeSegment.allocate(record1.remaining()).write(id1, record1);
+ activeSegment.allocate(record2.remaining()).write(id2, record2);
+ activeSegment.allocate(record3.remaining()).write(id3, record3);
+ activeSegment.allocate(record4.remaining()).write(id4, record4);
+
+ Segment.Tidier tidier = (Segment.Tidier)activeSegment.selfRef().tidier();
+ tidier.executor = ImmediateExecutor.INSTANCE;
+ OpOrder opOrder = new OpOrder();
+ tidier.await = opOrder.newBarrier();
+ tidier.await.issue();
+ activeSegment.close(null);
+
+ StaticSegment.SequentialReader<TimeUUID> reader = StaticSegment.sequentialReader(descriptor, TimeUUIDKeySupport.INSTANCE, 0);
+
+ // read all 4 entries sequentially and compare with originals
+ assertTrue(reader.advance());
+ assertEquals(id1, reader.key());
+ assertEquals(record1, reader.record());
+
+ assertTrue(reader.advance());
+ assertEquals(id2, reader.key());
+ assertEquals(record2, reader.record());
+
+ assertTrue(reader.advance());
+ assertEquals(id3, reader.key());
+ assertEquals(record3, reader.record());
+
+ assertTrue(reader.advance());
+ assertEquals(id4, reader.key());
+ assertEquals(record4, reader.record());
+
+ assertFalse(reader.advance());
+ }
+
+ private static Params params()
+ {
+ return TestParams.INSTANCE;
+ }
+}
diff --git a/journal/src/test/java/org/apache/cassandra/journal/TestParams.java b/journal/src/test/java/org/apache/cassandra/journal/TestParams.java
new file mode 100644
index 0000000..72c64b0
--- /dev/null
+++ b/journal/src/test/java/org/apache/cassandra/journal/TestParams.java
@@ -0,0 +1,75 @@
+/*
+ * 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.cassandra.journal;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.cassandra.service.accord.serializers.Version;
+
+public class TestParams implements Params
+{
+ public static final TestParams INSTANCE = new TestParams();
+
+ @Override
+ public int segmentSize()
+ {
+ return 32 << 20;
+ }
+
+ @Override
+ public FailurePolicy failurePolicy()
+ {
+ return FailurePolicy.STOP;
+ }
+
+ @Override
+ public FlushMode flushMode()
+ {
+ return FlushMode.GROUP;
+ }
+
+ @Override
+ public boolean enableCompaction()
+ {
+ return false;
+ }
+
+ @Override
+ public long compactionPeriod(TimeUnit units)
+ {
+ return units.convert(60, TimeUnit.SECONDS);
+ }
+
+ @Override
+ public long flushPeriod(TimeUnit units)
+ {
+ return units.convert(1, TimeUnit.SECONDS);
+ }
+
+ @Override
+ public long periodicBlockPeriod(TimeUnit units)
+ {
+ return units.convert(2, TimeUnit.SECONDS);
+ }
+
+ @Override
+ public int userVersion()
+ {
+ return Version.LATEST.version;
+ }
+}
diff --git a/journal/src/test/java/org/apache/cassandra/journal/TimeUUIDKeySupport.java b/journal/src/test/java/org/apache/cassandra/journal/TimeUUIDKeySupport.java
new file mode 100644
index 0000000..5694a29
--- /dev/null
+++ b/journal/src/test/java/org/apache/cassandra/journal/TimeUUIDKeySupport.java
@@ -0,0 +1,100 @@
+/*
+ * 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.cassandra.journal;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.zip.Checksum;
+
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.utils.TimeUUID;
+
+import static org.apache.cassandra.utils.FBUtilities.updateChecksumLong;
+
+class TimeUUIDKeySupport implements KeySupport<TimeUUID>
+{
+ static final TimeUUIDKeySupport INSTANCE = new TimeUUIDKeySupport();
+
+ @Override
+ public int serializedSize(int userVersion)
+ {
+ return 16;
+ }
+
+ @Override
+ public void serialize(TimeUUID key, DataOutputPlus out, int userVersion) throws IOException
+ {
+ out.writeLong(key.uuidTimestamp());
+ out.writeLong(key.lsb());
+ }
+
+ @Override
+ public void serialize(TimeUUID key, ByteBuffer out, int userVersion) throws IOException
+ {
+ out.putLong(key.uuidTimestamp());
+ out.putLong(key.lsb());
+ }
+
+ @Override
+ public TimeUUID deserialize(DataInputPlus in, int userVersion) throws IOException
+ {
+ long uuidTimestamp = in.readLong();
+ long lsb = in.readLong();
+ return new TimeUUID(uuidTimestamp, lsb);
+ }
+
+ @Override
+ public TimeUUID deserialize(ByteBuffer buffer, int position, int userVersion)
+ {
+ long uuidTimestamp = buffer.getLong(position);
+ long lsb = buffer.getLong(position + 8);
+ return new TimeUUID(uuidTimestamp, lsb);
+ }
+
+ @Override
+ public TimeUUID deserialize(ByteBuffer buffer, int userVersion)
+ {
+ long uuidTimestamp = buffer.getLong();
+ long lsb = buffer.getLong();
+ return new TimeUUID(uuidTimestamp, lsb);
+ }
+
+ @Override
+ public void updateChecksum(Checksum crc, TimeUUID key, int userVersion)
+ {
+ updateChecksumLong(crc, key.uuidTimestamp());
+ updateChecksumLong(crc, key.lsb());
+ }
+
+ @Override
+ public int compareWithKeyAt(TimeUUID key, ByteBuffer buffer, int position, int userVersion)
+ {
+ long uuidTimestamp = buffer.getLong(position);
+ long lsb = buffer.getLong(position + 8);
+ return key.uuidTimestamp() != uuidTimestamp
+ ? Long.compare(key.uuidTimestamp(), uuidTimestamp)
+ : Long.compare(key.lsb(), lsb);
+ }
+
+ @Override
+ public int compare(TimeUUID o1, TimeUUID o2)
+ {
+ return o1.compareTo(o2);
+ }
+}
diff --git a/journal/src/test/java/org/apache/cassandra/utils/TimeUUID.java b/journal/src/test/java/org/apache/cassandra/utils/TimeUUID.java
new file mode 100644
index 0000000..2b2b0c8
--- /dev/null
+++ b/journal/src/test/java/org/apache/cassandra/utils/TimeUUID.java
@@ -0,0 +1,64 @@
+package org.apache.cassandra.utils;
+
+import java.util.Objects;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicLong;
+
+public final class TimeUUID implements Comparable<TimeUUID>
+{
+ private final long uuidTimestamp;
+ private final long lsb;
+
+ public TimeUUID(long uuidTimestamp, long lsb)
+ {
+ this.uuidTimestamp = uuidTimestamp;
+ this.lsb = lsb;
+ }
+
+ public long uuidTimestamp()
+ {
+ return uuidTimestamp;
+ }
+
+ public long lsb()
+ {
+ return lsb;
+ }
+
+ @Override
+ public int compareTo(TimeUUID that)
+ {
+ int c = Long.compare(this.uuidTimestamp, that.uuidTimestamp);
+ return c != 0 ? c : Long.compare(this.lsb, that.lsb);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (!(o instanceof TimeUUID that)) return false;
+ return uuidTimestamp == that.uuidTimestamp && lsb == that.lsb;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(uuidTimestamp, lsb);
+ }
+
+ @Override
+ public String toString()
+ {
+ return Long.toHexString(uuidTimestamp) + ':' + Long.toHexString(lsb);
+ }
+
+ public static final class Generator
+ {
+ private static final AtomicLong NEXT = new AtomicLong(System.nanoTime());
+
+ public static TimeUUID nextTimeUUID()
+ {
+ return new TimeUUID(NEXT.incrementAndGet(), ThreadLocalRandom.current().nextLong());
+ }
+ }
+}
diff --git a/settings.gradle b/settings.gradle
index d040f74..f498fc6 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -3,3 +3,4 @@ include 'simulator-context'
include 'simulator-asm'
include 'simulator-core'
include 'integration-test'
+include 'journal'
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org