Panic on closed channel

pkg/walker/walker.go

@@ -2,10 +2,11 @@ package walker
 
 import (
 	"fmt"
-	"github.com/tg44/heptapod/pkg/utils"
 	"os"
 	"path/filepath"
 	"time"
+
+	"github.com/tg44/heptapod/pkg/utils"
 )
 
 // first ret is the excluding paths we found
@@ -27,60 +28,56 @@ type WalkJob struct {
 
 func Run(jobs []WalkJob, par int, bufferSize int, verbose int) []string {
 	defer utils.TimeTrack(time.Now(), "walker run", verbose)
-	if len(jobs) == 0 {
+	initialJobCount := len(jobs)
+	if initialJobCount == 0 {
 		return []string{}
 	}
-	spawn := make(chan WalkJob, bufferSize)
-	start := make(chan bool, bufferSize)
-	end := make(chan []string)
 
-	for _, j := range jobs {
-		spawn <- j
-	}
+	results := make(chan []string, bufferSize)
+	jobQueue := make(chan WalkJob, bufferSize)
+	spawn := make(chan WalkJob, bufferSize)
+	runningJobCount := 0
 
-	for w := 1; w <= par; w++ {
-		go worker(w, spawn, start, end, verbose)
+	for i, job := range jobs {
+		if (i + 1) <= par {
+			runningJobCount += 1
+			spawn <- job
+		} else {
+			jobQueue <- job
+		}
 	}
 
-	res := []string{}
-	//wait loop
-	<-start
-	for e := range end {
-		more := false
+	globalResult := []string{}
+	maxId := 0
+	for runningJobCount > 0 {
 		select {
-		case <-start:
-			more = true
-		default:
-			more = false
-		}
-		if !more {
-			close(start)
-			close(end)
-			close(spawn)
+		case singleResult := <-results:
+			globalResult = append(globalResult, singleResult...)
+			runningJobCount -= 1
+			if len(jobQueue) > 0 {
+				// check then act is safe since only this one thread received from jobQueue
+				runningJobCount += 1
+				spawn <- <-jobQueue
+			}
+		case j := <-spawn:
+			go walk(maxId, j.Rootpath, j.Walkers, j.AlreadyFiltered, results, jobQueue, verbose)
+			maxId += 1
 		}
-		res = append(res, e...)
-	}
-	return res
-}
-
-func worker(id int, spawn chan WalkJob, start chan bool, end chan []string, verbose int) {
-	for j := range spawn {
-		start <- true
-		walk(id, j.Rootpath, j.Walkers, j.AlreadyFiltered, spawn, start, end, verbose)
 	}
+	return globalResult
 }
 
-func walk(runnerId int, rootpath string, walkers []Walker, alreadyFiltered []string, spawn chan WalkJob, start chan bool, end chan []string, verbose int) {
+func walk(runnerId int, rootpath string, walkers []Walker, alreadyFiltered []string, results chan []string, jobQueue chan WalkJob, verbose int) {
 	defer utils.TimeTrack(time.Now(), fmt.Sprintf("(runner-%d) walk on %s", runnerId, rootpath), verbose)
 	hasNext := true
 	var res *utils.List = nil
 	path, err := utils.FixupPathsToHandleHome(rootpath)
 	if err != nil {
-		end <- []string{}
+		results <- []string{}
 		return
 	}
 	if len(walkers) == 0 {
-		end <- []string{}
+		results <- []string{}
 		return
 	}
 	l := &utils.List{path, nil}
@@ -149,7 +146,7 @@ func walk(runnerId int, rootpath string, walkers []Walker, alreadyFiltered []str
 								next = next.AddAsHead(f)
 							} else if len(keeps) > 0 {
 								//if some ignore happened we ignore the path, and spawn a new job with the nonignorant walkers
-								spawn <- WalkJob{f, keeps, alreadyFiltered}
+								jobQueue <- WalkJob{f, keeps, alreadyFiltered}
 							} else {
 								//if keeps is empty we let the next be nil
 							}
@@ -165,5 +162,5 @@ func walk(runnerId int, rootpath string, walkers []Walker, alreadyFiltered []str
 			l = next
 		}
 	}
-	end <- res.ToArray()
+	results <- res.ToArray()
 }