X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/8d73164d5ca50e7af1b3752b4251eaf9f11a9fc8..13c69930bbec6f691100e4374ba272905fd83f3e:/sdk/go/arvados/fs_collection.go diff --git a/sdk/go/arvados/fs_collection.go b/sdk/go/arvados/fs_collection.go index f6afadba5b..b996542abd 100644 --- a/sdk/go/arvados/fs_collection.go +++ b/sdk/go/arvados/fs_collection.go @@ -8,7 +8,6 @@ import ( "encoding/json" "fmt" "io" - "log" "os" "path" "regexp" @@ -116,14 +115,12 @@ func (fs *collectionFileSystem) newNode(name string, perm os.FileMode, modTime t } func (fs *collectionFileSystem) Sync() error { - log.Printf("cfs.Sync()") if fs.uuid == "" { return nil } txt, err := fs.MarshalManifest(".") if err != nil { - log.Printf("WARNING: (collectionFileSystem)Sync() failed: %s", err) - return err + return fmt.Errorf("sync failed: %s", err) } coll := &Collection{ UUID: fs.uuid, @@ -131,9 +128,9 @@ func (fs *collectionFileSystem) Sync() error { } err = fs.RequestAndDecode(nil, "PUT", "arvados/v1/collections/"+fs.uuid, fs.UpdateBody(coll), map[string]interface{}{"select": []string{"uuid"}}) if err != nil { - log.Printf("WARNING: (collectionFileSystem)Sync() failed: %s", err) + return fmt.Errorf("sync failed: update %s: %s", fs.uuid, err) } - return err + return nil } func (fs *collectionFileSystem) MarshalManifest(prefix string) (string, error) { @@ -549,9 +546,10 @@ func (dn *dirnode) Child(name string, replace func(inode) (inode, error)) (inode return dn.treenode.Child(name, replace) } -// sync flushes in-memory data (for the children with the given names, -// which must be children of dn) to persistent storage. Caller must -// have write lock on dn and the named children. +// sync flushes in-memory data and remote block references (for the +// children with the given names, which must be children of dn) to +// local persistent storage. Caller must have write lock on dn and the +// named children. func (dn *dirnode) sync(names []string) error { type shortBlock struct { fn *filenode @@ -588,37 +586,51 @@ func (dn *dirnode) sync(names []string) error { return nil } + localLocator := map[string]string{} for _, name := range names { fn, ok := dn.inodes[name].(*filenode) if !ok { continue } for idx, seg := range fn.segments { - seg, ok := seg.(*memSegment) - if !ok { - continue - } - if seg.Len() > maxBlockSize/2 { - if err := flush([]shortBlock{{fn, idx}}); err != nil { - return err + switch seg := seg.(type) { + case storedSegment: + loc, ok := localLocator[seg.locator] + if !ok { + var err error + loc, err = dn.fs.LocalLocator(seg.locator) + if err != nil { + return err + } + localLocator[seg.locator] = loc } - continue - } - if pendingLen+seg.Len() > maxBlockSize { - if err := flush(pending); err != nil { - return err + seg.locator = loc + fn.segments[idx] = seg + case *memSegment: + if seg.Len() > maxBlockSize/2 { + if err := flush([]shortBlock{{fn, idx}}); err != nil { + return err + } + continue + } + if pendingLen+seg.Len() > maxBlockSize { + if err := flush(pending); err != nil { + return err + } + pending = nil + pendingLen = 0 } - pending = nil - pendingLen = 0 + pending = append(pending, shortBlock{fn, idx}) + pendingLen += seg.Len() + default: + panic(fmt.Sprintf("can't sync segment type %T", seg)) } - pending = append(pending, shortBlock{fn, idx}) - pendingLen += seg.Len() } } return flush(pending) } -// caller must have read lock. +// caller must have write lock. func (dn *dirnode) marshalManifest(prefix string) (string, error) { var streamLen int64 type filepart struct { @@ -630,6 +642,17 @@ func (dn *dirnode) marshalManifest(prefix string) (string, error) { var subdirs string var blocks []string + if len(dn.inodes) == 0 { + if prefix == "." { + return "", nil + } + // Express the existence of an empty directory by + // adding an empty file named `\056`, which (unlike + // the more obvious spelling `.`) is accepted by the + // API's manifest validator. + return manifestEscape(prefix) + " d41d8cd98f00b204e9800998ecf8427e+0 0:0:\\056\n", nil + } + names := make([]string, 0, len(dn.inodes)) for name := range dn.inodes { names = append(names, name) @@ -758,8 +781,14 @@ func (dn *dirnode) loadManifest(txt string) error { } name := dirname + "/" + manifestUnescape(toks[2]) fnode, err := dn.createFileAndParents(name) - if err != nil { - return fmt.Errorf("line %d: cannot use path %q: %s", lineno, name, err) + if fnode == nil && err == nil && length == 0 { + // Special case: an empty file used as + // a marker to preserve an otherwise + // empty directory in a manifest. + continue + } + if err != nil || (fnode == nil && length != 0) { + return fmt.Errorf("line %d: cannot use path %q with length %d: %s", lineno, name, length, err) } // Map the stream offset/range coordinates to // block/offset/range coordinates and add @@ -817,15 +846,14 @@ func (dn *dirnode) loadManifest(txt string) error { return nil } -// only safe to call from loadManifest -- no locking +// only safe to call from loadManifest -- no locking. +// +// If path is a "parent directory exists" marker (the last path +// component is "."), the returned values are both nil. func (dn *dirnode) createFileAndParents(path string) (fn *filenode, err error) { var node inode = dn names := strings.Split(path, "/") basename := names[len(names)-1] - if !permittedName(basename) { - err = fmt.Errorf("invalid file part %q in path %q", basename, path) - return - } for _, name := range names[:len(names)-1] { switch name { case "", ".": @@ -856,6 +884,12 @@ func (dn *dirnode) createFileAndParents(path string) (fn *filenode, err error) { return } } + if basename == "." { + return + } else if !permittedName(basename) { + err = fmt.Errorf("invalid file part %q in path %q", basename, path) + return + } _, err = node.Child(basename, func(child inode) (inode, error) { switch child := child.(type) { case nil: