text:
package core

import (
	ctx "context"
	"fmt"
	"math"
	"reflect"
	"strings"
	"sync"
	"time"

	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
	"k8s.io/autoscaler/cluster-autoscaler/clusterstate"
	"k8s.io/autoscaler/cluster-autoscaler/context"
	core_utils "k8s.io/autoscaler/cluster-autoscaler/core/utils"
	"k8s.io/autoscaler/cluster-autoscaler/metrics"
	"k8s.io/autoscaler/cluster-autoscaler/processors"
	"k8s.io/autoscaler/cluster-autoscaler/processors/customresources"
	"k8s.io/autoscaler/cluster-autoscaler/simulator"
	"k8s.io/autoscaler/cluster-autoscaler/utils"
	"k8s.io/autoscaler/cluster-autoscaler/utils/daemonset"
	"k8s.io/autoscaler/cluster-autoscaler/utils/deletetaint"
	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
	"k8s.io/autoscaler/cluster-autoscaler/utils/kubernetes"
	kube_util "k8s.io/autoscaler/cluster-autoscaler/utils/kubernetes"
	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"

	apiv1 "k8s.io/api/core/v1"
	policyv1 "k8s.io/api/policy/v1beta1"
	kube_errors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/autoscaler/cluster-autoscaler/processors/status"
	"k8s.io/autoscaler/cluster-autoscaler/utils/gpu"
	kube_client "k8s.io/client-go/kubernetes"
	kube_record "k8s.io/client-go/tools/record"
	klog "k8s.io/klog/v2"
)

const (
	// ScaleDownDisabledKey is the name of annotation marking node as not eligible for scale down.
	ScaleDownDisabledKey = "cluster-autoscaler.kubernetes.io/scale-down-disabled"
	// DelayDeletionAnnotationPrefix is the prefix of annotation marking node as it needs to wait
	// for other K8s components before deleting node.
	DelayDeletionAnnotationPrefix = "delay-deletion.cluster-autoscaler.kubernetes.io/"
)

const (
	// MaxKubernetesEmptyNodeDeletionTime is the maximum time needed by Kubernetes to delete an empty node.
	MaxKubernetesEmptyNodeDeletionTime = 3 * time.Minute
	// MaxCloudProviderNodeDeletionTime is the maximum time needed by cloud provider to delete a node.
	MaxCloudProviderNodeDeletionTime = 5 * time.Minute
	// MaxPodEvictionTime is the maximum time CA tries to evict a pod before giving up.
	MaxPodEvictionTime = 2 * time.Minute
	// EvictionRetryTime is the time after CA retries failed pod eviction.
	EvictionRetryTime = 10 * time.Second
	// PodEvictionHeadroom is the extra time we wait to catch situations when the pod is ignoring SIGTERM and
	// is killed with SIGKILL after MaxGracefulTerminationTime
	PodEvictionHeadroom = 30 * time.Second
	// DaemonSetEvictionEmptyNodeTimeout is the time to evict all DaemonSet pods on empty node
	DaemonSetEvictionEmptyNodeTimeout = 10 * time.Second
	// DeamonSetTimeBetweenEvictionRetries is a time between retries to create eviction that uses for DaemonSet eviction for empty nodes
	DeamonSetTimeBetweenEvictionRetries = 3 * time.Second
)

// NodeDeletionTracker keeps track of node deletions.
type NodeDeletionTracker struct {
	sync.Mutex
	nonEmptyNodeDeleteInProgress bool
	// A map of node delete results by node name. It's being constantly emptied into ScaleDownStatus
	// objects in order to notify the ScaleDownStatusProcessor that the node drain has ended or that
	// an error occurred during the deletion process.
	nodeDeleteResults map[string]status.NodeDeleteResult
	// A map which keeps track of deletions in progress for nodepools.
	// Key is a node group id and value is a number of node deletions in progress.
	deletionsInProgress map[string]int
}

// Get current time. Proxy for unit tests.
var now func() time.Time = time.Now

// NewNodeDeletionTracker creates new NodeDeletionTracker.
func NewNodeDeletionTracker() *NodeDeletionTracker {
	return &NodeDeletionTracker{
		nodeDeleteResults:   make(map[string]status.NodeDeleteResult),
		deletionsInProgress: make(map[string]int),
	}
}

// IsNonEmptyNodeDeleteInProgress returns true if a non empty node is being deleted.
func (n *NodeDeletionTracker) IsNonEmptyNodeDeleteInProgress() bool {
	n.Lock()
	defer n.Unlock()
	return n.nonEmptyNodeDeleteInProgress
}

// SetNonEmptyNodeDeleteInProgress sets non empty node deletion in progress status.
func (n *NodeDeletionTracker) SetNonEmptyNodeDeleteInProgress(status bool) {
	n.Lock()
	defer n.Unlock()
	n.nonEmptyNodeDeleteInProgress = status
}

// StartDeletion increments node deletion in progress counter for the given nodegroup.
func (n *NodeDeletionTracker) StartDeletion(nodeGroupId string) {
	n.Lock()
	defer n.Unlock()
	n.deletionsInProgress[nodeGroupId]++
}

// EndDeletion decrements node deletion in progress counter for the given nodegroup.
func (n *NodeDeletionTracker) EndDeletion(nodeGroupId string) {
	n.Lock()
	defer n.Unlock()

	value, found := n.deletionsInProgress[nodeGroupId]
	if !found {
		klog.Errorf("This should never happen, counter for %s in DelayedNodeDeletionStatus wasn't found", nodeGroupId)
		return
	}
	if value <= 0 {
		klog.Errorf("This should never happen, counter for %s in DelayedNodeDeletionStatus isn't greater than 0, counter value is %d", nodeGroupId, value)
	}
	n.deletionsInProgress[nodeGroupId]--
	if n.deletionsInProgress[nodeGroupId] <= 0 {
		delete(n.deletionsInProgress, nodeGroupId)
	}
}

// GetDeletionsInProgress returns the number of deletions in progress for the given node group.
func (n *NodeDeletionTracker) GetDeletionsInProgress(nodeGroupId string) int {
	n.Lock()
	defer n.Unlock()
	return n.deletionsInProgress[nodeGroupId]
}

// AddNodeDeleteResult adds a node delete result to the result map.
func (n *NodeDeletionTracker) AddNodeDeleteResult(nodeName string, result status.NodeDeleteResult) {
	n.Lock()
	defer n.Unlock()
	n.nodeDeleteResults[nodeName] = result
}

// GetAndClearNodeDeleteResults returns the whole result map and replaces it with a new empty one.
func (n *NodeDeletionTracker) GetAndClearNodeDeleteResults() map[string]status.NodeDeleteResult {
	n.Lock()
	defer n.Unlock()
	results := n.nodeDeleteResults
	n.nodeDeleteResults = make(map[string]status.NodeDeleteResult)
	return results
}

type scaleDownResourcesLimits map[string]int64
type scaleDownResourcesDelta map[string]int64

// used as a value in scaleDownResourcesLimits if actual limit could not be obtained due to errors talking to cloud provider
const scaleDownLimitUnknown = math.MinInt64

func (sd *ScaleDown) computeScaleDownResourcesLeftLimits(nodes []*apiv1.Node, resourceLimiter *cloudprovider.ResourceLimiter, cp cloudprovider.CloudProvider, timestamp time.Time) scaleDownResourcesLimits {
	totalCores, totalMem := calculateScaleDownCoresMemoryTotal(nodes, timestamp)

	var totalResources map[string]int64
	var totalResourcesErr error
	if cloudprovider.ContainsCustomResources(resourceLimiter.GetResources()) {
		totalResources, totalResourcesErr = sd.calculateScaleDownCustomResourcesTotal(nodes, cp, timestamp)
	}

	resultScaleDownLimits := make(scaleDownResourcesLimits)
	for _, resource := range resourceLimiter.GetResources() {
		min := resourceLimiter.GetMin(resource)

		// we put only actual limits into final map. No entry means no limit.
		if min > 0 {
			switch {
			case resource == cloudprovider.ResourceNameCores:
				resultScaleDownLimits[resource] = computeAboveMin(totalCores, min)
			case resource == cloudprovider.ResourceNameMemory:
				resultScaleDownLimits[resource] = computeAboveMin(totalMem, min)
			case cloudprovider.IsCustomResource(resource):
				if totalResourcesErr != nil {
					resultScaleDownLimits[resource] = scaleDownLimitUnknown
				} else {
					resultScaleDownLimits[resource] = computeAboveMin(totalResources[resource], min)
				}
			default:
				klog.Errorf("Scale down limits defined for unsupported resource '%s'", resource)
			}
		}
	}
	return resultScaleDownLimits
}

func computeAboveMin(total int64, min int64) int64 {
	if total > min {
		return total - min
	}
	return 0

}

func calculateScaleDownCoresMemoryTotal(nodes []*apiv1.Node, timestamp time.Time) (int64, int64) {
	var coresTotal, memoryTotal int64
	for _, node := range nodes {
		if isNodeBeingDeleted(node, timestamp) {
			// Nodes being deleted do not count towards total cluster resources
			continue
		}
		cores, memory := core_utils.GetNodeCoresAndMemory(node)

		coresTotal += cores
		memoryTotal += memory
	}

	return coresTotal, memoryTotal
}

func (sd *ScaleDown) calculateScaleDownCustomResourcesTotal(nodes []*apiv1.Node, cp cloudprovider.CloudProvider, timestamp time.Time) (map[string]int64, error) {
	result := make(map[string]int64)
	ngCache := make(map[string][]customresources.CustomResourceTarget)
	for _, node := range nodes {
		if isNodeBeingDeleted(node, timestamp) {
			// Nodes being deleted do not count towards total cluster resources
			continue
		}
		nodeGroup, err := cp.NodeGroupForNode(node)
		if err != nil {
			return nil, errors.ToAutoscalerError(errors.CloudProviderError, err).AddPrefix("can not get node group for node %v when calculating cluster gpu usage", node.Name)
		}
		if nodeGroup == nil || reflect.ValueOf(nodeGroup).IsNil() {
			// We do not trust cloud providers to return properly constructed nil for interface type - hence the reflection check.
			// See https://golang.org/doc/faq#nil_error
			// TODO[lukaszos] consider creating cloud_provider sanitizer which will wrap cloud provider and ensure sane behaviour.
			nodeGroup = nil
		}

		var resourceTargets []customresources.CustomResourceTarget
		var cacheHit bool

		if nodeGroup != nil {
			resourceTargets, cacheHit = ngCache[nodeGroup.Id()]
		}
		if !cacheHit {
			resourceTargets, err = sd.processors.CustomResourcesProcessor.GetNodeResourceTargets(sd.context, node, nodeGroup)
			if err != nil {
				return nil, errors.ToAutoscalerError(errors.CloudProviderError, err).AddPrefix("can not get gpu count for node %v when calculating cluster gpu usage")
			}
			if nodeGroup != nil {
				ngCache[nodeGroup.Id()] = resourceTargets
			}
		}

		for _, resourceTarget := range resourceTargets {
			if resourceTarget.ResourceType == "" || resourceTarget.ResourceCount == 0 {
				continue
			}
			result[resourceTarget.ResourceType] += resourceTarget.ResourceCount
		}
	}

	return result, nil
}

func isNodeBeingDeleted(node *apiv1.Node, timestamp time.Time) bool {
	deleteTime, _ := deletetaint.GetToBeDeletedTime(node)
	return deleteTime != nil && (timestamp.Sub(*deleteTime) < MaxCloudProviderNodeDeletionTime || timestamp.Sub(*deleteTime) < MaxKubernetesEmptyNodeDeletionTime)
}

func noScaleDownLimitsOnResources() scaleDownResourcesLimits {
	return nil
}

func copyScaleDownResourcesLimits(source scaleDownResourcesLimits) scaleDownResourcesLimits {
	copy := scaleDownResourcesLimits{}
	for k, v := range source {
		copy[k] = v
	}
	return copy
}

func (sd *ScaleDown) computeScaleDownResourcesDelta(cp cloudprovider.CloudProvider, node *apiv1.Node, nodeGroup cloudprovider.NodeGroup, resourcesWithLimits []string) (scaleDownResourcesDelta, errors.AutoscalerError) {
	resultScaleDownDelta := make(scaleDownResourcesDelta)

	nodeCPU, nodeMemory := core_utils.GetNodeCoresAndMemory(node)
	resultScaleDownDelta[cloudprovider.ResourceNameCores] = nodeCPU
	resultScaleDownDelta[cloudprovider.ResourceNameMemory] = nodeMemory

	if cloudprovider.ContainsCustomResources(resourcesWithLimits) {
		resourceTargets, err := sd.processors.CustomResourcesProcessor.GetNodeResourceTargets(sd.context, node, nodeGroup)
		if err != nil {
			return scaleDownResourcesDelta{}, errors.ToAutoscalerError(errors.CloudProviderError, err).AddPrefix("Failed to get node %v custom resources: %v", node.Name)
		}
		for _, resourceTarget := range resourceTargets {
			resultScaleDownDelta[resourceTarget.ResourceType] = resourceTarget.ResourceCount
		}
	}
	return resultScaleDownDelta, nil
}

