framework: Lock files which could be read/written to concurrently

Add file locking to files that could be read and written to concurrently
by separate wa processes causing race conditions.

wa/framework/getters.py

@@ -31,7 +31,7 @@ import requests
 from wa import Parameter, settings, __file__ as _base_filepath
 from wa.framework.resource import ResourceGetter, SourcePriority, NO_ONE
 from wa.framework.exception import ResourceError
-from wa.utils.misc import (ensure_directory_exists as _d,
+from wa.utils.misc import (ensure_directory_exists as _d, lock_file,
                            ensure_file_directory_exists as _f, sha256, urljoin)
 from wa.utils.types import boolean, caseless_string
 
@@ -254,22 +254,23 @@ class Http(ResourceGetter):
         url = urljoin(self.url, owner_name, asset['path'])
         local_path = _f(os.path.join(settings.dependencies_directory, '__remote',
                                      owner_name, asset['path'].replace('/', os.sep)))
-        if os.path.exists(local_path) and not self.always_fetch:
-            local_sha = sha256(local_path)
-            if local_sha == asset['sha256']:
-                self.logger.debug('Local SHA256 matches; not re-downloading')
-                return local_path
-        self.logger.debug('Downloading {}'.format(url))
-        response = self.geturl(url, stream=True)
-        if response.status_code != http.client.OK:
-            message = 'Could not download asset "{}"; recieved "{} {}"'
-            self.logger.warning(message.format(url,
-                                               response.status_code,
-                                               response.reason))
-            return
-        with open(local_path, 'wb') as wfh:
-            for chunk in response.iter_content(chunk_size=self.chunk_size):
-                wfh.write(chunk)
+        with lock_file(local_path, timeout=5 * 60):
+            if os.path.exists(local_path) and not self.always_fetch:
+                local_sha = sha256(local_path)
+                if local_sha == asset['sha256']:
+                    self.logger.debug('Local SHA256 matches; not re-downloading')
+                    return local_path
+            self.logger.debug('Downloading {}'.format(url))
+            response = self.geturl(url, stream=True)
+            if response.status_code != http.client.OK:
+                message = 'Could not download asset "{}"; recieved "{} {}"'
+                self.logger.warning(message.format(url,
+                                                   response.status_code,
+                                                   response.reason))
+                return
+            with open(local_path, 'wb') as wfh:
+                for chunk in response.iter_content(chunk_size=self.chunk_size):
+                    wfh.write(chunk)
         return local_path
 
     def geturl(self, url, stream=False):

wa/framework/resource.py

@@ -25,6 +25,7 @@ from wa.framework.configuration import settings
 from wa.utils import log
 from wa.utils.misc import get_object_name
 from wa.utils.types import enum, list_or_string, prioritylist, version_tuple
+from wa.utils.misc import lock_file
 
 
 SourcePriority = enum(['package', 'remote', 'lan', 'local',
@@ -280,7 +281,8 @@ class ResourceResolver(object):
 
 def apk_version_matches(path, version):
     version = list_or_string(version)
-    info = ApkInfo(path)
+    with lock_file(path):
+        info = ApkInfo(path)
     for v in version:
         if info.version_name == v or info.version_code == v:
             return True
@@ -290,7 +292,8 @@ def apk_version_matches(path, version):
 
 
 def apk_version_matches_range(path, min_version=None, max_version=None):
-    info = ApkInfo(path)
+    with lock_file(path):
+        info = ApkInfo(path)
     return range_version_matching(info.version_name, min_version, max_version)
 
 
@@ -333,18 +336,21 @@ def file_name_matches(path, pattern):
 
 
 def uiauto_test_matches(path, uiauto):
-    info = ApkInfo(path)
+    with lock_file(path):
+        info = ApkInfo(path)
     return uiauto == ('com.arm.wa.uiauto' in info.package)
 
 
 def package_name_matches(path, package):
-    info = ApkInfo(path)
+    with lock_file(path):
+        info = ApkInfo(path)
     return info.package == package
 
 
 def apk_abi_matches(path, supported_abi, exact_abi=False):
     supported_abi = list_or_string(supported_abi)
-    info = ApkInfo(path)
+    with lock_file(path):
+        info = ApkInfo(path)
     # If no native code present, suitable for all devices.
     if not info.native_code:
         return True

wa/framework/target/info.py

@@ -23,6 +23,7 @@ from devlib.utils.android import AndroidProperties
 from wa.framework.configuration.core import settings
 from wa.framework.exception import ConfigError
 from wa.utils.serializer import read_pod, write_pod, Podable
+from wa.utils.misc import lock_file
 
 
 def cpuinfo_from_pod(pod):
@@ -280,13 +281,15 @@ def read_target_info_cache():
         os.makedirs(settings.cache_directory)
     if not os.path.isfile(settings.target_info_cache_file):
         return {}
-    return read_pod(settings.target_info_cache_file)
+    with lock_file(settings.target_info_cache_file):
+        return read_pod(settings.target_info_cache_file)
 
 
 def write_target_info_cache(cache):
     if not os.path.exists(settings.cache_directory):
         os.makedirs(settings.cache_directory)
-    write_pod(cache, settings.target_info_cache_file)
+    with lock_file(settings.target_info_cache_file):
+        write_pod(cache, settings.target_info_cache_file)
 
 
 def get_target_info_from_cache(system_id):

wa/framework/workload.py

@@ -32,6 +32,7 @@ from wa.framework.exception import WorkloadError, ConfigError
 from wa.utils.types import ParameterDict, list_or_string, version_tuple
 from wa.utils.revent import ReventRecorder
 from wa.utils.exec_control import once_per_instance
+from wa.utils.misc import lock_file
 
 
 class Workload(TargetedPlugin):
@@ -731,31 +732,33 @@ class PackageHandler(object):
             raise WorkloadError(msg)
 
         self.error_msg = None
-        if self.prefer_host_package:
-            self.resolve_package_from_host(context)
-            if not self.apk_file:
-                self.resolve_package_from_target()
-        else:
-            self.resolve_package_from_target()
-            if not self.apk_file:
+        with lock_file(os.path.join(self.owner.dependencies_directory, self.owner.name)):
+            if self.prefer_host_package:
                 self.resolve_package_from_host(context)
-
-        if self.apk_file:
-            self.apk_info = ApkInfo(self.apk_file)
-        else:
-            if self.error_msg:
-                raise WorkloadError(self.error_msg)
+                if not self.apk_file:
+                    self.resolve_package_from_target()
             else:
-                if self.package_name:
-                    message = 'Package "{package}" not found for workload {name} '\
-                              'on host or target.'
-                elif self.version:
-                    message = 'No matching package found for workload {name} '\
-                              '(version {version}) on host or target.'
+                self.resolve_package_from_target()
+                if not self.apk_file:
+                    self.resolve_package_from_host(context)
+
+            if self.apk_file:
+                with lock_file(self.apk_file):
+                    self.apk_info = ApkInfo(self.apk_file)
+            else:
+                if self.error_msg:
+                    raise WorkloadError(self.error_msg)
                 else:
-                    message = 'No matching package found for workload {name} on host or target'
-                raise WorkloadError(message.format(name=self.owner.name, version=self.version,
-                                                   package=self.package_name))
+                    if self.package_name:
+                        message = 'Package "{package}" not found for workload {name} '\
+                                  'on host or target.'
+                    elif self.version:
+                        message = 'No matching package found for workload {name} '\
+                                  '(version {version}) on host or target.'
+                    else:
+                        message = 'No matching package found for workload {name} on host or target'
+                    raise WorkloadError(message.format(name=self.owner.name, version=self.version,
+                                                       package=self.package_name))
 
     def resolve_package_from_host(self, context):
         self.logger.debug('Resolving package on host system')
@@ -898,8 +901,9 @@ class PackageHandler(object):
         package_info = self.target.get_package_info(package)
         apk_name = self._get_package_name(package_info.apk_path)
         host_path = os.path.join(self.owner.dependencies_directory, apk_name)
-        self.target.pull(package_info.apk_path, host_path,
-                         timeout=self.install_timeout)
+        with lock_file(host_path, timeout=self.install_timeout):
+            self.target.pull(package_info.apk_path, host_path,
+                             timeout=self.install_timeout)
         return host_path
 
     def teardown(self):

wa/instruments/misc.py

@@ -38,7 +38,7 @@ from wa import Instrument, Parameter, very_fast
 from wa.framework.exception import ConfigError
 from wa.framework.instrument import slow
 from wa.utils.diff import diff_sysfs_dirs, diff_interrupt_files
-from wa.utils.misc import as_relative
+from wa.utils.misc import as_relative, lock_file
 from wa.utils.misc import ensure_file_directory_exists as _f
 from wa.utils.misc import ensure_directory_exists as _d
 from wa.utils.types import list_of_strings
@@ -244,7 +244,8 @@ class ApkVersion(Instrument):
 
     def setup(self, context):
         if hasattr(context.workload, 'apk_file'):
-            self.apk_info = ApkInfo(context.workload.apk_file)
+            with lock_file(context.workload.apk_file):
+                self.apk_info = ApkInfo(context.workload.apk_file)
         else:
             self.apk_info = None