Concurrent Producer and Consumer in Go

Question

I want to create a producer/consumer with manager program in Go. For example: I have a 5 producers, 5 consumers and manager. Producers have their own local arrays, they iterate over them and send the elements to the manager. Consumers have their own local arrays with info that elements consume; they send them to the manager too. The Manager has it own array, where it stores what and how many elements there are (for example - if the producer sends 1 1 2 3 1 2 0 elements, the manager array looks like 1 3 2 1 (one 0, three 1, two 2 and one 3 ) and it handles producers' and consumers' requests - placing an element into the array (produce) or removing it (consume).

Is it possible to make a program like this in Go? I already did this in JAVA + CSP with channels to send info and guards in the manager to determine which procedure should be done first when producer and consumer try to process the same element (for example, a producer wants to add 1 to the manager array and at the same time a consumer wants to consume 1).

Any examples or advice are welcome, because I don't find any info about what I want to do. If needed I can give my JAVA+CSP code.

UPDATE. How about synchronization (not to take from empty array)? For example - if consumer wants to consume element from manager array that does not exist yet (for example consumer wants to consume '3', but manager don't have any of thems) but producer has this element and it will be produced after few iterations - how can I make consumers to check manager array again and again until producers work is finished? Should I need to create structs (or classes) for consumers elements and mark that they are used or not, or Go has specific methods to do this?

Answer 1

Here's a full example, including channel cleanup. After all the consumers and producers are finished, the Manager prints out the result (for which I've used a map rather than a slice, since I think it makes the code a tad easier).

package main

import "fmt"

// Consume processes the numbers in ns, sending them on ch after they're
// processed. When the routine is finished, it signals that on done.
func Consume(done chan bool, ch chan int, ns []int) {
    for i := range ns {
        ch <- i
    }
    done <- true
}

// Produce "creates" the numbers in ns, sending them on ch after they're
// produced. When the routine is finished, it signals that on done.
func Produce(done chan bool, ch chan int, ns []int) {
    for i := range ns {
        ch <- i
    }
    done <- true
}

// Manage creates consumers and producers for the given int slices.
// It returns once all consumers and producers are finished.
func Manage(cons, pros [][]int) {
    cch := make(chan int)
    pch := make(chan int)
    dch := make(chan bool)
    n := len(cons) + len(pros)
    data := make(map[int]int)
    for _, c := range cons {
        go Consume(dch, cch, c)
    }
    for _, p := range pros {
        go Produce(dch, pch, p)
    }
    for n > 0 {
        select {
        case c := <-cch:
            data[c] -= 1
        case c := <-pch:
            data[c] += 1
        case <-dch:
            n -= 1
        }
    }
    close(cch)
    close(pch)
    close(dch)
    fmt.Println(data)
}
func main() {
    cons := [][]int{{1, 3, 5}, {0, 1, 5}}
    pros := [][]int{{0, 1, 1}, {3, 5, 5, 7}}
    Manage(cons, pros)
}

Answer 2

I did a very similar example to what you're trying to do, check this gist github gist

The way I implemented that is using a single channel for my process consumer and another for my 2x processes that produce items, the for block controls the push from producers to the consumer and whenever the producers arent pushing anything the loop will default. The objects moved through the channels are slices, processes would produce and consume the headlines in each slice sent through the channel. I believe you can adjust this code to fit your example.