Refactor inmemory driver for Stat and WriteStream methods
This change started out as simply updating the existing inmemory driver to implement the new Stat call. After struggling with the map based implementation, it has been refactored to be a tree-based implementation. This process has exposed a few missing error cases in the StorageDriver API that should be addressed in the coming weeks.
This commit is contained in:
Stephen J Day
parent
ab9570f872
commit
2ebc373d91
3 changed files with 453 additions and 80 deletions
|
|
@ -5,9 +5,9 @@ import (
|
|||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/docker/docker-registry/storagedriver"
|
||||
"github.com/docker/docker-registry/storagedriver/factory"
|
||||
|
|
@ -29,13 +29,18 @@ func (factory *inMemoryDriverFactory) Create(parameters map[string]string) (stor
|
|||
// Driver is a storagedriver.StorageDriver implementation backed by a local map.
|
||||
// Intended solely for example and testing purposes.
|
||||
type Driver struct {
|
||||
storage map[string][]byte
|
||||
mutex sync.RWMutex
|
||||
root *dir
|
||||
mutex sync.RWMutex
|
||||
}
|
||||
|
||||
// New constructs a new Driver.
|
||||
func New() *Driver {
|
||||
return &Driver{storage: make(map[string][]byte)}
|
||||
return &Driver{root: &dir{
|
||||
common: common{
|
||||
p: "/",
|
||||
mod: time.Now(),
|
||||
},
|
||||
}}
|
||||
}
|
||||
|
||||
// Implement the storagedriver.StorageDriver interface.
|
||||
|
|
@ -44,18 +49,31 @@ func New() *Driver {
|
|||
func (d *Driver) GetContent(path string) ([]byte, error) {
|
||||
d.mutex.RLock()
|
||||
defer d.mutex.RUnlock()
|
||||
contents, ok := d.storage[path]
|
||||
if !ok {
|
||||
return nil, storagedriver.PathNotFoundError{Path: path}
|
||||
|
||||
rc, err := d.ReadStream(path, 0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return contents, nil
|
||||
defer rc.Close()
|
||||
|
||||
return ioutil.ReadAll(rc)
|
||||
}
|
||||
|
||||
// PutContent stores the []byte content at a location designated by "path".
|
||||
func (d *Driver) PutContent(path string, contents []byte) error {
|
||||
func (d *Driver) PutContent(p string, contents []byte) error {
|
||||
d.mutex.Lock()
|
||||
defer d.mutex.Unlock()
|
||||
d.storage[path] = contents
|
||||
|
||||
f, err := d.root.mkfile(p)
|
||||
if err != nil {
|
||||
// TODO(stevvooe): Again, we need to clarify when this is not a
|
||||
// directory in StorageDriver API.
|
||||
return fmt.Errorf("not a file")
|
||||
}
|
||||
|
||||
f.truncate()
|
||||
f.WriteAt(contents, 0)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -64,86 +82,104 @@ func (d *Driver) PutContent(path string, contents []byte) error {
|
|||
func (d *Driver) ReadStream(path string, offset int64) (io.ReadCloser, error) {
|
||||
d.mutex.RLock()
|
||||
defer d.mutex.RUnlock()
|
||||
contents, err := d.GetContent(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else if len(contents) <= int(offset) {
|
||||
return nil, storagedriver.InvalidOffsetError{Path: path, Offset: offset}
|
||||
|
||||
path = d.normalize(path)
|
||||
found := d.root.find(path)
|
||||
|
||||
if found.path() != path {
|
||||
return nil, storagedriver.PathNotFoundError{Path: path}
|
||||
}
|
||||
|
||||
src := contents[offset:]
|
||||
buf := make([]byte, len(src))
|
||||
copy(buf, src)
|
||||
return ioutil.NopCloser(bytes.NewReader(buf)), nil
|
||||
if found.isdir() {
|
||||
return nil, fmt.Errorf("%q is a directory", path)
|
||||
}
|
||||
|
||||
return ioutil.NopCloser(found.(*file).sectionReader(offset)), nil
|
||||
}
|
||||
|
||||
// WriteStream stores the contents of the provided io.ReadCloser at a location
|
||||
// designated by the given path.
|
||||
func (d *Driver) WriteStream(path string, offset, size int64, reader io.ReadCloser) error {
|
||||
defer reader.Close()
|
||||
d.mutex.RLock()
|
||||
defer d.mutex.RUnlock()
|
||||
func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (nn int64, err error) {
|
||||
d.mutex.Lock()
|
||||
defer d.mutex.Unlock()
|
||||
|
||||
resumableOffset, err := d.CurrentSize(path)
|
||||
if offset < 0 {
|
||||
return 0, storagedriver.InvalidOffsetError{Path: path, Offset: offset}
|
||||
}
|
||||
|
||||
normalized := d.normalize(path)
|
||||
|
||||
f, err := d.root.mkfile(normalized)
|
||||
if err != nil {
|
||||
return err
|
||||
return 0, fmt.Errorf("not a file")
|
||||
}
|
||||
|
||||
if offset > int64(resumableOffset) {
|
||||
return storagedriver.InvalidOffsetError{Path: path, Offset: offset}
|
||||
}
|
||||
var buf bytes.Buffer
|
||||
|
||||
contents, err := ioutil.ReadAll(reader)
|
||||
nn, err = buf.ReadFrom(reader)
|
||||
if err != nil {
|
||||
return err
|
||||
// TODO(stevvooe): This condition is odd and we may need to clarify:
|
||||
// we've read nn bytes from reader but have written nothing to the
|
||||
// backend. What is the correct return value? Really, the caller needs
|
||||
// to know that the reader has been advanced and reattempting the
|
||||
// operation is incorrect.
|
||||
return nn, err
|
||||
}
|
||||
|
||||
if offset > 0 {
|
||||
contents = append(d.storage[path][0:offset], contents...)
|
||||
}
|
||||
|
||||
d.storage[path] = contents
|
||||
return nil
|
||||
f.WriteAt(buf.Bytes(), offset)
|
||||
return nn, err
|
||||
}
|
||||
|
||||
// CurrentSize retrieves the curernt size in bytes of the object at the given
|
||||
// path.
|
||||
func (d *Driver) CurrentSize(path string) (uint64, error) {
|
||||
// Stat returns info about the provided path.
|
||||
func (d *Driver) Stat(path string) (storagedriver.FileInfo, error) {
|
||||
d.mutex.RLock()
|
||||
defer d.mutex.RUnlock()
|
||||
contents, ok := d.storage[path]
|
||||
if !ok {
|
||||
return 0, nil
|
||||
|
||||
normalized := d.normalize(path)
|
||||
found := d.root.find(path)
|
||||
|
||||
if found.path() != normalized {
|
||||
return nil, storagedriver.PathNotFoundError{Path: path}
|
||||
}
|
||||
return uint64(len(contents)), nil
|
||||
|
||||
fi := storagedriver.FileInfoFields{
|
||||
Path: path,
|
||||
IsDir: found.isdir(),
|
||||
ModTime: found.modtime(),
|
||||
}
|
||||
|
||||
if !fi.IsDir {
|
||||
fi.Size = int64(len(found.(*file).data))
|
||||
}
|
||||
|
||||
return storagedriver.FileInfoInternal{FileInfoFields: fi}, nil
|
||||
}
|
||||
|
||||
// List returns a list of the objects that are direct descendants of the given
|
||||
// path.
|
||||
func (d *Driver) List(path string) ([]string, error) {
|
||||
if path[len(path)-1] != '/' {
|
||||
path += "/"
|
||||
}
|
||||
subPathMatcher, err := regexp.Compile(fmt.Sprintf("^%s[^/]+", path))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
normalized := d.normalize(path)
|
||||
|
||||
found := d.root.find(normalized)
|
||||
|
||||
if !found.isdir() {
|
||||
return nil, fmt.Errorf("not a directory") // TODO(stevvooe): Need error type for this...
|
||||
}
|
||||
|
||||
d.mutex.RLock()
|
||||
defer d.mutex.RUnlock()
|
||||
// we use map to collect unique keys
|
||||
keySet := make(map[string]struct{})
|
||||
for k := range d.storage {
|
||||
if key := subPathMatcher.FindString(k); key != "" {
|
||||
keySet[key] = struct{}{}
|
||||
entries, err := found.(*dir).list(normalized)
|
||||
|
||||
if err != nil {
|
||||
switch err {
|
||||
case errNotExists:
|
||||
return nil, storagedriver.PathNotFoundError{Path: path}
|
||||
case errIsNotDir:
|
||||
return nil, fmt.Errorf("not a directory")
|
||||
default:
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
keys := make([]string, 0, len(keySet))
|
||||
for k := range keySet {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
return keys, nil
|
||||
return entries, nil
|
||||
}
|
||||
|
||||
// Move moves an object stored at sourcePath to destPath, removing the original
|
||||
|
|
@ -151,32 +187,37 @@ func (d *Driver) List(path string) ([]string, error) {
|
|||
func (d *Driver) Move(sourcePath string, destPath string) error {
|
||||
d.mutex.Lock()
|
||||
defer d.mutex.Unlock()
|
||||
contents, ok := d.storage[sourcePath]
|
||||
if !ok {
|
||||
return storagedriver.PathNotFoundError{Path: sourcePath}
|
||||
|
||||
normalizedSrc, normalizedDst := d.normalize(sourcePath), d.normalize(destPath)
|
||||
|
||||
err := d.root.move(normalizedSrc, normalizedDst)
|
||||
switch err {
|
||||
case errNotExists:
|
||||
return storagedriver.PathNotFoundError{Path: destPath}
|
||||
default:
|
||||
return err
|
||||
}
|
||||
d.storage[destPath] = contents
|
||||
delete(d.storage, sourcePath)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Delete recursively deletes all objects stored at "path" and its subpaths.
|
||||
func (d *Driver) Delete(path string) error {
|
||||
d.mutex.Lock()
|
||||
defer d.mutex.Unlock()
|
||||
var subPaths []string
|
||||
for k := range d.storage {
|
||||
if strings.HasPrefix(k, path) {
|
||||
subPaths = append(subPaths, k)
|
||||
}
|
||||
}
|
||||
|
||||
if len(subPaths) == 0 {
|
||||
normalized := d.normalize(path)
|
||||
|
||||
err := d.root.delete(normalized)
|
||||
switch err {
|
||||
case errNotExists:
|
||||
return storagedriver.PathNotFoundError{Path: path}
|
||||
default:
|
||||
return err
|
||||
}
|
||||
|
||||
for _, subPath := range subPaths {
|
||||
delete(d.storage, subPath)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Driver) normalize(p string) string {
|
||||
if !strings.HasPrefix(p, "/") {
|
||||
p = "/" + p // Ghetto path absolution.
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue