@@ -1,14 +1,22 @@
11import logging
22import pycurl
3+import threading
4+from itertools import cycle
35from StringIO import StringIO
6+from time import time
47from urllib import urlencode
58
69
10+CURL_SHARE = pycurl.CurlShare()
11+CURL_SHARE.setopt(pycurl.SH_SHARE, pycurl.LOCK_DATA_DNS)
12+MULTI_HANDLER = pycurl.CurlMulti()
13+
14+
715def __test_callback(*args):
816 print "callback called"
917
1018
11−def get_new_connection(source_address=None):
19+def get_connection(source_address=None):
1220 # pycurl initialization
1321 h = pycurl.Curl()
1422
@@ -18,46 +26,14 @@ def get_new_connection(source_address=None):
1826 # consistently use ipv4
1927 h.setopt(h.IPRESOLVE, pycurl.IPRESOLVE_V4)
2028
29+ h.setopt(pycurl.SHARE, CURL_SHARE)
30+
2131 if source_address:
2232 h.setopt(h.INTERFACE, source_address)
2333
2434 return h
2535
2636
27−class ConnectionSources(object):
28− def __init__(self, sources=None):
29− self.sources = []
30− if sources:
31− for s in sources:
32− self.sources.append(get_new_connection(s))
33− else:
34− self.sources.append(get_new_connection())
35− self.ptr = 0
36−
37− def get_source(self):
38− source = self.sources[self.ptr]
39− self.ptr = (self.ptr + 1) % len(self.sources)
40− return source
41−
42−
43−class ConnectionCache(object):
44− def __init__(self, source_ips=None):
45− self.source_ips = source_ips if source_ips else None
46− self.connections = {}
47−
48− def get_connection(self, key):
49− if key in self.connections:
50− return self.connections[key].get_source()
51−
52− sources = ConnectionSources(self.source_ips)
53− self.connections[key] = sources
54− return sources.get_source()
55−
56−
57−CONNECTION_CACHE = ConnectionCache()
58−MULTI_HANDLER = pycurl.CurlMulti()
59−
60−
6137class RequestContainer(object):
6238 def __init__(self,
6339 url,
@@ -110,29 +86,27 @@ class ResponseContainer(object):
11086
11187
11288class MultiRequest(object):
113− def __init__(self, connection_cache=None, multi_handler=None):
89+ def __init__(self, multi_handler=None, source_ips=None):
11490 self.requests = {}
11591
116− if connection_cache:
117− self.connection_cache = connection_cache
118− else:
119− self.connection_cache = CONNECTION_CACHE
120−
12192 if multi_handler:
12293 self._curl_multi_handler = multi_handler
12394 else:
12495 self._curl_multi_handler = MULTI_HANDLER
12596
126− def add(self, connection_name, url, **kwargs):
127− handle = self.connection_cache.get_connection(connection_name)
97+ self.source_ips = cycle(source_ips) if source_ips else cycle([None])
98+
99+ def add(self, url, **kwargs):
100+ handle = get_connection(next(self.source_ips))
128101 request_container = RequestContainer(url, handle, **kwargs)
129102 try:
130103 self._curl_multi_handler.add_handle(handle)
131104 except:
105+ print 'meep'
132106 pass
133107 self.requests[handle] = request_container
134108
135− def perform_requests(self):
109+ def send_requests(self):
136110 select_timeout = 0.1
137111
138112 # set timeout
@@ -154,6 +128,7 @@ class MultiRequest(object):
154128 else:
155129 h.unsetopt(h.COOKIE)
156130
131+ search_start = time()
157132 handles_num = len(self.requests)
158133 while handles_num:
159134 self._curl_multi_handler.select(select_timeout)
@@ -164,23 +139,35 @@ class MultiRequest(object):
164139 _, success_list, error_list = self._curl_multi_handler.info_read()
165140 # calling callbacks
166141 for h in success_list:
167− self.requests[h].finish()
142+ th = threading.Thread(
143+ target=self.requests[h].finish(),
144+ name='search_request',
145+ )
146+ th.start()
147+ # self.requests[h].finish()
168148 for h, err_code, err_string in error_list:
169149 logging.warn('Error on %s: "%s"', self.requests[h].url, err_string)
170150 handles_num -= len(success_list) + len(error_list)
171151 if ret != pycurl.E_CALL_MULTI_PERFORM:
172152 break
173153
154+ for th in threading.enumerate():
155+ if th.name == 'search_request':
156+ remaining_time = max(0.0, timeout - (time() - search_start))
157+ th.join(remaining_time)
158+ if th.isAlive():
159+ logging.warning('engine timeout: {0}'.format(th._engine_name))
160+
174161 # self._curl_multi_handler.close()
175162 return self.requests.values()
176163
177164
178165if __name__ == '__main__':
179166 r = MultiRequest()
180− r.add('a', 'http://httpbin.org/delay/0', headers={'User-Agent': 'x'})
181− r.add('d', 'http://127.0.0.1:7777/', headers={'User-Agent': 'x'})
182− r.add('b', 'http://httpbin.org/delay/0', cookies={'as': 'sa', 'bb': 'cc'})
183− r.add('c', 'http://httpbin.org/delay/0', callback=__test_callback, timeout=1.0, headers={'User-Agent': 'x'})
184− for v in r.perform_requests():
167+ r.add('http://httpbin.org/delay/0', headers={'User-Agent': 'x'})
168+ r.add('http://127.0.0.1:7777/', headers={'User-Agent': 'x'})
169+ r.add('http://httpbin.org/delay/0', cookies={'as': 'sa', 'bb': 'cc'})
170+ r.add('http://httpbin.org/delay/0', callback=__test_callback, timeout=1.0, headers={'User-Agent': 'x'})
171+ for v in r.send_requests():
185172 print v.url
186173 print v.response.text