//
// Name
//  $RCSfile: Partition.java,v $
// 
// Copyright
//      Copyright 2010-2011 Cloud Software Group, Inc. ALL RIGHTS RESERVED. 
//      Cloud Software Group, Inc. Confidential Information
//
// History
//  $Revision: 1.1.2.18 $ $Date: 2011/11/23 03:33:29 $
//
package com.kabira.platform.highavailability;

import com.kabira.platform.disteng.DEReplicationType;
import com.kabira.platform.disteng.DEProperties;
import java.util.Date;

/**
 * The Partition class. This non-managed class contains a snapshot of a
 * shared memory partition. Direct access to the Partition instance in
 * shared memory is not allowed to avoid transaction locking issues.
 * <p>
 * Note that partitions are only visible on their active and replica
 * nodes, not all nodes in the cluster.
 */
public final class Partition
{
    /**
     * The Partition states
     */
    public enum State
    {
        /** The partition has been defined, but not enabled */
        INITIAL,
        /** The partition is active */
        ACTIVE,
        /**
         * The partition definition is being updated. Entered when 
         * partition membership (i.e. the objects associated with the
         * partition) is being updated to split or merge partitions.
         */
        UPDATING,
        /**
         * The partition owner is being updated. Each object in the
         * partition is sent to the new owner and the location definition
         * of all instances are updated.
         */
        MIGRATING,
        /**
         * The partition replicas are being updated. Each object in the
         * partition is sent to the new replica list.
         */
        REPLICATING,
        /** 
         * The partition is not active because all associated nodes have
         * failed, the node was taken off-line because it is in a
         * non-quorum state, or the node was restarted.
         */
        UNAVAILABLE
    }

    /**
     * The Partition properties used when defining a partition
     */
    public final static class Properties
    {
        /**
         * Default constructor.
         */
        public Properties()
        {
            m_restoreFromNode = null;
            m_forceReplication = false;
            m_objectsLockedPerTransaction = DefaultObjectsLockedPerTransaction;
        }

        /**
         * Define the node that the partition should be restored from.
         * <p>
         * When this property is set, the partition defined on the given
         * remote node is loaded to the local node. This should be done
         * when restoring a node from a split-brain situation, where
         * <b>nodeName</b> is the node in the cluster where all objects
         * should be preserved, and the local node is the node being
         * restored. Any conflicts during restore will preserve the
         * objects on <b>nodeName</b>, and remove the conflicting objects
         * on the local node.
         * <p>
         * A restore is needed when multiple nodes are currently the active
         * node for a partition in a cluster due to a split-brain scenario.
         * In this case, the application needs to decide which active node
         * will be the node where the objects are preserved during a
         * restore. Note that the <b>nodeName</b> does not necessarily have
         * to be the node which becomes the partition's active node after the
         * restore completes.
         * <p>
         * The actual restore of the partition is done in the
         * enablePartitions() method when the JOIN_CLUSTER_RESTORE
         * EnableAction is used. If any other EnableAction is used,
         * object data isn't preserved, and no restoration of partition
         * objects is done.
         * <p>
         * If restoreFromNode isn't set after a split-brain scenario, the
         * runtime will perform a cluster wide broadcast to find the
         * current active node, and use that node to restore instances in
         * the partition. If multiple active nodes are found, the first
         * responder is chosen.
         * <p>
         * @param nodeName The remote node to use when restoring the
         *  partition's objects.
         *
         * @exception IllegalArgumentException
         * The nodeName was empty.
         * @see PartitionManager#definePartition(String,
         *              Partition.Properties, String, ReplicaNode [])
         * @see PartitionManager.EnableAction
         */
        public void restoreFromNode(String nodeName)
            throws IllegalArgumentException
        {
            if (nodeName.isEmpty())
            {
                throw new IllegalArgumentException(
                    "nodeName cannot be empty");
            }
            m_restoreFromNode = nodeName;
        }

        /**
         * Get the current restoreFromNode property value.
         * @return String containing value.
         */
        public final String getRestoreFromNode()
        {
            return m_restoreFromNode;
        }

