Is this a good way of managing parallel go routines when I care about ordering of results?

Question

I have a process with a number of stages that need to be completed in sequence. Each stage is largely parallelisable, involving looping over a large data structure and processing each item independently, and collating the results into the data structure used by the next stage.

My approach is like the following:

var wg sync.WaitGroup
var control_routine sync.WaitGroup
stage_complete = make(chan bool)
stage_one_channel := make(chan map[typeFoo][]typeBar, 16)
stage_two_channel := make(chan map[typeBaz][]typeQux, 16)
stage_one_results := make(map[typeFoo][]typeBar)
stage_two_results := make(map[typeBaz][]typeQux)

// A routine to receive results as they arrive from each stage
go func() {
    for  {
        select {
        case new_stage_one_results := <- stage_one_channel:
            // merge new_stage_one_results into stage_one_results...
        case new_stage_two_results := <- stage_two_channel:
            // merge new_stage_two_results into stage_two_results...
        ...
        case stay_alive := <-stage_complete:
            control_routine.Done()
            if !stay_alive {
                return
            }
        }
    }
}()

// Begin stage One
control_routine.Add(1)
for i, some_data := range my_data {
    wg.Add(1)
    go func(i, datum){
        defer wg.Done()
        some_stage_one_results = make(map[typeFoo][]typeBar)
        // do some heavy lifting with some_data which populates some_stage_one_results...
        stage_one_channel <- some_stage_one_results
    }()
}
wg.Wait()
stage_complete <- true
control_routine.Wait()
// End stage one

// More stages much like the first...

// Finally send the control routine a terminal signal and wait for it to finish
stage_complete <- false
control_routine.Wait()

This basically seems to work, but I need to be sure of the following assumption (which some behaviour is causing me to doubt).

Can I be sure that all go routines in stage N will completely finish and the data sent through stage_N_channel will all be processed in the control routine before the stage_complete channel is read from?

What I'm seeing is that sometimes, it looks like one or two results from a stage go missing, as if the control routine were moving on before processing them. Adding a one second sleep to the main routine after wg.Wait(), seems to help but I'm not sure is this is right.

Is there anything else unusual or potentially dangerous about this pattern? Is there another pattern I should consider using?

---

I finally settled on a solution where each stage resembles the following.

for i, some_data := range my_data {
  wg.Add(1)
  go func(i, some_data){
    // stage logic
    ...
    stage_one_channel <- some_stage_one_results
  }(i, some_data)
}

go func() {
    wg.Wait()
    close(stage_one_channel)
}()
for new_stage_one_data := <- stage_one_channel {
  // integrate received data
  ...
  wg.Done()
}

Answer 1

The answer to "Can I be sure...?" is no, there is nothing in your code that ensures that. That is why results sometimes go missing.

Missing results are a sign of a data race. You can use the Go race detector to find and fix these problems. Data races are categorically dangerous. They can corrupt memory.

You might be able to find and fix the problem by adding more synchronization, but your code is too complex as it is. Since you asked, a pattern you should probably be using is a particular producer-consumer pattern. It has two parts, 1) a producer goroutine that sends data to a channel and closes the channel after sending the last value, and 2) a consumer goroutine with a for loop that ranges over the channel. With this pattern you should be able to dispense with the select statement, the stage complete channel, and the control routine wait group.

Your code that begins for i, some_data := range my_data is what I'm calling a producer. It produces data on the channel stage_one_channel. Following wg.Wait(), you should replace the line stage_complete <- true with close(stage_one_channel).

Replace your for loop containing the select statement with a separate for loop for each stage, something like

for new_stage_one_results := range stage_one_channel {
    ...
}

When this for loop completes, you are sure that all data has been received from stage_one_channel and processed by the loop body. This is the synchronization your code is missing. Your current mechanism only waits for data to be sent, not received and processed.

When the last loop completes, your program can exit. You could retain some of your mechanism for waiting for this, but simpler would be swap what you currently have in your main goroutine for the "consumer" code that you currently have in that first function literal. That is, run your producers in one or more separate goroutines so you can have your consumers--the last one especially--in main. When the last consumer has processed the last data, you know your program is done, with nothing else to wait for.

Finally, let me reinforce, if go run -race says you have a problem, you have a problem.

Answer 2

I don't know anything about Go. I read your post out of curiosity and I think I can give you some pointers nonetheless.

1. It seems you removed too many things to create your simplified post. Your code does not make sense as it is. For example, wg.Add(1) is never called.

2. Variable names are a big part of software development. You should probably spend some time to find better names for wg and control_routine.

3. It's not clear from your post, but it seems that wg and control_routine are re-used in further steps. It's much cleaner to define new variables for different steps instead of reusing old variables.

4. It does not make sense to me that you have one go routine that processes all incoming data from all stages. I would only define a go method that accepts stage two items after all of stage one is done, etc. I would actually not just write the different steps one after the other, but define some kind of higher level construct to handle that stepping logic. I don't know Go, so I'm not sure if OO or something else is appropriate for that. You should at least make sure that each step if completely separated from the others, not reusing any older variables.

5. You explicitly kill the main processing go routine with the stay_alive business. I am wondering if this is standard go practice. It probably does take a tiny bit of resources if you don't kill that go routine when you are done. However, I am not sure it is worth doing since it does add quite a bit of extra complexity. Maybe someone who knows go could comment on that.

I can't tell you if you have some concurrency bug at the moment. It seems you forgot some bits when you created your post and your code is too complex for your task.