pyro4: Create pyro proxy factory which returns instaces scoped by the request.

rhodecode/lib/vcs/__init__.py

@@ -1,278 +1,278 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Various version Control System version lib (vcs) management abstraction layer
 for Python. Build with server client architecture.
 """
 
 
 VERSION = (0, 5, 0, 'dev')
 
 __version__ = '.'.join((str(each) for each in VERSION[:4]))
 
 __all__ = [
     'get_version', 'get_repo', 'get_backend',
     'VCSError', 'RepositoryError', 'CommitError'
     ]
 
 import atexit
 import logging
 import subprocess
 import time
 import urlparse
 from cStringIO import StringIO
 
 import pycurl
 import Pyro4
 from Pyro4.errors import CommunicationError
 
 from rhodecode.lib.vcs.conf import settings
 from rhodecode.lib.vcs.backends import get_repo, get_backend
 from rhodecode.lib.vcs.exceptions import (
     VCSError, RepositoryError, CommitError)
 
 
 log = logging.getLogger(__name__)
 
 
 def get_version():
     """
     Returns shorter version (digit parts only) as string.
     """
     return '.'.join((str(each) for each in VERSION[:3]))
 
 
 def connect_pyro4(server_and_port):
     from rhodecode.lib.vcs import connection, client
     from rhodecode.lib.middleware.utils import scm_app
 
-    git_remote = client.ThreadlocalProxyFactory(
+    git_remote = client.RequestScopeProxyFactory(
         settings.pyro_remote(settings.PYRO_GIT, server_and_port))
-    hg_remote = client.ThreadlocalProxyFactory(
+    hg_remote = client.RequestScopeProxyFactory(
         settings.pyro_remote(settings.PYRO_HG, server_and_port))
-    svn_remote = client.ThreadlocalProxyFactory(
+    svn_remote = client.RequestScopeProxyFactory(
         settings.pyro_remote(settings.PYRO_SVN, server_and_port))
 
     connection.Git = client.RepoMaker(proxy_factory=git_remote)
     connection.Hg = client.RepoMaker(proxy_factory=hg_remote)
     connection.Svn = client.RepoMaker(proxy_factory=svn_remote)
 
     scm_app.GIT_REMOTE_WSGI = Pyro4.Proxy(
         settings.pyro_remote(
             settings.PYRO_GIT_REMOTE_WSGI, server_and_port))
     scm_app.HG_REMOTE_WSGI = Pyro4.Proxy(
         settings.pyro_remote(
             settings.PYRO_HG_REMOTE_WSGI, server_and_port))
 
     @atexit.register
     def free_connection_resources():
         connection.Git = None
         connection.Hg = None
         connection.Svn = None
 
 
 def connect_http(server_and_port):
     from rhodecode.lib.vcs import connection, client_http
     from rhodecode.lib.middleware.utils import scm_app
 
     session_factory = client_http.ThreadlocalSessionFactory()
 
     connection.Git = client_http.RepoMaker(
         server_and_port, '/git', session_factory)
     connection.Hg = client_http.RepoMaker(
         server_and_port, '/hg', session_factory)
     connection.Svn = client_http.RepoMaker(
         server_and_port, '/svn', session_factory)
 
     scm_app.HG_REMOTE_WSGI = client_http.VcsHttpProxy(
         server_and_port, '/proxy/hg')
     scm_app.GIT_REMOTE_WSGI = client_http.VcsHttpProxy(
         server_and_port, '/proxy/git')
 
     @atexit.register
     def free_connection_resources():
         connection.Git = None
         connection.Hg = None
         connection.Svn = None
 
 
 def connect_vcs(server_and_port, protocol='pyro4'):
     """
     Initializes the connection to the vcs server.
 
     :param server_and_port: str, e.g. "localhost:9900"
     :param protocol: str, "pyro4" or "http"
     """
     if protocol == 'pyro4':
         connect_pyro4(server_and_port)
     elif protocol == 'http':
         connect_http(server_and_port)
 
 
 # TODO: johbo: This function should be moved into our test suite, there is
 # no reason to support starting the vcsserver in Enterprise itself.
 def start_vcs_server(server_and_port, protocol='pyro4', log_level=None):
     """
     Starts the vcs server in a subprocess.
     """
     log.info('Starting VCSServer as a sub process with %s protocol', protocol)
     if protocol == 'http':
         return _start_http_vcs_server(server_and_port, log_level)
     elif protocol == 'pyro4':
         return _start_pyro4_vcs_server(server_and_port, log_level)
 
 
 def _start_pyro4_vcs_server(server_and_port, log_level=None):
     _try_to_shutdown_running_server(server_and_port)
     host, port = server_and_port.rsplit(":", 1)
     host = host.strip('[]')
     args = [
         'vcsserver', '--port', port, '--host', host, '--locale', 'en_US.UTF-8',
         '--threadpool', '32']
     if log_level:
         args += ['--log-level', log_level]
     proc = subprocess.Popen(args)
 
     def cleanup_server_process():
         proc.kill()
     atexit.register(cleanup_server_process)
 
     server = create_vcsserver_proxy(server_and_port, protocol='pyro4')
     _wait_until_vcs_server_is_reachable(server)
 
 
 def _start_http_vcs_server(server_and_port, log_level=None):
     # TODO: mikhail: shutdown if an http server already runs
 
     host, port = server_and_port.rsplit(":", 1)
     args = [
         'pserve', 'vcsserver/development_pyramid.ini',
         'http_port=%s' % (port, ), 'http_host=%s' % (host, )]
     proc = subprocess.Popen(args)
 
     def cleanup_server_process():
         proc.kill()
     atexit.register(cleanup_server_process)
 
     server = create_vcsserver_proxy(server_and_port, protocol='http')
     _wait_until_vcs_server_is_reachable(server)
 
 
 def _wait_until_vcs_server_is_reachable(server):
     while xrange(80):  # max 40s of sleep
         try:
             server.ping()
             break
         except (CommunicationError, pycurl.error):
             pass
         time.sleep(0.5)
 
 
 def _try_to_shutdown_running_server(server_and_port):
     server = create_vcsserver_proxy(server_and_port)
     try:
         server.shutdown()
     except (CommunicationError, pycurl.error):
         return
 
     # TODO: Not sure why this is important, but without it the following start
     # of the server fails.
     server = create_vcsserver_proxy(server_and_port)
     server.ping()
 
 
 def create_vcsserver_proxy(server_and_port, protocol='pyro4'):
     if protocol == 'pyro4':
         return _create_vcsserver_proxy_pyro4(server_and_port)
     elif protocol == 'http':
         return _create_vcsserver_proxy_http(server_and_port)
 
 
 def _create_vcsserver_proxy_pyro4(server_and_port):
     server = Pyro4.Proxy(
         settings.pyro_remote(settings.PYRO_VCSSERVER, server_and_port))
     return server
 
 
 def _create_vcsserver_proxy_http(server_and_port):
     from rhodecode.lib.vcs import client_http
 
     session = _create_http_rpc_session()
     url = urlparse.urljoin('http://%s' % server_and_port, '/server')
     return client_http.RemoteObject(url, session)
 
 
 class CurlSession(object):
     """
     Modeled so that it provides a subset of the requests interface.
 
     This has been created so that it does only provide a minimal API for our
     needs. The parts which it provides are based on the API of the library
     `requests` which allows us to easily benchmark against it.
 
     Please have a look at the class :class:`requests.Session` when you extend
     it.
     """
 
     def __init__(self):
         curl = pycurl.Curl()
         # TODO: johbo: I did test with 7.19 of libcurl. This version has
         # trouble with 100 - continue being set in the expect header. This
         # can lead to massive performance drops, switching it off here.
         curl.setopt(curl.HTTPHEADER, ["Expect:"])
         curl.setopt(curl.TCP_NODELAY, True)
         curl.setopt(curl.PROTOCOLS, curl.PROTO_HTTP)
         self._curl = curl
 
     def post(self, url, data, allow_redirects=False):
         response_buffer = StringIO()
 
         curl = self._curl
         curl.setopt(curl.URL, url)
         curl.setopt(curl.POST, True)
         curl.setopt(curl.POSTFIELDS, data)
         curl.setopt(curl.FOLLOWLOCATION, allow_redirects)
         curl.setopt(curl.WRITEDATA, response_buffer)
         curl.perform()
 
         return CurlResponse(response_buffer)
 
 
 class CurlResponse(object):
     """
     The response of a request, modeled after the requests API.
 
     This class provides a subset of the response interface known from the
     library `requests`. It is intentionally kept similar, so that we can use
     `requests` as a drop in replacement for benchmarking purposes.
     """
 
     def __init__(self, response_buffer):
         self._response_buffer = response_buffer
 
     @property
     def content(self):
         return self._response_buffer.getvalue()
 
 
 def _create_http_rpc_session():
     session = CurlSession()
     return session

rhodecode/lib/vcs/client.py

@@ -1,284 +1,327 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Provides the implementation of various client utilities to reach the vcsserver.
 """
 
 
 import copy
 import logging
 import threading
 import urlparse
 import uuid
 import weakref
 from urllib2 import URLError
 
 import msgpack
 import Pyro4
 import requests