        /**
         * Determine how objects are replicated during a migrate of
         * an object partition.
         * <p>
         * When set to true, a migrate() or
         * definePartition()/enablePartition() will force the copy of
         * partitioned objects to all pre-existing replica nodes. The
         * default value for this property is false, objects are only
         * copied to new replicas as they are added since the objects
         * should already exist on the pre-existing replica nodes.
         * <p>
         * Normally, a migrate will skip the replication of objects to
         * pre-existing nodes in the partition's replica node list. This
         * allows applications to incrementally add replica nodes without
         * having to copy the objects to replicas that already exist in
         * the partition. However, if one or more replicas have gone
         * offline, or were not discovered when the partition was first
         * enabled, this property can be set to insure that objects are
         * pushed to all replicas in the cluster.
         * <p>
         * Warning: This is performance hostile, and should only be done
         * if the replica can't be manually taken offline and restored.
         * <p>
         * The value passed into definePartition() is stored and used in
         * failover. The value passed to migrate() overrides
         * the value passed to definePartition().
         *
         * @param enabled If true, force the copy of objects to all replicas
         *  when a migrate() or enablePartition() is executed.
         *
         */
        public void forceReplication(boolean enabled)
        {
            m_forceReplication = enabled;
        }

        /**
         * Get the current forceReplication property value.
         * @return boolean containing value.
         */
        public final boolean getForceReplication()
        {
            return m_forceReplication;
        }

        /**
         * Define the number of objects locked in a transaction when performing
         * a migrate() or update().
         * <p>
         * When distribution performs a migrate() or update(), or when it
         * performs a migrate() as part of failover, the work is done in
         * units of work defined by objectsLockedPerTransaction. This
         * allows applications to concurrently run while the work is in
         * progress, otherwise all partitioned instances would be locked
         * in the calling transaction, preventing application code from
         * establishing a write lock on instances on the active node, or
         * establishing a read lock on instances on replica nodes.
         * <p>
         * The default is defined by {@link #DefaultObjectsLockedPerTransaction}
         * <p>
         * If objectsLockedPerTransaction is set to zero, all instances will
         * be processed in the caller's transaction.
         * <p>
         * If the calling transaction has one or more partitioned
         * instances locked in the transaction,
         * objectsLockedPerTransaction is ignored, and all work is done
         * in the caller's transaction. To insure that migrate() or update()
         * minimizes the number of locks taken, they should be run in separate
         * transactions that have no partitioned instances locked. 
         * <p>
         * The value passed into definePartition() is stored and used in
         * failover. The value passed to migrate() and update() override
         * the value passed to definePartition().
         * 
         * @param objectsLockedPerTransaction Number of objects locked
         *  per transaction when sending data to remote nodes.
         * @exception IllegalArgumentException The objectsLockedPerTransaction
         *  value was negative.
         */
        public void setObjectsLockedPerTransaction(
            long objectsLockedPerTransaction)
        {
            if (objectsLockedPerTransaction < 0)
            {
                throw new IllegalArgumentException(
                    "invalid objectsLockedPerTransaction: " +
                    objectsLockedPerTransaction);
            }
            m_objectsLockedPerTransaction = objectsLockedPerTransaction;
        }

        /**
         * Get the current objectsLockedPerTransaction property value.
         * @return long containing value.
         */
        public final long getObjectsLockedPerTransaction()
        {
            return m_objectsLockedPerTransaction;
        }

        //
        // package private constructor
        //
        Properties(boolean forceReplication, long objectChunks)
        {
            m_restoreFromNode = null;
            m_forceReplication = forceReplication;
            m_objectsLockedPerTransaction = objectChunks;
        }

        String m_restoreFromNode;
        boolean m_forceReplication;
        long m_objectsLockedPerTransaction;
    }

    /** Index to the active node in the nodes array */
    public final static int     ActiveNodeIndex = 0;

    /**
     * The default number of objects locked per transaction when
     * performing a migrate or update.
     */
    public final static long    DefaultObjectsLockedPerTransaction = 1000;

    /**
     * Get the partition name.
     * @return Partition name.
     */
    public final String getName()
    {
        return m_name;
    }

    /**
     * Get the active node for the partition.
     * @return Active node name.
     */
    public final String getActiveNode()
    {
        assert( m_nodes.length >= 1 );
        return m_nodes[ActiveNodeIndex];
    }

    /**
     * Get the replica node list for the partition.
     * @return Array of replica node names.
     */
    public final String [] getReplicaNodes()
    {
        assert( m_nodes.length >= 1 );
        String [] replicas = new String[m_nodes.length - 1];
        for (int i = 0; i < m_nodes.length - 1; i++)
        {
            replicas[i] = m_nodes[i + 1];
        }
        return replicas;
    }

    /**
     * Get the complete node list for the partition.
     * <p>
     * The node list is a String array of node names, with the active
     * node being at ActiveNodeIndex, followed by a prioritized list of
     * replica nodes.
     * @return Array of node names.
     */
    public final String [] getNodeList()
    {
        return m_nodes;
    }

    /**
     * Get the current state for the partition.
     * @return Current state of partition.
     */
    public final State getCurrentState()
    {
        return m_currentState;
    }

