X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/056a3fa0fcace48ea0053ed5b89cd6e0d1ca792e..415e641036ebcfb944a37ffa14a101f2545104c3:/sdk/go/arvados/collection_fs.go

diff --git a/sdk/go/arvados/collection_fs.go b/sdk/go/arvados/collection_fs.go
index 79626500c1..45c0731b08 100644
--- a/sdk/go/arvados/collection_fs.go
+++ b/sdk/go/arvados/collection_fs.go
@@ -24,7 +24,11 @@ var (
 	ErrNegativeOffset    = errors.New("cannot seek to negative offset")
 	ErrFileExists        = errors.New("file exists")
 	ErrInvalidOperation  = errors.New("invalid operation")
+	ErrInvalidArgument   = errors.New("invalid argument")
 	ErrDirectoryNotEmpty = errors.New("directory not empty")
+	ErrWriteOnlyMode     = errors.New("file is O_WRONLY")
+	ErrSyncNotSupported  = errors.New("O_SYNC flag is not supported")
+	ErrIsDirectory       = errors.New("cannot rename file to overwrite existing directory")
 	ErrPermission        = os.ErrPermission
 
 	maxBlockSize = 1 << 26
@@ -84,15 +88,37 @@ func (fi fileinfo) Sys() interface{} {
 }
 
 // A CollectionFileSystem is an http.Filesystem plus Stat() and
-// support for opening writable files.
+// support for opening writable files. All methods are safe to call
+// from multiple goroutines.
 type CollectionFileSystem interface {
 	http.FileSystem
+
+	// analogous to os.Stat()
 	Stat(name string) (os.FileInfo, error)
+
+	// analogous to os.Create(): create/truncate a file and open it O_RDWR.
 	Create(name string) (File, error)
+
+	// Like os.OpenFile(): create or open a file or directory.
+	//
+	// If flag&os.O_EXCL==0, it opens an existing file or
+	// directory if one exists. If flag&os.O_CREATE!=0, it creates
+	// a new empty file or directory if one does not already
+	// exist.
+	//
+	// When creating a new item, perm&os.ModeDir determines
+	// whether it is a file or a directory.
+	//
+	// A file can be opened multiple times and used concurrently
+	// from multiple goroutines. However, each File object should
+	// be used by only one goroutine at a time.
 	OpenFile(name string, flag int, perm os.FileMode) (File, error)
+
 	Mkdir(name string, perm os.FileMode) error
 	Remove(name string) error
-	MarshalManifest(string) (string, error)
+	RemoveAll(name string) error
+	Rename(oldname, newname string) error
+	MarshalManifest(prefix string) (string, error)
 }
 
 type fileSystem struct {
@@ -174,7 +200,6 @@ func (fn *filenode) seek(startPtr filenodePtr) (ptr filenodePtr) {
 		// meaningless anyway
 		return
 	} else if ptr.off >= fn.fileinfo.size {
-		ptr.off = fn.fileinfo.size
 		ptr.extentIdx = len(fn.extents)
 		ptr.extentOff = 0
 		ptr.repacked = fn.repacked
@@ -231,8 +256,6 @@ func (fn *filenode) Readdir() []os.FileInfo {
 }
 
 func (fn *filenode) Read(p []byte, startPtr filenodePtr) (n int, ptr filenodePtr, err error) {
-	fn.RLock()
-	defer fn.RUnlock()
 	ptr = fn.seek(startPtr)
 	if ptr.off < 0 {
 		err = ErrNegativeOffset
@@ -272,8 +295,16 @@ func (fn *filenode) Stat() os.FileInfo {
 func (fn *filenode) Truncate(size int64) error {
 	fn.Lock()
 	defer fn.Unlock()
+	return fn.truncate(size)
+}
+
+func (fn *filenode) truncate(size int64) error {
+	if size == fn.fileinfo.size {
+		return nil
+	}
+	fn.repacked++
 	if size < fn.fileinfo.size {
-		ptr := fn.seek(filenodePtr{off: size, repacked: fn.repacked - 1})
+		ptr := fn.seek(filenodePtr{off: size})
 		for i := ptr.extentIdx; i < len(fn.extents); i++ {
 			if ext, ok := fn.extents[i].(*memExtent); ok {
 				fn.memsize -= int64(ext.Len())
@@ -292,7 +323,6 @@ func (fn *filenode) Truncate(size int64) error {
 			}
 		}
 		fn.fileinfo.size = size
-		fn.repacked++
 		return nil
 	}
 	for size > fn.fileinfo.size {
@@ -305,8 +335,6 @@ func (fn *filenode) Truncate(size int64) error {
 		} else if e, ok = fn.extents[len(fn.extents)-1].(writableExtent); !ok || e.Len() >= maxBlockSize {
 			e = &memExtent{}
 			fn.extents = append(fn.extents, e)
-		} else {
-			fn.repacked++
 		}
 		if maxgrow := int64(maxBlockSize - e.Len()); maxgrow < grow {
 			grow = maxgrow
@@ -318,9 +346,13 @@ func (fn *filenode) Truncate(size int64) error {
 	return nil
 }
 
+// Caller must hold lock.
 func (fn *filenode) Write(p []byte, startPtr filenodePtr) (n int, ptr filenodePtr, err error) {
-	fn.Lock()
-	defer fn.Unlock()
+	if startPtr.off > fn.fileinfo.size {
+		if err = fn.truncate(startPtr.off); err != nil {
+			return 0, startPtr, err
+		}
+	}
 	ptr = fn.seek(startPtr)
 	if ptr.off < 0 {
 		err = ErrNegativeOffset
@@ -497,11 +529,17 @@ type file struct {
 	inode
 	ptr        filenodePtr
 	append     bool
+	readable   bool
 	writable   bool
 	unreaddirs []os.FileInfo
 }
 
 func (f *file) Read(p []byte) (n int, err error) {
+	if !f.readable {
+		return 0, ErrWriteOnlyMode
+	}
+	f.inode.RLock()
+	defer f.inode.RUnlock()
 	n, f.ptr, err = f.inode.Read(p, f.ptr)
 	return
 }
@@ -520,9 +558,6 @@ func (f *file) Seek(off int64, whence int) (pos int64, err error) {
 	if ptr.off < 0 {
 		return f.ptr.off, ErrNegativeOffset
 	}
-	if ptr.off > size {
-		ptr.off = size
-	}
 	if ptr.off != f.ptr.off {
 		f.ptr = ptr
 		// force filenode to recompute f.ptr fields on next
@@ -540,6 +575,16 @@ func (f *file) Write(p []byte) (n int, err error) {
 	if !f.writable {
 		return 0, ErrReadOnlyFile
 	}
+	f.inode.Lock()
+	defer f.inode.Unlock()
+	if fn, ok := f.inode.(*filenode); ok && f.append {
+		f.ptr = filenodePtr{
+			off:       fn.fileinfo.size,
+			extentIdx: len(fn.extents),
+			extentOff: 0,
+			repacked:  fn.repacked,
+		}
+	}
 	n, f.ptr, err = f.inode.Write(p, f.ptr)
 	return
 }
@@ -699,6 +744,10 @@ func (dn *dirnode) marshalManifest(prefix string) (string, error) {
 			}
 			subdirs = subdirs + subdir
 		case *filenode:
+			if len(node.extents) == 0 {
+				segments = append(segments, m1segment{name: name})
+				break
+			}
 			for _, e := range node.extents {
 				switch e := e.(type) {
 				case storedExtent:
@@ -707,11 +756,18 @@ func (dn *dirnode) marshalManifest(prefix string) (string, error) {
 					} else {
 						blocks = append(blocks, e.locator)
 					}
-					segments = append(segments, m1segment{
+					next := m1segment{
 						name:   name,
 						offset: streamLen + int64(e.offset),
 						length: int64(e.length),
-					})
+					}
+					if prev := len(segments) - 1; prev >= 0 &&
+						segments[prev].name == name &&
+						segments[prev].offset+segments[prev].length == next.offset {
+						segments[prev].length += next.length
+					} else {
+						segments = append(segments, next)
+					}
 					streamLen += int64(e.size)
 				default:
 					// This can't happen: we
@@ -747,6 +803,8 @@ func (dn *dirnode) loadManifest(txt string) error {
 		lineno := i + 1
 		var extents []storedExtent
 		var anyFileTokens bool
+		var pos int64
+		var extIdx int
 		for i, token := range strings.Split(stream, " ") {
 			if i == 0 {
 				dirname = manifestUnescape(token)
@@ -790,26 +848,28 @@ func (dn *dirnode) loadManifest(txt string) error {
 				return fmt.Errorf("line %d: bad file segment %q", lineno, token)
 			}
 			name := path.Clean(dirname + "/" + manifestUnescape(toks[2]))
-			dn.makeParentDirs(name)
-			f, err := dn.OpenFile(name, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0700)
+			fnode, err := dn.createFileAndParents(name)
 			if err != nil {
-				return fmt.Errorf("line %d: cannot append to %q: %s", lineno, name, err)
-			}
-			if f.inode.Stat().IsDir() {
-				f.Close()
-				return fmt.Errorf("line %d: cannot append to %q: is a directory", lineno, name)
+				return fmt.Errorf("line %d: cannot use path %q: %s", lineno, name, err)
 			}
 			// Map the stream offset/range coordinates to
 			// block/offset/range coordinates and add
 			// corresponding storedExtents to the filenode
-			var pos int64
-			for _, e := range extents {
-				next := pos + int64(e.Len())
-				if next < offset {
+			if pos > offset {
+				// Can't continue where we left off.
+				// TODO: binary search instead of
+				// rewinding all the way (but this
+				// situation might be rare anyway)
+				extIdx, pos = 0, 0
+			}
+			for next := int64(0); extIdx < len(extents); extIdx, pos = extIdx+1, next {
+				e := extents[extIdx]
+				next = pos + int64(e.Len())
+				if next <= offset || e.Len() == 0 {
 					pos = next
 					continue
 				}
-				if pos > offset+length {
+				if pos >= offset+length {
 					break
 				}
 				var blkOff int
@@ -820,17 +880,18 @@ func (dn *dirnode) loadManifest(txt string) error {
 				if pos+int64(blkOff+blkLen) > offset+length {
 					blkLen = int(offset + length - pos - int64(blkOff))
 				}
-				f.inode.(*filenode).appendExtent(storedExtent{
+				fnode.appendExtent(storedExtent{
 					kc:      dn.kc,
 					locator: e.locator,
 					size:    e.size,
 					offset:  blkOff,
 					length:  blkLen,
 				})
-				pos = next
+				if next > offset+length {
+					break
+				}
 			}
-			f.Close()
-			if pos < offset+length {
+			if extIdx == len(extents) && pos < offset+length {
 				return fmt.Errorf("line %d: invalid segment in %d-byte stream: %q", lineno, pos, token)
 			}
 		}
@@ -845,21 +906,41 @@ func (dn *dirnode) loadManifest(txt string) error {
 	return nil
 }
 
-func (dn *dirnode) makeParentDirs(name string) (err error) {
-	names := strings.Split(name, "/")
-	for _, name := range names[:len(names)-1] {
-		f, err := dn.mkdir(name)
-		if err != nil {
-			return err
+// only safe to call from loadManifest -- no locking
+func (dn *dirnode) createFileAndParents(path string) (fn *filenode, err error) {
+	names := strings.Split(path, "/")
+	if basename := names[len(names)-1]; basename == "" || basename == "." || basename == ".." {
+		err = fmt.Errorf("invalid filename")
+		return
+	}
+	var node inode = dn
+	for i, name := range names {
+		dn, ok := node.(*dirnode)
+		if !ok {
+			err = ErrFileExists
+			return
 		}
-		defer f.Close()
-		var ok bool
-		dn, ok = f.inode.(*dirnode)
+		if name == "" || name == "." {
+			continue
+		}
+		if name == ".." {
+			node = dn.parent
+			continue
+		}
+		node, ok = dn.inodes[name]
 		if !ok {
-			return ErrFileExists
+			if i == len(names)-1 {
+				fn = dn.newFilenode(name, 0755)
+				return
+			}
+			node = dn.newDirnode(name, 0755)
 		}
 	}
-	return nil
+	var ok bool
+	if fn, ok = node.(*filenode); !ok {
+		err = ErrInvalidArgument
+	}
+	return
 }
 
 func (dn *dirnode) mkdir(name string) (*file, error) {
@@ -868,16 +949,24 @@ func (dn *dirnode) mkdir(name string) (*file, error) {
 
 func (dn *dirnode) Mkdir(name string, perm os.FileMode) error {
 	f, err := dn.mkdir(name)
-	if err != nil {
-		f.Close()
+	if err == nil {
+		err = f.Close()
 	}
 	return err
 }
 
 func (dn *dirnode) Remove(name string) error {
+	return dn.remove(name, false)
+}
+
+func (dn *dirnode) RemoveAll(name string) error {
+	return dn.remove(name, true)
+}
+
+func (dn *dirnode) remove(name string, recursive bool) error {
 	dirname, name := path.Split(name)
 	if name == "" || name == "." || name == ".." {
-		return ErrInvalidOperation
+		return ErrInvalidArgument
 	}
 	dn, ok := dn.lookupPath(dirname).(*dirnode)
 	if !ok {
@@ -891,7 +980,7 @@ func (dn *dirnode) Remove(name string) error {
 	case *dirnode:
 		node.RLock()
 		defer node.RUnlock()
-		if len(node.inodes) > 0 {
+		if !recursive && len(node.inodes) > 0 {
 			return ErrDirectoryNotEmpty
 		}
 	}
@@ -899,6 +988,79 @@ func (dn *dirnode) Remove(name string) error {
 	return nil
 }
 
+func (dn *dirnode) Rename(oldname, newname string) error {
+	olddir, oldname := path.Split(oldname)
+	if oldname == "" || oldname == "." || oldname == ".." {
+		return ErrInvalidArgument
+	}
+	olddirf, err := dn.OpenFile(olddir+".", os.O_RDONLY, 0)
+	if err != nil {
+		return fmt.Errorf("%q: %s", olddir, err)
+	}
+	defer olddirf.Close()
+	newdir, newname := path.Split(newname)
+	if newname == "." || newname == ".." {
+		return ErrInvalidArgument
+	} else if newname == "" {
+		// Rename("a/b", "c/") means Rename("a/b", "c/b")
+		newname = oldname
+	}
+	newdirf, err := dn.OpenFile(newdir+".", os.O_RDONLY, 0)
+	if err != nil {
+		return fmt.Errorf("%q: %s", newdir, err)
+	}
+	defer newdirf.Close()
+
+	// When acquiring locks on multiple nodes, all common
+	// ancestors must be locked first in order to avoid
+	// deadlock. This is assured by locking the path from root to
+	// newdir, then locking the path from root to olddir, skipping
+	// any already-locked nodes.
+	needLock := []sync.Locker{}
+	for _, f := range []*file{olddirf, newdirf} {
+		node := f.inode
+		needLock = append(needLock, node)
+		for node.Parent() != node {
+			node = node.Parent()
+			needLock = append(needLock, node)
+		}
+	}
+	locked := map[sync.Locker]bool{}
+	for i := len(needLock) - 1; i >= 0; i-- {
+		if n := needLock[i]; !locked[n] {
+			n.Lock()
+			defer n.Unlock()
+			locked[n] = true
+		}
+	}
+
+	olddn := olddirf.inode.(*dirnode)
+	newdn := newdirf.inode.(*dirnode)
+	oldinode, ok := olddn.inodes[oldname]
+	if !ok {
+		return os.ErrNotExist
+	}
+	if existing, ok := newdn.inodes[newname]; ok {
+		// overwriting an existing file or dir
+		if dn, ok := existing.(*dirnode); ok {
+			if !oldinode.Stat().IsDir() {
+				return ErrIsDirectory
+			}
+			dn.RLock()
+			defer dn.RUnlock()
+			if len(dn.inodes) > 0 {
+				return ErrDirectoryNotEmpty
+			}
+		}
+	} else {
+		newdn.fileinfo.size++
+	}
+	newdn.inodes[newname] = oldinode
+	delete(olddn.inodes, oldname)
+	olddn.fileinfo.size--
+	return nil
+}
+
 func (dn *dirnode) Parent() inode {
 	dn.RLock()
 	defer dn.RUnlock()
@@ -963,14 +1125,63 @@ func (dn *dirnode) lookupPath(path string) (node inode) {
 	return
 }
 
+func (dn *dirnode) newDirnode(name string, perm os.FileMode) *dirnode {
+	child := &dirnode{
+		parent: dn,
+		client: dn.client,
+		kc:     dn.kc,
+		fileinfo: fileinfo{
+			name: name,
+			mode: os.ModeDir | perm,
+		},
+	}
+	if dn.inodes == nil {
+		dn.inodes = make(map[string]inode)
+	}
+	dn.inodes[name] = child
+	dn.fileinfo.size++
+	return child
+}
+
+func (dn *dirnode) newFilenode(name string, perm os.FileMode) *filenode {
+	child := &filenode{
+		parent: dn,
+		fileinfo: fileinfo{
+			name: name,
+			mode: perm,
+		},
+	}
+	if dn.inodes == nil {
+		dn.inodes = make(map[string]inode)
+	}
+	dn.inodes[name] = child
+	dn.fileinfo.size++
+	return child
+}
+
+// OpenFile is analogous to os.OpenFile().
 func (dn *dirnode) OpenFile(name string, flag int, perm os.FileMode) (*file, error) {
+	if flag&os.O_SYNC != 0 {
+		return nil, ErrSyncNotSupported
+	}
 	dirname, name := path.Split(name)
 	dn, ok := dn.lookupPath(dirname).(*dirnode)
 	if !ok {
 		return nil, os.ErrNotExist
 	}
-	writeMode := flag&(os.O_RDWR|os.O_WRONLY|os.O_CREATE) != 0
-	if !writeMode {
+	var readable, writable bool
+	switch flag & (os.O_RDWR | os.O_RDONLY | os.O_WRONLY) {
+	case os.O_RDWR:
+		readable = true
+		writable = true
+	case os.O_RDONLY:
+		readable = true
+	case os.O_WRONLY:
+		writable = true
+	default:
+		return nil, fmt.Errorf("invalid flags 0x%x", flag)
+	}
+	if !writable {
 		// A directory can be opened via "foo/", "foo/.", or
 		// "foo/..".
 		switch name {
@@ -994,36 +1205,26 @@ func (dn *dirnode) OpenFile(name string, flag int, perm os.FileMode) (*file, err
 			return nil, os.ErrNotExist
 		}
 		if perm.IsDir() {
-			n = &dirnode{
-				parent: dn,
-				client: dn.client,
-				kc:     dn.kc,
-				fileinfo: fileinfo{
-					name: name,
-					mode: os.ModeDir | 0755,
-				},
-			}
+			n = dn.newDirnode(name, 0755)
 		} else {
-			n = &filenode{
-				parent: dn,
-				fileinfo: fileinfo{
-					name: name,
-					mode: 0755,
-				},
-			}
-		}
-		if dn.inodes == nil {
-			dn.inodes = make(map[string]inode)
+			n = dn.newFilenode(name, 0755)
 		}
-		dn.inodes[name] = n
-		dn.fileinfo.size++
 	} else if flag&os.O_EXCL != 0 {
 		return nil, ErrFileExists
+	} else if flag&os.O_TRUNC != 0 {
+		if !writable {
+			return nil, fmt.Errorf("invalid flag O_TRUNC in read-only mode")
+		} else if fn, ok := n.(*filenode); !ok {
+			return nil, fmt.Errorf("invalid flag O_TRUNC when opening directory")
+		} else {
+			fn.Truncate(0)
+		}
 	}
 	return &file{
 		inode:    n,
 		append:   flag&os.O_APPEND != 0,
-		writable: flag&(os.O_WRONLY|os.O_RDWR) != 0,
+		readable: readable,
+		writable: writable,
 	}, nil
 }