Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
4.9 kB · 221 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222package media
import ( "bufio" "fmt" "os" "path/filepath" "regexp" "runtime" "runtime/debug" "strconv" "strings" "sync" "syscall" "testing" "time"
"github.com/acarl005/stripansi" "github.com/cenkalti/backoff/v5" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/gstinit")
var streamplaceTestCount = 50
func init() { testRunsStr := os.Getenv("STREAMPLACE_TEST_COUNT") if testRunsStr != "" { var err error streamplaceTestCount, err = strconv.Atoi(testRunsStr) if err != nil { panic(fmt.Sprintf("STREAMPLACE_TEST_COUNT is not a number: %s", testRunsStr)) } }}
var LeakTestMutex sync.Mutex
const IgnoreLeaks = "STREAMPLACE_IGNORE_LEAKS"const ShowTrace = "STREAMPLACE_SHOW_TRACE"const GSTDebugNeeded = "leaks:9,GST_TRACER:9"const LeakLine = "GST_TRACER :0:: object-alive"
var LeakDoneRegex = regexp.MustCompile(`listed\s+(\d+)\s+alive\s+objects`)
var LeakReport = []string{}var LeakReportMutex sync.Mutexvar LeakDoneCh = make(chan struct{})
func TestMain(m *testing.M) { // When the parent test re-execs us as an isolated ingest worker (mirroring the // `streamplace ingest-worker` subcommand), act as that worker and exit — before // any leak-tracer setup, so the child stays a clean media pipeline. if len(os.Args) > 1 && os.Args[1] == "ingest-worker" { os.Exit(runIngestWorkerHelper()) } if len(os.Args) > 1 && os.Args[1] == "rtmp-push-worker" { os.Exit(runRTMPPushWorkerHelper()) } if os.Getenv(IgnoreLeaks) != "" { gstinit.InitGST() os.Exit(m.Run()) return } showTrace := os.Getenv(ShowTrace) != "" gstDebug := os.Getenv("GST_DEBUG") if gstDebug == "" { gstDebug = GSTDebugNeeded } else { gstDebug = fmt.Sprintf("%s,%s", gstDebug, GSTDebugNeeded) } os.Setenv("GST_DEBUG", gstDebug) os.Setenv("GST_TRACERS", "leaks") os.Setenv("GST_LEAKS_TRACER_SIG", "1") debug.SetGCPercent(5)
f, err := os.MkdirTemp("", "") if err != nil { panic(err) } fName := filepath.Join(f, "leak.log") err = syscall.Mkfifo(fName, 0640) if err != nil { panic(err) } os.Setenv("GST_DEBUG_FILE", fName)
go func() { pipe, err := os.OpenFile(fName, os.O_RDONLY, 0640) if err != nil { panic(err) } defer pipe.Close() // Read and print each line from FD scanner := bufio.NewScanner(pipe) for scanner.Scan() { lineAnsi := scanner.Text() line := stripansi.Strip(lineAnsi) if !strings.Contains(line, " TRACE ") || showTrace { fmt.Println(lineAnsi) } if strings.Contains(line, LeakLine) { LeakReportMutex.Lock() LeakReport = append(LeakReport, line) LeakReportMutex.Unlock() } else if LeakDoneRegex.MatchString(line) { LeakDoneCh <- struct{}{} } else { continue } } if err := scanner.Err(); err != nil { panic(err) } }() gstinit.InitGST() os.Exit(m.Run())}
func parseLeakReport() int { LeakReportMutex.Lock() // the typefind stuff is intentionally kept around, i guess. and each has one cap, so: typeFindCount := 0 for _, line := range LeakReport { if strings.Contains(line, "GstTypeFindFactory") { typeFindCount += 1 } } finalCount := len(LeakReport) - (typeFindCount * 2) if finalCount < 0 { finalCount = 0 } LeakReportMutex.Unlock() return finalCount}
// Often the GC is instant, but sometimes it takes a while. So, we retry a few times// with exponential backoff, giving the GC more time to do its thing.func getLeakCount(t *testing.T) int { ticker := backoff.NewTicker(backoff.NewExponentialBackOff()) defer ticker.Stop() var leaks int for i := 0; i < 10; i++ { leaks = getLeakCountInner(t) if leaks == 0 { return leaks } if i < 9 { <-ticker.C } } return leaks}
func getLeakCountInner(t *testing.T) int { if os.Getenv(IgnoreLeaks) != "" { return 0 } process, err := os.FindProcess(os.Getpid()) require.NoError(t, err)
LeakReportMutex.Lock() LeakReport = []string{} LeakReportMutex.Unlock()
flushes := 2
for range flushes { ch := make(chan struct{}) done := false go func() { thing := &[]byte{} runtime.SetFinalizer(thing, func(thing *[]byte) { done = true ch <- struct{}{} }) }()
go func() { runtime.GC() runtime.GC() for !done {
runtime.GC() runtime.GC() time.Sleep(500 * time.Millisecond) } <-ch }() }
err = process.Signal(os.Signal(syscall.SIGUSR1)) require.NoError(t, err)
<-LeakDoneCh
after := parseLeakReport() return after}
func checkGStreamerLeaks(t *testing.T, expected int) { if os.Getenv(IgnoreLeaks) != "" { return } leaks := getLeakCount(t) if leaks > expected { count := parseLeakReport() require.Equal(t, expected, count, "Leaks found") }}
func withNoGSTLeaks(t *testing.T, f func()) { LeakTestMutex.Lock() defer LeakTestMutex.Unlock() gstinit.InitGST() before := getLeakCount(t) defer checkGStreamerLeaks(t, before) // ignore := goleak.IgnoreCurrent() // defer goleak.VerifyNone(t, ignore) f()}