/*-
 *
 *  This file is part of Oracle NoSQL Database
 *  Copyright (C) 2011, 2014 Oracle and/or its affiliates.  All rights reserved.
 *
 *  Oracle NoSQL Database 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, version 3.
 *
 *  Oracle NoSQL Database 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 in the LICENSE file along with Oracle NoSQL Database.  If not,
 *  see <http://www.gnu.org/licenses/>.
 *
 *  An active Oracle commercial licensing agreement for this product
 *  supercedes this license.
 *
 *  For more information please contact:
 *
 *  Vice President Legal, Development
 *  Oracle America, Inc.
 *  5OP-10
 *  500 Oracle Parkway
 *  Redwood Shores, CA 94065
 *
 *  or
 *
 *  berkeleydb-info_us@oracle.com
 *
 *  [This line intentionally left blank.]
 *  [This line intentionally left blank.]
 *  [This line intentionally left blank.]
 *  [This line intentionally left blank.]
 *  [This line intentionally left blank.]
 *  [This line intentionally left blank.]
 *  EOF
 *
 */

package oracle.kv.impl.rep;

import java.util.EnumMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.logging.Level;
import java.util.logging.Logger;

import oracle.kv.impl.api.RequestDispatcher;
import oracle.kv.impl.api.rgstate.RepGroupState;
import oracle.kv.impl.api.rgstate.RepNodeState;
import oracle.kv.impl.api.rgstate.UpdateThread;
import oracle.kv.impl.metadata.Metadata;
import oracle.kv.impl.metadata.Metadata.MetadataType;
import oracle.kv.impl.metadata.MetadataInfo;
import oracle.kv.impl.rep.admin.RepNodeAdminAPI;
import oracle.kv.impl.topo.RepGroupId;
import oracle.kv.impl.topo.RepNodeId;
import oracle.kv.impl.topo.Topology;
import oracle.kv.impl.util.registry.RegistryUtils;

import com.sleepycat.je.rep.ReplicatedEnvironment;

/**
 * Thread to update the RNs metadata.
 */
public class MetadataUpdateThread extends UpdateThread {

    private final RepNode repNode;

    /**
     * The minimum number of threads used by the pool. This minimum number
     * permits progress in the presence of a few bad network connections.
     */
    private static final int MIN_POOL_SIZE = 6;

    /**
     *  The time interval between executions of a concurrent update pass.
     */
    private static final int UPDATE_THREAD_PERIOD_MS = 2000;

    /**
     * Used to ensure that there is only one pull request running at one time.
     */
    private final Map<MetadataType, Semaphore> pullsInProgress =
            new EnumMap<MetadataType, Semaphore>(MetadataType.class);

    /**
     * Next group to update. This starts at our own group + 1 and loops through
     * all of the groups.
     */
    private int nextGroup;

    /**
     * Creates the RN state update thread.
     *
     * @param requestDispatcher the request dispatcher associated with this
     * thread
     *
     * @param repNode the rep node
     *
     * @param logger the logger used by the update threads
     */
    public MetadataUpdateThread(RequestDispatcher requestDispatcher,
                                RepNode repNode,
                                final Logger logger) {
        super(requestDispatcher, UPDATE_THREAD_PERIOD_MS,
              requestDispatcher.getExceptionHandler(), logger);
        this.repNode = repNode;
        nextGroup = repNode.getRepNodeId().getGroupId() + 1;
        for (final MetadataType type : MetadataType.values()) {
            pullsInProgress.put(type, new Semaphore(1));
        }        
    }

