commit 33231500aa82861a438df098114ca442a4859d23
Author: Michael Hanselmann <hansmi@google.com>
Date:   Tue Jul 6 15:56:49 2010 +0200

    Convert RPC client to PycURL
    
    Instead of using our custom HTTP client, using PycURL's multi
    interface allows us to get rid of the HTTP client threadpool.
    The majority of the code is still in the ganeti.http.client
    module.
    
    A simple per-thread HTTP client pool gives cURL a chance to
    cache and retain as much information as possible (e.g. SSL certs).
    Unused HTTP clients (e.g. due to removed nodes) are deleted after
    25 requests going through the pool.
    
    Signed-off-by: Michael Hanselmann <hansmi@google.com>
    Reviewed-by: Iustin Pop <iustin@google.com>

diff --git a/Makefile.am b/Makefile.am
index f54d8e9..615ed85 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -389,6 +389,7 @@ python_tests = \
 	test/ganeti.rapi.client_unittest.py \
 	test/ganeti.rapi.resources_unittest.py \
 	test/ganeti.rapi.rlib2_unittest.py \
+	test/ganeti.rpc_unittest.py \
 	test/ganeti.serializer_unittest.py \
 	test/ganeti.ssh_unittest.py \
 	test/ganeti.uidpool_unittest.py \
diff --git a/daemons/ganeti-masterd b/daemons/ganeti-masterd
index 105dbe1..2ed6e79 100755
--- a/daemons/ganeti-masterd
+++ b/daemons/ganeti-masterd
@@ -513,7 +513,7 @@ def CheckMasterd(options, args):
     sys.exit(constants.EXIT_FAILURE)
 
 
-def ExecMasterd (options, args): # pylint: disable-msg=W0613
+def ExecMasterd(options, args): # pylint: disable-msg=W0613
   """Main master daemon function, executed with the PID file held.
 
   """
diff --git a/lib/http/client.py b/lib/http/client.py
index 697199d..8658aed 100644
--- a/lib/http/client.py
+++ b/lib/http/client.py
@@ -1,7 +1,7 @@
 #
 #
 
-# Copyright (C) 2007, 2008 Google Inc.
+# Copyright (C) 2007, 2008, 2010 Google Inc.
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -22,30 +22,17 @@
 
 """
 
-# pylint: disable-msg=E1103
+import logging
+import pycurl
+from cStringIO import StringIO
 
-# # E1103: %s %r has no %r member (but some types could not be
-# inferred), since _socketobject could be ssl or not and pylint
-# doesn't parse that
-
-
-import os
-import select
-import socket
-import errno
-import threading
-
-from ganeti import workerpool
 from ganeti import http
-from ganeti import utils
-
-
-HTTP_CLIENT_THREADS = 10
+from ganeti import compat
 
 
 class HttpClientRequest(object):
   def __init__(self, host, port, method, path, headers=None, post_data=None,
-               ssl_params=None, ssl_verify_peer=False, read_timeout=None):
+               read_timeout=None, curl_config_fn=None):
     """Describes an HTTP request.
 
     @type host: string
@@ -56,48 +43,52 @@ class HttpClientRequest(object):
     @param method: Method name
     @type path: string
     @param path: Request path
-    @type headers: dict or None
-    @param headers: Additional headers to send
+    @type headers: list or None
+    @param headers: Additional headers to send, list of strings
     @type post_data: string or None
     @param post_data: Additional data to send
-    @type ssl_params: HttpSslParams
-    @param ssl_params: SSL key and certificate
-    @type ssl_verify_peer: bool
-    @param ssl_verify_peer: Whether to compare our certificate with
-        server's certificate
     @type read_timeout: int
     @param read_timeout: if passed, it will be used as the read
         timeout while reading the response from the server
+    @type curl_config_fn: callable
+    @param curl_config_fn: Function to configure cURL object before request
+                           (Note: if the function configures the connection in
+                           a way where it wouldn't be efficient to reuse them,
+                           a "identity" property should be defined, see
+                           L{HttpClientRequest.identity})
 
     """
-    if post_data is not None:
-      assert method.upper() in (http.HTTP_POST, http.HTTP_PUT), \
-        "Only POST and GET requests support sending data"
-
     assert path.startswith("/"), "Path must start with slash (/)"
+    assert curl_config_fn is None or callable(curl_config_fn)
 
     # Request attributes
     self.host = host
     self.port = port
-    self.ssl_params = ssl_params
-    self.ssl_verify_peer = ssl_verify_peer
     self.method = method
     self.path = path
-    self.headers = headers
-    self.post_data = post_data
     self.read_timeout = read_timeout
+    self.curl_config_fn = curl_config_fn
 
+    if post_data is None:
+      self.post_data = ""
+    else:
+      self.post_data = post_data
+
+    if headers is None:
+      self.headers = []
+    elif isinstance(headers, dict):
+      # Support for old interface
+      self.headers = ["%s: %s" % (name, value)
+                      for name, value in headers.items()]
+    else:
+      self.headers = headers
+
+    # Response status
     self.success = None
     self.error = None
 
-    # Raw response
-    self.response = None
-
     # Response attributes
-    self.resp_version = None
     self.resp_status_code = None
-    self.resp_reason = None
-    self.resp_headers = None
     self.resp_body = None
 
   def __repr__(self):
@@ -108,310 +99,350 @@ class HttpClientRequest(object):
 
     return "<%s at %#x>" % (" ".join(status), id(self))
 
+  @property
+  def url(self):
+    """Returns the full URL for this requests.
 
