Come definiresti un gruppo di goroutine da eseguire contemporaneamente a Golang?

TL; TR: Per favore, vai all’ultima parte e dimmi come risolveresti questo problema.

Ho iniziato a usare Golang stamattina venendo da Python. Voglio chiamare un eseguibile closed-source da Go più volte, con un po ‘ di concorrenza, con diversi argomenti da linea di comando. Il mio codice risultante sta funzionando molto bene, ma mi piacerebbe avere il tuo input per migliorarlo. Dato che sono in una fase di apprendimento precoce, spiegherò anche il mio stream di lavoro.

Per semplicità, supponiamo qui che questo “programma closed-source esterno” sia zenity , uno strumento da riga di comando di Linux in grado di visualizzare windows di messaggi grafici dalla riga di comando.

Chiamare un file eseguibile da Go

Quindi, in Go, vorrei andare così:

 package main import "os/exec" func main() { cmd := exec.Command("zenity", "--info", "--text='Hello World'") cmd.Run() } 

Questo dovrebbe funzionare perfettamente. Nota che .Run() è un equivalente funzionale a .Start() seguito da .Wait() . Questo è grandioso, ma se volessi eseguire questo programma solo una volta, l’intera roba di programmazione non sarebbe valsa la pena. Quindi facciamolo più volte.

Chiamare un eseguibile più volte

Ora che ho avuto questo lavoro, mi piacerebbe chiamare il mio programma più volte, con argomenti personalizzati della riga di comando (qui solo i per semplicità).

 package main import ( "os/exec" "strconv" ) func main() { NumEl := 8 // Number of times the external program is called for i:=0; i<NumEl; i++ { cmd := exec.Command("zenity", "--info", "--text='Hello from iteration n." + strconv.Itoa(i) + "'") cmd.Run() } } 

Ok, ce l’abbiamo fatta! Ma non riesco ancora a vedere il vantaggio di Go over Python … Questo pezzo di codice viene effettivamente eseguito in modo seriale. Ho una CPU multi-core e mi piacerebbe approfittarne. Quindi aggiungiamo un po ‘di concorrenza con le goroutine.

Goroutine, o un modo per rendere il mio programma parallelo

a) Primo tentativo: aggiungi “vai” ovunque

