X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/64eeb3536c7b3ce082bc98cfc48ef045952f69e8..37e566e4a6246a1d9ca59489c3be6ec9f494e3b8:/sdk/go/arvados/fs_collection.go

diff --git a/sdk/go/arvados/fs_collection.go b/sdk/go/arvados/fs_collection.go
index d087fd0944..7903b4737c 100644
--- a/sdk/go/arvados/fs_collection.go
+++ b/sdk/go/arvados/fs_collection.go
@@ -44,9 +44,17 @@ type CollectionFileSystem interface {
 type collectionFileSystem struct {
 	fileSystem
 	uuid           string
-	savedPDH       atomic.Value
 	replicas       int
 	storageClasses []string
+
+	// PDH returned by the server as of last sync/load.
+	loadedPDH atomic.Value
+	// PDH of the locally generated manifest as of last
+	// sync/load. This can differ from loadedPDH after loading a
+	// version that was generated with different code and sorts
+	// filenames differently than we do, for example.
+	savedPDH atomic.Value
+
 	// guessSignatureTTL tracks a lower bound for the server's
 	// configured BlobSigningTTL. The guess is initially zero, and
 	// increases when we come across a signature with an expiry
@@ -74,7 +82,7 @@ func (c *Collection) FileSystem(client apiClient, kc keepClient) (CollectionFile
 			thr:       newThrottle(concurrentWriters),
 		},
 	}
-	fs.savedPDH.Store(c.PortableDataHash)
+	fs.loadedPDH.Store(c.PortableDataHash)
 	if r := c.ReplicationDesired; r != nil {
 		fs.replicas = *r
 	}
@@ -85,6 +93,7 @@ func (c *Collection) FileSystem(client apiClient, kc keepClient) (CollectionFile
 				name:    ".",
 				mode:    os.ModeDir | 0755,
 				modTime: modTime,
+				sys:     func() interface{} { return c },
 			},
 			inodes: make(map[string]inode),
 		},
