feat: proc tuner

util/proctuner/metric.go

@@ -0,0 +1,42 @@
+package proctuner
+
+import (
+	"runtime/metrics"
+)
+
+const latencyMetricName = "/sched/latencies:seconds"
+
+var (
+	metricSamples = []metrics.Sample{{Name: latencyMetricName}}
+)
+
+func fetchSchedLatency() (p50, p90, p99, max float64) {
+	metrics.Read(metricSamples)
+	histogram := metricSamples[0].Value.Float64Histogram()
+
+	var totalCount uint64
+	var latestIdx int
+	for idx, count := range histogram.Counts {
+		if count > 0 {
+			latestIdx = idx
+		}
+		totalCount += count
+	}
+	p50Count := totalCount / 2
+	p90Count := uint64(float64(totalCount) * 0.90)
+	p99Count := uint64(float64(totalCount) * 0.99)
+
+	var cursor uint64
+	for idx, count := range histogram.Counts {
+		cursor += count
+		if p99 == 0 && cursor >= p99Count {
+			p99 = histogram.Buckets[idx]
+		} else if p90 == 0 && cursor >= p90Count {
+			p90 = histogram.Buckets[idx]
+		} else if p50 == 0 && cursor >= p50Count {
+			p50 = histogram.Buckets[idx]
+		}
+	}
+	max = histogram.Buckets[latestIdx]
+	return
+}

util/proctuner/tuner.go

@@ -0,0 +1,121 @@
+package proctuner
+
+import (
+	"log"
+	"runtime"
+	"time"
+)
+
+const (
+	defaultLatencyAcceptable   float64 = 0.001 // 1ms
+	defaultP50LatencyThreshold float64 = 0.01  // 10ms
+	defaultP90LatencyThreshold float64 = 0.01  // 20ms
+	defaultP99LatencyThreshold float64 = 0.1   // 100ms
+	monitorFrequency                   = time.Second * 10
+	tuningFrequency                    = time.Minute
+)
+
+var originMaxProcs int
+var MaxProcs = runtime.NumCPU()
+
+type Option func(t *tuner)
+
+func WithMaxProcs(maxProcs int) Option {
+	return func(t *tuner) {
+		t.maxProcs = maxProcs
+	}
+}
+
+func WithP50LatencyThreshold(threshold float64) Option {
+	return func(t *tuner) {
+		t.p50Threshold = threshold
+	}
+}
+
+func WithP90LatencyThreshold(threshold float64) Option {
+	return func(t *tuner) {
+		t.p90Threshold = threshold
+	}
+}
+
+func WithP99LatencyThreshold(threshold float64) Option {
+	return func(t *tuner) {
+		t.p99Threshold = threshold
+	}
+}
+
+func Tuning(opts ...Option) error {
+	t := new(tuner)
+	t.acceptable = defaultLatencyAcceptable
+	t.p50Threshold = defaultP50LatencyThreshold
+	t.p90Threshold = defaultP90LatencyThreshold
+	t.p99Threshold = defaultP99LatencyThreshold
+	for _, opt := range opts {
+		opt(t)
+	}
+	maxProcs := t.maxProcs
+	originMaxProcs = runtime.GOMAXPROCS(0)
+	// default tuning
+	if maxProcs <= 0 {
+		maxProcs = originMaxProcs * 3
+	}
+	// reduce to MaxProcs
+	if maxProcs > MaxProcs {
+		maxProcs = MaxProcs
+	}
+	// no need to tuning
+	if maxProcs <= originMaxProcs {
+		return nil
+	}
+	t.minProcs = originMaxProcs
+	t.maxProcs = maxProcs
+	go t.tuning()
+	return nil
+}
+
+type tuner struct {
+	minProcs     int
+	maxProcs     int
+	acceptable   float64
+	p50Threshold float64
+	p90Threshold float64
+	p99Threshold float64
+}
+
+func (t *tuner) tuning() {
+	log.Printf("ProcTuning: MinProcs=%d MaxProcs=%d", t.minProcs, t.maxProcs)
+
+	var lastModify = time.Now()
+	var p50, p90, p99, max float64
+	var currentProcs = t.minProcs
+	for {
+		time.Sleep(monitorFrequency)
+		p50, p90, p99, max = fetchSchedLatency()
+		if p50 >= t.p50Threshold || p90 >= t.p90Threshold || p99 >= t.p99Threshold {
+			if currentProcs == t.maxProcs {
+				continue
+			}
+			now := time.Now()
+			if now.Sub(lastModify) < tuningFrequency {
+				continue
+			}
+			currentProcs += 1
+			runtime.GOMAXPROCS(currentProcs)
+			lastModify = now
+			log.Printf("ProcTuning: GOMAXPROCS from %d to %d", currentProcs-1, currentProcs)
+		} else if p99 <= t.acceptable {
+			if currentProcs <= t.minProcs {
+				continue
+			}
+			now := time.Now()
+			if now.Sub(lastModify) >= tuningFrequency {
+				continue
+			}
+			currentProcs -= 1
+			runtime.GOMAXPROCS(currentProcs)
+			lastModify = now
+			log.Printf("ProcTuning: GOMAXPROCS from %d to %d", currentProcs+1, currentProcs)
+		}
+		log.Printf("ProcTuning: Scheduler Latency[p50=%.6f,p90=%.6f,p99=%.6f,max=%.6f]", p50, p90, p99, max)
+	}
+}

util/proctuner/tuner_test.go

@@ -0,0 +1,33 @@
+package proctuner
+
+import (
+	"fmt"
+	"runtime"
+	"testing"
+	"time"
+)
+
+func BenchmarkTuner(b *testing.B) {
+	runtime.GOMAXPROCS(2)
+	_ = Tuning()
+
+	var running [5]int64
+	for i := 1; i <= 4; i++ {
+		go func(id int) {
+			sum := 0
+			for x := 0; ; x++ {
+				sum += x
+				if x%1000000 == 0 {
+					running[id]++
+				}
+			}
+		}(i)
+	}
+
+	total := 0
+	for x := 0; ; x++ {
+		time.Sleep(time.Second)
+		fmt.Println("main threads running", running[1:])
+		total += x
+	}
+}