-class _HttpClientToServerMessageWriter(http.HttpMessageWriter):
-  pass
-
+    """
+    # TODO: Support for non-SSL requests
+    return "https://%s:%s%s" % (self.host, self.port, self.path)
 
-class _HttpServerToClientMessageReader(http.HttpMessageReader):
-  # Length limits
-  START_LINE_LENGTH_MAX = 512
-  HEADER_LENGTH_MAX = 4096
+  @property
+  def identity(self):
+    """Returns identifier for retrieving a pooled connection for this request.
 
-  def ParseStartLine(self, start_line):
-    """Parses the status line sent by the server.
+    This allows cURL client objects to be re-used and to cache information
+    (e.g. SSL session IDs or connections).
 
     """
-    # Empty lines are skipped when reading
-    assert start_line
+    parts = [self.host, self.port]
 
-    try:
-      [version, status, reason] = start_line.split(None, 2)
-    except ValueError:
+    if self.curl_config_fn:
       try:
-        [version, status] = start_line.split(None, 1)
-        reason = ""
-      except ValueError:
-        version = http.HTTP_0_9
+        parts.append(self.curl_config_fn.identity)
+      except AttributeError:
+        pass
 
-    if version:
-      version = version.upper()
+    return "/".join(str(i) for i in parts)
 
-    # The status code is a three-digit number
-    try:
-      status = int(status)
-      if status < 100 or status > 999:
-        status = -1
-    except (TypeError, ValueError):
-      status = -1
 
-    if status == -1:
-      raise http.HttpError("Invalid status code (%r)" % start_line)
+class _HttpClient(object):
+  def __init__(self, curl_config_fn):
+    """Initializes this class.
 
-    return http.HttpServerToClientStartLine(version, status, reason)
+    @type curl_config_fn: callable
+    @param curl_config_fn: Function to configure cURL object after
+                           initialization
 
+    """
+    self._req = None
 
-class HttpClientRequestExecutor(http.HttpBase):
-  # Default headers
-  DEFAULT_HEADERS = {
-    http.HTTP_USER_AGENT: http.HTTP_GANETI_VERSION,
-    # TODO: For keep-alive, don't send "Connection: close"
-    http.HTTP_CONNECTION: "close",
-    }
+    curl = self._CreateCurlHandle()
+    curl.setopt(pycurl.VERBOSE, False)
+    curl.setopt(pycurl.NOSIGNAL, True)
+    curl.setopt(pycurl.USERAGENT, http.HTTP_GANETI_VERSION)
+    curl.setopt(pycurl.PROXY, "")
 
-  # Timeouts in seconds for socket layer
-  # TODO: Soft timeout instead of only socket timeout?
-  # TODO: Make read timeout configurable per OpCode?
-  CONNECT_TIMEOUT = 5
-  WRITE_TIMEOUT = 10
-  READ_TIMEOUT = None
-  CLOSE_TIMEOUT = 1
+    # Pass cURL object to external config function
+    if curl_config_fn:
+      curl_config_fn(curl)
 
-  def __init__(self, req):
-    """Initializes the HttpClientRequestExecutor class.
+    self._curl = curl
 
-    @type req: HttpClientRequest
-    @param req: Request object
+  @staticmethod
+  def _CreateCurlHandle():
+    """Returns a new cURL object.
 
     """
-    http.HttpBase.__init__(self)
-    self.request = req
-
-    try:
-      # TODO: Implement connection caching/keep-alive
-      self.sock = self._CreateSocket(req.ssl_params,
-                                     req.ssl_verify_peer)
+    return pycurl.Curl()
 
-      # Disable Python's timeout
-      self.sock.settimeout(None)
+  def GetCurlHandle(self):
+    """Returns the cURL object.
 
-      # Operate in non-blocking mode
-      self.sock.setblocking(0)
+    """
+    return self._curl
 
-      response_msg_reader = None
-      response_msg = None
-      force_close = True
+  def GetCurrentRequest(self):
+    """Returns the current request.
 
-      self._Connect()
-      try:
-        self._SendRequest()
-        (response_msg_reader, response_msg) = self._ReadResponse()
+    @rtype: L{HttpClientRequest} or None
 
-        # Only wait for server to close if we didn't have any exception.
-        force_close = False
-      finally:
-        # TODO: Keep-alive is not supported, always close connection
-        force_close = True
-        http.ShutdownConnection(self.sock, self.CLOSE_TIMEOUT,
-                                self.WRITE_TIMEOUT, response_msg_reader,
-                                force_close)
+    """
+    return self._req
 
-      self.sock.close()
-      self.sock = None
+  def StartRequest(self, req):
+    """Starts a request on this client.
 
-      req.response = response_msg
+    @type req: L{HttpClientRequest}
+    @param req: HTTP request
 