    /**
     * Updates metadata for one group.
     */
    @Override
    protected void doUpdate() {
        /* Select a group to query */
        final Topology topo = repNode.getTopology();

        if (topo == null) {
            return; /* Not yet initialized */
        }
        final Set<RepGroupId> groups = topo.getRepGroupIds();
        
        /*
         * Nothing to update if only one group. We don't shut down the update
         * thread because the store may expand.
         */
        if (groups.size() < 2) {
            return;
        }

        /*
         * Loop back to beginning.
         * TODO - This assumes that group IDs are contiguous, starting at 1.
         * When store contraction is implemented and group IDs can contain
         * holes the code will need to be modified.
         */
        if (nextGroup > groups.size()) {
            nextGroup = 1;
        }
        
        /* Skip our own group */
        if (nextGroup == repNode.getRepNodeId().getGroupId()) {
            nextGroup++;
            return;
        }
        
        final RepGroupState rgState =
                                getRepGroupState(new RepGroupId(nextGroup));

        threadPool.setMaximumPoolSize(
                 Math.max(MIN_POOL_SIZE, rgState.getRepNodeStates().size()/10));

        /* Check all types of metadata except Topology */
        for (final MetadataType checkType : MetadataType.values()) {
            if (shutdown.get()) {
                return;
            }
            if (checkType == MetadataType.TOPOLOGY) {
                continue;
            }
            try {
                checkMetadata(checkType, rgState);
            } catch (Exception e) {

                /*
                 * If shutdown, simply return, even if there was an
                 * exception
                 */
                if (shutdown.get()) {
                    return;
                }

                /* Log and go on with check and update other metadata types. */
                logOnFailure(repNode.getRepNodeId(), e, 
                             "Exception attempting to update " + checkType +
                             " metadata");
            }
        }
        nextGroup++;
    }

    /**
     * Checks if the specified metadata type for the group represented by
     * rgState needs updating. This may result in this node being updated.
     * 
     * @param type the metadata type to check
     * @param rgState the group state
     */
    private void checkMetadata(MetadataType type, RepGroupState rgState) {
        final int localSeqNum = repNode.getMetadataSeqNum(type);
        final int rgSeqNum = rgState.getSeqNum(type);
        
        /*
         * Check if we need an update. Note that except for the very first
         * check when reqSeqNum will be initialized to UNKNOWN_SEQ_NUM, we
         * will not know to pull unless our metadata is updated. This means
         * the system can't rely on pull for propagation.
         */
        assert Metadata.UNKNOWN_SEQ_NUM < Metadata.EMPTY_SEQUENCE_NUMBER;
        if (rgSeqNum >= localSeqNum) {
            return;
        }
        
        /*  Check with each node in the group */
        for (RepNodeState rnState : rgState.getRepNodeStates()) {
            if (shutdown.get()) {
                return;
            }
            if (needsResolution(rnState)) {
                continue;
            }
            threadPool.execute(new UpdateMetadata(rgState, rnState, type));
        }
    }

    /**
     * Task to update metadata to or from a RN.
     */
    private class UpdateMetadata implements Runnable {
        private final RepGroupState rgState;
        private final RepNodeState rnState;

        /* The metadata type to be pulled. */
        private final MetadataType type;

        public UpdateMetadata(RepGroupState rgState,
                              RepNodeState rnState,
                              MetadataType type) {
            this.rgState = rgState;
            this.rnState = rnState;
            this.type = type;
        }

        @Override
        public void run() {
            logger.log(Level.FINE, "Starting {0}", this);
            final RepNodeId targetRNId = rnState.getRepNodeId();
            try {
                final RegistryUtils regUtils = requestDispatcher.getRegUtils();
                if (regUtils == null) {
                    /*
                     * The request dispatcher has not initialized itself as
                     * yet. Retry later.
                     */
                    return;
                }

                final RepNodeAdminAPI rnAdmin =
                                        regUtils.getRepNodeAdmin(targetRNId);
                
                /* Start by getting the latest seq # from the RN */
                final int rgSeqNum = rgState.updateSeqNum(type,
                                            rnAdmin.getMetadataSeqNum(type));
                final int localSeqNum = repNode.getMetadataSeqNum(type);
                
                /* If we have newer MD, push, if behind, pull */
                if (localSeqNum > rgSeqNum) {
                    push(rgSeqNum, regUtils);
                } else if (localSeqNum < rgSeqNum) {
                    pull(localSeqNum, regUtils);
                }
            } catch (Exception e) {
                logOnFailure(targetRNId,  e,
                             "Exception updating " + targetRNId);
            }
        }
        
