pcm-coordinator/pkg/scheduler/algo/k8sStrategy.go

/*

 Copyright (c) [2023] [pcm]
 [pcm-coordinator] is licensed under Mulan PSL v2.
 You can use this software according to the terms and conditions of the Mulan PSL v2.
 You may obtain a copy of Mulan PSL v2 at:
          http://license.coscl.org.cn/MulanPSL2
 THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 EITHER EXPaRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details.

*/

package algo

import (
	"errors"
	"gonum.org/v1/gonum/mat"
	"math"
)

type k8sStrategy struct {
	ProviderList []*Provider
	Task         *Task
	StrategyList []*Strategy
}

func NewK8sStrategy(task *Task, providers ...*Provider) *k8sStrategy {
	var providerList []*Provider
	var res [][]int

	for _, p := range providers {
		p.GenMaxResourceNum(task)
		providerList = append(providerList, p)
	}

	back_trace_task(task.Replicas, 0, providerList, 0, &res, 0)

	var strategyList []*Strategy
	for _, r := range res {
		var path []int
		var pathlist [][]int

		var resourcePerProvider []int
		for j, p := range providerList {
			if r[j] > p.MaxReplicas {
				resourcePerProvider = append(resourcePerProvider, p.MaxReplicas)
			} else {
				resourcePerProvider = append(resourcePerProvider, r[j])
			}
		}
		back_trace_resource(resourcePerProvider, 0, path, &pathlist)
		strategy := NewStrategy()
		strategy.Tasksolution = r
		strategy.Resourcesolution = pathlist
		strategyList = append(strategyList, strategy)
	}

	return &k8sStrategy{ProviderList: providerList, Task: task, StrategyList: strategyList}
}

func (ps *k8sStrategy) computeMaxScore() (*Task, error) {
	maxStrategy := NewStrategy()
	var maxprofit float64

	//先计算出最大的利润值
	for _, strategy := range ps.StrategyList {
		for _, resourceSolu := range strategy.Resourcesolution {
			profit := computeProfit(ps.Task, strategy.Tasksolution, resourceSolu, ps.ProviderList)
			if profit > maxprofit {
				maxprofit = profit
			}
		}
	}

	for _, strategy := range ps.StrategyList {
		for _, resourceSolu := range strategy.Resourcesolution {
			profit := computeProfit(ps.Task, strategy.Tasksolution, resourceSolu, ps.ProviderList)
			highDegree := computeHighDegree(ps.Task, resourceSolu, ps.ProviderList)

			valueSum := profit/maxprofit + highDegree

			//将每个确定任务分配策略的最高的策略得分存储到里面
			if valueSum > maxStrategy.ValueSum {
				strategy.Profit = profit
				strategy.HighDegree = highDegree
			}

			if valueSum > maxStrategy.ValueSum {
				maxStrategy.ValueSum = valueSum
				maxStrategy.Tasksolution = strategy.Tasksolution
				newResourceSolu := [][]int{}
				newResourceSolu = append(newResourceSolu, resourceSolu)
				maxStrategy.Resourcesolution = newResourceSolu
				maxStrategy.Profit = profit
				maxStrategy.HighDegree = highDegree
			}
		}
	}

	if len(ps.ProviderList) == 0 {
		return nil, errors.New("empty providers")
	}

	ps.Task.MaxscoreStrategy = maxStrategy // 记录该任务的最终分配策略
	return ps.Task, nil
}