+from pyramid.threadlocal import get_current_request
 from Pyro4.errors import CommunicationError, ConnectionClosedError, DaemonError
 
 from rhodecode.lib.vcs import exceptions
 from rhodecode.lib.vcs.conf import settings
 
 log = logging.getLogger(__name__)
 
 
 # TODO: mikhail: Keep it in sync with vcsserver's
 # HTTPApplication.ALLOWED_EXCEPTIONS
 EXCEPTIONS_MAP = {
     'KeyError': KeyError,
     'URLError': URLError,
 }
 
 
 class HTTPRepoMaker(object):
     def __init__(self, server_and_port, backend_endpoint):
         self.url = urlparse.urljoin(
             'http://%s' % server_and_port, backend_endpoint)
 
     def __call__(self, path, config, with_wire=None):
         log.debug('HTTPRepoMaker call on %s', path)
         return HTTPRemoteRepo(path, config, self.url, with_wire=with_wire)
 
     def __getattr__(self, name):
         def f(*args, **kwargs):
             return self._call(name, *args, **kwargs)
         return f
 
     @exceptions.map_vcs_exceptions
     def _call(self, name, *args, **kwargs):
         payload = {
             'id': str(uuid.uuid4()),
             'method': name,
             'params': {'args': args, 'kwargs': kwargs}
         }
         return _remote_call(self.url, payload, EXCEPTIONS_MAP)
 
 
 class VcsHttpProxy(object):
 
     CHUNK_SIZE = 16384
 
     def __init__(self, server_and_port, backend_endpoint):
         adapter = requests.adapters.HTTPAdapter(max_retries=5)
         self.base_url = urlparse.urljoin(
             'http://%s' % server_and_port, backend_endpoint)
         self.session = requests.Session()
         self.session.mount('http://', adapter)
 
     def handle(self, environment, input_data, *args, **kwargs):
         data = {
             'environment': environment,
             'input_data': input_data,
             'args': args,
             'kwargs': kwargs
         }
         result = self.session.post(
             self.base_url, msgpack.packb(data), stream=True)
         return self._get_result(result)
 
     def _deserialize_and_raise(self, error):
         exception = Exception(error['message'])
         try:
             exception._vcs_kind = error['_vcs_kind']
         except KeyError:
             pass
         raise exception
 
     def _iterate(self, result):
         unpacker = msgpack.Unpacker()
         for line in result.iter_content(chunk_size=self.CHUNK_SIZE):
             unpacker.feed(line)
             for chunk in unpacker:
                 yield chunk
 
     def _get_result(self, result):
         iterator = self._iterate(result)
         error = iterator.next()
         if error:
             self._deserialize_and_raise(error)
 
         status = iterator.next()
         headers = iterator.next()
 
         return iterator, status, headers
 
 
 class HTTPRemoteRepo(object):
     def __init__(self, path, config, url, with_wire=None):
         self.url = url
         self._wire = {
             "path": path,
             "config": config,
             "context": str(uuid.uuid4()),
         }
         if with_wire:
             self._wire.update(with_wire)
 
     def __getattr__(self, name):
         def f(*args, **kwargs):
             return self._call(name, *args, **kwargs)
         return f
 
     @exceptions.map_vcs_exceptions
     def _call(self, name, *args, **kwargs):
         log.debug('Calling %s@%s', self.url, name)
         # TODO: oliver: This is currently necessary pre-call since the
         # config object is being changed for hooking scenarios
         wire = copy.deepcopy(self._wire)
         wire["config"] = wire["config"].serialize()
         payload = {
             'id': str(uuid.uuid4()),
             'method': name,
             'params': {'wire': wire, 'args': args, 'kwargs': kwargs}
         }
         return _remote_call(self.url, payload, EXCEPTIONS_MAP)
 
     def __getitem__(self, key):
         return self.revision(key)
 
 
 def _remote_call(url, payload, exceptions_map):
     response = requests.post(url, data=msgpack.packb(payload))
     response = msgpack.unpackb(response.content)
     error = response.get('error')
     if error:
         type_ = error.get('type', 'Exception')
         exc = exceptions_map.get(type_, Exception)
         exc = exc(error.get('message'))
         try:
             exc._vcs_kind = error['_vcs_kind']
         except KeyError:
             pass
         raise exc
     return response.get('result')
 
 
 class RepoMaker(object):
 
     def __init__(self, proxy_factory):
         self._proxy_factory = proxy_factory
 
     def __call__(self, path, config, with_wire=None):
         log.debug('RepoMaker call on %s', path)
         return RemoteRepo(
             path, config, remote_proxy=self._proxy_factory(),
             with_wire=with_wire)
 
     def __getattr__(self, name):
         remote_proxy = self._proxy_factory()
         func = _get_proxy_method(remote_proxy, name)
         return _wrap_remote_call(remote_proxy, func)
 
 
