<commit_before>/*
 * Copyright 2017-present Open Networking Laboratory
 *
 * 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
 *
 * 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 io.atomix.storage.journal;

import com.google.common.collect.Sets;
import io.atomix.serializer.Serializer;
import io.atomix.storage.StorageLevel;
import io.atomix.storage.buffer.Buffer;
import io.atomix.storage.buffer.FileBuffer;
import io.atomix.storage.buffer.HeapBuffer;
import io.atomix.storage.buffer.MappedBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.util.Collection;
import java.util.Map;
import java.util.NavigableMap;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;

/**
 * Segmented journal implementation.
 *
 * @author <a href="http://github.com/kuujo><NAME></a>
 */
public class SegmentedJournal<E> implements Journal<E> {

  /**
   * Returns a new segmented journal builder.
   *
   * @return A new segmented journal builder.
   */
  public static <E> Builder<E> newBuilder() {
    return new Builder<>();
  }

  private static final int DEFAULT_BUFFER_SIZE = 1024 * 1024;

  private final Logger log = LoggerFactory.getLogger(getClass());
  private final String name;
  private final StorageLevel storageLevel;
  private final File directory;
  private final Serializer serializer;
  private final int maxSegmentSize;
  private final int maxEntriesPerSegment;

  private final NavigableMap<Long, JournalSegment<E>> segments = new ConcurrentSkipListMap<>();
  private final Collection<SegmentedJournalReader<E>> readers = Sets.newConcurrentHashSet();
  private JournalSegment<E> currentSegment;

  private final SegmentedJournalWriter<E> writer;
  protected final ReadWriteLock lock = new ReentrantReadWriteLock();
  private volatile boolean open = true;

  public SegmentedJournal(
      String name,
      StorageLevel storageLevel,
      File directory,
      Serializer serializer,
      int maxSegmentSize,
      int maxEntriesPerSegment) {
    this.name = checkNotNull(name, "name cannot be null");
    this.storageLevel = checkNotNull(storageLevel, "storageLevel cannot be null");
    this.directory = checkNotNull(directory, "directory cannot be null");
    this.serializer = checkNotNull(serializer, "serializer cannot be null");
    this.maxSegmentSize = maxSegmentSize;
    this.maxEntriesPerSegment = maxEntriesPerSegment;
    open();
    this.writer = openWriter();
  }

  /**
   * Returns the segment file name prefix.
   *
   * @return The segment file name prefix.
   */
  public String name() {
    return name;
  }

  /**
   * Returns the storage directory.
   * <p>
   * The storage directory is the directory to which all segments write files. Segment files
   * for multiple logs may be stored in the storage directory, and files for each log instance will be identified
   * by the {@code prefix} provided when the log is opened.
   *
   * @return The storage directory.
   */
  public File directory() {
    return directory;
  }

  /**
   * Returns the storage level.
   * <p>
   * The storage level dictates how entries within individual journal segments should be stored.
   *
   * @return The storage level.
   */
  public StorageLevel storageLevel() {
    return storageLevel;
  }

  /**
   * Returns the maximum journal segment size.
   * <p>
   * The maximum segment size dictates the maximum size any segment in a segment may consume
   * in bytes.
   *
   * @return The maximum segment size in bytes.
   */
  public int maxSegmentSize() {
    return maxSegmentSize;
  }

  /**
   * Returns the maximum number of entries per segment.
   * <p>
   * The maximum entries per segment dictates the maximum number of entries
   * that are allowed to be stored in any segment in a journal.
   *
   * @return The maximum number of entries per segment.
   */
  public int maxEntriesPerSegment() {
    return maxEntriesPerSegment;
  }

  /**
   * Opens a new journal writer.
   *
   * @return A new journal writer.
   */
  protected SegmentedJournalWriter<E> openWriter() {
    return new SegmentedJournalWriter<>(this, lock.writeLock());
  }

