Godeps: bump peterbourgon/diskv dependency

This will allow us to take advantage of several new features like
`Import` (eliminating a copy from cas) and a "no cache" mode (to prevent
OOMs), as well as various bug fixes.

Godeps/Godeps.json

@@ -37,7 +37,7 @@
 		},
 		{
 			"ImportPath": "github.com/peterbourgon/diskv",
-			"Rev": "fbec614921b11c91804b7ea07ad4e53647ad0d94"
+			"Rev": "508f5671a72eeaef05cf8c24abe7fbc1c07faf69"
 		}
 	]
 }

Godeps/_workspace/src/github.com/peterbourgon/diskv/basic_test.go

@@ -95,7 +95,7 @@ func TestStrings(t *testing.T) {
 		}
 	}
 
-	for k := range d.Keys() {
+	for k := range d.Keys(nil) {
 		if _, present := keys[k]; present {
 			t.Logf("got: %s", k)
 			keys[k] = true

Godeps/_workspace/src/github.com/peterbourgon/diskv/compression.go

@@ -7,8 +7,8 @@ import (
 	"io"
 )
 
-// Compression is an interface that Diskv uses to implement compression of data.
-// Writer takes a destination io.Writer and returns a WriteCloser that
+// Compression is an interface that Diskv uses to implement compression of
+// data. Writer takes a destination io.Writer and returns a WriteCloser that
 // compresses all data written through it. Reader takes a source io.Reader and
 // returns a ReadCloser that decompresses all data read through it. You may
 // define these methods on your own type, or use one of the NewCompression
@@ -18,23 +18,6 @@ type Compression interface {
 	Reader(src io.Reader) (io.ReadCloser, error)
 }
 
-type genericCompression struct {
-	wf func(w io.Writer) (io.WriteCloser, error)
-	rf func(r io.Reader) (io.ReadCloser, error)
-}
-
-func (g *genericCompression) Writer(dst io.Writer) (io.WriteCloser, error) {
-	return g.wf(dst)
-}
-
-func (g *genericCompression) Reader(src io.Reader) (io.ReadCloser, error) {
-	return g.rf(src)
-}
-
-//
-//
-//
-
 // NewGzipCompression returns a Gzip-based Compression.
 func NewGzipCompression() Compression {
 	return NewGzipCompressionLevel(flate.DefaultCompression)
@@ -66,3 +49,16 @@ func NewZlibCompressionLevelDict(level int, dict []byte) Compression {
 		func(r io.Reader) (io.ReadCloser, error) { return zlib.NewReaderDict(r, dict) },
 	}
 }
+
+type genericCompression struct {
+	wf func(w io.Writer) (io.WriteCloser, error)
+	rf func(r io.Reader) (io.ReadCloser, error)
+}
+
+func (g *genericCompression) Writer(dst io.Writer) (io.WriteCloser, error) {
+	return g.wf(dst)
+}
+
+func (g *genericCompression) Reader(src io.Reader) (io.ReadCloser, error) {
+	return g.rf(src)
+}

Godeps/_workspace/src/github.com/peterbourgon/diskv/compression_test.go

@@ -16,7 +16,6 @@ func init() {
 func testCompressionWith(t *testing.T, c Compression, name string) {
 	d := New(Options{
 		BasePath:     "compression-test",
-		Transform:    func(string) []string { return []string{""} },
 		CacheSizeMax: 0,
 		Compression:  c,
 	})

Godeps/_workspace/src/github.com/peterbourgon/diskv/diskv.go

@@ -5,13 +5,15 @@ package diskv
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"os"
-	"path"
 	"path/filepath"
+	"strings"
 	"sync"
+	"syscall"
 )
 
 const (
@@ -21,7 +23,11 @@ const (
 )
 
 var (
-	defaultTransform = func(s string) []string { return []string{} }
+	defaultTransform   = func(s string) []string { return []string{} }
+	errCanceled        = errors.New("canceled")
+	errEmptyKey        = errors.New("empty key")
+	errBadKey          = errors.New("bad key")
+	errImportDirectory = errors.New("can't import a directory")
 )
 
 // TransformFunction transforms a key into a slice of strings, with each
@@ -59,28 +65,28 @@ type Diskv struct {
 // New returns an initialized Diskv structure, ready to use.
 // If the path identified by baseDir already contains data,
 // it will be accessible, but not yet cached.
-func New(options Options) *Diskv {
-	if options.BasePath == "" {
-		options.BasePath = defaultBasePath
+func New(o Options) *Diskv {
+	if o.BasePath == "" {
+		o.BasePath = defaultBasePath
 	}
-	if options.Transform == nil {
-		options.Transform = defaultTransform
+	if o.Transform == nil {
+		o.Transform = defaultTransform
 	}
-	if options.PathPerm == 0 {
-		options.PathPerm = defaultPathPerm
+	if o.PathPerm == 0 {
+		o.PathPerm = defaultPathPerm
 	}
-	if options.FilePerm == 0 {
-		options.FilePerm = defaultFilePerm
+	if o.FilePerm == 0 {
+		o.FilePerm = defaultFilePerm
 	}
 
 	d := &Diskv{
-		Options:   options,
+		Options:   o,
 		cache:     map[string][]byte{},
 		cacheSize: 0,
 	}
 
 	if d.Index != nil && d.IndexLess != nil {
-		d.Index.Initialize(d.IndexLess, d.Keys())
+		d.Index.Initialize(d.IndexLess, d.Keys(nil))
 	}
 
 	return d
@@ -90,7 +96,7 @@ func New(options Options) *Diskv {
 // available for reads. Write relies on the filesystem to perform an eventual
 // sync to physical media. If you need stronger guarantees, see WriteStream.
 func (d *Diskv) Write(key string, val []byte) error {
-	return d.write(key, bytes.NewBuffer(val), false)
+	return d.WriteStream(key, bytes.NewBuffer(val), false)
 }
 
 // WriteStream writes the data represented by the io.Reader to the disk, under
@@ -99,22 +105,20 @@ func (d *Diskv) Write(key string, val []byte) error {
 //
 // bytes.Buffer provides io.Reader semantics for basic data types.
 func (d *Diskv) WriteStream(key string, r io.Reader, sync bool) error {
-	return d.write(key, r, sync)
-}
-
-// write synchronously writes the key-value pair to disk,
-// making it immediately available for reads. write optionally
-// performs a Sync on the relevant file descriptor.
-func (d *Diskv) write(key string, r io.Reader, sync bool) error {
 	if len(key) <= 0 {
-		return fmt.Errorf("empty key")
+		return errEmptyKey
 	}
 
-	// TODO use atomic FS ops in write()
-
 	d.Lock()
 	defer d.Unlock()
-	if err := d.ensurePath(key); err != nil {
+
+	return d.writeStreamWithLock(key, r, sync)
+}
+
+// writeStream does no input validation checking.
+// TODO: use atomic FS ops.
+func (d *Diskv) writeStreamWithLock(key string, r io.Reader, sync bool) error {
+	if err := d.ensurePathWithLock(key); err != nil {
 		return fmt.Errorf("ensure path: %s", err)
 	}
 
@@ -124,7 +128,7 @@ func (d *Diskv) write(key string, r io.Reader, sync bool) error {
 		return fmt.Errorf("open file: %s", err)
 	}
 
-	var wc = io.WriteCloser(&nopWriteCloser{f})
+	wc := io.WriteCloser(&nopWriteCloser{f})
 	if d.Compression != nil {
 		wc, err = d.Compression.Writer(f)
 		if err != nil {
@@ -157,10 +161,54 @@ func (d *Diskv) write(key string, r io.Reader, sync bool) error {
 		d.Index.Insert(key)
 	}
 
-	delete(d.cache, key) // cache only on read
+	d.bustCacheWithLock(key) // cache only on read
+
 	return nil
 }
 
+// Import imports the source file into diskv under the destination key. If the
+// destination key already exists, it's overwritten. If move is true, the
+// source file is removed after a successful import.
+func (d *Diskv) Import(srcFilename, dstKey string, move bool) (err error) {
+	if dstKey == "" {
+		return errEmptyKey
+	}
+
+	if fi, err := os.Stat(srcFilename); err != nil {
+		return err
+	} else if fi.IsDir() {
+		return errImportDirectory
+	}
+
+	d.Lock()
+	defer d.Unlock()
+
+	if err := d.ensurePathWithLock(dstKey); err != nil {
+		return fmt.Errorf("ensure path: %s", err)
+	}
+
+	if move {
+		if err := syscall.Rename(srcFilename, d.completeFilename(dstKey)); err == nil {
+			d.bustCacheWithLock(dstKey)
+			return nil
+		} else if err != syscall.EXDEV {
+			// If it failed due to being on a different device, fall back to copying
+			return err
+		}
+	}
+
+	f, err := os.Open(srcFilename)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	err = d.writeStreamWithLock(dstKey, f, false)
+	if err == nil && move {
+		err = os.Remove(srcFilename)
+	}
+	return err
+}
+
 // Read reads the key and returns the value.
 // If the key is available in the cache, Read won't touch the disk.
 // If the key is not in the cache, Read will have the side-effect of
@@ -179,36 +227,39 @@ func (d *Diskv) Read(key string) ([]byte, error) {
 // ReadStream will use the cached value. Otherwise, it will return a handle to
 // the file on disk, and cache the data on read.
 //
-// If direct is true, ReadStream will always delete any cached value for the
+// If direct is true, ReadStream will lazily delete any cached value for the
 // key, and return a direct handle to the file on disk.
 //
-// ReadStream taps into the io.Reader stream prior to decompression, and
-// caches the compressed data.
+// If compression is enabled, ReadStream taps into the io.Reader stream prior
+// to decompression, and caches the compressed data.
 func (d *Diskv) ReadStream(key string, direct bool) (io.ReadCloser, error) {
 	d.RLock()
 	defer d.RUnlock()
 
 	if val, ok := d.cache[key]; ok {
-		if direct {
-			d.cacheSize -= uint64(len(val))
-			delete(d.cache, key)
-		} else {
+		if !direct {
 			buf := bytes.NewBuffer(val)
 			if d.Compression != nil {
 				return d.Compression.Reader(buf)
 			}
 			return ioutil.NopCloser(buf), nil
 		}
+
+		go func() {
+			d.Lock()
+			defer d.Unlock()
+			d.uncacheWithLock(key, uint64(len(val)))
+		}()
 	}
 
-	return d.read(key)
+	return d.readWithRLock(key)
 }
 
 // read ignores the cache, and returns an io.ReadCloser representing the
 // decompressed data for the given key, streamed from the disk. Clients should
 // acquire a read lock on the Diskv and check the cache themselves before
 // calling read.
-func (d *Diskv) read(key string) (io.ReadCloser, error) {
+func (d *Diskv) readWithRLock(key string) (io.ReadCloser, error) {
 	filename := d.completeFilename(key)
 
 	fi, err := os.Stat(filename)
@@ -224,7 +275,12 @@ func (d *Diskv) read(key string) (io.ReadCloser, error) {
 		return nil, err
 	}
 
-	r := newSiphon(f, d, key)
+	var r io.Reader
+	if d.CacheSizeMax > 0 {
+		r = newSiphon(f, d, key)
+	} else {
+		r = &closingReader{f}
+	}
 
 	var rc = io.ReadCloser(ioutil.NopCloser(r))
 	if d.Compression != nil {
@@ -237,6 +293,22 @@ func (d *Diskv) read(key string) (io.ReadCloser, error) {
 	return rc, nil
 }
 
+// closingReader provides a Reader that automatically closes the
+// embedded ReadCloser when it reaches EOF
+type closingReader struct {
+	rc io.ReadCloser
+}
+
+func (cr closingReader) Read(p []byte) (int, error) {
+	n, err := cr.rc.Read(p)
+	if err == io.EOF {
+		if closeErr := cr.rc.Close(); closeErr != nil {
+			return n, closeErr // close must succeed for Read to succeed
+		}
+	}
+	return n, err
+}
+
 // siphon is like a TeeReader: it copies all data read through it to an
 // internal buffer, and moves that buffer to the cache at EOF.
 type siphon struct {
@@ -282,11 +354,7 @@ func (d *Diskv) Erase(key string) error {
 	d.Lock()
 	defer d.Unlock()
 
-	// erase from cache
-	if val, ok := d.cache[key]; ok {
-		d.cacheSize -= uint64(len(val))
-		delete(d.cache, key)
-	}
+	d.bustCacheWithLock(key)
 
 	// erase from index
 	if d.Index != nil {
@@ -297,17 +365,17 @@ func (d *Diskv) Erase(key string) error {
 	filename := d.completeFilename(key)
 	if s, err := os.Stat(filename); err == nil {
 		if !!s.IsDir() {
-			return fmt.Errorf("bad key")
+			return errBadKey
 		}
 		if err = os.Remove(filename); err != nil {
-			return err
+			return fmt.Errorf("remove: %s", err)
 		}
 	} else {
-		return err
+		return fmt.Errorf("stat: %s", err)
 	}
 
 	// clean up and return
-	d.pruneDirs(key)
+	d.pruneDirsWithLock(key)
 	return nil
 }
 
@@ -344,12 +412,27 @@ func (d *Diskv) Has(key string) bool {
 	return true
 }
 
-// Keys returns a channel that will yield every key accessible by the store in
-// undefined order.
-func (d *Diskv) Keys() <-chan string {
+// Keys returns a channel that will yield every key accessible by the store,
+// in undefined order. If a cancel channel is provided, closing it will
+// terminate and close the keys channel.
+func (d *Diskv) Keys(cancel <-chan struct{}) <-chan string {
+	return d.KeysPrefix("", cancel)
+}
+
+// KeysPrefix returns a channel that will yield every key accessible by the
+// store with the given prefix, in undefined order. If a cancel channel is
+// provided, closing it will terminate and close the keys channel. If the
+// provided prefix is the empty string, all keys will be yielded.
+func (d *Diskv) KeysPrefix(prefix string, cancel <-chan struct{}) <-chan string {
+	var prepath string
+	if prefix == "" {
+		prepath = d.BasePath
+	} else {
+		prepath = d.pathFor(prefix)
+	}
 	c := make(chan string)
 	go func() {
-		filepath.Walk(d.BasePath, walker(c))
+		filepath.Walk(prepath, walker(c, prefix, cancel))
 		close(c)
 	}()
 	return c
@@ -357,49 +440,54 @@ func (d *Diskv) Keys() <-chan string {
 
 // walker returns a function which satisfies the filepath.WalkFunc interface.
 // It sends every non-directory file entry down the channel c.
-func walker(c chan string) func(path string, info os.FileInfo, err error) error {
+func walker(c chan<- string, prefix string, cancel <-chan struct{}) filepath.WalkFunc {
 	return func(path string, info os.FileInfo, err error) error {
-		if err == nil && !info.IsDir() {
-			c <- info.Name()
+		if err != nil {
+			return err
 		}
-		return nil // "pass"
+
+		if info.IsDir() || !strings.HasPrefix(info.Name(), prefix) {
+			return nil // "pass"
+		}
+
+		select {
+		case c <- info.Name():
+		case <-cancel:
+			return errCanceled
+		}
+
+		return nil
 	}
 }
 
 // pathFor returns the absolute path for location on the filesystem where the
 // data for the given key will be stored.
 func (d *Diskv) pathFor(key string) string {
-	return path.Join(d.BasePath, path.Join(d.Transform(key)...))
+	return filepath.Join(d.BasePath, filepath.Join(d.Transform(key)...))
 }
 
-// ensureDir is a helper function that generates all necessary directories on
-// the filesystem for the given key.
-func (d *Diskv) ensurePath(key string) error {
+// ensurePathWithLock is a helper function that generates all necessary
+// directories on the filesystem for the given key.
+func (d *Diskv) ensurePathWithLock(key string) error {
 	return os.MkdirAll(d.pathFor(key), d.PathPerm)
 }
 
 // completeFilename returns the absolute path to the file for the given key.
 func (d *Diskv) completeFilename(key string) string {
-	return fmt.Sprintf("%s%c%s", d.pathFor(key), os.PathSeparator, key)
+	return filepath.Join(d.pathFor(key), key)
 }
 
 // cacheWithLock attempts to cache the given key-value pair in the store's
 // cache. It can fail if the value is larger than the cache's maximum size.
 func (d *Diskv) cacheWithLock(key string, val []byte) error {
 	valueSize := uint64(len(val))
-	if err := d.ensureCacheSpaceFor(valueSize); err != nil {
+	if err := d.ensureCacheSpaceWithLock(valueSize); err != nil {
 		return fmt.Errorf("%s; not caching", err)
 	}
 
 	// be very strict about memory guarantees
 	if (d.cacheSize + valueSize) > d.CacheSizeMax {
-		panic(
-			fmt.Sprintf(
-				"failed to make room for value (%d/%d)",
-				valueSize,
-				d.CacheSizeMax,
-			),
-		)
+		panic(fmt.Sprintf("failed to make room for value (%d/%d)", valueSize, d.CacheSizeMax))
 	}
 
 	d.cache[key] = val
@@ -414,13 +502,23 @@ func (d *Diskv) cacheWithoutLock(key string, val []byte) error {
 	return d.cacheWithLock(key, val)
 }
 
-// pruneDirs deletes empty directories in the path walk leading to the key k.
-// Typically this function is called after an Erase is made.
-func (d *Diskv) pruneDirs(key string) error {
+func (d *Diskv) bustCacheWithLock(key string) {
+	if val, ok := d.cache[key]; ok {
+		d.uncacheWithLock(key, uint64(len(val)))
+	}
+}
+
+func (d *Diskv) uncacheWithLock(key string, sz uint64) {
+	d.cacheSize -= sz
+	delete(d.cache, key)
+}
+
+// pruneDirsWithLock deletes empty directories in the path walk leading to the
+// key k. Typically this function is called after an Erase is made.
+func (d *Diskv) pruneDirsWithLock(key string) error {
 	pathlist := d.Transform(key)
 	for i := range pathlist {
-		pslice := pathlist[:len(pathlist)-i]
-		dir := path.Join(d.BasePath, path.Join(pslice...))
+		dir := filepath.Join(d.BasePath, filepath.Join(pathlist[:len(pathlist)-i]...))
 
 		// thanks to Steven Blenkinsop for this snippet
 		switch fi, err := os.Stat(dir); true {
@@ -430,7 +528,7 @@ func (d *Diskv) pruneDirs(key string) error {
 			panic(fmt.Sprintf("corrupt dirstate at %s", dir))
 		}
 
-		nlinks, err := filepath.Glob(fmt.Sprintf("%s%c*", dir, os.PathSeparator))
+		nlinks, err := filepath.Glob(filepath.Join(dir, "*"))
 		if err != nil {
 			return err
 		} else if len(nlinks) > 0 {
@@ -444,31 +542,25 @@ func (d *Diskv) pruneDirs(key string) error {
 	return nil
 }
 
-// ensureCacheSpaceFor deletes entries from the cache in arbitrary order until
-// the cache has at least valueSize bytes available.
-func (d *Diskv) ensureCacheSpaceFor(valueSize uint64) error {
+// ensureCacheSpaceWithLock deletes entries from the cache in arbitrary order
+// until the cache has at least valueSize bytes available.
+func (d *Diskv) ensureCacheSpaceWithLock(valueSize uint64) error {
 	if valueSize > d.CacheSizeMax {
-		return fmt.Errorf(
-			"value size (%d bytes) too large for cache (%d bytes)",
-			valueSize,
-			d.CacheSizeMax,
-		)
+		return fmt.Errorf("value size (%d bytes) too large for cache (%d bytes)", valueSize, d.CacheSizeMax)
 	}
 
 	safe := func() bool { return (d.cacheSize + valueSize) <= d.CacheSizeMax }
+
 	for key, val := range d.cache {
 		if safe() {
 			break
 		}
-		delete(d.cache, key)            // delete is safe, per spec
-		d.cacheSize -= uint64(len(val)) // len should return uint :|
+
+		d.uncacheWithLock(key, uint64(len(val)))
 	}
+
 	if !safe() {
-		panic(fmt.Sprintf(
-			"%d bytes still won't fit in the cache! (max %d bytes)",
-			valueSize,
-			d.CacheSizeMax,
-		))
+		panic(fmt.Sprintf("%d bytes still won't fit in the cache! (max %d bytes)", valueSize, d.CacheSizeMax))
 	}
 
 	return nil
@@ -477,8 +569,8 @@ func (d *Diskv) ensureCacheSpaceFor(valueSize uint64) error {
 // nopWriteCloser wraps an io.Writer and provides a no-op Close method to
 // satisfy the io.WriteCloser interface.
 type nopWriteCloser struct {
-	w io.Writer
+	io.Writer
 }
 
-func (wc *nopWriteCloser) Write(p []byte) (int, error) { return wc.w.Write(p) }
+func (wc *nopWriteCloser) Write(p []byte) (int, error) { return wc.Writer.Write(p) }
 func (wc *nopWriteCloser) Close() error                { return nil }

Godeps/_workspace/src/github.com/peterbourgon/diskv/examples/content-addressable-store/cas.go

@@ -43,7 +43,7 @@ func main() {
 	}
 
 	var keyCount int
-	for key := range d.Keys() {
+	for key := range d.Keys(nil) {
 		val, err := d.Read(key)
 		if err != nil {
 			panic(fmt.Sprintf("key %s had no value", key))

Godeps/_workspace/src/github.com/peterbourgon/diskv/import_test.go

@@ -0,0 +1,76 @@
+package diskv_test
+
+import (
+	"bytes"
+	"io/ioutil"
+	"os"
+
+	"github.com/peterbourgon/diskv"
+
+	"testing"
+)
+
+func TestImportMove(t *testing.T) {
+	b := []byte(`0123456789`)
+	f, err := ioutil.TempFile("", "temp-test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := f.Write(b); err != nil {
+		t.Fatal(err)
+	}
+	f.Close()
+
+	d := diskv.New(diskv.Options{
+		BasePath: "test-import-move",
+	})
+	defer d.EraseAll()
+
+	key := "key"
+
+	if err := d.Write(key, []byte(`TBD`)); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := d.Import(f.Name(), key, true); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := os.Stat(f.Name()); err == nil || !os.IsNotExist(err) {
+		t.Errorf("expected temp file to be gone, but err = %v", err)
+	}
+
+	if !d.Has(key) {
+		t.Errorf("%q not present", key)
+	}
+
+	if buf, err := d.Read(key); err != nil || bytes.Compare(b, buf) != 0 {
+		t.Errorf("want %q, have %q (err = %v)", string(b), string(buf), err)
+	}
+}
+
+func TestImportCopy(t *testing.T) {
+	b := []byte(`¡åéîòü!`)
+
+	f, err := ioutil.TempFile("", "temp-test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := f.Write(b); err != nil {
+		t.Fatal(err)
+	}
+	f.Close()
+
+	d := diskv.New(diskv.Options{
+		BasePath: "test-import-copy",
+	})
+	defer d.EraseAll()
+
+	if err := d.Import(f.Name(), "key", false); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := os.Stat(f.Name()); err != nil {
+		t.Errorf("expected temp file to remain, but got err = %v", err)
+	}
+}

Godeps/_workspace/src/github.com/peterbourgon/diskv/index.go

@@ -33,8 +33,8 @@ func (s llrbString) Less(i llrb.Item) bool {
 // using Petar Maymounkov's LLRB tree.
 type LLRBIndex struct {
 	sync.RWMutex
-	less LessFunction
-	tree *llrb.LLRB
+	LessFunction
+	*llrb.LLRB
 }
 
 // Initialize populates the LLRB tree with data from the keys channel,
@@ -42,28 +42,28 @@ type LLRBIndex struct {
 func (i *LLRBIndex) Initialize(less LessFunction, keys <-chan string) {
 	i.Lock()
 	defer i.Unlock()
-	i.less = less
-	i.tree = rebuild(less, keys)
+	i.LessFunction = less
+	i.LLRB = rebuild(less, keys)
 }
 
 // Insert inserts the given key (only) into the LLRB tree.
 func (i *LLRBIndex) Insert(key string) {
 	i.Lock()
 	defer i.Unlock()
-	if i.tree == nil || i.less == nil {
+	if i.LLRB == nil || i.LessFunction == nil {
 		panic("uninitialized index")
 	}
-	i.tree.ReplaceOrInsert(llrbString{s: key, l: i.less})
+	i.LLRB.ReplaceOrInsert(llrbString{s: key, l: i.LessFunction})
 }
 
 // Delete removes the given key (only) from the LLRB tree.
 func (i *LLRBIndex) Delete(key string) {
 	i.Lock()
 	defer i.Unlock()
-	if i.tree == nil || i.less == nil {
+	if i.LLRB == nil || i.LessFunction == nil {
 		panic("uninitialized index")
 	}
-	i.tree.Delete(llrbString{s: key, l: i.less})
+	i.LLRB.Delete(llrbString{s: key, l: i.LessFunction})
 }
 
 // Keys yields a maximum of n keys in order. If the passed 'from' key is empty,
@@ -74,19 +74,19 @@ func (i *LLRBIndex) Keys(from string, n int) []string {
 	i.RLock()
 	defer i.RUnlock()
 
-	if i.tree == nil || i.less == nil {
+	if i.LLRB == nil || i.LessFunction == nil {
 		panic("uninitialized index")
 	}
 
-	if i.tree.Len() <= 0 {
+	if i.LLRB.Len() <= 0 {
 		return []string{}
 	}
 
-	llrbFrom := llrbString{s: from, l: i.less}
+	llrbFrom := llrbString{s: from, l: i.LessFunction}
 	skipFirst := true
-	if len(from) <= 0 || !i.tree.Has(llrbFrom) {
+	if len(from) <= 0 || !i.LLRB.Has(llrbFrom) {
 		// no such key, so start at the top
-		llrbFrom = i.tree.Min().(llrbString)
+		llrbFrom = i.LLRB.Min().(llrbString)
 		skipFirst = false
 	}
 
@@ -95,7 +95,7 @@ func (i *LLRBIndex) Keys(from string, n int) []string {
 		keys = append(keys, i.(llrbString).s)
 		return len(keys) < n
 	}
-	i.tree.AscendGreaterOrEqual(llrbFrom, iterator)
+	i.LLRB.AscendGreaterOrEqual(llrbFrom, iterator)
 
 	if skipFirst && len(keys) > 0 {
 		keys = keys[1:]

Godeps/_workspace/src/github.com/peterbourgon/diskv/issues_test.go

@@ -3,6 +3,7 @@ package diskv
 import (
 	"bytes"
 	"io/ioutil"
+	"sync"
 	"testing"
 	"time"
 )
@@ -72,3 +73,49 @@ func TestIssue2B(t *testing.T) {
 	}
 	t.Logf("ReadStream('abc') returned error: %v", err)
 }
+
+// Ensure ReadStream with direct=true isn't racy.
+func TestIssue17(t *testing.T) {
+	var (
+		basePath = "test-data"
+	)
+
+	dWrite := New(Options{
+		BasePath:     basePath,
+		CacheSizeMax: 0,
+	})
+	defer dWrite.EraseAll()
+
+	dRead := New(Options{
+		BasePath:     basePath,
+		CacheSizeMax: 50,
+	})
+
+	cases := map[string]string{
+		"a": `1234567890`,
+		"b": `2345678901`,
+		"c": `3456789012`,
+		"d": `4567890123`,
+		"e": `5678901234`,
+	}
+
+	for k, v := range cases {
+		if err := dWrite.Write(k, []byte(v)); err != nil {
+			t.Fatalf("during write: %s", err)
+		}
+		dRead.Read(k) // ensure it's added to cache
+	}
+
+	var wg sync.WaitGroup
+	start := make(chan struct{})
+	for k, v := range cases {
+		wg.Add(1)
+		go func(k, v string) {
+			<-start
+			dRead.ReadStream(k, true)
+			wg.Done()
+		}(k, v)
+	}
+	close(start)
+	wg.Wait()
+}

Godeps/_workspace/src/github.com/peterbourgon/diskv/keys_test.go

@@ -0,0 +1,231 @@
+package diskv_test
+
+import (
+	"reflect"
+	"runtime"
+	"strings"
+	"testing"
+
+	"github.com/coreos/rocket/Godeps/_workspace/src/github.com/peterbourgon/diskv"
+)
+
+var (
+	keysTestData = map[string]string{
+		"ab01cd01": "When we started building CoreOS",
+		"ab01cd02": "we looked at all the various components available to us",
+		"ab01cd03": "re-using the best tools",
+		"ef01gh04": "and building the ones that did not exist",
+		"ef02gh05": "We believe strongly in the Unix philosophy",
+		"xxxxxxxx": "tools should be independently useful",
+	}
+
+	prefixes = []string{
+		"", // all
+		"a",
+		"ab",
+		"ab0",
+		"ab01",
+		"ab01cd0",
+		"ab01cd01",
+		"ab01cd01x", // none
+		"b",         // none
+		"b0",        // none
+		"0",         // none
+		"01",        // none
+		"e",
+		"ef",
+		"efx", // none
+		"ef01gh0",
+		"ef01gh04",
+		"ef01gh05",
+		"ef01gh06", // none
+	}
+)
+
+func TestKeysFlat(t *testing.T) {
+	transform := func(s string) []string {
+		if s == "" {
+			t.Fatalf(`transform should not be called with ""`)
+		}
+		return []string{}
+	}
+	d := diskv.New(diskv.Options{
+		BasePath:  "test-data",
+		Transform: transform,
+	})
+	defer d.EraseAll()
+
+	for k, v := range keysTestData {
+		d.Write(k, []byte(v))
+	}
+
+	checkKeys(t, d.Keys(nil), keysTestData)
+}
+
+func TestKeysNested(t *testing.T) {
+	d := diskv.New(diskv.Options{
+		BasePath:  "test-data",
+		Transform: blockTransform(2),
+	})
+	defer d.EraseAll()
+
+	for k, v := range keysTestData {
+		d.Write(k, []byte(v))
+	}
+
+	checkKeys(t, d.Keys(nil), keysTestData)
+}
+
+func TestKeysPrefixFlat(t *testing.T) {
+	d := diskv.New(diskv.Options{
+		BasePath: "test-data",
+	})
+	defer d.EraseAll()
+
+	for k, v := range keysTestData {
+		d.Write(k, []byte(v))
+	}
+
+	for _, prefix := range prefixes {
+		checkKeys(t, d.KeysPrefix(prefix, nil), filterPrefix(keysTestData, prefix))
+	}
+}
+
+func TestKeysPrefixNested(t *testing.T) {
+	d := diskv.New(diskv.Options{
+		BasePath:  "test-data",
+		Transform: blockTransform(2),
+	})
+	defer d.EraseAll()
+
+	for k, v := range keysTestData {
+		d.Write(k, []byte(v))
+	}
+
+	for _, prefix := range prefixes {
+		checkKeys(t, d.KeysPrefix(prefix, nil), filterPrefix(keysTestData, prefix))
+	}
+}
+
+func TestKeysCancel(t *testing.T) {
+	d := diskv.New(diskv.Options{
+		BasePath: "test-data",
+	})
+	defer d.EraseAll()
+
+	for k, v := range keysTestData {
+		d.Write(k, []byte(v))
+	}
+
+	var (
+		cancel      = make(chan struct{})
+		received    = 0
+		cancelAfter = len(keysTestData) / 2
+	)
+
+	for key := range d.Keys(cancel) {
+		received++
+
+		if received >= cancelAfter {
+			close(cancel)
+			runtime.Gosched() // allow walker to detect cancel
+		}
+
+		t.Logf("received %d: %q", received, key)
+	}
+
+	if want, have := cancelAfter, received; want != have {
+		t.Errorf("want %d, have %d")
+	}
+}
+
+func checkKeys(t *testing.T, c <-chan string, want map[string]string) {
+	for k := range c {
+		if _, ok := want[k]; !ok {
+			t.Errorf("%q yielded but not expected", k)
+			continue
+		}
+
+		delete(want, k)
+		t.Logf("%q yielded OK", k)
+	}
+
+	if len(want) != 0 {
+		t.Errorf("%d expected key(s) not yielded: %s", len(want), strings.Join(flattenKeys(want), ", "))
+	}
+}
+
+func blockTransform(blockSize int) func(string) []string {
+	return func(s string) []string {
+		var (
+			sliceSize = len(s) / blockSize
+			pathSlice = make([]string, sliceSize)
+		)
+		for i := 0; i < sliceSize; i++ {
+			from, to := i*blockSize, (i*blockSize)+blockSize
+			pathSlice[i] = s[from:to]
+		}
+		return pathSlice
+	}
+}
+
+func filterPrefix(in map[string]string, prefix string) map[string]string {
+	out := map[string]string{}
+	for k, v := range in {
+		if strings.HasPrefix(k, prefix) {
+			out[k] = v
+		}
+	}
+	return out
+}
+
+func TestFilterPrefix(t *testing.T) {
+	input := map[string]string{
+		"all":        "",
+		"and":        "",
+		"at":         "",
+		"available":  "",
+		"best":       "",
+		"building":   "",
+		"components": "",
+		"coreos":     "",
+		"did":        "",
+		"exist":      "",
+		"looked":     "",
+		"not":        "",
+		"ones":       "",
+		"re-using":   "",
+		"started":    "",
+		"that":       "",
+		"the":        "",
+		"to":         "",
+		"tools":      "",
+		"us":         "",
+		"various":    "",
+		"we":         "",
+		"when":       "",
+	}
+
+	for prefix, want := range map[string]map[string]string{
+		"a":    map[string]string{"all": "", "and": "", "at": "", "available": ""},
+		"al":   map[string]string{"all": ""},
+		"all":  map[string]string{"all": ""},
+		"alll": map[string]string{},
+		"c":    map[string]string{"components": "", "coreos": ""},
+		"co":   map[string]string{"components": "", "coreos": ""},
+		"com":  map[string]string{"components": ""},
+	} {
+		have := filterPrefix(input, prefix)
+		if !reflect.DeepEqual(want, have) {
+			t.Errorf("%q: want %v, have %v", prefix, flattenKeys(want), flattenKeys(have))
+		}
+	}
+}
+
+func flattenKeys(m map[string]string) []string {
+	a := make([]string, 0, len(m))
+	for k := range m {
+		a = append(a, k)
+	}
+	return a
+}

Godeps/_workspace/src/github.com/peterbourgon/diskv/speed_test.go

@@ -88,15 +88,15 @@ func benchWrite(b *testing.B, size int, withIndex bool) {
 	b.StopTimer()
 }
 
-func BenchmarkWrite_32B_NoIndex(b *testing.B) {
+func BenchmarkWrite__32B_NoIndex(b *testing.B) {
 	benchWrite(b, 32, false)
 }
 
-func BenchmarkWrite_1KB_NoIndex(b *testing.B) {
+func BenchmarkWrite__1KB_NoIndex(b *testing.B) {
 	benchWrite(b, 1024, false)
 }
 
-func BenchmarkWrite_4KB_NoIndex(b *testing.B) {
+func BenchmarkWrite__4KB_NoIndex(b *testing.B) {
 	benchWrite(b, 4096, false)
 }
 
@@ -104,15 +104,15 @@ func BenchmarkWrite_10KB_NoIndex(b *testing.B) {
 	benchWrite(b, 10240, false)
 }
 
-func BenchmarkWrite_32B_WithIndex(b *testing.B) {
+func BenchmarkWrite__32B_WithIndex(b *testing.B) {
 	benchWrite(b, 32, true)
 }
 
-func BenchmarkWrite_1KB_WithIndex(b *testing.B) {
+func BenchmarkWrite__1KB_WithIndex(b *testing.B) {
 	benchWrite(b, 1024, true)
 }
 
-func BenchmarkWrite_4KB_WithIndex(b *testing.B) {
+func BenchmarkWrite__4KB_WithIndex(b *testing.B) {
 	benchWrite(b, 4096, true)
 }
 
@@ -120,15 +120,15 @@ func BenchmarkWrite_10KB_WithIndex(b *testing.B) {
 	benchWrite(b, 10240, true)
 }
 
-func BenchmarkRead_32B_NoCache(b *testing.B) {
+func BenchmarkRead__32B_NoCache(b *testing.B) {
 	benchRead(b, 32, 0)
 }
 
-func BenchmarkRead_1KB_NoCache(b *testing.B) {
+func BenchmarkRead__1KB_NoCache(b *testing.B) {
 	benchRead(b, 1024, 0)
 }
 
-func BenchmarkRead_4KB_NoCache(b *testing.B) {
+func BenchmarkRead__4KB_NoCache(b *testing.B) {
 	benchRead(b, 4096, 0)
 }
 
@@ -136,15 +136,15 @@ func BenchmarkRead_10KB_NoCache(b *testing.B) {
 	benchRead(b, 10240, 0)
 }
 
-func BenchmarkRead_32B_WithCache(b *testing.B) {
+func BenchmarkRead__32B_WithCache(b *testing.B) {
 	benchRead(b, 32, keyCount*32*2)
 }
 
-func BenchmarkRead_1KB_WithCache(b *testing.B) {
+func BenchmarkRead__1KB_WithCache(b *testing.B) {
 	benchRead(b, 1024, keyCount*1024*2)
 }
 
-func BenchmarkRead_4KB_WithCache(b *testing.B) {
+func BenchmarkRead__4KB_WithCache(b *testing.B) {
 	benchRead(b, 4096, keyCount*4096*2)
 }
 

Godeps/_workspace/src/github.com/peterbourgon/diskv/stream_test.go

@@ -9,7 +9,6 @@ import (
 func TestBasicStreamCaching(t *testing.T) {
 	d := New(Options{
 		BasePath:     "test-data",
-		Transform:    func(string) []string { return []string{} },
 		CacheSizeMax: 1024,
 	})
 	defer d.EraseAll()
@@ -45,18 +44,15 @@ func TestBasicStreamCaching(t *testing.T) {
 
 func TestReadStreamDirect(t *testing.T) {
 	var (
-		basePath  = "test-data"
-		transform = func(string) []string { return []string{} }
+		basePath = "test-data"
 	)
 	dWrite := New(Options{
 		BasePath:     basePath,
-		Transform:    transform,
 		CacheSizeMax: 0,
 	})
 	defer dWrite.EraseAll()
 	dRead := New(Options{
 		BasePath:     basePath,
-		Transform:    transform,
 		CacheSizeMax: 1024,
 	})
 

cas/cas.go

@@ -122,7 +122,7 @@ func (ds Store) ReadIndex(i Index) error {
 func (ds Store) Dump(hex bool) {
 	for _, s := range ds.stores {
 		var keyCount int
-		for key := range s.Keys() {
+		for key := range s.Keys(nil) {
 			val, err := s.Read(key)
 			if err != nil {
 				panic(fmt.Sprintf("key %s had no value", key))