<commit_before>/* This file is part of VoltDB.
 * Copyright (C) 2008-2019 VoltDB Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */

package org.voltdb.iv2;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

import org.voltcore.logging.VoltLogger;
import org.voltcore.messaging.HostMessenger;
import org.voltcore.messaging.Mailbox;
import org.voltcore.messaging.TransactionInfoBaseMessage;
import org.voltcore.messaging.VoltMessage;
import org.voltcore.utils.CoreUtils;
import org.voltdb.ClientResponseImpl;
import org.voltdb.CommandLog;
import org.voltdb.CommandLog.DurabilityListener;
import org.voltdb.RealVoltDB;
import org.voltdb.SnapshotCompletionInterest;
import org.voltdb.SnapshotCompletionMonitor;
import org.voltdb.SystemProcedureCatalog;
import org.voltdb.VoltDB;
import org.voltdb.VoltDBInterface;
import org.voltdb.VoltTable;
import org.voltdb.client.ClientResponse;
import org.voltdb.dtxn.TransactionState;
import org.voltdb.exceptions.SerializableException;
import org.voltdb.exceptions.TransactionRestartException;
import org.voltdb.iv2.SiteTasker.SiteTaskerRunnable;
import org.voltdb.messaging.BorrowTaskMessage;
import org.voltdb.messaging.CompleteTransactionMessage;
import org.voltdb.messaging.CompleteTransactionResponseMessage;
import org.voltdb.messaging.DummyTransactionResponseMessage;
import org.voltdb.messaging.DummyTransactionTaskMessage;
import org.voltdb.messaging.DumpMessage;
import org.voltdb.messaging.DumpPlanThenExitMessage;
import org.voltdb.messaging.FragmentResponseMessage;
import org.voltdb.messaging.FragmentTaskMessage;
import org.voltdb.messaging.InitiateResponseMessage;
import org.voltdb.messaging.Iv2InitiateTaskMessage;
import org.voltdb.messaging.Iv2LogFaultMessage;
import org.voltdb.messaging.MPBacklogFlushMessage;
import org.voltdb.messaging.MultiPartitionParticipantMessage;
import org.voltdb.messaging.RepairLogTruncationMessage;
import org.voltdb.utils.MiscUtils;
import org.voltdb.utils.VoltTrace;

import com.google_voltpatches.common.collect.Sets;
import com.google_voltpatches.common.primitives.Ints;
import com.google_voltpatches.common.primitives.Longs;
import com.google_voltpatches.common.util.concurrent.ListenableFuture;
import com.google_voltpatches.common.util.concurrent.SettableFuture;

public class SpScheduler extends Scheduler implements SnapshotCompletionInterest
{
    static final VoltLogger tmLog = new VoltLogger("TM");
    static final VoltLogger hostLog = new VoltLogger("HOST");
    static class DuplicateCounterKey implements Comparable<DuplicateCounterKey> {
        private final long m_txnId;
        private final long m_spHandle;

        DuplicateCounterKey(long txnId, long spHandle) {
            m_txnId = txnId;
            m_spHandle = spHandle;
        }

        @Override
        public boolean equals(Object o) {
            try {
                DuplicateCounterKey other = (DuplicateCounterKey) o;
                return (m_txnId == other.m_txnId && m_spHandle == other.m_spHandle);
            }
            catch (Exception e) {
                return false;
            }
        }

        // Only care about comparing TXN ID part for sorting in updateReplicas
        @Override
        public int compareTo(DuplicateCounterKey o) {
            if (m_txnId < o.m_txnId) {
                return -1;
            } else if (m_txnId > o.m_txnId) {
                return 1;
            } else {
                if (m_spHandle < o.m_spHandle) {
                    return -1;
                }
                else if (m_spHandle > o.m_spHandle) {
                    return 1;
                }
                else {
                    return 0;
                }
            }
        }

        @Override
        public int hashCode() {
            assert(false) : "Hashing this is unsafe as it can't promise no collisions.";
            throw new UnsupportedOperationException(
                    "Hashing this is unsafe as it can't promise no collisions.");
        }

        @Override
        public String toString() {
            return "[txn:" + TxnEgo.txnIdToString(m_txnId) + ", spHandle:" + TxnEgo.txnIdToString(m_spHandle) + "]";
        }

        public boolean isSpTransaction() {
            return (TxnEgo.getPartitionId(m_txnId) != MpInitiator.MP_INIT_PID);
        }
    };

    public interface DurableUniqueIdListener {
        /**
         * Notify listener of last durable Single-Part and Multi-Part uniqueIds
         */
        public void lastUniqueIdsMadeDurable(long spUniqueId, long mpUniqueId);
    }

    private List<Long> m_replicaHSIds = new ArrayList<>();
    long m_sendToHSIds[] = new long[0];
    private final TransactionTaskQueue m_pendingTasks;
    private final Map<Long, TransactionState> m_outstandingTxns =
        new HashMap<Long, TransactionState>();
    private final TreeMap<DuplicateCounterKey, DuplicateCounter> m_duplicateCounters =
        new TreeMap<DuplicateCounterKey, DuplicateCounter>();
    // MP fragment tasks or completion tasks pending durability
    private final Map<Long, Queue<TransactionTask>> m_mpsPendingDurability =
        new HashMap<Long, Queue<TransactionTask>>();
    private CommandLog m_cl;
    private final SnapshotCompletionMonitor m_snapMonitor;
    private BufferedReadLog m_bufferedReadLog = null;