type scaleDownLimitsCheckResult struct {
	exceeded          bool
	exceededResources []string
}

func scaleDownLimitsNotExceeded() scaleDownLimitsCheckResult {
	return scaleDownLimitsCheckResult{false, []string{}}
}

func (limits *scaleDownResourcesLimits) checkScaleDownDeltaWithinLimits(delta scaleDownResourcesDelta) scaleDownLimitsCheckResult {
	exceededResources := sets.NewString()
	for resource, resourceDelta := range delta {
		resourceLeft, found := (*limits)[resource]
		if found {
			if (resourceDelta > 0) && (resourceLeft == scaleDownLimitUnknown || resourceDelta > resourceLeft) {
				exceededResources.Insert(resource)
			}
		}
	}
	if len(exceededResources) > 0 {
		return scaleDownLimitsCheckResult{true, exceededResources.List()}
	}

	return scaleDownLimitsNotExceeded()
}

func (limits *scaleDownResourcesLimits) tryDecrementLimitsByDelta(delta scaleDownResourcesDelta) scaleDownLimitsCheckResult {
	result := limits.checkScaleDownDeltaWithinLimits(delta)
	if result.exceeded {
		return result
	}
	for resource, resourceDelta := range delta {
		resourceLeft, found := (*limits)[resource]
		if found {
			(*limits)[resource] = resourceLeft - resourceDelta
		}
	}
	return scaleDownLimitsNotExceeded()
}

// ScaleDown is responsible for maintaining the state needed to perform unneeded node removals.
type ScaleDown struct {
	context                *context.AutoscalingContext
	processors             *processors.AutoscalingProcessors
	clusterStateRegistry   *clusterstate.ClusterStateRegistry
	unneededNodes          map[string]time.Time
	unneededNodesList      []*apiv1.Node
	unremovableNodes       map[string]time.Time
	podLocationHints       map[string]string
	nodeUtilizationMap     map[string]simulator.UtilizationInfo
	usageTracker           *simulator.UsageTracker
	nodeDeletionTracker    *NodeDeletionTracker
	unremovableNodeReasons map[string]*simulator.UnremovableNode
}

// NewScaleDown builds new ScaleDown object.
func NewScaleDown(context *context.AutoscalingContext, processors *processors.AutoscalingProcessors, clusterStateRegistry *clusterstate.ClusterStateRegistry) *ScaleDown {
	return &ScaleDown{
		context:                context,
		processors:             processors,
		clusterStateRegistry:   clusterStateRegistry,
		unneededNodes:          make(map[string]time.Time),
		unremovableNodes:       make(map[string]time.Time),
		podLocationHints:       make(map[string]string),
		nodeUtilizationMap:     make(map[string]simulator.UtilizationInfo),
		usageTracker:           simulator.NewUsageTracker(),
		unneededNodesList:      make([]*apiv1.Node, 0),
		nodeDeletionTracker:    NewNodeDeletionTracker(),
		unremovableNodeReasons: make(map[string]*simulator.UnremovableNode),
	}
}

// CleanUp cleans up the internal ScaleDown state.
func (sd *ScaleDown) CleanUp(timestamp time.Time) {
	// Use default ScaleDownUnneededTime as in this context the value
	// doesn't apply to any specific NodeGroup.
	sd.usageTracker.CleanUp(timestamp.Add(-sd.context.NodeGroupDefaults.ScaleDownUnneededTime))
	sd.clearUnremovableNodeReasons()
}

// CleanUpUnneededNodes clears the list of unneeded nodes.
func (sd *ScaleDown) CleanUpUnneededNodes() {
	sd.unneededNodesList = make([]*apiv1.Node, 0)
	sd.unneededNodes = make(map[string]time.Time)
}

func (sd *ScaleDown) checkNodeUtilization(timestamp time.Time, node *apiv1.Node, nodeInfo *schedulerframework.NodeInfo) (simulator.UnremovableReason, *simulator.UtilizationInfo) {
	// Skip nodes that were recently checked.
	if _, found := sd.unremovableNodes[node.Name]; found {
		return simulator.RecentlyUnremovable, nil
	}

	// Skip nodes marked to be deleted, if they were marked recently.
	// Old-time marked nodes are again eligible for deletion - something went wrong with them
	// and they have not been deleted.
	if isNodeBeingDeleted(node, timestamp) {
		klog.V(1).Infof("Skipping %s from delete consideration - the node is currently being deleted", node.Name)
		return simulator.CurrentlyBeingDeleted, nil
	}

	// Skip nodes marked with no scale down annotation
	if hasNoScaleDownAnnotation(node) {
		klog.V(1).Infof("Skipping %s from delete consideration - the node is marked as no scale down", node.Name)
		return simulator.ScaleDownDisabledAnnotation, nil
	}

	utilInfo, err := simulator.CalculateUtilization(node, nodeInfo, sd.context.IgnoreDaemonSetsUtilization, sd.context.IgnoreMirrorPodsUtilization, sd.context.CloudProvider.GPULabel(), timestamp)
	if err != nil {
		klog.Warningf("Failed to calculate utilization for %s: %v", node.Name, err)
	}

	nodeGroup, err := sd.context.CloudProvider.NodeGroupForNode(node)
	if err != nil {
		return simulator.UnexpectedError, nil
	}
	if nodeGroup == nil || reflect.ValueOf(nodeGroup).IsNil() {
		// We should never get here as non-autoscaled nodes should not be included in scaleDownCandidates list
		// (and the default PreFilteringScaleDownNodeProcessor would indeed filter them out).
		klog.Warningf("Skipped %s from delete consideration - the node is not autoscaled", node.Name)
		return simulator.NotAutoscaled, nil
	}

	underutilized, err := sd.isNodeBelowUtilizationThreshold(node, nodeGroup, utilInfo)
	if err != nil {
		klog.Warningf("Failed to check utilization thresholds for %s: %v", node.Name, err)
		return simulator.UnexpectedError, nil
	}
	if !underutilized {
		klog.V(4).Infof("Node %s is not suitable for removal - %s utilization too big (%f)", node.Name, utilInfo.ResourceName, utilInfo.Utilization)
		return simulator.NotUnderutilized, &utilInfo
	}

	klog.V(4).Infof("Node %s - %s utilization %f", node.Name, utilInfo.ResourceName, utilInfo.Utilization)

	return simulator.NoReason, &utilInfo
}

// UpdateUnneededNodes calculates which nodes are not needed, i.e. all pods can be scheduled somewhere else,
// and updates unneededNodes map accordingly. It also computes information where pods can be rescheduled and
// node utilization level. The computations are made only for the nodes managed by CA.
// * destinationNodes are the nodes that can potentially take in any pods that are evicted because of a scale down.
// * scaleDownCandidates are the nodes that are being considered for scale down.
// * timestamp is the current timestamp.
// * pdbs is a list of pod disruption budgets.
func (sd *ScaleDown) UpdateUnneededNodes(
	destinationNodes []*apiv1.Node,
	scaleDownCandidates []*apiv1.Node,
	timestamp time.Time,
	pdbs []*policyv1.PodDisruptionBudget,
) errors.AutoscalerError {

	// Only scheduled non expendable pods and pods waiting for lower priority pods preemption can prevent node delete.
	// Extract cluster state from snapshot for initial analysis
	allNodeInfos, err := sd.context.ClusterSnapshot.NodeInfos().List()
	if err != nil {
		// This should never happen, List() returns err only because scheduler interface requires it.
		return errors.ToAutoscalerError(errors.InternalError, err)
	}

	sd.updateUnremovableNodes(timestamp)

	skipped := 0
	utilizationMap := make(map[string]simulator.UtilizationInfo)
	currentlyUnneededNodeNames := make([]string, 0, len(scaleDownCandidates))

	// Phase1 - look at the nodes utilization. Calculate the utilization
	// only for the managed nodes.
	for _, node := range scaleDownCandidates {
		nodeInfo, err := sd.context.ClusterSnapshot.NodeInfos().Get(node.Name)
		if err != nil {
			klog.Errorf("Can't retrieve scale-down candidate %s from snapshot, err: %v", node.Name, err)
			sd.addUnremovableNodeReason(node, simulator.UnexpectedError)
			continue
		}

		reason, utilInfo := sd.checkNodeUtilization(timestamp, node, nodeInfo)
		if utilInfo != nil {
			utilizationMap[node.Name] = *utilInfo
		}
		if reason != simulator.NoReason {
			// For logging purposes.
			if reason == simulator.RecentlyUnremovable {
				skipped++
			}

			sd.addUnremovableNodeReason(node, reason)
			continue
		}

		currentlyUnneededNodeNames = append(currentlyUnneededNodeNames, node.Name)
	}

	if skipped > 0 {
		klog.V(1).Infof("Scale-down calculation: ignoring %v nodes unremovable in the last %v", skipped, sd.context.AutoscalingOptions.UnremovableNodeRecheckTimeout)
	}

	emptyNodesToRemove := sd.getEmptyNodesToRemoveNoResourceLimits(currentlyUnneededNodeNames, timestamp)

	emptyNodes := make(map[string]bool)
	for _, empty := range emptyNodesToRemove {
		emptyNodes[empty.Node.Name] = true
	}

	currentlyUnneededNonEmptyNodes := make([]string, 0, len(currentlyUnneededNodeNames))
	for _, node := range currentlyUnneededNodeNames {
		if !emptyNodes[node] {
			currentlyUnneededNonEmptyNodes = append(currentlyUnneededNonEmptyNodes, node)
		}
	}

	// Phase2 - check which nodes can be probably removed using fast drain.
	currentCandidates, currentNonCandidates := sd.chooseCandidates(currentlyUnneededNonEmptyNodes)

	destinations := make([]string, 0, len(destinationNodes))
	for _, destinationNode := range destinationNodes {
		destinations = append(destinations, destinationNode.Name)
	}

	// Look for nodes to remove in the current candidates
	nodesToRemove, unremovable, newHints, simulatorErr := simulator.FindNodesToRemove(
		currentCandidates,
		destinations,
		sd.context.ListerRegistry,
		sd.context.ClusterSnapshot,
		sd.context.PredicateChecker,
		sd.podLocationHints,
		sd.usageTracker,
		timestamp,
		pdbs)
	if simulatorErr != nil {
		return sd.markSimulationError(simulatorErr, timestamp)
	}

	additionalCandidatesCount := sd.context.ScaleDownNonEmptyCandidatesCount - len(nodesToRemove)
	if additionalCandidatesCount > len(currentNonCandidates) {
		additionalCandidatesCount = len(currentNonCandidates)
	}
	// Limit the additional candidates pool size for better performance.
	additionalCandidatesPoolSize := int(math.Ceil(float64(len(allNodeInfos)) * sd.context.ScaleDownCandidatesPoolRatio))
	if additionalCandidatesPoolSize < sd.context.ScaleDownCandidatesPoolMinCount {
		additionalCandidatesPoolSize = sd.context.ScaleDownCandidatesPoolMinCount
	}
	if additionalCandidatesPoolSize > len(currentNonCandidates) {
		additionalCandidatesPoolSize = len(currentNonCandidates)
	}
	if additionalCandidatesCount > 0 {
		// Look for additional nodes to remove among the rest of nodes.
		klog.V(3).Infof("Finding additional %v candidates for scale down.", additionalCandidatesCount)
		additionalNodesToRemove, additionalUnremovable, additionalNewHints, simulatorErr :=
			simulator.FindNodesToRemove(
				currentNonCandidates[:additionalCandidatesPoolSize],
				destinations,
				sd.context.ListerRegistry,
				sd.context.ClusterSnapshot,
				sd.context.PredicateChecker,
				sd.podLocationHints,
				sd.usageTracker,
				timestamp,
				pdbs)
		if simulatorErr != nil {
			return sd.markSimulationError(simulatorErr, timestamp)
		}
		if len(additionalNodesToRemove) > additionalCandidatesCount {
			additionalNodesToRemove = additionalNodesToRemove[:additionalCandidatesCount]
		}
		nodesToRemove = append(nodesToRemove, additionalNodesToRemove...)
		unremovable = append(unremovable, additionalUnremovable...)
		for key, value := range additionalNewHints {
			newHints[key] = value
		}
	}

	for _, empty := range emptyNodesToRemove {
		nodesToRemove = append(nodesToRemove, simulator.NodeToBeRemoved{Node: empty.Node, PodsToReschedule: []*apiv1.Pod{}})
	}

	// Update the timestamp map.
	result := make(map[string]time.Time)
	unneededNodesList := make([]*apiv1.Node, 0, len(nodesToRemove))
	for _, node := range nodesToRemove {
		name := node.Node.Name
		unneededNodesList = append(unneededNodesList, node.Node)
		if val, found := sd.unneededNodes[name]; !found {
			result[name] = timestamp
		} else {
			result[name] = val
		}
	}

	// Add nodes to unremovable map
	if len(unremovable) > 0 {
		unremovableTimeout := timestamp.Add(sd.context.AutoscalingOptions.UnremovableNodeRecheckTimeout)
		for _, unremovableNode := range unremovable {
			sd.unremovableNodes[unremovableNode.Node.Name] = unremovableTimeout
			sd.addUnremovableNode(unremovableNode)
		}
		klog.V(1).Infof("%v nodes found to be unremovable in simulation, will re-check them at %v", len(unremovable), unremovableTimeout)
	}

	// This method won't always check all nodes, so let's give a generic reason for all nodes that weren't checked.
	for _, node := range scaleDownCandidates {
		_, unremovableReasonProvided := sd.unremovableNodeReasons[node.Name]
		_, unneeded := result[node.Name]
		if !unneeded && !unremovableReasonProvided {
			sd.addUnremovableNodeReason(node, simulator.NotUnneededOtherReason)
		}
	}

	// Update state and metrics
	sd.unneededNodesList = unneededNodesList
	sd.unneededNodes = result
	sd.podLocationHints = newHints
	sd.nodeUtilizationMap = utilizationMap
	sd.clusterStateRegistry.UpdateScaleDownCandidates(sd.unneededNodesList, timestamp)
	metrics.UpdateUnneededNodesCount(len(sd.unneededNodesList))
	return nil
}