-      req.resp_version = req.response.start_line.version
-      req.resp_status_code = req.response.start_line.code
-      req.resp_reason = req.response.start_line.reason
-      req.resp_headers = req.response.headers
-      req.resp_body = req.response.body
+    """
+    assert not self._req, "Another request is already started"
+
+    self._req = req
+    self._resp_buffer = StringIO()
+
+    url = req.url
+    method = req.method
+    post_data = req.post_data
+    headers = req.headers
+
+    # PycURL requires strings to be non-unicode
+    assert isinstance(method, str)
+    assert isinstance(url, str)
+    assert isinstance(post_data, str)
+    assert compat.all(isinstance(i, str) for i in headers)
+
+    # Configure cURL object for request
+    curl = self._curl
+    curl.setopt(pycurl.CUSTOMREQUEST, str(method))
+    curl.setopt(pycurl.URL, url)
+    curl.setopt(pycurl.POSTFIELDS, post_data)
+    curl.setopt(pycurl.WRITEFUNCTION, self._resp_buffer.write)
+    curl.setopt(pycurl.HTTPHEADER, headers)
+
+    if req.read_timeout is None:
+      curl.setopt(pycurl.TIMEOUT, 0)
+    else:
+      curl.setopt(pycurl.TIMEOUT, int(req.read_timeout))
 
-      req.success = True
-      req.error = None
+    # Pass cURL object to external config function
+    if req.curl_config_fn:
+      req.curl_config_fn(curl)
 
-    except http.HttpError, err:
-      req.success = False
-      req.error = str(err)
+  def Done(self, errmsg):
+    """Finishes a request.
 
-  def _Connect(self):
-    """Non-blocking connect to host with timeout.
+    @type errmsg: string or None
+    @param errmsg: Error message if request failed
 
     """
-    connected = False
-    while True:
-      try:
-        connect_error = self.sock.connect_ex((self.request.host,
-                                              self.request.port))
-      except socket.gaierror, err:
-        raise http.HttpError("Connection failed: %s" % str(err))
+    req = self._req
+    assert req, "No request"
 
-      if connect_error == errno.EINTR:
-        # Mask signals
-        pass
+    logging.debug("Request %s finished, errmsg=%s", req, errmsg)
 
-      elif connect_error == 0:
-        # Connection established
-        connected = True
-        break
+    curl = self._curl
 
-      elif connect_error == errno.EINPROGRESS:
-        # Connection started
-        break
+    req.success = not bool(errmsg)
+    req.error = errmsg
 
-      raise http.HttpError("Connection failed (%s: %s)" %
-                             (connect_error, os.strerror(connect_error)))
+    # Get HTTP response code
+    req.resp_status_code = curl.getinfo(pycurl.RESPONSE_CODE)
+    req.resp_body = self._resp_buffer.getvalue()
 
-    if not connected:
-      # Wait for connection
-      event = utils.WaitForFdCondition(self.sock, select.POLLOUT,
-                                       self.CONNECT_TIMEOUT)
-      if event is None:
-        raise http.HttpError("Timeout while connecting to server")
+    # Reset client object
+    self._req = None
+    self._resp_buffer = None
 
-      # Get error code
-      connect_error = self.sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
-      if connect_error != 0:
-        raise http.HttpError("Connection failed (%s: %s)" %
-                               (connect_error, os.strerror(connect_error)))
+    # Ensure no potentially large variables are referenced
+    curl.setopt(pycurl.POSTFIELDS, "")
+    curl.setopt(pycurl.WRITEFUNCTION, lambda _: None)
 
-    # Enable TCP keep-alive
-    self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
 
-    # If needed, Linux specific options are available to change the TCP
-    # keep-alive settings, see "man 7 tcp" for TCP_KEEPCNT, TCP_KEEPIDLE and
-    # TCP_KEEPINTVL.
+class _PooledHttpClient:
+  """Data structure for HTTP client pool.
 
-    # Do the secret SSL handshake
-    if self.using_ssl:
-      self.sock.set_connect_state() # pylint: disable-msg=E1103
-      try:
-        http.Handshake(self.sock, self.WRITE_TIMEOUT)
-      except http.HttpSessionHandshakeUnexpectedEOF:
-        raise http.HttpError("Server closed connection during SSL handshake")
+  """
+  def __init__(self, identity, client):
+    """Initializes this class.
 
-  def _SendRequest(self):
-    """Sends request to server.
+    @type identity: string
+    @param identity: Client identifier for pool
+    @type client: L{_HttpClient}
+    @param client: HTTP client
 
     """
-    # Headers
-    send_headers = self.DEFAULT_HEADERS.copy()
+    self.identity = identity
+    self.client = client
+    self.lastused = 0
 
-    if self.request.headers:
-      send_headers.update(self.request.headers)
+  def __repr__(self):
+    status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),
+              "id=%s" % self.identity,
+              "lastuse=%s" % self.lastused,
+              repr(self.client)]
 
-    send_headers[http.HTTP_HOST] = "%s:%s" % (self.request.host,
-                                              self.request.port)
+    return "<%s at %#x>" % (" ".join(status), id(self))
 
-    # Response message
-    msg = http.HttpMessage()
 
-    # Combine request line. We only support HTTP/1.0 (no chunked transfers and
-    # no keep-alive).
-    # TODO: For keep-alive, change to HTTP/1.1
-    msg.start_line = \
-      http.HttpClientToServerStartLine(method=self.request.method.upper(),
-                                       path=self.request.path,
-                                       version=http.HTTP_1_0)
-    msg.headers = send_headers
-    msg.body = self.request.post_data
+class HttpClientPool:
+  """A simple HTTP client pool.
 
