remote: protect all remote client/session creation code with locks

With @cached_property last write wins, so all threads will use the same
session. However, several might be created concurrently wasting
resources and even potentially asking something from the user.

This change guarantees only one client instance will be created for a
remote.

Author: Suor

dvc/remote/azure.py

@@ -4,10 +4,10 @@
 import logging
 import os
 import re
-from datetime import datetime
-from datetime import timedelta
+from datetime import datetime, timedelta
+import threading
 
-from funcy import cached_property
+from funcy import cached_property, wrap_prop
 
 from dvc.config import Config
 from dvc.path_info import CloudURLInfo
@@ -64,6 +64,7 @@ def __init__(self, repo, config):
             else self.path_cls.from_parts(scheme=self.scheme, netloc=bucket)
         )
 
+    @wrap_prop(threading.Lock())
     @cached_property
     def blob_service(self):
         from azure.storage.blob import BlockBlobService

dvc/remote/gs.py

@@ -5,8 +5,9 @@
 from functools import wraps
 import io
 import os.path
+import threading
 
-from funcy import cached_property
+from funcy import cached_property, wrap_prop
 
 from dvc.config import Config
 from dvc.exceptions import DvcException
@@ -91,6 +92,7 @@ def __init__(self, repo, config):
         self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None)
         self.credentialpath = config.get(Config.SECTION_GCP_CREDENTIALPATH)
 
+    @wrap_prop(threading.Lock())
     @cached_property
     def gs(self):
         from google.cloud.storage import Client

dvc/remote/http.py

@@ -1,8 +1,9 @@
 from __future__ import unicode_literals
 
 import logging
+import threading
 
-from funcy import cached_property
+from funcy import cached_property, wrap_prop
 
 from dvc.config import Config
 from dvc.config import ConfigError
@@ -81,6 +82,7 @@ def get_file_checksum(self, path_info):
 
         return etag
 
+    @wrap_prop(threading.Lock())
     @cached_property
     def _session(self):
         import requests

dvc/remote/oss.py

@@ -3,6 +3,9 @@
 
 import logging
 import os
+import threading
+
+from funcy import cached_property, wrap_prop
 
 from dvc.config import Config
 from dvc.path_info import CloudURLInfo
@@ -61,30 +64,29 @@ def __init__(self, repo, config):
             or "defaultSecret"
         )
 
-        self._bucket = None
-
-    @property
+    @wrap_prop(threading.Lock())
+    @cached_property
     def oss_service(self):
         import oss2
 
-        if self._bucket is None:
-            logger.debug("URL {}".format(self.path_info))
-            logger.debug("key id {}".format(self.key_id))
-            logger.debug("key secret {}".format(self.key_secret))
-            auth = oss2.Auth(self.key_id, self.key_secret)
-            self._bucket = oss2.Bucket(
-                auth, self.endpoint, self.path_info.bucket
+        logger.debug("URL {}".format(self.path_info))
+        logger.debug("key id {}".format(self.key_id))
+        logger.debug("key secret {}".format(self.key_secret))
+
+        auth = oss2.Auth(self.key_id, self.key_secret)
+        bucket = oss2.Bucket(auth, self.endpoint, self.path_info.bucket)
+
+        # Ensure bucket exists
+        try:
+            bucket.get_bucket_info()
+        except oss2.exceptions.NoSuchBucket:
+            bucket.create_bucket(
+                oss2.BUCKET_ACL_PUBLIC_READ,
+                oss2.models.BucketCreateConfig(
+                    oss2.BUCKET_STORAGE_CLASS_STANDARD
+                ),
             )
-            try:  # verify that bucket exists
-                self._bucket.get_bucket_info()
-            except oss2.exceptions.NoSuchBucket:
-                self._bucket.create_bucket(
-                    oss2.BUCKET_ACL_PUBLIC_READ,
-                    oss2.models.BucketCreateConfig(
-                        oss2.BUCKET_STORAGE_CLASS_STANDARD
-                    ),
-                )
-        return self._bucket
+        return bucket
 
     def remove(self, path_info):
         if path_info.scheme != self.scheme:

dvc/remote/pool.py

@@ -1,7 +1,8 @@
 from collections import deque
 from contextlib import contextmanager
+import threading
 
-from funcy import memoize
+from funcy import memoize, wrap_with
 
 
 @contextmanager
@@ -17,6 +18,7 @@ def get_connection(conn_func, *args, **kwargs):
         pool.release(conn)
 
 
+@wrap_with(threading.Lock())
 @memoize
 def get_pool(conn_func, *args, **kwargs):
     return Pool(conn_func, *args, **kwargs)

dvc/remote/s3.py

@@ -3,8 +3,9 @@
 
 import logging
 import os
+import threading
 
-from funcy import cached_property
+from funcy import cached_property, wrap_prop
 
 from dvc.config import Config
 from dvc.exceptions import DvcException
@@ -56,6 +57,7 @@ def __init__(self, repo, config):
         if shared_creds:
             os.environ.setdefault("AWS_SHARED_CREDENTIALS_FILE", shared_creds)
 
+    @wrap_prop(threading.Lock())
     @cached_property
     def s3(self):
         import boto3

setup.py

@@ -74,7 +74,7 @@ def run(self):
     "humanize>=0.5.1",
     "PyYAML>=5.1.2",
     "ruamel.yaml>=0.16.1",
-    "funcy>=1.12",
+    "funcy>=1.14",
     "pathspec>=0.6.0",
     "shortuuid>=0.5.0",
     "tqdm>=4.38.0,<5",