  /**
   * Opens the segments.
   */
  private void open() {
    // Load existing log segments from disk.
    for (JournalSegment segment : loadSegments()) {
      segments.put(segment.descriptor().index(), segment);
    }

    // If a segment doesn't already exist, create an initial segment starting at index 1.
    if (!segments.isEmpty()) {
      currentSegment = segments.lastEntry().getValue();
    } else {
      JournalSegmentDescriptor descriptor = JournalSegmentDescriptor.newBuilder()
          .withId(1)
          .withIndex(1)
          .withMaxSegmentSize(maxSegmentSize)
          .withMaxEntries(maxEntriesPerSegment)
          .build();

      currentSegment = createSegment(descriptor);
      currentSegment.descriptor().update(System.currentTimeMillis());

      segments.put(1L, currentSegment);
    }
  }

  /**
   * Asserts that the manager is open.
   *
   * @throws IllegalStateException if the segment manager is not open
   */
  private void assertOpen() {
    checkState(currentSegment != null, "journal not open");
  }

  /**
   * Resets the current segment, creating a new segment if necessary.
   */
  private synchronized void resetCurrentSegment() {
    JournalSegment lastSegment = getLastSegment();
    if (lastSegment != null) {
      currentSegment = lastSegment;
    } else {
      JournalSegmentDescriptor descriptor = JournalSegmentDescriptor.newBuilder()
          .withId(1)
          .withIndex(1)
          .withMaxSegmentSize(maxSegmentSize)
          .withMaxEntries(maxEntriesPerSegment)
          .build();

      currentSegment = createSegment(descriptor);

      segments.put(1L, currentSegment);
    }
  }

  /**
   * Returns the first segment in the log.
   *
   * @throws IllegalStateException if the segment manager is not open
   */
  synchronized JournalSegment<E> getFirstSegment() {
    assertOpen();
    Map.Entry<Long, JournalSegment<E>> segment = segments.firstEntry();
    return segment != null ? segment.getValue() : null;
  }

  /**
   * Returns the last segment in the log.
   *
   * @throws IllegalStateException if the segment manager is not open
   */
  synchronized JournalSegment<E> getLastSegment() {
    assertOpen();
    Map.Entry<Long, JournalSegment<E>> segment = segments.lastEntry();
    return segment != null ? segment.getValue() : null;
  }

  /**
   * Returns the segment prior to the segment with the given ID.
   *
   * @param index The segment index with which to look up the prior segment.
   * @return The prior segment for the given index.
   */
  JournalSegment<E> getPreviousSegment(long index) {
    Map.Entry<Long, JournalSegment<E>> previousSegment = segments.lowerEntry(index);
    return previousSegment != null ? previousSegment.getValue() : null;
  }

  /**
   * Creates and returns the next segment.
   *
   * @return The next segment.
   * @throws IllegalStateException if the segment manager is not open
   */
  synchronized JournalSegment<E> getNextSegment() {
    assertOpen();
    JournalSegment lastSegment = getLastSegment();
    JournalSegmentDescriptor descriptor = JournalSegmentDescriptor.newBuilder()
        .withId(lastSegment != null ? lastSegment.descriptor().id() + 1 : 1)
        .withIndex(currentSegment.lastIndex() + 1)
        .withMaxSegmentSize(maxSegmentSize)
        .withMaxEntries(maxEntriesPerSegment)
        .build();

    currentSegment = createSegment(descriptor);

    segments.put(descriptor.index(), currentSegment);
    return currentSegment;
  }

  /**
   * Returns the segment following the segment with the given ID.
   *
   * @param index The segment index with which to look up the next segment.
   * @return The next segment for the given index.
   */
  JournalSegment<E> getNextSegment(long index) {
    Map.Entry<Long, JournalSegment<E>> nextSegment = segments.higherEntry(index);
    return nextSegment != null ? nextSegment.getValue() : null;
  }

  /**
   * Returns the segment for the given index.
   *
   * @param index The index for which to return the segment.
   * @throws IllegalStateException if the segment manager is not open
   */
  synchronized JournalSegment<E> getSegment(long index) {
    assertOpen();
    // Check if the current segment contains the given index first in order to prevent an unnecessary map lookup.
    if (currentSegment != null && index > currentSegment.index()) {
      return currentSegment;
    }

    // If the index is in another segment, get the entry with the next lowest first index.
    Map.Entry<Long, JournalSegment<E>> segment = segments.floorEntry(index);
    return segment != null ? segment.getValue() : null;
  }

