sdk/go/arvados/fs_base.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: Apache-2.0
   4
   5 package arvados
   6
   7 import (
   8         "errors"
   9         "fmt"
  10         "io"
  11         "io/fs"
  12         "log"
  13         "net/http"
  14         "os"
  15         "path"
  16         "strings"
  17         "sync"
  18         "time"
  19 )
  20
  21 var (
  22         ErrReadOnlyFile      = errors.New("read-only file")
  23         ErrNegativeOffset    = errors.New("cannot seek to negative offset")
  24         ErrFileExists        = errors.New("file exists")
  25         ErrInvalidOperation  = errors.New("invalid operation")
  26         ErrInvalidArgument   = errors.New("invalid argument")
  27         ErrDirectoryNotEmpty = errors.New("directory not empty")
  28         ErrWriteOnlyMode     = errors.New("file is O_WRONLY")
  29         ErrSyncNotSupported  = errors.New("O_SYNC flag is not supported")
  30         ErrIsDirectory       = errors.New("cannot rename file to overwrite existing directory")
  31         ErrNotADirectory     = errors.New("not a directory")
  32         ErrPermission        = os.ErrPermission
  33         DebugLocksPanicMode  = false
  34 )
  35
  36 type syncer interface {
  37         Sync() error
  38 }
  39
  40 func debugPanicIfNotLocked(l sync.Locker, writing bool) {
  41         if !DebugLocksPanicMode {
  42                 return
  43         }
  44         race := false
  45         if rl, ok := l.(interface {
  46                 RLock()
  47                 RUnlock()
  48         }); ok && writing {
  49                 go func() {
  50                         // Fail if we can grab the read lock during an
  51                         // operation that purportedly has write lock.
  52                         rl.RLock()
  53                         race = true
  54                         rl.RUnlock()
  55                 }()
  56         } else {
  57                 go func() {
  58                         l.Lock()
  59                         race = true
  60                         l.Unlock()
  61                 }()
  62         }
  63         time.Sleep(100)
  64         if race {
  65                 panic("bug: caller-must-have-lock func called, but nobody has lock")
  66         }
  67 }
  68
  69 // A File is an *os.File-like interface for reading and writing files
  70 // in a FileSystem.
  71 type File interface {
  72         io.Reader
  73         io.Writer
  74         io.Closer
  75         io.Seeker
  76         Size() int64
  77         Readdir(int) ([]os.FileInfo, error)
  78         Stat() (os.FileInfo, error)
  79         Truncate(int64) error
  80         Sync() error
  81         // Create a snapshot of a file or directory tree, which can
  82         // then be spliced onto a different path or a different
  83         // collection.
  84         Snapshot() (*Subtree, error)
  85         // Replace this file or directory with the given snapshot.
  86         // The target must be inside a collection: Splice returns an
  87         // error if the File is a virtual file or directory like
  88         // by_id, a project directory, .arvados#collection,
  89         // etc. Splice can replace directories with regular files and
  90         // vice versa, except it cannot replace the root directory of
  91         // a collection with a regular file.
  92         Splice(snapshot *Subtree) error
  93 }
  94
  95 // A Subtree is a detached part of a filesystem tree that can be
  96 // spliced into a filesystem via (File)Splice().
  97 type Subtree struct {
  98         inode inode
  99 }
 100
 101 // A FileSystem is an http.Filesystem plus Stat() and support for
 102 // opening writable files. All methods are safe to call from multiple
 103 // goroutines.
 104 type FileSystem interface {
 105         http.FileSystem
 106         fsBackend
 107
 108         rootnode() inode
 109
 110         // filesystem-wide lock: used by Rename() to prevent deadlock
 111         // while locking multiple inodes.
 112         locker() sync.Locker
 113
 114         // throttle for limiting concurrent background writers
 115         throttle() *throttle
 116
 117         // create a new node with nil parent.
 118         newNode(name string, perm os.FileMode, modTime time.Time) (node inode, err error)
 119
 120         // analogous to os.Stat()
 121         Stat(name string) (os.FileInfo, error)
 122
 123         // analogous to os.Create(): create/truncate a file and open it O_RDWR.
 124         Create(name string) (File, error)
 125
 126         // Like os.OpenFile(): create or open a file or directory.
 127         //
 128         // If flag&os.O_EXCL==0, it opens an existing file or
 129         // directory if one exists. If flag&os.O_CREATE!=0, it creates
 130         // a new empty file or directory if one does not already
 131         // exist.
 132         //
 133         // When creating a new item, perm&os.ModeDir determines
 134         // whether it is a file or a directory.
 135         //
 136         // A file can be opened multiple times and used concurrently
 137         // from multiple goroutines. However, each File object should
 138         // be used by only one goroutine at a time.
 139         OpenFile(name string, flag int, perm os.FileMode) (File, error)
 140
 141         Mkdir(name string, perm os.FileMode) error
 142         Remove(name string) error
 143         RemoveAll(name string) error
 144         Rename(oldname, newname string) error
 145
 146         // Write buffered data from memory to storage, returning when
 147         // all updates have been saved to persistent storage.
 148         Sync() error
 149
 150         // Write buffered data from memory to storage, but don't wait
 151         // for all writes to finish before returning. If shortBlocks
 152         // is true, flush everything; otherwise, if there's less than
 153         // a full block of buffered data at the end of a stream, leave
 154         // it buffered in memory in case more data can be appended. If
 155         // path is "", flush all dirs/streams; otherwise, flush only
 156         // the specified dir/stream.
 157         Flush(path string, shortBlocks bool) error
 158
 159         // Estimate current memory usage.
 160         MemorySize() int64
 161 }
 162
 163 type fsFS struct {
 164         FileSystem
 165 }
 166
 167 // FS returns an fs.FS interface to the given FileSystem, to enable
 168 // the use of fs.WalkDir, etc.
 169 func FS(fs FileSystem) fs.FS { return fsFS{fs} }
 170 func (fs fsFS) Open(path string) (fs.File, error) {
 171         f, err := fs.FileSystem.Open(path)
 172         return f, err
 173 }
 174
 175 type inode interface {
 176         SetParent(parent inode, name string)
 177         Parent() inode
 178         FS() FileSystem
 179         Read([]byte, filenodePtr) (int, filenodePtr, error)
 180         Write([]byte, filenodePtr) (int, filenodePtr, error)
 181         Truncate(int64) error
 182         IsDir() bool
 183         Readdir() ([]os.FileInfo, error)
 184         Size() int64
 185         FileInfo() os.FileInfo
 186         // Create a snapshot of this node and its descendants.
 187         Snapshot() (inode, error)
 188         // Replace this node with a copy of the provided snapshot.
 189         // Caller may provide the same snapshot to multiple Splice
 190         // calls, but must not modify the snapshot concurrently.
 191         Splice(inode) error
 192
 193         // Child() performs lookups and updates of named child nodes.
 194         //
 195         // (The term "child" here is used strictly. This means name is
 196         // not "." or "..", and name does not contain "/".)
 197         //
 198         // If replace is non-nil, Child calls replace(x) where x is
 199         // the current child inode with the given name. If possible,
 200         // the child inode is replaced with the one returned by
 201         // replace().
 202         //
 203         // If replace(x) returns an inode (besides x or nil) that is
 204         // subsequently returned by Child(), then Child()'s caller
 205         // must ensure the new child's name and parent are set/updated
 206         // to Child()'s name argument and its receiver respectively.
 207         // This is not necessarily done before replace(x) returns, but
 208         // it must be done before Child()'s caller releases the
 209         // parent's lock.
 210         //
 211         // Nil represents "no child". replace(nil) signifies that no
 212         // child with this name exists yet. If replace() returns nil,
 213         // the existing child should be deleted if possible.
 214         //
 215         // An implementation of Child() is permitted to ignore
 216         // replace() or its return value. For example, a regular file
 217         // inode does not have children, so Child() always returns
 218         // nil.
 219         //
 220         // Child() returns the child, if any, with the given name: if
 221         // a child was added or changed, the new child is returned.
 222         //
 223         // Caller must have lock (or rlock if replace is nil).
 224         Child(name string, replace func(inode) (inode, error)) (inode, error)
 225
 226         sync.Locker
 227         RLock()
 228         RUnlock()
 229         MemorySize() int64
 230 }
 231
 232 type fileinfo struct {
 233         name    string
 234         mode    os.FileMode
 235         size    int64
 236         modTime time.Time
 237 }
 238
 239 // Name implements os.FileInfo.
 240 func (fi fileinfo) Name() string {
 241         return fi.name
 242 }
 243
 244 // ModTime implements os.FileInfo.
 245 func (fi fileinfo) ModTime() time.Time {
 246         return fi.modTime
 247 }
 248
 249 // Mode implements os.FileInfo.
 250 func (fi fileinfo) Mode() os.FileMode {
 251         return fi.mode
 252 }
 253
 254 // IsDir implements os.FileInfo.
 255 func (fi fileinfo) IsDir() bool {
 256         return fi.mode&os.ModeDir != 0
 257 }
 258
 259 // Size implements os.FileInfo.
 260 func (fi fileinfo) Size() int64 {
 261         return fi.size
 262 }
 263
 264 // Sys implements os.FileInfo.
 265 func (fi fileinfo) Sys() interface{} {
 266         return nil
 267 }
 268
 269 type nullnode struct{}
 270
 271 func (*nullnode) Mkdir(string, os.FileMode) error {
 272         return ErrInvalidOperation
 273 }
 274
 275 func (*nullnode) Read([]byte, filenodePtr) (int, filenodePtr, error) {
 276         return 0, filenodePtr{}, ErrInvalidOperation
 277 }
 278
 279 func (*nullnode) Write([]byte, filenodePtr) (int, filenodePtr, error) {
 280         return 0, filenodePtr{}, ErrInvalidOperation
 281 }
 282
 283 func (*nullnode) Truncate(int64) error {
 284         return ErrInvalidOperation
 285 }
 286
 287 func (*nullnode) FileInfo() os.FileInfo {
 288         return fileinfo{}
 289 }
 290
 291 func (*nullnode) IsDir() bool {
 292         return false
 293 }
 294
 295 func (*nullnode) Readdir() ([]os.FileInfo, error) {
 296         return nil, ErrInvalidOperation
 297 }
 298
 299 func (*nullnode) Child(name string, replace func(inode) (inode, error)) (inode, error) {
 300         return nil, ErrNotADirectory
 301 }
 302
 303 func (*nullnode) MemorySize() int64 {
 304         // Types that embed nullnode should report their own size, but
 305         // if they don't, we at least report a non-zero size to ensure
 306         // a large tree doesn't get reported as 0 bytes.
 307         return 64
 308 }
 309
 310 func (*nullnode) Snapshot() (inode, error) {
 311         return nil, ErrInvalidOperation
 312 }
 313
 314 func (*nullnode) Splice(inode) error {
 315         return ErrInvalidOperation
 316 }
 317
 318 type treenode struct {
 319         fs       FileSystem
 320         parent   inode
 321         inodes   map[string]inode
 322         fileinfo fileinfo
 323         sync.RWMutex
 324         nullnode
 325 }
 326
 327 func (n *treenode) FS() FileSystem {
 328         return n.fs
 329 }
 330
 331 func (n *treenode) SetParent(p inode, name string) {
 332         n.Lock()
 333         defer n.Unlock()
 334         n.parent = p
 335         n.fileinfo.name = name
 336 }
 337
 338 func (n *treenode) Parent() inode {
 339         n.RLock()
 340         defer n.RUnlock()
 341         return n.parent
 342 }
 343
 344 func (n *treenode) IsDir() bool {
 345         return true
 346 }
 347
 348 func (n *treenode) Child(name string, replace func(inode) (inode, error)) (child inode, err error) {
 349         debugPanicIfNotLocked(n, false)
 350         child = n.inodes[name]
 351         if name == "" || name == "." || name == ".." {
 352                 err = ErrInvalidArgument
 353                 return
 354         }
 355         if replace == nil {
 356                 return
 357         }
 358         newchild, err := replace(child)
 359         if err != nil {
 360                 return
 361         }
 362         if newchild == nil {
 363                 debugPanicIfNotLocked(n, true)
 364                 delete(n.inodes, name)
 365         } else if newchild != child {
 366                 debugPanicIfNotLocked(n, true)
 367                 n.inodes[name] = newchild
 368                 n.fileinfo.modTime = time.Now()
 369                 child = newchild
 370         }
 371         return
 372 }
 373
 374 func (n *treenode) Size() int64 {
 375         return n.FileInfo().Size()
 376 }
 377
 378 func (n *treenode) FileInfo() os.FileInfo {
 379         n.Lock()
 380         defer n.Unlock()
 381         n.fileinfo.size = int64(len(n.inodes))
 382         return n.fileinfo
 383 }
 384
 385 func (n *treenode) Readdir() (fi []os.FileInfo, err error) {
 386         n.RLock()
 387         defer n.RUnlock()
 388         fi = make([]os.FileInfo, 0, len(n.inodes))
 389         for _, inode := range n.inodes {
 390                 fi = append(fi, inode.FileInfo())
 391         }
 392         return
 393 }
 394
 395 func (n *treenode) Sync() error {
 396         n.RLock()
 397         defer n.RUnlock()
 398         for _, inode := range n.inodes {
 399                 syncer, ok := inode.(syncer)
 400                 if !ok {
 401                         return ErrInvalidOperation
 402                 }
 403                 err := syncer.Sync()
 404                 if err != nil {
 405                         return err
 406                 }
 407         }
 408         return nil
 409 }
 410
 411 func (n *treenode) MemorySize() (size int64) {
 412         // To avoid making other callers wait while we count the
 413         // entire filesystem size, we lock the node only long enough
 414         // to copy the list of children. We accept that the resulting
 415         // size will sometimes be misleading (e.g., we will
 416         // double-count an item that moves from A to B after we check
 417         // A's size but before we check B's size).
 418         n.RLock()
 419         debugPanicIfNotLocked(n, false)
 420         todo := make([]inode, 0, len(n.inodes))
 421         for _, inode := range n.inodes {
 422                 todo = append(todo, inode)
 423         }
 424         n.RUnlock()
 425         for _, inode := range todo {
 426                 size += inode.MemorySize()
 427         }
 428         return 64 + size
 429 }
 430
 431 type fileSystem struct {
 432         root inode
 433         fsBackend
 434         mutex sync.Mutex
 435         thr   *throttle
 436 }
 437
 438 func (fs *fileSystem) rootnode() inode {
 439         return fs.root
 440 }
 441
 442 func (fs *fileSystem) throttle() *throttle {
 443         return fs.thr
 444 }
 445
 446 func (fs *fileSystem) locker() sync.Locker {
 447         return &fs.mutex
 448 }
 449
 450 // OpenFile is analogous to os.OpenFile().
 451 func (fs *fileSystem) OpenFile(name string, flag int, perm os.FileMode) (File, error) {
 452         return fs.openFile(name, flag, perm)
 453 }
 454
 455 func (fs *fileSystem) openFile(name string, flag int, perm os.FileMode) (*filehandle, error) {
 456         if flag&os.O_SYNC != 0 {
 457                 return nil, ErrSyncNotSupported
 458         }
 459         dirname, name := path.Split(name)
 460         parent, err := rlookup(fs.root, dirname)
 461         if err != nil {
 462                 return nil, err
 463         }
 464         var readable, writable bool
 465         switch flag & (os.O_RDWR | os.O_RDONLY | os.O_WRONLY) {
 466         case os.O_RDWR:
 467                 readable = true
 468                 writable = true
 469         case os.O_RDONLY:
 470                 readable = true
 471         case os.O_WRONLY:
 472                 writable = true
 473         default:
 474                 return nil, fmt.Errorf("invalid flags 0x%x", flag)
 475         }
 476         if parent.IsDir() {
 477                 // A directory can be opened via "foo/", "foo/.", or
 478                 // "foo/..".
 479                 switch name {
 480                 case ".", "":
 481                         return &filehandle{inode: parent, readable: readable, writable: writable}, nil
 482                 case "..":
 483                         return &filehandle{inode: parent.Parent(), readable: readable, writable: writable}, nil
 484                 }
 485         }
 486         createMode := flag&os.O_CREATE != 0
 487         // We always need to take Lock() here, not just RLock(). Even
 488         // if we know we won't be creating a file, parent might be a
 489         // lookupnode, which sometimes populates its inodes map during
 490         // a Child() call.
 491         parent.Lock()
 492         defer parent.Unlock()
 493         n, err := parent.Child(name, nil)
 494         if err != nil {
 495                 return nil, err
 496         } else if n == nil {
 497                 if !createMode {
 498                         return nil, os.ErrNotExist
 499                 }
 500                 n, err = parent.Child(name, func(inode) (repl inode, err error) {
 501                         repl, err = parent.FS().newNode(name, perm|0755, time.Now())
 502                         if err != nil {
 503                                 return
 504                         }
 505                         repl.SetParent(parent, name)
 506                         return
 507                 })
 508                 if err != nil {
 509                         return nil, err
 510                 } else if n == nil {
 511                         // Parent rejected new child, but returned no error
 512                         return nil, ErrInvalidArgument
 513                 }
 514         } else if flag&os.O_EXCL != 0 {
 515                 return nil, ErrFileExists
 516         } else if flag&os.O_TRUNC != 0 {
 517                 if !writable {
 518                         return nil, fmt.Errorf("invalid flag O_TRUNC in read-only mode")
 519                 } else if n.IsDir() {
 520                         return nil, fmt.Errorf("invalid flag O_TRUNC when opening directory")
 521                 } else if err := n.Truncate(0); err != nil {
 522                         return nil, err
 523                 }
 524         }
 525         return &filehandle{
 526                 inode:    n,
 527                 append:   flag&os.O_APPEND != 0,
 528                 readable: readable,
 529                 writable: writable,
 530         }, nil
 531 }
 532
 533 func (fs *fileSystem) Open(name string) (http.File, error) {
 534         return fs.OpenFile(name, os.O_RDONLY, 0)
 535 }
 536
 537 func (fs *fileSystem) Create(name string) (File, error) {
 538         return fs.OpenFile(name, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0)
 539 }
 540
 541 func (fs *fileSystem) Mkdir(name string, perm os.FileMode) error {
 542         dirname, name := path.Split(name)
 543         n, err := rlookup(fs.root, dirname)
 544         if err != nil {
 545                 return err
 546         }
 547         n.Lock()
 548         defer n.Unlock()
 549         if child, err := n.Child(name, nil); err != nil {
 550                 return err
 551         } else if child != nil {
 552                 return os.ErrExist
 553         }
 554
 555         _, err = n.Child(name, func(inode) (repl inode, err error) {
 556                 repl, err = n.FS().newNode(name, perm|os.ModeDir, time.Now())
 557                 if err != nil {
 558                         return
 559                 }
 560                 repl.SetParent(n, name)
 561                 return
 562         })
 563         return err
 564 }
 565
 566 func (fs *fileSystem) Stat(name string) (os.FileInfo, error) {
 567         node, err := rlookup(fs.root, name)
 568         if err != nil {
 569                 return nil, err
 570         }
 571         return node.FileInfo(), nil
 572 }
 573
 574 func (fs *fileSystem) Rename(oldname, newname string) error {
 575         olddir, oldname := path.Split(oldname)
 576         if oldname == "" || oldname == "." || oldname == ".." {
 577                 return ErrInvalidArgument
 578         }
 579         olddirf, err := fs.openFile(olddir+".", os.O_RDONLY, 0)
 580         if err != nil {
 581                 return fmt.Errorf("%q: %s", olddir, err)
 582         }
 583         defer olddirf.Close()
 584
 585         newdir, newname := path.Split(newname)
 586         if newname == "." || newname == ".." {
 587                 return ErrInvalidArgument
 588         } else if newname == "" {
 589                 // Rename("a/b", "c/") means Rename("a/b", "c/b")
 590                 newname = oldname
 591         }
 592         newdirf, err := fs.openFile(newdir+".", os.O_RDONLY, 0)
 593         if err != nil {
 594                 return fmt.Errorf("%q: %s", newdir, err)
 595         }
 596         defer newdirf.Close()
 597
 598         // TODO: If the nearest common ancestor ("nca") of olddirf and
 599         // newdirf is on a different filesystem than fs, we should
 600         // call nca.FS().Rename() instead of proceeding. Until then
 601         // it's awkward for filesystems to implement their own Rename
 602         // methods effectively: the only one that runs is the one on
 603         // the root FileSystem exposed to the caller (webdav, fuse,
 604         // etc).
 605
 606         // When acquiring locks on multiple inodes, avoid deadlock by
 607         // locking the entire containing filesystem first.
 608         cfs := olddirf.inode.FS()
 609         cfs.locker().Lock()
 610         defer cfs.locker().Unlock()
 611
 612         if cfs != newdirf.inode.FS() {
 613                 // Moving inodes across filesystems is not (yet)
 614                 // supported. Locking inodes from different
 615                 // filesystems could deadlock, so we must error out
 616                 // now.
 617                 return ErrInvalidOperation
 618         }
 619
 620         // To ensure we can test reliably whether we're about to move
 621         // a directory into itself, lock all potential common
 622         // ancestors of olddir and newdir.
 623         needLock := []sync.Locker{}
 624         for _, node := range []inode{olddirf.inode, newdirf.inode} {
 625                 needLock = append(needLock, node)
 626                 for node.Parent() != node && node.Parent().FS() == node.FS() {
 627                         node = node.Parent()
 628                         needLock = append(needLock, node)
 629                 }
 630         }
 631         locked := map[sync.Locker]bool{}
 632         for i := len(needLock) - 1; i >= 0; i-- {
 633                 if n := needLock[i]; !locked[n] {
 634                         n.Lock()
 635                         defer n.Unlock()
 636                         locked[n] = true
 637                 }
 638         }
 639
 640         _, err = olddirf.inode.Child(oldname, func(oldinode inode) (inode, error) {
 641                 if oldinode == nil {
 642                         return oldinode, os.ErrNotExist
 643                 }
 644                 if locked[oldinode] {
 645                         // oldinode cannot become a descendant of itself.
 646                         return oldinode, ErrInvalidArgument
 647                 }
 648                 if oldinode.FS() != cfs && newdirf.inode != olddirf.inode {
 649                         // moving a mount point to a different parent
 650                         // is not (yet) supported.
 651                         return oldinode, ErrInvalidArgument
 652                 }
 653                 accepted, err := newdirf.inode.Child(newname, func(existing inode) (inode, error) {
 654                         if existing != nil && existing.IsDir() {
 655                                 return existing, ErrIsDirectory
 656                         }
 657                         return oldinode, nil
 658                 })
 659                 if err != nil {
 660                         // Leave oldinode in olddir.
 661                         return oldinode, err
 662                 }
 663                 accepted.SetParent(newdirf.inode, newname)
 664                 return nil, nil
 665         })
 666         return err
 667 }
 668
 669 func (fs *fileSystem) Remove(name string) error {
 670         return fs.remove(strings.TrimRight(name, "/"), false)
 671 }
 672
 673 func (fs *fileSystem) RemoveAll(name string) error {
 674         err := fs.remove(strings.TrimRight(name, "/"), true)
 675         if os.IsNotExist(err) {
 676                 // "If the path does not exist, RemoveAll returns
 677                 // nil." (see "os" pkg)
 678                 err = nil
 679         }
 680         return err
 681 }
 682
 683 func (fs *fileSystem) remove(name string, recursive bool) error {
 684         dirname, name := path.Split(name)
 685         if name == "" || name == "." || name == ".." {
 686                 return ErrInvalidArgument
 687         }
 688         dir, err := rlookup(fs.root, dirname)
 689         if err != nil {
 690                 return err
 691         }
 692         dir.Lock()
 693         defer dir.Unlock()
 694         _, err = dir.Child(name, func(node inode) (inode, error) {
 695                 if node == nil {
 696                         return nil, os.ErrNotExist
 697                 }
 698                 if !recursive && node.IsDir() && node.Size() > 0 {
 699                         return node, ErrDirectoryNotEmpty
 700                 }
 701                 return nil, nil
 702         })
 703         return err
 704 }
 705
 706 func (fs *fileSystem) Sync() error {
 707         if syncer, ok := fs.root.(syncer); ok {
 708                 return syncer.Sync()
 709         }
 710         return ErrInvalidOperation
 711 }
 712
 713 func (fs *fileSystem) Flush(string, bool) error {
 714         log.Printf("TODO: flush fileSystem")
 715         return ErrInvalidOperation
 716 }
 717
 718 func (fs *fileSystem) MemorySize() int64 {
 719         return fs.root.MemorySize()
 720 }
 721
 722 // rlookup (recursive lookup) returns the inode for the file/directory
 723 // with the given name (which may contain "/" separators). If no such
 724 // file/directory exists, the returned node is nil.
 725 func rlookup(start inode, path string) (node inode, err error) {
 726         node = start
 727         for _, name := range strings.Split(path, "/") {
 728                 if node.IsDir() {
 729                         if name == "." || name == "" {
 730                                 continue
 731                         }
 732                         if name == ".." {
 733                                 node = node.Parent()
 734                                 continue
 735                         }
 736                 }
 737                 node, err = func() (inode, error) {
 738                         node.Lock()
 739                         defer node.Unlock()
 740                         return node.Child(name, nil)
 741                 }()
 742                 if node == nil || err != nil {
 743                         break
 744                 }
 745         }
 746         if node == nil && err == nil {
 747                 err = os.ErrNotExist
 748         }
 749         return
 750 }
 751
 752 func permittedName(name string) bool {
 753         return name != "" && name != "." && name != ".." && !strings.Contains(name, "/")
 754 }
 755
 756 // Snapshot returns a Subtree that's a copy of the given path. It
 757 // returns an error if the path is not inside a collection.
 758 func Snapshot(fs FileSystem, path string) (*Subtree, error) {
 759         f, err := fs.OpenFile(path, os.O_RDONLY, 0)
 760         if err != nil {
 761                 return nil, err
 762         }
 763         defer f.Close()
 764         return f.Snapshot()
 765 }
 766
 767 // Splice inserts newsubtree at the indicated target path.
 768 //
 769 // Splice returns an error if target is not inside a collection.
 770 //
 771 // Splice returns an error if target is the root of a collection and
 772 // newsubtree is a snapshot of a file.
 773 func Splice(fs FileSystem, target string, newsubtree *Subtree) error {
 774         f, err := fs.OpenFile(target, os.O_WRONLY, 0)
 775         if os.IsNotExist(err) {
 776                 f, err = fs.OpenFile(target, os.O_CREATE|os.O_WRONLY, 0700)
 777         }
 778         if err != nil {
 779                 return fmt.Errorf("open %s: %w", target, err)
 780         }
 781         defer f.Close()
 782         return f.Splice(newsubtree)
 783 }