// isNodeBelowUtilizationThreshold determines if a given node utilization is below threshold.
func (sd *ScaleDown) isNodeBelowUtilizationThreshold(node *apiv1.Node, nodeGroup cloudprovider.NodeGroup, utilInfo simulator.UtilizationInfo) (bool, error) {
	var threshold float64
	var err error
	if gpu.NodeHasGpu(sd.context.CloudProvider.GPULabel(), node) {
		threshold, err = sd.processors.NodeGroupConfigProcessor.GetScaleDownGpuUtilizationThreshold(sd.context, nodeGroup)
		if err != nil {
			return false, err
		}
	} else {
		threshold, err = sd.processors.NodeGroupConfigProcessor.GetScaleDownUtilizationThreshold(sd.context, nodeGroup)
		if err != nil {
			return false, err
		}
	}
	if utilInfo.Utilization >= threshold {
		return false, nil
	}
	return true, nil
}

// updateUnremovableNodes updates unremovableNodes map according to current
// state of the cluster. Removes from the map nodes that are no longer in the
// nodes list.
func (sd *ScaleDown) updateUnremovableNodes(timestamp time.Time) {
	if len(sd.unremovableNodes) <= 0 {
		return
	}
	newUnremovableNodes := make(map[string]time.Time, len(sd.unremovableNodes))
	for oldUnremovable, ttl := range sd.unremovableNodes {
		if _, err := sd.context.ClusterSnapshot.NodeInfos().Get(oldUnremovable); err != nil {
			// Not logging on error level as most likely cause is that node is no longer in the cluster.
			klog.Infof("Can't retrieve node %s from snapshot, removing from unremovable map, err: %v", oldUnremovable, err)
			continue
		}
		if ttl.After(timestamp) {
			// Keep nodes that are still in the cluster and haven't expired yet.
			newUnremovableNodes[oldUnremovable] = ttl
		}
	}
	sd.unremovableNodes = newUnremovableNodes
}

func (sd *ScaleDown) clearUnremovableNodeReasons() {
	sd.unremovableNodeReasons = make(map[string]*simulator.UnremovableNode)
}

func (sd *ScaleDown) addUnremovableNodeReason(node *apiv1.Node, reason simulator.UnremovableReason) {
	sd.unremovableNodeReasons[node.Name] = &simulator.UnremovableNode{Node: node, Reason: reason, BlockingPod: nil}
}

func (sd *ScaleDown) addUnremovableNode(unremovableNode *simulator.UnremovableNode) {
	sd.unremovableNodeReasons[unremovableNode.Node.Name] = unremovableNode
}

func (sd *ScaleDown) getUnremovableNodesCount() map[simulator.UnremovableReason]int {
	reasons := make(map[simulator.UnremovableReason]int)

	for _, node := range sd.unremovableNodeReasons {
		reasons[node.Reason]++
	}

	return reasons
}

// markSimulationError indicates a simulation error by clearing  relevant scale
// down state and returning an appropriate error.
func (sd *ScaleDown) markSimulationError(simulatorErr errors.AutoscalerError,
	timestamp time.Time) errors.AutoscalerError {
	klog.Errorf("Error while simulating node drains: %v", simulatorErr)
	sd.unneededNodesList = make([]*apiv1.Node, 0)
	sd.unneededNodes = make(map[string]time.Time)
	sd.nodeUtilizationMap = make(map[string]simulator.UtilizationInfo)
	sd.clusterStateRegistry.UpdateScaleDownCandidates(sd.unneededNodesList, timestamp)
	return simulatorErr.AddPrefix("error while simulating node drains: ")
}

// chooseCandidates splits nodes into current candidates for scale-down and the
// rest. Current candidates are unneeded nodes from the previous run that are
// still in the nodes list.
func (sd *ScaleDown) chooseCandidates(nodes []string) (candidates []string, nonCandidates []string) {
	// Number of candidates should not be capped. We will look for nodes to remove
	// from the whole set of nodes.
	if sd.context.ScaleDownNonEmptyCandidatesCount <= 0 {
		return nodes, nil
	}
	for _, node := range nodes {
		if _, found := sd.unneededNodes[node]; found {
			candidates = append(candidates, node)
		} else {
			nonCandidates = append(nonCandidates, node)
		}
	}
	return candidates, nonCandidates
}

func (sd *ScaleDown) mapNodesToStatusScaleDownNodes(nodes []*apiv1.Node, nodeGroups map[string]cloudprovider.NodeGroup, evictedPodLists map[string][]*apiv1.Pod) []*status.ScaleDownNode {
	var result []*status.ScaleDownNode
	for _, node := range nodes {
		result = append(result, &status.ScaleDownNode{
			Node:        node,
			NodeGroup:   nodeGroups[node.Name],
			UtilInfo:    sd.nodeUtilizationMap[node.Name],
			EvictedPods: evictedPodLists[node.Name],
		})
	}
	return result
}

// SoftTaintUnneededNodes manage soft taints of unneeded nodes.
func (sd *ScaleDown) SoftTaintUnneededNodes(allNodes []*apiv1.Node) (errors []error) {
	defer metrics.UpdateDurationFromStart(metrics.ScaleDownSoftTaintUnneeded, time.Now())
	apiCallBudget := sd.context.AutoscalingOptions.MaxBulkSoftTaintCount
	timeBudget := sd.context.AutoscalingOptions.MaxBulkSoftTaintTime
	skippedNodes := 0
	startTime := now()
	for _, node := range allNodes {
		if deletetaint.HasToBeDeletedTaint(node) {
			// Do not consider nodes that are scheduled to be deleted
			continue
		}
		alreadyTainted := deletetaint.HasDeletionCandidateTaint(node)
		_, unneeded := sd.unneededNodes[node.Name]

		// Check if expected taints match existing taints
		if unneeded != alreadyTainted {
			if apiCallBudget <= 0 || now().Sub(startTime) >= timeBudget {
				skippedNodes++
				continue
			}
			apiCallBudget--
			if unneeded && !alreadyTainted {
				err := deletetaint.MarkDeletionCandidate(node, sd.context.ClientSet)
				if err != nil {
					errors = append(errors, err)
					klog.Warningf("Soft taint on %s adding error %v", node.Name, err)
				}
			}
			if !unneeded && alreadyTainted {
				_, err := deletetaint.CleanDeletionCandidate(node, sd.context.ClientSet)
				if err != nil {
					errors = append(errors, err)
					klog.Warningf("Soft taint on %s removal error %v", node.Name, err)
				}
			}
		}
	}
	if skippedNodes > 0 {
		klog.V(4).Infof("Skipped adding/removing soft taints on %v nodes - API call limit exceeded", skippedNodes)
	}
	return
}