  /**
   * Removes a segment.
   *
   * @param segment The segment to remove.
   */
  synchronized void removeSegment(JournalSegment segment) {
    segments.remove(segment.index());
    segment.close();
    segment.delete();
    resetCurrentSegment();
  }

  /**
   * Creates a new segment.
   */
  private JournalSegment<E> createSegment(JournalSegmentDescriptor descriptor) {
    switch (storageLevel) {
      case MEMORY:
        return createMemorySegment(descriptor);
      case DISK:
        return createDiskSegment(descriptor);
      default:
        throw new AssertionError();
    }
  }

  /**
   * Creates a new segment instance.
   *
   * @param segmentFile The segment file.
   * @param descriptor The segment descriptor.
   * @return The segment instance.
   */
  protected JournalSegment<E> newSegment(JournalSegmentFile segmentFile, JournalSegmentDescriptor descriptor) {
    return new JournalSegment<>(segmentFile, descriptor, serializer, this);
  }

  /**
   * Creates a new segment.
   */
  private JournalSegment<E> createDiskSegment(JournalSegmentDescriptor descriptor) {
    File segmentFile = JournalSegmentFile.createSegmentFile(name, directory, descriptor.id(), descriptor.version());
    Buffer buffer = MappedBuffer.allocate(segmentFile, Math.min(DEFAULT_BUFFER_SIZE, descriptor.maxSegmentSize()), Integer.MAX_VALUE);
    descriptor.copyTo(buffer);
    JournalSegment<E> segment = newSegment(new JournalSegmentFile(segmentFile), descriptor);
    log.debug("Created disk segment: {}", segment);
    return segment;
  }

  /**
   * Creates a new segment.
   */
  private JournalSegment<E> createMemorySegment(JournalSegmentDescriptor descriptor) {
    File segmentFile = JournalSegmentFile.createSegmentFile(name, directory, descriptor.id(), descriptor.version());
    Buffer buffer = HeapBuffer.allocate(Math.min(DEFAULT_BUFFER_SIZE, descriptor.maxSegmentSize()), Integer.MAX_VALUE);
    descriptor.copyTo(buffer);
    JournalSegment<E> segment = newSegment(new JournalSegmentFile(segmentFile), descriptor);
    log.debug("Created memory segment: {}", segment);
    return segment;
  }

  /**
   * Loads a segment.
   */
  private JournalSegment loadSegment(long segmentId, long segmentVersion) {
    switch (storageLevel) {
      case MEMORY:
        return loadMemorySegment(segmentId, segmentVersion);
      case DISK:
        return loadDiskSegment(segmentId, segmentVersion);
      default:
        throw new AssertionError();
    }
  }

  /**
   * Loads a segment.
   */
  private JournalSegment<E> loadDiskSegment(long segmentId, long segmentVersion) {
    File file = JournalSegmentFile.createSegmentFile(name, directory, segmentId, segmentVersion);
    Buffer buffer = MappedBuffer.allocate(file, Math.min(DEFAULT_BUFFER_SIZE, maxSegmentSize), Integer.MAX_VALUE);
    JournalSegmentDescriptor descriptor = new JournalSegmentDescriptor(buffer);
    JournalSegment<E> segment = newSegment(new JournalSegmentFile(file), descriptor);
    log.debug("Loaded disk segment: {} ({})", descriptor.id(), file.getName());
    return segment;
  }

  /**
   * Loads a segment.
   */
  private JournalSegment<E> loadMemorySegment(long segmentId, long segmentVersion) {
    File file = JournalSegmentFile.createSegmentFile(name, directory, segmentId, segmentVersion);
    Buffer buffer = HeapBuffer.allocate(Math.min(DEFAULT_BUFFER_SIZE, maxSegmentSize), Integer.MAX_VALUE);
    JournalSegmentDescriptor descriptor = new JournalSegmentDescriptor(buffer);
    JournalSegment<E> segment = newSegment(new JournalSegmentFile(file), descriptor);
    log.debug("Loaded memory segment: {}", descriptor.id());
    return segment;
  }