func computeProfit(task *Task, tasksolution []int, resourcesolution []int, providerList []*Provider) float64 {
	var timeexecution int //记录任务的实际最大执行时间
	var costSum float64   //该任务在多个云厂商所需支付的成本总价

	for i, provider := range providerList {

		//如果该厂商分的任务为0，则直接跳过该厂商，循环到下一厂商
		if tasksolution[i] == 0 {
			continue
		}

		//先计算下该云厂商的执行时间ddl，并替换任务的最大执行时间,向上取整
		t := math.Ceil(float64(tasksolution[i])/float64(resourcesolution[i])) * float64(task.Time)
		if int(t) > timeexecution {
			timeexecution = int(t)
		}

		//计算前几份资源多执行任务
		forOneMoreTaskNUm := tasksolution[i] % resourcesolution[i]

		for j := 0; j < resourcesolution[i]; j++ {
			if j < forOneMoreTaskNUm {
				t = math.Ceil(float64(tasksolution[i])/float64(resourcesolution[i])) * float64(task.Time)
			} else {
				t = math.Floor(float64(tasksolution[i])/float64(resourcesolution[i])) * float64(task.Time)
			}

			//如果这份资源分的的任务数
			cost := (provider.CpuCost*task.Cpu + provider.MemCost*task.Mem + provider.DiskCost*task.Disk) * t * (math.Pow(float64(j+1), math.Log2(provider.LearnIndex)))
			costSum += cost
		}

	}

	//计算用户的支付价格pay
	pay := task.Pr
	if timeexecution == task.Time { //没有排队等待，且只有一个副本直接执行或者多个副本完全并行执行
		if pay < costSum {
			pay = costSum
		}
	} else if timeexecution >= task.T0 && timeexecution <= task.T1 { //有排队时间或者任务存在串行执行
		if task.T1 == task.T0 { //仅有一个副本，时间中有排队时间
			e := math.Exp(float64(-task.B) * float64(timeexecution-task.T1))
			pay = (1 - 1/(1+e)) * task.Pr
		} else { //多个副本
			e := math.Exp(float64(-task.B) * float64(timeexecution-task.T1) / float64(task.T1-task.T0))
			pay = (1 - 1/(1+e)) * task.Pr
		}

		if pay < costSum {
			pay = costSum
		}
	} else { //超出用户满意度的完全串行时间
		pay = 1 / 2 * task.Pr
		if pay < costSum {
			pay = costSum
		}
	}

	profitSum := pay - costSum
	return profitSum
}

func computeHighDegree(task *Task, resourcesolution []int, providerList []*Provider) float64 {
	var highDegreeSum float64
	// 依次计算每个云厂商的资源可用度
	for i, provider := range providerList {
		// 定义两个四维向量
		// 未来任务资源需求比例
		futureDemand := mat.NewVecDense(3, []float64{1, 1, 1})

		// 定义假设按此方案分配后的剩余资源可用量，时间虽然有差异，但是先按那个时刻算吧，这里可能还要改一下
		nowLeft_cpu := provider.CpuAvail - task.Cpu*float64(resourcesolution[i])
		nowLeft_mem := provider.MemAvail - task.Mem*float64(resourcesolution[i])
		nowLeft_disk := provider.DiskAvail - task.Disk*float64(resourcesolution[i])

		nowLeft := mat.NewVecDense(3, []float64{nowLeft_cpu, nowLeft_mem, nowLeft_disk})
		// 使用余弦相似度计算两个比值的相近度
		// 计算向量的内积
		dot_product := mat.Dot(futureDemand, nowLeft)

		// 计算向量的模长
		magnitude1 := mat.Norm(futureDemand, 2)
		magnitude2 := mat.Norm(nowLeft, 2)

		// 计算余弦相似度
		//临时处理被除数为0的特殊情况
		var cosineSimilarity = 0.0
		if magnitude1 != 0 && magnitude2 != 0 {
			cosineSimilarity = dot_product / (magnitude1 * magnitude2)
		}

		highDegreeSum += cosineSimilarity
	}

	return highDegreeSum / float64(len(providerList))
}

func back_trace_task(ReplicaNum int, DoneReplicasNum int, providerList []*Provider, staclu int, res *[][]int, sum int) {
	//var count int = 0
	pnum := len(providerList)

	//所有的任务数都已经进行分配
	if DoneReplicasNum == ReplicaNum {
		var a []int
		for i := 0; i < pnum; i++ {
			a = append(a, providerList[i].CurReplicas)
		}
		*res = append(*res, a)
		//(*res)[0] = append((*res)[0], a)
		//count += 1
		return
	}

	//遍历完所有的云厂商序号
	if staclu >= pnum {
		return
	}

	if providerList[staclu].CurReplicas < providerList[staclu].MaxTaskCanRun {
		providerList[staclu].CurReplicas += 1
		back_trace_task(ReplicaNum, DoneReplicasNum+1, providerList, staclu, res, sum)
		providerList[staclu].CurReplicas -= 1
		back_trace_task(ReplicaNum, DoneReplicasNum, providerList, staclu+1, res, sum)
	} else {
		back_trace_task(ReplicaNum, DoneReplicasNum, providerList, staclu+1, res, sum)
	}

}

func back_trace_resource(list []int, i int, path []int, pathlist *[][]int) {
	if i == len(list) {
		var pathCopy = make([]int, len(path))
		copy(pathCopy, path)
		*pathlist = append(*pathlist, pathCopy)
		return
	}

	if list[i] == 0 {
		path = append(path, 0)
		back_trace_resource(list, i+1, path, pathlist)
		path = path[:len(path)-1]
	} else {
		for j := 1; j < list[i]+1; j++ {
			path = append(path, j)
			back_trace_resource(list, i+1, path, pathlist)
			path = path[:len(path)-1]
		}
	}

}