    /**
     * Get the ReplicaNode instance for the given replica node index.
     * @param idx Index into the ReplicaNode array.
     * @return ReplicaNode instance.
     * @exception IndexOutOfBoundsException
     *  The idx isn't valid.
     */
    public final ReplicaNode getReplicaNode(int idx)
        throws IndexOutOfBoundsException
    {
        // m_nodes[0] is active node, so we need to disallow -1
        if (idx < 0) throw new IndexOutOfBoundsException(Integer.toString(idx));
        return new ReplicaNode(m_nodes[idx + 1],
            PartitionManager.mapDEReplicationType(
                m_replicationTypes[idx + 1]));
    }

    /**
     * Get the ReplicaNode instance for the given replica node name.
     * @param nodeName Name of replica node.
     * @return ReplicaNode instance, or null if not found.
     * @exception IllegalArgumentException The nodeName was invalid.
     */
    public final ReplicaNode getReplicaNode(final String nodeName)
        throws IllegalArgumentException
    {
        if (nodeName.isEmpty())
        {
            throw new IllegalArgumentException("nodeName cannot be empty");
        }
        for (int idx = 1; idx < m_nodes.length; idx++)
        {
            if (m_nodes[idx].equals(nodeName))
            {
                return getReplicaNode(idx - 1);
            }
        }
        return null;
    }

    /**
     * Get the last time the state for the partition was updated
     * @return Date of last state change for partition.
     */
    public final Date getLastStateChangeTime()
    {
        return m_lastUpdated;
    }

    /**
     * Get the properties currently defined for the partition.
     * <p>
     * The restoreFromNode property is not stored in the runtime since it
     * has no meaning outside of the initial definePartition(). So this method
     * will return a Properties instance with a null restoreFromNode value.
     *
     * @return Properties instance for partition.
     */
    public final Properties getProperties()
    {
        return new Properties(m_forceReplication, m_objectChunks);
    }

    /**
     * Use {@link #update(Properties)} instead.
     */
    @Deprecated
    public final void update() throws NodeMismatch, NotActiveNode
    {
        update(null);
    }

    /**
     * Update the partition.
     * <p>
     * This method is used to re-partition all instances that exist in a
     * partition. For each instance in the partition, the {@link
     * PartitionMapper} defined for that type is accessed, and the
     * instance re-assigned to the partition that the {@link
     * PartitionMapper#getPartition} method returns. The update() method
     * must be called on the current active node for the partition.
     * <p>
     * To split a partition, the applications should create the new
     * partition, install a new PartitionMapper for all types managed by
     * the partition, and call the update() method for the existing
     * partition.
     * <p>
     * To merge two or more partitions, the applications should install a
     * new PartitionMapper for all types managed by the partition(s), and
     * call the update() method for all the partitions that need to
     * be merged.
     * <p>
     * It is important that all partitions returned when executing the
     * PartitionMapper's getPartition() call for a type all contain
     * identical node lists. If the partition returned contains a
     * different node list than this partition, a NodeMismatch exception
     * is thrown, and the update() terminates. To fix this, perform a
     * migrate() on all partitions that will be split or merged to insure
     * that they have identical node lists before performing an update().
     *
     * @param partitionProperties Optional properties for the partition.
     *  If null, the default property values are used.
     * 
     * @exception NodeMismatch
     *  The re-partitioning of instances was done using partitions that
     *  have different node lists.
     * @exception NotActiveNode
     *  The current node is not the active node for the partition.
     */
    public final void update(final Properties partitionProperties)
            throws NodeMismatch, NotActiveNode
    {
        DEProperties props = new DEProperties();

        if (partitionProperties == null)
        {
            props.forceReplication = false;
            props.objectChunks = Partition.DefaultObjectsLockedPerTransaction;
        }
        else
        {
            props.forceReplication = partitionProperties.m_forceReplication;
            props.objectChunks =
                partitionProperties.m_objectsLockedPerTransaction;
        }

        PartitionManager.updatePartition(m_name, props);
    }

    /**
     * Use {@link #migrate(Properties, String, ReplicaNode [])} instead.
     */
    @Deprecated
    public final void migrate(String [] nodes)
        throws NotActiveNode, IllegalArgumentException
    {
        PartitionManager.validateNodeList(nodes);
        DEReplicationType [] drl = new DEReplicationType[nodes.length];
        for (int i = 0; i < drl.length; i++)
        {
            drl[i] = DEReplicationType.DataSynchronous;
        }

        DEProperties props = new DEProperties();
        props.forceReplication = false;
        props.objectChunks = Partition.DefaultObjectsLockedPerTransaction;

        PartitionManager.migratePartition(m_name, props, nodes, drl);
    }

