Leopard Implementation for NPFU

Author: Tingjia

.gitignore

@@ -0,0 +1,8 @@
+*.log
+*.json
+imgs/ol-min
+
+ol
+min-image/spin
+
+default-ol/*

BilliBenchClient/client.go

@@ -0,0 +1,363 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"math"
+	"net/http"
+	"os"
+	"strconv"
+	"sync"
+	"time"
+)
+
+type LambdaFunction struct {
+	MaxCPU            int `json:"maxCPU"`
+	MaxMemory         int `json:"maxMemory"`
+	IdleMemory        int `json:"idleMemory"`
+	ColdStartDuration int `json:"coldStartDuration"`
+
+	AvgCPU_orig    float64 `json:"avgCPU_orig"`
+	AvgCPU_uniform float64 `json:"avgCPU_uniform"`
+	AvgCPU_norm_50 float64 `json:"avgCPU_norm_50"`
+	AvgCPU_norm_70 float64 `json:"avgCPU_norm_70"`
+	AvgCPU_norm_30 float64 `json:"avgCPU_norm_30"`
+	AvgCPU_norm_90 float64 `json:"avgCPU_norm_90"`
+	AvgCPU_norm_10 float64 `json:"avgCPU_norm_10"`
+
+	AvgMemory  float64 `json:"avgMemory"`
+	AvgRuntime float64 `json:"avgRuntime"`
+
+	TotalInvocationNumber int `json:"totalInvocationNumber"`
+}
+
+type LambdaRegisterMsg struct {
+	FuncName string `json:"funcName"`
+
+	// Knobs
+	MaxCPU      int  `json:"maxCPU"`
+	MaxMem      int  `json:"maxMem"`
+	Preemptible bool `json:"preemptible"`
+
+	// ColdStart information
+	IdleMemory        int `json:"idleMemory"`
+	ColdStartDuration int `json:"coldStartDuration"`
+
+	// Stat
+	AvgCPUUtil float64 `json:"avgCPUUtil"`
+	AvgMemUtil float64 `json:"avgMemUtil"`
+	AvgRuntime float64 `json:"avgRuntime"`
+}
+
+type LambdaCallMsg struct {
+	FuncName     string  `json:"funcName"`
+	IdealRuntime float64 `json:"idealRuntime"`
+	EventStr     string  `json:"eventStr"`
+	InvokeId     string  `json:"invokeId"`
+}
+
+var Conf *Config
+
+type Config struct {
+	AdmissionControl string
+	Mem2CPU          int
+	boss_addr        string
+	boss_port        string
+	requestLog       *log.Logger
+	load             float64
+	cpudist          string
+}
+
+func RegisterFunctions(lambda_functions map[string]LambdaFunction, batch_set []string, client *http.Client) {
+	var maxCPU int
+	var maxMemory int
+	var preemptible bool
+
+	for lambda_name, lambda_function := range lambda_functions {
+		lambda_name_int, err := strconv.Atoi(lambda_name)
+		maxCPU = lambda_name_int%8 + 1
+		maxMemory = lambda_function.MaxMemory
+		if Conf.AdmissionControl == "StaticLinear" {
+			preemptible = false
+			if maxCPU*Conf.Mem2CPU < maxMemory {
+				maxCPU = int(math.Ceil(float64(maxMemory) / float64(Conf.Mem2CPU)))
+			}
+			maxMemory = maxCPU * Conf.Mem2CPU
+		} else if Conf.AdmissionControl == "Static" {
+			preemptible = false
+		} else {
+			preemptible = false
+			for _, str := range batch_set {
+				if str == lambda_name {
+					preemptible = true
+					break
+				}
+			}
+		}
+
+		avgCPU := 0.0
+		switch Conf.cpudist {
+		case "orig":
+			avgCPU = lambda_function.AvgCPU_orig
+		case "uniform":
+			avgCPU = lambda_function.AvgCPU_uniform
+		case "norm_10":
+			avgCPU = lambda_function.AvgCPU_norm_10
+		case "norm_30":
+			avgCPU = lambda_function.AvgCPU_norm_30
+		case "norm_50":
+			avgCPU = lambda_function.AvgCPU_norm_50
+		case "norm_70":
+			avgCPU = lambda_function.AvgCPU_norm_70
+		case "norm_90":
+			avgCPU = lambda_function.AvgCPU_norm_90
+		}
+
+		msg := LambdaRegisterMsg{
+			FuncName:    lambda_name,
+			MaxCPU:      maxCPU,
+			MaxMem:      maxMemory,
+			Preemptible: preemptible,
+
+			IdleMemory:        lambda_function.IdleMemory,
+			ColdStartDuration: lambda_function.ColdStartDuration,
+
+			AvgCPUUtil: avgCPU * float64(lambda_name_int%8+1),
+			AvgMemUtil: lambda_function.AvgMemory,
+			AvgRuntime: lambda_function.AvgRuntime,
+		}
+
+		jsonData, err := json.Marshal(msg)
+		if err != nil {
+			panic(fmt.Sprintf("Error marshaling JSON: %v", err))
+		}
+
+		url := fmt.Sprintf("http://%s:%s/register/", Conf.boss_addr, Conf.boss_port)
+		req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
+		if err != nil {
+			panic(fmt.Sprintf("error creating request: %v", err))
+		}
+		req.Header.Set("Content-Type", "application/json")
+
+		fmt.Printf("%s\n", jsonData)
+		resp, err := client.Do(req)
+		if err != nil {
+			panic(fmt.Sprintf("error sending request: %v", err))
+		}
+		resp.Body.Close()
+
+		fmt.Println("Response status:", resp.Status)
+	}
+
+	return
+}
+
+func SendLambdaCallReq(fname string, lambda_call map[string]interface{}, client *http.Client, wg *sync.WaitGroup, nextId int) {
+	defer wg.Done()
+	event := make(map[string]interface{})
+	event["durationArray"] = lambda_call["durationArray"].([]interface{})
+	event["memArray"] = lambda_call["memArray"].([]interface{})
+	event["cpuArray"] = lambda_call[fmt.Sprintf("parallelArray_%s", Conf.cpudist)].([]interface{})
+
+	eventStr, err := json.Marshal(event)
+	if err != nil {
+		panic(fmt.Sprintf("Error marshaling JSON: %v", err))
+	}
+
+	msg := LambdaCallMsg{
+		FuncName:     fname,
+		IdealRuntime: lambda_call["IdealRuntime"].(float64),
+		EventStr:     string(eventStr),
+		InvokeId:     fmt.Sprintf("%sIK%d", fname, nextId),
+	}
+
+	jsonData, err := json.Marshal(msg)
+	if err != nil {
+		panic(fmt.Sprintf("Error marshaling JSON: %v", err))
+	}
+
+	url := fmt.Sprintf("http://%s:%s/run/", Conf.boss_addr, Conf.boss_port)
+	req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
+	if err != nil {
+		panic(fmt.Sprintf("error creating request: %v", err))
+	}
+	req.Header.Set("Content-Type", "application/json")
+	fmt.Printf("%s\n", jsonData)
+	Conf.requestLog.Printf("%s", jsonData)
+	resp, err := client.Do(req)
+	if err != nil {
+		panic(fmt.Sprintf("error sending request: %v", err))
+	}
+	resp.Body.Close()
+
+}
+
+func conditionalMultiple(index, divisor, falseVal, trueVal int) int {
+	if index%divisor == 0 {
+		return falseVal
+	}
+	return trueVal
+}
+
+func CallFunction(fname string, lambda_call_list [][]interface{}, client *http.Client, wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	var function_wg sync.WaitGroup
+	// timeInterval := int(lambda_calls[0].(float64))
+	nextId := 0
+	multiple := 0
+	for index, value := range lambda_call_list {
+		timeInterval := int(value[0].(float64))
+		time.Sleep(time.Duration(timeInterval) * time.Millisecond)
+
+		switch Conf.load {
+		case 0.25:
+			multiple = conditionalMultiple(index, 4, 1, 0)
+		case 0.5:
+			multiple = conditionalMultiple(index, 2, 0, 1)
+		case 0.75:
+			multiple = conditionalMultiple(index, 4, 0, 1)
+		case 1:
+			multiple = 1
+		case 1.25:
+			multiple = conditionalMultiple(index, 4, 2, 1)
+		case 1.5:
+			multiple = conditionalMultiple(index, 2, 1, 2)
+		case 1.75:
+			multiple = conditionalMultiple(index, 4, 1, 2)
+		case 2:
+			multiple = 2
+		case 2.5:
+			multiple = conditionalMultiple(index, 2, 2, 3)
+		case 3:
+			multiple = 3
+		}
+
+		for i := 0; i < multiple; i += 1 {
+			function_wg.Add(1)
+			lambda_call := value[1].(map[string]interface{})
+			go SendLambdaCallReq(fname, lambda_call, client, &function_wg, nextId)
+			nextId += 1
+		}
+
+	}
+
+	function_wg.Wait()
+}
+
+type Addresses struct {
+	Cpudist          string   `json:"cpudist"`
+	Batchset         string   `json:"batchset"`
+	AdmissionControl string   `json:"admissionControl"`
+	Controller       string   `json:"controller"`
+	Workers          []string `json:"workers"`
+	Load             float64  `json:"load"`
+	Mem2CPU          float64  `json:"mem2CPU"`
+}
+
+func main() {
+	file, err := os.Open("/users/Tingjia/project/addresses.json")
+	if err != nil {
+		log.Fatalf("Failed to open file: %s", err)
+	}
+	defer file.Close()
+
+	byteValue, err := io.ReadAll(file)
+	if err != nil {
+		log.Fatalf("Failed to read file: %s", err)
+	}
+
+	var addresses Addresses
+	err = json.Unmarshal(byteValue, &addresses)
+	if err != nil {
+		log.Fatalf("Failed to unmarshal JSON: %s", err)
+	}
+
+	Conf = &Config{
+		AdmissionControl: addresses.AdmissionControl, // StaticLinear Static FLEX
+		Mem2CPU:          int(addresses.Mem2CPU),
+		boss_addr:        addresses.Controller,
+		boss_port:        "5000",
+		load:             addresses.Load,
+		cpudist:          addresses.Cpudist,
+	}
+
+	requestLogFile, _ := os.Create("request.log")
+	requestLog := log.New(requestLogFile, "", 0)
+	requestLog.SetFlags(log.Lmicroseconds)
+	Conf.requestLog = requestLog
+
+	client := &http.Client{}
+
+	var lambda_functions map[string]LambdaFunction
+	var lambda_calls map[string][][]interface{}
+	// Get the function statistic and meta data
+	jsonFile, err := os.Open("workloads/Functions.json")
+	if err != nil {
+		log.Fatalf("Failed to open JSON file: %s", err)
+	}
+	defer jsonFile.Close()
+
+	byteValue, err = io.ReadAll(jsonFile)
+	if err != nil {
+		log.Fatalf("Failed to read JSON file: %s", err)
+	}
+
+	err = json.Unmarshal(byteValue, &lambda_functions)
+	if err != nil {
+		fmt.Println("Error parsing JSON:", err)
+		return
+	}
+
+	jsonFile, err = os.Open(fmt.Sprintf("workloads/Batch_set.json"))
+	if err != nil {
+		log.Fatalf("Failed to open JSON file: %s", err)
+	}
+	defer jsonFile.Close()
+
+	// Read the file content
+	byteValue, err = io.ReadAll(jsonFile)
+	if err != nil {
+		log.Fatalf("Failed to read JSON file: %s", err)
+	}
+
+	var batchFuncs map[string][]string
+
+	// Parse the JSON data
+	if err := json.Unmarshal(byteValue, &batchFuncs); err != nil {
+		log.Fatalf("Failed to parse JSON: %s", err)
+	}
+
+	RegisterFunctions(lambda_functions, batchFuncs[addresses.Batchset], client)
+
+	// Get the request call for each function
+	jsonFile, err = os.Open(fmt.Sprintf("workloads/AzureWorkloads_0.json"))
+	if err != nil {
+		log.Fatalf("Failed to open JSON file: %s", err)
+	}
+	defer jsonFile.Close()
+
+	byteValue, err = io.ReadAll(jsonFile)
+	if err != nil {
+		log.Fatalf("Failed to read JSON file: %s", err)
+	}
+
+	err = json.Unmarshal(byteValue, &lambda_calls)
+	if err != nil {
+		fmt.Println("Error parsing JSON:", err)
+		return
+	}
+
+	var wg sync.WaitGroup
+	for fname, lambda_call_list := range lambda_calls {
+		wg.Add(1)
+		go CallFunction(fname, lambda_call_list, client, &wg)
+	}
+	time.Sleep(20 * time.Minute)
+	os.Exit(1)
+
+	// wg.Wait()
+}