15652: Add collectionfs Flush() method to control memory use.

[arvados.git] / services / crunch-run / copier.go

diff --git a/services/crunch-run/copier.go b/services/crunch-run/copier.go
index b6093ce092933fbbf52dd39903540a28515471ba..555a2654d8c1513d79cd99b0b493acc42ffc9319 100644 (file)
--- a/services/crunch-run/copier.go
+++ b/services/crunch-run/copier.go
@@ -54,6 +54,7 @@ type copier struct {
        keepClient    IKeepClient
        hostOutputDir string
        ctrOutputDir  string
+       binds         []string
        mounts        map[string]arvados.Mount
        secretMounts  map[string]arvados.Mount
        logger        printfer
@@ -69,22 +70,33 @@ type copier struct {
 func (cp *copier) Copy() (string, error) {
        err := cp.walkMount("", cp.ctrOutputDir, limitFollowSymlinks, true)
        if err != nil {
-               return "", err
+               return "", fmt.Errorf("error scanning files to copy to output: %v", err)
        }
        fs, err := (&arvados.Collection{ManifestText: cp.manifest}).FileSystem(cp.client, cp.keepClient)
        if err != nil {
-               return "", err
+               return "", fmt.Errorf("error creating Collection.FileSystem: %v", err)
        }
        for _, d := range cp.dirs {
                err = fs.Mkdir(d, 0777)
-               if err != nil {
-                       return "", err
+               if err != nil && err != os.ErrExist {
+                       return "", fmt.Errorf("error making directory %q in output collection: %v", d, err)
                }
        }
+       var lastparentdir string
        for _, f := range cp.files {
+               // If a dir has just had its last file added, do a
+               // full Flush. Otherwise, do a partial Flush (write
+               // full-size blocks, but leave the last short block
+               // open so f's data can be packed with it).
+               dir, _ := filepath.Split(f.dst)
+               if err := fs.Flush(dir != lastparentdir); err != nil {
+                       return "", fmt.Errorf("error flushing output collection file data: %v", err)
+               }
+               lastparentdir = dir
+
                err = cp.copyFile(fs, f)
                if err != nil {
-                       return "", err
+                       return "", fmt.Errorf("error copying file %q into output collection: %v", f, err)
                }
        }
        return fs.MarshalManifest(".")
@@ -157,8 +169,12 @@ func (cp *copier) walkMount(dest, src string, maxSymlinks int, walkMountsBelow b
                        return err
                }
                cp.manifest += mft.Extract(srcRelPath, dest).Text
-       case srcRoot == cp.ctrOutputDir:
-               f, err := os.Open(filepath.Join(cp.hostOutputDir, ".arvados#collection"))
+       default:
+               hostRoot, err := cp.hostRoot(srcRoot)
+               if err != nil {
+                       return err
+               }
+               f, err := os.Open(filepath.Join(hostRoot, ".arvados#collection"))
                if err != nil {
                        return err
                }
@@ -170,8 +186,6 @@ func (cp *copier) walkMount(dest, src string, maxSymlinks int, walkMountsBelow b
                }
                mft := manifest.Manifest{Text: coll.ManifestText}
                cp.manifest += mft.Extract(srcRelPath, dest).Text
-       default:
-               return fmt.Errorf("cannot output %q as %q: writable collection mounted at %q", src, dest, srcRoot)
        }
        if walkMountsBelow {
                return cp.walkMountsBelow(dest, src)
@@ -185,7 +199,7 @@ func (cp *copier) walkMountsBelow(dest, src string) error {
                if !strings.HasPrefix(mnt, src+"/") {
                        continue
                }
-               if mntinfo.Kind == "text" || mntinfo.Kind == "json" {
+               if cp.copyRegularFiles(mntinfo) {
                        // These got copied into the nearest parent
                        // mount as regular files during setup, so
                        // they get copied as regular files when we
@@ -284,15 +298,15 @@ func (cp *copier) walkHostFS(dest, src string, maxSymlinks int, includeMounts bo
                        if _, isSecret := cp.secretMounts[src]; isSecret {
                                continue
                        }
-                       if mntinfo, isMount := cp.mounts[src]; isMount && mntinfo.Kind != "text" && mntinfo.Kind != "json" {
+                       if mntinfo, isMount := cp.mounts[src]; isMount && !cp.copyRegularFiles(mntinfo) {
                                // If a regular file/dir somehow
                                // exists at a path that's also a
                                // mount target, ignore the file --
                                // the mount has already been included
                                // with walkMountsBelow().
                                //
-                               // (...except json/text mounts, which
-                               // are handled as regular files.)
+                               // (...except mount types that are
+                               // handled as regular files.)
                                continue
                        }
                        err = cp.walkHostFS(dest, src, maxSymlinks, false)
@@ -316,6 +330,25 @@ func (cp *copier) walkHostFS(dest, src string, maxSymlinks int, includeMounts bo
        return fmt.Errorf("Unsupported file type (mode %o) in output dir: %q", fi.Mode(), src)
 }
 
+// Return the host path that was mounted at the given path in the
+// container.
+func (cp *copier) hostRoot(ctrRoot string) (string, error) {
+       if ctrRoot == cp.ctrOutputDir {
+               return cp.hostOutputDir, nil
+       }
+       for _, bind := range cp.binds {
+               tokens := strings.Split(bind, ":")
+               if len(tokens) >= 2 && tokens[1] == ctrRoot {
+                       return tokens[0], nil
+               }
+       }
+       return "", fmt.Errorf("not bind-mounted: %q", ctrRoot)
+}
+
+func (cp *copier) copyRegularFiles(m arvados.Mount) bool {
+       return m.Kind == "text" || m.Kind == "json" || (m.Kind == "collection" && m.Writable)
+}
+
 func (cp *copier) getManifest(pdh string) (*manifest.Manifest, error) {
        if mft, ok := cp.manifestCache[pdh]; ok {
                return mft, nil