new file mode 100644

@@ -0,0 +1,57 @@

+import httplib

+from urlparse import urlparse

+from threading import Semaphore

+from logging import debug, info, warning, error

+

+from Config import Config

+

+__all__ = [ "ConnMan" ]

+

+class http_connection(object):

+    def __init__(self, id, hostname, ssl):

+        self.hostname = hostname

+        self.ssl = ssl

+        self.id = id

+        self.counter = 0

+        if not ssl:

+            self.c = httplib.HTTPConnection(hostname)

+        else:

+            self.c = httplib.HTTPSConnection(hostname)

+

+class ConnMan(object):

+    conn_pool_sem = Semaphore()

+    conn_pool = {}

+    conn_max_counter = 800    ## AWS closes connection after some ~90 requests

+

+    @staticmethod

+    def get(hostname, ssl = None):

+        if ssl == None:

+            ssl = Config().use_https

+        conn = None

+        conn_id = "http%s://%s" % (ssl and "s" or "", hostname)

+        ConnMan.conn_pool_sem.acquire()

+        if not ConnMan.conn_pool.has_key(conn_id):

+            ConnMan.conn_pool[conn_id] = []

+        if len(ConnMan.conn_pool[conn_id]):

+            conn = ConnMan.conn_pool[conn_id].pop()

+            debug("ConnMan.get(): re-using connection: %s#%d" % (conn.id, conn.counter))

+        ConnMan.conn_pool_sem.release()

+        if not conn:

+            debug("ConnMan.get(): creating new connection: %s" % conn_id)

+            conn = http_connection(conn_id, hostname, ssl)

+            conn.c.connect()

+        conn.counter += 1

+        return conn

+

+    @staticmethod

+    def put(conn):

+        if conn.counter >= ConnMan.conn_max_counter:

+            conn.c.close()

+            debug("ConnMan.put(): closing over-used connection")

+            return

+

+        ConnMan.conn_pool_sem.acquire()

+        ConnMan.conn_pool[conn.id].append(conn)

+        ConnMan.conn_pool_sem.release()

+        debug("ConnMan.put(): connection put back to pool (%s#%d)" % (conn.id, conn.counter))

+

@@ -27,6 +27,7 @@ from Config import Config

 from Exceptions import *

 from MultiPart import MultiPartUpload

 from S3Uri import S3Uri

+from ConnMan import ConnMan

 try:

     import magic, gzip

@@ -668,18 +669,18 @@ class S3(object):

             # "Stringify" all headers

             for header in headers.keys():

                 headers[header] = str(headers[header])

-            conn = self.get_connection(resource['bucket'])

+            conn = ConnMan.get(self.get_hostname(resource['bucket']))

             uri = self.format_uri(resource)

             debug("Sending request method_string=%r, uri=%r, headers=%r, body=(%i bytes)" % (method_string, uri, headers, len(body or "")))

-            conn.request(method_string, uri, body, headers)

+            conn.c.request(method_string, uri, body, headers)

             response = {}

-            http_response = conn.getresponse()

+            http_response = conn.c.getresponse()

             response["status"] = http_response.status

             response["reason"] = http_response.reason

             response["headers"] = convertTupleListToDict(http_response.getheaders())

             response["data"] =  http_response.read()

             debug("Response: " + str(response))

-            conn.close()

+            ConnMan.put(conn)

         except Exception, e:

             if retries:

                 warning("Retrying failed request: %s (%s)" % (resource['uri'], e))

@@ -722,12 +723,11 @@ class S3(object):

             info("Sending file '%s', please wait..." % file.name)

         timestamp_start = time.time()

         try:

-            conn = self.get_connection(resource['bucket'])

-            conn.connect()

-            conn.putrequest(method_string, self.format_uri(resource))

+            conn = ConnMan.get(self.get_hostname(resource['bucket']))

+            conn.c.putrequest(method_string, self.format_uri(resource))

             for header in headers.keys():

-                conn.putheader(header, str(headers[header]))

-            conn.endheaders()

+                conn.c.putheader(header, str(headers[header]))

+            conn.c.endheaders()

         except Exception, e:

             if self.config.progress_meter:

                 progress.done("failed")

@@ -750,7 +750,7 @@ class S3(object):

                 else:

                     data = buffer

                 md5_hash.update(data)

-                conn.send(data)

+                conn.c.send(data)

                 if self.config.progress_meter:

                     progress.update(delta_position = len(data))

                 size_left -= len(data)

@@ -758,13 +758,13 @@ class S3(object):

                     time.sleep(throttle)

             md5_computed = md5_hash.hexdigest()

             response = {}

-            http_response = conn.getresponse()

+            http_response = conn.c.getresponse()

             response["status"] = http_response.status

             response["reason"] = http_response.reason

             response["headers"] = convertTupleListToDict(http_response.getheaders())

             response["data"] = http_response.read()

             response["size"] = size_total

-            conn.close()

+            ConnMan.put(conn)

             debug(u"Response: %s" % response)

         except Exception, e:

             if self.config.progress_meter:

@@ -787,7 +787,7 @@ class S3(object):

         response["speed"] = response["elapsed"] and float(response["size"]) / response["elapsed"] or float(-1)

         if self.config.progress_meter:

-            ## The above conn.close() takes some time -> update() progress meter

+            ## Finalising the upload takes some time -> update() progress meter

             ## to correct the average speed. Otherwise people will complain that

             ## 'progress' and response["speed"] are inconsistent ;-)

             progress.update()

@@ -862,17 +862,16 @@ class S3(object):

             info("Receiving file '%s', please wait..." % stream.name)

         timestamp_start = time.time()

         try:

-            conn = self.get_connection(resource['bucket'])

-            conn.connect()

-            conn.putrequest(method_string, self.format_uri(resource))

+            conn = ConnMan.get(self.get_hostname(resource['bucket']))

+            conn.c.putrequest(method_string, self.format_uri(resource))

             for header in headers.keys():

-                conn.putheader(header, str(headers[header]))

+                conn.c.putheader(header, str(headers[header]))

             if start_position > 0:

                 debug("Requesting Range: %d .. end" % start_position)

-                conn.putheader("Range", "bytes=%d-" % start_position)

-            conn.endheaders()

+                conn.c.putheader("Range", "bytes=%d-" % start_position)

+            conn.c.endheaders()

             response = {}

-            http_response = conn.getresponse()

+            http_response = conn.c.getresponse()

             response["status"] = http_response.status

             response["reason"] = http_response.reason

             response["headers"] = convertTupleListToDict(http_response.getheaders())

@@ -928,7 +927,7 @@ class S3(object):

                 ## Call progress meter from here...

                 if self.config.progress_meter:

                     progress.update(delta_position = len(data))

-            conn.close()

+            ConnMan.put(conn)

         except Exception, e:

             if self.config.progress_meter:

                 progress.done("failed")