diff --git a/schedulers/proactive_dynamic_capping/README.md b/schedulers/proactive_dynamic_capping/README.md new file mode 100644 index 0000000..60f4431 --- /dev/null +++ b/schedulers/proactive_dynamic_capping/README.md @@ -0,0 +1,10 @@ +##Proactive Dynamic Capping + +Perform Cluster wide dynamic capping. + +Offer 2 methods: + 1. First Come First Serve -- For each task that needs to be scheduled, in the order in which it arrives, compute the cluster wide cap. + 2. Rank based cluster wide capping -- Sort a given set of tasks to be scheduled, in ascending order of requested watts, and then compute the cluster wide cap for each of the tasks in the ordered set. + +#Note + The github.com folder contains a library that is required to compute the median of a given set of values. diff --git a/schedulers/proactive_dynamic_capping/main.go b/schedulers/proactive_dynamic_capping/main.go new file mode 100644 index 0000000..d705017 --- /dev/null +++ b/schedulers/proactive_dynamic_capping/main.go @@ -0,0 +1,99 @@ +package main + +import ( + "constants" + "fmt" + "math/rand" + "task" + "proactive_dynamic_capping" + ) + +func sample_available_power() map[string]float64{ + return map[string]float64{ + "stratos-001":100.0, + "stratos-002":150.0, + "stratos-003":80.0, + "stratos-004":90.0, + } +} + +func get_random_power(min, max int) int { + return rand.Intn(max - min) + min +} + +func cap_value_one_task_fcfs(capper *proactive_dynamic_capping.Capper) { + fmt.Println("==== FCFS, Number of tasks: 1 ====") + available_power := sample_available_power() + tsk := task.NewTask("gouravr/minife:v5", "minife:v5", "stratos-001", + "minife_command", 4.0, 10, 50, 1) + if cap_value, err := capper.Fcfs_determine_cap(available_power, tsk); err == nil { + fmt.Println("task = " + tsk.String()) + fmt.Printf("cap value = %f\n", cap_value) + } +} + +func cap_value_window_size_tasks_fcfs(capper *proactive_dynamic_capping.Capper) { + fmt.Println() + fmt.Println("==== FCFS, Number of tasks: 3 (window size) ====") + available_power := sample_available_power() + for i := 0; i < constants.Window_size; i++ { + tsk := task.NewTask("gouravr/minife:v5", "minife:v5", "stratos-001", + "minife_command", 4.0, 10, get_random_power(30, 150), 1) + fmt.Printf("task%d = %s\n", i, tsk.String()) + if cap_value, err := capper.Fcfs_determine_cap(available_power, tsk); err == nil { + fmt.Printf("CAP: %f\n", cap_value) + } + } +} + +func cap_value_more_than_window_size_tasks_fcfs(capper *proactive_dynamic_capping.Capper) { + fmt.Println() + fmt.Println("==== FCFS, Number of tasks: >3 (> window_size) ====") + available_power := sample_available_power() + for i := 0; i < constants.Window_size + 2; i++ { + tsk := task.NewTask("gouravr/minife:v5", "minife:v5", "stratos-001", + "minife_command", 4.0, 10, get_random_power(30, 150), 1) + fmt.Printf("task%d = %s\n", i, tsk.String()) + if cap_value, err := capper.Fcfs_determine_cap(available_power, tsk); err == nil { + fmt.Printf("CAP: %f\n", cap_value) + } + } +} + +func cap_values_for_ranked_tasks(capper *proactive_dynamic_capping.Capper) { + fmt.Println() + fmt.Println("==== Ranked, Number of tasks: 5 (window size + 2) ====") + available_power := sample_available_power() + var tasks_to_schedule []*task.Task + for i := 0; i < constants.Window_size + 2; i++ { + tasks_to_schedule = append(tasks_to_schedule, + task.NewTask("gouravr/minife:v5", "minife:v5", "stratos-001", + "minife_command", 4.0, 10, get_random_power(30, 150), 1)) + } + // Printing the tasks that need to be scheduled. + index := 0 + for _, tsk := range tasks_to_schedule { + fmt.Printf("task%d = %s\n", index, tsk.String()) + index++ + } + if sorted_tasks_to_be_scheduled, cwcv, err := capper.Ranked_determine_cap(available_power, tasks_to_schedule); err == nil { + fmt.Printf("The cap values are: ") + fmt.Println(cwcv) + fmt.Println("The order of tasks to be scheduled :-") + for _, tsk := range sorted_tasks_to_be_scheduled { + fmt.Println(tsk.String()) + } + } +} + +func main() { + capper := proactive_dynamic_capping.GetInstance() + cap_value_one_task_fcfs(capper) + capper.Clear() + cap_value_window_size_tasks_fcfs(capper) + capper.Clear() + cap_value_more_than_window_size_tasks_fcfs(capper) + capper.Clear() + cap_values_for_ranked_tasks(capper) + capper.Clear() +} diff --git a/schedulers/proactive_dynamic_capping/src/constants/constants.go b/schedulers/proactive_dynamic_capping/src/constants/constants.go new file mode 100644 index 0000000..0b1a0cc --- /dev/null +++ b/schedulers/proactive_dynamic_capping/src/constants/constants.go @@ -0,0 +1,39 @@ +/* +Constants that are used across scripts +1. The available hosts = stratos-00x (x varies from 1 to 8) +2. cap_margin = percentage of the requested power to allocate +3. power_threshold = overloading factor +4. total_power = total power per node +5. window_size = number of tasks to consider for computation of the dynamic cap. +*/ +package constants + +var Hosts = []string{"stratos-001", "stratos-002", + "stratos-003", "stratos-004", + "stratos-005", "stratos-006", + "stratos-007", "stratos-008"} + +/* + Margin with respect to the required power for a job. + So, if power required = 10W, the node would be capped to 75%*10W. + This value can be changed upon convenience. +*/ +var Cap_margin = 0.75 + +// Lower bound of the power threshold for a tasks +var Power_threshold = 0.6 + +// Total power per node +var Total_power = map[string]float64 { + "stratos-001": 100.0, + "stratos-002": 150.0, + "stratos-003": 80.0, + "stratos-004": 90.0, + "stratos-005": 200.0, + "stratos-006": 100.0, + "stratos-007": 175.0, + "stratos-008": 175.0, +} + +// Window size for running average +var Window_size = 3 diff --git a/schedulers/proactive_dynamic_capping/src/github.com/montanaflynn/stats b/schedulers/proactive_dynamic_capping/src/github.com/montanaflynn/stats new file mode 160000 index 0000000..60dcacf --- /dev/null +++ b/schedulers/proactive_dynamic_capping/src/github.com/montanaflynn/stats @@ -0,0 +1 @@ +Subproject commit 60dcacf48f43d6dd654d0ed94120ff5806c5ca5c diff --git a/schedulers/proactive_dynamic_capping/src/proactive_dynamic_capping/capper.go b/schedulers/proactive_dynamic_capping/src/proactive_dynamic_capping/capper.go new file mode 100644 index 0000000..4e183f3 --- /dev/null +++ b/schedulers/proactive_dynamic_capping/src/proactive_dynamic_capping/capper.go @@ -0,0 +1,235 @@ +/* +Cluster wide dynamic capping +Step1. Compute running average of tasks in window. +Step2. Compute what percentage of available power of each node, is the running average. +Step3. Compute the median of the percentages and this is the percentage that the cluster needs to be cpaped at. + +1. First Fit Scheduling -- Perform the above steps for each task that needs to be scheduled. +2. Rank based Scheduling -- Sort a set of tasks to be scheduled, in ascending order of power, and then perform the above steps for each of them in the sorted order. +*/ + +package proactive_dynamic_capping + +import ( + "constants" + "container/list" + "errors" + "github.com/montanaflynn/stats" + "task" + "sort" + "sync" +) + +// Structure containing utility data structures used to compute cluster wide dyanmic cap. +type Capper struct { + // window of tasks. + window_of_tasks list.List + // The current sum of requested powers of the tasks in the window. + current_sum float64 + // The current number of tasks in the window. + number_of_tasks_in_window int +} + +// Defining constructor for Capper. +func NewCapper() *Capper { + return &Capper{current_sum: 0.0, number_of_tasks_in_window: 0} +} + +// For locking on operations that may result in race conditions. +var mutex sync.Mutex + +// Singleton instance of Capper +var singleton_capper *Capper +// Retrieve the singleton instance of Capper. +func GetInstance() *Capper { + if singleton_capper == nil { + mutex.Lock() + singleton_capper = NewCapper() + mutex.Unlock() + } else { + // Do nothing + } + return singleton_capper +} + +// Clear and initialize all the members of Capper. +func (capper Capper) Clear() { + capper.window_of_tasks.Init() + capper.current_sum = 0 + capper.number_of_tasks_in_window = 0 +} + +// Compute the average of watts of all the tasks in the window. +func (capper Capper) average() float64 { + return capper.current_sum / float64(capper.window_of_tasks.Len()) +} + +/* + Compute the running average + + Using Capper#window_of_tasks to store the tasks in the window. Task at position 0 (oldest task) removed when window is full and new task arrives. +*/ +func (capper Capper) running_average_of_watts(tsk *task.Task) float64 { + var average float64 + if capper.number_of_tasks_in_window < constants.Window_size { + capper.window_of_tasks.PushBack(tsk) + capper.number_of_tasks_in_window++ + capper.current_sum += float64(tsk.Watts) + } else { + task_to_remove_element := capper.window_of_tasks.Front() + if task_to_remove, ok := task_to_remove_element.Value.(*task.Task); ok { + capper.current_sum -= float64(task_to_remove.Watts) + capper.window_of_tasks.Remove(task_to_remove_element) + } + capper.window_of_tasks.PushBack(tsk) + capper.current_sum += float64(tsk.Watts) + } + average = capper.average() + return average +} + +/* + Calculating cap value + + 1. Sorting the values of running_average_available_power_percentage in ascending order. + 2. Computing the median of the above sorted values. + 3. The median is now the cap value. +*/ +func (capper Capper) get_cap(running_average_available_power_percentage map[string]float64) float64 { + var values []float64 + // Validation + if running_average_available_power_percentage == nil { + return 100.0 + } + for _, apower := range running_average_available_power_percentage { + values = append(values, apower) + } + // sorting the values in ascending order + sort.Float64s(values) + // Calculating the median + if median, err := stats.Median(values); err == nil { + return median + } + // should never reach here. If here, then just setting the cap value to be 100 + return 100.0 +} + +// In place sorting of tasks to be scheduled based on the requested watts. +func qsort_tasks(low int, high int, tasks_to_sort []*task.Task) { + i := low + j := high + // calculating the pivot + pivot_index := low + (high - low)/2 + pivot := tasks_to_sort[pivot_index] + for i <= j { + for tasks_to_sort[i].Watts < pivot.Watts { + i++ + } + for tasks_to_sort[j].Watts > pivot.Watts { + j-- + } + if i <= j { + temp := tasks_to_sort[i] + tasks_to_sort[i] = tasks_to_sort[j] + tasks_to_sort[j] = temp + i++ + j-- + } + } + if low < j { + qsort_tasks(low, j, tasks_to_sort) + } + if i < high { + qsort_tasks(i, high, tasks_to_sort) + } +} + +// Sorting tasks in ascending order of requested watts. +func (capper Capper) sort_tasks(tasks_to_sort []*task.Task) { + qsort_tasks(0, len(tasks_to_sort)-1, tasks_to_sort) +} + +/* +Remove entry for finished task. +Electron needs to call this whenever a task completes so that the finished task no longer contributes to the computation of the cluster wide cap. +*/ +func (capper Capper) Task_finished(finished_task *task.Task) { + // If the window is empty then just return. Should not be entering this condition as it would mean that there is a bug. + if capper.window_of_tasks.Len() == 0 { + return + } + + // Checking whether the finished task is currently present in the window of tasks. + var task_element_to_remove *list.Element + for task_element := capper.window_of_tasks.Front(); task_element != nil; task_element = task_element.Next() { + if tsk, ok := task_element.Value.(*task.Task); ok { + if task.Compare(tsk, finished_task) { + task_element_to_remove = task_element + } + } + } + + // If finished task is there in the window of tasks, then we need to remove the task from the same and modify the members of Capper accordingly. + if task_to_remove, ok := task_element_to_remove.Value.(*task.Task); ok { + capper.window_of_tasks.Remove(task_element_to_remove) + capper.number_of_tasks_in_window -= 1 + capper.current_sum -= float64(task_to_remove.Watts) + } +} + +// Ranked based scheduling +func (capper Capper) Ranked_determine_cap(available_power map[string]float64, tasks_to_schedule []*task.Task) ([]*task.Task, map[int]float64, error) { + // Validation + if available_power == nil || len(tasks_to_schedule) == 0 { + return nil, nil, errors.New("No available power and no tasks to schedule.") + } else { + // Need to sort the tasks in ascending order of requested power + capper.sort_tasks(tasks_to_schedule) + + // Now, for each task in the sorted set of tasks, we need to use the Fcfs_determine_cap logic. + cluster_wide_cap_values := make(map[int]float64) + index := 0 + for _, tsk := range tasks_to_schedule { + /* + Note that even though Fcfs_determine_cap is called, we have sorted the tasks aprior and thus, the tasks are scheduled in the sorted fashion. + Calling Fcfs_determine_cap(...) just to avoid redundant code. + */ + if cap, err := capper.Fcfs_determine_cap(available_power, tsk); err == nil { + cluster_wide_cap_values[index] = cap + } else { + return nil, nil, err + } + index++ + } + // Now returning the sorted set of tasks and the cluster wide cap values for each task that is launched. + return tasks_to_schedule, cluster_wide_cap_values, nil + } +} + +// First come first serve scheduling. +func (capper Capper) Fcfs_determine_cap(available_power map[string]float64, new_task *task.Task) (float64, error) { + // Validation + if available_power == nil { + // If no power available power, then capping the cluster at 100%. Electron might choose to queue the task. + return 100.0, errors.New("No available power.") + } else { + mutex.Lock() + // Need to calcualte the running average + running_average := capper.running_average_of_watts(new_task) + // What percent of available power for each node is the running average + running_average_available_power_percentage := make(map[string]float64) + for node, apower := range available_power { + if apower >= running_average { + running_average_available_power_percentage[node] = (running_average/apower) * 100 + } else { + // We don't consider this node in the offers + } + } + + // Determine the cluster wide cap value. + cap_value := capper.get_cap(running_average_available_power_percentage) + // Electron has to now cap the cluster to this value before launching the next task. + mutex.Unlock() + return cap_value, nil + } +} diff --git a/schedulers/proactive_dynamic_capping/src/task/task.go b/schedulers/proactive_dynamic_capping/src/task/task.go new file mode 100644 index 0000000..47d8aa5 --- /dev/null +++ b/schedulers/proactive_dynamic_capping/src/task/task.go @@ -0,0 +1,73 @@ +package task + +import ( + "constants" + "encoding/json" + "reflect" + "strconv" + "utilities" +) + +/* + Blueprint for the task. + Members: + image: + name: + host: + cmd: + cpu: + ram: + watts: + inst: +*/ +type Task struct { + Image string + Name string + Host string + CMD string + CPU float64 + RAM int + Watts int + Inst int +} + +// Defining a constructor for Task +func NewTask(image string, name string, host string, + cmd string, cpu float64, ram int, watts int, inst int) *Task { + return &Task{Image: image, Name: name, Host: host, CPU: cpu, + RAM: ram, Watts: watts, Inst: inst} +} + +// Update the host on which the task needs to be scheduled. +func (task Task) Update_host(new_host string) { + // Validation + if _, ok := constants.Total_power[new_host]; ok { + task.Host = new_host + } +} + +// Stringify task instance +func (task Task) String() string { + task_map := make(map[string]string) + task_map["image"] = task.Image + task_map["name"] = task.Name + task_map["host"] = task.Host + task_map["cmd"] = task.CMD + task_map["cpu"] = utils.FloatToString(task.CPU) + task_map["ram"] = strconv.Itoa(task.RAM) + task_map["watts"] = strconv.Itoa(task.Watts) + task_map["inst"] = strconv.Itoa(task.Inst) + + json_string, _ := json.Marshal(task_map) + return string(json_string) +} + +// Compare one task to another. 2 tasks are the same if all the corresponding members are the same. +func Compare(task *Task, other_task *Task) bool { + // If comparing the same pointers (checking the addresses). + if task == other_task { + return true + } + // Checking member equality + return reflect.DeepEqual(*task, *other_task) +} diff --git a/schedulers/proactive_dynamic_capping/src/utilities/utils.go b/schedulers/proactive_dynamic_capping/src/utilities/utils.go new file mode 100644 index 0000000..5f2e341 --- /dev/null +++ b/schedulers/proactive_dynamic_capping/src/utilities/utils.go @@ -0,0 +1,9 @@ +package utils + +import "strconv" + +// Convert float64 to string +func FloatToString(input float64) string { + // Precision is 2, Base is 64 + return strconv.FormatFloat(input, 'f', 2, 64) +}