    // Need to track when command log replay is complete (even if not performed) so that
    // we know when we can start writing viable replay sets to the fault log.
    boolean m_replayComplete = false;
    // The DurabilityListener is not thread-safe. Access it only on the Site thread.
    private final DurabilityListener m_durabilityListener;
    // Generator of pre-IV2ish timestamp based unique IDs
    private final UniqueIdGenerator m_uniqueIdGenerator;

    // the current not-needed-any-more point of the repair log.
    long m_repairLogTruncationHandle = Long.MIN_VALUE;
    // the truncation handle last sent to the replicas
    long m_lastSentTruncationHandle = Long.MIN_VALUE;
    // the max schedule transaction sphandle, multi-fragments mp txn counts one
    long m_maxScheduledTxnSpHandle = Long.MIN_VALUE;

    // the checkpoint transaction sphandle upon MigratePartitionLeader is initiated
    long m_migratePartitionLeaderCheckPoint = Long.MIN_VALUE;

    //The RepairLog is the same instance as the one initialized in InitiatorMailbox.
    //Iv2IniatiateTaskMessage, FragmentTaskMessage and CompleteTransactionMessage
    //are to be added to the repair log when these messages get updated transaction ids.
    protected RepairLog m_repairLog;

    private final boolean IS_KSAFE_CLUSTER;

    SpScheduler(int partitionId, SiteTaskerQueue taskQueue, SnapshotCompletionMonitor snapMonitor, boolean scoreboardEnabled)
    {
        super(partitionId, taskQueue);
        m_pendingTasks = new TransactionTaskQueue(m_tasks, scoreboardEnabled);
        m_snapMonitor = snapMonitor;
        m_durabilityListener = new SpDurabilityListener(this, m_pendingTasks);
        m_uniqueIdGenerator = new UniqueIdGenerator(partitionId, 0);
        m_bufferedReadLog = new BufferedReadLog();
        m_repairLogTruncationHandle = getCurrentTxnId();
        // initialized as current txn id in order to release the initial reads into the system
        m_maxScheduledTxnSpHandle = getCurrentTxnId();
        IS_KSAFE_CLUSTER = VoltDB.instance().getKFactor() > 0;
    }

    public void initializeScoreboard(int siteId, Mailbox mailBox) {
        m_pendingTasks.initializeScoreboard(siteId, mailBox);
    }

    @Override
    public void setLeaderState(boolean isLeader)
    {
        super.setLeaderState(isLeader);
        m_snapMonitor.addInterest(this);
        VoltDBInterface db = VoltDB.instance();
        if (isLeader && db instanceof RealVoltDB ) {
            SpInitiator init = (SpInitiator)((RealVoltDB)db).getInitiator(m_partitionId);
            if (init.m_term != null) {
                ((SpTerm)init.m_term).setPromoting(false);
            }
        }
    }

    @Override
    public void setMaxSeenTxnId(long maxSeenTxnId)
    {
        super.setMaxSeenTxnId(maxSeenTxnId);
        writeIv2ViableReplayEntry();
    }

    @Override
    public void configureDurableUniqueIdListener(final DurableUniqueIdListener listener, final boolean install) {
        m_tasks.offer(new SiteTaskerRunnable() {
            @Override
            void run()
            {
                m_durabilityListener.configureUniqueIdListener(listener, install);
            }
            private SiteTaskerRunnable init(DurableUniqueIdListener listener){
                taskInfo = listener.getClass().getSimpleName();
                return this;
            }
        }.init(listener));
    }

    @Override
    public void shutdown()
    {
        m_tasks.offer(m_nullTask);
    }

