8.8 Example: Concurrent Directory Traversal
In this section, we’ll build a program that reports the disk usage of
one or more directories specified on the command line, like the Unix
du command.
Most of its work is done by the walkDir function below, which
enumerates the entries of the directory dir using the
dirents helper function.
// walkDir recursively walks the file tree rooted at dir
// and sends the size of each found file on fileSizes.
func walkDir(dir string, fileSizes chan<- int64) {
for _, entry := range dirents(dir) {
if entry.IsDir() {
subdir := filepath.Join(dir, entry.Name())
walkDir(subdir, fileSizes)
} else {
fileSizes <- entry.Size()
}
}
}
// dirents returns the entries of directory dir.
func dirents(dir string) []os.FileInfo {
entries, err := ioutil.ReadDir(dir)
if err != nil {
fmt.Fprintf(os.Stderr, "du1: %v
", err)
return nil
}
return entries
}
The ioutil.ReadDir function
returns a slice of os.FileInfo—the same information that a call to
os.Stat returns for a single file.
For each subdirectory, walkDir recursively calls itself, and
for each file, walkDir sends a message on the fileSizes
channel.
The message is the size of the file in bytes.
The main function, shown below, uses two goroutines.
The background goroutine calls walkDir for each directory
specified on the command line and finally closes the fileSizes
channel.
The main goroutine computes the sum of the file sizes it receives from
the channel and finally prints the total.
// The du1 command computes the disk usage of the files in a directory.
package main
import (
"flag"
"fmt"
"io/ioutil"
"os"
"path/filepath"
)
func main() {
// Determine the initial directories.
flag.Parse()
roots := flag.Args()
if len(roots) == 0 {
roots = []string{"."}
}
// Traverse the file tree.
fileSizes := make(chan int64)
go func() {
for _, root := range roots {
walkDir(root, fileSizes)
}
close(fileSizes)
}()
// Print the results.
var nfiles, nbytes int64
for size := range fileSizes {
nfiles++
nbytes += size
}
printDiskUsage(nfiles, nbytes)
}
func printDiskUsage(nfiles, nbytes int64) {
fmt.Printf("%d files %.1f GB
", nfiles, float64(nbytes)/1e9)
}
This program pauses for a long while before printing its result:
$ go build gopl.io/ch8/du1 $ ./du1 $HOME /usr /bin /etc 213201 files 62.7 GB
The program would be nicer if it kept us informed of its progress.
However, simply moving the printDiskUsage call into the loop
would cause it to print thousands of lines of output.
The variant of du below prints the totals periodically, but
only if the -v flag is specified since not all users will want to
see progress messages.
The background goroutine that loops over roots remains unchanged.
The main goroutine now uses a ticker to generate events every
500ms, and a select statement to wait for either a file
size message, in which case it updates the totals, or a tick
event, in which case it prints the current totals.
If the -v flag is not specified, the tick channel
remains nil, and its case in the select is
effectively disabled.
var verbose = flag.Bool("v", false, "show verbose progress messages")
func main() {
// ...start background goroutine...
// Print the results periodically.
var tick <-chan time.Time
if *verbose {
tick = time.Tick(500 * time.Millisecond)
}
var nfiles, nbytes int64
loop:
for {
select {
case size, ok := <-fileSizes:
if !ok {
break loop // fileSizes was closed
}
nfiles++
nbytes += size
case <-tick:
printDiskUsage(nfiles, nbytes)
}
}
printDiskUsage(nfiles, nbytes) // final totals
}
Since the program no longer uses a range loop, the
first select case must explicitly test whether
the fileSizes channel has been closed, using the two-result
form of receive operation.
If the channel has been closed, the program breaks out of the loop.
The labeled break statement breaks out of both
the select and the for loop;
an unlabeled break
would break out of only the select, causing the loop to begin
the next iteration.
The program now gives us a leisurely stream of updates:
$ go build gopl.io/ch8/du2 $ ./du2 -v $HOME /usr /bin /etc 28608 files 8.3 GB 54147 files 10.3 GB 93591 files 15.1 GB 127169 files 52.9 GB 175931 files 62.2 GB 213201 files 62.7 GB
However, it still takes too long to finish.
There’s no reason why all the calls to walkDir can’t be done
concurrently, thereby exploiting parallelism in the disk system.
The third version of du, below, creates a new goroutine for
each call to walkDir.
It uses a sync.WaitGroup (§8.5)
to count the number of calls to walkDir that are still active, and
a closer goroutine
to close the fileSizes channel when the
counter drops to zero.
func main() {
// ...determine roots...
// Traverse each root of the file tree in parallel.
fileSizes := make(chan int64)
var n sync.WaitGroup
for _, root := range roots {
n.Add(1)
go walkDir(root, &n, fileSizes)
}
go func() {
n.Wait()
close(fileSizes)
}()
// ...select loop...
}
func walkDir(dir string, n *sync.WaitGroup, fileSizes chan<- int64) {
defer n.Done()
for _, entry := range dirents(dir) {
if entry.IsDir() {
n.Add(1)
subdir := filepath.Join(dir, entry.Name())
go walkDir(subdir, n, fileSizes)
} else {
fileSizes <- entry.Size()
}
}
}
Since this program creates many thousands of goroutines at its peak,
we have to change dirents to use a counting semaphore to prevent it
from opening too many files at once, just as we did for the web
crawler in Section 8.6:
// sema is a counting semaphore for limiting concurrency in dirents.
var sema = make(chan struct{}, 20)
// dirents returns the entries of directory dir.
func dirents(dir string) []os.FileInfo {
sema <- struct{}{} // acquire token
defer func() { <-sema }() // release token
// ...
This version runs several times faster than the previous one, though there is a lot of variability from system to system.
Exercise 8.9:
Write a version of du that computes and periodically displays
separate totals for each of the root directories.