  /**
   * Loads all segments from disk.
   *
   * @return A collection of segments for the log.
   */
  protected Collection<JournalSegment> loadSegments() {
    // Ensure log directories are created.
    directory.mkdirs();

    TreeMap<Long, JournalSegment> segments = new TreeMap<>();

    // Iterate through all files in the log directory.
    for (File file : directory.listFiles(File::isFile)) {

      // If the file looks like a segment file, attempt to load the segment.
      if (JournalSegmentFile.isSegmentFile(name, file)) {
        JournalSegmentFile segmentFile = new JournalSegmentFile(file);
        JournalSegmentDescriptor descriptor = new JournalSegmentDescriptor(FileBuffer.allocate(file, JournalSegmentDescriptor.BYTES));

        // Load the segment.
        JournalSegment segment = loadSegment(descriptor.id(), descriptor.version());

        // If a segment with an equal or lower index has already been loaded, ensure this segment is not superseded
        // by the earlier segment. This can occur due to segments being combined during log compaction.
        Map.Entry<Long, JournalSegment> previousEntry = segments.floorEntry(segment.index());
        if (previousEntry != null) {

          // If an existing descriptor exists with a lower index than this segment's first index, check to determine
          // whether this segment's first index is contained in that existing index. If it is, determine which segment
          // should take precedence based on segment versions.
          JournalSegment previousSegment = previousEntry.getValue();

          // If the two segments start at the same index, the segment with the higher version number is used.
          if (previousSegment.index() == segment.index()) {
            if (segment.descriptor().version() > previousSegment.descriptor().version()) {
              log.debug("Replaced segment {} with newer version: {} ({})", previousSegment.descriptor().id(), segment.descriptor().version(), segmentFile.file().getName());
              segments.remove(previousEntry.getKey());
              previousSegment.close();
              previousSegment.delete();
            } else {
              segment.close();
              segment.delete();
              continue;
            }
          }
          // If the existing segment's entries overlap with the loaded segment's entries, the existing segment always
          // supersedes the loaded segment. Log compaction processes ensure this is always the case.
          else if (previousSegment.index() + previousSegment.length() > segment.index()) {
            segment.close();
            segment.delete();
            continue;
          }
        }

        // Add the segment to the segments list.
        log.debug("Found segment: {} ({})", segment.descriptor().id(), segmentFile.file().getName());
        segments.put(segment.index(), segment);

        // Ensure any segments later in the log with which this segment overlaps are removed.
        Map.Entry<Long, JournalSegment> nextEntry = segments.higherEntry(segment.index());
        while (nextEntry != null) {
          if (nextEntry.getValue().index() < segment.index() + segment.length()) {
            segments.remove(nextEntry.getKey());
            nextEntry = segments.higherEntry(segment.index());
          } else {
            break;
          }
        }

        descriptor.close();
      }
    }
    return segments.values();
  }

  /**
   * Resets journal readers.
   *
   * @param index The index at which to reset readers.
   */
  void resetReaders(long index) {
    for (SegmentedJournalReader<E> reader : readers) {
      if (reader.getNextIndex() >= index) {
        reader.reset(index);
      }
    }
  }

  @Override
  public SegmentedJournalWriter<E> writer() {
    return writer;
  }

  @Override
  public SegmentedJournalReader<E> openReader(long index) {
    return new SegmentedJournalReader<>(this, lock.readLock(), index);
  }

  @Override
  public boolean isOpen() {
    return open;
  }

  @Override
  public void compact(long index) {
    SortedMap<Long, JournalSegment<E>> compactSegments = segments.headMap(index);
    if (!compactSegments.isEmpty()) {
      log.info("{} - Compacting {} segment(s)", name, compactSegments.size());
      writer.getLock().lock();
      try {
        for (JournalSegment segment : compactSegments.values()) {
          log.debug("Deleting segment: {}", segment);
          segment.close();
          segment.delete();
        }
        compactSegments.clear();
      } finally {
        writer.getLock().unlock();
      }
    }
  }

