Get responses from multiple go routines into an array

Question

I need to fetch responses from multiple go routines and put them into an array. I know that channels could be used for this, however I am not sure how I can make sure that all go routines have finished processing the results. Thus I am using a waitgroup.

Code

func main() {
  log.Info("Collecting ints")
  var results []int32
  for _, broker := range e.BrokersByBrokerID {
      wg.Add(1)
      go getInt32(&wg)
  }
  wg.Wait()
  log.info("Collected")
}

func getInt32(wg *sync.WaitGroup) (int32, error) {
  defer wg.Done()

  // Just to show that this method may just return an error and no int32
  err := broker.Open(config)
  if err != nil && err != sarama.ErrAlreadyConnected {
    return 0, fmt.Errorf("Cannot connect to broker '%v': %s", broker.ID(), err)
  }
  defer broker.Close()

  return 1003, nil
}

My question

How can I put all the response int32 (which may return an error) into my int32 array, making sure that all go routines have finished their processing work and returned either the error or the int?

Answer 1

If you don't process the return values of the function launched as a goroutine, they are discarded. See What happens to return value from goroutine.

You may use a slice to collect the results, where each goroutine could receive the index to put the results to, or alternatively the address of the element. See Can I concurrently write different slice elements. Note that if you use this, the slice must be pre-allocated and only the element belonging to the goroutine may be written, you can't "touch" other elements and you can't append to the slice.

Or you may use a channel, on which the goroutines send values that include the index or ID of the item they processed, so the collecting goroutine can identify or order them. See How to collect values from N goroutines executed in a specific order?

If processing should stop on the first error encountered, see Close multiple goroutine if an error occurs in one in go

Here's an example how it could look like when using a channel. Note that no waitgroup is needed here, because we know that we expect as many values on the channel as many goroutines we launch.

type result struct {
    task int32
    data int32
    err  error
}

func main() {
    tasks := []int32{1, 2, 3, 4}

    ch := make(chan result)

    for _, task := range tasks {
        go calcTask(task, ch)
    }

    // Collect results:
    results := make([]result, len(tasks))

    for i := range results {
        results[i] = <-ch
    }

    fmt.Printf("Results: %+v\n", results)
}

func calcTask(task int32, ch chan<- result) {
    if task > 2 {
        // Simulate failure
        ch <- result{task: task, err: fmt.Errorf("task %v failed", task)}
        return
    }

    // Simulate success
    ch <- result{task: task, data: task * 2, err: nil}
}

Output (try ot on the Go Playground):

Results: [{task:4 data:0 err:0x40e130} {task:1 data:2 err:<nil>} {task:2 data:4 err:<nil>} {task:3 data:0 err:0x40e138}]

Answer 2

I also believe you have to use channel, it must be something like this:

package main

import (
    "fmt"
    "log"
    "sync"
)

var (
    BrokersByBrokerID = []int32{1, 2, 3}
)

type result struct {
    data string
    err string // you must use error type here
}

func main()  {
    var wg sync.WaitGroup
    var results []result
    ch := make(chan result)

    for _, broker := range BrokersByBrokerID {
        wg.Add(1)
        go getInt32(ch, &wg, broker)
    }

    go func() {
        for v := range ch {
            results = append(results, v)
        }
    }()

    wg.Wait()
    close(ch)

    log.Printf("collected %v", results)
}

func getInt32(ch chan result, wg *sync.WaitGroup, broker int32) {
    defer wg.Done()

    if broker == 1 {
        ch <- result{err: fmt.Sprintf("error: gor broker 1")}
        return
    }

    ch <- result{data: fmt.Sprintf("broker %d - ok", broker)}
}

Result will look like this:

2019/02/05 15:26:28 collected [{broker 3 - ok } {broker 2 - ok } { error: gor broker 1}]

Answer 3

package main

import (
    "fmt"
    "log"
    "sync"
)

var (
    BrokersByBrokerID = []int{1, 2, 3, 4}
)

type result struct {
    data string
    err  string // you must use error type here
}

func main() {
    var wg sync.WaitGroup
    var results []int
    ch := make(chan int)
    done := make(chan bool)
    for _, broker := range BrokersByBrokerID {
        wg.Add(1)

        go func(i int) {
            defer wg.Done()
            ch <- i
            if i == 4 {
                done <- true
            }

        }(broker)
    }
L:
    for {
        select {
        case v := <-ch:

            results = append(results, v)
            if len(results) == 4 {
                //<-done
                close(ch)
                break L
                
            }

        case _ = <-done:
            break
        }
    }

    fmt.Println("STOPPED")
    //<-done
    wg.Wait()

    log.Printf("collected %v", results)

}

Answer 4

Thank cn007b and Edenshaw. My answer is based on their answers. As Edenshaw commented, need another sync.Waitgroup for goroutine which getting results from channel, or you may get an incomplete array.

package main

import (
    "fmt"
    "sync"

    "encoding/json"
)

type Resp struct {
    id int
}

func main() {
    var wg sync.WaitGroup
    chanRes := make(chan interface{}, 3)
    for i := 0; i < 3; i++ {
        wg.Add(1)
        resp := &Resp{}
        go func(i int, resp *Resp) {
            defer wg.Done()
            resp.id = i
            chanRes <- resp
        }(i, resp)
    }

    res := make([]interface{}, 0)

    var wg2 sync.WaitGroup
    wg2.Add(1)
    go func() {
        defer wg2.Done()
        for v := range chanRes {
            res = append(res, v.(*Resp).id)
        }
    }()

    wg.Wait()
    close(chanRes)
    wg2.Wait()

    resStr, _ := json.Marshal(res)
    fmt.Println(string(resStr))
}

Answer 5

package main

import (
    "fmt"
    "log"
    "sync"
    "time"
)

var (
    BrokersByBrokerID = []int{1, 2, 3, 4}
)

type result struct {
    data string
    err  string // you must use error type here
}

func main() {    
    var wg sync.WaitGroup.   
    var results []int  
    ch := make(chan int)  
    done := make(chan bool) 
    for _, broker := range BrokersByBrokerID {                      
       wg.Add(1)

        go func(i int) {
            defer wg.Done()
            ch <- i
            if i == 4 {
                done <- true
            } 

        }(broker)
    }
    
    for v := range ch {

        results = append(results, v)
        if len(results) == 4 {
            close(ch)
        }

    }

    fmt.Println("STOPPED")
    <-done
    wg.Wait()
    
    log.Printf("collected %v", results)


}


</pre>