caches: make gevent curl connection cache friendly.

- gevent threads needs it's own cache, and connection
- added strict detection of cache problem in gevent curl

rhodecode/lib/vcs/backends/base.py

@@ -176,7 +176,6 @@ class BaseRepository(object):
     EMPTY_COMMIT_ID = '0' * 40
 
     path = None
-    _remote = None
 
     def __init__(self, repo_path, config=None, create=False, **kwargs):
         """
@@ -221,6 +220,10 @@ class BaseRepository(object):
                 config.set(section, key, val)
         return config
 
+    @LazyProperty
+    def _remote(self):
+        raise NotImplementedError
+
     @LazyProperty
     def EMPTY_COMMIT(self):
         return EmptyCommit(self.EMPTY_COMMIT_ID)

rhodecode/lib/vcs/backends/git/repository.py

@@ -62,14 +62,17 @@ class GitRepository(BaseRepository):
 
         self.path = safe_str(os.path.abspath(repo_path))
         self.config = config if config else self.get_default_config()
-        self._remote = connection.Git(
-            self.path, self.config, with_wire=with_wire)
+        self.with_wire = with_wire
 
         self._init_repo(create, src_url, update_after_clone, bare)
 
         # caches
         self._commit_ids = {}
 
+    @LazyProperty
+    def _remote(self):
+        return connection.Git(self.path, self.config, with_wire=self.with_wire)
+
     @LazyProperty
     def bare(self):
         return self._remote.bare()

rhodecode/lib/vcs/backends/hg/repository.py

@@ -77,15 +77,17 @@ class MercurialRepository(BaseRepository):
         # special requirements
         self.config = config if config else self.get_default_config(
             default=[('extensions', 'largefiles', '1')])
-
-        self._remote = connection.Hg(
-            self.path, self.config, with_wire=with_wire)
+        self.with_wire = with_wire
 
         self._init_repo(create, src_url, update_after_clone)
 
         # caches
         self._commit_ids = {}
 
+    @LazyProperty
+    def _remote(self):
+        return connection.Hg(self.path, self.config, with_wire=self.with_wire)
+
     @LazyProperty
     def commit_ids(self):
         """

rhodecode/lib/vcs/backends/svn/repository.py

@@ -72,11 +72,13 @@ class SubversionRepository(base.BaseRepository):
                  **kwargs):
         self.path = safe_str(os.path.abspath(repo_path))
         self.config = config if config else self.get_default_config()
-        self._remote = connection.Svn(
-            self.path, self.config)
 
         self._init_repo(create, src_url)
 
+    @LazyProperty
+    def _remote(self):
+        return connection.Svn(self.path, self.config)
+
     def _init_repo(self, create, src_url):
         if create and os.path.exists(self.path):
             raise RepositoryError(

rhodecode/lib/vcs/geventcurl.py

@@ -27,6 +27,7 @@ class in a way that is compatible with gevent.
 import logging
 import gevent
 import pycurl
+import greenlet
 
 # Import everything from pycurl.
 # This allows us to use this module as a drop in replacement of pycurl.
@@ -230,6 +231,12 @@ class GeventCurl(object):
         This perform method is compatible with gevent because it uses gevent
         synchronization mechanisms to wait for the request to finish.
         """
+        if getattr(self._curl, 'waiter', None) is not None:
+            current = greenlet.getcurrent()
+            msg = 'This curl object is already used by another greenlet, {}, \n' \
+                  'this is {}'.format(self._curl.waiter, current)
+            raise Exception(msg)
+
         waiter = self._curl.waiter = Waiter()
         try:
             self._multi.add_handle(self._curl)

rhodecode/model/db.py

@@ -2339,8 +2339,11 @@ class Repository(Base, BaseModel):
         def get_instance_cached(repo_id):
             return self._get_instance()
 
+        # we must use thread scoped cache here,
+        # because each thread of gevent needs it's own connection and cache
         inv_context_manager = rc_cache.InvalidationContext(
-            uid=cache_namespace_uid, invalidation_namespace=invalidation_namespace)
+            uid=cache_namespace_uid, invalidation_namespace=invalidation_namespace,
+            thread_scoped=True)
         with inv_context_manager as invalidation_context:
             args = (self.repo_id,)
             # re-compute and store cache if we get invalidate signal