  @Override
  public void close() {
    segments.values().forEach(segment -> {
      log.debug("Closing segment: {}", segment);
      segment.close();<commit_msg>Ensure current segment cannot be removed during compation.<commit_after>/*
 * Copyright 2017-present Open Networking Laboratory
 *
 * 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
 *
 * 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 io.atomix.storage.journal;

import com.google.common.collect.Sets;
import io.atomix.serializer.Serializer;
import io.atomix.storage.StorageLevel;
import io.atomix.storage.buffer.Buffer;
import io.atomix.storage.buffer.FileBuffer;
import io.atomix.storage.buffer.HeapBuffer;
import io.atomix.storage.buffer.MappedBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.util.Collection;
import java.util.Map;
import java.util.NavigableMap;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;

/**
 * Segmented journal implementation.
 *
 * @author <a href="http://github.com/kuujo><NAME></a>
 */
public class SegmentedJournal<E> implements Journal<E> {

  /**
   * Returns a new segmented journal builder.
   *
   * @return A new segmented journal builder.
   */
  public static <E> Builder<E> newBuilder() {
    return new Builder<>();
  }

  private static final int DEFAULT_BUFFER_SIZE = 1024 * 1024;

  private final Logger log = LoggerFactory.getLogger(getClass());
  private final String name;
  private final StorageLevel storageLevel;
  private final File directory;
  private final Serializer serializer;
  private final int maxSegmentSize;
  private final int maxEntriesPerSegment;

  private final NavigableMap<Long, JournalSegment<E>> segments = new ConcurrentSkipListMap<>();
  private final Collection<SegmentedJournalReader<E>> readers = Sets.newConcurrentHashSet();
  private volatile JournalSegment<E> currentSegment;

  private final SegmentedJournalWriter<E> writer;
  protected final ReadWriteLock lock = new ReentrantReadWriteLock();
  private volatile boolean open = true;

  public SegmentedJournal(
      String name,
      StorageLevel storageLevel,
      File directory,
      Serializer serializer,
      int maxSegmentSize,
      int maxEntriesPerSegment) {
    this.name = checkNotNull(name, "name cannot be null");
    this.storageLevel = checkNotNull(storageLevel, "storageLevel cannot be null");
    this.directory = checkNotNull(directory, "directory cannot be null");
    this.serializer = checkNotNull(serializer, "serializer cannot be null");
    this.maxSegmentSize = maxSegmentSize;
    this.maxEntriesPerSegment = maxEntriesPerSegment;
    open();
    this.writer = openWriter();
  }

  /**
   * Returns the segment file name prefix.
   *
   * @return The segment file name prefix.
   */
  public String name() {
    return name;
  }

  /**
   * Returns the storage directory.
   * <p>
   * The storage directory is the directory to which all segments write files. Segment files
   * for multiple logs may be stored in the storage directory, and files for each log instance will be identified
   * by the {@code prefix} provided when the log is opened.
   *
   * @return The storage directory.
   */
  public File directory() {
    return directory;
  }

  /**
   * Returns the storage level.
   * <p>
   * The storage level dictates how entries within individual journal segments should be stored.
   *
   * @return The storage level.
   */
  public StorageLevel storageLevel() {
    return storageLevel;
  }

  /**
   * Returns the maximum journal segment size.
   * <p>
   * The maximum segment size dictates the maximum size any segment in a segment may consume
   * in bytes.
   *
   * @return The maximum segment size in bytes.
   */
  public int maxSegmentSize() {
    return maxSegmentSize;
  }

  /**
   * Returns the maximum number of entries per segment.
   * <p>
   * The maximum entries per segment dictates the maximum number of entries
   * that are allowed to be stored in any segment in a journal.
   *
   * @return The maximum number of entries per segment.
   */
  public int maxEntriesPerSegment() {
    return maxEntriesPerSegment;
  }

  /**
   * Opens a new journal writer.
   *
   * @return A new journal writer.
   */
  protected SegmentedJournalWriter<E> openWriter() {
    return new SegmentedJournalWriter<>(this, lock.writeLock());
  }

  /**
   * Opens the segments.
   */
  private void open() {
    // Load existing log segments from disk.
    for (JournalSegment<E> segment : loadSegments()) {
      segments.put(segment.descriptor().index(), segment);
    }

    // If a segment doesn't already exist, create an initial segment starting at index 1.
    if (!segments.isEmpty()) {
      currentSegment = segments.lastEntry().getValue();
    } else {
      JournalSegmentDescriptor descriptor = JournalSegmentDescriptor.newBuilder()
          .withId(1)
          .withIndex(1)
          .withMaxSegmentSize(maxSegmentSize)
          .withMaxEntries(maxEntriesPerSegment)
          .build();

      currentSegment = createSegment(descriptor);
      currentSegment.descriptor().update(System.currentTimeMillis());

      segments.put(1L, currentSegment);
    }
  }

  /**
   * Asserts that the manager is open.
   *
   * @throws IllegalStateException if the segment manager is not open
   */
  private void assertOpen() {
    checkState(currentSegment != null, "journal not open");
  }

  /**
   * Resets the current segment, creating a new segment if necessary.
   */
  private synchronized void resetCurrentSegment() {
    JournalSegment<E> lastSegment = getLastSegment();
    if (lastSegment != null) {
      currentSegment = lastSegment;
    } else {
      JournalSegmentDescriptor descriptor = JournalSegmentDescriptor.newBuilder()
          .withId(1)
          .withIndex(1)
          .withMaxSegmentSize(maxSegmentSize)
          .withMaxEntries(maxEntriesPerSegment)
          .build();

      currentSegment = createSegment(descriptor);

      segments.put(1L, currentSegment);
    }
  }

  /**
   * Returns the first segment in the log.
   *
   * @throws IllegalStateException if the segment manager is not open
   */
  synchronized JournalSegment<E> getFirstSegment() {
    assertOpen();
    Map.Entry<Long, JournalSegment<E>> segment = segments.firstEntry();
    return segment != null ? segment.getValue() : null;
  }

  /**
   * Returns the last segment in the log.
   *
   * @throws IllegalStateException if the segment manager is not open
   */
  synchronized JournalSegment<E> getLastSegment() {
    assertOpen();
    Map.Entry<Long, JournalSegment<E>> segment = segments.lastEntry();
    return segment != null ? segment.getValue() : null;
  }

  /**
   * Returns the segment prior to the segment with the given ID.
   *
   * @param index The segment index with which to look up the prior segment.
   * @return The prior segment for the given index.
   */
  JournalSegment<E> getPreviousSegment(long index) {
    Map.Entry<Long, JournalSegment<E>> previousSegment = segments.lowerEntry(index);
    return previousSegment != null ? previousSegment.getValue() : null;
  }

  /**
   * Creates and returns the next segment.
   *
   * @return The next segment.
   * @throws IllegalStateException if the segment manager is not open
   */
  synchronized JournalSegment<E> getNextSegment() {
    assertOpen();
    JournalSegment lastSegment = getLastSegment();
    JournalSegmentDescriptor descriptor = JournalSegmentDescriptor.newBuilder()
        .withId(lastSegment != null ? lastSegment.descriptor().id() + 1 : 1)
        .withIndex(currentSegment.lastIndex() + 1)
        .withMaxSegmentSize(maxSegmentSize)
        .withMaxEntries(maxEntriesPerSegment)
        .build();

    currentSegment = createSegment(descriptor);

    segments.put(descriptor.index(), currentSegment);
    return currentSegment;
  }

  /**
   * Returns the segment following the segment with the given ID.
   *
   * @param index The segment index with which to look up the next segment.
   * @return The next segment for the given index.
   */
  JournalSegment<E> getNextSegment(long index) {
    Map.Entry<Long, JournalSegment<E>> nextSegment = segments.higherEntry(index);
    return nextSegment != null ? nextSegment.getValue() : null;
  }

  /**
   * Returns the segment for the given index.
   *
   * @param index The index for which to return the segment.
   * @throws IllegalStateException if the segment manager is not open
   */
  synchronized JournalSegment<E> getSegment(long index) {
    assertOpen();
    // Check if the current segment contains the given index first in order to prevent an unnecessary map lookup.
    if (currentSegment != null && index > currentSegment.index()) {
      return currentSegment;
    }

    // If the index is in another segment, get the entry with the next lowest first index.
    Map.Entry<Long, JournalSegment<E>> segment = segments.floorEntry(index);
    return segment != null ? segment.getValue() : null;
  }

  /**
   * Removes a segment.
   *
   * @param segment The segment to remove.
   */
  synchronized void removeSegment(JournalSegment segment) {
    segments.remove(segment.index());
    segment.close();
    segment.delete();
    resetCurrentSegment();
  }

  /**
   * Creates a new segment.
   */
  private JournalSegment<E> createSegment(JournalSegmentDescriptor descriptor) {
    switch (storageLevel) {
      case MEMORY:
        return createMemorySegment(descriptor);
      case DISK:
        return createDiskSegment(descriptor);
      default:
        throw new AssertionError();
    }
  }

  /**
   * Creates a new segment instance.
   *
   * @param segmentFile The segment file.
   * @param descriptor The segment descriptor.
   * @return The segment instance.
   */
  protected JournalSegment<E> newSegment(JournalSegmentFile segmentFile, JournalSegmentDescriptor descriptor) {
    return new JournalSegment<>(segmentFile, descriptor, serializer, this);
  }

  /**
   * Creates a new segment.
   */
  private JournalSegment<E> createDiskSegment(JournalSegmentDescriptor descriptor) {
    File segmentFile = JournalSegmentFile.createSegmentFile(name, directory, descriptor.id(), descriptor.version());
    Buffer buffer = MappedBuffer.allocate(segmentFile, Math.min(DEFAULT_BUFFER_SIZE, descriptor.maxSegmentSize()), Integer.MAX_VALUE);
    descriptor.copyTo(buffer);
    JournalSegment<E> segment = newSegment(new JournalSegmentFile(segmentFile), descriptor);
    log.debug("Created disk segment: {}", segment);
    return segment;
  }

  /**
   * Creates a new segment.
   */
  private JournalSegment<E> createMemorySegment(JournalSegmentDescriptor descriptor) {
    File segmentFile = JournalSegmentFile.createSegmentFile(name, directory, descriptor.id(), descriptor.version());
    Buffer buffer = HeapBuffer.allocate(Math.min(DEFAULT_BUFFER_SIZE, descriptor.maxSegmentSize()), Integer.MAX_VALUE);
    descriptor.copyTo(buffer);
    JournalSegment<E> segment = newSegment(new JournalSegmentFile(segmentFile), descriptor);
    log.debug("Created memory segment: {}", segment);
    return segment;
  }

  /**
   * Loads a segment.
   */
  private JournalSegment<E> loadSegment(long segmentId, long segmentVersion) {
    switch (storageLevel) {
      case MEMORY:
        return loadMemorySegment(segmentId, segmentVersion);
      case DISK:
        return loadDiskSegment(segmentId, segmentVersion);
      default:
        throw new AssertionError();
    }
  }

  /**
   * Loads a segment.
   */
  private JournalSegment<E> loadDiskSegment(long segmentId, long segmentVersion) {
    File file = JournalSegmentFile.createSegmentFile(name, directory, segmentId, segmentVersion);
    Buffer buffer = MappedBuffer.allocate(file, Math.min(DEFAULT_BUFFER_SIZE, maxSegmentSize), Integer.MAX_VALUE);
    JournalSegmentDescriptor descriptor = new JournalSegmentDescriptor(buffer);
    JournalSegment<E> segment = newSegment(new JournalSegmentFile(file), descriptor);
    log.debug("Loaded disk segment: {} ({})", descriptor.id(), file.getName());
    return segment;
  }

  /**
   * Loads a segment.
   */
  private JournalSegment<E> loadMemorySegment(long segmentId, long segmentVersion) {
    File file = JournalSegmentFile.createSegmentFile(name, directory, segmentId, segmentVersion);
    Buffer buffer = HeapBuffer.allocate(Math.min(DEFAULT_BUFFER_SIZE, maxSegmentSize), Integer.MAX_VALUE);
    JournalSegmentDescriptor descriptor = new JournalSegmentDescriptor(buffer);
    JournalSegment<E> segment = newSegment(new JournalSegmentFile(file), descriptor);
    log.debug("Loaded memory segment: {}", descriptor.id());
    return segment;
  }

  /**
   * Loads all segments from disk.
   *
   * @return A collection of segments for the log.
   */
  protected Collection<JournalSegment<E>> loadSegments() {
    // Ensure log directories are created.
    directory.mkdirs();

    TreeMap<Long, JournalSegment<E>> segments = new TreeMap<>();

    // Iterate through all files in the log directory.
    for (File file : directory.listFiles(File::isFile)) {

      // If the file looks like a segment file, attempt to load the segment.
      if (JournalSegmentFile.isSegmentFile(name, file)) {
        JournalSegmentFile segmentFile = new JournalSegmentFile(file);
        JournalSegmentDescriptor descriptor = new JournalSegmentDescriptor(FileBuffer.allocate(file, JournalSegmentDescriptor.BYTES));

        // Load the segment.
        JournalSegment<E> segment = loadSegment(descriptor.id(), descriptor.version());

        // If a segment with an equal or lower index has already been loaded, ensure this segment is not superseded
        // by the earlier segment. This can occur due to segments being combined during log compaction.
        Map.Entry<Long, JournalSegment<E>> previousEntry = segments.floorEntry(segment.index());
        if (previousEntry != null) {

          // If an existing descriptor exists with a lower index than this segment's first index, check to determine
          // whether this segment's first index is contained in that existing index. If it is, determine which segment
          // should take precedence based on segment versions.
          JournalSegment previousSegment = previousEntry.getValue();

          // If the two segments start at the same index, the segment with the higher version number is used.
          if (previousSegment.index() == segment.index()) {
            if (segment.descriptor().version() > previousSegment.descriptor().version()) {
              log.debug("Replaced segment {} with newer version: {} ({})", previousSegment.descriptor().id(), segment.descriptor().version(), segmentFile.file().getName());
              segments.remove(previousEntry.getKey());
              previousSegment.close();
              previousSegment.delete();
            } else {
              segment.close();
              segment.delete();
              continue;
            }
          }
          // If the existing segment's entries overlap with the loaded segment's entries, the existing segment always
          // supersedes the loaded segment. Log compaction processes ensure this is always the case.
          else if (previousSegment.index() + previousSegment.length() > segment.index()) {
            segment.close();
            segment.delete();
            continue;
          }
        }

        // Add the segment to the segments list.
        log.debug("Found segment: {} ({})", segment.descriptor().id(), segmentFile.file().getName());
        segments.put(segment.index(), segment);

        // Ensure any segments later in the log with which this segment overlaps are removed.
        Map.Entry<Long, JournalSegment<E>> nextEntry = segments.higherEntry(segment.index());
        while (nextEntry != null) {
          if (nextEntry.getValue().index() < segment.index() + segment.length()) {
            segments.remove(nextEntry.getKey());
            nextEntry = segments.higherEntry(segment.index());
          } else {
            break;
          }
        }

        descriptor.close();
      }
    }
    return segments.values();
  }

  /**
   * Resets journal readers.
   *
   * @param index The index at which to reset readers.
   */
  void resetReaders(long index) {
    for (SegmentedJournalReader<E> reader : readers) {
      if (reader.getNextIndex() >= index) {
        reader.reset(index);
      }
    }
  }

  @Override
  public SegmentedJournalWriter<E> writer() {
    return writer;
  }

  @Override
  public SegmentedJournalReader<E> openReader(long index) {
    SegmentedJournalReader<E> reader = new SegmentedJournalReader<>(this, lock.readLock(), index);
    readers.add(reader);
    return reader;
  }

  void closeReader(SegmentedJournalReader<E> reader) {
    readers.remove(reader);
  }

  @Override
  public boolean isOpen() {
    return open;
  }

  @Override
  public void compact(long index) {
    Map.Entry<Long, JournalSegment<E>> segmentEntry = segments.floorEntry(index);
    if (segmentEntry != null) {
      SortedMap<Long, JournalSegment<E>> compactSegments = segments.headMap(segmentEntry.getValue().index());
      if (!compactSegments.isEmpty()) {
        log.info("{} - Compacting {} segment(s)", name, compactSegments.size());
        writer.getLock().lock();
        try {
          for (JournalSegment segment : compactSegments.values()) {
            log.debug("Deleting segment: {}", segment);
            segment.close();
            segment.delete();
          }
          compactSegments.clear();
        } finally {
          writer.getLock().unlock();
        }
      }
    }
  }

  @Override
  public void close() {
    segments.values().forEach(segment -> {
      log.debug("Closing segment: {}", segment);
      segment.close();
{'fraction_non_alphanumeric': 0.10137106987500688, 'fraction_numerical': 0.0015417653212928804, 'mean_word_length': 3.2697778300223344, 'pattern_counts': {'":': 0, '<': 135, '<?xml version=': 0, '>': 143, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6543948', 'n_tokens_mistral': 9552, 'n_tokens_neox': 9390, 'n_words': 2915}