    // This is going to run in the BabySitter's thread.  This and deliver are synchronized by
    // virtue of both being called on InitiatorMailbox and not directly called.
    // (That is, InitiatorMailbox's API, used by BabySitter, is synchronized on the same
    // lock deliver() is synchronized on.)
    @Override
    public long[] updateReplicas(List<Long> replicas, Map<Integer, Long> partitionMasters,
            TransactionState snapshotTransactionState)
    {
        if (tmLog.isDebugEnabled()) {
            tmLog.debug("[SpScheduler.updateReplicas] replicas to " + CoreUtils.hsIdCollectionToString(replicas) +
                    " on " + CoreUtils.hsIdToString(m_mailbox.getHSId())
             + " from " + CoreUtils.hsIdCollectionToString(m_replicaHSIds));
        }
        long[] replicasAdded = new long[0];
        if (m_replicaHSIds.size() > 0 && replicas.size() > m_replicaHSIds.size()) {
            // Remember the rejoin sites before update replicas set
            Set<Long> rejoinHSIds = Sets.difference(new HashSet<Long>(replicas),
                                                  new HashSet<Long>(m_replicaHSIds));
            replicasAdded = Longs.toArray(rejoinHSIds);
        }
        // First - correct the official replica set.
        m_replicaHSIds = replicas;
        // Update the list of remote replicas that we'll need to send to
        List<Long> sendToHSIds = new ArrayList<Long>(m_replicaHSIds);
        sendToHSIds.remove(m_mailbox.getHSId());
        m_sendToHSIds = Longs.toArray(sendToHSIds);

        // A new site joins in, forward the current txn (stream snapshot save) message to new site
        if (m_isLeader && snapshotTransactionState != null) {
            // Look up the DuplicateCounter for this snapshots fragment
            DuplicateCounterKey key = new DuplicateCounterKey(snapshotTransactionState.txnId,
                    snapshotTransactionState.m_spHandle);
            DuplicateCounter duplicateCounter = m_duplicateCounters.get(key);
            assert (duplicateCounter != null);
            duplicateCounter.addReplicas(replicasAdded);
            // Forward fragment message to new replica
            m_mailbox.send(replicasAdded, duplicateCounter.getOpenMessage());
        }

        // Cleanup duplicate counters and collect DONE counters
        // in this list for further processing.
        List<DuplicateCounterKey> doneCounters = new LinkedList<DuplicateCounterKey>();
        for (Entry<DuplicateCounterKey, DuplicateCounter> entry : m_duplicateCounters.entrySet()) {
            DuplicateCounter counter = entry.getValue();
            int result = counter.updateReplicas(m_replicaHSIds);
            if (result == DuplicateCounter.DONE) {
                doneCounters.add(entry.getKey());
            }
        }

        //notify the new partition leader that the old leader has completed the Txns if needed
        //after duplicate counters are cleaned. m_mailbox can be MockMailBox which is used for
        //unit test
        if (!m_isLeader && m_mailbox instanceof InitiatorMailbox) {
            ((InitiatorMailbox)m_mailbox).notifyNewLeaderOfTxnDoneIfNeeded();
        }

        // Maintain the CI invariant that responses arrive in txnid order.
        Collections.sort(doneCounters);
        for (DuplicateCounterKey key : doneCounters) {
            DuplicateCounter counter = m_duplicateCounters.remove(key);

            final TransactionState txn = m_outstandingTxns.get(key.m_txnId);
            if (txn == null || txn.isDone()) {
                m_outstandingTxns.remove(key.m_txnId);
                // for MP write txns, we should use it's first SpHandle in the TransactionState
                // for SP write txns, we can just use the SpHandle from the DuplicateCounterKey
                long safeSpHandle = txn == null ? key.m_spHandle: txn.m_spHandle;
                setRepairLogTruncationHandle(safeSpHandle, false);
            }

            VoltMessage resp = counter.getLastResponse();
            if (resp != null) {
                // MPI is tracking deps per partition HSID.  We need to make
                // sure we write ours into the message getting sent to the MPI
                if (resp instanceof FragmentResponseMessage) {
                    FragmentResponseMessage fresp = (FragmentResponseMessage)resp;
                    fresp.setExecutorSiteId(m_mailbox.getHSId());
                }
                m_mailbox.send(counter.m_destinationId, resp);
            }
            else {
                hostLog.warn("TXN " + counter.getTxnId() + " lost all replicas and " +
                        "had no responses.  This should be impossible?");
            }
        }

        long uniqueId;
        if (snapshotTransactionState == null) {
            uniqueId = m_uniqueIdGenerator.getLastUniqueId();
        } else {
            // When there is a snapshot transaction state use the lastSpUniqueId from that
            uniqueId = snapshotTransactionState.m_lastSpUniqueId;
        }
        SettableFuture<Boolean> written = writeIv2ViableReplayEntry(uniqueId);

        // Get the fault log status here to ensure the leader has written it to disk
        // before initiating transactions again.
        blockFaultLogWriteStatus(written);

        return replicasAdded;
    }

    /**
     * Poll the replay sequencer and process the messages until it returns null
     */
    private void deliverReadyTxns() {
        // First, pull all the sequenced messages, if any.
        VoltMessage m = m_replaySequencer.poll();
        while(m != null) {
            deliver(m);
            m = m_replaySequencer.poll();
        }
        // Then, try to pull all the drainable messages, if any.
        m = m_replaySequencer.drain();
        while (m != null) {
            if (m instanceof Iv2InitiateTaskMessage) {
                // Send IGNORED response for all SPs
                Iv2InitiateTaskMessage task = (Iv2InitiateTaskMessage) m;
                final InitiateResponseMessage response = new InitiateResponseMessage(task);
                response.setResults(new ClientResponseImpl(ClientResponse.UNEXPECTED_FAILURE,
                            new VoltTable[0],
                            ClientResponseImpl.IGNORED_TRANSACTION));
                m_mailbox.send(response.getInitiatorHSId(), response);
            }
            m = m_replaySequencer.drain();
        }
    }

