[add] performance mode support (dev)

trheyi · Dec 27, 2023 · 85b6b69 · 85b6b69
1 parent 06857d8
commit 85b6b69
Show file tree

Hide file tree

Showing 4 changed files with 107 additions and 50 deletions.
diff --git a/runtime/v8/dispatcher.go b/runtime/v8/dispatcher.go
@@ -18,11 +18,18 @@ type RunnerMap struct {
 // Dispatcher is a runner dispatcher
 type Dispatcher struct {
 	availables RunnerMap
+	health     *Health
 	total      uint
 	min        uint
 	max        uint
 }
 
+// Health is the health check
+type Health struct {
+	missing uint // the missing runners
+	total   uint // the total runners
+}
+
 // the global dispatcher instance
 // initialize when the v8 start
 var dispatcher *Dispatcher = nil
@@ -31,10 +38,11 @@ var dispatcher *Dispatcher = nil
 func NewDispatcher(min, max uint) *Dispatcher {
 
 	// Test the min and max
-	min = 200
-	max = 200
+	// min = 10
+	// max = 200
 	return &Dispatcher{
 		availables: RunnerMap{data: make(map[uint]*Runner), mutex: &sync.RWMutex{}},
+		health:     &Health{missing: 0, total: 0},
 		total:      0,
 		min:        min,
 		max:        max,
@@ -46,7 +54,8 @@ func (dispatcher *Dispatcher) Start() error {
 	for i := uint(0); i < dispatcher.min; i++ {
 		dispatcher.create()
 	}
-	log.Info("[dispatcher] the dispatcher is started. runners %d", dispatcher.total)
+	log.Trace("[dispatcher] the dispatcher is started. runners %d", dispatcher.total)
+	go dispatcher.Scaling()
 	return nil
 }
 
@@ -66,7 +75,7 @@ func (dispatcher *Dispatcher) offline(runner *Runner) {
 
 func (dispatcher *Dispatcher) _offline(runner *Runner) {
 	delete(dispatcher.availables.data, runner.id)
-	log.Info("[dispatcher] runner %p offline (%d/%d) (%d/%d)", runner, len(dispatcher.availables.data), dispatcher.total, tempCount, cacheCount)
+	log.Trace("[dispatcher] runner %d offline (%d/%d) %s", runner.id, len(dispatcher.availables.data), dispatcher.total, dispatcher.health)
 	// create a new runner if the total runners are less than max
 	// if dispatcher.total < dispatcher.max {
 	// 	go dispatcher.create()
@@ -77,44 +86,87 @@ func (dispatcher *Dispatcher) online(runner *Runner) {
 	dispatcher.availables.mutex.Lock()
 	defer dispatcher.availables.mutex.Unlock()
 	dispatcher.availables.data[runner.id] = runner
-	log.Info("[dispatcher] runner %p online (%d/%d) (%d/%d)", runner, len(dispatcher.availables.data), dispatcher.total, tempCount, cacheCount)
+	log.Trace("[dispatcher] runner %d online (%d/%d) %s", runner.id, len(dispatcher.availables.data), dispatcher.total, dispatcher.health)
 }
 
 func (dispatcher *Dispatcher) create() {
 	runner := NewRunner(true)
-	go runner.Start()
+	ready := make(chan bool)
+	go runner.Start(ready)
+	<-ready
 	dispatcher.total++
-	log.Info("[dispatcher] runner %p create (%d/%d) (%d/%d)", runner, len(dispatcher.availables.data), dispatcher.total, tempCount, cacheCount)
+	log.Trace("[dispatcher] runner %d create (%d/%d) %s", runner.id, len(dispatcher.availables.data), dispatcher.total, dispatcher.health)
 }
 
 func (dispatcher *Dispatcher) destory(runner *Runner) {
 	dispatcher.availables.mutex.Lock()
 	defer dispatcher.availables.mutex.Unlock()
 	delete(dispatcher.availables.data, runner.id)
 	dispatcher.total--
-	log.Info("[dispatcher] runner %p destory (%d,%d) (%d/%d)", runner, len(dispatcher.availables.data), dispatcher.total, tempCount, cacheCount)
+	log.Trace("[dispatcher] runner %d destory (%d,%d) %s", runner.id, len(dispatcher.availables.data), dispatcher.total, dispatcher.health)
 }
 
 // Select select a free v8 runner
 func (dispatcher *Dispatcher) Select(timeout time.Duration) (*Runner, error) {
 	dispatcher.availables.mutex.Lock()
 	defer dispatcher.availables.mutex.Unlock()
+	go dispatcher.totalCount()
+
 	for _, runner := range dispatcher.availables.data {
-		cacheCount++
 		dispatcher._offline(runner)
-		fmt.Println("--------------------", runner.id)
-		fmt.Println("Select a free v8 runner id", runner.id, "count", len(dispatcher.availables.data))
+		log.Debug(fmt.Sprintf("--- %d -----------------\n", runner.id))
+		log.Debug(fmt.Sprintln("1.  Select a free v8 runner id", runner.id, "count", len(dispatcher.availables.data)))
 		return runner, nil
 	}
 
-	tempCount++
+	go dispatcher.missingCount()
 	runner := NewRunner(false)
-	go runner.Start()
+	ready := make(chan bool)
+	go runner.Start(ready)
+	<-ready
 	return runner, nil
 }
 
 // Scaling scale the v8 runners, check every 10 seconds
 // Release the v8 runners if the free runners are more than max size
 // Create a new v8 runner if the free runners are less than min size
 func (dispatcher *Dispatcher) Scaling() {
+
+	log.Info("[dispatcher] the dispatcher is scaling. min %d, max %d", dispatcher.min, dispatcher.max)
+
+	// check the free runners every 30 seconds
+	// @todo: release the free runners if the free runners are more than min size
+	ticker := time.NewTicker(30 * time.Second)
+	for {
+		select {
+		case <-ticker.C:
+			percent := float64(dispatcher.health.missing) / float64(dispatcher.health.total)
+			log.Trace("[dispatcher] the health percent is %f", percent)
+			dispatcher.health.Reset()
+			if percent > 0.2 {
+				for i := uint(0); i < dispatcher.max-dispatcher.total; i++ {
+					dispatcher.create()
+				}
+				continue
+			}
+		}
+	}
+}
+
+func (dispatcher *Dispatcher) missingCount() {
+	dispatcher.health.missing = dispatcher.health.missing + 1
+}
+
+func (dispatcher *Dispatcher) totalCount() {
+	dispatcher.health.total = dispatcher.health.total + 1
+}
+
+func (health *Health) String() string {
+	return fmt.Sprintf("missing: %d, total: %d", health.missing, health.total)
+}
+
+// Reset reset the health
+func (health *Health) Reset() {
+	health.missing = 0
+	health.total = 0
 }
diff --git a/runtime/v8/isolate.go b/runtime/v8/isolate.go
@@ -30,8 +30,7 @@ var isoReady chan *store.Isolate
 // in performance mode, the minSize isolates will be created
 func initialize() {
 
-	fmt.Println("initialize mode:", runtimeOption.Mode)
-
+	log.Info("[V8] initialize mode: %s", runtimeOption.Mode)
 	v8go.YaoInit(uint(runtimeOption.HeapSizeLimit / 1024 / 1024))
 
 	// Performance mode

diff --git a/runtime/v8/option.go b/runtime/v8/option.go
@@ -21,11 +21,11 @@ func EnablePrecompile() {
 func (option *Option) Validate() {
 
 	if option.MinSize == 0 {
-		option.MinSize = 2
+		option.MinSize = 50
 	}
 
 	if option.MaxSize == 0 {
-		option.MaxSize = 10
+		option.MaxSize = 100
 	}
 
 	if option.DefaultTimeout == 0 {
@@ -44,14 +44,14 @@ func (option *Option) Validate() {
 		option.Mode = "standard"
 	}
 
-	if option.MinSize > 100 {
-		log.Warn("[V8] the maximum value of initSize is 100")
-		option.MinSize = 100
+	if option.MinSize > 500 {
+		log.Warn("[V8] the maximum value of initSize is 500")
+		option.MinSize = 500
 	}
 
-	if option.MaxSize > 100 {
-		log.Warn("[V8] the maximum value of maxSize is 100")
-		option.MaxSize = 100
+	if option.MaxSize > 500 {
+		log.Warn("[V8] the maximum value of maxSize is 500")
+		option.MaxSize = 500
 	}
 
 	if option.MinSize > option.MaxSize {

diff --git a/runtime/v8/runner.go b/runtime/v8/runner.go
@@ -5,7 +5,6 @@ import (
 	"time"
 
 	"github.com/yaoapp/gou/runtime/v8/bridge"
-	"github.com/yaoapp/kun/exception"
 	"github.com/yaoapp/kun/log"
 	"rogchap.com/v8go"
 )
@@ -28,9 +27,6 @@ type Runner struct {
 
 var seq uint = 0
 
-var tempCount uint64 = 0
-var cacheCount uint64 = 0
-
 const (
 	// RunnerStatusInit is the runner status init
 	RunnerStatusInit uint8 = iota
@@ -68,39 +64,40 @@ const (
 
 // NewRunner create a new v8 runner
 func NewRunner(keepalive bool) *Runner {
-	iso := v8go.YaoNewIsolate()
-	tmpl := MakeTemplate(iso)
-	ctx := v8go.NewContext(iso, tmpl)
-	status := RunnerStatusInit
-	if !keepalive {
-		status = RunnerStatusReady
-	}
-
 	seq++
 	return &Runner{
 		id:        seq,
-		iso:       iso,
-		ctx:       ctx,
+		iso:       nil,
+		ctx:       nil,
 		signal:    make(chan uint8, 2),
 		keepalive: keepalive,
-		status:    status,
+		status:    RunnerStatusInit,
 	}
 }
 
 // Start start the v8 runner
-func (runner *Runner) Start() error {
+func (runner *Runner) Start(ready chan bool) error {
 
 	// Set the status to free
 	if runner.status != RunnerStatusInit {
-		return fmt.Errorf("[runner] you can't start a runner with status: [%d]", runner.status)
+		err := fmt.Errorf("[runner] you can't start a runner with status: [%d]", runner.status)
+		log.Error(err.Error())
+		return err
 	}
 
+	iso := v8go.YaoNewIsolate()
+	tmpl := MakeTemplate(iso)
+	ctx := v8go.NewContext(iso, tmpl)
+	runner.iso = iso
+	runner.ctx = ctx
+
 	runner.status = RunnerStatusReady
 	if runner.keepalive {
 		dispatcher.online(runner)
 	}
 
 	ticker := time.NewTicker(time.Millisecond * 50)
+	ready <- true
 
 	// Command loop
 	for {
@@ -155,7 +152,7 @@ func (runner *Runner) Exec(script *Script) interface{} {
 	runner.status = RunnerStatusRunning
 	runner.script = script
 	runner.chResp = make(chan interface{})
-	fmt.Println("Exec a script to the v8 runner to execute", runner.id)
+	log.Debug(fmt.Sprintln("2.  Exec a script to the v8 runner to execute", "runner.id:", runner.id, "status:", runner.status, "keepalive:", runner.keepalive, len(runner.signal)))
 
 	runner.signal <- RunnerCommandExec
 	select {
@@ -174,19 +171,26 @@ func (runner *Runner) exec() {
 	defer func() {
 		go func() {
 			if !runner.keepalive {
+				log.Debug(fmt.Sprintln("3.1 Send a destory signal to the v8 runner", "runner.id:", runner.id, "status:", runner.status, runner.keepalive))
 				runner.signal <- RunnerCommandDestroy
-				fmt.Println("Send a destory signal to the v8 runner done", runner.id, runner.status)
+				log.Debug(fmt.Sprintln("3.2 Send a destory signal to the v8 runner done", "runner.id:", runner.id, "status:", runner.status))
 				return
 			}
 			runner.signal <- RunnerCommandClose
-			fmt.Println("Send a close signal to the v8 runner done", runner.id, runner.status)
+			log.Debug(fmt.Sprintln("3.  Send a close signal to the v8 runner done", "runner.id:", runner.id, "status:", runner.status, runner.keepalive))
 		}()
 	}()
 
+	// runner.chResp <- "OK"
+	runner._exec()
+}
+
+func (runner *Runner) _exec() {
+
 	// Create instance of the script
 	instance, err := runner.iso.CompileUnboundScript(runner.script.Source, runner.script.File, v8go.CompileOptions{})
 	if err != nil {
-		exception.New("scripts.%s.%s %s", 500, runner.script.ID, runner.method, err.Error()).Throw()
+		runner.chResp <- err
 		return
 	}
 	v, err := instance.Run(runner.ctx)
@@ -203,15 +207,14 @@ func (runner *Runner) exec() {
 		Global: runner.global,
 	})
 	if err != nil {
-		exception.New("scripts.%s.%s %s", 500, runner.script.ID, runner.method, err.Error()).Throw()
+		runner.chResp <- err
 		return
 	}
 
 	// Run the method
 	jsArgs, err := bridge.JsValues(runner.ctx, runner.args)
 	if err != nil {
-		log.Error("scripts.%s.%s %s", runner.script.ID, runner.method, err.Error())
-		exception.New(err.Error(), 500).Throw()
+		runner.chResp <- err
 		return
 
 	}
@@ -234,9 +237,8 @@ func (runner *Runner) exec() {
 
 func (runner *Runner) close() {
 
-	fmt.Println("close the runner", runner.id, runner.status)
-	fmt.Println("--------------------", runner.id, "END")
-	fmt.Println("")
+	log.Debug(fmt.Sprintln("4.  close the runner", "runner.id:", runner.id, "status:", runner.status))
+	log.Debug(fmt.Sprintf("--- %d end -----------------\n\n", runner.id))
 
 	if runner.keepalive {
 		runner.reset()
@@ -256,6 +258,10 @@ func (runner *Runner) close() {
 
 // destory the runner
 func (runner *Runner) destory() {
+
+	log.Debug(fmt.Sprintln("4.  destory the runner", "runner.id:", runner.id, "status:", runner.status))
+	log.Debug(fmt.Sprintf("--- %d end -----------------\n\n", runner.id))
+
 	runner.status = RunnerStatusDestroy
 	if runner.signal != nil {
 		close(runner.signal)