Switch to multithreading, lock around layer pushes (#2)

Author: omesser

clients/logging/__init__.py

@@ -115,7 +115,7 @@ def log_and_raise(self, severity, error_msg, *args, **kwargs):
 
         # format the exception into the raised error message if we got one
         if 'exc' in kwargs:
-            error_msg = '{0}: {1}'.format(error_msg, kwargs['exc'].lower())
+            error_msg = '{0}: {1}'.format(error_msg, kwargs['exc'])
 
         exception_type = kwargs.get('exc_type', RuntimeError)
 

core/processor.py

@@ -64,8 +64,9 @@ def process(self):
             manifest = self._get_manifest(tmp_dir_name)
             self._logger.debug('Extracted archive manifest', manifest=manifest)
 
-            # prepare proc pool, note tarfile is not thread safe https://bugs.python.org/issue23649
-            with multiprocessing.pool.Pool(processes=self._parallel) as pool:
+            # prepare thread pool, note tarfile is not thread safe https://bugs.python.org/issue23649
+            # so if full extraction is not done beforehand, this is not safe
+            with multiprocessing.pool.ThreadPool(processes=self._parallel) as pool:
                 for image_config in manifest:
                     res = pool.apply_async(
                         process_image,

core/registry.py

@@ -5,6 +5,7 @@
 import hashlib
 import urllib.parse
 import time
+import threading
 
 import humanfriendly
 import requests
@@ -41,6 +42,8 @@ def __init__(
         if self._login:
             self._basicauth = requests.auth.HTTPBasicAuth(self._login, self._password)
 
+        self._layer_locks = {}
+
         self._logger.debug(
             'Initialized',
             registry_url=self._registry_url,
@@ -78,10 +81,7 @@ def process_image(self, tmp_dir_name, image_config):
             # push individual image layers
             layers = image_config["Layers"]
             for layer in layers:
-                self._logger.info('Pushing layer', layer=layer)
-                push_url = self._initialize_push(image)
-                layer_path = os.path.join(tmp_dir_name, layer)
-                self._push_layer(layer_path, push_url)
+                self._process_layer(layer, image, tmp_dir_name)
 
             # then, push image config
             self._logger.info(
@@ -150,6 +150,24 @@ def _push_manifest(self, manifest, image, tag):
                 content=response.content,
             )
 
+    def _process_layer(self, layer, image, tmp_dir_name):
+
+        # isolate layer key
+        layer_key = os.path.dirname(layer)
+
+        # pushing the layer in parallel from different images might result in 500 internal server error
+        self._logger.debug('Acquiring layer lock', layer_key=layer_key)
+        self._layer_locks.setdefault(layer_key, threading.Lock())
+        self._layer_locks[layer_key].acquire()
+        try:
+            self._logger.info('Pushing layer', layer=layer)
+            push_url = self._initialize_push(image)
+            layer_path = os.path.join(tmp_dir_name, layer)
+            self._push_layer(layer_path, push_url)
+        finally:
+            self._logger.debug('Releasing layer lock', layer_key=layer_key)
+            self._layer_locks[layer_key].release()
+
     def _initialize_push(self, repository):
         """
         Request a push URL for the image repository for a layer or manifest

dockerregistrypusher.py

@@ -63,7 +63,7 @@ def register_arguments(parser):
     parser.add_argument(
         '-p',
         '--parallel',
-        help='Control parallelism (multi-processing)',
+        help='Control parallelism (threads)',
         type=int,
         default=1,
     )