-class ThreadlocalProxyFactory(object):
+class RequestScopeProxyFactory(object):
     """
-    Creates one Pyro4 proxy per thread on demand.
+    This factory returns pyro proxy instances based on a per request scope.
+    It returns the same instance if called from within the same request and
+    different instances if called from different requests.
     """
 
     def __init__(self, remote_uri):
         self._remote_uri = remote_uri
-        self._thread_local = threading.local()
+        self._proxy_pool = []
+        self._borrowed_proxies = {}
+
+    def __call__(self, request=None):
+        """
+        Wrapper around `getProxy`.
+        """
+        request = request or get_current_request()
+        return self.getProxy(request)
+
+    def getProxy(self, request=None):
+        """
+        Call this to get the pyro proxy instance for the request.
+        """
+        request = request or get_current_request()
+
+        # Return already borrowed proxy for this request
+        if request in self._borrowed_proxies:
+            return self._borrowed_proxies[request]
 
-    def __call__(self):
-        if not hasattr(self._thread_local, 'proxy'):
-            self._thread_local.proxy = Pyro4.Proxy(self._remote_uri)
-        return self._thread_local.proxy
+        # Get proxy from pool or create new instance.
+        try:
+            proxy = self._proxy_pool.pop()
+        except IndexError:
+            log.info('Creating new proxy for remote_uri=%s', self._remote_uri)
+            proxy = Pyro4.Proxy(self._remote_uri)
+
+        # Store proxy instance as borrowed and add request callback.
+        self._borrowed_proxies[request] = proxy
+        request.add_finished_callback(self._returnProxy)
+
+        return proxy
+
+    def _returnProxy(self, request=None):
+        """
+        Callback that gets called by pyramid when the request is finished.
+        It puts the proxy back into the pool.
+        """
+        request = request or get_current_request()
+
+        if request in self._borrowed_proxies:
+            proxy = self._borrowed_proxies.pop(request)
+            self._proxy_pool.append(proxy)
+        else:
+            log.warn('Return proxy for remote_uri=%s but no proxy borrowed '
+                     'for this request.', self._remote_uri)
 
 
 class RemoteRepo(object):
 
     def __init__(self, path, config, remote_proxy, with_wire=None):
         self._wire = {
             "path": path,
             "config": config,
             "context": uuid.uuid4(),
         }
         if with_wire:
             self._wire.update(with_wire)
         self._remote_proxy = remote_proxy
         self.refs = RefsWrapper(self)
 
     def __getattr__(self, name):
         log.debug('Calling %s@%s', self._remote_proxy, name)
         # TODO: oliver: This is currently necessary pre-call since the
         # config object is being changed for hooking scenarios
         wire = copy.deepcopy(self._wire)
         wire["config"] = wire["config"].serialize()
 
         try:
             func = _get_proxy_method(self._remote_proxy, name)
         except DaemonError as e:
             if e.message == 'unknown object':
                 raise exceptions.VCSBackendNotSupportedError
             else:
                 raise
 
         return _wrap_remote_call(self._remote_proxy, func, wire)
 
     def __getitem__(self, key):
         return self.revision(key)
 
 
 def _get_proxy_method(proxy, name):
     try:
         return getattr(proxy, name)
     except CommunicationError:
         raise CommunicationError(
             'Unable to connect to remote pyro server %s' % proxy)
 
 
 def _wrap_remote_call(proxy, func, *args):
     all_args = list(args)
 
     @exceptions.map_vcs_exceptions
     def caller(*args, **kwargs):
         all_args.extend(args)
         try:
             return func(*all_args, **kwargs)
         except ConnectionClosedError:
             log.debug('Connection to VCSServer closed, trying to reconnect.')
             proxy._pyroReconnect(tries=settings.PYRO_RECONNECT_TRIES)
 
             return func(*all_args, **kwargs)
 
     return caller
 
 
 class RefsWrapper(object):
 
     def __init__(self, repo):
         self._repo = weakref.proxy(repo)
 
     def __setitem__(self, key, value):
         self._repo._assign_ref(key, value)
 
 
 class FunctionWrapper(object):
 
     def __init__(self, func, wire):
         self._func = func
         self._wire = wire
 
     @exceptions.map_vcs_exceptions
     def __call__(self, *args, **kwargs):
         return self._func(self._wire, *args, **kwargs)