Wyciek pamięci Golang dotyczący Goroutines

Question

Mam program Go, który działa nieprzerwanie i polega całkowicie na wątku goroutines + 1 manager. Główna nić po prostu nazywa goroutiny i inaczej śpi.

Występuje przeciek pamięci. Program zużywa coraz więcej pamięci, aż do wyczerpania wszystkich 16GB RAM + 32GB SWAP, a następnie każdej paniki goroutine. To właśnie pamięć systemu operacyjnego wywołuje panikę, zwykle panika jest fork/exec ./anotherapp: cannot allocate memory podczas próby wykonania anotherapp.

Gdy to się stanie, wszystkie wątki robocze będą panikować, a następnie zostaną odzyskane i zrestartowane. Tak więc każda goroutine wpadnie w panikę, zostanie odzyskana i uruchomiona ponownie ... w którym to momencie zużycie pamięci nie zmniejszy się, pozostanie na poziomie 48 GB, mimo że nie ma praktycznie nic przydzielonego. Oznacza to, że wszystkie goryle będą zawsze panikować, ponieważ nigdy nie ma wystarczającej ilości pamięci, dopóki cały plik wykonywalny nie zostanie całkowicie zabity i zrestartowany.

Cała sprawa jest około 50.000 linii, ale rzeczywisty obszar problematyczny jest następująca:

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 
} 

O ile widzę, gdy funkcja do (ostatnia linia) kończy się, albo poprzez gotowych lub wpadając w panikę, kończy się funkcja runit, która kończy się całkowicie, co oznacza, że ​​cała pamięć z tej goroutine powinna być teraz wolna. Oto co się dzieje. Zdarza się, że ta aplikacja zużywa coraz więcej pamięci, aż stanie się niezdolna do działania, wszystkie paniki gorylami, a pamięć nie zmniejsza się.

Profilowanie nie ujawnia niczego podejrzanego. Wydaje się, że wyciek wykracza poza zasięg profilera.

Answer 1

Proszę rozważyć odwracanie wzór, patrz here lub poniżej ....

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() 
}