Add new initial command stage to mutex all reads of the installspace.

catkin_tools/execution/executor.py

@@ -59,9 +59,6 @@ def async_job(verb, job, threadpool, locks, event_queue, log_path):
     # Jobs start occuping a jobserver job
     occupying_job = True
 
-    # Load environment for this job
-    job_env = job.getenv(os.environ)
-
     # Execute each stage of this job
     for stage in job.stages:
         # Logger reference in this scope for error reporting
@@ -102,7 +99,7 @@ def async_job(verb, job, threadpool, locks, event_queue, log_path):
                     while True:
                         try:
                             # Update the environment for this stage (respects overrides)
-                            stage.update_env(job_env)
+                            stage.update_env(job.env)
 
                             # Get the logger
                             protocol_type = stage.logger_factory(verb, job.jid, stage.label, event_queue, log_path)

catkin_tools/execution/jobs.py

@@ -24,7 +24,7 @@ class Job(object):
 
     """A Job is a series of operations, each of which is considered a "stage" of the job."""
 
-    def __init__(self, jid, deps, env_loader, stages, continue_on_failure=True):
+    def __init__(self, jid, deps, env, stages, continue_on_failure=True):
         """
         jid: Unique job identifier
         deps: Dependencies (in terms of other jid's)
@@ -33,7 +33,7 @@ def __init__(self, jid, deps, env_loader, stages, continue_on_failure=True):
         """
         self.jid = jid
         self.deps = deps
-        self.env_loader = env_loader
+        self.env = env
         self.stages = stages
         self.continue_on_failure = continue_on_failure
 
@@ -48,6 +48,3 @@ def all_deps_succeeded(self, completed_jobs):
     def any_deps_failed(self, completed_jobs):
         """Return True if any dependencies which have been completed have failed."""
         return any([not completed_jobs.get(dep_id, True) for dep_id in self.deps])
-
-    def getenv(self, env):
-        return self.env_loader(env)

catkin_tools/jobs/catkin.py

@@ -35,7 +35,7 @@
 from .commands.make import MAKE_EXEC
 
 from .utils import copyfiles
-from .utils import get_env_loader
+from .utils import load_env
 from .utils import makedirs
 from .utils import rmfiles
 
