<commit_before>/*
 * Copyright 2015-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.protocols.raft.impl;

import com.google.common.primitives.Longs;
import io.atomix.protocols.raft.RaftException;
import io.atomix.protocols.raft.RaftServer;
import io.atomix.protocols.raft.cluster.MemberId;
import io.atomix.protocols.raft.service.RaftService;
import io.atomix.protocols.raft.service.ServiceId;
import io.atomix.protocols.raft.service.ServiceType;
import io.atomix.protocols.raft.service.impl.DefaultServiceContext;
import io.atomix.protocols.raft.session.RaftSessionMetadata;
import io.atomix.protocols.raft.session.SessionId;
import io.atomix.protocols.raft.session.impl.RaftSessionContext;
import io.atomix.protocols.raft.session.impl.RaftSessionManager;
import io.atomix.protocols.raft.storage.log.RaftLog;
import io.atomix.protocols.raft.storage.log.RaftLogReader;
import io.atomix.protocols.raft.storage.log.entry.CloseSessionEntry;
import io.atomix.protocols.raft.storage.log.entry.CommandEntry;
import io.atomix.protocols.raft.storage.log.entry.ConfigurationEntry;
import io.atomix.protocols.raft.storage.log.entry.InitializeEntry;
import io.atomix.protocols.raft.storage.log.entry.KeepAliveEntry;
import io.atomix.protocols.raft.storage.log.entry.MetadataEntry;
import io.atomix.protocols.raft.storage.log.entry.OpenSessionEntry;
import io.atomix.protocols.raft.storage.log.entry.QueryEntry;
import io.atomix.protocols.raft.storage.log.entry.RaftLogEntry;
import io.atomix.storage.journal.Indexed;
import io.atomix.utils.concurrent.ComposableFuture;
import io.atomix.utils.concurrent.Futures;
import io.atomix.utils.concurrent.ThreadContext;
import io.atomix.utils.logging.ContextualLoggerFactory;
import io.atomix.utils.logging.LoggerContext;
import org.slf4j.Logger;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Supplier;
import java.util.stream.Collectors;

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

/**
 * Internal server state machine.
 * <p>
 * The internal state machine handles application of commands to the user provided {@link RaftService}
 * and keeps track of internal state like sessions and the various indexes relevant to log compaction.
 */
public class RaftServiceManager implements AutoCloseable {
  private static final long COMPACT_INTERVAL_MILLIS = 1000 * 10;

  private final Logger logger;
  private final RaftContext raft;
  private final ScheduledExecutorService threadPool;
  private final ThreadContext threadContext;
  private final RaftLog log;
  private final RaftLogReader reader;
  private final RaftSessionManager sessionManager = new RaftSessionManager();
  private final Map<String, DefaultServiceContext> services = new HashMap<>();
  private long lastCompacted;

  public RaftServiceManager(RaftContext raft, ScheduledExecutorService threadPool, ThreadContext threadContext) {
    this.raft = checkNotNull(raft, "state cannot be null");
    this.log = raft.getLog();
    this.reader = log.openReader(1, RaftLogReader.Mode.COMMITS);
    this.threadPool = threadPool;
    this.threadContext = threadContext;
    this.logger = ContextualLoggerFactory.getLogger(getClass(), LoggerContext.builder(RaftServer.class)
        .addValue(raft.getName())
        .build());
    threadContext.schedule(Duration.ofMillis(COMPACT_INTERVAL_MILLIS), this::compactLog);
  }

  /**
   * Returns the session manager.
   *
   * @return The session manager.
   */
  public RaftSessionManager getSessions() {
    return sessionManager;
  }

  /**
   * Applies all commits up to the given index.
   * <p>
   * Calls to this method are assumed not to expect a result. This allows some optimizations to be
   * made internally since linearizable events don't have to be waited to complete the command.
   *
   * @param index The index up to which to apply commits.
   */
  public void applyAll(long index) {
    // Don't attempt to apply indices that have already been applied.
    if (index > raft.getLastApplied()) {
      raft.getThreadContext().execute(() -> apply(index));
    }
  }

