env: synchronize updates to environments (#14621)

Updates to environments were not multi-process safe, which prevented them from taking advantage of parallel builds as implemented in #13100.  This is a minimal set of changes to enable `spack install` in an environment to be parallelized:

- [x] add an internal lock, stored in the `.spack-env` directory, 
      to synchronize updates to `spack.yaml` and `spack.lock`
- [x] add `Environment.write_transaction` interface for this lock
- [x] makes use of `Environment.write_transaction` in `install`, 
      `add`, and `remove` commands

- `uninstall` is not synchronized yet; that is left for a future PR.

lib/spack/spack/cmd/add.py

@@ -25,6 +25,7 @@ def setup_parser(subparser):
 def add(parser, args):
     env = ev.get_env(args, 'add', required=True)
 
+    with env.write_transaction():
         for spec in spack.cmd.parse_specs(args.specs):
             if not env.add(spec, args.list_name):
                 tty.msg("Package {0} was already added to {1}"

lib/spack/spack/cmd/concretize.py

@@ -18,6 +18,7 @@ def setup_parser(subparser):
 
 def concretize(parser, args):
     env = ev.get_env(args, 'concretize', required=True)
+    with env.write_transaction():
         concretized_specs = env.concretize(force=args.force)
         ev.display_specs(concretized_specs)
         env.write()

lib/spack/spack/cmd/install.py

@@ -223,8 +223,13 @@ def install_spec(cli_args, kwargs, abstract_spec, spec):
         # handle active environment, if any
         env = ev.get_env(cli_args, 'install')
         if env:
-            env.install(abstract_spec, spec, **kwargs)
+            with env.write_transaction():
-            env.write()
+                concrete = env.concretize_and_add(
+                    abstract_spec, spec)
+                env.write(regenerate_views=False)
+            env._install(concrete, **kwargs)
+            with env.write_transaction():
+                env.regenerate_views()
         else:
             spec.package.do_install(**kwargs)
 
@@ -259,6 +264,7 @@ def install(parser, args, **kwargs):
         env = ev.get_env(args, 'install')
         if env:
             if not args.only_concrete:
+                with env.write_transaction():
                     concretized_specs = env.concretize()
                     ev.display_specs(concretized_specs)
 
@@ -268,6 +274,9 @@ def install(parser, args, **kwargs):
 
             tty.msg("Installing environment %s" % env.name)
             env.install_all(args)
+            with env.write_transaction():
+                # It is not strictly required to synchronize view regeneration
+                # but doing so can prevent redundant work in the filesystem.
                 env.regenerate_views()
             return
         else:

lib/spack/spack/cmd/remove.py

@@ -31,6 +31,7 @@ def setup_parser(subparser):
 def remove(parser, args):
     env = ev.get_env(args, 'remove', required=True)
 
+    with env.write_transaction():
         if args.all:
             env.clear()
         else:

lib/spack/spack/environment.py

@@ -36,6 +36,7 @@
 from spack.spec import Spec
 from spack.spec_list import SpecList, InvalidSpecConstraintError
 from spack.variant import UnknownVariantError
+import spack.util.lock as lk
 
 #: environment variable used to indicate the active environment
 spack_env_var = 'SPACK_ENV'
@@ -557,6 +558,9 @@ def __init__(self, path, init_file=None, with_view=None):
                 path to the view.
         """
         self.path = os.path.abspath(path)
+
+        self.txlock = lk.Lock(self._transaction_lock_path)
+
         # This attribute will be set properly from configuration
         # during concretization
         self.concretization = None
@@ -571,6 +575,29 @@ def __init__(self, path, init_file=None, with_view=None):
                 else:
                     self._read_manifest(f, raw_yaml=default_manifest_yaml)
         else:
+            self._read()
+
+        if with_view is False:
+            self.views = {}
+        elif with_view is True:
+            self.views = {
+                default_view_name: ViewDescriptor(self.view_path_default)}
+        elif isinstance(with_view, six.string_types):
+            self.views = {default_view_name: ViewDescriptor(with_view)}
+        # If with_view is None, then defer to the view settings determined by
+        # the manifest file
+
+    def _re_read(self):
+        """Reinitialize the environment object if it has been written (this
+           may not be true if the environment was just created in this running
+           instance of Spack)."""
+        if not os.path.exists(self.manifest_path):
+            return
+
+        self.clear()
+        self._read()
+
+    def _read(self):
         default_manifest = not os.path.exists(self.manifest_path)
         if default_manifest:
             # No manifest, use default yaml
@@ -592,15 +619,9 @@ def __init__(self, path, init_file=None, with_view=None):
                         self.lock_path, self._lock_backup_v1_path))
                 shutil.copy(self.lock_path, self._lock_backup_v1_path)
 
-        if with_view is False:
+    def write_transaction(self):
-            self.views = {}
+        """Get a write lock context manager for use in a `with` block."""
-        elif with_view is True:
+        return lk.WriteTransaction(self.txlock, acquire=self._re_read)
-            self.views = {
-                default_view_name: ViewDescriptor(self.view_path_default)}
-        elif isinstance(with_view, six.string_types):
-            self.views = {default_view_name: ViewDescriptor(with_view)}
-        # If with_view is None, then defer to the view settings determined by
-        # the manifest file
 
     def _read_manifest(self, f, raw_yaml=None):
         """Read manifest file and set up user specs."""
@@ -694,6 +715,13 @@ def manifest_path(self):
         """Path to spack.yaml file in this environment."""
         return os.path.join(self.path, manifest_name)
 
+    @property
+    def _transaction_lock_path(self):
+        """The location of the lock file used to synchronize multiple
+        processes updating the same environment.
+        """
+        return os.path.join(self.path, 'transaction_lock')
+
     @property
     def lock_path(self):
         """Path to spack.lock file in this environment."""
@@ -986,11 +1014,18 @@ def _concretize_separately(self):
                 concretized_specs.append((uspec, concrete))
         return concretized_specs
 
-    def install(self, user_spec, concrete_spec=None, **install_args):
+    def concretize_and_add(self, user_spec, concrete_spec=None):
-        """Install a single spec into an environment.
+        """Concretize and add a single spec to the environment.
 
-        This will automatically concretize the single spec, but it won't
+        Concretize the provided ``user_spec`` and add it along with the
-        affect other as-yet unconcretized specs.
+        concretized result to the environment. If the given ``user_spec`` was
+        already present in the environment, this does not add a duplicate.
+        The concretized spec will be added unless the ``user_spec`` was
+        already present and an associated concrete spec was already present.
+
+        Args:
+            concrete_spec: if provided, then it is assumed that it is the
+                result of concretizing the provided ``user_spec``
         """
         if self.concretization == 'together':
             msg = 'cannot install a single spec in an environment that is ' \
@@ -1001,7 +1036,6 @@ def install(self, user_spec, concrete_spec=None, **install_args):
 
         spec = Spec(user_spec)
 
-        with spack.store.db.read_transaction():
         if self.add(spec):
             concrete = concrete_spec or spec.concretized()
             self._add_concrete_spec(spec, concrete)
@@ -1016,22 +1050,7 @@ def install(self, user_spec, concrete_spec=None, **install_args):
                 concrete = spec.concretized()
                 self._add_concrete_spec(spec, concrete)
 
-            self._install(concrete, **install_args)
+        return concrete
-
-    def _install(self, spec, **install_args):
-        spec.package.do_install(**install_args)
-
-        # Make sure log directory exists
-        log_path = self.log_path
-        fs.mkdirp(log_path)
-
-        with fs.working_dir(self.path):
-            # Link the resulting log file into logs dir
-            build_log_link = os.path.join(
-                log_path, '%s-%s.log' % (spec.name, spec.dag_hash(7)))
-            if os.path.lexists(build_log_link):
-                os.remove(build_log_link)
-            os.symlink(spec.package.build_log_path, build_log_link)
 
     @property
     def default_view(self):
@@ -1131,6 +1150,33 @@ def _add_concrete_spec(self, spec, concrete, new=True):
         self.concretized_order.append(h)
         self.specs_by_hash[h] = concrete
 
+    def install(self, user_spec, concrete_spec=None, **install_args):
+        """Install a single spec into an environment.
+
+        This will automatically concretize the single spec, but it won't
+        affect other as-yet unconcretized specs.
+        """
+        concrete = self.concretize_and_add(user_spec, concrete_spec)
+
+        self._install(concrete, **install_args)
+
+    def _install(self, spec, **install_args):
+        # "spec" must be concrete
+        spec.package.do_install(**install_args)
+
+        if not spec.external:
+            # Make sure log directory exists
+            log_path = self.log_path
+            fs.mkdirp(log_path)
+
+            with fs.working_dir(self.path):
+                # Link the resulting log file into logs dir
+                build_log_link = os.path.join(
+                    log_path, '%s-%s.log' % (spec.name, spec.dag_hash(7)))
+                if os.path.lexists(build_log_link):
+                    os.remove(build_log_link)
+                os.symlink(spec.package.build_log_path, build_log_link)
+
     def install_all(self, args=None):
         """Install all concretized specs in an environment.
 
@@ -1138,10 +1184,20 @@ def install_all(self, args=None):
         that needs to be done separately with a call to write().
 
         """
+
+        # If "spack install" is invoked repeatedly for a large environment
+        # where all specs are already installed, the operation can take
+        # a large amount of time due to repeatedly acquiring and releasing
+        # locks, this does an initial check across all specs within a single
+        # DB read transaction to reduce time spent in this case.
+        uninstalled_specs = []
         with spack.store.db.read_transaction():
             for concretized_hash in self.concretized_order:
                 spec = self.specs_by_hash[concretized_hash]
+                if not spec.package.installed:
+                    uninstalled_specs.append(spec)
 
+        for spec in uninstalled_specs:
             # Parse cli arguments and construct a dictionary
             # that will be passed to Package.do_install API
             kwargs = dict()
@@ -1150,14 +1206,6 @@ def install_all(self, args=None):
 
             self._install(spec, **kwargs)
 
-                if not spec.external:
-                    # Link the resulting log file into logs dir
-                    log_name = '%s-%s' % (spec.name, spec.dag_hash(7))
-                    build_log_link = os.path.join(self.log_path, log_name)
-                    if os.path.lexists(build_log_link):
-                        os.remove(build_log_link)
-                    os.symlink(spec.package.build_log_path, build_log_link)
-
     def all_specs_by_hash(self):
         """Map of hashes to spec for all specs in this environment."""
         # Note this uses dag-hashes calculated without build deps as keys,