@@ -350,10 +350,23 @@ def create_catkin_build_job(context, package, package_path, dependencies, force_
     install_space = context.package_install_space(package)
     # Package metadata path
     metadata_path = context.package_metadata_path(package)
+    # Environment dictionary for the job, which will be built
+    # up by the executions in the getenv stage.
+    job_env = dict(os.environ)
 
     # Create job stages
     stages = []
 
+    # Get environment for job.
+    stages.append(FunctionStage(
+        'getenv',
+        load_env,
+        locked_resource='installspace',
+        job_env=job_env,
+        package=package,
+        context=context
+    ))
+
     # Create package build space
     stages.append(FunctionStage(
         'mkdir',
@@ -478,7 +491,7 @@ def create_catkin_build_job(context, package, package_path, dependencies, force_
     return Job(
         jid=package.name,
         deps=dependencies,
-        env_loader=get_env_loader(package, context),
+        env=job_env,
         stages=stages)
 
 
@@ -493,12 +506,26 @@ def create_catkin_clean_job(
         clean_install):
     """Generate a Job that cleans a catkin package"""
 
-    stages = []
-
     # Package build space path
     build_space = context.package_build_space(package)
     # Package metadata path
     metadata_path = context.package_metadata_path(package)
+    # Environment dictionary for the job, which will be built
+    # up by the executions in the getenv stage.
+    job_env = dict(os.environ)
+
+    # Create job stages
+    stages = []
+
+    # Get environment for job.
+    stages.append(FunctionStage(
+        'getenv',
+        load_env,
+        locked_resource='installspace',
+        job_env=job_env,
+        package=package,
+        context=context
+    ))
 
     # Remove installed files
     if clean_install:
@@ -566,7 +593,7 @@ def create_catkin_clean_job(
     return Job(
         jid=package.name,
         deps=dependencies,
-        env_loader=get_env_loader(package, context),
+        env=job_env,
         stages=stages)
 
 

catkin_tools/jobs/cmake.py

@@ -31,7 +31,7 @@
 from .commands.make import MAKE_EXEC
 
 from .utils import copyfiles
-from .utils import get_env_loader
+from .utils import load_env
 from .utils import makedirs
 from .utils import rmfiles
 
@@ -217,6 +217,9 @@ def create_cmake_build_job(context, package, package_path, dependencies, force_c
     build_space = context.package_build_space(package)
     # Package metadata path
     metadata_path = context.package_metadata_path(package)
+    # Environment dictionary for the job, which will be built
+    # up by the executions in the getenv stage.
+    job_env = dict(os.environ)
 
     # Get actual staging path
     dest_path = context.package_dest_path(package)
@@ -225,6 +228,16 @@ def create_cmake_build_job(context, package, package_path, dependencies, force_c
     # Create job stages
     stages = []
 
+    # Get environment for job.
+    stages.append(FunctionStage(
+        'getenv',
+        load_env,
+        locked_resource='installspace',
+        job_env=job_env,
+        package=package,
+        context=context
+    ))
+
     # Create package build space
     stages.append(FunctionStage(
         'mkdir',
@@ -321,7 +334,7 @@ def create_cmake_build_job(context, package, package_path, dependencies, force_c
     return Job(
         jid=package.name,
         deps=dependencies,
-        env_loader=get_env_loader(package, context),
+        env=job_env,
         stages=stages)
 
 
@@ -340,9 +353,24 @@ def create_cmake_clean_job(
     build_space = context.package_build_space(package)
     # Package metadata path
     metadata_path = context.package_metadata_path(package)
+    # Environment dictionary for the job, which will be built
+    # up by the executions in the getenv stage.
+    job_env = dict(os.environ)
 
+    # Create job stages
     stages = []
 
+    # Get environment for job.
+    stages.append(FunctionStage(
+        'getenv',
+        load_env,
+        locked_resource='installspace',
+        job_env=job_env,
+        package=package,
+        context=context
+    ))
+
+    # Remove installed files
     if clean_install and context.install:
         installed_files = get_installed_files(context.package_metadata_path(package))
         stages.append(FunctionStage(
@@ -381,7 +409,7 @@ def create_cmake_clean_job(
     return Job(
         jid=package.name,
         deps=dependencies,
-        env_loader=get_env_loader(package, context),
+        env=job_env,
         stages=stages)
 
 

catkin_tools/jobs/utils.py

@@ -50,32 +50,25 @@ def get_env_loaders(package, context):
     return sources
 
 
-def get_env_loader(package, context):
-    """This function returns a function object which extends a base environment
-    based on a set of environments to load."""
-
-    def load_env(base_env):
-        # Copy the base environment to extend
-        job_env = dict(base_env)
-        # Get the paths to the env loaders
-        env_loader_paths = get_env_loaders(package, context)
-        # If DESTDIR is set, set _CATKIN_SETUP_DIR as well
-        if context.destdir is not None:
-            job_env['_CATKIN_SETUP_DIR'] = context.package_dest_path(package)
-
-        for env_loader_path in env_loader_paths:
-            # print(' - Loading resultspace env from: {}'.format(env_loader_path))
-            resultspace_env = get_resultspace_environment(
-                os.path.split(env_loader_path)[0],
-                base_env=job_env,
-                quiet=True,
-                cached=context.use_env_cache,
-                strict=False)
-            job_env.update(resultspace_env)
-
-        return job_env
-
-    return load_env
+def load_env(logger, event_queue, job_env, package, context):
+    # Get the paths to the env loaders
+    env_loader_paths = get_env_loaders(package, context)
+    # If DESTDIR is set, set _CATKIN_SETUP_DIR as well
+    if context.destdir is not None:
+        job_env['_CATKIN_SETUP_DIR'] = context.package_dest_path(package)
+
+    for env_loader_path in env_loader_paths:
+        if logger:
+            logger.out('Loading environment from: {}'.format(env_loader_path))
+        resultspace_env = get_resultspace_environment(
+            os.path.split(env_loader_path)[0],
+            base_env=job_env,
+            quiet=True,
+            cached=context.use_env_cache,
+            strict=False)
+        job_env.update(resultspace_env)
+
+    return 0
 
 
 def makedirs(logger, event_queue, path):

catkin_tools/verbs/catkin_build/cli.py

@@ -48,7 +48,7 @@
 
 import catkin_tools.execution.job_server as job_server
 
-from catkin_tools.jobs.utils import get_env_loader
+from catkin_tools.jobs.utils import load_env
 
 from catkin_tools.metadata import find_enclosing_workspace
 from catkin_tools.metadata import get_metadata
@@ -228,7 +228,8 @@ def print_build_env(context, package_name):
     # Load the environment used by this package for building
     for pth, pkg in workspace_packages.items():
         if pkg.name == package_name:
-            environ = get_env_loader(pkg, context)(os.environ)
+            environ = dict(os.environ)
+            load_env(None, None, environ, pkg, context)
             print(format_env_dict(environ))
             return 0
     print('[build] Error: Package `{}` not in workspace.'.format(package_name),