  /**
   * Applies the entry at the given index to the state machine.
   * <p>
   * Calls to this method are assumed to expect a result. This means linearizable session events
   * triggered by the application of the command at the given index will be awaited before completing
   * the returned future.
   *
   * @param index The index to apply.
   * @return A completable future to be completed once the commit has been applied.
   */
  public <T> CompletableFuture<T> apply(long index) {
    ComposableFuture<T> future = new ComposableFuture<>();
    threadContext.execute(() -> {
      // Apply entries prior to this entry.
      while (reader.hasNext()) {
        long nextIndex = reader.getNextIndex();

        // Validate that the next entry can be applied.
        long lastApplied = raft.getLastApplied();
        if (nextIndex > lastApplied + 1 && nextIndex != reader.getFirstIndex()) {
          logger.error("Cannot apply non-sequential index {} unless it's the first entry in the log: {}", nextIndex, reader.getFirstIndex());
          future.completeExceptionally(new IndexOutOfBoundsException("Cannot apply non-sequential index unless it's the first entry in the log"));
          return;
        } else if (nextIndex < lastApplied) {
          logger.error("Cannot apply duplicate entry at index {}", nextIndex);
          future.completeExceptionally(new IndexOutOfBoundsException("Cannot apply duplicate entry at index " + nextIndex));
          return;
        }

        // If the next index is less than or equal to the given index, read and apply the entry.
        if (nextIndex < index) {
          Indexed<RaftLogEntry> entry = reader.next();
          apply(entry);
          raft.setLastApplied(nextIndex);
        }
        // If the next index is equal to the applied index, apply it and return the result.
        else if (nextIndex == index) {
          // Read the entry from the log. If the entry is non-null then apply it, otherwise
          // simply update the last applied index and return a null result.
          try {
            Indexed<RaftLogEntry> entry = reader.next();
            if (entry.index() != index) {
              throw new IllegalStateException("inconsistent index applying entry " + index + ": " + entry);
            }
            this.<T>apply(entry).whenComplete(future);
            return;
          } finally {
            raft.setLastApplied(nextIndex);
          }
        }
        // If the applied index has been passed, return a null result.
        else {
          raft.setLastApplied(nextIndex);
          future.complete(null);
          return;
        }
      }
      future.completeExceptionally(new IndexOutOfBoundsException("Cannot commit index " + index));<commit_msg>Log error message when exception occurs applying an entry.<commit_after>/*
 * Copyright 2015-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.protocols.raft.impl;

import com.google.common.primitives.Longs;
import io.atomix.protocols.raft.RaftException;
import io.atomix.protocols.raft.RaftServer;
import io.atomix.protocols.raft.cluster.MemberId;
import io.atomix.protocols.raft.service.RaftService;
import io.atomix.protocols.raft.service.ServiceId;
import io.atomix.protocols.raft.service.ServiceType;
import io.atomix.protocols.raft.service.impl.DefaultServiceContext;
import io.atomix.protocols.raft.session.RaftSessionMetadata;
import io.atomix.protocols.raft.session.SessionId;
import io.atomix.protocols.raft.session.impl.RaftSessionContext;
import io.atomix.protocols.raft.session.impl.RaftSessionManager;
import io.atomix.protocols.raft.storage.log.RaftLog;
import io.atomix.protocols.raft.storage.log.RaftLogReader;
import io.atomix.protocols.raft.storage.log.entry.CloseSessionEntry;
import io.atomix.protocols.raft.storage.log.entry.CommandEntry;
import io.atomix.protocols.raft.storage.log.entry.ConfigurationEntry;
import io.atomix.protocols.raft.storage.log.entry.InitializeEntry;
import io.atomix.protocols.raft.storage.log.entry.KeepAliveEntry;
import io.atomix.protocols.raft.storage.log.entry.MetadataEntry;
import io.atomix.protocols.raft.storage.log.entry.OpenSessionEntry;
import io.atomix.protocols.raft.storage.log.entry.QueryEntry;
import io.atomix.protocols.raft.storage.log.entry.RaftLogEntry;
import io.atomix.storage.journal.Indexed;
import io.atomix.utils.concurrent.ComposableFuture;
import io.atomix.utils.concurrent.Futures;
import io.atomix.utils.concurrent.ThreadContext;
import io.atomix.utils.logging.ContextualLoggerFactory;
import io.atomix.utils.logging.LoggerContext;
import org.slf4j.Logger;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Supplier;
import java.util.stream.Collectors;

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

/**
 * Internal server state machine.
 * <p>
 * The internal state machine handles application of commands to the user provided {@link RaftService}
 * and keeps track of internal state like sessions and the various indexes relevant to log compaction.
 */
public class RaftServiceManager implements AutoCloseable {
  private static final long COMPACT_INTERVAL_MILLIS = 1000 * 10;

  private final Logger logger;
  private final RaftContext raft;
  private final ScheduledExecutorService threadPool;
  private final ThreadContext threadContext;
  private final RaftLog log;
  private final RaftLogReader reader;
  private final RaftSessionManager sessionManager = new RaftSessionManager();
  private final Map<String, DefaultServiceContext> services = new HashMap<>();
  private long lastCompacted;

  public RaftServiceManager(RaftContext raft, ScheduledExecutorService threadPool, ThreadContext threadContext) {
    this.raft = checkNotNull(raft, "state cannot be null");
    this.log = raft.getLog();
    this.reader = log.openReader(1, RaftLogReader.Mode.COMMITS);
    this.threadPool = threadPool;
    this.threadContext = threadContext;
    this.logger = ContextualLoggerFactory.getLogger(getClass(), LoggerContext.builder(RaftServer.class)
        .addValue(raft.getName())
        .build());
    threadContext.schedule(Duration.ofMillis(COMPACT_INTERVAL_MILLIS), this::compactLog);
  }

  /**
   * Returns the session manager.
   *
   * @return The session manager.
   */
  public RaftSessionManager getSessions() {
    return sessionManager;
  }

  /**
   * Applies all commits up to the given index.
   * <p>
   * Calls to this method are assumed not to expect a result. This allows some optimizations to be
   * made internally since linearizable events don't have to be waited to complete the command.
   *
   * @param index The index up to which to apply commits.
   */
  public void applyAll(long index) {
    // Don't attempt to apply indices that have already been applied.
    if (index > raft.getLastApplied()) {
      raft.getThreadContext().execute(() -> apply(index));
    }
  }

  /**
   * Applies the entry at the given index to the state machine.
   * <p>
   * Calls to this method are assumed to expect a result. This means linearizable session events
   * triggered by the application of the command at the given index will be awaited before completing
   * the returned future.
   *
   * @param index The index to apply.
   * @return A completable future to be completed once the commit has been applied.
   */
  public <T> CompletableFuture<T> apply(long index) {
    ComposableFuture<T> future = new ComposableFuture<>();
    threadContext.execute(() -> {
      // Apply entries prior to this entry.
      while (reader.hasNext()) {
        long nextIndex = reader.getNextIndex();

        // Validate that the next entry can be applied.
        long lastApplied = raft.getLastApplied();
        if (nextIndex > lastApplied + 1 && nextIndex != reader.getFirstIndex()) {
          logger.error("Cannot apply non-sequential index {} unless it's the first entry in the log: {}", nextIndex, reader.getFirstIndex());
          future.completeExceptionally(new IndexOutOfBoundsException("Cannot apply non-sequential index unless it's the first entry in the log"));
          return;
        } else if (nextIndex < lastApplied) {
          logger.error("Cannot apply duplicate entry at index {}", nextIndex);
          future.completeExceptionally(new IndexOutOfBoundsException("Cannot apply duplicate entry at index " + nextIndex));
          return;
        }

        // If the next index is less than or equal to the given index, read and apply the entry.
        if (nextIndex < index) {
          Indexed<RaftLogEntry> entry = reader.next();
          try {
            apply(entry);
          } catch (Exception e) {
            logger.error("Failed to apply {}: {}", entry, e);
          } finally {
            raft.setLastApplied(nextIndex);
          }
        }
        // If the next index is equal to the applied index, apply it and return the result.
        else if (nextIndex == index) {
          // Read the entry from the log. If the entry is non-null then apply it, otherwise
          // simply update the last applied index and return a null result.
          Indexed<RaftLogEntry> entry = reader.next();
          try {
            if (entry.index() != index) {
              throw new IllegalStateException("inconsistent index applying entry " + index + ": " + entry);
            }
            this.<T>apply(entry).whenComplete(future);
            return;
          } catch (Exception e) {
            logger.error("Failed to apply {}: {}", entry, e);
          } finally {
            raft.setLastApplied(nextIndex);
          }
        }
        // If the applied index has been passed, return a null result.
        else {
          raft.setLastApplied(nextIndex);
          future.complete(null);
          return;
        }
      }
      future.completeExceptionally(new IndexOutOfBoundsException("Cannot commit index " + index));
{'fraction_non_alphanumeric': 0.09258661849151559, 'fraction_numerical': 0.002193689915478418, 'mean_word_length': 3.5215869311551926, 'pattern_counts': {'":': 2, '<': 31, '<?xml version=': 0, '>': 35, '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': '4157632', 'n_tokens_mistral': 4285, 'n_tokens_neox': 4198, 'n_words': 1289}