// TryToScaleDown tries to scale down the cluster. It returns a result inside a ScaleDownStatus indicating if any node was
// removed and error if such occurred.
func (sd *ScaleDown) TryToScaleDown(
	currentTime time.Time,
	pdbs []*policyv1.PodDisruptionBudget,
) (*status.ScaleDownStatus, errors.AutoscalerError) {

	scaleDownStatus := &status.ScaleDownStatus{NodeDeleteResults: sd.nodeDeletionTracker.GetAndClearNodeDeleteResults()}
	nodeDeletionDuration := time.Duration(0)
	findNodesToRemoveDuration := time.Duration(0)
	defer updateScaleDownMetrics(time.Now(), &findNodesToRemoveDuration, &nodeDeletionDuration)

	allNodeInfos, errSnapshot := sd.context.ClusterSnapshot.NodeInfos().List()
	if errSnapshot != nil {
		// This should never happen, List() returns err only because scheduler interface requires it.
		return scaleDownStatus, errors.ToAutoscalerError(errors.InternalError, errSnapshot)
	}

	nodesWithoutMaster := filterOutMasters(allNodeInfos)
	nodesWithoutMasterNames := make([]string, 0, len(nodesWithoutMaster))
	for _, node := range nodesWithoutMaster {
		nodesWithoutMasterNames = append(nodesWithoutMasterNames, node.Name)
	}

	candidateNames := make([]string, 0)
	readinessMap := make(map[string]bool)
	candidateNodeGroups := make(map[string]cloudprovider.NodeGroup)
	gpuLabel := sd.context.CloudProvider.GPULabel()
	availableGPUTypes := sd.context.CloudProvider.GetAvailableGPUTypes()

	resourceLimiter, errCP := sd.context.CloudProvider.GetResourceLimiter()
	if errCP != nil {
		scaleDownStatus.Result = status.ScaleDownError
		return scaleDownStatus, errors.ToAutoscalerError(errors.CloudProviderError, errCP)
	}

	scaleDownResourcesLeft := sd.computeScaleDownResourcesLeftLimits(nodesWithoutMaster, resourceLimiter, sd.context.CloudProvider, currentTime)

	nodeGroupSize := utils.GetNodeGroupSizeMap(sd.context.CloudProvider)
	resourcesWithLimits := resourceLimiter.GetResources()
	for nodeName, unneededSince := range sd.unneededNodes {
		klog.V(2).Infof("%s was unneeded for %s", nodeName, currentTime.Sub(unneededSince).String())

		nodeInfo, err := sd.context.ClusterSnapshot.NodeInfos().Get(nodeName)
		if err != nil {
			klog.Errorf("Can't retrieve unneeded node %s from snapshot, err: %v", nodeName, err)
			continue
		}

		node := nodeInfo.Node()

		// Check if node is marked with no scale down annotation.
		if hasNoScaleDownAnnotation(node) {
			klog.V(4).Infof("Skipping %s - scale down disabled annotation found", node.Name)
			sd.addUnremovableNodeReason(node, simulator.ScaleDownDisabledAnnotation)
			continue
		}

		ready, _, _ := kube_util.GetReadinessState(node)
		readinessMap[node.Name] = ready

		nodeGroup, err := sd.context.CloudProvider.NodeGroupForNode(node)
		if err != nil {
			klog.Errorf("Error while checking node group for %s: %v", node.Name, err)
			sd.addUnremovableNodeReason(node, simulator.UnexpectedError)
			continue
		}
		if nodeGroup == nil || reflect.ValueOf(nodeGroup).IsNil() {
			klog.V(4).Infof("Skipping %s - no node group config", node.Name)
			sd.addUnremovableNodeReason(node, simulator.NotAutoscaled)
			continue
		}

		if ready {
			// Check how long a ready node was underutilized.
			unneededTime, err := sd.processors.NodeGroupConfigProcessor.GetScaleDownUnneededTime(sd.context, nodeGroup)
			if err != nil {
				klog.Errorf("Error trying to get ScaleDownUnneededTime for node %s (in group: %s)", node.Name, nodeGroup.Id())
				continue
			}
			if !unneededSince.Add(unneededTime).Before(currentTime) {
				sd.addUnremovableNodeReason(node, simulator.NotUnneededLongEnough)
				continue
			}
		} else {
			// Unready nodes may be deleted after a different time than underutilized nodes.
			unreadyTime, err := sd.processors.NodeGroupConfigProcessor.GetScaleDownUnreadyTime(sd.context, nodeGroup)
			if err != nil {
				klog.Errorf("Error trying to get ScaleDownUnnreadyTime for node %s (in group: %s)", node.Name, nodeGroup.Id())
				continue
			}
			if !unneededSince.Add(unreadyTime).Before(currentTime) {
				sd.addUnremovableNodeReason(node, simulator.NotUnreadyLongEnough)
				continue
			}
		}

		size, found := nodeGroupSize[nodeGroup.Id()]
		if !found {
			klog.Errorf("Error while checking node group size %s: group size not found in cache", nodeGroup.Id())
			sd.addUnremovableNodeReason(node, simulator.UnexpectedError)
			continue
		}

		deletionsInProgress := sd.nodeDeletionTracker.GetDeletionsInProgress(nodeGroup.Id())
		if size-deletionsInProgress <= nodeGroup.MinSize() {
			klog.V(1).Infof("Skipping %s - node group min size reached", node.Name)
			sd.addUnremovableNodeReason(node, simulator.NodeGroupMinSizeReached)
			continue
		}

		scaleDownResourcesDelta, err := sd.computeScaleDownResourcesDelta(sd.context.CloudProvider, node, nodeGroup, resourcesWithLimits)
		if err != nil {
			klog.Errorf("Error getting node resources: %v", err)
			sd.addUnremovableNodeReason(node, simulator.UnexpectedError)
			continue
		}

		checkResult := scaleDownResourcesLeft.checkScaleDownDeltaWithinLimits(scaleDownResourcesDelta)
		if checkResult.exceeded {
			klog.V(4).Infof("Skipping %s - minimal limit exceeded for %v", node.Name, checkResult.exceededResources)
			sd.addUnremovableNodeReason(node, simulator.MinimalResourceLimitExceeded)
			continue
		}

		candidateNames = append(candidateNames, node.Name)
		candidateNodeGroups[node.Name] = nodeGroup
	}

	if len(candidateNames) == 0 {
		klog.V(1).Infof("No candidates for scale down")
		scaleDownStatus.Result = status.ScaleDownNoUnneeded
		return scaleDownStatus, nil
	}

	// Trying to delete empty nodes in bulk. If there are no empty nodes then CA will
	// try to delete not-so-empty nodes, possibly killing some pods and allowing them
	// to recreate on other nodes.
	emptyNodesToRemove := sd.getEmptyNodesToRemove(candidateNames, scaleDownResourcesLeft, currentTime)
	emptyNodesToRemove = sd.processors.ScaleDownSetProcessor.GetNodesToRemove(sd.context, emptyNodesToRemove, sd.context.MaxEmptyBulkDelete)
	if len(emptyNodesToRemove) > 0 {
		nodeDeletionStart := time.Now()
		deletedNodes, err := sd.scheduleDeleteEmptyNodes(emptyNodesToRemove, sd.context.ClientSet, sd.context.Recorder, readinessMap, candidateNodeGroups)
		nodeDeletionDuration = time.Now().Sub(nodeDeletionStart)

		// TODO: Give the processor some information about the nodes that failed to be deleted.
		scaleDownStatus.ScaledDownNodes = sd.mapNodesToStatusScaleDownNodes(deletedNodes, candidateNodeGroups, make(map[string][]*apiv1.Pod))
		if len(deletedNodes) > 0 {
			scaleDownStatus.Result = status.ScaleDownNodeDeleteStarted
		} else {
			scaleDownStatus.Result = status.ScaleDownError
		}
		if err != nil {
			return scaleDownStatus, err.AddPrefix("failed to delete at least one empty node: ")
		}
		return scaleDownStatus, nil
	}

	findNodesToRemoveStart := time.Now()

	// We look for only 1 node so new hints may be incomplete.
	nodesToRemove, unremovable, _, err := simulator.FindNodesToRemove(
		candidateNames,
		nodesWithoutMasterNames,
		sd.context.ListerRegistry,
		sd.context.ClusterSnapshot,
		sd.context.PredicateChecker,
		sd.podLocationHints,
		sd.usageTracker,
		time.Now(),
		pdbs)
	findNodesToRemoveDuration = time.Now().Sub(findNodesToRemoveStart)

	for _, unremovableNode := range unremovable {
		sd.addUnremovableNode(unremovableNode)
	}

	if err != nil {
		scaleDownStatus.Result = status.ScaleDownError
		return scaleDownStatus, err.AddPrefix("Find node to remove failed: ")
	}
	nodesToRemove = sd.processors.ScaleDownSetProcessor.GetNodesToRemove(sd.context, nodesToRemove, 1)
	if len(nodesToRemove) == 0 {
		klog.V(1).Infof("No node to remove")
		scaleDownStatus.Result = status.ScaleDownNoNodeDeleted
		return scaleDownStatus, nil
	}
	toRemove := nodesToRemove[0]
	utilization := sd.nodeUtilizationMap[toRemove.Node.Name]
	podNames := make([]string, 0, len(toRemove.PodsToReschedule))
	for _, pod := range toRemove.PodsToReschedule {
		podNames = append(podNames, pod.Namespace+"/"+pod.Name)
	}
	klog.V(0).Infof("Scale-down: removing node %s, utilization: %v, pods to reschedule: %s", toRemove.Node.Name, utilization,
		strings.Join(podNames, ","))
	sd.context.LogRecorder.Eventf(apiv1.EventTypeNormal, "ScaleDown", "Scale-down: removing node %s, utilization: %v, pods to reschedule: %s",
		toRemove.Node.Name, utilization, strings.Join(podNames, ","))

	// Nothing super-bad should happen if the node is removed from tracker prematurely.
	simulator.RemoveNodeFromTracker(sd.usageTracker, toRemove.Node.Name, sd.unneededNodes)
	nodeDeletionStart := time.Now()

	// Starting deletion.
	nodeDeletionDuration = time.Now().Sub(nodeDeletionStart)
	sd.nodeDeletionTracker.SetNonEmptyNodeDeleteInProgress(true)

	go func() {
		// Finishing the delete process once this goroutine is over.
		var result status.NodeDeleteResult
		defer func() { sd.nodeDeletionTracker.AddNodeDeleteResult(toRemove.Node.Name, result) }()
		defer sd.nodeDeletionTracker.SetNonEmptyNodeDeleteInProgress(false)
		nodeGroup, found := candidateNodeGroups[toRemove.Node.Name]
		if !found {
			result = status.NodeDeleteResult{ResultType: status.NodeDeleteErrorFailedToDelete, Err: errors.NewAutoscalerError(
				errors.InternalError, "failed to find node group for %s", toRemove.Node.Name)}
			return
		}
		result = sd.deleteNode(toRemove.Node, toRemove.PodsToReschedule, toRemove.DaemonSetPods, nodeGroup)
		if result.ResultType != status.NodeDeleteOk {
			klog.Errorf("Failed to delete %s: %v", toRemove.Node.Name, result.Err)
			return
		}
		if readinessMap[toRemove.Node.Name] {
			metrics.RegisterScaleDown(1, gpu.GetGpuTypeForMetrics(gpuLabel, availableGPUTypes, toRemove.Node, nodeGroup), metrics.Underutilized)
		} else {
			metrics.RegisterScaleDown(1, gpu.GetGpuTypeForMetrics(gpuLabel, availableGPUTypes, toRemove.Node, nodeGroup), metrics.Unready)
		}
	}()

	scaleDownStatus.ScaledDownNodes = sd.mapNodesToStatusScaleDownNodes([]*apiv1.Node{toRemove.Node}, candidateNodeGroups, map[string][]*apiv1.Pod{toRemove.Node.Name: toRemove.PodsToReschedule})
	scaleDownStatus.Result = status.ScaleDownNodeDeleteStarted
	return scaleDownStatus, nil
}

// updateScaleDownMetrics registers duration of different parts of scale down.
// Separates time spent on finding nodes to remove, deleting nodes and other operations.
func updateScaleDownMetrics(scaleDownStart time.Time, findNodesToRemoveDuration *time.Duration, nodeDeletionDuration *time.Duration) {
	stop := time.Now()
	miscDuration := stop.Sub(scaleDownStart) - *nodeDeletionDuration - *findNodesToRemoveDuration
	metrics.UpdateDuration(metrics.ScaleDownNodeDeletion, *nodeDeletionDuration)
	metrics.UpdateDuration(metrics.ScaleDownFindNodesToRemove, *findNodesToRemoveDuration)
	metrics.UpdateDuration(metrics.ScaleDownMiscOperations, miscDuration)
}

func (sd *ScaleDown) getEmptyNodesToRemoveNoResourceLimits(candidates []string, timestamp time.Time) []simulator.NodeToBeRemoved {
	return sd.getEmptyNodesToRemove(candidates, noScaleDownLimitsOnResources(), timestamp)
}

// This functions finds empty nodes among passed candidates and returns a list of empty nodes
// that can be deleted at the same time.
func (sd *ScaleDown) getEmptyNodesToRemove(candidates []string, resourcesLimits scaleDownResourcesLimits,
	timestamp time.Time) []simulator.NodeToBeRemoved {

	emptyNodes := simulator.FindEmptyNodesToRemove(sd.context.ClusterSnapshot, candidates, timestamp)
	availabilityMap := make(map[string]int)
	nodesToRemove := make([]simulator.NodeToBeRemoved, 0)
	resourcesLimitsCopy := copyScaleDownResourcesLimits(resourcesLimits) // we do not want to modify input parameter
	resourcesNames := sets.StringKeySet(resourcesLimits).List()
	for _, nodeName := range emptyNodes {
		nodeInfo, err := sd.context.ClusterSnapshot.NodeInfos().Get(nodeName)
		if err != nil {
			klog.Errorf("Can't retrieve node %s from snapshot, err: %v", nodeName, err)
			continue
		}
		node := nodeInfo.Node()
		nodeGroup, err := sd.context.CloudProvider.NodeGroupForNode(node)
		if err != nil {
			klog.Errorf("Failed to get group for %s", nodeName)
			continue
		}
		if nodeGroup == nil || reflect.ValueOf(nodeGroup).IsNil() {
			continue
		}
		var available int
		var found bool
		if available, found = availabilityMap[nodeGroup.Id()]; !found {
			// Will be cached.
			size, err := nodeGroup.TargetSize()
			if err != nil {
				klog.Errorf("Failed to get size for %s: %v ", nodeGroup.Id(), err)
				continue
			}
			deletionsInProgress := sd.nodeDeletionTracker.GetDeletionsInProgress(nodeGroup.Id())
			available = size - nodeGroup.MinSize() - deletionsInProgress
			if available < 0 {
				available = 0
			}
			availabilityMap[nodeGroup.Id()] = available
		}
		if available > 0 {
			resourcesDelta, err := sd.computeScaleDownResourcesDelta(sd.context.CloudProvider, node, nodeGroup, resourcesNames)
			if err != nil {
				klog.Errorf("Error: %v", err)
				continue
			}
			checkResult := resourcesLimitsCopy.tryDecrementLimitsByDelta(resourcesDelta)
			if checkResult.exceeded {
				continue
			}
			available--
			availabilityMap[nodeGroup.Id()] = available
			nodesToRemove = append(nodesToRemove, simulator.NodeToBeRemoved{
				Node: node,
			})
		}
	}

	return nodesToRemove
}

func (sd *ScaleDown) scheduleDeleteEmptyNodes(emptyNodesToRemove []simulator.NodeToBeRemoved, client kube_client.Interface,
	recorder kube_record.EventRecorder, readinessMap map[string]bool,
	candidateNodeGroups map[string]cloudprovider.NodeGroup) ([]*apiv1.Node, errors.AutoscalerError) {
	deletedNodes := []*apiv1.Node{}
	for _, empty := range emptyNodesToRemove {
		klog.V(0).Infof("Scale-down: removing empty node %s", empty.Node.Name)
		sd.context.LogRecorder.Eventf(apiv1.EventTypeNormal, "ScaleDownEmpty", "Scale-down: removing empty node %s", empty.Node.Name)
		simulator.RemoveNodeFromTracker(sd.usageTracker, empty.Node.Name, sd.unneededNodes)
		nodeGroup, found := candidateNodeGroups[empty.Node.Name]
		if !found {
			return deletedNodes, errors.NewAutoscalerError(
				errors.CloudProviderError, "failed to find node group for %s", empty.Node.Name)
		}
		taintErr := deletetaint.MarkToBeDeleted(empty.Node, client, sd.context.CordonNodeBeforeTerminate)
		if taintErr != nil {
			recorder.Eventf(empty.Node, apiv1.EventTypeWarning, "ScaleDownFailed", "failed to mark the node as toBeDeleted/unschedulable: %v", taintErr)
			return deletedNodes, errors.ToAutoscalerError(errors.ApiCallError, taintErr)
		}
		deletedNodes = append(deletedNodes, empty.Node)
		go func(nodeToDelete *apiv1.Node, nodeGroupForDeletedNode cloudprovider.NodeGroup, evictByDefault bool) {
			sd.nodeDeletionTracker.StartDeletion(nodeGroupForDeletedNode.Id())
			defer sd.nodeDeletionTracker.EndDeletion(nodeGroupForDeletedNode.Id())
			var result status.NodeDeleteResult
			defer func() { sd.nodeDeletionTracker.AddNodeDeleteResult(nodeToDelete.Name, result) }()

			var deleteErr errors.AutoscalerError
			// If we fail to delete the node we want to remove delete taint
			defer func() {
				if deleteErr != nil {
					deletetaint.CleanToBeDeleted(nodeToDelete, client, sd.context.CordonNodeBeforeTerminate)
					recorder.Eventf(nodeToDelete, apiv1.EventTypeWarning, "ScaleDownFailed", "failed to delete empty node: %v", deleteErr)
				} else {
					sd.context.LogRecorder.Eventf(apiv1.EventTypeNormal, "ScaleDownEmpty", "Scale-down: empty node %s removed", nodeToDelete.Name)
				}
			}()
			if err := evictDaemonSetPods(sd.context.ClusterSnapshot, nodeToDelete, client, sd.context.MaxGracefulTerminationSec, time.Now(), DaemonSetEvictionEmptyNodeTimeout, DeamonSetTimeBetweenEvictionRetries, recorder, evictByDefault); err != nil {
				klog.Warningf("error while evicting DS pods from an empty node: %v", err)
			}
			deleteErr = waitForDelayDeletion(nodeToDelete, sd.context.ListerRegistry.AllNodeLister(), sd.context.AutoscalingOptions.NodeDeletionDelayTimeout)
			if deleteErr != nil {
				klog.Errorf("Problem with empty node deletion: %v", deleteErr)
				result = status.NodeDeleteResult{ResultType: status.NodeDeleteErrorFailedToDelete, Err: deleteErr}
				return
			}
			deleteErr = deleteNodeFromCloudProvider(nodeToDelete, sd.context.CloudProvider,
				sd.context.Recorder, sd.clusterStateRegistry)
			if deleteErr != nil {
				klog.Errorf("Problem with empty node deletion: %v", deleteErr)
				result = status.NodeDeleteResult{ResultType: status.NodeDeleteErrorFailedToDelete, Err: deleteErr}
				return
			}
			if readinessMap[nodeToDelete.Name] {
				metrics.RegisterScaleDown(1, gpu.GetGpuTypeForMetrics(sd.context.CloudProvider.GPULabel(), sd.context.CloudProvider.GetAvailableGPUTypes(), nodeToDelete, nodeGroupForDeletedNode), metrics.Empty)
			} else {
				metrics.RegisterScaleDown(1, gpu.GetGpuTypeForMetrics(sd.context.CloudProvider.GPULabel(), sd.context.CloudProvider.GetAvailableGPUTypes(), nodeToDelete, nodeGroupForDeletedNode), metrics.Unready)
			}
			result = status.NodeDeleteResult{ResultType: status.NodeDeleteOk}
		}(empty.Node, nodeGroup, sd.context.DaemonSetEvictionForEmptyNodes)
	}
	return deletedNodes, nil
}