Riscriviamo il nostro codice per rendere le cose più facili da chiamare e riutilizzare e aggiungere la famosa parola chiave go :

 package main import ( "os/exec" "strconv" ) func main() { NumEl := 8 for i:=0; i<NumEl; i++ { go callProg(i) // <--- There! } } func callProg(i int) { cmd := exec.Command("zenity", "--info", "--text='Hello from iteration n." + strconv.Itoa(i) + "'") cmd.Run() } 

Niente! Qual è il problema? Tutte le goroutine vengono eseguite contemporaneamente. Non so davvero perché la zenity non sia eseguita ma AFAIK, il programma Go è uscito prima che il programma esterno di zenity potesse essere inizializzato. Ciò è stato confermato dall’uso del tempo. time.Sleep : attendere un paio di secondi è stato sufficiente per consentire l’avvio dell’istanza 8 di zenity. Non so se questo può essere considerato un bug però.

Per peggiorare le cose, il vero programma che in realtà mi piacerebbe chiamare richiede un po ‘di tempo per essere eseguito. Se eseguo 8 istanze di questo programma in parallelo sulla mia CPU a 4 core, exec.Command un po ‘di tempo facendo un sacco di cambio di contesto … Non so come si comportano Go goroutines, ma exec.Command lancerà zenity 8 volte in 8 fili diversi. Per renderlo ancora peggio, voglio eseguire questo programma più di 100.000 volte. Fare tutto questo in una volta in goroutine non sarà affatto efficiente. Comunque, mi piacerebbe sfruttare la mia CPU a 4 core!

b) Secondo tentativo: utilizzare pool di goroutine

Le risorse online tendono a raccomandare l’uso di sync.WaitGroup per questo tipo di lavoro. Il problema con questo approccio è che si sta fondamentalmente lavorando con lotti di goroutine: se creo WaitGroup di 4 membri, il programma Go attenderà che tutti e 4 i programmi esterni finiscano prima di chiamare un nuovo batch di 4 programmi. Questo non è efficiente: la CPU è sprecata, ancora una volta.

Altre risorse consigliano l’uso di un canale bufferizzato per eseguire il lavoro:

 package main import ( "os/exec" "strconv" ) func main() { NumEl := 8 // Number of times the external program is called NumCore := 4 // Number of available cores c := make(chan bool, NumCore - 1) for i:=0; i<NumEl; i++ { go callProg(i, c) c <- true // At the NumCoreth iteration, c is blocking } } func callProg(i int, c chan bool) { defer func () {<- c}() cmd := exec.Command("zenity", "--info", "--text='Hello from iteration n." + strconv.Itoa(i) + "'") cmd.Run() } 

Questo sembra brutto. I canali non erano destinati a questo scopo: sto sfruttando un effetto collaterale. Adoro il concetto di defer ma odio dover dichiarare una funzione (anche una lambda) per estrarre un valore dal canale fittizio che ho creato. Oh, e ovviamente, usare un canale fittizio è di per sé brutto.

c) Terzo tentativo: morire quando tutti i bambini sono morti

Ora abbiamo quasi finito. Devo solo prendere in considerazione un altro effetto collaterale: il programma Go si chiude prima che tutti i pop-up di zenity siano chiusi. Questo perché quando il ciclo è finito (all’ottava iterazione), nulla impedisce al programma di finire. Questa volta, sync.WaitGroup sarà utile.

 package main import ( "os/exec" "strconv" "sync" ) func main() { NumEl := 8 // Number of times the external program is called NumCore := 4 // Number of available cores c := make(chan bool, NumCore - 1) wg := new(sync.WaitGroup) wg.Add(NumEl) // Set the number of goroutines to (0 + NumEl) for i:=0; i<NumEl; i++ { go callProg(i, c, wg) c <- true // At the NumCoreth iteration, c is blocking } wg.Wait() // Wait for all the children to die close(c) } func callProg(i int, c chan bool, wg *sync.WaitGroup) { defer func () { <- c wg.Done() // Decrease the number of alive goroutines }() cmd := exec.Command("zenity", "--info", "--text='Hello from iteration n." + strconv.Itoa(i) + "'") cmd.Run() } 

Fatto.

Le mie domande

  • Conosci qualche altro modo corretto per limitare il numero di goroutine eseguite contemporaneamente?

Non intendo discussioni; come Go gestisce internamente le goroutine non è rilevante. Intendo davvero limitare il numero di goroutine lanciate contemporaneamente: exec.Command crea un nuovo thread ogni volta che viene chiamato, quindi dovrei controllare il numero di volte in cui viene chiamato.

  • Ti sembra il codice adatto?
  • Sai come evitare l’uso di un canale fittizio in quel caso?

Non riesco a convincermi che questi canali fittizi sono la strada da percorrere.

Creo 4 goroutini di lavoratori che leggono i compiti da un canale comune. Le goroutine che sono più veloci di altre (perché sono programmate in modo diverso o capita di ottenere compiti semplici) riceveranno più compiti da questo canale rispetto ad altri. In aggiunta a ciò, vorrei usare sync.WaitGroup per aspettare che tutti i lavoratori finissero . La parte rimanente è solo la creazione dei compiti. Qui puoi vedere un’implementazione di esempio di questo approccio:

 package main import ( "os/exec" "strconv" "sync" ) func main() { tasks := make(chan *exec.Cmd, 64) // spawn four worker goroutines var wg sync.WaitGroup for i := 0; i < 4; i++ { wg.Add(1) go func() { for cmd := range tasks { cmd.Run() } wg.Done() }() } // generate some tasks for i := 0; i < 10; i++ { tasks <- exec.Command("zenity", "--info", "--text='Hello from iteration n."+strconv.Itoa(i)+"'") } close(tasks) // wait for the workers to finish wg.Wait() } 

Ci sono probabilmente altri possibili approcci, ma penso che questa sia una soluzione molto pulita e facile da capire.

Un semplice approccio alla limitazione (esegui f() N volte ma maxConcurrency massimo contemporaneamente), solo uno schema:

 package main import ( "sync" ) const maxConcurrency = 4 // for example var throttle = make(chan int, maxConcurrency) func main() { const N = 100 // for example var wg sync.WaitGroup for i := 0; i < N; i++ { throttle <- 1 // whatever number wg.Add(1) go f(i, &wg, throttle) } wg.Wait() } func f(i int, wg *sync.WaitGroup, throttle chan int) { defer wg.Done() // whatever processing println(i) <-throttle } 

Terreno di gioco

Probabilmente non chiamerei il canale del throttle "dummy". IMHO è un modo elegante (ovviamente non è la mia invenzione), come limitare la concorrenza.

BTW: Si noti che si sta ignorando l'errore restituito da cmd.Run() .

prova questo: https://github.com/korovkin/limiter

  limiter := NewConcurrencyLimiter(10) limiter.Execute(func() { zenity(...) }) limiter.Wait()