在多个线程中运行一个函数

Question

I have implemented a function contractGraph which calculates a minimal cut of a graph using randomized contraction. I am running it a specified number of times and calculating the minimum cut:

minCut := 0
for i := 0; i < totalCount; i++ {
    _minCut := contractGraph(graph)
    if minCut == 0 || _minCut < minCut {
        minCut = _minCut
    }
}

contractGraph does CPU intensive calculations, but the program uses only one CPU core on my machine. I want to modify it, so at any time 4 parallel executions of contractGraph happen, the results are put in channel and are read synchronously and the minimum is calculated.

I tried:

func worker(graph Graph, i int, workerChan <- chan bool, minCutChan chan <- int) {
    defer func () { <- workerChan }()
    min_cut := contractGraph(graph)
    minCutChan <- min_cut
}


func workerRunner(graph Graph, minCutChan chan int, totalCount int, workerCount int) {
    workerChan := make(chan bool, workerCount)
    for i := 0; i < totalCount; i++ {
        go worker(graph, i, workerChan, minCutChan)
    }
}

    minCutChan := make(chan int)
    go workerRunner(graph, minCutChan, totalCount, 4)

    // read the resulting min cuts
    minCut := 0
    for _minCut := range minCutChan {
        if minCut == 0 || _minCut < minCut {
            minCut = _minCut
        }
    }

But still only one core is used and I get at the end:

fatal error: all goroutines are asleep - deadlock!

Also I don't like having to channels, I think it should be possible to have only one channel with the results.

What pattern would you recommend to use?

Answer 1

You forgot to close the minCutChan so main is stuck into range and all the go routines have completed.

to not use the channel you can use sync.WaitGroup

EDIT: To handle the totalCount I would use atomic.AddInt64 see the new updated examples:

see a working mock example with these edits: http://play.golang.org/p/WyCQrWK5aa

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

type Graph struct {
}

func contractGraph(Graph) int { return 0 }

func worker(wg *sync.WaitGroup, graph Graph, i int, minCutChan chan<- int) {
    defer wg.Done()
    for {
        count := atomic.AddInt64(&totalCount, -1) 
        if count < 0 {
            break
        }
        fmt.Println("Worker Iteration", count)
        min_cut := contractGraph(graph)
        minCutChan <- min_cut
    }
}

func workerRunner(graph Graph, minCutChan chan int, workerCount int) {
    wg := new(sync.WaitGroup)
    wg.Add(workerCount)
    for i := 0; i < workerCount; i++ {
        go worker(wg, graph, i, minCutChan)
    }
    wg.Wait()
    close(minCutChan)
}

var totalCount int64

func main() {
    workerCount := 4
    graph := Graph{}
    totalCount = 100
    minCutChan := make(chan int, workerCount+1)
    go workerRunner(graph, minCutChan, workerCount)

    go func() {
    }()

    // read the resulting min cuts
    minCut := 0
    for _minCut := range minCutChan {
        if minCut == 0 || _minCut < minCut {
            minCut = _minCut
        }
    }
    fmt.Println(minCut)
}

even more in go style is to spin the workers inside an anonymous function:

http://play.golang.org/p/nT0uUutQyS

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

type Graph struct {
}

func contractGraph(Graph) int { return 0 }

var totalCount int64

func workerRunner(graph Graph, minCutChan chan int, workerCount int) {
    var wg sync.WaitGroup
    wg.Add(workerCount)
    for i := 0; i < workerCount; i++ {
        go func() {
            defer wg.Done()
            for {
                count := atomic.AddInt64(&totalCount, -1)
                if count < 0 {
                    break
                }
                fmt.Println("Worker Iteration", count)

                min_cut := contractGraph(graph)
                minCutChan <- min_cut
            }
        }()
    }
    wg.Wait()
    close(minCutChan)
}

func main() {
    workerCount := 4
    totalCount = 100
    graph := Graph{}
    minCutChan := make(chan int, workerCount+1)
    go workerRunner(graph, minCutChan, workerCount)

    // read the resulting min cuts
    minCut := 0
    for _minCut := range minCutChan {
        if minCut == 0 || _minCut < minCut {
            minCut = _minCut
        }
    }
    fmt.Println(minCut)
}