// backend\performance.go package main import ( "bytes" "context" "encoding/json" "io" "math" "net/http" "os" "runtime" "strconv" "strings" "sync" "sync/atomic" "time" gocpu "github.com/shirou/gopsutil/v3/cpu" godisk "github.com/shirou/gopsutil/v3/disk" gohost "github.com/shirou/gopsutil/v3/host" ) // Server-Startzeit für Perf/Uptime var serverStartedAt = time.Now() var lastCPUUsageBits uint64 // atomic float64 bits var cpuTemperatureCache = struct { sync.Mutex value float64 available bool sampledAt time.Time }{} func setLastCPUUsage(v float64) { atomic.StoreUint64(&lastCPUUsageBits, math.Float64bits(v)) } func getLastCPUUsage() float64 { return math.Float64frombits(atomic.LoadUint64(&lastCPUUsageBits)) } func getMaxCPUTemperatureC() (float64, bool) { now := time.Now() cpuTemperatureCache.Lock() if now.Sub(cpuTemperatureCache.sampledAt) < 15*time.Second { value := cpuTemperatureCache.value available := cpuTemperatureCache.available cpuTemperatureCache.Unlock() return value, available } cpuTemperatureCache.Unlock() ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() temps, err := gohost.SensorsTemperaturesWithContext(ctx) if err != nil || len(temps) == 0 { cpuTemperatureCache.Lock() cpuTemperatureCache.value = 0 cpuTemperatureCache.available = false cpuTemperatureCache.sampledAt = now cpuTemperatureCache.Unlock() return 0, false } maxCPU := 0.0 maxAny := 0.0 for _, temp := range temps { v := temp.Temperature if math.IsNaN(v) || math.IsInf(v, 0) || v <= 0 || v > 150 { continue } if v > maxAny { maxAny = v } key := strings.ToLower(strings.TrimSpace(temp.SensorKey)) if strings.Contains(key, "cpu") || strings.Contains(key, "core") || strings.Contains(key, "package") || strings.Contains(key, "tdie") || strings.Contains(key, "tctl") || strings.Contains(key, "k10temp") { if v > maxCPU { maxCPU = v } } } value := maxCPU if value <= 0 { value = maxAny } available := value > 0 cpuTemperatureCache.Lock() cpuTemperatureCache.value = value cpuTemperatureCache.available = available cpuTemperatureCache.sampledAt = now cpuTemperatureCache.Unlock() return value, available } func startCPUUsageSampler(ctx context.Context) { _, _ = gocpu.Percent(200*time.Millisecond, false) t := time.NewTicker(2 * time.Second) go func() { defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: p, err := gocpu.Percent(0, false) if err != nil || len(p) == 0 { continue } usage := p[0] if math.IsNaN(usage) || math.IsInf(usage, 0) || usage < 0 { continue } if usage > 100 { usage = 100 } setLastCPUUsage(usage) } } }() } // ---------- Feste ffmpeg/ffprobe Limiter ---------- type FixedLimiter chan struct{} func NewFixedLimiter(n int) FixedLimiter { if n < 1 { n = 1 } return make(FixedLimiter, n) } func (l FixedLimiter) Acquire(ctx context.Context) error { if l == nil { return nil } if ctx == nil { ctx = context.Background() } select { case l <- struct{}{}: return nil case <-ctx.Done(): return ctx.Err() } } func (l FixedLimiter) Release() { if l == nil { return } select { case <-l: default: } } func (l FixedLimiter) Limit() int { return cap(l) } func (l FixedLimiter) InUse() int { return len(l) } var ( previewSem FixedLimiter thumbSem FixedLimiter genSem FixedLimiter durSem FixedLimiter ) func envInt(name string) (int, bool) { v := strings.TrimSpace(os.Getenv(name)) if v == "" { return 0, false } n, err := strconv.Atoi(v) if err != nil { return 0, false } return n, true } func initFFmpegLimiters() { previewN := 2 thumbN := 4 genN := 2 durN := 8 if n, ok := envInt("PREVIEW_WORKERS"); ok && n > 0 { previewN = n } if n, ok := envInt("THUMB_WORKERS"); ok && n > 0 { thumbN = n } if n, ok := envInt("GEN_WORKERS"); ok && n > 0 { genN = n } if n, ok := envInt("DUR_WORKERS"); ok && n > 0 { durN = n } previewSem = NewFixedLimiter(previewN) thumbSem = NewFixedLimiter(thumbN) genSem = NewFixedLimiter(genN) durSem = NewFixedLimiter(durN) setLastCPUUsage(0) startCPUUsageSampler(context.Background()) } func buildPerfSnapshot() map[string]any { var ms runtime.MemStats runtime.ReadMemStats(&ms) s := getSettings() recordDir, _ := resolvePathRelativeToApp(s.RecordDir) var diskFreeBytes uint64 var diskTotalBytes uint64 var diskUsedPercent float64 diskPath := recordDir if recordDir != "" { if u, err := godisk.Usage(recordDir); err == nil && u != nil { diskFreeBytes = u.Free diskTotalBytes = u.Total diskUsedPercent = u.UsedPercent } } // ✅ Dynamische Disk-Schwellen (2× inFlight, Resume = +3GB) pauseGB, resumeGB, inFlight, pauseNeed, resumeNeed := computeDiskThresholds() resp := map[string]any{ "ts": time.Now().UTC().Format(time.RFC3339Nano), "serverMs": time.Now().UTC().UnixMilli(), "uptimeSec": time.Since(serverStartedAt).Seconds(), "cpuPercent": func() float64 { v := getLastCPUUsage() if math.IsNaN(v) || math.IsInf(v, 0) || v < 0 { return 0 } if v > 100 { return 100 } return v }(), "cpuTemperatureC": func() float64 { v, ok := getMaxCPUTemperatureC() if !ok { return 0 } return math.Round(v*10) / 10 }(), "cpuTemperatureAvailable": func() bool { _, ok := getMaxCPUTemperatureC() return ok }(), "diskPath": diskPath, "diskFreeBytes": diskFreeBytes, "diskTotalBytes": diskTotalBytes, "diskUsedPercent": diskUsedPercent, "diskEmergency": atomic.LoadInt32(&diskEmergency) == 1, "diskPauseBelowGB": pauseGB, "diskResumeAboveGB": resumeGB, "diskInFlightBytes": inFlight, "diskInFlightHuman": formatBytesSI(u64ToI64(inFlight)), "diskPauseNeedBytes": pauseNeed, "diskPauseNeedHuman": formatBytesSI(u64ToI64(pauseNeed)), "diskResumeNeedBytes": resumeNeed, "diskResumeNeedHuman": formatBytesSI(u64ToI64(resumeNeed)), "goroutines": runtime.NumGoroutine(), "mem": map[string]any{ "alloc": ms.Alloc, "heapAlloc": ms.HeapAlloc, "heapInuse": ms.HeapInuse, "sys": ms.Sys, "numGC": ms.NumGC, }, } sem := map[string]any{} if previewSem != nil { sem["preview"] = map[string]any{ "inUse": previewSem.InUse(), "limit": previewSem.Limit(), } } if thumbSem != nil { sem["thumb"] = map[string]any{ "inUse": thumbSem.InUse(), "limit": thumbSem.Limit(), } } if genSem != nil { sem["gen"] = map[string]any{ "inUse": genSem.InUse(), "limit": genSem.Limit(), } } if durSem != nil { sem["dur"] = map[string]any{ "inUse": durSem.InUse(), "limit": durSem.Limit(), } } if len(sem) > 0 { resp["sem"] = sem } return resp } func pingHandler(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet && r.Method != http.MethodHead { http.Error(w, "Nur GET/HEAD erlaubt", http.StatusMethodNotAllowed) return } w.Header().Set("Cache-Control", "no-store") w.WriteHeader(http.StatusNoContent) } func perfHandler(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "Nur GET erlaubt", http.StatusMethodNotAllowed) return } resp := buildPerfSnapshot() w.Header().Set("Content-Type", "application/json") w.Header().Set("Cache-Control", "no-store") _ = json.NewEncoder(w).Encode(resp) } func perfStreamHandler(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "Nur GET erlaubt", http.StatusMethodNotAllowed) return } fl, ok := w.(http.Flusher) if !ok { http.Error(w, "Streaming nicht unterstützt", http.StatusInternalServerError) return } ms := 5000 if q := r.URL.Query().Get("ms"); q != "" { if v, err := strconv.Atoi(q); err == nil { if v < 1000 { v = 1000 } if v > 30000 { v = 30000 } ms = v } } w.Header().Set("Content-Type", "text/event-stream; charset=utf-8") w.Header().Set("Cache-Control", "no-store") w.Header().Set("Connection", "keep-alive") w.Header().Set("X-Accel-Buffering", "no") ctx := r.Context() send := func() error { payload := buildPerfSnapshot() var buf bytes.Buffer if err := json.NewEncoder(&buf).Encode(payload); err != nil { return err } _, _ = io.WriteString(w, "event: perf\n") _, _ = io.WriteString(w, "data: ") _, _ = w.Write(buf.Bytes()) _, _ = io.WriteString(w, "\n") fl.Flush() return nil } _ = send() t := time.NewTicker(time.Duration(ms) * time.Millisecond) hb := time.NewTicker(15 * time.Second) defer t.Stop() defer hb.Stop() for { select { case <-ctx.Done(): return case <-t.C: _ = send() case <-hb.C: _, _ = io.WriteString(w, ": keep-alive\n\n") fl.Flush() } } }