/* This file is part of VoltDB.
 * Copyright (C) 2008-2012 VoltDB Inc.
 *
 * VoltDB is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * VoltDB 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */

package org.voltdb.rejoin;

import java.io.File;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import org.voltcore.logging.VoltLogger;
import org.voltcore.messaging.HostMessenger;
import org.voltcore.messaging.VoltMessage;
import org.voltcore.utils.CoreUtils;
import org.voltdb.VoltDB;
import org.voltdb.messaging.RejoinMessage;
import org.voltdb.messaging.RejoinMessage.Type;
import org.voltdb.utils.VoltFile;

/**
 * Sequentially rejoins each site. This rejoin coordinator is accessed by sites
 * sequentially, so no need to synchronize.
 */
public class SequentialRejoinCoordinator extends RejoinCoordinator {
    private static final VoltLogger rejoinLog = new VoltLogger("REJOIN");

    // contains all sites that haven't started streaming snapshot
    private final Queue<Long> m_pendingSites;
    // contains all sites that haven't finished replaying transactions
    private final Queue<Long> m_rejoiningSites = new LinkedList<Long>();

    public SequentialRejoinCoordinator(HostMessenger messenger,
                                       List<Long> sites,
                                       String voltroot) {
        super(messenger);
        m_pendingSites = new LinkedList<Long>(sites);
        if (m_pendingSites.isEmpty()) {
            VoltDB.crashLocalVoltDB("No execution sites to rejoin", false, null);
        }

        // clear overflow dir in case there are files left from previous runs
        try {
            File overflowDir = new File(voltroot, "rejoin_overflow");
            if (overflowDir.exists()) {
                VoltFile.recursivelyDelete(overflowDir);
            }
        } catch (Exception e) {
            VoltDB.crashLocalVoltDB("Fail to clear rejoin overflow directory", false, e);
        }
    }

    /**
     * Send rejoin initiation message to the local site
     * @param HSId
     */
    private void initiateRejoinOnSite(long HSId) {
        RejoinMessage msg = new RejoinMessage(getHSId(),
                                              RejoinMessage.Type.INITIATION);
        send(HSId, msg);
    }

    @Override
    public void startRejoin() {
        long firstSite = m_pendingSites.poll();
        m_rejoiningSites.add(firstSite);
        String HSIdString = CoreUtils.hsIdToString(firstSite);
        rejoinLog.debug("Start rejoining site " + HSIdString);
        initiateRejoinOnSite(firstSite);
    }

    private void onSnapshotStreamFinished(long HSId) {
        rejoinLog.info("Finished streaming snapshot to site " +
                CoreUtils.hsIdToString(HSId));

        if (!m_pendingSites.isEmpty()) {
            long nextSite = m_pendingSites.poll();
            m_rejoiningSites.add(nextSite);
            String HSIdString = CoreUtils.hsIdToString(nextSite);
            rejoinLog.debug("Start rejoining site " + HSIdString);
            initiateRejoinOnSite(nextSite);
        }
    }

    private void onReplayFinished(long HSId) {
        if (!m_rejoiningSites.remove(HSId)) {
            VoltDB.crashLocalVoltDB("Unknown site " + CoreUtils.hsIdToString(HSId) +
                                    " finished rejoin", false, null);
        }
        rejoinLog.info("Finished rejoining site " + CoreUtils.hsIdToString(HSId));

        if (m_rejoiningSites.isEmpty()) {
            // no more sites to rejoin, we're done
            VoltDB.instance().onExecutionSiteRejoinCompletion(0l);
        }
    }

    @Override
    public void deliver(VoltMessage message) {
        if (!(message instanceof RejoinMessage)) {
            VoltDB.crashLocalVoltDB("Unknown message type " +
                    message.getClass().toString() + " sent to the rejoin coordinator",
                    false, null);
        }

        RejoinMessage rm = (RejoinMessage) message;
        Type type = rm.getType();
        if (type == RejoinMessage.Type.SNAPSHOT_FINISHED) {
            onSnapshotStreamFinished(rm.m_sourceHSId);
        } else if (type == RejoinMessage.Type.REPLAY_FINISHED) {
            onReplayFinished(rm.m_sourceHSId);
        } else {
            VoltDB.crashLocalVoltDB("Wrong rejoin message of type " + type +
                                    " sent to the rejoin coordinator", false, null);
        }
    }
}

@@ -749,16 +749,44 @@
 o.File;%0A
+import java.util.ArrayList;%0A
 import j
@@ -3892,80 +3892,484 @@
     
-rejoinLog.info(%22Finished rejoining site %22 + CoreUtils.hsIdToString(HSId)
+String msg = %22Finished rejoining site %22 + CoreUtils.hsIdToString(HSId);%0A        ArrayList%3CLong%3E remainingSites = new ArrayList%3CLong%3E(m_pendingSites);%0A        remainingSites.addAll(m_rejoiningSites);%0A        if (!remainingSites.isEmpty()) %7B%0A            msg += %22. Remaining sites to rejoin: %22 +%0A                    CoreUtils.hsIdCollectionToString(remainingSites);%0A        %7D%0A        else %7B%0A            msg += %22. All sites completed rejoin.%22;%0A        %7D%0A        rejoinLog.info(msg
 );%0A%0A

Include remaining sites to rejoin in the rejoin coordinator log.