    /**
     * Sequence the message for replay if it's for CL or DR.
     *
     * @param message
     * @return true if the message can be delivered directly to the scheduler,
     * false if the message is queued
     */
    @Override
    public boolean sequenceForReplay(VoltMessage message)
    {
        boolean canDeliver = false;
        long sequenceWithUniqueId = Long.MIN_VALUE;

        boolean commandLog = (message instanceof TransactionInfoBaseMessage &&
                (((TransactionInfoBaseMessage)message).isForReplay()));

        boolean sentinel = message instanceof MultiPartitionParticipantMessage;

        boolean replay = commandLog || sentinel;
        boolean sequenceForReplay = m_isLeader && replay;

        if (replay) {
            sequenceWithUniqueId = ((TransactionInfoBaseMessage)message).getUniqueId();
        }

        if (sequenceForReplay) {
            InitiateResponseMessage dupe = m_replaySequencer.dedupe(sequenceWithUniqueId,
                    (TransactionInfoBaseMessage) message);
            if (dupe != null) {
                // Duplicate initiate task message, send response
                m_mailbox.send(dupe.getInitiatorHSId(), dupe);
            }
            else if (!m_replaySequencer.offer(sequenceWithUniqueId, (TransactionInfoBaseMessage) message)) {
                canDeliver = true;
            }
            else {
                deliverReadyTxns();
            }

            // If it's a DR sentinel, send an acknowledgement
            if (sentinel && !commandLog) {
                MultiPartitionParticipantMessage mppm = (MultiPartitionParticipantMessage) message;
                final InitiateResponseMessage response = new InitiateResponseMessage(mppm);
                ClientResponseImpl clientResponse =
                        new ClientResponseImpl(ClientResponseImpl.UNEXPECTED_FAILURE,
                                new VoltTable[0], ClientResponseImpl.IGNORED_TRANSACTION);
                response.setResults(clientResponse);
                m_mailbox.send(response.getInitiatorHSId(), response);
            }
        }
        else {
            if (replay) {
                // Update last seen and last polled uniqueId for replicas
                m_replaySequencer.updateLastSeenUniqueId(sequenceWithUniqueId,
                        (TransactionInfoBaseMessage) message);
                m_replaySequencer.updateLastPolledUniqueId(sequenceWithUniqueId,
                        (TransactionInfoBaseMessage) message);
            }

            canDeliver = true;
        }

        return canDeliver;
    }

    // SpInitiators will see every message type.  The Responses currently come
    // from local work, but will come from replicas when replication is
    // implemented
    @Override
    public void deliver(VoltMessage message)
    {
        if (message instanceof Iv2InitiateTaskMessage) {
            handleIv2InitiateTaskMessage((Iv2InitiateTaskMessage)message);
        }
        else if (message instanceof InitiateResponseMessage) {
            handleInitiateResponseMessage((InitiateResponseMessage)message);
        }
        else if (message instanceof FragmentTaskMessage) {
            handleFragmentTaskMessage((FragmentTaskMessage)message);
        }
        else if (message instanceof FragmentResponseMessage) {
            handleFragmentResponseMessage((FragmentResponseMessage)message);
        }
        else if (message instanceof CompleteTransactionMessage) {
            handleCompleteTransactionMessage((CompleteTransactionMessage)message);
        }
        else if (message instanceof CompleteTransactionResponseMessage) {
            handleCompleteTransactionResponseMessage((CompleteTransactionResponseMessage) message);
        }
        else if (message instanceof BorrowTaskMessage) {
            handleBorrowTaskMessage((BorrowTaskMessage)message);
        }
        else if (message instanceof Iv2LogFaultMessage) {
            handleIv2LogFaultMessage((Iv2LogFaultMessage)message);
        }
        else if (message instanceof DumpMessage) {
            handleDumpMessage((DumpMessage)message);
        } else if (message instanceof DumpPlanThenExitMessage) {
            handleDumpPlanMessage((DumpPlanThenExitMessage)message);
        }
        else if (message instanceof DummyTransactionTaskMessage) {
            handleDummyTransactionTaskMessage((DummyTransactionTaskMessage) message);
        }
        else if (message instanceof DummyTransactionResponseMessage) {
            handleDummyTransactionResponseMessage((DummyTransactionResponseMessage)message);
        }
        else if (message instanceof MPBacklogFlushMessage) {
            cleanupTransactionBacklogOnRepair();
        }
        else {
            throw new RuntimeException("UNKNOWN MESSAGE TYPE, BOOM!");
        }
    }

