atune: modify server streaming rpc to bidirectional streaming rpc in tuning command

api/profile/profile.proto

@@ -20,7 +20,7 @@ service ProfileMgr {
 	rpc Profile(ProfileInfo) returns (stream AckCheck) {}
 	rpc CheckInitProfile(ProfileInfo) returns (stream AckCheck) {}
     	rpc Analysis(AnalysisMessage) returns (stream AckCheck) {}
-	rpc Tuning(ProfileInfo) returns (stream AckCheck) {}
+	rpc Tuning(stream ProfileInfo) returns (stream AckCheck) {}
 	rpc ListWorkload(ProfileInfo) returns (stream ListMessage) {}
 	rpc UpgradeProfile(ProfileInfo) returns (stream AckCheck) {}
 	rpc InfoProfile(ProfileInfo) returns (stream ProfileInfo) {}

common/project/projet.go

@@ -15,14 +15,10 @@ package project
 
 import (
 	"atune/common/log"
-	"atune/common/utils"
 	"fmt"
-	"io/ioutil"
 	"os/exec"
 	"strconv"
 	"strings"
-
-	yaml "gopkg.in/yaml.v2"
 )
 
 // Evaluate :store the evaluate object
@@ -78,59 +74,30 @@ type YamlPrjObj struct {
 	Info YamlObj `yaml:"info"`
 }
 
-// LoadProject method load the tuning yaml
-func LoadProject(path string) (*YamlPrjSvr, error) {
-	exist, err := utils.PathExist(path)
-	if err != nil {
-		return nil, err
-	}
-
-	if !exist {
-		return nil, fmt.Errorf("the path %s doesn't exist", path)
-	}
-
-	return newProject(path)
-}
-
-func newProject(path string) (*YamlPrjSvr, error) {
-	info, err := ioutil.ReadFile(path)
-	if err != nil {
-		return nil, err
-	}
-
-	prj := new(YamlPrjSvr)
-	if err := yaml.Unmarshal(info, prj); err != nil {
-		return nil, fmt.Errorf("parse %s failed : %s", path, err)
-	}
-
-	return prj, nil
-}
-
 // BenchMark method call the benchmark script
 func (y *YamlPrjCli) BenchMark() (string, error) {
 	benchStr := make([]string, 0)
 
-	cmd := exec.Command("sh", "-c", y.Benchmark)
-	benchOutByte, err := cmd.CombinedOutput()
-
+	benchOutByte, err := ExecCommand(y.Benchmark)
 	if err != nil {
 		return "", fmt.Errorf("failed to run benchmark, err: %v", err)
 	}
+
 	for _, evaluation := range y.Evaluations {
 		newScript := strings.Replace(evaluation.Info.Get, "$out", string(benchOutByte), -1)
-		cmd := exec.Command("sh", "-c", newScript)
-		bout, err := cmd.Output()
+		bout, err := ExecCommand(newScript)
 		if err != nil {
-			log.Error(err)
+			err = fmt.Errorf("faild to exec %s, err: %v", newScript, err)
 			return strings.Join(benchStr, ","), err
 		}
-		floatout, err := strconv.ParseFloat(strings.Replace(string(bout), "\n", "", -1), 64)
+
+		floatOut, err := strconv.ParseFloat(strings.Replace(string(bout), "\n", "", -1), 64)
 		if err != nil {
-			log.Error(err)
+			err = fmt.Errorf("faild to parse float, err: %v", err)
 			return strings.Join(benchStr, ","), err
 		}
 
-		out := strconv.FormatFloat((floatout * float64(evaluation.Info.Weight) / 100), 'f', -1, 64)
+		out := strconv.FormatFloat(floatOut*float64(evaluation.Info.Weight)/100, 'f', -1, 64)
 		if evaluation.Info.Type == "negative" {
 			out = "-" + out
 		}
@@ -208,3 +175,4 @@ func ExecCommand(script string) ([]byte, error) {
 	out, err := cmd.CombinedOutput()
 	return out, err
 }
+

common/tuning/optimizer.go

@@ -25,6 +25,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"time"
@@ -32,43 +33,39 @@ import (
 
 // Optimizer : the type implement the bayes serch service
 type Optimizer struct {
-	Prj *project.YamlPrjSvr
+	Prj             *project.YamlPrjSvr
+	Content         []byte
+	Iter            int
+	MaxIter         int
+	OptimizerPutURL string
+	FinalEval       string
+	MinEvalSum      float64
+	RespPutIns      *models.RespPutBody
+	StartIterTime   string
 }
 
-//BenchMark : the benchmark data
-type BenchMark struct {
-	Content []byte
-}
+// InitTuned method for init tuning
+func (o *Optimizer) InitTuned(ch chan *PB.AckCheck) error {
+	clientIter, err := strconv.Atoi(string(o.Content))
+	if err != nil {
+		return err
+	}
 
-var optimizerPutURL string
-var optimization *Optimizer
-var finalEval string
-var minEvalSum float64
-var respPutIns *models.RespPutBody
-var iter int
-var maxIter int
-var startIterTime string
+	log.Infof("begin to dynamic optimizer search, client ask iterations:%d", clientIter)
+	ch <- &PB.AckCheck{Name: fmt.Sprintf("begin to dynamic optimizer search")}
 
-// InitTuned method for init tuning
-func (o *Optimizer) InitTuned(ch chan *PB.AckCheck, askIter int) error {
 	//dynamic profle setting
-	maxIter = askIter
-	if maxIter > o.Prj.Maxiterations {
-		maxIter = o.Prj.Maxiterations
+	o.MaxIter = clientIter
+	if o.MaxIter > o.Prj.Maxiterations {
+		o.MaxIter = o.Prj.Maxiterations
 		log.Infof("project:%s max iterations:%d", o.Prj.Project, o.Prj.Maxiterations)
 		ch <- &PB.AckCheck{Name: fmt.Sprintf("server project %s max iterations %d\n",
 			o.Prj.Project, o.Prj.Maxiterations)}
 	}
 
-	exist, err := utils.PathExist(config.DefaultTempPath)
-	if err != nil {
+	if err := utils.CreateDir(config.DefaultTempPath, 0750); err != nil {
 		return err
 	}
-	if !exist {
-		if err = os.MkdirAll(config.DefaultTempPath, 0750); err != nil {
-			return err
-		}
-	}
 
 	projectName := fmt.Sprintf("project %s\n", o.Prj.Project)
 	err = utils.WriteFile(config.TuningFile, projectName, config.FilePerm,
@@ -78,12 +75,10 @@ func (o *Optimizer) InitTuned(ch chan *PB.AckCheck, askIter int) error {
 		return err
 	}
 
-	initConfigure := ""
+	initConfigure := make([]string, 0)
 	optimizerBody := new(models.OptimizerPostBody)
-	optimizerBody.MaxEval = maxIter
-
+	optimizerBody.MaxEval = o.MaxIter
 	optimizerBody.Knobs = make([]models.Knob, 0)
-
 	for _, item := range o.Prj.Object {
 		knob := new(models.Knob)
 		knob.Dtype = item.Info.Dtype
@@ -100,11 +95,11 @@ func (o *Optimizer) InitTuned(ch chan *PB.AckCheck, askIter int) error {
 		if err != nil {
 			return fmt.Errorf("faild to exec %s, err: %v", item.Info.GetScript, err)
 		}
-		initConfigure += strings.TrimSpace(knob.Name+"="+string(out)) + ","
+		initConfigure = append(initConfigure, strings.TrimSpace(knob.Name+"="+string(out)))
 	}
 
 	err = utils.WriteFile(path.Join(config.DefaultTempPath,
-		o.Prj.Project+config.TuningRestoreConfig), initConfigure,
+		o.Prj.Project+config.TuningRestoreConfig), strings.Join(initConfigure, ","),
 		config.FilePerm, os.O_WRONLY|os.O_CREATE|os.O_TRUNC)
 	if err != nil {
 		log.Error(err)
@@ -116,21 +111,17 @@ func (o *Optimizer) InitTuned(ch chan *PB.AckCheck, askIter int) error {
 		return err
 	}
 	if respPostIns.Status != "OK" {
-		log.Errorf(respPostIns.Status)
-		return fmt.Errorf("create task failed: %s", respPostIns.Status)
+		err := fmt.Errorf("create task failed: %s", respPostIns.Status)
+		log.Errorf(err.Error())
+		return err
 	}
 
-	log.Infof("create task id is %s", respPostIns.TaskID)
 	url := config.GetURL(config.OptimizerURI)
-	optimizerPutURL = fmt.Sprintf("%s/%s", url, respPostIns.TaskID)
-
-	log.Infof("optimizer put url is: %s", optimizerPutURL)
+	o.OptimizerPutURL = fmt.Sprintf("%s/%s", url, respPostIns.TaskID)
+	log.Infof("optimizer put url is: %s", o.OptimizerPutURL)
 
-	optimization = o
-	iter = 0
-
-	benchmark := BenchMark{Content: nil}
-	if _, err := benchmark.DynamicTuned(ch); err != nil {
+	o.Content = nil
+	if err := o.DynamicTuned(ch); err != nil {
 		return err
 	}
 
@@ -140,70 +131,72 @@ func (o *Optimizer) InitTuned(ch chan *PB.AckCheck, askIter int) error {
 /*
 DynamicTuned method using bayes algorithm to search the best performance parameters
 */
-func (bench *BenchMark) DynamicTuned(ch chan *PB.AckCheck) (bool, error) {
+func (o *Optimizer) DynamicTuned(ch chan *PB.AckCheck) error {
 	var evalValue string
 	var err error
-	if bench.Content != nil {
-		evalValue, err = bench.evalParsing(ch)
+	if o.Content != nil {
+		evalValue, err = o.evalParsing(ch)
 		if err != nil {
-			return true, err
+			return err
 		}
 	}
 
-	os.Setenv("ITERATION", strconv.Itoa(iter))
+	os.Setenv("ITERATION", strconv.Itoa(o.Iter))
 
 	optPutBody := new(models.OptimizerPutBody)
-	optPutBody.Iterations = iter
+	optPutBody.Iterations = o.Iter
 	optPutBody.Value = evalValue
-	respPutIns, err = optPutBody.Put(optimizerPutURL)
+	o.RespPutIns, err = optPutBody.Put(o.OptimizerPutURL)
 	if err != nil {
 		log.Errorf("get setting parameter error: %v", err)
-		return true, err
+		return err
 	}
 
-	log.Infof("setting params is: %s", respPutIns.Param)
-	if err := optimization.Prj.RunSet(respPutIns.Param); err != nil {
+	log.Infof("setting params is: %s", o.RespPutIns.Param)
+	if err := o.Prj.RunSet(o.RespPutIns.Param); err != nil {
 		log.Error(err)
-		return true, err
+		return err
 	}
-
 	log.Info("set the parameter success")
-	if err := optimization.Prj.RestartProject(); err != nil {
+
+	if err := o.Prj.RestartProject(); err != nil {
 		log.Error(err)
-		return true, err
+		return err
 	}
 	log.Info("restart project success")
 
-	startIterTime = time.Now().Format(config.DefaultTimeFormat)
+	o.StartIterTime = time.Now().Format(config.DefaultTimeFormat)
 
-	if iter == maxIter {
-		finalEval := strings.Replace(finalEval, "=-", "=", -1)
+	if o.Iter == o.MaxIter {
+		finalEval := strings.Replace(o.FinalEval, "=-", "=", -1)
 		optimizationTerm := fmt.Sprintf("\n The final optimization result is: %s\n"+
-			" The final evaluation value is: %s", respPutIns.Param, finalEval)
+			" The final evaluation value is: %s", o.RespPutIns.Param, finalEval)
 		log.Info(optimizationTerm)
-		ch <- &PB.AckCheck{Name: optimizationTerm, Status: utils.SUCCESS}
+		ch <- &PB.AckCheck{Name: optimizationTerm}
 
-		if err = deleteTask(optimizerPutURL); err != nil {
+		if err = deleteTask(o.OptimizerPutURL); err != nil {
 			log.Error(err)
 		}
-		return true, nil
 	}
 
-	iter++
-	return false, nil
+	o.Iter++
+	ch <- &PB.AckCheck{Status: utils.SUCCESS}
+
+	return nil
 }
 
 //restore tuning config
-func (o *Optimizer) RestoreConfigTuned() error {
+func (o *Optimizer) RestoreConfigTuned(ch chan *PB.AckCheck) error {
 	tuningRestoreConf := path.Join(config.DefaultTempPath, o.Prj.Project+config.TuningRestoreConfig)
 	exist, err := utils.PathExist(tuningRestoreConf)
 	if err != nil {
 		return err
 	}
 	if !exist {
-		log.Errorf("%s project has not been executed the dynamic optimizer search", o.Prj.Project)
-		return fmt.Errorf("%s project has not been executed the dynamic optimizer search",
-			o.Prj.Project)
+		err := fmt.Errorf("%s project has not been executed "+
+			"the dynamic optimizer search", o.Prj.Project)
+		log.Errorf(err.Error())
+		return err
 	}
 
 	content, err := ioutil.ReadFile(tuningRestoreConf)
@@ -218,27 +211,25 @@ func (o *Optimizer) RestoreConfigTuned() error {
 		return err
 	}
 
-	log.Infof("restore %s project params success", o.Prj.Project)
+	result := fmt.Sprintf("restore %s project params success", o.Prj.Project)
+	ch <- &PB.AckCheck{Name: result, Status: utils.SUCCESS}
+	log.Infof(result)
 	return nil
 }
 
-func (bench *BenchMark) evalParsing(ch chan *PB.AckCheck) (string, error) {
-	eval := string(bench.Content)
+func (o *Optimizer) evalParsing(ch chan *PB.AckCheck) (string, error) {
+	eval := string(o.Content)
 	positiveEval := strings.Replace(eval, "=-", "=", -1)
 	optimizationTerm := fmt.Sprintf("The %dth optimization result is: %s\n"+
-		" The %dth evaluation value is: %s", iter, respPutIns.Param, iter, positiveEval)
+		" The %dth evaluation value is: %s", o.Iter, o.RespPutIns.Param, o.Iter, positiveEval)
 	ch <- &PB.AckCheck{Name: optimizationTerm}
 	log.Info(optimizationTerm)
 
 	endIterTime := time.Now().Format(config.DefaultTimeFormat)
 	iterInfo := make([]string, 0)
-	iterInfo = append(iterInfo, strconv.Itoa(iter))
-	iterInfo = append(iterInfo, startIterTime)
-	iterInfo = append(iterInfo, endIterTime)
-	iterInfo = append(iterInfo, positiveEval)
-	iterInfo = append(iterInfo, respPutIns.Param)
+	iterInfo = append(iterInfo, strconv.Itoa(o.Iter), o.StartIterTime, endIterTime,
+		positiveEval, o.RespPutIns.Param)
 	output := strings.Join(iterInfo, "|")
-
 	err := utils.WriteFile(config.TuningFile, output+"\n", config.FilePerm,
 		os.O_APPEND|os.O_WRONLY)
 	if err != nil {
@@ -264,9 +255,9 @@ func (bench *BenchMark) evalParsing(ch chan *PB.AckCheck) (string, error) {
 		evalValue = append(evalValue, kvs[1])
 	}
 
-	if iter == 1 || evalSum < minEvalSum {
-		minEvalSum = evalSum
-		finalEval = eval
+	if o.Iter == 1 || evalSum < o.MinEvalSum {
+		o.MinEvalSum = evalSum
+		o.FinalEval = eval
 	}
 	return strings.Join(evalValue, ","), nil
 }
@@ -280,3 +271,34 @@ func deleteTask(url string) error {
 	defer resp.Body.Close()
 	return nil
 }
+
+//check server prj
+func CheckServerPrj(data string, optimizer *Optimizer) error {
+	var prjs []*project.YamlPrjSvr
+	err := filepath.Walk(config.DefaultTuningPath, func(path string, info os.FileInfo, err error) error {
+		if !info.IsDir() {
+			prj := new(project.YamlPrjSvr)
+			if err := utils.ParseFile(path, "yaml", &prj); err != nil {
+				return fmt.Errorf("load %s faild, err: %v", path, err)
+			}
+			log.Infof("project:%s load %s success", prj.Project, path)
+			prjs = append(prjs, prj)
+		}
+		return nil
+	})
+
+	if err != nil {
+		return err
+	}
+
+	for _, prj := range prjs {
+		if data == prj.Project {
+			log.Infof("find Project:%s", prj.Project)
+			optimizer.Prj = prj
+			return nil
+		}
+	}
+
+	return fmt.Errorf("project:%s not found", data)
+}
+

common/utils/utils.go

@@ -16,8 +16,11 @@ package utils
 import (
 	PB "atune/api/profile"
 	"bufio"
+	"encoding/xml"
 	"fmt"
+	"gopkg.in/yaml.v2"
 	"io"
+	"io/ioutil"
 	"math/rand"
 	"net"
 	"os"
@@ -306,3 +309,49 @@ func WriteFile(filename string, data string, perm os.FileMode, wrapper int) erro
 	return err
 }
 
+//parse yaml or xml file
+func ParseFile(filePath string, fileFormat string, out interface{}) error {
+	exist, err := PathExist(filePath)
+	if err != nil {
+		return err
+	}
+
+	if !exist {
+		return fmt.Errorf("%s is not exist", filePath)
+	}
+
+	data, err := ioutil.ReadFile(filePath)
+	if err != nil {
+		return err
+	}
+
+	switch fileFormat {
+	case "yaml":
+		if err := yaml.Unmarshal(data, out); err != nil {
+			return err
+		}
+	case "xml":
+		if err := xml.Unmarshal(data, out); err != nil {
+			return err
+		}
+	default:
+		return fmt.Errorf("this conversion mode of %s is not supported", fileFormat)
+	}
+
+	return nil
+}
+
+//create dir if the dir is not exist
+func CreateDir(dir string, perm os.FileMode) error {
+	exist, err := PathExist(dir)
+	if err != nil {
+		return err
+	}
+	if !exist {
+		if err = os.MkdirAll(dir, perm); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+

modules/client/profile/profile_tuning.go

@@ -21,13 +21,11 @@ import (
 	"atune/common/utils"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"strconv"
 	"strings"
 
 	"github.com/urfave/cli"
 	CTX "golang.org/x/net/context"
-	yaml "gopkg.in/yaml.v2"
 )
 
 var profileTunningCommand = cli.Command{
@@ -82,99 +80,92 @@ func profileTunning(ctx *cli.Context) error {
 	}
 
 	yamlPath := ctx.Args().Get(0)
-	exist, err := utils.PathExist(yamlPath)
-	if err != nil {
-		return err
-	}
-
-	if !exist {
-		return fmt.Errorf("project file %s is not exist", yamlPath)
-	}
-
 	if !strings.HasSuffix(yamlPath, ".yaml") && !strings.HasSuffix(yamlPath, ".yml") {
-		return fmt.Errorf("project file is not ends with yaml or yml")
+		return fmt.Errorf("error: %s is not ends with yaml or yml", yamlPath)
 	}
 
-	data, err := ioutil.ReadFile(yamlPath)
-	if err != nil {
-		return err
-	}
 	prj := project.YamlPrjCli{}
-	if err := yaml.Unmarshal(data, &prj); err != nil {
+	if err := utils.ParseFile(yamlPath, "yaml", &prj); err != nil {
 		return err
 	}
-
-	if len(prj.Project) < 1 || len(prj.Project) > 128 {
-		return fmt.Errorf("project name must be no less than 1 and no greater than 128 in yaml or yml")
-	}
-
-	if len(prj.Benchmark) < 1 {
-		return fmt.Errorf("benchmark must be specified in yaml or yml")
-	}
-
-	if len(prj.Evaluations) > 10 {
-		return fmt.Errorf("evaluations must be no greater than 10 in project %s", prj.Project)
-	}
-
-	_, err = runTuningRPC(ctx, &PB.ProfileInfo{Name: prj.Project, Content: []byte(strconv.Itoa(prj.Iterations))})
-	if err != nil {
+	if err := checkTuningPrjYaml(prj); err != nil {
 		return err
 	}
 
-	iter := 0
-	for iter < prj.Iterations {
-		benchmarkByte, err := prj.BenchMark()
-		if err != nil {
-			return err
+	err := runTuningRPC(ctx, func(stream PB.ProfileMgr_TuningClient) error {
+		content := &PB.ProfileInfo{Name: prj.Project, Content: []byte(strconv.Itoa(prj.Iterations))}
+		if err := stream.Send(content); err != nil {
+			return fmt.Errorf("client sends failure, error: %v", err)
 		}
 
-		var status string
-		status, err = runTuningRPC(ctx, &PB.ProfileInfo{Content: []byte(benchmarkByte)})
-		if err != nil {
-			return err
+		iter := 0
+		for {
+			reply, err := stream.Recv()
+			if err == io.EOF {
+				break
+			}
+			if err != nil {
+				return err
+			}
+			if reply.Status != utils.SUCCESS {
+				utils.Print(reply)
+				continue
+			}
+			if iter >= prj.Iterations {
+				break
+			}
+
+			benchmarkByte, err := prj.BenchMark()
+			if err != nil {
+				return err
+			}
+			if err := stream.Send(&PB.ProfileInfo{Content: []byte(benchmarkByte)}); err != nil {
+				return fmt.Errorf("client sends failure, error: %v", err)
+			}
+			iter++
 		}
+		return nil
+	})
 
-		if status == utils.SUCCESS {
-			break
-		}
-
-		iter++
+	if err != nil {
+		return err
 	}
 	return nil
 }
 
-func runTuningRPC(ctx *cli.Context, info *PB.ProfileInfo) (string, error) {
-	c, err := client.NewClientFromContext(ctx)
-	if err != nil {
-		return "", err
-	}
-
-	defer c.Close()
-	svc := PB.NewProfileMgrClient(c.Connection())
-	stream, err := svc.Tuning(CTX.Background(), info)
-	if err != nil {
-		return "", err
+func checkRestoreConfig(ctx *cli.Context) error {
+	if err := checkTuningCtx(ctx); err != nil {
+		return err
 	}
 
-	for {
-		reply, err := stream.Recv()
-
-		if err == io.EOF {
-			break
+	err := runTuningRPC(ctx, func(stream PB.ProfileMgr_TuningClient) error {
+		content := &PB.ProfileInfo{Name: ctx.String("project")}
+		if err := stream.Send(content); err != nil {
+			return fmt.Errorf("client sends failure, error: %v", err)
 		}
 
-		if err != nil {
-			return "", err
+		for {
+			reply, err := stream.Recv()
+			if err == io.EOF {
+				break
+			}
+			if err != nil {
+				return err
+			}
+			utils.Print(reply)
+
+			if reply.Status == utils.SUCCESS {
+				break
+			}
 		}
+		return nil
+	})
 
-		utils.Print(reply)
-
-		if reply.Status == utils.SUCCESS {
-			return reply.Status, nil
-		}
+	if err != nil {
+		return err
 	}
 
-	return "", nil
+	return nil
 }
 
 func checkTuningCtx(ctx *cli.Context) error {
@@ -190,15 +181,49 @@ func checkTuningCtx(ctx *cli.Context) error {
 	return nil
 }
 
-func checkRestoreConfig(ctx *cli.Context) error {
-	err := checkTuningCtx(ctx)
+func checkTuningPrjYaml(prj project.YamlPrjCli) error {
+	if len(prj.Project) < 1 || len(prj.Project) > 128 {
+		return fmt.Errorf("error: project name must be no less than 1 " +
+			"and no greater than 128 in yaml or yml")
+	}
+
+	if len(prj.Benchmark) < 1 {
+		return fmt.Errorf("error: benchmark must be specified in yaml or yml")
+	}
+
+	if len(prj.Evaluations) > 10 {
+		return fmt.Errorf("error: evaluations must be no greater than 10 "+
+			"in project %s", prj.Project)
+	}
+
+	if prj.Iterations < 11 {
+		return fmt.Errorf("error: iterations must be greater than 10 "+
+			"in project %s", prj.Project)
+	}
+	return nil
+}
+
+func runTuningRPC(ctx *cli.Context, sendTask func(stream PB.ProfileMgr_TuningClient) error) error {
+	c, err := client.NewClientFromContext(ctx)
 	if err != nil {
 		return err
 	}
-	_, err = runTuningRPC(ctx, &PB.ProfileInfo{Name: ctx.String("project")})
+
+	defer c.Close()
+	svc := PB.NewProfileMgrClient(c.Connection())
+	stream, err := svc.Tuning(CTX.Background())
 	if err != nil {
 		return err
 	}
+
+	if err := sendTask(stream); err != nil {
+		return err
+	}
+
+	if err := stream.CloseSend(); err != nil {
+		return fmt.Errorf("failed to close stream, error: %v", err)
+	}
+
 	return nil
 }
 

modules/server/profile/profile.go

@@ -21,7 +21,6 @@ import (
 	"atune/common/log"
 	"atune/common/models"
 	"atune/common/profile"
-	"atune/common/project"
 	"atune/common/registry"
 	"atune/common/schedule"
 	SVC "atune/common/service"
@@ -32,11 +31,11 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"os/exec"
 	"path"
-	"path/filepath"
 	"regexp"
 	"sort"
 	"strconv"
@@ -450,92 +449,59 @@ func (s *ProfileServer) Analysis(message *PB.AnalysisMessage, stream PB.ProfileM
 }
 
 // Tuning method calling the bayes search method to tuned parameters
-func (s *ProfileServer) Tuning(profileInfo *PB.ProfileInfo, stream PB.ProfileMgr_TuningServer) error {
-	//dynamic profle setting
-	data := profileInfo.GetName()
-	content := profileInfo.GetContent()
+func (s *ProfileServer) Tuning(stream PB.ProfileMgr_TuningServer) error {
+	if !s.TryLock() {
+		return fmt.Errorf("dynamic optimizer search or analysis has been in running")
+	}
+	defer s.Unlock()
 
 	ch := make(chan *PB.AckCheck)
+	defer close(ch)
 	go func() {
 		for value := range ch {
 			_ = stream.Send(value)
 		}
 	}()
 
-	// data == "" means in tuning process
-	if data == "" {
-		benchmark := tuning.BenchMark{Content: content}
-		isEnd, err := benchmark.DynamicTuned(ch)
-		if isEnd {
-			s.Unlock()
+	var optimizer = tuning.Optimizer{}
+	for {
+		reply, err := stream.Recv()
+		if err == io.EOF {
+			break
 		}
 		if err != nil {
 			return err
 		}
-		return nil
-	}
 
-	prjfound := false
-	var prjs []*project.YamlPrjSvr
-	err := filepath.Walk(config.DefaultTuningPath, func(path string, info os.FileInfo, err error) error {
-		if !info.IsDir() {
-			prj, ret := project.LoadProject(path)
-			if ret != nil {
-				log.Errorf("load %s faild(%s)", path, ret)
-				return ret
+		data := reply.Name
+		content := reply.Content
+
+		// data == "" means in tuning process
+		if data == "" {
+			optimizer.Content = content
+			err := optimizer.DynamicTuned(ch)
+			if err != nil {
+				return err
 			}
-			log.Infof("project:%s load %s success", prj.Project, path)
-			prjs = append(prjs, prj)
+			continue
 		}
-		return nil
-	})
 
-	if err != nil {
-		return err
-	}
-
-	var prj *project.YamlPrjSvr
-	for _, prj = range prjs {
-		if data == prj.Project {
-			log.Infof("found Project:%v", prj.Project)
-			prjfound = true
-			break
+		if err := tuning.CheckServerPrj(data, &optimizer); err != nil {
+			return err
 		}
-	}
-
-	if !prjfound {
-		log.Errorf("project:%s not found", data)
-		return fmt.Errorf("project:%s not found", data)
-	}
 
-	//content == nil means in restore config
-	if content == nil {
-		if !s.TryLock() {
-			return fmt.Errorf("dynamic optimizer search or analysis has been in running")
+		//content == nil means in restore config
+		if content == nil {
+			if err := optimizer.RestoreConfigTuned(ch); err != nil {
+				return err
+			}
+			continue
 		}
-		optimizer := tuning.Optimizer{Prj: prj}
-		err := optimizer.RestoreConfigTuned()
-		s.Unlock()
-		if err != nil {
+
+		optimizer.Content = content
+		if err = optimizer.InitTuned(ch); err != nil {
 			return err
 		}
-		return nil
-	}
-
-	log.Info("begin to dynamic optimizer search")
-	_ = stream.Send(&PB.AckCheck{Name: fmt.Sprintf("begin to dynamic optimizer search")})
-
-	if !s.TryLock() {
-		return fmt.Errorf("dynamic optimizer search or analysis has been in running")
-	}
-
-	optimizer := tuning.Optimizer{Prj: prj}
-	iter, _ := strconv.Atoi(string(content))
-	log.Infof("client ask iterations:%d", iter)
-	err = optimizer.InitTuned(ch, iter)
-	if err != nil {
-		s.Unlock()
-		return err
 	}
 
 	return nil