fix: goroutine stuck on edge case

Author: kevwan

.gitignore

@@ -8,7 +8,8 @@
 # dev files
 .idea
 
-# Test binary, built with `go test -c`
+# For test
+**/testdata
 *.test
 
 # Output of the go coverage tool, specifically when used with LiteIDE

mapreduce.go

@@ -286,33 +286,21 @@ func drain(channel <-chan interface{}) {
 
 func executeMappers(mCtx mapperContext) {
 	var wg sync.WaitGroup
-	pc := &onceChan{channel: make(chan interface{})}
 	defer func() {
-		// in case panic happens when processing last item, for loop not handling it.
-		select {
-		case r := <-pc.channel:
-			mCtx.panicChan.write(r)
-		default:
-		}
-
 		wg.Wait()
 		close(mCtx.collector)
 		drain(mCtx.source)
 	}()
 
+	var failed int32
 	pool := make(chan struct{}, mCtx.workers)
 	writer := newGuardedWriter(mCtx.ctx, mCtx.collector, mCtx.doneChan)
-	for {
+	for atomic.LoadInt32(&failed) == 0 {
 		select {
 		case <-mCtx.ctx.Done():
 			return
 		case <-mCtx.doneChan:
 			return
-		case r := <-pc.channel:
-			// make sure this method quit ASAP,
-			// without this case branch, all the items from source will be consumed.
-			mCtx.panicChan.write(r)
-			return
 		case pool <- struct{}{}:
 			item, ok := <-mCtx.source
 			if !ok {
@@ -324,7 +312,8 @@ func executeMappers(mCtx mapperContext) {
 			go func() {
 				defer func() {
 					if r := recover(); r != nil {
-						pc.write(r)
+						atomic.AddInt32(&failed, 1)
+						mCtx.panicChan.write(r)
 					}
 					wg.Done()
 					<-pool