    /**
     * Migrate the partition.
     * <p>
     * This method is used to migrate all instances that exist in a
     * partition. For each instance in the partition, the object is
     * migrated as needed to all nodes in the replicas array, and to the
     * new activeNode if it is different from the current active node. This
     * method must be called on the currently active node for the partition.
     * <p>
     * Any properties passed in only apply to the current migrate() command,
     * properties passed into definePartition() are saved and used during
     * failover. 
     * 
     * @param partitionProperties Optional properties for the partition.
     *  If null, the default property values are used.
     * @param activeNode Active node after migrate completes.
     * @param replicas An ordered list of replica nodes for the partition.
     *  Should be passed in as a null instance or a zero length array if
     *  no replicas exists.
     *
     * @exception NotActiveNode
     *  The current node is not the active node for the partition.
     * @exception IllegalArgumentException
     * The activeNode or replica array was invalid.
     */
    public final void migrate(
        final Properties partitionProperties,
        String activeNode,
        ReplicaNode [] replicas)
            throws NotActiveNode, IllegalArgumentException
    {
        PartitionManager.validateReplicaList(replicas);

        int nodeLength = (replicas == null) ? 1 : replicas.length + 1;
        String [] nodes = new String[nodeLength];

        nodes[0] = activeNode;
        if (replicas != null )
        {
            for (int i = 0; i < replicas.length; i++)
            {
                assert( i + 1 < nodes.length );
                nodes[i + 1] = replicas[i].nodeName;
            }
        }
        PartitionManager.validateNodeList(nodes);

        DEReplicationType [] drl = new DEReplicationType[nodes.length];
        drl[0] = DEReplicationType.DataSynchronous;
        if (replicas != null)
        {
            for (int i = 0; i < replicas.length; i++)
            {
                assert( i + 1 < nodes.length );
                drl[i + 1] = PartitionManager.mapReplicationType(
                    replicas[i].replicationType);
            }
        }
        boolean forceReplication = (partitionProperties != null)
            ? partitionProperties.m_forceReplication : false;

        DEProperties props = new DEProperties();
        if (partitionProperties == null)
        {
            props.forceReplication = false;
            props.objectChunks = Partition.DefaultObjectsLockedPerTransaction;
        }
        else
        {
            props.forceReplication = partitionProperties.m_forceReplication;
            props.objectChunks =
                partitionProperties.m_objectsLockedPerTransaction;
        }
        PartitionManager.migratePartition(m_name, props, nodes, drl);
    }

    /**
     * Associated a notifier with the partition.
     * <p>
     * This method is used to associate a user defined {@link PartitionNotifier}
     * instance with a partition.
     * <p>
     * PartitionNotifier instances are local to a node, this method
     * should be executed on all nodes which need to determine if a
     * partition state change occurs.
     * <p>
     * The same PartitionNotifier instance can be associated with multiple
     * partitions. Multiple PartitionNotifier instances can be installed for
     * a given Partition. When a state change occurs, all instances are
     * executed, no order is guaranteed when executing the notifiers.
     * To remove a notifier, the PartitionNotifier instance should be deleted.
     * 
     * @param partitionNotifier User defined notifier instance.
     *
     */
    public final void setNotifier(PartitionNotifier partitionNotifier)
    {
        PartitionManager.setNotifier(m_name, partitionNotifier);
    }

    /** The name of the partition */
    private final String         m_name;

    /** The list of nodes for a partition */
    private final String []      m_nodes;

    /** The list of nodes for a partition */
    private final DEReplicationType [] m_replicationTypes;

    /** The current state of the partition */
    private final State          m_currentState;

    /** The time it was updated */
    private final Date          m_lastUpdated;

    /** stored property */
    private final boolean       m_forceReplication;

    /** stored property */
    private final long          m_objectChunks;

    //
    // FIX THIS: Add stats attributes...
    //

    //
    // Package private constructor
    //
    Partition(
        String name,
        String [] nodes,
        DEReplicationType [] replicationTypes,
        State currentState,
        Date lastUpdated,
        boolean forceReplication,
        long objectChunks)
    {
        this.m_name = name;
        this.m_nodes = nodes;
        this.m_replicationTypes = replicationTypes;
        this.m_currentState = currentState;
        this.m_lastUpdated = lastUpdated;
        this.m_forceReplication = forceReplication;
        this.m_objectChunks = objectChunks;
    }
    //
    // Never should be called
    //
    private Partition()
    {
        this.m_name = null;
        this.m_nodes = null;
        this.m_replicationTypes = null;
        this.m_currentState = null;
        this.m_lastUpdated = null;
        this.m_forceReplication = false;
        this.m_objectChunks = 0;
    }
}