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 b' class BaseRepository(object):'
     EMPTY_COMMIT_ID = '0' * 40
 
     path = None
-    _remote = None
 
     def __init__(self, repo_path, config=None, create=False, **kwargs):
         """
@@ -222,6 +221,10 b' class BaseRepository(object):'
         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,8 +62,7 b' 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)
 
@@ -71,6 +70,10 b' class GitRepository(BaseRepository):'
         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,9 +77,7 b' 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)
 
@@ -87,6 +85,10 b' class MercurialRepository(BaseRepository'
         self._commit_ids = {}
 
     @LazyProperty
+    def _remote(self):
+        return connection.Hg(self.path, self.config, with_wire=self.with_wire)
+
+    @LazyProperty
     def commit_ids(self):
         """
         Returns list of commit ids, in ascending order.  Being lazy

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

@@ -72,11 +72,13 b' class SubversionRepository(base.BaseRepo'
                  **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 b' class in a way that is compatible with g'
 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 b' 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 b' 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