    // SpScheduler expects to see InitiateTaskMessages corresponding to single-partition
    // procedures only.
    private void handleIv2InitiateTaskMessage(Iv2InitiateTaskMessage message)
    {
        if (!message.isSinglePartition()) {
            throw new RuntimeException("SpScheduler.handleIv2InitiateTaskMessage " +
                    "should never receive multi-partition initiations.");
        }
        final String procedureName = message.getStoredProcedureName();
        long newSpHandle;
        long uniqueId = Long.MIN_VALUE;
        Iv2InitiateTaskMessage msg = message;
        if (m_isLeader || message.isReadOnly()) {

            /*
             * If this is for CL replay or DR, update the unique ID generator
             */
            if (message.isForReplay()) {
                uniqueId = message.getUniqueId();
                try {
                    m_uniqueIdGenerator.updateMostRecentlyGeneratedUniqueId(uniqueId);
                }
                catch (Exception e) {
                    hostLog.fatal(e.getMessage());
                    hostLog.fatal("Invocation: " + message);
                    VoltDB.crashLocalVoltDB(e.getMessage(), true, e);
                }<commit_msg>ENG-17721: Turn Runtime Exception to crashLocalVoltDB and log the er… (#6511)<commit_after>/* This file is part of VoltDB.
 * Copyright (C) 2008-2019 VoltDB Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */

package org.voltdb.iv2;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

import org.voltcore.logging.VoltLogger;
import org.voltcore.messaging.HostMessenger;
import org.voltcore.messaging.Mailbox;
import org.voltcore.messaging.TransactionInfoBaseMessage;
import org.voltcore.messaging.VoltMessage;
import org.voltcore.utils.CoreUtils;
import org.voltdb.ClientResponseImpl;
import org.voltdb.CommandLog;
import org.voltdb.CommandLog.DurabilityListener;
import org.voltdb.RealVoltDB;
import org.voltdb.SnapshotCompletionInterest;
import org.voltdb.SnapshotCompletionMonitor;
import org.voltdb.SystemProcedureCatalog;
import org.voltdb.VoltDB;
import org.voltdb.VoltDBInterface;
import org.voltdb.VoltTable;
import org.voltdb.client.ClientResponse;
import org.voltdb.dtxn.TransactionState;
import org.voltdb.exceptions.SerializableException;
import org.voltdb.exceptions.TransactionRestartException;
import org.voltdb.iv2.SiteTasker.SiteTaskerRunnable;
import org.voltdb.messaging.BorrowTaskMessage;
import org.voltdb.messaging.CompleteTransactionMessage;
import org.voltdb.messaging.CompleteTransactionResponseMessage;
import org.voltdb.messaging.DummyTransactionResponseMessage;
import org.voltdb.messaging.DummyTransactionTaskMessage;
import org.voltdb.messaging.DumpMessage;
import org.voltdb.messaging.DumpPlanThenExitMessage;
import org.voltdb.messaging.FragmentResponseMessage;
import org.voltdb.messaging.FragmentTaskMessage;
import org.voltdb.messaging.InitiateResponseMessage;
import org.voltdb.messaging.Iv2InitiateTaskMessage;
import org.voltdb.messaging.Iv2LogFaultMessage;
import org.voltdb.messaging.MPBacklogFlushMessage;
import org.voltdb.messaging.MultiPartitionParticipantMessage;
import org.voltdb.messaging.RepairLogTruncationMessage;
import org.voltdb.utils.MiscUtils;
import org.voltdb.utils.VoltTrace;

import com.google_voltpatches.common.collect.Sets;
import com.google_voltpatches.common.primitives.Ints;
import com.google_voltpatches.common.primitives.Longs;
import com.google_voltpatches.common.util.concurrent.ListenableFuture;
import com.google_voltpatches.common.util.concurrent.SettableFuture;

public class SpScheduler extends Scheduler implements SnapshotCompletionInterest
{
    static final VoltLogger tmLog = new VoltLogger("TM");
    static final VoltLogger hostLog = new VoltLogger("HOST");
    static class DuplicateCounterKey implements Comparable<DuplicateCounterKey> {
        private final long m_txnId;
        private final long m_spHandle;

        DuplicateCounterKey(long txnId, long spHandle) {
            m_txnId = txnId;
            m_spHandle = spHandle;
        }

        @Override
        public boolean equals(Object o) {
            try {
                DuplicateCounterKey other = (DuplicateCounterKey) o;
                return (m_txnId == other.m_txnId && m_spHandle == other.m_spHandle);
            }
            catch (Exception e) {
                return false;
            }
        }

        // Only care about comparing TXN ID part for sorting in updateReplicas
        @Override
        public int compareTo(DuplicateCounterKey o) {
            if (m_txnId < o.m_txnId) {
                return -1;
            } else if (m_txnId > o.m_txnId) {
                return 1;
            } else {
                if (m_spHandle < o.m_spHandle) {
                    return -1;
                }
                else if (m_spHandle > o.m_spHandle) {
                    return 1;
                }
                else {
                    return 0;
                }
            }
        }

        @Override
        public int hashCode() {
            assert(false) : "Hashing this is unsafe as it can't promise no collisions.";
            throw new UnsupportedOperationException(
                    "Hashing this is unsafe as it can't promise no collisions.");
        }

        @Override
        public String toString() {
            return "[txn:" + TxnEgo.txnIdToString(m_txnId) + ", spHandle:" + TxnEgo.txnIdToString(m_spHandle) + "]";
        }

        public boolean isSpTransaction() {
            return (TxnEgo.getPartitionId(m_txnId) != MpInitiator.MP_INIT_PID);
        }
    };

    public interface DurableUniqueIdListener {
        /**
         * Notify listener of last durable Single-Part and Multi-Part uniqueIds
         */
        public void lastUniqueIdsMadeDurable(long spUniqueId, long mpUniqueId);
    }

    private List<Long> m_replicaHSIds = new ArrayList<>();
    long m_sendToHSIds[] = new long[0];
    private final TransactionTaskQueue m_pendingTasks;
    private final Map<Long, TransactionState> m_outstandingTxns =
        new HashMap<Long, TransactionState>();
    private final TreeMap<DuplicateCounterKey, DuplicateCounter> m_duplicateCounters =
        new TreeMap<DuplicateCounterKey, DuplicateCounter>();
    // MP fragment tasks or completion tasks pending durability
    private final Map<Long, Queue<TransactionTask>> m_mpsPendingDurability =
        new HashMap<Long, Queue<TransactionTask>>();
    private CommandLog m_cl;
    private final SnapshotCompletionMonitor m_snapMonitor;
    private BufferedReadLog m_bufferedReadLog = null;

    // Need to track when command log replay is complete (even if not performed) so that
    // we know when we can start writing viable replay sets to the fault log.
    boolean m_replayComplete = false;
    // The DurabilityListener is not thread-safe. Access it only on the Site thread.
    private final DurabilityListener m_durabilityListener;
    // Generator of pre-IV2ish timestamp based unique IDs
    private final UniqueIdGenerator m_uniqueIdGenerator;

    // the current not-needed-any-more point of the repair log.
    long m_repairLogTruncationHandle = Long.MIN_VALUE;
    // the truncation handle last sent to the replicas
    long m_lastSentTruncationHandle = Long.MIN_VALUE;
    // the max schedule transaction sphandle, multi-fragments mp txn counts one
    long m_maxScheduledTxnSpHandle = Long.MIN_VALUE;

    // the checkpoint transaction sphandle upon MigratePartitionLeader is initiated
    long m_migratePartitionLeaderCheckPoint = Long.MIN_VALUE;

    //The RepairLog is the same instance as the one initialized in InitiatorMailbox.
    //Iv2IniatiateTaskMessage, FragmentTaskMessage and CompleteTransactionMessage
    //are to be added to the repair log when these messages get updated transaction ids.
    protected RepairLog m_repairLog;

    private final boolean IS_KSAFE_CLUSTER;

    SpScheduler(int partitionId, SiteTaskerQueue taskQueue, SnapshotCompletionMonitor snapMonitor, boolean scoreboardEnabled)
    {
        super(partitionId, taskQueue);
        m_pendingTasks = new TransactionTaskQueue(m_tasks, scoreboardEnabled);
        m_snapMonitor = snapMonitor;
        m_durabilityListener = new SpDurabilityListener(this, m_pendingTasks);
        m_uniqueIdGenerator = new UniqueIdGenerator(partitionId, 0);
        m_bufferedReadLog = new BufferedReadLog();
        m_repairLogTruncationHandle = getCurrentTxnId();
        // initialized as current txn id in order to release the initial reads into the system
        m_maxScheduledTxnSpHandle = getCurrentTxnId();
        IS_KSAFE_CLUSTER = VoltDB.instance().getKFactor() > 0;
    }

    public void initializeScoreboard(int siteId, Mailbox mailBox) {
        m_pendingTasks.initializeScoreboard(siteId, mailBox);
    }

    @Override
    public void setLeaderState(boolean isLeader)
    {
        super.setLeaderState(isLeader);
        m_snapMonitor.addInterest(this);
        VoltDBInterface db = VoltDB.instance();
        if (isLeader && db instanceof RealVoltDB ) {
            SpInitiator init = (SpInitiator)((RealVoltDB)db).getInitiator(m_partitionId);
            if (init.m_term != null) {
                ((SpTerm)init.m_term).setPromoting(false);
            }
        }
    }

    @Override
    public void setMaxSeenTxnId(long maxSeenTxnId)
    {
        super.setMaxSeenTxnId(maxSeenTxnId);
        writeIv2ViableReplayEntry();
    }

    @Override
    public void configureDurableUniqueIdListener(final DurableUniqueIdListener listener, final boolean install) {
        m_tasks.offer(new SiteTaskerRunnable() {
            @Override
            void run()
            {
                m_durabilityListener.configureUniqueIdListener(listener, install);
            }
            private SiteTaskerRunnable init(DurableUniqueIdListener listener){
                taskInfo = listener.getClass().getSimpleName();
                return this;
            }
        }.init(listener));
    }

    @Override
    public void shutdown()
    {
        m_tasks.offer(m_nullTask);
    }

    // This is going to run in the BabySitter's thread.  This and deliver are synchronized by
    // virtue of both being called on InitiatorMailbox and not directly called.
    // (That is, InitiatorMailbox's API, used by BabySitter, is synchronized on the same
    // lock deliver() is synchronized on.)
    @Override
    public long[] updateReplicas(List<Long> replicas, Map<Integer, Long> partitionMasters,
            TransactionState snapshotTransactionState)
    {
        if (tmLog.isDebugEnabled()) {
            tmLog.debug("[SpScheduler.updateReplicas] replicas to " + CoreUtils.hsIdCollectionToString(replicas) +
                    " on " + CoreUtils.hsIdToString(m_mailbox.getHSId())
             + " from " + CoreUtils.hsIdCollectionToString(m_replicaHSIds));
        }
        long[] replicasAdded = new long[0];
        if (m_replicaHSIds.size() > 0 && replicas.size() > m_replicaHSIds.size()) {
            // Remember the rejoin sites before update replicas set
            Set<Long> rejoinHSIds = Sets.difference(new HashSet<Long>(replicas),
                                                  new HashSet<Long>(m_replicaHSIds));
            replicasAdded = Longs.toArray(rejoinHSIds);
        }
        // First - correct the official replica set.
        m_replicaHSIds = replicas;
        // Update the list of remote replicas that we'll need to send to
        List<Long> sendToHSIds = new ArrayList<Long>(m_replicaHSIds);
        sendToHSIds.remove(m_mailbox.getHSId());
        m_sendToHSIds = Longs.toArray(sendToHSIds);

        // A new site joins in, forward the current txn (stream snapshot save) message to new site
        if (m_isLeader && snapshotTransactionState != null) {
            // Look up the DuplicateCounter for this snapshots fragment
            DuplicateCounterKey key = new DuplicateCounterKey(snapshotTransactionState.txnId,
                    snapshotTransactionState.m_spHandle);
            DuplicateCounter duplicateCounter = m_duplicateCounters.get(key);
            assert (duplicateCounter != null);
            duplicateCounter.addReplicas(replicasAdded);
            // Forward fragment message to new replica
            m_mailbox.send(replicasAdded, duplicateCounter.getOpenMessage());
        }

        // Cleanup duplicate counters and collect DONE counters
        // in this list for further processing.
        List<DuplicateCounterKey> doneCounters = new LinkedList<DuplicateCounterKey>();
        for (Entry<DuplicateCounterKey, DuplicateCounter> entry : m_duplicateCounters.entrySet()) {
            DuplicateCounter counter = entry.getValue();
            int result = counter.updateReplicas(m_replicaHSIds);
            if (result == DuplicateCounter.DONE) {
                doneCounters.add(entry.getKey());
            }
        }

        //notify the new partition leader that the old leader has completed the Txns if needed
        //after duplicate counters are cleaned. m_mailbox can be MockMailBox which is used for
        //unit test
        if (!m_isLeader && m_mailbox instanceof InitiatorMailbox) {
            ((InitiatorMailbox)m_mailbox).notifyNewLeaderOfTxnDoneIfNeeded();
        }

        // Maintain the CI invariant that responses arrive in txnid order.
        Collections.sort(doneCounters);
        for (DuplicateCounterKey key : doneCounters) {
            DuplicateCounter counter = m_duplicateCounters.remove(key);

            final TransactionState txn = m_outstandingTxns.get(key.m_txnId);
            if (txn == null || txn.isDone()) {
                m_outstandingTxns.remove(key.m_txnId);
                // for MP write txns, we should use it's first SpHandle in the TransactionState
                // for SP write txns, we can just use the SpHandle from the DuplicateCounterKey
                long safeSpHandle = txn == null ? key.m_spHandle: txn.m_spHandle;
                setRepairLogTruncationHandle(safeSpHandle, false);
            }

            VoltMessage resp = counter.getLastResponse();
            if (resp != null) {
                // MPI is tracking deps per partition HSID.  We need to make
                // sure we write ours into the message getting sent to the MPI
                if (resp instanceof FragmentResponseMessage) {
                    FragmentResponseMessage fresp = (FragmentResponseMessage)resp;
                    fresp.setExecutorSiteId(m_mailbox.getHSId());
                }
                m_mailbox.send(counter.m_destinationId, resp);
            }
            else {
                hostLog.warn("TXN " + counter.getTxnId() + " lost all replicas and " +
                        "had no responses.  This should be impossible?");
            }
        }

        long uniqueId;
        if (snapshotTransactionState == null) {
            uniqueId = m_uniqueIdGenerator.getLastUniqueId();
        } else {
            // When there is a snapshot transaction state use the lastSpUniqueId from that
            uniqueId = snapshotTransactionState.m_lastSpUniqueId;
        }
        SettableFuture<Boolean> written = writeIv2ViableReplayEntry(uniqueId);

        // Get the fault log status here to ensure the leader has written it to disk
        // before initiating transactions again.
        blockFaultLogWriteStatus(written);

        return replicasAdded;
    }

    /**
     * Poll the replay sequencer and process the messages until it returns null
     */
    private void deliverReadyTxns() {
        // First, pull all the sequenced messages, if any.
        VoltMessage m = m_replaySequencer.poll();
        while(m != null) {
            deliver(m);
            m = m_replaySequencer.poll();
        }
        // Then, try to pull all the drainable messages, if any.
        m = m_replaySequencer.drain();
        while (m != null) {
            if (m instanceof Iv2InitiateTaskMessage) {
                // Send IGNORED response for all SPs
                Iv2InitiateTaskMessage task = (Iv2InitiateTaskMessage) m;
                final InitiateResponseMessage response = new InitiateResponseMessage(task);
                response.setResults(new ClientResponseImpl(ClientResponse.UNEXPECTED_FAILURE,
                            new VoltTable[0],
                            ClientResponseImpl.IGNORED_TRANSACTION));
                m_mailbox.send(response.getInitiatorHSId(), response);
            }
            m = m_replaySequencer.drain();
        }
    }

    /**
     * Sequence the message for replay if it's for CL or DR.
     *
     * @param message
     * @return true if the message can be delivered directly to the scheduler,
     * false if the message is queued
     */
    @Override
    public boolean sequenceForReplay(VoltMessage message)
    {
        boolean canDeliver = false;
        long sequenceWithUniqueId = Long.MIN_VALUE;

        boolean commandLog = (message instanceof TransactionInfoBaseMessage &&
                (((TransactionInfoBaseMessage)message).isForReplay()));

        boolean sentinel = message instanceof MultiPartitionParticipantMessage;

        boolean replay = commandLog || sentinel;
        boolean sequenceForReplay = m_isLeader && replay;

        if (replay) {
            sequenceWithUniqueId = ((TransactionInfoBaseMessage)message).getUniqueId();
        }

        if (sequenceForReplay) {
            InitiateResponseMessage dupe = m_replaySequencer.dedupe(sequenceWithUniqueId,
                    (TransactionInfoBaseMessage) message);
            if (dupe != null) {
                // Duplicate initiate task message, send response
                m_mailbox.send(dupe.getInitiatorHSId(), dupe);
            }
            else if (!m_replaySequencer.offer(sequenceWithUniqueId, (TransactionInfoBaseMessage) message)) {
                canDeliver = true;
            }
            else {
                deliverReadyTxns();
            }

            // If it's a DR sentinel, send an acknowledgement
            if (sentinel && !commandLog) {
                MultiPartitionParticipantMessage mppm = (MultiPartitionParticipantMessage) message;
                final InitiateResponseMessage response = new InitiateResponseMessage(mppm);
                ClientResponseImpl clientResponse =
                        new ClientResponseImpl(ClientResponseImpl.UNEXPECTED_FAILURE,
                                new VoltTable[0], ClientResponseImpl.IGNORED_TRANSACTION);
                response.setResults(clientResponse);
                m_mailbox.send(response.getInitiatorHSId(), response);
            }
        }
        else {
            if (replay) {
                // Update last seen and last polled uniqueId for replicas
                m_replaySequencer.updateLastSeenUniqueId(sequenceWithUniqueId,
                        (TransactionInfoBaseMessage) message);
                m_replaySequencer.updateLastPolledUniqueId(sequenceWithUniqueId,
                        (TransactionInfoBaseMessage) message);
            }

            canDeliver = true;
        }

        return canDeliver;
    }

    // SpInitiators will see every message type.  The Responses currently come
    // from local work, but will come from replicas when replication is
    // implemented
    @Override
    public void deliver(VoltMessage message)
    {
        if (message instanceof Iv2InitiateTaskMessage) {
            handleIv2InitiateTaskMessage((Iv2InitiateTaskMessage)message);
        }
        else if (message instanceof InitiateResponseMessage) {
            handleInitiateResponseMessage((InitiateResponseMessage)message);
        }
        else if (message instanceof FragmentTaskMessage) {
            handleFragmentTaskMessage((FragmentTaskMessage)message);
        }
        else if (message instanceof FragmentResponseMessage) {
            handleFragmentResponseMessage((FragmentResponseMessage)message);
        }
        else if (message instanceof CompleteTransactionMessage) {
            handleCompleteTransactionMessage((CompleteTransactionMessage)message);
        }
        else if (message instanceof CompleteTransactionResponseMessage) {
            handleCompleteTransactionResponseMessage((CompleteTransactionResponseMessage) message);
        }
        else if (message instanceof BorrowTaskMessage) {
            handleBorrowTaskMessage((BorrowTaskMessage)message);
        }
        else if (message instanceof Iv2LogFaultMessage) {
            handleIv2LogFaultMessage((Iv2LogFaultMessage)message);
        }
        else if (message instanceof DumpMessage) {
            handleDumpMessage((DumpMessage)message);
        } else if (message instanceof DumpPlanThenExitMessage) {
            handleDumpPlanMessage((DumpPlanThenExitMessage)message);
        }
        else if (message instanceof DummyTransactionTaskMessage) {
            handleDummyTransactionTaskMessage((DummyTransactionTaskMessage) message);
        }
        else if (message instanceof DummyTransactionResponseMessage) {
            handleDummyTransactionResponseMessage((DummyTransactionResponseMessage)message);
        }
        else if (message instanceof MPBacklogFlushMessage) {
            cleanupTransactionBacklogOnRepair();
        }
        else {
            throw new RuntimeException("UNKNOWN MESSAGE TYPE, BOOM!");
        }
    }

    // SpScheduler expects to see InitiateTaskMessages corresponding to single-partition
    // procedures only.
    private void handleIv2InitiateTaskMessage(Iv2InitiateTaskMessage message)
    {
        if (!message.isSinglePartition()) {
            VoltDB.crashLocalVoltDB("SpScheduler.handleIv2InitiateTaskMessage " +
                    "should never receive multi-partition initiations. Invocation: " + message, true, null);
        }
        final String procedureName = message.getStoredProcedureName();
        long newSpHandle;
        long uniqueId = Long.MIN_VALUE;
        Iv2InitiateTaskMessage msg = message;
        if (m_isLeader || message.isReadOnly()) {

            /*
             * If this is for CL replay or DR, update the unique ID generator
             */
            if (message.isForReplay()) {
                uniqueId = message.getUniqueId();
                try {
                    m_uniqueIdGenerator.updateMostRecentlyGeneratedUniqueId(uniqueId);
                }
                catch (Exception e) {
                    hostLog.fatal(e.getMessage());
                    hostLog.fatal("Invocation: " + message);
                    VoltDB.crashLocalVoltDB(e.getMessage(), true, e);
                }
{'fraction_non_alphanumeric': 0.07376536312849162, 'fraction_numerical': 0.002078212290502793, 'mean_word_length': 2.571793439221007, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2264009', 'n_tokens_mistral': 12149, 'n_tokens_neox': 11729, 'n_words': 3018}