// Create eviction object for all DaemonSet pods on the node
func evictDaemonSetPods(clusterSnapshot simulator.ClusterSnapshot, nodeToDelete *apiv1.Node, client kube_client.Interface, maxGracefulTerminationSec int, timeNow time.Time, dsEvictionTimeout time.Duration, waitBetweenRetries time.Duration,
	recorder kube_record.EventRecorder, evictByDefault bool) error {
	nodeInfo, err := clusterSnapshot.NodeInfos().Get(nodeToDelete.Name)
	if err != nil {
		return fmt.Errorf("failed to get node info for %s", nodeToDelete.Name)
	}
	_, daemonSetPods, _, err := simulator.FastGetPodsToMove(nodeInfo, true, true, []*policyv1.PodDisruptionBudget{}, timeNow)
	if err != nil {
		return fmt.Errorf("failed to get DaemonSet pods for %s (error: %v)", nodeToDelete.Name, err)
	}

	daemonSetPods = daemonset.PodsToEvict(daemonSetPods, evictByDefault)

	dsEviction := make(chan status.PodEvictionResult, len(daemonSetPods))

	// Perform eviction of DaemonSet pods
	for _, daemonSetPod := range daemonSetPods {
		go func(podToEvict *apiv1.Pod) {
			dsEviction <- evictPod(podToEvict, true, client, recorder, maxGracefulTerminationSec, timeNow.Add(dsEvictionTimeout), waitBetweenRetries)
		}(daemonSetPod)
	}
	// Wait for creating eviction of DaemonSet pods
	var failedPodErrors []string
	for range daemonSetPods {
		select {
		case status := <-dsEviction:
			if status.Err != nil {
				failedPodErrors = append(failedPodErrors, status.Err.Error())
			}
		// adding waitBetweenRetries in order to have a bigger time interval than evictPod()
		case <-time.After(dsEvictionTimeout):
			return fmt.Errorf("failed to create DaemonSet eviction for %v seconds on the %s", dsEvictionTimeout, nodeToDelete.Name)
		}
	}
	if len(failedPodErrors) > 0 {

		return fmt.Errorf("following DaemonSet pod failed to evict on the %s:\n%s", nodeToDelete.Name, fmt.Errorf(strings.Join(failedPodErrors, "\n")))
	}
	return nil
}

func (sd *ScaleDown) deleteNode(node *apiv1.Node, pods []*apiv1.Pod, daemonSetPods []*apiv1.Pod,
	nodeGroup cloudprovider.NodeGroup) status.NodeDeleteResult {
	deleteSuccessful := false
	drainSuccessful := false

	if err := deletetaint.MarkToBeDeleted(node, sd.context.ClientSet, sd.context.CordonNodeBeforeTerminate); err != nil {
		sd.context.Recorder.Eventf(node, apiv1.EventTypeWarning, "ScaleDownFailed", "failed to mark the node as toBeDeleted/unschedulable: %v", err)
		return status.NodeDeleteResult{ResultType: status.NodeDeleteErrorFailedToMarkToBeDeleted, Err: errors.ToAutoscalerError(errors.ApiCallError, err)}
	}

	sd.nodeDeletionTracker.StartDeletion(nodeGroup.Id())
	defer sd.nodeDeletionTracker.EndDeletion(nodeGroup.Id())

	// If we fail to evict all the pods from the node we want to remove delete taint
	defer func() {
		if !deleteSuccessful {
			deletetaint.CleanToBeDeleted(node, sd.context.ClientSet, sd.context.CordonNodeBeforeTerminate)
			if !drainSuccessful {
				sd.context.Recorder.Eventf(node, apiv1.EventTypeWarning, "ScaleDownFailed", "failed to drain the node, aborting ScaleDown")
			} else {
				sd.context.Recorder.Eventf(node, apiv1.EventTypeWarning, "ScaleDownFailed", "failed to delete the node")
			}
		}
	}()

	sd.context.Recorder.Eventf(node, apiv1.EventTypeNormal, "ScaleDown", "marked the node as toBeDeleted/unschedulable")

	daemonSetPods = daemonset.PodsToEvict(daemonSetPods, sd.context.DaemonSetEvictionForOccupiedNodes)

	// attempt drain
	evictionResults, err := drainNode(node, pods, daemonSetPods, sd.context.ClientSet, sd.context.Recorder, sd.context.MaxGracefulTerminationSec, MaxPodEvictionTime, EvictionRetryTime, PodEvictionHeadroom)
	if err != nil {
		return status.NodeDeleteResult{ResultType: status.NodeDeleteErrorFailedToEvictPods, Err: err, PodEvictionResults: evictionResults}
	}
	drainSuccessful = true

	if typedErr := waitForDelayDeletion(node, sd.context.ListerRegistry.AllNodeLister(), sd.context.AutoscalingOptions.NodeDeletionDelayTimeout); typedErr != nil {
		return status.NodeDeleteResult{ResultType: status.NodeDeleteErrorFailedToDelete, Err: typedErr}
	}

	// attempt delete from cloud provider

	if typedErr := deleteNodeFromCloudProvider(node, sd.context.CloudProvider, sd.context.Recorder, sd.clusterStateRegistry); typedErr != nil {
		return status.NodeDeleteResult{ResultType: status.NodeDeleteErrorFailedToDelete, Err: typedErr}
	}

	deleteSuccessful = true // Let the deferred function know there is no need to cleanup
	return status.NodeDeleteResult{ResultType: status.NodeDeleteOk}
}

func evictPod(podToEvict *apiv1.Pod, isDaemonSetPod bool, client kube_client.Interface, recorder kube_record.EventRecorder,
	maxGracefulTerminationSec int, retryUntil time.Time, waitBetweenRetries time.Duration) status.PodEvictionResult {
	recorder.Eventf(podToEvict, apiv1.EventTypeNormal, "ScaleDown", "deleting pod for node scale down")

	maxTermination := int64(apiv1.DefaultTerminationGracePeriodSeconds)
	if podToEvict.Spec.TerminationGracePeriodSeconds != nil {
		if *podToEvict.Spec.TerminationGracePeriodSeconds < int64(maxGracefulTerminationSec) {
			maxTermination = *podToEvict.Spec.TerminationGracePeriodSeconds
		} else {
			maxTermination = int64(maxGracefulTerminationSec)
		}
	}

	var lastError error
	for first := true; first || time.Now().Before(retryUntil); time.Sleep(waitBetweenRetries) {
		first = false
		eviction := &policyv1.Eviction{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: podToEvict.Namespace,
				Name:      podToEvict.Name,
			},
			DeleteOptions: &metav1.DeleteOptions{
				GracePeriodSeconds: &maxTermination,
			},
		}
		lastError = client.CoreV1().Pods(podToEvict.Namespace).Evict(ctx.TODO(), eviction)
		if lastError == nil || kube_errors.IsNotFound(lastError) {
			return status.PodEvictionResult{Pod: podToEvict, TimedOut: false, Err: nil}
		}
	}
	if !isDaemonSetPod {
		klog.Errorf("Failed to evict pod %s, error: %v", podToEvict.Name, lastError)
		recorder.Eventf(podToEvict, apiv1.EventTypeWarning, "ScaleDownFailed", "failed to delete pod for ScaleDown")
	}
	return status.PodEvictionResult{Pod: podToEvict, TimedOut: true, Err: fmt.Errorf("failed to evict pod %s/%s within allowed timeout (last error: %v)", podToEvict.Namespace, podToEvict.Name, lastError)}
}

// Performs drain logic on the node. Marks the node as unschedulable and later removes all pods, giving
// them up to MaxGracefulTerminationTime to finish.
func drainNode(node *apiv1.Node, pods []*apiv1.Pod, daemonSetPods []*apiv1.Pod, client kube_client.Interface, recorder kube_record.EventRecorder,
	maxGracefulTerminationSec int, maxPodEvictionTime time.Duration, waitBetweenRetries time.Duration,
	podEvictionHeadroom time.Duration) (evictionResults map[string]status.PodEvictionResult, err error) {

	evictionResults = make(map[string]status.PodEvictionResult)
	retryUntil := time.Now().Add(maxPodEvictionTime)
	confirmations := make(chan status.PodEvictionResult, len(pods))
	daemonSetConfirmations := make(chan status.PodEvictionResult, len(daemonSetPods))
	for _, pod := range pods {
		evictionResults[pod.Name] = status.PodEvictionResult{Pod: pod, TimedOut: true, Err: nil}
		go func(podToEvict *apiv1.Pod) {
			confirmations <- evictPod(podToEvict, false, client, recorder, maxGracefulTerminationSec, retryUntil, waitBetweenRetries)
		}(pod)
	}

	// Perform eviction of daemonset. We don't want to raise an error if daemonsetPod wasn't evict properly
	for _, daemonSetPod := range daemonSetPods {
		go func(podToEvict *apiv1.Pod) {
			daemonSetConfirmations <- evictPod(podToEvict, true, client, recorder, maxGracefulTerminationSec, retryUntil, waitBetweenRetries)
		}(daemonSetPod)

	}

	podsEvictionCounter := 0
	for i := 0; i < len(pods)+len(daemonSetPods); i++ {
		select {
		case evictionResult := <-confirmations:
			podsEvictionCounter++
			evictionResults[evictionResult.Pod.Name] = evictionResult
			if evictionResult.WasEvictionSuccessful() {
				metrics.RegisterEvictions(1)
			}
		case <-daemonSetConfirmations:
		case <-time.After(retryUntil.Sub(time.Now()) + 5*time.Second):
			if podsEvictionCounter < len(pods) {
				// All pods initially had results with TimedOut set to true, so the ones that didn't receive an actual result are correctly marked as timed out.
				return evictionResults, errors.NewAutoscalerError(errors.ApiCallError, "Failed to drain node %s/%s: timeout when waiting for creating evictions", node.Namespace, node.Name)
			}
			klog.Infof("Timeout when waiting for creating daemonSetPods eviction")
		}
	}

	evictionErrs := make([]error, 0)
	for _, result := range evictionResults {
		if !result.WasEvictionSuccessful() {
			evictionErrs = append(evictionErrs, result.Err)
		}
	}
	if len(evictionErrs) != 0 {
		return evictionResults, errors.NewAutoscalerError(errors.ApiCallError, "Failed to drain node %s/%s, due to following errors: %v", node.Namespace, node.Name, evictionErrs)
	}

	// Evictions created successfully, wait maxGracefulTerminationSec + podEvictionHeadroom to see if pods really disappeared.
	var allGone bool
	for start := time.Now(); time.Now().Sub(start) < time.Duration(maxGracefulTerminationSec)*time.Second+podEvictionHeadroom; time.Sleep(5 * time.Second) {
		allGone = true
		for _, pod := range pods {
			podreturned, err := client.CoreV1().Pods(pod.Namespace).Get(ctx.TODO(), pod.Name, metav1.GetOptions{})
			if err == nil && (podreturned == nil || podreturned.Spec.NodeName == node.Name) {
				klog.Errorf("Not deleted yet %s/%s", pod.Namespace, pod.Name)
				allGone = false
				break
			}
			if err != nil && !kube_errors.IsNotFound(err) {
				klog.Errorf("Failed to check pod %s/%s: %v", pod.Namespace, pod.Name, err)
				allGone = false
				break
			}
		}
		if allGone {
			klog.V(1).Infof("All pods removed from %s", node.Name)
			// Let the deferred function know there is no need for cleanup
			return evictionResults, nil
		}
	}

	for _, pod := range pods {
		podReturned, err := client.CoreV1().Pods(pod.Namespace).Get(ctx.TODO(), pod.Name, metav1.GetOptions{})
		if err == nil && (podReturned == nil || podReturned.Spec.NodeName == node.Name) {
			evictionResults[pod.Name] = status.PodEvictionResult{Pod: pod, TimedOut: true, Err: nil}
		} else if err != nil && !kube_errors.IsNotFound(err) {
			evictionResults[pod.Name] = status.PodEvictionResult{Pod: pod, TimedOut: true, Err: err}
		} else {
			evictionResults[pod.Name] = status.PodEvictionResult{Pod: pod, TimedOut: false, Err: nil}
		}
	}

	return evictionResults, errors.NewAutoscalerError(errors.TransientError, "Failed to drain node %s/%s: pods remaining after timeout", node.Namespace, node.Name)
}