        /**
         * Push metadata changes to a RN. Incremental changes are
         * sent if available, otherwise, the entire metadata is pushed.
         */
        private void push(int rgSeqNum, RegistryUtils regUtils) {
            final RepNodeId targetRNId = rnState.getRepNodeId();
            try {
                final RepNodeAdminAPI rnAdmin =
                                        regUtils.getRepNodeAdmin(targetRNId);
                final Metadata<?> md = repNode.getMetadata(type);
                
                /* Attempt to update the node with deltas */
                final MetadataInfo info = md.getChangeInfo(rgSeqNum);
                if (!info.isEmpty()) {
                    int updatedSeqNum = rnAdmin.updateMetadata(info);
                    rgState.updateSeqNum(type, updatedSeqNum);

                    if (updatedSeqNum >= rgSeqNum) {
                        logger.log(Level.FINE,
                                   "Pushed {0} metadata changes [{1}] to {2}",
                                   new Object[]{type, info, targetRNId});
                        return;
                    }
                }
                
                /* Cannot generate a delta so send entire metadata */
                rnAdmin.updateMetadata(md);
                rgState.updateSeqNum(type, md.getSequenceNumber());
                logger.log(Level.INFO, "Pushed {0} to {1}", // FINE
                           new Object[]{md, targetRNId});

            } catch (Exception e) {
                logOnFailure(targetRNId,  e,
                             "Failed pushing " +  type + " metadata to " +
                             targetRNId + ", target metadata seq number:" +
                             rgSeqNum);
            }
        }
        
        /**
         * Pull metadata from the node.
         */
        private void pull(int localSeqNum, RegistryUtils regUtils) {  
            final ReplicatedEnvironment repEnv = repNode.getEnv(1);
            if ((repEnv == null) || !repEnv.getState().isMaster()) {
                return;
            }
            
            if (!pullsInProgress.get(type).tryAcquire()) {
                /* A pull is already in progress. */
                return;
            }

            final RepNodeId targetRNId = rnState.getRepNodeId();
            try {
                final RepNodeAdminAPI rnAdmin =
                                    regUtils.getRepNodeAdmin(targetRNId);
                
                /* Attempt to update from deltas */
                final MetadataInfo info =
                                        rnAdmin.getMetadata(type, localSeqNum);
                if (!info.isEmpty()) {
                    rgState.updateSeqNum(type, info.getSourceSeqNum());
                    repNode.updateMetadata(info);
                    logger.log(Level.FINE,
                               "Pulled {0} metadata changes from {1}, " +
                               "updated to: {2}",
                               new Object[]{type, targetRNId,
                                            repNode.getMetadata(type)});
                    return;
                }
                
                /* Deltas are not available, try getting the entire metadata */
                final Metadata<?> md = rnAdmin.getMetadata(type);
                if (md != null) {
                    rgState.updateSeqNum(type, md.getSequenceNumber());
                    if (repNode.updateMetadata(md)) {
                        logger.log(Level.FINE,
                                   "Pulled {0} from {1}",
                                   new Object[]{md, targetRNId});
                    } else {
                        /*
                         * The update could have failed for numerous,
                         * non-serious reasons, so simply log.
                         */
                        logger.log(Level.FINE,
                                   "Unable to update {0} pulled from {1}",
                                    new Object[]{md, targetRNId});
                    }
                }
            } catch (Exception e) {
                logOnFailure(targetRNId, e,
                             "Failed pulling " + type + " metadata from " +
                             targetRNId);
            } finally {
                pullsInProgress.get(type).release();
            }
        }
        
        @Override
        public String toString() {
            return "UpdateThread[" + type + ", " + rnState.getRepNodeId() + "]";
        }
    }
}