-    try:
-      _HttpClientToServerMessageWriter(self.sock, msg, self.WRITE_TIMEOUT)
-    except http.HttpSocketTimeout:
-      raise http.HttpError("Timeout while sending request")
-    except socket.error, err:
-      raise http.HttpError("Error sending request: %s" % err)
+  Supports one pooled connection per identity (see
+  L{HttpClientRequest.identity}).
 
-  def _ReadResponse(self):
-    """Read response from server.
+  """
+  #: After how many generations to drop unused clients
+  _MAX_GENERATIONS_DROP = 25
+
+  def __init__(self, curl_config_fn):
+    """Initializes this class.
+
+    @type curl_config_fn: callable
+    @param curl_config_fn: Function to configure cURL object after
+                           initialization
 
     """
-    response_msg = http.HttpMessage()
+    self._curl_config_fn = curl_config_fn
+    self._generation = 0
+    self._pool = {}
 
-    if self.request.read_timeout is None:
-      timeout = self.READ_TIMEOUT
-    else:
-      timeout = self.request.read_timeout
+  @staticmethod
+  def _GetHttpClientCreator():
+    """Returns callable to create HTTP client.
 
-    try:
-      response_msg_reader = \
-        _HttpServerToClientMessageReader(self.sock, response_msg, timeout)
-    except http.HttpSocketTimeout:
-      raise http.HttpError("Timeout while reading response")
-    except socket.error, err:
-      raise http.HttpError("Error reading response: %s" % err)
+    """
+    return _HttpClient
 
-    return (response_msg_reader, response_msg)
+  def _Get(self, identity):
+    """Gets an HTTP client from the pool.
 
+    @type identity: string
+    @param identity: Client identifier
 
-class _HttpClientPendingRequest(object):
-  """Data class for pending requests.
+    """
+    try:
+      pclient  = self._pool.pop(identity)
+    except KeyError:
+      # Need to create new client
+      client = self._GetHttpClientCreator()(self._curl_config_fn)
+      pclient = _PooledHttpClient(identity, client)
+      logging.debug("Created new client %s", pclient)
+    else:
+      logging.debug("Reusing client %s", pclient)
 
-  """
-  def __init__(self, request):
-    self.request = request
+    assert pclient.identity == identity
 
-    # Thread synchronization
-    self.done = threading.Event()
+    return pclient
 
-  def __repr__(self):
-    status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),
-              "req=%r" % self.request]
+  def _StartRequest(self, req):
+    """Starts a request.
 
-    return "<%s at %#x>" % (" ".join(status), id(self))
+    @type req: L{HttpClientRequest}
+    @param req: HTTP request
 
+    """
+    logging.debug("Starting request %r", req)
+    pclient = self._Get(req.identity)
 
-class HttpClientWorker(workerpool.BaseWorker):
-  """HTTP client worker class.
+    assert req.identity not in self._pool
 
-  """
-  def RunTask(self, pend_req): # pylint: disable-msg=W0221
-    try:
-      HttpClientRequestExecutor(pend_req.request)
-    finally:
-      pend_req.done.set()
+    pclient.client.StartRequest(req)
+    pclient.lastused = self._generation
+
+    return pclient
 
+  def _Return(self, pclients):
+    """Returns HTTP clients to the pool.
 
-class HttpClientWorkerPool(workerpool.WorkerPool):
-  def __init__(self, manager):
-    workerpool.WorkerPool.__init__(self, "HttpClient",
-                                   HTTP_CLIENT_THREADS,
-                                   HttpClientWorker)
-    self.manager = manager
+    """
+    for pc in pclients:
+      logging.debug("Returning client %s to pool", pc)
+      assert pc.identity not in self._pool
+      assert pc not in self._pool.values()
+      self._pool[pc.identity] = pc
+
+    # Check for unused clients
+    for pc in self._pool.values():
+      if (pc.lastused + self._MAX_GENERATIONS_DROP) < self._generation:
+        logging.debug("Removing client %s which hasn't been used"
+                      " for %s generations",
+                      pc, self._MAX_GENERATIONS_DROP)
+        self._pool.pop(pc.identity, None)
+
+    assert compat.all(pc.lastused >= (self._generation -
+                                      self._MAX_GENERATIONS_DROP)
+                      for pc in self._pool.values())
+
+  @staticmethod
+  def _CreateCurlMultiHandle():
+    """Creates new cURL multi handle.
 
+    """
+    return pycurl.CurlMulti()
 
-class HttpClientManager(object):
-  """Manages HTTP requests.
+  def ProcessRequests(self, requests):
+    """Processes any number of HTTP client requests using pooled objects.
 
-  """
-  def __init__(self):
-    self._wpool = HttpClientWorkerPool(self)
+    @type requests: list of L{HttpClientRequest}
+    @param requests: List of all requests
 
-  def __del__(self):
-    self.Shutdown()
+    """
+    multi = self._CreateCurlMultiHandle()
 