// Removes the given node from cloud provider. No extra pre-deletion actions are executed on
// the Kubernetes side.
func deleteNodeFromCloudProvider(node *apiv1.Node, cloudProvider cloudprovider.CloudProvider,
	recorder kube_record.EventRecorder, registry *clusterstate.ClusterStateRegistry) errors.AutoscalerError {
	nodeGroup, err := cloudProvider.NodeGroupForNode(node)
	if err != nil {
		return errors.NewAutoscalerError(
			errors.CloudProviderError, "failed to find node group for %s: %v", node.Name, err)
	}
	if nodeGroup == nil || reflect.ValueOf(nodeGroup).IsNil() {
		return errors.NewAutoscalerError(errors.InternalError, "picked node that doesn't belong to a node group: %s", node.Name)
	}
	if err = nodeGroup.DeleteNodes([]*apiv1.Node{node}); err != nil {
		return errors.NewAutoscalerError(errors.CloudProviderError, "failed to delete %s: %v", node.Name, err)
	}
	recorder.Eventf(node, apiv1.EventTypeNormal, "ScaleDown", "node removed by cluster autoscaler")
	registry.RegisterScaleDown(&clusterstate.ScaleDownRequest{
		NodeGroup:          nodeGroup,
		NodeName:           node.Name,
		Time:               time.Now(),
		ExpectedDeleteTime: time.Now().Add(MaxCloudProviderNodeDeletionTime),
	})
	return nil
}

func waitForDelayDeletion(node *apiv1.Node, nodeLister kubernetes.NodeLister, timeout time.Duration) errors.AutoscalerError {
	if timeout != 0 && hasDelayDeletionAnnotation(node) {
		klog.V(1).Infof("Wait for removing %s annotations on node %v", DelayDeletionAnnotationPrefix, node.Name)
		err := wait.Poll(5*time.Second, timeout, func() (bool, error) {
			klog.V(5).Infof("Waiting for removing %s annotations on node %v", DelayDeletionAnnotationPrefix, node.Name)
			freshNode, err := nodeLister.Get(node.Name)
			if err != nil || freshNode == nil {
				return false, fmt.Errorf("failed to get node %v: %v", node.Name, err)
			}
			return !hasDelayDeletionAnnotation(freshNode), nil
		})
		if err != nil && err != wait.ErrWaitTimeout {
			return errors.ToAutoscalerError(errors.ApiCallError, err)
		}
		if err == wait.ErrWaitTimeout {
			klog.Warningf("Delay node deletion timed out for node %v, delay deletion annotation wasn't removed within %v, this might slow down scale down.", node.Name, timeout)
		} else {
			klog.V(2).Infof("Annotation %s removed from node %v", DelayDeletionAnnotationPrefix, node.Name)
		}
	}
	return nil
}

func hasDelayDeletionAnnotation(node *apiv1.Node) bool {
	for annotation := range node.Annotations {
		if strings.HasPrefix(annotation, DelayDeletionAnnotationPrefix) {
			return true
		}
	}
	return false
}

func hasNoScaleDownAnnotation(node *apiv1.Node) bool {
	return node.Annotations[ScaleDownDisabledKey] == "true"
}

const (
	apiServerLabelKey   = "component"
	apiServerLabelValue = "kube-apiserver"
)

func isMasterNode(nodeInfo *schedulerframework.NodeInfo) bool {
	for _, podInfo := range nodeInfo.Pods {
		if podInfo.Pod.Namespace == metav1.NamespaceSystem && podInfo.Pod.Labels[apiServerLabelKey] == apiServerLabelValue {
			return true
		}
	}
	return false
}

func filterOutMasters(nodeInfos []*schedulerframework.NodeInfo) []*apiv1.Node {
	result := make([]*apiv1.Node, 0, len(nodeInfos))
	for _, nodeInfo := range nodeInfos {
		if !isMasterNode(nodeInfo) {
			result = append(result, nodeInfo.Node())
		}
	}
	return result
}


text:
export default {
  transitionDuration: 300,
  /**
   * 触发一个事件
   * mouseenter, mouseleave, mouseover, keyup, change, click 等
   * @param  {Element} elm
   * @param  {String} name
   * @param  {*} opts
   */
  triggerEvent: function (elm, name, ...opts) {
    let eventName
    if (/^mouse|click/.test(name)) {
      eventName = 'MouseEvents'
    } else if (/^key/.test(name)) {
      eventName = 'KeyboardEvent'
    } else {
      eventName = 'HTMLEvents'
    }
    const evt = document.createEvent(eventName)
    evt.initEvent(name, ...opts)
    elm.dispatchEvent
      ? elm.dispatchEvent(evt)
      : elm.fireEvent('on' + name, evt)
    return elm
  }
}

redpajama_set_name: RedPajamaGithub


text:
package com.amazonaws.services.codepipeline.model.transform;

import java.util.Map;
import java.util.Map.Entry;

import com.amazonaws.services.codepipeline.model.*;
import com.amazonaws.transform.SimpleTypeJsonUnmarshallers.*;
import com.amazonaws.transform.*;

import com.fasterxml.jackson.core.JsonToken;
import static com.fasterxml.jackson.core.JsonToken.*;

/**
 * OutputArtifact JSON Unmarshaller
 */
public class OutputArtifactJsonUnmarshaller implements
        Unmarshaller<OutputArtifact, JsonUnmarshallerContext> {

    public OutputArtifact unmarshall(JsonUnmarshallerContext context)
            throws Exception {
        OutputArtifact outputArtifact = new OutputArtifact();

        int originalDepth = context.getCurrentDepth();
        String currentParentElement = context.getCurrentParentElement();
        int targetDepth = originalDepth + 1;

        JsonToken token = context.getCurrentToken();
        if (token == null)
            token = context.nextToken();
        if (token == VALUE_NULL)
            return null;

        while (true) {
            if (token == null)
                break;

            if (token == FIELD_NAME || token == START_OBJECT) {
                if (context.testExpression("name", targetDepth)) {
                    context.nextToken();
                    outputArtifact.setName(StringJsonUnmarshaller.getInstance()
                            .unmarshall(context));
                }
            } else if (token == END_ARRAY || token == END_OBJECT) {
                if (context.getLastParsedParentElement() == null
                        || context.getLastParsedParentElement().equals(
                                currentParentElement)) {
                    if (context.getCurrentDepth() <= originalDepth)
                        break;
                }
            }
            token = context.nextToken();
        }

        return outputArtifact;
    }

    private static OutputArtifactJsonUnmarshaller instance;

    public static OutputArtifactJsonUnmarshaller getInstance() {
        if (instance == null)
            instance = new OutputArtifactJsonUnmarshaller();
        return instance;
    }
}


text:
package com.model;

import org.hibernate.annotations.OnDelete;
import org.hibernate.annotations.OnDeleteAction;

import javax.persistence.*;
import java.util.Date;

/**
 * Created by Shreya on 7/18/2017.
 */
@Entity
public class Resource {
    private Integer resourceId;
    private String description;
    private User createdBy;
    private Topic topic;
    private Date dateCreated;
    private Date lastUpdated;

    public Resource() {
    }

    public Resource(String description, User createdBy, Topic topic, Date dateCreated, Date lastUpdated) {

        this.description = description;
        this.createdBy = createdBy;
        this.topic = topic;
        this.dateCreated = dateCreated;
        this.lastUpdated = lastUpdated;
    }

    @Id
    @GeneratedValue(strategy = GenerationType.TABLE)
    @Column(name = "resource_id")
    public Integer getResourceId() {
        return resourceId;
    }

    public void setResourceId(Integer resourceId) {
        this.resourceId = resourceId;
    }

    @Column(name = "description")
    public String getDescription() {
        return description;
    }

    public void setDescription(String description) {
        this.description = description;
    }

    @ManyToOne(targetEntity = User.class)
    public User getCreatedBy() {
        return createdBy;
    }

    public void setCreatedBy(User createdBy) {
        this.createdBy = createdBy;
    }

    @ManyToOne(targetEntity = Topic.class)
    @OnDelete(action = OnDeleteAction.CASCADE)
    public Topic getTopic() {
        return topic;
    }

    public void setTopic(Topic topic) {
        this.topic = topic;
    }

    @Temporal(TemporalType.DATE)
    @Column(name = "date_created")
    public Date getDateCreated() {
        return dateCreated;
    }

    public void setDateCreated(Date dateCreated) {
        this.dateCreated = dateCreated;
    }

    @Temporal(TemporalType.DATE)
    @Column(name = "last_updated")
    public Date getLastUpdated() {
        return lastUpdated;
    }

    public void setLastUpdated(Date lastUpdated) {
        this.lastUpdated = lastUpdated;
    }

    @Override
    public String toString() {
        return "Resource : " + topic + " " + description;
    }
}


text:
Create Comment

<form method="post" action="<?php echo $action ?>">
Subject

<input type="text" name="subject" />

Body

<textarea name="body"></textarea>

<input type="submit" />
</form>


text:
package org.apache.cassandra.gms;

import java.io.*;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.cassandra.io.IVersionedSerializer;

/**
 * This message gets sent out as a result of the receipt of a GossipDigestSynMessage by an
 * endpoint. This is the 2 stage of the 3 way messaging in the Gossip protocol.
 */

class GossipDigestAckMessage
{
    private static IVersionedSerializer<GossipDigestAckMessage> serializer_;
    static
    {
        serializer_ = new GossipDigestAckMessageSerializer();
    }

    List<GossipDigest> gDigestList_ = new ArrayList<GossipDigest>();
    Map<InetAddress, EndpointState> epStateMap_ = new HashMap<InetAddress, EndpointState>();

    static IVersionedSerializer<GossipDigestAckMessage> serializer()
    {
        return serializer_;
    }

    GossipDigestAckMessage(List<GossipDigest> gDigestList, Map<InetAddress, EndpointState> epStateMap)
    {
        gDigestList_ = gDigestList;
        epStateMap_ = epStateMap;
    }

    List<GossipDigest> getGossipDigestList()
    {
        return gDigestList_;
    }

    Map<InetAddress, EndpointState> getEndpointStateMap()
    {
        return epStateMap_;
    }
}

class GossipDigestAckMessageSerializer implements IVersionedSerializer<GossipDigestAckMessage>
{
    public void serialize(GossipDigestAckMessage gDigestAckMessage, DataOutput dos, int version) throws IOException
    {
        GossipDigestSerializationHelper.serialize(gDigestAckMessage.gDigestList_, dos, version);
        dos.writeBoolean(true); // 0.6 compatibility
        EndpointStatesSerializationHelper.serialize(gDigestAckMessage.epStateMap_, dos, version);
    }

    public GossipDigestAckMessage deserialize(DataInput dis, int version) throws IOException
    {
        List<GossipDigest> gDigestList = GossipDigestSerializationHelper.deserialize(dis, version);
        dis.readBoolean(); // 0.6 compatibility
        Map<InetAddress, EndpointState> epStateMap = EndpointStatesSerializationHelper.deserialize(dis, version);
        return new GossipDigestAckMessage(gDigestList, epStateMap);
    }

    public long serializedSize(GossipDigestAckMessage gossipDigestAckMessage, int version)
    {
        throw new UnsupportedOperationException();
    }
}


text:
namespace autofill_assistant {

// Action to edit password. It is only possible if Userdata contains selected
// login.
class EditPasswordAction : public Action {
 public:
  explicit EditPasswordAction(ActionDelegate* delegate,
                              const ActionProto& proto);
  ~EditPasswordAction() override;

  EditPasswordAction(const EditPasswordAction&) = delete;
  EditPasswordAction& operator=(const EditPasswordAction&) = delete;

 private:
  // Overrides Action:
  void InternalProcessAction(ProcessActionCallback callback) override;

  void OnPasswordEdited(bool success);
  void EndAction(const ClientStatus& status);

  ProcessActionCallback callback_;
  base::WeakPtrFactory<EditPasswordAction> weak_ptr_factory_{this};
};

}  // namespace autofill_assistant
#endif  // COMPONENTS_AUTOFILL_ASSISTANT_BROWSER_ACTIONS_EDIT_PASSWORD_ACTION_H_


