Perdita di memoria Golang relativa alle goroutine

Question

Ho un programma Go che funziona continuamente e fa affidamento interamente sulle goroutine + 1 thread manager. Il thread principale chiama semplicemente goroutine e dorme altrimenti.

C'è una perdita di memoria. Il programma utilizza sempre più memoria finché non scarica tutti i 16 GB di RAM + 32 GB di SWAP e poi ogni panico di goroutine. In realtà è la memoria del sistema operativo che provoca il panico, di solito il panico è fork/exec ./anotherapp: cannot allocate memory quando provo a eseguire anotherapp.

In questo caso, tutti i thread di lavoro verranno presi dal panico, recuperati e riavviati. Quindi ogni goroutine andrà nel panico, sarà recuperata e riavviata ... a quel punto l'utilizzo della memoria non diminuirà, rimane a 48 GB anche se ora non c'è praticamente nulla a disposizione. Ciò significa che tutte le goroutine saranno sempre nel panico perché non c'è mai abbastanza memoria, finché l'intero eseguibile non viene ucciso e riavviato completamente.

L'intera cosa è circa 50.000 linee, ma l'area effettiva problematica è il seguente:

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 
} 

Per quanto posso vedere, quando la funzione do (ultima riga) termina, sia avendo finito o dopo aver preso il panico, termina la funzione runit, che termina interamente la goroutine, il che significa che tutta la memoria da quella goroutine dovrebbe essere ora libera. Questo è ora ciò che accade. Quello che succede è che questa app usa sempre più memoria e più memoria fino a quando non diventa in grado di funzionare, tutto il panico delle goroutines runit, e tuttavia la memoria non diminuisce.

Il profilo non rivela nulla di sospetto. La perdita sembra essere al di fuori dell'ambito del profiler.

Answer 1

perche non invertendo il modello, vedere here o al di sotto ....

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