C'è un modo elegante per mettere in pausa e riprendere qualsiasi altra goroutine in golang?

Question

Nel mio caso, ho migliaia di goroutine che funzionano contemporaneamente come work(). Ho anche avuto una goroutine sync(). Quando inizia sync, ho bisogno di un'altra goroutine per mettere in pausa per un po 'dopo che il processo di sincronizzazione è terminato. Ecco il mio codice:

var channels []chan int 
var channels_mutex sync.Mutex 

func work() { 
    channel := make(chan int, 1) 
    channels_mutex.Lock() 
    channels = append(channels, channel) 
    channels_mutex.Unlock() 
    for { 
    for { 
     sync_stat := <- channel // blocked here 
     if sync_stat == 0 { // if sync complete 
     break 
     } 
    } 
    // Do some jobs 
    if (some condition) { 
     return 
    } 
    } 
} 

func sync() { 
    channels_mutex.Lock() 
    // do some sync 

    for int i := 0; i != len(channels); i++ { 
    channels[i] <- 0 
    } 
    channels_mutex.Unlock() 
} 

Ora il problema è, dal momento che è sempre <- bloccando sul letto, ogni volta che va a sync_stat := <- channel sta bloccando. So che se il canale è stato chiuso non verrà bloccato, ma poiché devo utilizzare questo canale fino alle uscite work() e non ho trovato alcun modo per riaprire un canale chiuso.

Mi sospetto su un modo sbagliato, quindi qualsiasi aiuto è apprezzato. Esiste un modo "elegante" di mettere in pausa lo & riprendere qualsiasi altra goroutine in golang?

Answer 1

Se ho capito bene, vuoi N numero di lavoratori e un controller, che può mettere in pausa, riprendere e fermare i lavoratori a volontà. Il seguente codice farà proprio questo.

package main 

import (
    "fmt" 
    "runtime" 
    "sync" 
) 

// Possible worker states. 
const (
    Stopped = 0 
    Paused = 1 
    Running = 2 
) 

// Maximum number of workers. 
const WorkerCount = 1000 

func main() { 
    // Launch workers. 
    var wg sync.WaitGroup 
    wg.Add(WorkerCount + 1) 

    workers := make([]chan int, WorkerCount) 
    for i := range workers { 
     workers[i] = make(chan int, 1) 

     go func(i int) { 
      worker(i, workers[i]) 
      wg.Done() 
     }(i) 
    } 

    // Launch controller routine. 
    go func() { 
     controller(workers) 
     wg.Done() 
    }() 

    // Wait for all goroutines to finish. 
    wg.Wait() 
} 

func worker(id int, ws <-chan int) { 
    state := Paused // Begin in the paused state. 

    for { 
     select { 
     case state = <-ws: 
      switch state { 
      case Stopped: 
       fmt.Printf("Worker %d: Stopped\n", id) 
       return 
      case Running: 
       fmt.Printf("Worker %d: Running\n", id) 
      case Paused: 
       fmt.Printf("Worker %d: Paused\n", id) 
      } 

     default: 
      // We use runtime.Gosched() to prevent a deadlock in this case. 
      // It will not be needed of work is performed here which yields 
      // to the scheduler. 
      runtime.Gosched() 

      if state == Paused { 
       break 
      } 

      // Do actual work here. 
     } 
    } 
} 

// controller handles the current state of all workers. They can be 
// instructed to be either running, paused or stopped entirely. 
func controller(workers []chan int) { 
    // Start workers 
    setState(workers, Running) 

    // Pause workers. 
    setState(workers, Paused) 

    // Unpause workers. 
    setState(workers, Running) 

    // Shutdown workers. 
    setState(workers, Stopped) 
} 

// setState changes the state of all given workers. 
func setState(workers []chan int, state int) { 
    for _, w := range workers { 
     w <- state 
    } 
}