text:
import ast
import urlparse

from scrapy import log
from scrapy.selector import Selector
from scrapy.http import Request
from scrapy.contrib.loader import ItemLoader
from scrapy.contrib.loader.processor import TakeFirst
from scrapy.exceptions import CloseSpider

from dynamic_scraper.spiders.django_base_spider import DjangoBaseSpider
from dynamic_scraper.models import ScraperElem
from dynamic_scraper.utils.scheduler import Scheduler
from dynamic_scraper.utils import processors

class DjangoSpider(DjangoBaseSpider):

    def __init__(self, *args, **kwargs):
        self.mandatory_vars.append('scraped_obj_class')
        self.mandatory_vars.append('scraped_obj_item_class')

        super(DjangoSpider, self).__init__(self, *args, **kwargs)
        self._set_config(**kwargs)
        self._check_scraper_config()

        self._set_start_urls(self.scrape_url)
        self.scheduler = Scheduler(self.scraper.scraped_obj_class.scraper_scheduler_conf)
        self.from_detail_page = False
        self.loader = None
        self.items_read_count = 0
        self.items_save_count = 0

        msg = "Spider for " + self.ref_object.__class__.__name__ + " \"" + str(self.ref_object) + "\" (" + str(self.ref_object.pk) + ") initialized."
        self.log(msg, log.INFO)

    def _set_config(self, **kwargs):
        log_msg = ""
        #max_items_read
        if 'max_items_read' in kwargs:
            try:
                self.conf['MAX_ITEMS_READ'] = int(kwargs['max_items_read'])
            except ValueError:
                raise CloseSpider("You have to provide an integer value as max_items_read parameter!")
            if len(log_msg) > 0:
                log_msg += ", "
            log_msg += "max_items_read " + str(self.conf['MAX_ITEMS_READ'])
        else:
            self.conf['MAX_ITEMS_READ'] = self.scraper.max_items_read
        #max_items_save
        if 'max_items_save' in kwargs:
            try:
                self.conf['MAX_ITEMS_SAVE'] = int(kwargs['max_items_save'])
            except ValueError:
                raise CloseSpider("You have to provide an integer value as max_items_save parameter!")
            if len(log_msg) > 0:
                log_msg += ", "
            log_msg += "max_items_save " + str(self.conf['MAX_ITEMS_SAVE'])
        else:
            self.conf['MAX_ITEMS_SAVE'] = self.scraper.max_items_save

        super(DjangoSpider, self)._set_config(log_msg, **kwargs)

    def _check_scraper_config(self):
        try:
            self.scraper.get_base_elem()
        except ScraperElem.DoesNotExist:
            raise CloseSpider('Please define a base scraper elem in your database!')
        try:
            self.scraper.get_detail_page_url_elem()
        except ScraperElem.DoesNotExist:
            raise CloseSpider('Please define a detail page url scraper elem in your database!')
        if(len(self.scraper.get_base_elems()) > 1):
            raise CloseSpider('A scraper can\'t have more than one base scraper elem!')
        if(len(self.scraper.get_detail_page_url_elems()) > 1):
            raise CloseSpider('A scraper can\'t have more than one detail page url scraper elem!')

    def _set_start_urls(self, scrape_url):

        if self.scraper.pagination_type != 'N':
            if not self.scraper.pagination_append_str:
                raise CloseSpider('Please provide a pagination_append_str for pagination (e.g. "/archive/{page}/")!')
            if self.scraper.pagination_append_str.find('{page}') == -1:
                raise CloseSpider('Pagination_append_str has to contain "{page}" as placeholder for page replace!')
            if not self.scraper.pagination_page_replace:
                raise CloseSpider('Please provide a pagination_page_replace context corresponding to pagination_type!')

        if self.scraper.pagination_type == 'R':
            try:
                pages = self.scraper.pagination_page_replace
                pages = pages.split(',')
                if len(pages) > 3:
                    raise Exception
                pages = range(*map(int, pages))
            except Exception:
                raise CloseSpider('Pagination_page_replace for pagination_type "RANGE_FUNCT" ' +\
                                  'has to be provided as python range function arguments ' +\
                                  '[start], stop[, step] (e.g. "1, 50, 10", no brackets)!')

        if self.scraper.pagination_type == 'F':
            try:
                pages = self.scraper.pagination_page_replace
                pages = pages.strip(', ')
                pages = ast.literal_eval("[" + pages + ",]")
            except SyntaxError:
                raise CloseSpider('Wrong pagination_page_replace format for pagination_type "FREE_LIST", ' +\
                                  "Syntax: 'Replace string 1', 'Another replace string 2', 'A number 3', ...")

        if self.scraper.pagination_type != 'N':
            append_str = self.scraper.pagination_append_str
            if scrape_url[-1:] == '/' and append_str[0:1] == '/':
                append_str = append_str[1:]

            for page in pages:
                url = scrape_url + append_str.format(page=page)
                self.start_urls.append(url)
            if not self.scraper.pagination_on_start:
                self.start_urls.append(scrape_url)

        if self.scraper.pagination_type == 'N':
            self.start_urls.append(scrape_url)

    def _set_loader_context(self, context_str):
        try:
            context_str = context_str.strip(', ')
            context = ast.literal_eval("{" + context_str + "}")
            context['spider'] = self
            self.loader.context = context
        except SyntaxError:
            self.log("Wrong context definition format: " + context_str, log.ERROR)

    def _get_processors(self, procs_str):
        procs = [TakeFirst(), processors.string_strip,]
        if not procs_str:
            return procs
        procs_tmp = list(procs_str.split(','))
        for p in procs_tmp:
            p = p.strip()
            if hasattr(processors, p):
                procs.append(getattr(processors, p))
            else:
                self.log("Processor '%s' is not defined!" % p, log.ERROR)
        procs = tuple(procs)
        return procs

    def _scrape_item_attr(self, scraper_elem):
        if(self.from_detail_page == scraper_elem.from_detail_page):
            procs = self._get_processors(scraper_elem.processors)
            self._set_loader_context(scraper_elem.proc_ctxt)

            static_ctxt = self.loader.context.get('static', '')
            if processors.static in procs and static_ctxt:
                self.loader.add_value(scraper_elem.scraped_obj_attr.name, static_ctxt)
            elif(scraper_elem.reg_exp):
                self.loader.add_xpath(scraper_elem.scraped_obj_attr.name, scraper_elem.x_path, *procs,  re=scraper_elem.reg_exp)
            else:
                self.loader.add_xpath(scraper_elem.scraped_obj_attr.name, scraper_elem.x_path, *procs)
            msg  = '{0: <20}'.format(scraper_elem.scraped_obj_attr.name)
            c_values = self.loader.get_collected_values(scraper_elem.scraped_obj_attr.name)
            if len(c_values) > 0:
                msg += "'" + c_values[0] + "'"
            else:
                msg += u'None'
            self.log(msg, log.DEBUG)

    def _set_loader(self, response, xs, item):
        if not xs:
            self.from_detail_page = True
            item = response.request.meta['item']
            self.loader = ItemLoader(item=item, response=response)
            self.loader.default_output_processor = TakeFirst()
        else:
            self.from_detail_page = False
            self.loader = ItemLoader(item=item, selector=xs)
            self.loader.default_output_processor = TakeFirst()

    def parse_item(self, response, xs=None):
        self._set_loader(response, xs, self.scraped_obj_item_class())
        if not self.from_detail_page:
            self.items_read_count += 1

        elems = self.scraper.get_scrape_elems()

        for elem in elems:
            self._scrape_item_attr(elem)
        # Dealing with Django Char- and TextFields defining blank field as null
        item = self.loader.load_item()
        for key, value in item.items():
            if value == None and \
               self.scraped_obj_class()._meta.get_field(key).blank and \
               not self.scraped_obj_class()._meta.get_field(key).null:
                item[key] = ''

        return item

    def parse(self, response):
        """
        Parsing the page and performing other actions
        """
        #xs = Selector(response) #deprecated, use response.xpath
        base_elem = self.scraper.get_base_elem()
        url_elem = self.scraper.get_detail_page_url_elem()
        base_objects = response.xpath(base_elem.x_path)

        if(len(base_objects) == 0):
            self.log("No base objects found!", log.ERROR)

        if(self.conf['MAX_ITEMS_READ']):
            items_left = min(len(base_objects), self.conf['MAX_ITEMS_READ'] - self.items_read_count)
            base_objects = base_objects[0:items_left]

        for obj in base_objects:
            item_num = self.items_read_count + 1
            self.log("Starting to crawl item %s." % str(item_num), log.INFO)
            item = self.parse_item(response, obj)
            url_name = url_elem.scraped_obj_attr.name
            if(item and url_name in item):
                url = item[url_name]
                cnt = self.scraped_obj_class.objects.filter(url=item[url_name]).count()
                cnt1 = self.scraper.get_standard_update_elems_from_detail_page().count()
                cnt2 = self.scraper.get_from_detail_page_scrape_elems().count()
                # Mark item as DOUBLE item
                if cnt > 0:
                    item[url_name] = 'DOUBLE' + item[url_name]
                # (DOUBLE item with no standard update elements to be scraped from detail page) or
                # generally no attributes scraped from detail page
                if (cnt > 0 and cnt1 == 0) or cnt2 == 0:
                    yield item
                else:
                    yield Request(urlparse.urljoin(response.url,url), callback=self.parse_item, meta={'item':item})
            else:
                self.log("Detail page url elem could not be read!", log.ERROR)


text:
using System;
using System.Collections.Generic;
using System.Linq;
using PushNotifications.Contracts;
using PushNotifications.Contracts.PushNotifications;
using PushNotifications.Contracts.PushNotifications.Delivery;
using PushNotifications.Delivery.FireBase.Logging;
using PushNotifications.Delivery.FireBase.Models;
using RestSharp;
using RestSharp.Serializers;
using static PushNotifications.Delivery.FireBase.Models.FireBaseResponseModel;

namespace PushNotifications.Delivery.FireBase
{
    public sealed class FireBaseDelivery : IPushNotificationDelivery
    {
        static ILog log = LogProvider.GetLogger(typeof(FireBaseDelivery));

        readonly string serverKey;

        readonly IRestClient restClient;

        readonly ISerializer serializer;

        IPushNotificationAggregator pushNotificationAggregator;

        public FireBaseDelivery(IRestClient restClient, ISerializer serializer, string serverKey)
        {
            if (ReferenceEquals(null, restClient) == true) throw new ArgumentNullException(nameof(restClient));
            if (ReferenceEquals(null, serializer) == true) throw new ArgumentNullException(nameof(serializer));
            if (string.IsNullOrEmpty(serverKey) == true) throw new ArgumentNullException(nameof(serverKey));

            this.restClient = restClient;
            this.serializer = serializer;
            this.serverKey = serverKey;
        }

        public void UseAggregator(IPushNotificationAggregator pushNotificationAggregator)
        {
            if (ReferenceEquals(null, pushNotificationAggregator) == true) throw new ArgumentNullException(nameof(pushNotificationAggregator));
            this.pushNotificationAggregator = pushNotificationAggregator;
        }

        public SendTokensResult Send(SubscriptionToken token, NotificationForDelivery notification)
        {
            if (ReferenceEquals(null, token) == true) throw new ArgumentNullException(nameof(token));
            if (ReferenceEquals(null, notification) == true) throw new ArgumentNullException(nameof(notification));

            if (ReferenceEquals(null, pushNotificationAggregator) == false)
                return pushNotificationAggregator.Queue(token, notification);

            return Send(new List<SubscriptionToken> { token }, notification);
        }

        public SendTokensResult Send(IList<SubscriptionToken> tokens, NotificationForDelivery notification)
        {
            if (ReferenceEquals(null, tokens) == true) throw new ArgumentNullException(nameof(tokens));
            if (ReferenceEquals(null, notification) == true) throw new ArgumentNullException(nameof(notification));
            if (tokens.Count == 0) throw new ArgumentException("Tokens are missing");

            const string resource = "fcm/send";

            log.Debug(() => $"[FireBase] sending '{tokens.Count}' PN for notification '{notification.Id}' with body '{notification.NotificationPayload.Body}'");

            List<string> tokensAsStrings = tokens.Select(x => x.ToString()).ToList();
            NotificationPayload payload = notification.NotificationPayload;
            Dictionary<string, object> data = notification.NotificationData;
            string badge = payload.Badge > 0 ? payload.Badge.ToString() : "1";
            var fireBaseSendNotificationModel = new FireBaseSendNotificationModel(payload.Title, payload.Body, payload.Sound, badge);
            var model = new FireBaseSendModel(tokensAsStrings, fireBaseSendNotificationModel, data, notification.ExpiresAt);
            IRestRequest request = CreateRestRequest(resource, Method.POST).AddJsonBody(model);
            IRestResponse<FireBaseResponseModel> response = restClient.Execute<FireBaseResponseModel>(request);

            if (response.StatusCode == System.Net.HttpStatusCode.OK)
            {
                if (response.HasDataFailure())
                {
                    List<FireBaseResponseResultModel> firebaseResponseModel = response.Data.Results;
                    return ExpiredTokensDetector.GetNotRegisteredTokens(tokens, firebaseResponseModel);
                }

                log.Info($"[FireBase] success: PN with body {notification.NotificationPayload?.Body} was sent to {tokens.Count} tokens");
                return SendTokensResult.Success;
            }
            else
            {
                response.LogFireBaseError(() => $"[FireBase] failure: status code '{response.StatusCode}'. PN body '{notification.NotificationPayload.Body}'");
                return SendTokensResult.Failed;
            }
        }

