environment views: single pass view generation (#29443)

Reduces the number of stat calls to a bare minimum:
- Single pass over src prefixes
- Handle projection clashes in memory

Symlinked directories in the src prefixes are now conditionally
transformed into directories with symlinks in the dst dir. Notably
`intel-mkl`, `cuda` and `qt` has top-level symlinked directories that
previously resulted in empty directories in the view. We now avoid
cycles and possible exponential blowup by only expanding symlinks that:
- point to dirs deeper in the folder structure;
- are a fixed depth of 2.

14 files changed, 950 insertions, 25 deletions

diff --git a/lib/spack/llnl/util/filesystem.py b/lib/spack/llnl/util/filesystem.py
index 179e1703b9..a0129e3550 100644
--- a/lib/spack/llnl/util/filesystem.py
+++ b/lib/spack/llnl/util/filesystem.py
@@ -1044,6 +1044,79 @@ def traverse_tree(source_root, dest_root, rel_path='', **kwargs):
         yield (source_path, dest_path)
 
 
+def lexists_islink_isdir(path):
+    """Computes the tuple (lexists(path), islink(path), isdir(path)) in a minimal
+    number of stat calls."""
+    # First try to lstat, so we know if it's a link or not.
+    try:
+        lst = os.lstat(path)
+    except (IOError, OSError):
+        return False, False, False
+
+    is_link = stat.S_ISLNK(lst.st_mode)
+
+    # Check whether file is a dir.
+    if not is_link:
+        is_dir = stat.S_ISDIR(lst.st_mode)
+        return True, is_link, is_dir
+
+    # Check whether symlink points to a dir.
+    try:
+        st = os.stat(path)
+        is_dir = stat.S_ISDIR(st.st_mode)
+    except (IOError, OSError):
+        # Dangling symlink (i.e. it lexists but not exists)
+        is_dir = False
+
+    return True, is_link, is_dir
+
+
+def visit_directory_tree(root, visitor, rel_path='', depth=0):
+    """
+    Recurses the directory root depth-first through a visitor pattern
+
+    The visitor interface is as follows:
+    - visit_file(root, rel_path, depth)
+    - before_visit_dir(root, rel_path, depth) -> bool
+        if True, descends into this directory
+    - before_visit_symlinked_dir(root, rel_path, depth) -> bool
+        if True, descends into this directory
+    - after_visit_dir(root, rel_path, depth) -> void
+        only called when before_visit_dir returns True
+    - after_visit_symlinked_dir(root, rel_path, depth) -> void
+        only called when before_visit_symlinked_dir returns True
+    """
+    dir = os.path.join(root, rel_path)
+
+    if sys.version_info >= (3, 5, 0):
+        dir_entries = sorted(os.scandir(dir), key=lambda d: d.name)  # novermin
+    else:
+        dir_entries = os.listdir(dir)
+        dir_entries.sort()
+
+    for f in dir_entries:
+        if sys.version_info >= (3, 5, 0):
+            rel_child = os.path.join(rel_path, f.name)
+            islink, isdir = f.is_symlink(), f.is_dir()
+        else:
+            rel_child = os.path.join(rel_path, f)
+            lexists, islink, isdir = lexists_islink_isdir(os.path.join(dir, f))
+            if not lexists:
+                continue
+
+        if not isdir:
+            # Handle files
+            visitor.visit_file(root, rel_child, depth)
+        elif not islink and visitor.before_visit_dir(root, rel_child, depth):
+            # Handle ordinary directories
+            visit_directory_tree(root, visitor, rel_child, depth + 1)
+            visitor.after_visit_dir(root, rel_child, depth)
+        elif islink and visitor.before_visit_symlinked_dir(root, rel_child, depth):
+            # Handle symlinked directories
+            visit_directory_tree(root, visitor, rel_child, depth + 1)
+            visitor.after_visit_symlinked_dir(root, rel_child, depth)
+
+
 @system_path_filter
 def set_executable(path):
     mode = os.stat(path).st_mode
diff --git a/lib/spack/llnl/util/link_tree.py b/lib/spack/llnl/util/link_tree.py
index bd91a1dabc..58cc30c2b1 100644
--- a/lib/spack/llnl/util/link_tree.py
+++ b/lib/spack/llnl/util/link_tree.py
@@ -10,6 +10,7 @@ from __future__ import print_function
 import filecmp
 import os
 import shutil
+from collections import OrderedDict
 
 import llnl.util.tty as tty
 from llnl.util.filesystem import mkdirp, touch, traverse_tree
@@ -30,6 +31,246 @@ def remove_link(src, dest):
         os.remove(dest)
 
 
+class MergeConflict:
+    """
+    The invariant here is that src_a and src_b are both mapped
+    to dst:
+
+        project(src_a) == project(src_b) == dst
+    """
+    def __init__(self, dst, src_a=None, src_b=None):
+        self.dst = dst
+        self.src_a = src_a
+        self.src_b = src_b
+
+
+class SourceMergeVisitor(object):
+    """
+    Visitor that produces actions:
+    - An ordered list of directories to create in dst
+    - A list of files to link in dst
+    - A list of merge conflicts in dst/
+    """
+    def __init__(self, ignore=None):
+        self.ignore = ignore if ignore is not None else lambda f: False
+
+        # When mapping <src root> to <dst root>/<projection>, we need
+        # to prepend the <projection> bit to the relative path in the
+        # destination dir.
+        self.projection = ''
+
+        # When a file blocks another file, the conflict can sometimes
+        # be resolved / ignored (e.g. <prefix>/LICENSE or
+        # or <site-packages>/<namespace>/__init__.py conflicts can be
+        # ignored).
+        self.file_conflicts = []
+
+        # When we have to create a dir where a file is, or a file
+        # where a dir is, we have fatal errors, listed here.
+        self.fatal_conflicts = []
+
+        # What directories we have to make; this is an ordered set,
+        # so that we have a fast lookup and can run mkdir in order.
+        self.directories = OrderedDict()
+
+        # Files to link. Maps dst_rel to (src_rel, src_root)
+        self.files = OrderedDict()
+
+    def before_visit_dir(self, root, rel_path, depth):
+        """
+        Register a directory if dst / rel_path is not blocked by a file or ignored.
+        """
+        proj_rel_path = os.path.join(self.projection, rel_path)
+
+        if self.ignore(rel_path):
+            # Don't recurse when dir is ignored.
+            return False
+        elif proj_rel_path in self.files:
+            # Can't create a dir where a file is.
+            src_a_root, src_a_relpath = self.files[proj_rel_path]
+            self.fatal_conflicts.append(MergeConflict(
+                dst=proj_rel_path,
+                src_a=os.path.join(src_a_root, src_a_relpath),
+                src_b=os.path.join(root, rel_path)))
+            return False
+        elif proj_rel_path in self.directories:
+            # No new directory, carry on.
+            return True
+        else:
+            # Register new directory.
+            self.directories[proj_rel_path] = (root, rel_path)
+            return True
+
+    def after_visit_dir(self, root, rel_path, depth):
+        pass
+
+    def before_visit_symlinked_dir(self, root, rel_path, depth):
+        """
+        Replace symlinked dirs with actual directories when possible in low depths,
+        otherwise handle it as a file (i.e. we link to the symlink).
+
+        Transforming symlinks into dirs makes it more likely we can merge directories,
+        e.g. when <prefix>/lib -> <prefix>/subdir/lib.
+
+        We only do this when the symlink is pointing into a subdirectory from the
+        symlink's directory, to avoid potential infinite recursion; and only at a
+        constant level of nesting, to avoid potential exponential blowups in file
+        duplication.
+        """
+        if self.ignore(rel_path):
+            return False
+
+        # Only follow symlinked dirs in <prefix>/**/**/*
+        if depth > 1:
+            handle_as_dir = False
+        else:
+            # Only follow symlinked dirs when pointing deeper
+            src = os.path.join(root, rel_path)
+            real_parent = os.path.realpath(os.path.dirname(src))
+            real_child = os.path.realpath(src)
+            handle_as_dir = real_child.startswith(real_parent)
+
+        if handle_as_dir:
+            return self.before_visit_dir(root, rel_path, depth)
+
+        self.visit_file(root, rel_path, depth)
+        return False
+
+    def after_visit_symlinked_dir(self, root, rel_path, depth):
+        pass
+
+    def visit_file(self, root, rel_path, depth):
+        proj_rel_path = os.path.join(self.projection, rel_path)
+
+        if self.ignore(rel_path):
+            pass
+        elif proj_rel_path in self.directories:
+            # Can't create a file where a dir is; fatal error
+            src_a_root, src_a_relpath = self.directories[proj_rel_path]
+            self.fatal_conflicts.append(MergeConflict(
+                dst=proj_rel_path,
+                src_a=os.path.join(src_a_root, src_a_relpath),
+                src_b=os.path.join(root, rel_path)))
+        elif proj_rel_path in self.files:
+            # In some cases we can resolve file-file conflicts
+            src_a_root, src_a_relpath = self.files[proj_rel_path]
+            self.file_conflicts.append(MergeConflict(
+                dst=proj_rel_path,
+                src_a=os.path.join(src_a_root, src_a_relpath),
+                src_b=os.path.join(root, rel_path)))
+        else:
+            # Otherwise register this file to be linked.
+            self.files[proj_rel_path] = (root, rel_path)
+
+    def set_projection(self, projection):
+        self.projection = os.path.normpath(projection)
+
+        # Todo, is this how to check in general for empty projection?
+        if self.projection == '.':
+            self.projection = ''
+            return
+
+        # If there is a projection, we'll also create the directories
+        # it consists of, and check whether that's causing conflicts.
+        path = ''
+        for part in self.projection.split(os.sep):
+            path = os.path.join(path, part)
+            if path not in self.files:
+                self.directories[path] = ('<projection>', path)
+            else:
+                # Can't create a dir where a file is.
+                src_a_root, src_a_relpath = self.files[path]
+                self.fatal_conflicts.append(MergeConflict(
+                    dst=path,
+                    src_a=os.path.join(src_a_root, src_a_relpath),
+                    src_b=os.path.join('<projection>', path)))
+
+
+class DestinationMergeVisitor(object):
+    """DestinatinoMergeVisitor takes a SourceMergeVisitor
+    and:
+
+    a. registers additional conflicts when merging
+       to the destination prefix
+    b. removes redundant mkdir operations when
+       directories already exist in the destination
+       prefix.
+
+    This also makes sure that symlinked directories
+    in the target prefix will never be merged with
+    directories in the sources directories.
+    """
+    def __init__(self, source_merge_visitor):
+        self.src = source_merge_visitor
+
+    def before_visit_dir(self, root, rel_path, depth):
+        # If destination dir is a file in a src dir, add a conflict,
+        # and don't traverse deeper
+        if rel_path in self.src.files:
+            src_a_root, src_a_relpath = self.src.files[rel_path]
+            self.src.fatal_conflicts.append(MergeConflict(
+                rel_path,
+                os.path.join(src_a_root, src_a_relpath),
+                os.path.join(root, rel_path)))
+            return False
+
+        # If destination dir was also a src dir, remove the mkdir
+        # action, and traverse deeper.
+        if rel_path in self.src.directories:
+            del self.src.directories[rel_path]
+            return True
+
+        # If the destination dir does not appear in the src dir,
+        # don't descend into it.
+        return False
+
+    def after_visit_dir(self, root, rel_path, depth):
+        pass
+
+    def before_visit_symlinked_dir(self, root, rel_path, depth):
+        """
+        Symlinked directories in the destination prefix should
+        be seen as files; we should not accidentally merge
+        source dir with a symlinked dest dir.
+        """
+        # Always conflict
+        if rel_path in self.src.directories:
+            src_a_root, src_a_relpath = self.src.directories[rel_path]
+            self.src.fatal_conflicts.append(MergeConflict(
+                rel_path,
+                os.path.join(src_a_root, src_a_relpath),
+                os.path.join(root, rel_path)))
+
+        if rel_path in self.src.files:
+            src_a_root, src_a_relpath = self.src.files[rel_path]
+            self.src.fatal_conflicts.append(MergeConflict(
+                rel_path,
+                os.path.join(src_a_root, src_a_relpath),
+                os.path.join(root, rel_path)))
+
+        # Never descend into symlinked target dirs.
+        return False
+
+    def after_visit_symlinked_dir(self, root, rel_path, depth):
+        pass
+
+    def visit_file(self, root, rel_path, depth):
+        # Can't merge a file if target already exists
+        if rel_path in self.src.directories:
+            src_a_root, src_a_relpath = self.src.directories[rel_path]
+            self.src.fatal_conflicts.append(MergeConflict(
+                rel_path,
+                os.path.join(src_a_root, src_a_relpath),
+                os.path.join(root, rel_path)))
+
+        elif rel_path in self.src.files:
+            src_a_root, src_a_relpath = self.src.files[rel_path]
+            self.src.fatal_conflicts.append(MergeConflict(
+                rel_path,
+                os.path.join(src_a_root, src_a_relpath),
+                os.path.join(root, rel_path)))
+
+
 class LinkTree(object):
     """Class to create trees of symbolic links from a source directory.
 
@@ -138,7 +379,7 @@ class LinkTree(object):
         conflict = self.find_conflict(
             dest_root, ignore=ignore, ignore_file_conflicts=ignore_conflicts)
         if conflict:
-            raise MergeConflictError(conflict)
+            raise SingleMergeConflictError(conflict)
 
         self.merge_directories(dest_root, ignore)
         existing = []
@@ -170,7 +411,24 @@ class LinkTree(object):
 
 
 class MergeConflictError(Exception):
+    pass
 
+
+class SingleMergeConflictError(MergeConflictError):
     def __init__(self, path):
         super(MergeConflictError, self).__init__(
             "Package merge blocked by file: %s" % path)
+
+
+class MergeConflictSummary(MergeConflictError):
+    def __init__(self, conflicts):
+        """
+        A human-readable summary of file system view merge conflicts (showing only the
+        first 3 issues.)
+        """
+        msg = "{0} fatal error(s) when merging prefixes:\n".format(len(conflicts))
+        # show the first 3 merge conflicts.
+        for conflict in conflicts[:3]:
+            msg += "    `{0}` and `{1}` both project to `{2}`".format(
+                conflict.src_a, conflict.src_b, conflict.dst)
+        super(MergeConflictSummary, self).__init__(msg)
diff --git a/lib/spack/spack/build_systems/python.py b/lib/spack/spack/build_systems/python.py
index 6aa225ba6c..d7e00d84b5 100644
--- a/lib/spack/spack/build_systems/python.py
+++ b/lib/spack/spack/build_systems/python.py
@@ -216,7 +216,7 @@ class PythonPackage(PackageBase):
 
         return conflicts
 
-    def add_files_to_view(self, view, merge_map):
+    def add_files_to_view(self, view, merge_map, skip_if_exists=False):
         bin_dir = self.spec.prefix.bin
         python_prefix = self.extendee_spec.prefix
         python_is_external = self.extendee_spec.external
diff --git a/lib/spack/spack/environment/environment.py b/lib/spack/spack/environment/environment.py
index ea1dc85d4a..5acac8c390 100644
--- a/lib/spack/spack/environment/environment.py
+++ b/lib/spack/spack/environment/environment.py
@@ -44,7 +44,7 @@ import spack.util.path
 import spack.util.spack_json as sjson
 import spack.util.spack_yaml as syaml
 from spack.filesystem_view import (
-    YamlFilesystemView,
+    SimpleFilesystemView,
     inverse_view_func_parser,
     view_func_parser,
 )
@@ -444,18 +444,16 @@ class ViewDescriptor(object):
                 rooted at that path. Default None. This should only be used to
                 regenerate the view, and cannot be used to access specs.
         """
-        root = self._current_root
-        if new:
-            root = new
+        root = new if new else self._current_root
         if not root:
             # This can only be hit if we write a future bug
             msg = ("Attempting to get nonexistent view from environment. "
                    "View root is at %s" % self.root)
             raise SpackEnvironmentViewError(msg)
-        return YamlFilesystemView(root, spack.store.layout,
-                                  ignore_conflicts=True,
-                                  projections=self.projections,
-                                  link=self.link_type)
+        return SimpleFilesystemView(root, spack.store.layout,
+                                    ignore_conflicts=True,
+                                    projections=self.projections,
+                                    link=self.link_type)
 
     def __contains__(self, spec):
         """Is the spec described by the view descriptor
diff --git a/lib/spack/spack/filesystem_view.py b/lib/spack/spack/filesystem_view.py
index d6b3d3ed8f..7c14e3d044 100644
--- a/lib/spack/spack/filesystem_view.py
+++ b/lib/spack/spack/filesystem_view.py
@@ -5,6 +5,7 @@
 
 import collections
 import functools as ft
+import itertools
 import os
 import re
 import shutil
@@ -12,9 +13,20 @@ import sys
 
 from llnl.util import tty
 from llnl.util.compat import filter, map, zip
-from llnl.util.filesystem import mkdirp, remove_dead_links, remove_empty_directories
+from llnl.util.filesystem import (
+    mkdirp,
+    remove_dead_links,
+    remove_empty_directories,
+    visit_directory_tree,
+)
 from llnl.util.lang import index_by, match_predicate
-from llnl.util.link_tree import LinkTree, MergeConflictError
+from llnl.util.link_tree import (
+    DestinationMergeVisitor,
+    LinkTree,
+    MergeConflictSummary,
+    SingleMergeConflictError,
+    SourceMergeVisitor,
+)
 from llnl.util.symlink import symlink
 from llnl.util.tty.color import colorize
 
@@ -413,7 +425,7 @@ class YamlFilesystemView(FilesystemView):
             conflicts.extend(pkg.view_file_conflicts(self, merge_map))
 
         if conflicts:
-            raise MergeConflictError(conflicts[0])
+            raise SingleMergeConflictError(conflicts[0])
 
         # merge directories with the tree
         tree.merge_directories(view_dst, ignore_file)
@@ -730,6 +742,137 @@ class YamlFilesystemView(FilesystemView):
                      'Skipping already activated package: %s' % spec.name)
 
 
+class SimpleFilesystemView(FilesystemView):
+    """A simple and partial implementation of FilesystemView focused on
+    performance and immutable views, where specs cannot be removed after they
+    were added."""
+
+    def __init__(self, root, layout, **kwargs):
+        super(SimpleFilesystemView, self).__init__(root, layout, **kwargs)
+
+    def add_specs(self, *specs, **kwargs):
+        assert all((s.concrete for s in specs))
+        if len(specs) == 0:
+            return
+
+        # Drop externals
+        for s in specs:
+            if s.external:
+                tty.warn('Skipping external package: ' + s.short_spec)
+        specs = [s for s in specs if not s.external]
+
+        if kwargs.get("exclude", None):
+            specs = set(filter_exclude(specs, kwargs["exclude"]))
+
+        # Ignore spack meta data folder.
+        def skip_list(file):
+            return os.path.basename(file) == spack.store.layout.metadata_dir
+
+        visitor = SourceMergeVisitor(ignore=skip_list)
+
+        # Gather all the directories to be made and files to be linked
+        for spec in specs:
+            src_prefix = spec.package.view_source()
+            visitor.set_projection(self.get_relative_projection_for_spec(spec))
+            visit_directory_tree(src_prefix, visitor)
+
+        # Check for conflicts in destination dir.
+        visit_directory_tree(self._root, DestinationMergeVisitor(visitor))
+
+        # Throw on fatal dir-file conflicts.
+        if visitor.fatal_conflicts:
+            raise MergeConflictSummary(visitor.fatal_conflicts)
+
+        # Inform about file-file conflicts.
+        if visitor.file_conflicts:
+            if self.ignore_conflicts:
+                tty.debug("{0} file conflicts".format(len(visitor.file_conflicts)))
+            else:
+                raise MergeConflictSummary(visitor.file_conflicts)
+
+        tty.debug("Creating {0} dirs and {1} links".format(
+            len(visitor.directories),
+            len(visitor.files)))
+
+        # Make the directory structure
+        for dst in visitor.directories:
+            os.mkdir(os.path.join(self._root, dst))
+
+        # Then group the files to be linked by spec...
+        # For compatibility, we have to create a merge_map dict mapping
+        # full_src => full_dst
+        files_per_spec = itertools.groupby(
+            visitor.files.items(), key=lambda item: item[1][0])
+
+        for (spec, (src_root, rel_paths)) in zip(specs, files_per_spec):
+            merge_map = dict()
+            for dst_rel, (_, src_rel) in rel_paths:
+                full_src = os.path.join(src_root, src_rel)
+                full_dst = os.path.join(self._root, dst_rel)
+                merge_map[full_src] = full_dst
+            spec.package.add_files_to_view(self, merge_map, skip_if_exists=True)
+
+        # Finally create the metadata dirs.
+        self.link_metadata(specs)
+
+    def link_metadata(self, specs):
+        metadata_visitor = SourceMergeVisitor()
+
+        for spec in specs:
+            src_prefix = os.path.join(
+                spec.package.view_source(),
+                spack.store.layout.metadata_dir)
+            proj = os.path.join(
+                self.get_relative_projection_for_spec(spec),
+                spack.store.layout.metadata_dir,
+                spec.name)
+            metadata_visitor.set_projection(proj)
+            visit_directory_tree(src_prefix, metadata_visitor)
+
+        # Check for conflicts in destination dir.
+        visit_directory_tree(self._root, DestinationMergeVisitor(metadata_visitor))
+
+        # Throw on dir-file conflicts -- unlikely, but who knows.
+        if metadata_visitor.fatal_conflicts:
+            raise MergeConflictSummary(metadata_visitor.fatal_conflicts)
+
+        # We are strict here for historical reasons
+        if metadata_visitor.file_conflicts:
+            raise MergeConflictSummary(metadata_visitor.file_conflicts)
+
+        for dst in metadata_visitor.directories:
+            os.mkdir(os.path.join(self._root, dst))
+
+        for dst_relpath, (src_root, src_relpath) in metadata_visitor.files.items():
+            self.link(os.path.join(src_root, src_relpath),
+                      os.path.join(self._root, dst_relpath))
+
+    def get_relative_projection_for_spec(self, spec):
+        # Extensions are placed by their extendee, not by their own spec
+        if spec.package.extendee_spec:
+            spec = spec.package.extendee_spec
+
+        p = spack.projections.get_projection(self.projections, spec)
+        return spec.format(p) if p else ''
+
+    def get_projection_for_spec(self, spec):
+        """
+           Return the projection for a spec in this view.
+
+           Relies on the ordering of projections to avoid ambiguity.
+        """
+        spec = spack.spec.Spec(spec)
+        # Extensions are placed by their extendee, not by their own spec
+        locator_spec = spec
+        if spec.package.extendee_spec:
+            locator_spec = spec.package.extendee_spec
+
+        proj = spack.projections.get_projection(self.projections, locator_spec)
+        if proj:
+            return os.path.join(self._root, locator_spec.format(proj))
+        return self._root
+
+
 #####################
 # utility functions #
 #####################
diff --git a/lib/spack/spack/package.py b/lib/spack/spack/package.py
index de903c4ffa..46683cc60b 100644
--- a/lib/spack/spack/package.py
+++ b/lib/spack/spack/package.py
@@ -479,14 +479,26 @@ class PackageViewMixin(object):
         """
         return set(dst for dst in merge_map.values() if os.path.lexists(dst))
 
-    def add_files_to_view(self, view, merge_map):
+    def add_files_to_view(self, view, merge_map, skip_if_exists=False):
         """Given a map of package files to destination paths in the view, add
         the files to the view. By default this adds all files. Alternative
         implementations may skip some files, for example if other packages
         linked into the view already include the file.
-        """
-        for src, dst in merge_map.items():
-            if not os.path.lexists(dst):
+
+        Args:
+            view (spack.filesystem_view.FilesystemView): the view that's updated
+            merge_map (dict): maps absolute source paths to absolute dest paths for
+                all files in from this package.
+            skip_if_exists (bool): when True, don't link files in view when they
+                already exist. When False, always link files, without checking
+                if they already exist.
+        """
+        if skip_if_exists:
+            for src, dst in merge_map.items():
+                if not os.path.lexists(dst):
+                    view.link(src, dst, spec=self.spec)
+        else:
+            for src, dst in merge_map.items():
                 view.link(src, dst, spec=self.spec)
 
     def remove_files_from_view(self, view, merge_map):
diff --git a/lib/spack/spack/test/cmd/env.py b/lib/spack/spack/test/cmd/env.py
index 7315a9d7f1..7853cb16f2 100644
--- a/lib/spack/spack/test/cmd/env.py
+++ b/lib/spack/spack/test/cmd/env.py
@@ -1147,16 +1147,49 @@ def test_env_updates_view_install(
 
 def test_env_view_fails(
         tmpdir, mock_packages, mock_stage, mock_fetch, install_mockery):
+    # We currently ignore file-file conflicts for the prefix merge,
+    # so in principle there will be no errors in this test. But
+    # the .spack metadata dir is handled separately and is more strict.
+    # It also throws on file-file conflicts. That's what we're checking here
+    # by adding the same package twice to a view.
     view_dir = tmpdir.join('view')
     env('create', '--with-view=%s' % view_dir, 'test')
     with ev.read('test'):
         add('libelf')
         add('libelf cflags=-g')
-        with pytest.raises(llnl.util.link_tree.MergeConflictError,
-                           match='merge blocked by file'):
+        with pytest.raises(llnl.util.link_tree.MergeConflictSummary,
+                           match=spack.store.layout.metadata_dir):
             install('--fake')
 
 
+def test_env_view_fails_dir_file(
+        tmpdir, mock_packages, mock_stage, mock_fetch, install_mockery):
+    # This environment view fails to be created because a file
+    # and a dir are in the same path. Test that it mentions the problematic path.
+    view_dir = tmpdir.join('view')
+    env('create', '--with-view=%s' % view_dir, 'test')
+    with ev.read('test'):
+        add('view-dir-file')
+        add('view-dir-dir')
+        with pytest.raises(llnl.util.link_tree.MergeConflictSummary,
+                           match=os.path.join('bin', 'x')):
+            install()
+
+
+def test_env_view_succeeds_symlinked_dir_file(
+        tmpdir, mock_packages, mock_stage, mock_fetch, install_mockery):
+    # A symlinked dir and an ordinary dir merge happily
+    view_dir = tmpdir.join('view')
+    env('create', '--with-view=%s' % view_dir, 'test')
+    with ev.read('test'):
+        add('view-dir-symlinked-dir')
+        add('view-dir-dir')
+        install()
+        x_dir = os.path.join(str(view_dir), 'bin', 'x')
+        assert os.path.exists(os.path.join(x_dir, 'file_in_dir'))
+        assert os.path.exists(os.path.join(x_dir, 'file_in_symlinked_dir'))
+
+
 def test_env_without_view_install(
         tmpdir, mock_stage, mock_fetch, install_mockery):
     # Test enabling a view after installing specs
@@ -1193,9 +1226,10 @@ env:
         install('--fake')
 
     e = ev.read('test')
-    # Try retrieving the view object
-    view = e.default_view.view()
-    assert view.get_spec('mpileaks')
+
+    # Check that metadata folder for this spec exists
+    assert os.path.isdir(os.path.join(e.default_view.view()._root,
+                         '.spack', 'mpileaks'))
 
 
 def test_env_updates_view_install_package(
diff --git a/lib/spack/spack/test/conftest.py b/lib/spack/spack/test/conftest.py
index c946898076..32c73af18c 100644
--- a/lib/spack/spack/test/conftest.py
+++ b/lib/spack/spack/test/conftest.py
@@ -1576,3 +1576,40 @@ def brand_new_binary_cache():
     yield
     spack.binary_distribution.binary_index = llnl.util.lang.Singleton(
         spack.binary_distribution._binary_index)
+
+
+@pytest.fixture()
+def noncyclical_dir_structure(tmpdir):
+    """
+    Create some non-trivial directory structure with
+    symlinks to dirs and dangling symlinks, but no cycles::
+
+        .
+        |-- a/
+        |   |-- d/
+        |   |-- file_1
+        |   |-- to_file_1 -> file_1
+        |   `-- to_c -> ../c
+        |-- b -> a
+        |-- c/
+        |   |-- dangling_link -> nowhere
+        |   `-- file_2
+        `-- file_3
+    """
+    d, j = tmpdir.mkdir('nontrivial-dir'), os.path.join
+
+    with d.as_cwd():
+        os.mkdir(j('a'))
+        os.mkdir(j('a', 'd'))
+        with open(j('a', 'file_1'), 'wb'):
+            pass
+        os.symlink(j('file_1'), j('a', 'to_file_1'))
+        os.symlink(j('..', 'c'), j('a', 'to_c'))
+        os.symlink(j('a'), j('b'))
+        os.mkdir(j('c'))
+        os.symlink(j('nowhere'), j('c', 'dangling_link'))
+        with open(j('c', 'file_2'), 'wb'):
+            pass
+        with open(j('file_3'), 'wb'):
+            pass
+    yield d
diff --git a/lib/spack/spack/test/llnl/util/filesystem.py b/lib/spack/spack/test/llnl/util/filesystem.py
index b5a4594efb..916d83a352 100644
--- a/lib/spack/spack/test/llnl/util/filesystem.py
+++ b/lib/spack/spack/test/llnl/util/filesystem.py
@@ -698,3 +698,171 @@ def test_is_nonsymlink_exe_with_shebang(tmpdir):
         assert not fs.is_nonsymlink_exe_with_shebang('executable_but_not_script')
         assert not fs.is_nonsymlink_exe_with_shebang('not_executable_with_shebang')
         assert not fs.is_nonsymlink_exe_with_shebang('symlink_to_executable_script')
+
+
+def test_lexists_islink_isdir(tmpdir):
+    root = str(tmpdir)
+
+    # Create a directory and a file, an a bunch of symlinks.
+    dir = os.path.join(root, "dir")
+    file = os.path.join(root, "file")
+    nonexistent = os.path.join(root, "does_not_exist")
+    symlink_to_dir = os.path.join(root, "symlink_to_dir")
+    symlink_to_file = os.path.join(root, "symlink_to_file")
+    dangling_symlink = os.path.join(root, "dangling_symlink")
+    symlink_to_dangling_symlink = os.path.join(root, "symlink_to_dangling_symlink")
+    symlink_to_symlink_to_dir = os.path.join(root, "symlink_to_symlink_to_dir")
+    symlink_to_symlink_to_file = os.path.join(root, "symlink_to_symlink_to_file")
+
+    os.mkdir(dir)
+    with open(file, "wb") as f:
+        f.write(b"file")
+
+    os.symlink("dir", symlink_to_dir)
+    os.symlink("file", symlink_to_file)
+    os.symlink("does_not_exist", dangling_symlink)
+    os.symlink("dangling_symlink", symlink_to_dangling_symlink)
+    os.symlink("symlink_to_dir", symlink_to_symlink_to_dir)
+    os.symlink("symlink_to_file", symlink_to_symlink_to_file)
+
+    assert fs.lexists_islink_isdir(dir) == (True, False, True)
+    assert fs.lexists_islink_isdir(file) == (True, False, False)
+    assert fs.lexists_islink_isdir(nonexistent) == (False, False, False)
+    assert fs.lexists_islink_isdir(symlink_to_dir) == (True, True, True)
+    assert fs.lexists_islink_isdir(symlink_to_file) == (True, True, False)
+    assert fs.lexists_islink_isdir(symlink_to_dangling_symlink) == (True, True, False)
+    assert fs.lexists_islink_isdir(symlink_to_symlink_to_dir) == (True, True, True)
+    assert fs.lexists_islink_isdir(symlink_to_symlink_to_file) == (True, True, False)
+
+
+class RegisterVisitor(object):
+    """A directory visitor that keeps track of all visited paths"""
+    def __init__(self, root, follow_dirs=True, follow_symlink_dirs=True):
+        self.files = []
+        self.dirs_before = []
+        self.symlinked_dirs_before = []
+        self.dirs_after = []
+        self.symlinked_dirs_after = []
+
+        self.root = root
+        self.follow_dirs = follow_dirs
+        self.follow_symlink_dirs = follow_symlink_dirs
+
+    def check(self, root, rel_path, depth):
+        # verify the (root, rel_path, depth) make sense.
+        assert root == self.root and depth + 1 == len(rel_path.split(os.sep))
+
+    def visit_file(self, root, rel_path, depth):
+        self.check(root, rel_path, depth)
+        self.files.append(rel_path)
+
+    def before_visit_dir(self, root, rel_path, depth):
+        self.check(root, rel_path, depth)
+        self.dirs_before.append(rel_path)
+        return self.follow_dirs
+
+    def before_visit_symlinked_dir(self, root, rel_path, depth):
+        self.check(root, rel_path, depth)
+        self.symlinked_dirs_before.append(rel_path)
+        return self.follow_symlink_dirs
+
+    def after_visit_dir(self, root, rel_path, depth):
+        self.check(root, rel_path, depth)
+        self.dirs_after.append(rel_path)
+
+    def after_visit_symlinked_dir(self, root, rel_path, depth):
+        self.check(root, rel_path, depth)
+        self.symlinked_dirs_after.append(rel_path)
+
+
+@pytest.mark.skipif(sys.platform == 'win32', reason="Requires symlinks")
+def test_visit_directory_tree_follow_all(noncyclical_dir_structure):
+    root = str(noncyclical_dir_structure)
+    visitor = RegisterVisitor(root, follow_dirs=True, follow_symlink_dirs=True)
+    fs.visit_directory_tree(root, visitor)
+    j = os.path.join
+    assert visitor.files == [
+        j('a', 'file_1'),
+        j('a', 'to_c', 'dangling_link'),
+        j('a', 'to_c', 'file_2'),
+        j('a', 'to_file_1'),
+        j('b', 'file_1'),
+        j('b', 'to_c', 'dangling_link'),
+        j('b', 'to_c', 'file_2'),
+        j('b', 'to_file_1'),
+        j('c', 'dangling_link'),
+        j('c', 'file_2'),
+        j('file_3'),
+    ]
+    assert visitor.dirs_before == [
+        j('a'),
+        j('a', 'd'),
+        j('b', 'd'),
+        j('c'),
+    ]
+    assert visitor.dirs_after == [
+        j('a', 'd'),
+        j('a'),
+        j('b', 'd'),
+        j('c'),
+    ]
+    assert visitor.symlinked_dirs_before == [
+        j('a', 'to_c'),
+        j('b'),
+        j('b', 'to_c'),
+    ]
+    assert visitor.symlinked_dirs_after == [
+        j('a', 'to_c'),
+        j('b', 'to_c'),
+        j('b'),
+    ]
+
+
+@pytest.mark.skipif(sys.platform == 'win32', reason="Requires symlinks")
+def test_visit_directory_tree_follow_dirs(noncyclical_dir_structure):
+    root = str(noncyclical_dir_structure)
+    visitor = RegisterVisitor(root, follow_dirs=True, follow_symlink_dirs=False)
+    fs.visit_directory_tree(root, visitor)
+    j = os.path.join
+    assert visitor.files == [
+        j('a', 'file_1'),
+        j('a', 'to_file_1'),
+        j('c', 'dangling_link'),
+        j('c', 'file_2'),
+        j('file_3'),
+    ]
+    assert visitor.dirs_before == [
+        j('a'),
+        j('a', 'd'),
+        j('c'),
+    ]
+    assert visitor.dirs_after == [
+        j('a', 'd'),
+        j('a'),
+        j('c'),
+    ]
+    assert visitor.symlinked_dirs_before == [
+        j('a', 'to_c'),
+        j('b'),
+    ]
+    assert not visitor.symlinked_dirs_after
+
+
+@pytest.mark.skipif(sys.platform == 'win32', reason="Requires symlinks")
+def test_visit_directory_tree_follow_none(noncyclical_dir_structure):
+    root = str(noncyclical_dir_structure)
+    visitor = RegisterVisitor(root, follow_dirs=False, follow_symlink_dirs=False)
+    fs.visit_directory_tree(root, visitor)
+    j = os.path.join
+    assert visitor.files == [
+        j('file_3'),
+    ]
+    assert visitor.dirs_before == [
+        j('a'),
+        j('c'),
+    ]
+    assert not visitor.dirs_after
+    assert visitor.symlinked_dirs_before == [
+        j('b'),
+    ]
+    assert not visitor.symlinked_dirs_after
diff --git a/lib/spack/spack/test/llnl/util/link_tree.py b/lib/spack/spack/test/llnl/util/link_tree.py
index aef9aafc6d..296a2cd263 100644
--- a/lib/spack/spack/test/llnl/util/link_tree.py
+++ b/lib/spack/spack/test/llnl/util/link_tree.py
@@ -8,8 +8,8 @@ import sys
 
 import pytest
 
-from llnl.util.filesystem import mkdirp, touchp, working_dir
-from llnl.util.link_tree import LinkTree
+from llnl.util.filesystem import mkdirp, touchp, visit_directory_tree, working_dir
+from llnl.util.link_tree import DestinationMergeVisitor, LinkTree, SourceMergeVisitor
 from llnl.util.symlink import islink
 
 from spack.stage import Stage
@@ -173,3 +173,141 @@ def test_ignore(stage, link_tree):
 
         assert os.path.isfile('source/.spec')
         assert os.path.isfile('dest/.spec')
+
+
+def test_source_merge_visitor_does_not_follow_symlinked_dirs_at_depth(tmpdir):
+    """Given an dir structure like this::
+
+        .
+        `-- a
+            |-- b
+            |   |-- c
+            |   |   |-- d
+            |   |   |   `-- file
+            |   |   `-- symlink_d -> d
+            |   `-- symlink_c -> c
+            `-- symlink_b -> b
+
+    The SoureMergeVisitor will expand symlinked dirs to directories, but only
+    to fixed depth, to avoid exponential explosion. In our current defaults,
+    symlink_b will be expanded, but symlink_c and symlink_d will not.
+    """
+    j = os.path.join
+    with tmpdir.as_cwd():
+        os.mkdir(j('a'))
+        os.mkdir(j('a', 'b'))
+        os.mkdir(j('a', 'b', 'c'))
+        os.mkdir(j('a', 'b', 'c', 'd'))
+        os.symlink(j('b'), j('a', 'symlink_b'))
+        os.symlink(j('c'), j('a', 'b', 'symlink_c'))
+        os.symlink(j('d'), j('a', 'b', 'c', 'symlink_d'))
+        with open(j('a', 'b', 'c', 'd', 'file'), 'wb'):
+            pass
+
+    visitor = SourceMergeVisitor()
+    visit_directory_tree(str(tmpdir), visitor)
+    assert [p for p in visitor.files.keys()] == [
+        j('a', 'b', 'c', 'd', 'file'),
+        j('a', 'b', 'c', 'symlink_d'),  # treated as a file, not expanded
+        j('a', 'b', 'symlink_c'),  # treated as a file, not expanded
+        j('a', 'symlink_b', 'c', 'd', 'file'),  # symlink_b was expanded
+        j('a', 'symlink_b', 'c', 'symlink_d'),  # symlink_b was expanded
+        j('a', 'symlink_b', 'symlink_c')  # symlink_b was expanded
+    ]
+    assert [p for p in visitor.directories.keys()] == [
+        j('a'),
+        j('a', 'b'),
+        j('a', 'b', 'c'),
+        j('a', 'b', 'c', 'd'),
+        j('a', 'symlink_b'),
+        j('a', 'symlink_b', 'c'),
+        j('a', 'symlink_b', 'c', 'd'),
+    ]
+
+
+def test_source_merge_visitor_cant_be_cyclical(tmpdir):
+    """Given an dir structure like this::
+
+        .
+        |-- a
+        |   `-- symlink_b -> ../b
+        |   `-- symlink_symlink_b -> symlink_b
+        `-- b
+            `-- symlink_a -> ../a
+
+    The SoureMergeVisitor will not expand `a/symlink_b`, `a/symlink_symlink_b` and
+    `b/symlink_a` to avoid recursion. The general rule is: only expand symlinked dirs
+    pointing deeper into the directory structure.
+    """
+    j = os.path.join
+    with tmpdir.as_cwd():
+        os.mkdir(j('a'))
+        os.symlink(j('..', 'b'), j('a', 'symlink_b'))
+        os.symlink(j('symlink_b'), j('a', 'symlink_b_b'))
+        os.mkdir(j('b'))
+        os.symlink(j('..', 'a'), j('b', 'symlink_a'))
+
+    visitor = SourceMergeVisitor()
+    visit_directory_tree(str(tmpdir), visitor)
+    assert [p for p in visitor.files.keys()] == [
+        j('a', 'symlink_b'),
+        j('a', 'symlink_b_b'),
+        j('b', 'symlink_a')
+    ]
+    assert [p for p in visitor.directories.keys()] == [
+        j('a'),
+        j('b')
+    ]
+
+
+def test_destination_merge_visitor_always_errors_on_symlinked_dirs(tmpdir):
+    """When merging prefixes into a non-empty destination folder, and
+    this destination folder has a symlinked dir where the prefix has a dir,
+    we should never merge any files there, but register a fatal error."""
+    j = os.path.join
+
+    # Here example_a and example_b are symlinks.
+    with tmpdir.mkdir('dst').as_cwd():
+        os.mkdir('a')
+        os.symlink('a', 'example_a')
+        os.symlink('a', 'example_b')
+
+    # Here example_a is a directory, and example_b is a (non-expanded) symlinked
+    # directory.
+    with tmpdir.mkdir('src').as_cwd():
+        os.mkdir('example_a')
+        with open(j('example_a', 'file'), 'wb'):
+            pass
+        os.symlink('..', 'example_b')
+
+    visitor = SourceMergeVisitor()
+    visit_directory_tree(str(tmpdir.join('src')), visitor)
+    visit_directory_tree(str(tmpdir.join('dst')), DestinationMergeVisitor(visitor))
+
+    assert visitor.fatal_conflicts
+    conflicts = [c.dst for c in visitor.fatal_conflicts]
+    assert 'example_a' in conflicts
+    assert 'example_b' in conflicts
+
+
+def test_destination_merge_visitor_file_dir_clashes(tmpdir):
+    """Tests whether non-symlink file-dir and dir-file clashes as registered as fatal
+    errors"""
+    with tmpdir.mkdir('a').as_cwd():
+        os.mkdir('example')
+
+    with tmpdir.mkdir('b').as_cwd():
+        with open('example', 'wb'):
+            pass
+
+    a_to_b = SourceMergeVisitor()
+    visit_directory_tree(str(tmpdir.join('a')), a_to_b)
+    visit_directory_tree(str(tmpdir.join('b')), DestinationMergeVisitor(a_to_b))
+    assert a_to_b.fatal_conflicts
+    assert a_to_b.fatal_conflicts[0].dst == 'example'
+
+    b_to_a = SourceMergeVisitor()
+    visit_directory_tree(str(tmpdir.join('b')), b_to_a)
+    visit_directory_tree(str(tmpdir.join('a')), DestinationMergeVisitor(b_to_a))
+    assert b_to_a.fatal_conflicts
+    assert b_to_a.fatal_conflicts[0].dst == 'example'
diff --git a/var/spack/repos/builtin.mock/packages/view-dir-dir/package.py b/var/spack/repos/builtin.mock/packages/view-dir-dir/package.py
new file mode 100644
index 0000000000..b85a81caa4
--- /dev/null
+++ b/var/spack/repos/builtin.mock/packages/view-dir-dir/package.py
@@ -0,0 +1,21 @@
+# Copyright 2013-2022 Lawrence Livermore National Security, LLC and other
+# Spack Project Developers. See the top-level COPYRIGHT file for details.
+#
+# SPDX-License-Identifier: (Apache-2.0 OR MIT)
+
+import os
+
+
+class ViewDirDir(Package):
+    """Installs a <prefix>/bin/x where x is a dir, in contrast to view-dir-file."""
+    homepage = "http://www.spack.org"
+    url = "http://www.spack.org/downloads/aml-1.0.tar.gz"
+    has_code = False
+
+    version('0.1.0', sha256='cc89a8768693f1f11539378b21cdca9f0ce3fc5cb564f9b3e4154a051dcea69b')
+
+    def install(self, spec, prefix):
+        os.mkdir(os.path.join(prefix, 'bin'))
+        os.mkdir(os.path.join(prefix, 'bin', 'x'))
+        with open(os.path.join(prefix, 'bin', 'x', 'file_in_dir'), 'wb') as f:
+            f.write(b'hello world')
diff --git a/var/spack/repos/builtin.mock/packages/view-dir-file/package.py b/var/spack/repos/builtin.mock/packages/view-dir-file/package.py
new file mode 100644
index 0000000000..a47e02ca24
--- /dev/null
+++ b/var/spack/repos/builtin.mock/packages/view-dir-file/package.py
@@ -0,0 +1,20 @@
+# Copyright 2013-2022 Lawrence Livermore National Security, LLC and other
+# Spack Project Developers. See the top-level COPYRIGHT file for details.
+#
+# SPDX-License-Identifier: (Apache-2.0 OR MIT)
+
+import os
+
+
+class ViewDirFile(Package):
+    """Installs a <prefix>/bin/x where x is a file, in contrast to view-dir-dir"""
+    homepage = "http://www.spack.org"
+    url = "http://www.spack.org/downloads/aml-1.0.tar.gz"
+    has_code = False
+
+    version('0.1.0', sha256='cc89a8768693f1f11539378b21cdca9f0ce3fc5cb564f9b3e4154a051dcea69b')
+
+    def install(self, spec, prefix):
+        os.mkdir(os.path.join(prefix, 'bin'))
+        with open(os.path.join(prefix, 'bin', 'x'), 'wb') as f:
+            f.write(b'file')
diff --git a/var/spack/repos/builtin.mock/packages/view-dir-symlinked-dir/package.py b/var/spack/repos/builtin.mock/packages/view-dir-symlinked-dir/package.py
new file mode 100644
index 0000000000..ae0d90541b
--- /dev/null
+++ b/var/spack/repos/builtin.mock/packages/view-dir-symlinked-dir/package.py
@@ -0,0 +1,23 @@
+# Copyright 2013-2022 Lawrence Livermore National Security, LLC and other
+# Spack Project Developers. See the top-level COPYRIGHT file for details.
+#
+# SPDX-License-Identifier: (Apache-2.0 OR MIT)
+
+import os
+
+
+class ViewDirSymlinkedDir(Package):
+    """Installs <prefix>/bin/x/file_in_symlinked_dir where x -> y is a symlinked dir.
+    This should be mergeable with view-dir-dir, but not with view-dir-file."""
+    homepage = "http://www.spack.org"
+    url = "http://www.spack.org/downloads/aml-1.0.tar.gz"
+    has_code = False
+
+    version('0.1.0', sha256='cc89a8768693f1f11539378b21cdca9f0ce3fc5cb564f9b3e4154a051dcea69b')
+
+    def install(self, spec, prefix):
+        os.mkdir(os.path.join(prefix, 'bin'))
+        os.mkdir(os.path.join(prefix, 'bin', 'y'))
+        with open(os.path.join(prefix, 'bin', 'y', 'file_in_symlinked_dir'), 'wb') as f:
+            f.write(b'hello world')
+        os.symlink('y', os.path.join(prefix, 'bin', 'x'))
diff --git a/var/spack/repos/builtin/packages/python/package.py b/var/spack/repos/builtin/packages/python/package.py
index c97407d9fc..7f96898287 100644
--- a/var/spack/repos/builtin/packages/python/package.py
+++ b/var/spack/repos/builtin/packages/python/package.py
@@ -1365,7 +1365,7 @@ config.update(get_paths())
                                             self.spec
                                         ))
 
-    def add_files_to_view(self, view, merge_map):
+    def add_files_to_view(self, view, merge_map, skip_if_exists=False):
         bin_dir = self.spec.prefix.bin if sys.platform != 'win32'\
             else self.spec.prefix
         for src, dst in merge_map.items():