-  def ExecRequests(self, requests):
-    """Execute HTTP requests.
+    # For client cleanup
+    self._generation += 1
 
-    This function can be called from multiple threads at the same time.
+    assert compat.all((req.error is None and
+                       req.success is None and
+                       req.resp_status_code is None and
+                       req.resp_body is None)
+                      for req in requests)
 
-    @type requests: List of HttpClientRequest instances
-    @param requests: The requests to execute
-    @rtype: List of HttpClientRequest instances
-    @return: The list of requests passed in
+    curl_to_pclient = {}
+    for req in requests:
+      pclient = self._StartRequest(req)
+      curl = pclient.client.GetCurlHandle()
+      curl_to_pclient[curl] = pclient
+      multi.add_handle(curl)
+      assert pclient.client.GetCurrentRequest() == req
+      assert pclient.lastused >= 0
 
-    """
-    # _HttpClientPendingRequest is used for internal thread synchronization
-    pending = [_HttpClientPendingRequest(req) for req in requests]
+    assert len(curl_to_pclient) == len(requests)
 
-    try:
-      # Add requests to queue
-      for pend_req in pending:
-        self._wpool.AddTask(pend_req)
+    done_count = 0
+    while True:
+      (ret, _) = multi.perform()
+      assert ret in (pycurl.E_MULTI_OK, pycurl.E_CALL_MULTI_PERFORM)
+
+      if ret == pycurl.E_CALL_MULTI_PERFORM:
+        # cURL wants to be called again
+        continue
+
+      while True:
+        (remaining_messages, successful, failed) = multi.info_read()
+
+        for curl in successful:
+          multi.remove_handle(curl)
+          done_count += 1
+          pclient = curl_to_pclient[curl]
+          req = pclient.client.GetCurrentRequest()
+          pclient.client.Done(None)
+          assert req.success
+          assert not pclient.client.GetCurrentRequest()
+
+        for curl, errnum, errmsg in failed:
+          multi.remove_handle(curl)
+          done_count += 1
+          pclient = curl_to_pclient[curl]
+          req = pclient.client.GetCurrentRequest()
+          pclient.client.Done("Error %s: %s" % (errnum, errmsg))
+          assert req.error
+          assert not pclient.client.GetCurrentRequest()
+
+        if remaining_messages == 0:
+          break
+
+      assert done_count <= len(requests)
+
+      if done_count == len(requests):
+        break
 
-    finally:
-      # In case of an exception we should still wait for the rest, otherwise
-      # another thread from the worker pool could modify the request object
-      # after we returned.
+      # Wait for I/O. The I/O timeout shouldn't be too long so that HTTP
+      # timeouts, which are only evaluated in multi.perform, aren't
+      # unnecessarily delayed.
+      multi.select(1.0)
 
-      # And wait for them to finish
-      for pend_req in pending:
-        pend_req.done.wait()
+    assert compat.all(pclient.client.GetCurrentRequest() is None
+                      for pclient in curl_to_pclient.values())
 
-    # Return original list
-    return requests
+    # Return clients to pool
+    self._Return(curl_to_pclient.values())
 
-  def Shutdown(self):
-    self._wpool.Quiesce()
-    self._wpool.TerminateWorkers()
+    assert done_count == len(requests)
+    assert compat.all(req.error is not None or
+                      (req.success and
+                       req.resp_status_code is not None and
+                       req.resp_body is not None)
+                      for req in requests)
diff --git a/lib/rpc.py b/lib/rpc.py
index 00a6602..f5c4333 100644
--- a/lib/rpc.py
+++ b/lib/rpc.py
@@ -34,6 +34,8 @@ import os
 import logging
 import zlib
 import base64
+import pycurl
+import threading
 
 from ganeti import utils
 from ganeti import objects
@@ -47,8 +49,12 @@ from ganeti import netutils
 import ganeti.http.client  # pylint: disable-msg=W0611
 
 
-# Module level variable
-_http_manager = None
+# Timeout for connecting to nodes (seconds)
+_RPC_CONNECT_TIMEOUT = 5
+
+_RPC_CLIENT_HEADERS = [
+  "Content-type: %s" % http.HTTP_APP_JSON,
+  ]
 
 # Various time constants for the timeout table
 _TMO_URGENT = 60 # one minute