        public bool SendToTopic(Topic topic, NotificationForDelivery notification)
        {
            if (ReferenceEquals(null, topic) == true) throw new ArgumentNullException(nameof(topic));
            if (ReferenceEquals(null, notification) == true) throw new ArgumentNullException(nameof(notification));

            const string resource = "fcm/send";

            log.Debug(() => $"[FireBase] sending PN to topic: {topic} for notification '{notification.Id}' with body '{notification.NotificationPayload.Body}'");
            var payload = notification.NotificationPayload;
            var data = notification.NotificationData;
            string badge = payload.Badge > 0 ? payload.Badge.ToString() : "1";

            var fireBaseSendNotificationModel = new FireBaseSendNotificationModel(payload.Title, payload.Body, payload.Sound, badge);
            var model = new FireBaseTopicSendModel(topic, fireBaseSendNotificationModel, data, notification.ExpiresAt);

            var request = CreateRestRequest(resource, Method.POST).AddJsonBody(model);
            var result = restClient.Execute<FireBaseResponseModel>(request);

            if (result.StatusCode != System.Net.HttpStatusCode.OK || result.HasDataFailure())
            {
                result.LogFireBaseError(() => $"[FireBase] failure: status code '{result.StatusCode}'. PN body '{notification.NotificationPayload.Body}'");
                return false;
            }

            log.Info($"[FireBase] success: PN with body {notification.NotificationPayload?.Body} was sent to {topic} topic");
            return true;
        }

        IRestRequest CreateRestRequest(string resource, Method method)
        {
            var request = new RestRequest(resource, method);
            request.RequestFormat = DataFormat.Json;
            request.AddHeader("Authorization", $"key={serverKey}");
            request.JsonSerializer = serializer;

            return request;
        }
    }
}


text:
require 'json'

COHORT_FILE = ARGV[0]
DONT_PAIR_FILE = ARGV[1]

USERS     = JSON.parse(File.read(COHORT_FILE))
DONT_PAIR = JSON.parse(File.read(DONT_PAIR_FILE)) if DONT_PAIR_FILE

def make_pair pairs
  user = pairs.keys.shuffle.first
  previous_pairs = pairs.delete(user)
  available_pairs = pairs.keys - previous_pairs
  if DONT_PAIR.include? user
    available_pairs -= DONT_PAIR
  end

  if available_pairs.empty?
    available_pairs = pairs.keys
    USERS[user] = []
  end

  if pair = available_pairs.shuffle.first
    pairs.delete(pair)

    USERS[user] << pair
    USERS[pair] << user

    "Pair: #{user}, #{pair}"
  else
    "Solo: #{user}"
  end
end

to_pair = USERS.clone
puts make_pair(to_pair) until to_pair.empty?

File.open(COHORT_FILE, 'w') { |file| file.write JSON.generate(USERS) }


text:
package msdingfield.easyflow.graph;

import static msdingfield.easyflow.graph.DiamondOperations.a;
import static msdingfield.easyflow.graph.DiamondOperations.b1;
import static msdingfield.easyflow.graph.DiamondOperations.b2;
import static msdingfield.easyflow.graph.DiamondOperations.c;
import static org.junit.Assert.assertEquals;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executor;

import msdingfield.easyflow.execution.Task;
import msdingfield.easyflow.graph.Graph;
import msdingfield.easyflow.graph.GraphTaskBuilder;
import msdingfield.easyflow.graph.TaskFactory;
import msdingfield.easyflow.reflect.Context;

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

@RunWith(Parameterized.class)
public class GraphExecutionTest {

	public static class Params {
		public Set<AddOperation> ops = null;

		public String inputKey = null;
		public Integer inputValue = null;

		public Integer aValue = null;
		public Integer b1Value = null;
		public Integer b2Value = null;
		public Integer cValue = null;
	}

	@Parameters
	public static ArrayList<Params[]> getParamStructs() {
		ArrayList<Params[]> paramArray = Lists.newArrayList();
		for (final Params test : tests) {
			paramArray.add(new Params[] {test});
		}
		return paramArray;
	}

	public GraphExecutionTest(final Params param) {
		params = param;
	}

	private Params params = null;

	public static Params[] tests = {

		// Empty System
		new Params(){{
			ops = Sets.newHashSet();
			inputKey = "a";
			inputValue = 99;
			aValue = 99;
		}},

		// One Operation
		new Params(){{
			ops = Sets.newHashSet(a);
			inputKey = "input";
			inputValue = 1;
			aValue = 3;
		}},

		// Two Operations Linear
		new Params(){{
			ops = Sets.newHashSet(a, b1);
			inputKey = "input";
			inputValue = 1;
			aValue = 3;
			b1Value = 7;
		}},

		// Two Operations Independent
		new Params(){{
			ops = Sets.newHashSet(b2, b1);
			inputKey = "a";
			inputValue = 1;
			aValue = 1;
			b1Value = 5;
			b2Value = 9;
		}},

		// Fork
		new Params(){{
			ops = Sets.newHashSet(a, b2, b1);
			inputKey = "input";
			inputValue = 1;
			aValue = 3;
			b1Value = 7;
			b2Value = 11;
		}},

		// Join
		new Params(){{
			ops = Sets.newHashSet(c, b2, b1);
			inputKey = "a";
			inputValue = 1;
			aValue = 1;
			b1Value = 5;
			b2Value = 9;
			cValue = 30;
		}},

		// Diamond
		new Params(){{
			ops = Sets.newHashSet(a, c, b2, b1);
			inputKey = "input";
			inputValue = 1;
			aValue = 3;
			b1Value = 7;
			b2Value = 11;
			cValue = 34;
		}}
	};

	private Context context = new Context();

	@Before
	public void setup() {
		context = new Context();
	}

	@Test
	public void test() throws InterruptedException {
		final Graph<AddOperation> graph = new Graph<AddOperation>(params.ops);

		context.setEdgeValue(params.inputKey, params.inputValue);
		GraphTaskBuilder
		.graph(graph)
		.taskFactory(AddOperation.getTaskFactory(context))
		.build()
		.schedule().join();

		assertEquals(params.aValue, context.getEdgeValue("a"));
		assertEquals(params.b1Value, context.getEdgeValue("b.1"));
		assertEquals(params.b2Value, context.getEdgeValue("b.2"));
		assertEquals(params.cValue, context.getEdgeValue("c"));
	}

	public static final class AddOperationTaskFactory implements TaskFactory<AddOperation> {
		private Context context;
		public AddOperationTaskFactory(final Context context) {
			this.context = context;
		}

		@Override
		public Task create(final Executor executor, final AddOperation op) {
			final Task task = new Task(executor);
			task.addWorker(new Runnable(){
				@Override public void run() {
					op.execute(context);
				}});
			return task;
		}

	}
}


text:
function KeepFilter=FilterMap(CADS_object,varargin)
%UNTITLED Summary of this function goes here
%   Detailed explanation goes here
VarCutoff=Inf;
numTiles=1;
ResidueTargetMode=false();
try
    contact_cutoff=CADS_object(1).Settings.Structure.ContactCutoff;
catch
    contact_cutoff=3.4;
end
AtomFilterType='anything';
% InterActionFilter='any';

if nargin >1
    for ind=1:length(varargin)/2
        switch varargin{2*ind-1}
            case 'numBins'
                numBins=varargin{2*ind};
            case 'VarCutoff'
                VarCutoff=varargin{2*ind};
            case 'FilterRes'
                FilterRes=varargin{2*ind};
            case 'numTiles'
                numTiles=varargin{2*ind};
            case 'Subsets'
                Subsets=varargin{2*ind};
            case 'ContactCutoff'
                contact_cutoff=varargin{2*ind};
            case 'AtomFilterType'
                AtomFilterType=varargin{2*ind};
%             case 'InterActionFilter'
%                 InterActionFilter=varargin{2*ind};
            case 'ResidueTarget'
                ResidueTarget=varargin{2*ind};
                ResidueTargetMode=true;
        end
    end
end

if exist('numBins','var')
    [~, Bounds] = cSequenceColor(CADS_object,varargin{:});
    if numTiles<numBins
        VarCutoff=Bounds(numTiles);
    else
        warning('numTiles was too high, ignoring')
    end
end
%
% if exist('Subsets','var')
%     subsetsmode=true;
% else
%     subsetsmode=false;
% end
numDataSets=length(CADS_object);
ChosenSubsets=cell(numDataSets,1);
MapUC=cell(numDataSets,1);
KeepFilter=cell(numDataSets,1);
for Pro_ind=1:numDataSets
    if exist('Subsets','var') && ~isempty(Subsets) && ~isempty(Subsets{Pro_ind})
        subsetsmode=true;
        ChosenSubsets{Pro_ind}=Subsets{Pro_ind};
        numSubsets=length(ChosenSubsets{Pro_ind});
    elseif ~exist('Subsets','var') && ~isempty(CADS_object(Pro_ind).Subsets)
        subsetsmode=true;
        ChosenSubsets{Pro_ind}=CADS_object(Pro_ind).Subsets;
        numSubsets=length(ChosenSubsets{Pro_ind});
    else
        subsetsmode=false;
        numSubsets=1;
    end
%     names=CADS_object(Pro_ind).Species;
    %     CAOut(length(names))=ConAnalOutput();
    %     MapUC=cell(1,length(names));

    for model=1:length(CADS_object(Pro_ind).PDB)
        %         pdb1 = PDBentry([names{model},'Subject']);
        %         pdb2 = PDBentry([names{model},'Target']);
        %         pdb1.PDBfromStruct(CADS_object(Pro_ind).PDB(model),'newID1')
        %         pdb2.PDBfromStruct(Target(model),'newID2')
        if exist('FilterRes','var')
            KeepRes=logical(sum(vertcat(FilterRes{Pro_ind}{:}),1));
            VarCutoff=FilterRes{Pro_ind};
        else
            if isnumeric(CADS_object(Pro_ind).Results(model).Variability)...
                    && ~isempty(CADS_object(Pro_ind).Results(model).Variability)
                KeepRes=CADS_object(Pro_ind).Results(model).Variability<=VarCutoff;
            else
                KeepRes=true(length(CADS_object(Pro_ind).Keep(model).PDB),1);
            end
        end

        for subset=1:numSubsets
            if subsetsmode
                KeepRes2=false(size(KeepRes));
                KeepRes2(ChosenSubsets{Pro_ind}{subset})=true;
                KeepRes3=KeepRes&KeepRes2;
            else
                KeepRes3=KeepRes;
            end

            if sum(KeepRes3)
                if ResidueTargetMode
                    [MapUC{Pro_ind}{model}{subset}]=MapContacts(CADS_object(Pro_ind).PDB(model),...
                        ResidueTarget{model},[],contact_cutoff,...
                        CADS_object(Pro_ind).Keep(model).Alignment(KeepRes3),...
                        'AtomFilterType',AtomFilterType,varargin{:});
                else
                    [MapUC{Pro_ind}{model}{subset}]=MapContacts(CADS_object(Pro_ind).PDB(model),...
                        CADS_object(Pro_ind).Target(model),[],contact_cutoff,...
                        CADS_object(Pro_ind).Keep(model).Alignment(KeepRes3),...
                        'AtomFilterType',AtomFilterType,varargin{:});
                end
            else
                MapUC{Pro_ind}{model}{subset}=struct('contacts',{{}},'residues',{{}},'Y',{{}});
%                 KeepFilter{Pro_ind}{model}{subset}=[];
            end
            if ~isempty(MapUC{Pro_ind}{model}{subset}.Y)
                [~,Keep]=ismember([MapUC{Pro_ind}{model}{subset}.Y{:,1}],...
                CADS_object(Pro_ind).Keep(model).Alignment);
            else
                Keep=0;
            end

            KeepRes4=false(size(KeepRes));
            if sum(Keep)>0
                KeepRes4(Keep)=true;
            end
            KeepFilter{Pro_ind}{model}{subset}=KeepRes4;

        end
        CADS_object(Pro_ind).Results(model).AddContactMapFiltered([MapUC{Pro_ind}{model}{:}],{VarCutoff});
    end
end
end


text:
package concurrent;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author yujianjian  2017-08-03 上午8:51
 */
public class TheadPoolDemo {
    public static void main(String[] args) {
        ThreadFactory threadFactory = new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return null;
            }
        };
        RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();

        ExecutorService exec = new ThreadPoolExecutor(50, 50, 0L,
                                                      TimeUnit.MILLISECONDS,
                                                      new ArrayBlockingQueue<>(1000),
                                                      threadFactory,rejectedExecutionHandler);
    }
}