golangの他のゴルーチンを一時停止＆再開するためのエレガントな方法はありますか？

Question

私の場合、私は同時にwork()として働いている何千ものゴルーチンを持っています。私もsync()のゴルーチンを持っていました。 syncが起動すると、同期ジョブが完了してからしばらくの間、他のゴルーチンが一時停止する必要があります。ここに私のコードは次のとおりです。

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

が今の問題は<-は常に読み取りでブロックされているので、ある、すべての時間がsync_stat := <- channelに行くブロックしています。私はチャンネルが閉鎖されているかどうかを知っていますが、それはブロックされませんが、work()出口までこのチャンネルを使用しなければならないので、閉鎖されたチャンネルを開く方法が見つかりませんでした。

私は間違った方法で自分自身を疑うので、どんな助けにも感謝します。 &ゴーランの他のゴルーチンを再開するには、いくつかの "エレガントな"方法がありますか？

Answer 1

あなたが正しく理解している場合は、N人の作業者と1人のコントローラが必要です。これにより、休暇を中断し、再開し、停止することができます。次のコードはそれだけです。

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