@@ -72,29 +78,62 @@ _TIMEOUTS = {
 def Init():
   """Initializes the module-global HTTP client manager.
 
-  Must be called before using any RPC function.
+  Must be called before using any RPC function and while exactly one thread is
+  running.
 
   """
-  global _http_manager # pylint: disable-msg=W0603
-
-  assert not _http_manager, "RPC module initialized more than once"
+  # curl_global_init(3) and curl_global_cleanup(3) must be called with only
+  # one thread running. This check is just a safety measure -- it doesn't
+  # cover all cases.
+  assert threading.activeCount() == 1, \
+         "Found more than one active thread when initializing pycURL"
 
-  http.InitSsl()
+  logging.info("Using PycURL %s", pycurl.version)
 
-  _http_manager = http.client.HttpClientManager()
+  pycurl.global_init(pycurl.GLOBAL_ALL)
 
 
 def Shutdown():
   """Stops the module-global HTTP client manager.
 
-  Must be called before quitting the program.
+  Must be called before quitting the program and while exactly one thread is
+  running.
 
   """
-  global _http_manager # pylint: disable-msg=W0603
+  pycurl.global_cleanup()
+
+
+def _ConfigRpcCurl(curl):
+  noded_cert = str(constants.NODED_CERT_FILE)
 
-  if _http_manager:
-    _http_manager.Shutdown()
-    _http_manager = None
+  curl.setopt(pycurl.FOLLOWLOCATION, False)
+  curl.setopt(pycurl.CAINFO, noded_cert)
+  curl.setopt(pycurl.SSL_VERIFYHOST, 0)
+  curl.setopt(pycurl.SSL_VERIFYPEER, True)
+  curl.setopt(pycurl.SSLCERTTYPE, "PEM")
+  curl.setopt(pycurl.SSLCERT, noded_cert)
+  curl.setopt(pycurl.SSLKEYTYPE, "PEM")
+  curl.setopt(pycurl.SSLKEY, noded_cert)
+  curl.setopt(pycurl.CONNECTTIMEOUT, _RPC_CONNECT_TIMEOUT)
+
+
+class _RpcThreadLocal(threading.local):
+  def GetHttpClientPool(self):
+    """Returns a per-thread HTTP client pool.
+
+    @rtype: L{http.client.HttpClientPool}
+
+    """
+    try:
+      pool = self.hcp
+    except AttributeError:
+      pool = http.client.HttpClientPool(_ConfigRpcCurl)
+      self.hcp = pool
+
+    return pool
+
+
+_thread_local = _RpcThreadLocal()
 
 
 def _RpcTimeout(secs):
@@ -218,11 +257,7 @@ class Client:
     self.procedure = procedure
     self.body = body
     self.port = port
-    self.nc = {}
-
-    self._ssl_params = \
-      http.HttpSslParams(ssl_key_path=constants.NODED_CERT_FILE,
-                         ssl_cert_path=constants.NODED_CERT_FILE)
+    self._request = {}
 
   def ConnectList(self, node_list, address_list=None, read_timeout=None):
     """Add a list of nodes to the target nodes.
@@ -260,28 +295,28 @@ class Client:
     if read_timeout is None:
       read_timeout = _TIMEOUTS[self.procedure]
 
-    self.nc[name] = \
-      http.client.HttpClientRequest(address, self.port, http.HTTP_PUT,
-                                    "/%s" % self.procedure,
-                                    post_data=self.body,
-                                    ssl_params=self._ssl_params,
-                                    ssl_verify_peer=True,
+    self._request[name] = \
+      http.client.HttpClientRequest(str(address), self.port,
+                                    http.HTTP_PUT, str("/%s" % self.procedure),
+                                    headers=_RPC_CLIENT_HEADERS,
+                                    post_data=str(self.body),
                                     read_timeout=read_timeout)
 
-  def GetResults(self):
+  def GetResults(self, http_pool=None):
     """Call nodes and return results.
 
     @rtype: list
     @return: List of RPC results
 
     """
-    assert _http_manager, "RPC module not initialized"
+    if not http_pool:
+      http_pool = _thread_local.GetHttpClientPool()
 
-    _http_manager.ExecRequests(self.nc.values())
+    http_pool.ProcessRequests(self._request.values())
 
     results = {}
 
-    for name, req in self.nc.iteritems():
+    for name, req in self._request.iteritems():
       if req.success and req.resp_status_code == http.HTTP_OK:
         results[name] = RpcResult(data=serializer.LoadJson(req.resp_body),
                                   node=name, call=self.procedure)
diff --git a/test/ganeti.http_unittest.py b/test/ganeti.http_unittest.py
index 7d0d477..3180c7c 100755
--- a/test/ganeti.http_unittest.py
+++ b/test/ganeti.http_unittest.py
@@ -87,12 +87,6 @@ class TestMisc(unittest.TestCase):
     self.assert_(message_reader_class.START_LINE_LENGTH_MAX > 0)
     self.assert_(message_reader_class.HEADER_LENGTH_MAX > 0)
 
-  def testClientSizeLimits(self):
-    """Test HTTP client size limits"""
-    message_reader_class = http.client._HttpServerToClientMessageReader
-    self.assert_(message_reader_class.START_LINE_LENGTH_MAX > 0)
-    self.assert_(message_reader_class.HEADER_LENGTH_MAX > 0)
-
   def testFormatAuthHeader(self):
     self.assertEqual(http.auth._FormatAuthHeader("Basic", {}),
                      "Basic")
@@ -330,5 +324,76 @@ class TestReadPasswordFile(testutils.GanetiTestCase):
     self.assertEqual(users["user2"].options, ["write", "read"])
 
 
+class TestClientRequest(unittest.TestCase):
+  def testRepr(self):
+    cr = http.client.HttpClientRequest("localhost", 1234, "GET", "/version",
+                                       headers=[], post_data="Hello World")
+    self.assert_(repr(cr).startswith("<"))
+
+  def testNoHeaders(self):
+    cr = http.client.HttpClientRequest("localhost", 1234, "GET", "/version",
+                                       headers=None)
+    self.assert_(isinstance(cr.headers, list))
+    self.assertEqual(cr.headers, [])
+    self.assertEqual(cr.url, "https://localhost:1234/version")
+
+  def testOldStyleHeaders(self):
+    headers = {
+      "Content-type": "text/plain",
+      "Accept": "text/html",
+      }
+    cr = http.client.HttpClientRequest("localhost", 16481, "GET", "/vg_list",
+                                       headers=headers)
+    self.assert_(isinstance(cr.headers, list))
+    self.assertEqual(sorted(cr.headers), [
+      "Accept: text/html",
+      "Content-type: text/plain",
+      ])
+    self.assertEqual(cr.url, "https://localhost:16481/vg_list")
+
+  def testNewStyleHeaders(self):
+    headers = [
+      "Accept: text/html",
+      "Content-type: text/plain; charset=ascii",
+      "Server: httpd 1.0",
+      ]
+    cr = http.client.HttpClientRequest("localhost", 1234, "GET", "/version",
+                                       headers=headers)
+    self.assert_(isinstance(cr.headers, list))
+    self.assertEqual(sorted(cr.headers), sorted(headers))
+    self.assertEqual(cr.url, "https://localhost:1234/version")
+
+  def testPostData(self):
+    cr = http.client.HttpClientRequest("localhost", 1234, "GET", "/version",
+                                       post_data="Hello World")
+    self.assertEqual(cr.post_data, "Hello World")
+
+  def testNoPostData(self):
+    cr = http.client.HttpClientRequest("localhost", 1234, "GET", "/version")
+    self.assertEqual(cr.post_data, "")
+
+  def testIdentity(self):
+    # These should all use different connections, hence also have a different
+    # identity
+    cr1 = http.client.HttpClientRequest("localhost", 1234, "GET", "/version")
+    cr2 = http.client.HttpClientRequest("localhost", 9999, "GET", "/version")
+    cr3 = http.client.HttpClientRequest("node1", 1234, "GET", "/version")
+    cr4 = http.client.HttpClientRequest("node1", 9999, "GET", "/version")
+
+    self.assertEqual(len(set([cr1.identity, cr2.identity,
+                              cr3.identity, cr4.identity])), 4)
+
+    # But this one should have the same
+    cr1vglist = http.client.HttpClientRequest("localhost", 1234,
+                                              "GET", "/vg_list")
+    self.assertEqual(cr1.identity, cr1vglist.identity)
+
+
+class TestClient(unittest.TestCase):
+  def test(self):
+    pool = http.client.HttpClientPool(None)
+    self.assertFalse(pool._pool)
+
+
 if __name__ == '__main__':
   testutils.GanetiTestProgram()
diff --git a/test/ganeti.rpc_unittest.py b/test/ganeti.rpc_unittest.py
new file mode 100755
index 0000000..acbd1ab
--- /dev/null
+++ b/test/ganeti.rpc_unittest.py
@@ -0,0 +1,223 @@
+#!/usr/bin/python
+#
+
+# Copyright (C) 2010 Google Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# 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 General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301, USA.
+
+
+"""Script for testing ganeti.rpc"""
+
+import os
+import sys
+import unittest
+
+from ganeti import constants
+from ganeti import compat
+from ganeti import rpc
+from ganeti import http
+from ganeti import errors
+from ganeti import serializer
+
+import testutils
+
+
+class TestTimeouts(unittest.TestCase):
+  def test(self):
+    names = [name[len("call_"):] for name in dir(rpc.RpcRunner)
+             if name.startswith("call_")]
+    self.assertEqual(len(names), len(rpc._TIMEOUTS))
+    self.assertFalse([name for name in names
+                      if not (rpc._TIMEOUTS[name] is None or
+                              rpc._TIMEOUTS[name] > 0)])
+
+
+class FakeHttpPool:
+  def __init__(self, response_fn):
+    self._response_fn = response_fn
+    self.reqcount = 0
+
+  def ProcessRequests(self, reqs):
+    for req in reqs:
+      self.reqcount += 1
+      self._response_fn(req)
+
+
+class TestClient(unittest.TestCase):
+  def _GetVersionResponse(self, req):
+    self.assertEqual(req.host, "localhost")
+    self.assertEqual(req.port, 24094)
+    self.assertEqual(req.path, "/version")
+    req.success = True
+    req.resp_status_code = http.HTTP_OK
+    req.resp_body = serializer.DumpJson((True, 123))
+
+  def testVersionSuccess(self):
+    client = rpc.Client("version", None, 24094)
+    client.ConnectNode("localhost")
+    pool = FakeHttpPool(self._GetVersionResponse)
+    result = client.GetResults(http_pool=pool)
+    self.assertEqual(result.keys(), ["localhost"])
+    lhresp = result["localhost"]
+    self.assertFalse(lhresp.offline)
+    self.assertEqual(lhresp.node, "localhost")
+    self.assertFalse(lhresp.fail_msg)
+    self.assertEqual(lhresp.payload, 123)
+    self.assertEqual(lhresp.call, "version")
+    lhresp.Raise("should not raise")
+    self.assertEqual(pool.reqcount, 1)
+
+  def _GetMultiVersionResponse(self, req):
+    self.assert_(req.host.startswith("node"))
+    self.assertEqual(req.port, 23245)
+    self.assertEqual(req.path, "/version")
+    req.success = True
+    req.resp_status_code = http.HTTP_OK
+    req.resp_body = serializer.DumpJson((True, 987))
+
+  def testMultiVersionSuccess(self):
+    nodes = ["node%s" % i for i in range(50)]
+    client = rpc.Client("version", None, 23245)
+    client.ConnectList(nodes)
+
+    pool = FakeHttpPool(self._GetMultiVersionResponse)
+    result = client.GetResults(http_pool=pool)
+    self.assertEqual(sorted(result.keys()), sorted(nodes))
+
+    for name in nodes:
+      lhresp = result[name]
+      self.assertFalse(lhresp.offline)
+      self.assertEqual(lhresp.node, name)
+      self.assertFalse(lhresp.fail_msg)
+      self.assertEqual(lhresp.payload, 987)
+      self.assertEqual(lhresp.call, "version")
+      lhresp.Raise("should not raise")
+
+    self.assertEqual(pool.reqcount, len(nodes))
+
+  def _GetVersionResponseFail(self, req):
+    self.assertEqual(req.path, "/version")
+    req.success = True
+    req.resp_status_code = http.HTTP_OK
+    req.resp_body = serializer.DumpJson((False, "Unknown error"))
+
+  def testVersionFailure(self):
+    client = rpc.Client("version", None, 5903)
+    client.ConnectNode("aef9ur4i.example.com")
+    pool = FakeHttpPool(self._GetVersionResponseFail)
+    result = client.GetResults(http_pool=pool)
+    self.assertEqual(result.keys(), ["aef9ur4i.example.com"])
+    lhresp = result["aef9ur4i.example.com"]
+    self.assertFalse(lhresp.offline)
+    self.assertEqual(lhresp.node, "aef9ur4i.example.com")
+    self.assert_(lhresp.fail_msg)
+    self.assertFalse(lhresp.payload)
+    self.assertEqual(lhresp.call, "version")
+    self.assertRaises(errors.OpExecError, lhresp.Raise, "failed")
+    self.assertEqual(pool.reqcount, 1)
+
+  def _GetHttpErrorResponse(self, httperrnodes, failnodes, req):
+    self.assertEqual(req.path, "/vg_list")
+    self.assertEqual(req.port, 15165)
+
+    if req.host in httperrnodes:
+      req.success = False
+      req.error = "Node set up for HTTP errors"
+
+    elif req.host in failnodes:
+      req.success = True
+      req.resp_status_code = 404
+      req.resp_body = serializer.DumpJson({
+        "code": 404,
+        "message": "Method not found",
+        "explain": "Explanation goes here",
+        })
+    else:
+      req.success = True
+      req.resp_status_code = http.HTTP_OK
+      req.resp_body = serializer.DumpJson((True, hash(req.host)))
+
+  def testHttpError(self):
+    nodes = ["uaf6pbbv%s" % i for i in range(50)]
+
+    httperrnodes = set(nodes[1::7])
+    self.assertEqual(len(httperrnodes), 7)
+
+    failnodes = set(nodes[2::3]) - httperrnodes
+    self.assertEqual(len(failnodes), 14)
+
+    self.assertEqual(len(set(nodes) - failnodes - httperrnodes), 29)
+
+    client = rpc.Client("vg_list", None, 15165)
+    client.ConnectList(nodes)
+
+    pool = FakeHttpPool(compat.partial(self._GetHttpErrorResponse,
+                                       httperrnodes, failnodes))
+    result = client.GetResults(http_pool=pool)
+    self.assertEqual(sorted(result.keys()), sorted(nodes))
+
+    for name in nodes:
+      lhresp = result[name]
+      self.assertFalse(lhresp.offline)
+      self.assertEqual(lhresp.node, name)
+      self.assertEqual(lhresp.call, "vg_list")
+
+      if name in httperrnodes:
+        self.assert_(lhresp.fail_msg)
+        self.assertRaises(errors.OpExecError, lhresp.Raise, "failed")
+      elif name in failnodes:
+        self.assert_(lhresp.fail_msg)
+        self.assertRaises(errors.OpPrereqError, lhresp.Raise, "failed",
+                          prereq=True, ecode=errors.ECODE_INVAL)
+      else:
+        self.assertFalse(lhresp.fail_msg)
+        self.assertEqual(lhresp.payload, hash(name))
+        lhresp.Raise("should not raise")
+
+    self.assertEqual(pool.reqcount, len(nodes))
+
+  def _GetInvalidResponseA(self, req):
+    self.assertEqual(req.path, "/version")
+    req.success = True
+    req.resp_status_code = http.HTTP_OK
+    req.resp_body = serializer.DumpJson(("This", "is", "an", "invalid",
+                                         "response", "!", 1, 2, 3))
+
+  def _GetInvalidResponseB(self, req):
+    self.assertEqual(req.path, "/version")
+    req.success = True
+    req.resp_status_code = http.HTTP_OK
+    req.resp_body = serializer.DumpJson("invalid response")
+
+  def testInvalidResponse(self):
+    client = rpc.Client("version", None, 19978)
+    for fn in [self._GetInvalidResponseA, self._GetInvalidResponseB]:
+      client.ConnectNode("oqo7lanhly.example.com")
+      pool = FakeHttpPool(fn)
+      result = client.GetResults(http_pool=pool)
+      self.assertEqual(result.keys(), ["oqo7lanhly.example.com"])
+      lhresp = result["oqo7lanhly.example.com"]
+      self.assertFalse(lhresp.offline)
+      self.assertEqual(lhresp.node, "oqo7lanhly.example.com")
+      self.assert_(lhresp.fail_msg)
+      self.assertFalse(lhresp.payload)
+      self.assertEqual(lhresp.call, "version")
+      self.assertRaises(errors.OpExecError, lhresp.Raise, "failed")
+      self.assertEqual(pool.reqcount, 1)
+
+
+if __name__ == "__main__":
+  testutils.GanetiTestProgram()
