Link to go playground https://go.dev/play/p/ctQDpDW6pui This code has been based on suggestions and conversations in this thread here
Architecture:
- A read method creates a channel shared with producer. Read method adds to the channel and producer reads from it.
- A producer and consumer share a channel. Producer adds data, consumer consumes.
- The producer and consumer use error channels to communicate errors to main method.
- A hard requirement is if any worker - the producer or consumer - encounters an error then all workers should halt. This caused me to use contexts to cancel the producer and consumers.
- The producers and consumers communicate error via error channel.
- A go-routine called handleAllErrors consumes the errorChannel and uses context's cancel to call quits and shut down the remaining producers and consumers.
Issues
- I have tried hard to overcome challenges of deadlock and to the best of my knowledge all deadlock issues have been resolved however the code appears a some improvement.
- I am especially concerned of the race conditions of draining queues when the consumers are interrupted.
- Eg: if producers are still computing, while all the consumers get interrupted the code could blocks/deadlocks.
- Eg: if all producers are interrupted while reader is still feeding it, it may blocks.
Please review the code, help figure any deadlocks I missed and suggest a cleaner approach to some deadlocks I tried to work-around.
package operation
import (
"context"
"fmt"
"sync"
)
func mapperreducer_so() {
a1 := []int{1, 2, 3, 4, 5}
a2 := []int{5, 4, 3, 1, 1}
a3 := []int{6, 7, 8, 9}
a4 := []int{1, 2, 3, 4, 5}
a5 := []int{5, 4, 3, 1, 1}
a6 := []int{6, 7, 18, 9}
arrayOfArray := [][]int{a1, a2, a3, a4, a5, a6}
ctx, cancel := context.WithCancel(context.Background())
ch1 := read(ctx, arrayOfArray)
messageCh := make(chan int)
errCh := make(chan error)
producerWg := &sync.WaitGroup{}
for i := 0; i < 3; i++ {
producerWg.Add(1)
producer(ctx, producerWg, ch1, messageCh, errCh)
}
consumerWg := &sync.WaitGroup{}
for i := 0; i < 3; i++ {
consumerWg.Add(1)
consumer(ctx, consumerWg, messageCh, errCh)
}
firstError := handleAllErrors(ctx, cancel, errCh)
producerWg.Wait()
close(messageCh)
consumerWg.Wait()
close(errCh)
fmt.Println(<-firstError)
}
func read(ctx context.Context, arrayOfArray [][]int) <-chan []int {
ch := make(chan []int)
go func() {
defer close(ch)
for i := 0; i < len(arrayOfArray); i++ {
select {
case <-ctx.Done():
return
case ch <- arrayOfArray[i]:
}
}
}()
return ch
}
func producer(ctx context.Context, wg *sync.WaitGroup, in <-chan []int, messageCh chan<- int, errCh chan<- error) {
go func() {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case arr, ok := <-in:
if !ok {
return
}
for i := 0; i < len(arr); i++ {
// simulating an error.
//if arr[i] == 10 {
// errCh <- fmt.Errorf("producer interrupted")
//}
select {
case <-ctx.Done():
return
case messageCh <- 2 * arr[i]:
}
}
}
}
}()
}
func consumer(ctx context.Context, wg *sync.WaitGroup, messageCh <-chan int, errCh chan<- error) {
go func() {
wg.Done()
for {
select {
case <-ctx.Done():
return
case n, ok := <-messageCh:
if !ok {
return
}
fmt.Println("consumed: ", n)
// simulating erros
//if n == 10 {
// errCh <- fmt.Errorf("output error during write")
//}
}
}
}()
}
func handleAllErrors(ctx context.Context, cancel context.CancelFunc, errCh chan error) <-chan error {
firstErrCh := make(chan error, 1)
isFirstError := true
go func() {
defer close(firstErrCh)
for err := range errCh {
select {
case <-ctx.Done():
default:
cancel()
}
if isFirstError {
firstErrCh <- err
isFirstError = !isFirstError
}
}
}()
return firstErrCh
}