@@ -93,6 +102,13 @@ func (c *Collection) FileSystem(client apiClient, kc keepClient) (CollectionFile
 	if err := root.loadManifest(c.ManifestText); err != nil {
 		return nil, err
 	}
+
+	txt, err := root.marshalManifest(context.Background(), ".", false)
+	if err != nil {
+		return nil, err
+	}
+	fs.savedPDH.Store(PortableDataHash(txt))
+
 	backdateTree(root, modTime)
 	fs.root = root
 	return fs, nil
@@ -289,44 +305,72 @@ func (fs *collectionFileSystem) Truncate(int64) error {
 	return ErrInvalidOperation
 }
 
-// Check for and incorporate upstream changes -- unless that has
-// already been done recently, in which case this func is a no-op.
-func (fs *collectionFileSystem) checkChangesOnServer() error {
-	if fs.uuid == "" && fs.savedPDH.Load() == "" {
-		return nil
+// Check for and incorporate upstream changes. If force==false, this
+// is a no-op except once every ttl/100 or so.
+//
+// Return value is true if new content was loaded from upstream and
+// any unsaved local changes have been discarded.
+func (fs *collectionFileSystem) checkChangesOnServer(force bool) (bool, error) {
+	if fs.uuid == "" && fs.loadedPDH.Load() == "" {
+		return false, nil
 	}
 
-	// First try UUID if any, then last known PDH. Stop if all
-	// signatures are new enough.
-	checkingAll := false
-	for _, id := range []string{fs.uuid, fs.savedPDH.Load().(string)} {
-		if id == "" {
-			continue
-		}
-
-		fs.lockCheckChanges.Lock()
-		if !checkingAll && fs.holdCheckChanges.After(time.Now()) {
-			fs.lockCheckChanges.Unlock()
-			return nil
-		}
-		remain, ttl := fs.signatureTimeLeft()
-		if remain > 0.01 && !checkingAll {
-			fs.holdCheckChanges = time.Now().Add(ttl / 100)
-		}
+	fs.lockCheckChanges.Lock()
+	if !force && fs.holdCheckChanges.After(time.Now()) {
 		fs.lockCheckChanges.Unlock()
+		return false, nil
+	}
+	remain, ttl := fs.signatureTimeLeft()
+	if remain > 0.01 {
+		fs.holdCheckChanges = time.Now().Add(ttl / 100)
+	}
+	fs.lockCheckChanges.Unlock()
 
-		if remain >= 0.5 {
-			break
+	if !force && remain >= 0.5 {
+		// plenty of time left on current signatures
+		return false, nil
+	}
+
+	loadedPDH, _ := fs.loadedPDH.Load().(string)
+	getparams := map[string]interface{}{"select": []string{"portable_data_hash", "manifest_text"}}
+	if fs.uuid != "" {
+		var coll Collection
+		err := fs.RequestAndDecode(&coll, "GET", "arvados/v1/collections/"+fs.uuid, nil, getparams)
+		if err != nil {
+			return false, err
+		}
+		if coll.PortableDataHash != loadedPDH {
+			// collection has changed upstream since we
+			// last loaded or saved. Refresh local data,
+			// losing any unsaved local changes.
+			newfs, err := coll.FileSystem(fs.fileSystem.fsBackend, fs.fileSystem.fsBackend)
+			if err != nil {
+				return false, err
+			}
+			snap, err := Snapshot(newfs, "/")
+			if err != nil {
+				return false, err
+			}
+			err = Splice(fs, "/", snap)
+			if err != nil {
+				return false, err
+			}
+			fs.loadedPDH.Store(coll.PortableDataHash)
+			fs.savedPDH.Store(newfs.(*collectionFileSystem).savedPDH.Load())
+			return true, nil
 		}
-		checkingAll = true
+		fs.updateSignatures(coll.ManifestText)
+		return false, nil
+	}
+	if loadedPDH != "" {
 		var coll Collection
-		err := fs.RequestAndDecode(&coll, "GET", "arvados/v1/collections/"+id, nil, map[string]interface{}{"select": []string{"portable_data_hash", "manifest_text"}})
+		err := fs.RequestAndDecode(&coll, "GET", "arvados/v1/collections/"+loadedPDH, nil, getparams)
 		if err != nil {
-			continue
+			return false, err
 		}
 		fs.updateSignatures(coll.ManifestText)
 	}
-	return nil
+	return false, nil
 }
 
 // Refresh signature on a single locator, if necessary. Assume caller
@@ -338,11 +382,12 @@ func (fs *collectionFileSystem) refreshSignature(locator string) string {
 	if err != nil || exp.Sub(time.Now()) > time.Minute {
 		// Synchronous update is not needed. Start an
 		// asynchronous update if needed.
-		go fs.checkChangesOnServer()
+		go fs.checkChangesOnServer(false)
 		return locator
 	}
+	loadedPDH, _ := fs.loadedPDH.Load().(string)
 	var manifests string
-	for _, id := range []string{fs.uuid, fs.savedPDH.Load().(string)} {
+	for _, id := range []string{fs.uuid, loadedPDH} {
 		if id == "" {
 			continue
 		}
@@ -367,18 +412,19 @@ func (fs *collectionFileSystem) refreshSignature(locator string) string {
 }
 
 func (fs *collectionFileSystem) Sync() error {
-	err := fs.checkChangesOnServer()
+	refreshed, err := fs.checkChangesOnServer(true)
 	if err != nil {
 		return err
 	}
-	if fs.uuid == "" {
+	if refreshed || fs.uuid == "" {
 		return nil
 	}
 	txt, err := fs.MarshalManifest(".")
 	if err != nil {
 		return fmt.Errorf("sync failed: %s", err)
 	}
-	if PortableDataHash(txt) == fs.savedPDH.Load() {
+	savingPDH := PortableDataHash(txt)
+	if savingPDH == fs.savedPDH.Load() {
 		// No local changes since last save or initial load.
 		return nil
 	}
@@ -402,15 +448,16 @@ func (fs *collectionFileSystem) Sync() error {
 		"select": selectFields,
 	})
 	if err != nil {
-		return fmt.Errorf("sync failed: update %s: %s", fs.uuid, err)
+		return fmt.Errorf("sync failed: update %s: %w", fs.uuid, err)
 	}
 	fs.updateSignatures(coll.ManifestText)
-	fs.savedPDH.Store(coll.PortableDataHash)
+	fs.loadedPDH.Store(coll.PortableDataHash)
+	fs.savedPDH.Store(savingPDH)
 	return nil
 }
 
 func (fs *collectionFileSystem) Flush(path string, shortBlocks bool) error {
-	node, err := rlookup(fs.fileSystem.root, path)
+	node, err := rlookup(fs.fileSystem.root, path, nil)
 	if err != nil {
 		return err
 	}
@@ -442,21 +489,27 @@ func (fs *collectionFileSystem) Flush(path string, shortBlocks bool) error {
 }
 
 func (fs *collectionFileSystem) MemorySize() int64 {
-	fs.fileSystem.root.Lock()
-	defer fs.fileSystem.root.Unlock()
 	return fs.fileSystem.root.(*dirnode).MemorySize()
 }
 
 func (fs *collectionFileSystem) MarshalManifest(prefix string) (string, error) {
 	fs.fileSystem.root.Lock()
 	defer fs.fileSystem.root.Unlock()
-	return fs.fileSystem.root.(*dirnode).marshalManifest(context.TODO(), prefix)
+	return fs.fileSystem.root.(*dirnode).marshalManifest(context.TODO(), prefix, true)
 }
 
 func (fs *collectionFileSystem) Size() int64 {
 	return fs.fileSystem.root.(*dirnode).TreeSize()
 }
 
+func (fs *collectionFileSystem) Snapshot() (inode, error) {
+	return fs.fileSystem.root.Snapshot()
+}
+
+func (fs *collectionFileSystem) Splice(r inode) error {
+	return fs.fileSystem.root.Splice(r)
+}
+
 // filenodePtr is an offset into a file that is (usually) efficient to
 // seek to. Specifically, if filenode.repacked==filenodePtr.repacked
 // then
@@ -480,9 +533,9 @@ type filenodePtr struct {
 //
 // After seeking:
 //
-//     ptr.segmentIdx == len(filenode.segments) // i.e., at EOF
-//     ||
-//     filenode.segments[ptr.segmentIdx].Len() > ptr.segmentOff
+//	ptr.segmentIdx == len(filenode.segments) // i.e., at EOF
+//	||
+//	filenode.segments[ptr.segmentIdx].Len() > ptr.segmentOff
 func (fn *filenode) seek(startPtr filenodePtr) (ptr filenodePtr) {
 	ptr = startPtr
 	if ptr.off < 0 {
@@ -567,6 +620,16 @@ func (fn *filenode) FS() FileSystem {
 	return fn.fs
 }
 
+func (fn *filenode) MemorySize() (size int64) {
+	fn.RLock()
+	defer fn.RUnlock()
+	size = 64
+	for _, seg := range fn.segments {
+		size += seg.memorySize()
+	}
+	return
+}
+
 // Read reads file data from a single segment, starting at startPtr,
 // into p. startPtr is assumed not to be up-to-date. Caller must have
 // RLock or Lock.
@@ -876,6 +939,47 @@ func (fn *filenode) waitPrune() {
 	}
 }
 
+func (fn *filenode) Snapshot() (inode, error) {
+	fn.RLock()
+	defer fn.RUnlock()
+	segments := make([]segment, 0, len(fn.segments))
+	for _, seg := range fn.segments {
+		segments = append(segments, seg.Slice(0, seg.Len()))
+	}
+	return &filenode{
+		fileinfo: fn.fileinfo,
+		segments: segments,
+	}, nil
+}
+
+func (fn *filenode) Splice(repl inode) error {
+	repl, err := repl.Snapshot()
+	if err != nil {
+		return err
+	}
+	fn.parent.Lock()
+	defer fn.parent.Unlock()
+	fn.Lock()
+	defer fn.Unlock()
+	_, err = fn.parent.Child(fn.fileinfo.name, func(inode) (inode, error) { return repl, nil })
+	if err != nil {
+		return err
+	}
+	switch repl := repl.(type) {
+	case *dirnode:
+		repl.parent = fn.parent
+		repl.fileinfo.name = fn.fileinfo.name
+		repl.setTreeFS(fn.fs)
+	case *filenode:
+		repl.parent = fn.parent
+		repl.fileinfo.name = fn.fileinfo.name
+		repl.fs = fn.fs
+	default:
+		return fmt.Errorf("cannot splice snapshot containing %T: %w", repl, ErrInvalidArgument)
+	}
+	return nil
+}
+
 type dirnode struct {
 	fs *collectionFileSystem
 	treenode
@@ -1100,23 +1204,16 @@ func (dn *dirnode) flush(ctx context.Context, names []string, opts flushOpts) er
 	return cg.Wait()
 }
 
-// caller must have write lock.
 func (dn *dirnode) MemorySize() (size int64) {
-	for _, name := range dn.sortedNames() {
-		node := dn.inodes[name]
-		node.Lock()
-		defer node.Unlock()
-		switch node := node.(type) {
-		case *dirnode:
-			size += node.MemorySize()
-		case *filenode:
-			for _, seg := range node.segments {
-				switch seg := seg.(type) {
-				case *memSegment:
-					size += int64(seg.Len())
-				}
-			}
-		}
+	dn.RLock()
+	todo := make([]inode, 0, len(dn.inodes))
+	for _, node := range dn.inodes {
+		todo = append(todo, node)
+	}
+	dn.RUnlock()
+	size = 64
+	for _, node := range todo {
+		size += node.MemorySize()
 	}
 	return
 }
@@ -1132,7 +1229,7 @@ func (dn *dirnode) sortedNames() []string {
 }
 
 // caller must have write lock.
-func (dn *dirnode) marshalManifest(ctx context.Context, prefix string) (string, error) {
+func (dn *dirnode) marshalManifest(ctx context.Context, prefix string, flush bool) (string, error) {
 	cg := newContextGroup(ctx)
 	defer cg.Cancel()
 
@@ -1179,7 +1276,7 @@ func (dn *dirnode) marshalManifest(ctx context.Context, prefix string) (string,
 	for i, name := range dirnames {
 		i, name := i, name
 		cg.Go(func() error {
-			txt, err := dn.inodes[name].(*dirnode).marshalManifest(cg.Context(), prefix+"/"+name)
+			txt, err := dn.inodes[name].(*dirnode).marshalManifest(cg.Context(), prefix+"/"+name, flush)
 			subdirs[i] = txt
 			return err
 		})
@@ -1195,7 +1292,10 @@ func (dn *dirnode) marshalManifest(ctx context.Context, prefix string) (string,
 
 		var fileparts []filepart
 		var blocks []string
-		if err := dn.flush(cg.Context(), filenames, flushOpts{sync: true, shortBlocks: true}); err != nil {
+		if !flush {
+			// skip flush -- will fail below if anything
+			// needed flushing
+		} else if err := dn.flush(cg.Context(), filenames, flushOpts{sync: true, shortBlocks: true}); err != nil {
 			return err
 		}
 		for _, name := range filenames {
@@ -1226,10 +1326,12 @@ func (dn *dirnode) marshalManifest(ctx context.Context, prefix string) (string,
 					}
 					streamLen += int64(seg.size)
 				default:
-					// This can't happen: we
-					// haven't unlocked since
+					// We haven't unlocked since
 					// calling flush(sync=true).
-					panic(fmt.Sprintf("can't marshal segment type %T", seg))
+					// Evidently the caller passed
+					// flush==false but there were
+					// local changes.
+					return fmt.Errorf("can't marshal segment type %T", seg)
 				}
 			}
 		}
@@ -1356,7 +1458,7 @@ func (dn *dirnode) loadManifest(txt string) error {
 				segIdx, pos = 0, 0
 			}
 			for ; segIdx < len(segments); segIdx++ {
-				seg := segments[segIdx]
+				seg := &segments[segIdx]
 				next := pos + int64(seg.Len())
 				if next <= offset || seg.Len() == 0 {
 					pos = next
@@ -1489,12 +1591,93 @@ func (dn *dirnode) TreeSize() (bytes int64) {
 	return
 }
 
+func (dn *dirnode) Snapshot() (inode, error) {
+	return dn.snapshot()
+}
+
+func (dn *dirnode) snapshot() (*dirnode, error) {
+	dn.RLock()
+	defer dn.RUnlock()
+	snap := &dirnode{
+		treenode: treenode{
+			inodes:   make(map[string]inode, len(dn.inodes)),
+			fileinfo: dn.fileinfo,
+		},
+	}
+	for name, child := range dn.inodes {
+		dupchild, err := child.Snapshot()
+		if err != nil {
+			return nil, err
+		}
+		snap.inodes[name] = dupchild
+		dupchild.SetParent(snap, name)
+	}
+	return snap, nil
+}
+
+func (dn *dirnode) Splice(repl inode) error {
+	repl, err := repl.Snapshot()
+	if err != nil {
+		return fmt.Errorf("cannot copy snapshot: %w", err)
+	}
+	switch repl := repl.(type) {
+	default:
+		return fmt.Errorf("cannot splice snapshot containing %T: %w", repl, ErrInvalidArgument)
+	case *dirnode:
+		dn.Lock()
+		defer dn.Unlock()
+		dn.inodes = repl.inodes
+		dn.setTreeFS(dn.fs)
+	case *filenode:
+		dn.parent.Lock()
+		defer dn.parent.Unlock()
+		removing, err := dn.parent.Child(dn.fileinfo.name, nil)
+		if err != nil {
+			return fmt.Errorf("cannot use Splice to replace a top-level directory with a file: %w", ErrInvalidOperation)
+		} else if removing != dn {
+			// If ../thisdirname is not this dirnode, it
+			// must be an inode that wraps a dirnode, like
+			// a collectionFileSystem or deferrednode.
+			if deferred, ok := removing.(*deferrednode); ok {
+				// More useful to report the type of
+				// the wrapped node rather than just
+				// *deferrednode. (We know the real
+				// inode is already loaded because dn
+				// is inside it.)
+				removing = deferred.realinode()
+			}
+			return fmt.Errorf("cannot use Splice to attach a file at top level of %T: %w", removing, ErrInvalidOperation)
+		}
+		dn.Lock()
+		defer dn.Unlock()
+		_, err = dn.parent.Child(dn.fileinfo.name, func(inode) (inode, error) { return repl, nil })
+		if err != nil {
+			return fmt.Errorf("error replacing filenode: dn.parent.Child(): %w", err)
+		}
+		repl.fs = dn.fs
+	}
+	return nil
+}
+
+func (dn *dirnode) setTreeFS(fs *collectionFileSystem) {
+	dn.fs = fs
+	for _, child := range dn.inodes {
+		switch child := child.(type) {
+		case *dirnode:
+			child.setTreeFS(fs)
+		case *filenode:
+			child.fs = fs
+		}
+	}
+}
+
 type segment interface {
 	io.ReaderAt
 	Len() int
 	// Return a new segment with a subsection of the data from this
 	// one. length<0 means length=Len()-off.
 	Slice(off int, length int) segment
+	memorySize() int64
 }
 
 type memSegment struct {
@@ -1573,6 +1756,10 @@ func (me *memSegment) ReadAt(p []byte, off int64) (n int, err error) {
 	return
 }
 
+func (me *memSegment) memorySize() int64 {
+	return 64 + int64(len(me.buf))
+}
+
 type storedSegment struct {
 	kc      fsBackend
 	locator string
@@ -1610,6 +1797,10 @@ func (se storedSegment) ReadAt(p []byte, off int64) (n int, err error) {
 	return se.kc.ReadAt(se.locator, p, int(off)+se.offset)
 }
 
+func (se storedSegment) memorySize() int64 {
+	return 64 + int64(len(se.locator))
+}
+
 func canonicalName(name string) string {
 	name = path.Clean("/" + name)
 	if name == "/" || name == "./" {