env: synchronize updates to environments (spack#14621)

Updates to environments were not multi-process safe, which prevented them from taking advantage of parallel builds as implemented in spack#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.

Author: scheibelp

lib/spack/spack/cmd/add.py

@@ -25,10 +25,11 @@ def setup_parser(subparser):
 def add(parser, args):
     env = ev.get_env(args, 'add', required=True)
 
-    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}"
-                    .format(spec.name, env.name))
-        else:
-            tty.msg('Adding %s to environment %s' % (spec, env.name))
-    env.write()
+    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}"
+                        .format(spec.name, env.name))
+            else:
+                tty.msg('Adding %s to environment %s' % (spec, env.name))
+        env.write()

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)
-    concretized_specs = env.concretize(force=args.force)
-    ev.display_specs(concretized_specs)
-    env.write()
+    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)
-            env.write()
+            with env.write_transaction():
+                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,16 +264,20 @@ def install(parser, args, **kwargs):
         env = ev.get_env(args, 'install')
         if env:
             if not args.only_concrete:
-                concretized_specs = env.concretize()
-                ev.display_specs(concretized_specs)
+                with env.write_transaction():
+                    concretized_specs = env.concretize()
+                    ev.display_specs(concretized_specs)
 
-                # save view regeneration for later, so that we only do it
-                # once, as it can be slow.
-                env.write(regenerate_views=False)
+                    # save view regeneration for later, so that we only do it
+                    # once, as it can be slow.
+                    env.write(regenerate_views=False)
 
             tty.msg("Installing environment %s" % env.name)
             env.install_all(args)
-            env.regenerate_views()
+            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:
             tty.die("install requires a package argument or a spack.yaml file")

lib/spack/spack/cmd/remove.py

@@ -31,10 +31,11 @@ def setup_parser(subparser):
 def remove(parser, args):
     env = ev.get_env(args, 'remove', required=True)
 
-    if args.all:
-        env.clear()
-    else:
-        for spec in spack.cmd.parse_specs(args.specs):
-            tty.msg('Removing %s from environment %s' % (spec, env.name))
-            env.remove(spec, args.list_name, force=args.force)
-    env.write()
+    with env.write_transaction():
+        if args.all:
+            env.clear()
+        else:
+            for spec in spack.cmd.parse_specs(args.specs):
+                tty.msg('Removing %s from environment %s' % (spec, env.name))
+                env.remove(spec, args.list_name, force=args.force)
+        env.write()

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,26 +575,7 @@ def __init__(self, path, init_file=None, with_view=None):
                 else:
                     self._read_manifest(f, raw_yaml=default_manifest_yaml)
         else:
-            default_manifest = not os.path.exists(self.manifest_path)
-            if default_manifest:
-                # No manifest, use default yaml
-                self._read_manifest(default_manifest_yaml)
-            else:
-                with open(self.manifest_path) as f:
-                    self._read_manifest(f)
-
-            if os.path.exists(self.lock_path):
-                with open(self.lock_path) as f:
-                    read_lock_version = self._read_lockfile(f)
-                if default_manifest:
-                    # No manifest, set user specs from lockfile
-                    self._set_user_specs_from_lockfile()
-
-                if read_lock_version == 1:
-                    tty.debug(
-                        "Storing backup of old lockfile {0} at {1}".format(
-                            self.lock_path, self._lock_backup_v1_path))
-                    shutil.copy(self.lock_path, self._lock_backup_v1_path)
+            self._read()
 
         if with_view is False:
             self.views = {}
@@ -602,6 +587,42 @@ def __init__(self, path, init_file=None, with_view=None):
         # 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
+            self._read_manifest(default_manifest_yaml)
+        else:
+            with open(self.manifest_path) as f:
+                self._read_manifest(f)
+
+        if os.path.exists(self.lock_path):
+            with open(self.lock_path) as f:
+                read_lock_version = self._read_lockfile(f)
+            if default_manifest:
+                # No manifest, set user specs from lockfile
+                self._set_user_specs_from_lockfile()
+
+            if read_lock_version == 1:
+                tty.debug(
+                    "Storing backup of old lockfile {0} at {1}".format(
+                        self.lock_path, self._lock_backup_v1_path))
+                shutil.copy(self.lock_path, self._lock_backup_v1_path)
+
+    def write_transaction(self):
+        """Get a write lock context manager for use in a `with` block."""
+        return lk.WriteTransaction(self.txlock, acquire=self._re_read)
+
     def _read_manifest(self, f, raw_yaml=None):
         """Read manifest file and set up user specs."""
         if raw_yaml:
@@ -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):
-        """Install a single spec into an environment.
+    def concretize_and_add(self, user_spec, concrete_spec=None):
+        """Concretize and add a single spec to the environment.
 
-        This will automatically concretize the single spec, but it won't
-        affect other as-yet unconcretized specs.
+        Concretize the provided ``user_spec`` and add it along with the
+        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,37 +1036,21 @@ 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()
+        if self.add(spec):
+            concrete = concrete_spec or spec.concretized()
+            self._add_concrete_spec(spec, concrete)
+        else:
+            # spec might be in the user_specs, but not installed.
+            # TODO: Redo name-based comparison for old style envs
+            spec = next(
+                s for s in self.user_specs if s.satisfies(user_spec)
+            )
+            concrete = self.specs_by_hash.get(spec.build_hash())
+            if not concrete:
+                concrete = spec.concretized()
                 self._add_concrete_spec(spec, concrete)
-            else:
-                # spec might be in the user_specs, but not installed.
-                # TODO: Redo name-based comparison for old style envs
-                spec = next(
-                    s for s in self.user_specs if s.satisfies(user_spec)
-                )
-                concrete = self.specs_by_hash.get(spec.build_hash())
-                if not concrete:
-                    concrete = spec.concretized()
-                    self._add_concrete_spec(spec, concrete)
 
-            self._install(concrete, **install_args)
-
-    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)
+        return concrete
 
     @property
     def default_view(self):
@@ -1131,32 +1150,61 @@ 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.
 
         Note: this does not regenerate the views for the environment;
         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()
+            if args:
+                spack.cmd.install.update_kwargs_from_args(args, kwargs)
 
-                # Parse cli arguments and construct a dictionary
-                # that will be passed to Package.do_install API
-                kwargs = dict()
-                if args:
-                    spack.cmd.install.update_kwargs_from_args(args, kwargs)
-
-                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)
+            self._install(spec, **kwargs)
 
     def all_specs_by_hash(self):
         """Map of hashes to spec for all specs in this environment."""