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         // Source data structure: *Collection, *Group, or
 238         // nil. Currently populated only for project dirs and
 239         // top-level collection dirs; *not* populated for
 240         // /by_id/{uuid} dirs (only subdirs below that). Does not stay
 241         // up to date with upstream changes.
 242         //
 243         // Intended to support keep-web's properties-as-s3-metadata
 244         // feature (https://dev.arvados.org/issues/19088).
 245         sys interface{}
 246 }
 247
 248 // Name implements os.FileInfo.
 249 func (fi fileinfo) Name() string {
 250         return fi.name
 251 }
 252
 253 // ModTime implements os.FileInfo.
 254 func (fi fileinfo) ModTime() time.Time {
 255         return fi.modTime
 256 }
 257
 258 // Mode implements os.FileInfo.
 259 func (fi fileinfo) Mode() os.FileMode {
 260         return fi.mode
 261 }
 262
 263 // IsDir implements os.FileInfo.
 264 func (fi fileinfo) IsDir() bool {
 265         return fi.mode&os.ModeDir != 0
 266 }
 267
 268 // Size implements os.FileInfo.
 269 func (fi fileinfo) Size() int64 {
 270         return fi.size
 271 }
 272
 273 // Sys implements os.FileInfo. See comment in fileinfo struct.
 274 func (fi fileinfo) Sys() interface{} {
 275         return fi.sys
 276 }
 277
 278 type nullnode struct{}
 279
 280 func (*nullnode) Mkdir(string, os.FileMode) error {
 281         return ErrInvalidOperation
 282 }
 283
 284 func (*nullnode) Read([]byte, filenodePtr) (int, filenodePtr, error) {
 285         return 0, filenodePtr{}, ErrInvalidOperation
 286 }
 287
 288 func (*nullnode) Write([]byte, filenodePtr) (int, filenodePtr, error) {
 289         return 0, filenodePtr{}, ErrInvalidOperation
 290 }
 291
 292 func (*nullnode) Truncate(int64) error {
 293         return ErrInvalidOperation
 294 }
 295
 296 func (*nullnode) FileInfo() os.FileInfo {
 297         return fileinfo{}
 298 }
 299
 300 func (*nullnode) IsDir() bool {
 301         return false
 302 }
 303
 304 func (*nullnode) Readdir() ([]os.FileInfo, error) {
 305         return nil, ErrInvalidOperation
 306 }
 307
 308 func (*nullnode) Child(name string, replace func(inode) (inode, error)) (inode, error) {
 309         return nil, ErrNotADirectory
 310 }
 311
 312 func (*nullnode) MemorySize() int64 {
 313         // Types that embed nullnode should report their own size, but
 314         // if they don't, we at least report a non-zero size to ensure
 315         // a large tree doesn't get reported as 0 bytes.
 316         return 64
 317 }
 318
 319 func (*nullnode) Snapshot() (inode, error) {
 320         return nil, ErrInvalidOperation
 321 }
 322
 323 func (*nullnode) Splice(inode) error {
 324         return ErrInvalidOperation
 325 }
 326
 327 type treenode struct {
 328         fs       FileSystem
 329         parent   inode
 330         inodes   map[string]inode
 331         fileinfo fileinfo
 332         sync.RWMutex
 333         nullnode
 334 }
 335
 336 func (n *treenode) FS() FileSystem {
 337         return n.fs
 338 }
 339
 340 func (n *treenode) SetParent(p inode, name string) {
 341         n.Lock()
 342         defer n.Unlock()
 343         n.parent = p
 344         n.fileinfo.name = name
 345 }
 346
 347 func (n *treenode) Parent() inode {
 348         n.RLock()
 349         defer n.RUnlock()
 350         return n.parent
 351 }
 352
 353 func (n *treenode) IsDir() bool {
 354         return true
 355 }
 356
 357 func (n *treenode) Child(name string, replace func(inode) (inode, error)) (child inode, err error) {
 358         debugPanicIfNotLocked(n, false)
 359         child = n.inodes[name]
 360         if name == "" || name == "." || name == ".." {
 361                 err = ErrInvalidArgument
 362                 return
 363         }
 364         if replace == nil {
 365                 return
 366         }
 367         newchild, err := replace(child)
 368         if err != nil {
 369                 return
 370         }
 371         if newchild == nil {
 372                 debugPanicIfNotLocked(n, true)
 373                 delete(n.inodes, name)
 374         } else if newchild != child {
 375                 debugPanicIfNotLocked(n, true)
 376                 n.inodes[name] = newchild
 377                 n.fileinfo.modTime = time.Now()
 378                 child = newchild
 379         }
 380         return
 381 }
 382
 383 func (n *treenode) Size() int64 {
 384         return n.FileInfo().Size()
 385 }
 386
 387 func (n *treenode) FileInfo() os.FileInfo {
 388         n.Lock()
 389         defer n.Unlock()
 390         n.fileinfo.size = int64(len(n.inodes))
 391         return n.fileinfo
 392 }
 393
 394 func (n *treenode) Readdir() (fi []os.FileInfo, err error) {
 395         n.RLock()
 396         defer n.RUnlock()
 397         fi = make([]os.FileInfo, 0, len(n.inodes))
 398         for _, inode := range n.inodes {
 399                 fi = append(fi, inode.FileInfo())
 400         }
 401         return
 402 }
 403
 404 func (n *treenode) Sync() error {
 405         n.RLock()
 406         defer n.RUnlock()
 407         for _, inode := range n.inodes {
 408                 syncer, ok := inode.(syncer)
 409                 if !ok {
 410                         return ErrInvalidOperation
 411                 }
 412                 err := syncer.Sync()
 413                 if err != nil {
 414                         return err
 415                 }
 416         }
 417         return nil
 418 }
 419
 420 func (n *treenode) MemorySize() (size int64) {
 421         n.RLock()
 422         defer n.RUnlock()
 423         debugPanicIfNotLocked(n, false)
 424         for _, inode := range n.inodes {
 425                 size += inode.MemorySize()
 426         }
 427         return 64 + size
 428 }
 429
 430 type fileSystem struct {
 431         root inode
 432         fsBackend
 433         mutex sync.Mutex
 434         thr   *throttle
 435 }
 436
 437 func (fs *fileSystem) rootnode() inode {
 438         return fs.root
 439 }
 440
 441 func (fs *fileSystem) throttle() *throttle {
 442         return fs.thr
 443 }
 444
 445 func (fs *fileSystem) locker() sync.Locker {
 446         return &fs.mutex
 447 }
 448
 449 // OpenFile is analogous to os.OpenFile().
 450 func (fs *fileSystem) OpenFile(name string, flag int, perm os.FileMode) (File, error) {
 451         return fs.openFile(name, flag, perm)
 452 }
 453
 454 func (fs *fileSystem) openFile(name string, flag int, perm os.FileMode) (*filehandle, error) {
 455         if flag&os.O_SYNC != 0 {
 456                 return nil, ErrSyncNotSupported
 457         }
 458         dirname, name := path.Split(name)
 459         parent, err := rlookup(fs.root, dirname)
 460         if err != nil {
 461                 return nil, err
 462         }
 463         var readable, writable bool
 464         switch flag & (os.O_RDWR | os.O_RDONLY | os.O_WRONLY) {
 465         case os.O_RDWR:
 466                 readable = true
 467                 writable = true
 468         case os.O_RDONLY:
 469                 readable = true
 470         case os.O_WRONLY:
 471                 writable = true
 472         default:
 473                 return nil, fmt.Errorf("invalid flags 0x%x", flag)
 474         }
 475         if parent.IsDir() {
 476                 // A directory can be opened via "foo/", "foo/.", or
 477                 // "foo/..".
 478                 switch name {
 479                 case ".", "":
 480                         return &filehandle{inode: parent, readable: readable, writable: writable}, nil
 481                 case "..":
 482                         return &filehandle{inode: parent.Parent(), readable: readable, writable: writable}, nil
 483                 }
 484         }
 485         createMode := flag&os.O_CREATE != 0
 486         // We always need to take Lock() here, not just RLock(). Even
 487         // if we know we won't be creating a file, parent might be a
 488         // lookupnode, which sometimes populates its inodes map during
 489         // a Child() call.
 490         parent.Lock()
 491         defer parent.Unlock()
 492         n, err := parent.Child(name, nil)
 493         if err != nil {
 494                 return nil, err
 495         } else if n == nil {
 496                 if !createMode {
 497                         return nil, os.ErrNotExist
 498                 }
 499                 n, err = parent.Child(name, func(inode) (repl inode, err error) {
 500                         repl, err = parent.FS().newNode(name, perm|0755, time.Now())
 501                         if err != nil {
 502                                 return
 503                         }
 504                         repl.SetParent(parent, name)
 505                         return
 506                 })
 507                 if err != nil {
 508                         return nil, err
 509                 } else if n == nil {
 510                         // Parent rejected new child, but returned no error
 511                         return nil, ErrInvalidArgument
 512                 }
 513         } else if flag&os.O_EXCL != 0 {
 514                 return nil, ErrFileExists
 515         } else if flag&os.O_TRUNC != 0 {
 516                 if !writable {
 517                         return nil, fmt.Errorf("invalid flag O_TRUNC in read-only mode")
 518                 } else if n.IsDir() {
 519                         return nil, fmt.Errorf("invalid flag O_TRUNC when opening directory")
 520                 } else if err := n.Truncate(0); err != nil {
 521                         return nil, err
 522                 }
 523         }
 524         return &filehandle{
 525                 inode:    n,
 526                 append:   flag&os.O_APPEND != 0,
 527                 readable: readable,
 528                 writable: writable,
 529         }, nil
 530 }
 531
 532 func (fs *fileSystem) Open(name string) (http.File, error) {
 533         return fs.OpenFile(name, os.O_RDONLY, 0)
 534 }
 535
 536 func (fs *fileSystem) Create(name string) (File, error) {
 537         return fs.OpenFile(name, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0)
 538 }
 539
 540 func (fs *fileSystem) Mkdir(name string, perm os.FileMode) error {
 541         dirname, name := path.Split(name)
 542         n, err := rlookup(fs.root, dirname)
 543         if err != nil {
 544                 return err
 545         }
 546         n.Lock()
 547         defer n.Unlock()
 548         if child, err := n.Child(name, nil); err != nil {
 549                 return err
 550         } else if child != nil {
 551                 return os.ErrExist
 552         }
 553
 554         _, err = n.Child(name, func(inode) (repl inode, err error) {
 555                 repl, err = n.FS().newNode(name, perm|os.ModeDir, time.Now())
 556                 if err != nil {
 557                         return
 558                 }
 559                 repl.SetParent(n, name)
 560                 return
 561         })
 562         return err
 563 }
 564
 565 func (fs *fileSystem) Stat(name string) (os.FileInfo, error) {
 566         node, err := rlookup(fs.root, name)
 567         if err != nil {
 568                 return nil, err
 569         }
 570         return node.FileInfo(), nil
 571 }
 572
 573 func (fs *fileSystem) Rename(oldname, newname string) error {
 574         olddir, oldname := path.Split(oldname)
 575         if oldname == "" || oldname == "." || oldname == ".." {
 576                 return ErrInvalidArgument
 577         }
 578         olddirf, err := fs.openFile(olddir+".", os.O_RDONLY, 0)
 579         if err != nil {
 580                 return fmt.Errorf("%q: %s", olddir, err)
 581         }
 582         defer olddirf.Close()
 583
 584         newdir, newname := path.Split(newname)
 585         if newname == "." || newname == ".." {
 586                 return ErrInvalidArgument
 587         } else if newname == "" {
 588                 // Rename("a/b", "c/") means Rename("a/b", "c/b")
 589                 newname = oldname
 590         }
 591         newdirf, err := fs.openFile(newdir+".", os.O_RDONLY, 0)
 592         if err != nil {
 593                 return fmt.Errorf("%q: %s", newdir, err)
 594         }
 595         defer newdirf.Close()
 596
 597         // TODO: If the nearest common ancestor ("nca") of olddirf and
 598         // newdirf is on a different filesystem than fs, we should
 599         // call nca.FS().Rename() instead of proceeding. Until then
 600         // it's awkward for filesystems to implement their own Rename
 601         // methods effectively: the only one that runs is the one on
 602         // the root FileSystem exposed to the caller (webdav, fuse,
 603         // etc).
 604
 605         // When acquiring locks on multiple inodes, avoid deadlock by
 606         // locking the entire containing filesystem first.
 607         cfs := olddirf.inode.FS()
 608         cfs.locker().Lock()
 609         defer cfs.locker().Unlock()
 610
 611         if cfs != newdirf.inode.FS() {
 612                 // Moving inodes across filesystems is not (yet)
 613                 // supported. Locking inodes from different
 614                 // filesystems could deadlock, so we must error out
 615                 // now.
 616                 return ErrInvalidOperation
 617         }
 618
 619         // To ensure we can test reliably whether we're about to move
 620         // a directory into itself, lock all potential common
 621         // ancestors of olddir and newdir.
 622         needLock := []sync.Locker{}
 623         for _, node := range []inode{olddirf.inode, newdirf.inode} {
 624                 needLock = append(needLock, node)
 625                 for node.Parent() != node && node.Parent().FS() == node.FS() {
 626                         node = node.Parent()
 627                         needLock = append(needLock, node)
 628                 }
 629         }
 630         locked := map[sync.Locker]bool{}
 631         for i := len(needLock) - 1; i >= 0; i-- {
 632                 if n := needLock[i]; !locked[n] {
 633                         n.Lock()
 634                         defer n.Unlock()
 635                         locked[n] = true
 636                 }
 637         }
 638
 639         _, err = olddirf.inode.Child(oldname, func(oldinode inode) (inode, error) {
 640                 if oldinode == nil {
 641                         return oldinode, os.ErrNotExist
 642                 }
 643                 if locked[oldinode] {
 644                         // oldinode cannot become a descendant of itself.
 645                         return oldinode, ErrInvalidArgument
 646                 }
 647                 if oldinode.FS() != cfs && newdirf.inode != olddirf.inode {
 648                         // moving a mount point to a different parent
 649                         // is not (yet) supported.
 650                         return oldinode, ErrInvalidArgument
 651                 }
 652                 accepted, err := newdirf.inode.Child(newname, func(existing inode) (inode, error) {
 653                         if existing != nil && existing.IsDir() {
 654                                 return existing, ErrIsDirectory
 655                         }
 656                         return oldinode, nil
 657                 })
 658                 if err != nil {
 659                         // Leave oldinode in olddir.
 660                         return oldinode, err
 661                 }
 662                 accepted.SetParent(newdirf.inode, newname)
 663                 return nil, nil
 664         })
 665         return err
 666 }
 667
 668 func (fs *fileSystem) Remove(name string) error {
 669         return fs.remove(strings.TrimRight(name, "/"), false)
 670 }
 671
 672 func (fs *fileSystem) RemoveAll(name string) error {
 673         err := fs.remove(strings.TrimRight(name, "/"), true)
 674         if os.IsNotExist(err) {
 675                 // "If the path does not exist, RemoveAll returns
 676                 // nil." (see "os" pkg)
 677                 err = nil
 678         }
 679         return err
 680 }
 681
 682 func (fs *fileSystem) remove(name string, recursive bool) error {
 683         dirname, name := path.Split(name)
 684         if name == "" || name == "." || name == ".." {
 685                 return ErrInvalidArgument
 686         }
 687         dir, err := rlookup(fs.root, dirname)
 688         if err != nil {
 689                 return err
 690         }
 691         dir.Lock()
 692         defer dir.Unlock()
 693         _, err = dir.Child(name, func(node inode) (inode, error) {
 694                 if node == nil {
 695                         return nil, os.ErrNotExist
 696                 }
 697                 if !recursive && node.IsDir() && node.Size() > 0 {
 698                         return node, ErrDirectoryNotEmpty
 699                 }
 700                 return nil, nil
 701         })
 702         return err
 703 }
 704
 705 func (fs *fileSystem) Sync() error {
 706         if syncer, ok := fs.root.(syncer); ok {
 707                 return syncer.Sync()
 708         }
 709         return ErrInvalidOperation
 710 }
 711
 712 func (fs *fileSystem) Flush(string, bool) error {
 713         log.Printf("TODO: flush fileSystem")
 714         return ErrInvalidOperation
 715 }
 716
 717 func (fs *fileSystem) MemorySize() int64 {
 718         return fs.root.MemorySize()
 719 }
 720
 721 // rlookup (recursive lookup) returns the inode for the file/directory
 722 // with the given name (which may contain "/" separators). If no such
 723 // file/directory exists, the returned node is nil.
 724 func rlookup(start inode, path string) (node inode, err error) {
 725         node = start
 726         for _, name := range strings.Split(path, "/") {
 727                 if node.IsDir() {
 728                         if name == "." || name == "" {
 729                                 continue
 730                         }
 731                         if name == ".." {
 732                                 node = node.Parent()
 733                                 continue
 734                         }
 735                 }
 736                 node, err = func() (inode, error) {
 737                         node.Lock()
 738                         defer node.Unlock()
 739                         return node.Child(name, nil)
 740                 }()
 741                 if node == nil || err != nil {
 742                         break
 743                 }
 744         }
 745         if node == nil && err == nil {
 746                 err = os.ErrNotExist
 747         }
 748         return
 749 }
 750
 751 func permittedName(name string) bool {
 752         return name != "" && name != "." && name != ".." && !strings.Contains(name, "/")
 753 }
 754
 755 // Snapshot returns a Subtree that's a copy of the given path. It
 756 // returns an error if the path is not inside a collection.
 757 func Snapshot(fs FileSystem, path string) (*Subtree, error) {
 758         f, err := fs.OpenFile(path, os.O_RDONLY, 0)
 759         if err != nil {
 760                 return nil, err
 761         }
 762         defer f.Close()
 763         return f.Snapshot()
 764 }
 765
 766 // Splice inserts newsubtree at the indicated target path.
 767 //
 768 // Splice returns an error if target is not inside a collection.
 769 //
 770 // Splice returns an error if target is the root of a collection and
 771 // newsubtree is a snapshot of a file.
 772 func Splice(fs FileSystem, target string, newsubtree *Subtree) error {
 773         f, err := fs.OpenFile(target, os.O_WRONLY, 0)
 774         if os.IsNotExist(err) {
 775                 f, err = fs.OpenFile(target, os.O_CREATE|os.O_WRONLY, 0700)
 776         }
 777         if err != nil {
 778                 return fmt.Errorf("open %s: %w", target, err)
 779         }
 780         defer f.Close()
 781         return f.Splice(newsubtree)
 782 }