Golang Memory Leak in Bezug auf Goroutines

Question

Ich habe ein Go-Programm, das kontinuierlich läuft und vollständig auf goroutines + 1 manager Thread basiert. Der Haupt-Thread ruft einfach Goroutines an und schläft ansonsten.

Es liegt ein Speicherleck vor. Das Programm verwendet mehr und mehr Speicher, bis es alle 16 GB RAM + 32 GB SWAP ableitet und dann jede goroutine in Panik gerät. Es ist tatsächlich OS-Speicher, der die Panik verursacht, in der Regel ist die Panik fork/exec ./anotherapp: cannot allocate memory, wenn ich versuche, anotherapp auszuführen.

Wenn dies passiert, werden alle Worker-Threads in Panik versetzt und wiederhergestellt und neu gestartet. So wird jede goroutine in Panik geraten, wiederhergestellt und neu gestartet ... an diesem Punkt wird die Speichernutzung nicht abnehmen, sie bleibt bei 48GB, obwohl jetzt praktisch nichts zugewiesen ist. Das bedeutet, dass alle Goroutinen immer in Panik geraten, da nie genug Speicher zur Verfügung steht, bis die gesamte ausführbare Datei beendet und vollständig neu gestartet wurde.

Das Ganze ist etwa 50.000 Zeilen, aber der eigentliche Problembereich ist wie folgt:

type queue struct { 
    identifier string 
    type bool 
} 

func main() { 

    // Set number of gorountines that can be run 
    var xthreads int32 = 10 
    var usedthreads int32 
    runtime.GOMAXPROCS(14) 
    ready := make(chan *queue, 5) 

    // Start the manager goroutine, which prepared identifiers in the background ready for processing, always with 5 waiting to go 
    go manager(ready) 

    // Start creating goroutines to process as they are ready 
    for obj := range ready { // loops through "ready" channel and waits when there is nothing 

     // This section uses atomic instead of a blocking channel in an earlier attempt to stop the memory leak, but it didn't work 
     for atomic.LoadInt32(&usedthreads) >= xthreads { 
      time.Sleep(time.Second) 
     } 
     debug.FreeOSMemory() // Try to clean up the memory, also did not stop the leak 
     atomic.AddInt32(&usedthreads, 1) // Mark goroutine as started 

     // Unleak obj, probably unnecessary, but just to be safe 
     copy := new(queue) 
     copy.identifier = unleak.String(obj.identifier) // unleak is a 3rd party package that makes a copy of the string 
     copy.type = obj.type 
     go runit(copy, &usedthreads) // Start the processing thread 

    } 

    fmt.Println(`END`) // This should never happen as the channels are never closed 
} 

func manager(ready chan *queue) { 
    // This thread communicates with another server and fills the "ready" channel 
} 

// This is the goroutine 
func runit(obj *queue, threadcount *int32) { 
    defer func() { 
     if r := recover(); r != nil { 
      // Panicked 
      erstring := fmt.Sprint(r) 
      reportFatal(obj.identifier, erstring) 
     } else { 
      // Completed successfully 
      reportDone(obj.identifier) 
     } 
     atomic.AddInt32(threadcount, -1) // Mark goroutine as finished 
    }() 
    do(obj) // This function does the actual processing 
} 

Soweit ich sehen kann, wenn die do Funktion (letzte Zeile) endet, entweder durch mit fertiger oder In Panik geraten endet dann die runit Funktion, die die Goroutine vollständig beendet, was bedeutet, dass der gesamte Speicher von dieser Goroutine nun frei sein sollte. Das ist jetzt was passiert. Was passiert ist, dass diese App nur mehr und mehr und mehr Speicher verwendet, bis es nicht mehr funktionsfähig ist, und die Speicher nicht abnimmt.

Profiling zeigt nichts Verdächtiges. Das Leck scheint außerhalb des Bereichs des Profilers zu liegen.

Answer 1

Bitte beachten Sie die Muster invertiert, siehe here oder unten ....

package main 

import (
    "log" 
    "math/rand" 
    "sync" 
    "time" 
) 

// I do work 
func worker(id int, work chan int) { 
    for i := range work { 
     // Work simulation 
     log.Printf("Worker %d, sleeping for %d seconds\n", id, i) 
     time.Sleep(time.Duration(rand.Intn(i)) * time.Second) 
    } 
} 

// Return some fake work 
func getWork() int { 
    return rand.Intn(2) + 1 
} 

func main() { 
    wg := new(sync.WaitGroup) 
    work := make(chan int) 

    // run 10 workers 
    for i := 0; i < 10; i++ { 
     wg.Add(1) 
     go func(i int) { 
      worker(i, work) 
      wg.Done() 
     }(i) 
    } 

    // main "thread" 
    for i := 0; i < 100; i++ { 
     work <- getWork() 
    } 

    // signal there is no more work to be done 
    close(work) 

    // Wait for the workers to exit 
    wg.Wait() 
}