diff --git a/Godeps/Godeps.json b/Godeps/Godeps.json index e93732f..083a3e8 100644 --- a/Godeps/Godeps.json +++ b/Godeps/Godeps.json @@ -37,7 +37,7 @@ }, { "ImportPath": "github.com/peterbourgon/diskv", - "Rev": "fbec614921b11c91804b7ea07ad4e53647ad0d94" + "Rev": "508f5671a72eeaef05cf8c24abe7fbc1c07faf69" } ] } diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/basic_test.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/basic_test.go index 46e6bc0..ef0d32f 100644 --- a/Godeps/_workspace/src/github.com/peterbourgon/diskv/basic_test.go +++ b/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 diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/compression.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/compression.go index 8b27bbd..5192b02 100644 --- a/Godeps/_workspace/src/github.com/peterbourgon/diskv/compression.go +++ b/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) +} diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/compression_test.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/compression_test.go index be9cbba..2d61420 100644 --- a/Godeps/_workspace/src/github.com/peterbourgon/diskv/compression_test.go +++ b/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, }) diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/diskv.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/diskv.go index 705da56..512b535 100644 --- a/Godeps/_workspace/src/github.com/peterbourgon/diskv/diskv.go +++ b/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 } diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/examples/content-addressable-store/cas.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/examples/content-addressable-store/cas.go index 27cd573..b985697 100644 --- a/Godeps/_workspace/src/github.com/peterbourgon/diskv/examples/content-addressable-store/cas.go +++ b/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)) diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/import_test.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/import_test.go new file mode 100644 index 0000000..a08ac7c --- /dev/null +++ b/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) + } +} diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/index.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/index.go index 9530431..fd42b40 100644 --- a/Godeps/_workspace/src/github.com/peterbourgon/diskv/index.go +++ b/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:] diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/issues_test.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/issues_test.go index a1ba9e6..0b0b109 100644 --- a/Godeps/_workspace/src/github.com/peterbourgon/diskv/issues_test.go +++ b/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() +} diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/keys_test.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/keys_test.go new file mode 100644 index 0000000..7675eaf --- /dev/null +++ b/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 +} diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/speed_test.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/speed_test.go index 6cc731e..43837c0 100644 --- a/Godeps/_workspace/src/github.com/peterbourgon/diskv/speed_test.go +++ b/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) } diff --git a/Godeps/_workspace/src/github.com/peterbourgon/diskv/stream_test.go b/Godeps/_workspace/src/github.com/peterbourgon/diskv/stream_test.go index b5fb878..7991dbf 100644 --- a/Godeps/_workspace/src/github.com/peterbourgon/diskv/stream_test.go +++ b/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, }) diff --git a/cas/cas.go b/cas/cas.go index 5be7b1d..a95dd5c 100644 --- a/cas/cas.go +++ b/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))