Playing around with pipelines
@@ -9,13 +9,16 @@ import (
)
type FileJob struct {
+ Filename string
Location string
Content []byte
+ Count int64
}
// A buffered channel that we can send work requests on.
var FileReadJobQueue = make(chan FileJob, runtime.NumCPU()*10)
var FileProcessJobQueue = make(chan FileJob, runtime.NumCPU()*10)
+var FileSummaryJobQueue = make(chan FileJob, runtime.NumCPU()*10)
func readFile(filepath string) []byte {
// TODO only read as deep into the file as we need
@@ -38,7 +41,7 @@ func walkDirectory(directory string) {
if f.IsDir() {
directories = append(directories, f.Name())
} else {
- FileReadJobQueue <- FileJob{Location: filepath.Join(directory, f.Name())}
+ FileReadJobQueue <- FileJob{Filename: f.Name(), Location: filepath.Join(directory, f.Name())}
}
}
@@ -49,8 +52,10 @@ func walkDirectory(directory string) {
func fileProcessorWorker() {
for {
+ // Blocks till it gets something
res := <-FileProcessJobQueue
+ // Do some pointless work
count := 0
for _, i := range res.Content {
if i == 0 {
@@ -58,15 +63,17 @@ func fileProcessorWorker() {
}
}
- fmt.Println(res.Location, len(res.Content), count)
+ fmt.Println(res.Filename, res.Location, len(res.Content), count)
+ FileSummaryJobQueue <- FileJob{Filename: res.Filename, Location: res.Location, Count: int64(count)}
}
}
func fileReaderWorker() {
for {
+ // Blocks till it gets something
res := <-FileReadJobQueue
content := readFile(res.Location)
- FileProcessJobQueue <- FileJob{Location: res.Location, Content: content}
+ FileProcessJobQueue <- FileJob{Filename: res.Filename, Location: res.Location, Content: content}
}
}
@@ -76,8 +83,8 @@ func main() {
go fileReaderWorker()
}
- fmt.Println("Main")
walkDirectory("../")
+ fmt.Println("Finished running...")
// Once done lets print it all out
output := []string{
