@@ -1,14 +1,54 @@
1+import atexit
12import sys
3+import threading
4+import asyncio
5+import logging
6+import concurrent.futures
27from time import time
38from itertools import cycle
4−from threading import local
59
6−import requests
10+import httpcore
11+import httpx
12+import h2.exceptions
13+from httpx_socks import AsyncProxyTransport
14+from python_socks import parse_proxy_url
15+import python_socks._errors
716
817from searx import settings
918from searx import logger
1019from searx.raise_for_httperror import raise_for_httperror
1120
21+# Optional uvloop (support Python 3.6)
22+try:
23+ import uvloop
24+except ImportError:
25+ pass
26+else:
27+ uvloop.install()
28+
29+# queue.SimpleQueue: Support Python 3.6
30+try:
31+ from queue import SimpleQueue
32+except ImportError:
33+ from queue import Empty
34+ from collections import deque
35+
36+ class SimpleQueue:
37+ """Minimal backport of queue.SimpleQueue"""
38+
39+ def __init__(self):
40+ self._queue = deque()
41+ self._count = threading.Semaphore(0)
42+
43+ def put(self, item):
44+ self._queue.append(item)
45+ self._count.release()
46+
47+ def get(self, timeout=None):
48+ if not self._count.acquire(True, timeout):
49+ raise Empty
50+ return self._queue.popleft()
51+
1252
1353logger = logger.getChild('poolrequests')
1454
@@ -31,99 +71,63 @@ if not getattr(ssl, "HAS_SNI", False):
3171 sys.exit(1)
3272
3373
34−class HTTPAdapterWithConnParams(requests.adapters.HTTPAdapter):
35−
36− def __init__(self, pool_connections=requests.adapters.DEFAULT_POOLSIZE,
37− pool_maxsize=requests.adapters.DEFAULT_POOLSIZE,
38− max_retries=requests.adapters.DEFAULT_RETRIES,
39− pool_block=requests.adapters.DEFAULT_POOLBLOCK,
40− **conn_params):
41− if max_retries == requests.adapters.DEFAULT_RETRIES:
42− self.max_retries = requests.adapters.Retry(0, read=False)
43− else:
44− self.max_retries = requests.adapters.Retry.from_int(max_retries)
45− self.config = {}
46− self.proxy_manager = {}
47−
48− super().__init__()
49−
50− self._pool_connections = pool_connections
51− self._pool_maxsize = pool_maxsize
52− self._pool_block = pool_block
53− self._conn_params = conn_params
54−
55− self.init_poolmanager(pool_connections, pool_maxsize, block=pool_block, **conn_params)
56−
57− def __setstate__(self, state):
58− # Can't handle by adding 'proxy_manager' to self.__attrs__ because
59− # because self.poolmanager uses a lambda function, which isn't pickleable.
60− self.proxy_manager = {}
61− self.config = {}
62−
63− for attr, value in state.items():
64− setattr(self, attr, value)
74+LOOP = None
75+CLIENTS = dict()
76+THREADLOCAL = threading.local()
77+LIMITS = httpx.Limits(
78+ # Magic number kept from previous code
79+ max_connections=settings['outgoing'].get('pool_connections', 100),
80+ # Picked from constructor
81+ max_keepalive_connections=settings['outgoing'].get('pool_maxsize', 10),
82+ #
83+ keepalive_expiry=settings['outgoing'].get('keepalive_expiry', 5.0)
84+)
85+# default parameters for AsyncHTTPTransport
86+# see https://github.com/encode/httpx/blob/e05a5372eb6172287458b37447c30f650047e1b8/httpx/_transports/default.py#L108-L121 # noqa
87+TRANSPORT_KWARGS = {
88+ 'http2': settings['outgoing'].get('http2', False),
89+ 'retries': 0,
90+ 'trust_env': False,
91+ 'backend': 'asyncio'
92+}
93+# requests compatibility when reading proxy settings from settings.yml
94+PROXY_PATTERN_MAPPING = {
95+ 'http': 'https://',
96+ 'https:': 'https://'
97+}
98+# default maximum redirect
99+# from https://github.com/psf/requests/blob/8c211a96cdbe9fe320d63d9e1ae15c5c07e179f8/requests/models.py#L55
100+DEFAULT_REDIRECT_LIMIT = 30
65101
66− self.init_poolmanager(self._pool_connections, self._pool_maxsize,
67− block=self._pool_block, **self._conn_params)
68102
69−
70−threadLocal = local()
71−connect = settings['outgoing'].get('pool_connections', 100) # Magic number kept from previous code
72−maxsize = settings['outgoing'].get('pool_maxsize', requests.adapters.DEFAULT_POOLSIZE) # Picked from constructor
73103if settings['outgoing'].get('source_ips'):
74− http_adapters = cycle(HTTPAdapterWithConnParams(pool_connections=connect, pool_maxsize=maxsize,
75− source_address=(source_ip, 0))
76− for source_ip in settings['outgoing']['source_ips'])
77− https_adapters = cycle(HTTPAdapterWithConnParams(pool_connections=connect, pool_maxsize=maxsize,
78− source_address=(source_ip, 0))
79− for source_ip in settings['outgoing']['source_ips'])
104+ LOCAL_ADDRESS_CYCLE = cycle(settings['outgoing'].get('source_ips'))
80105else:
81− http_adapters = cycle((HTTPAdapterWithConnParams(pool_connections=connect, pool_maxsize=maxsize), ))
82− https_adapters = cycle((HTTPAdapterWithConnParams(pool_connections=connect, pool_maxsize=maxsize), ))
83−
84−
85−class SessionSinglePool(requests.Session):
86−
87− def __init__(self):
88− super().__init__()
89−
90− # reuse the same adapters
91− self.adapters.clear()
92−
93− https_adapter = threadLocal.__dict__.setdefault('https_adapter', next(https_adapters))
94− self.mount('https://', https_adapter)
95− if get_enable_http_protocol():
96− http_adapter = threadLocal.__dict__.setdefault('http_adapter', next(http_adapters))
97− self.mount('http://', http_adapter)
98−
99− def close(self):
100− """Call super, but clear adapters since there are managed globaly"""
101− self.adapters.clear()
102− super().close()
106+ LOCAL_ADDRESS_CYCLE = cycle((None, ))
103107
104108
105109def set_timeout_for_thread(timeout, start_time=None):
106− threadLocal.timeout = timeout
107− threadLocal.start_time = start_time
110+ THREADLOCAL.timeout = timeout
111+ THREADLOCAL.start_time = start_time
108112
109113
110114def set_enable_http_protocol(enable_http):
111− threadLocal.enable_http = enable_http
115+ THREADLOCAL.enable_http = enable_http
112116
113117
114118def get_enable_http_protocol():
115119 try:
116− return threadLocal.enable_http
120+ return THREADLOCAL.enable_http
117121 except AttributeError:
118122 return False
119123
120124
121125def reset_time_for_thread():
122− threadLocal.total_time = 0
126+ THREADLOCAL.total_time = 0
123127
124128
125129def get_time_for_thread():
126− return threadLocal.total_time
130+ return THREADLOCAL.total_time
127131
128132
129133def get_proxy_cycles(proxy_settings):
@@ -152,22 +156,197 @@ def get_global_proxies():
152156 return get_proxies(GLOBAL_PROXY_CYCLES)
153157
154158
159+async def close_connections_for_url(connection_pool: httpcore.AsyncConnectionPool, url: httpcore._utils.URL):
160+ origin = httpcore._utils.url_to_origin(url)
161+ logger.debug('Drop connections for %r', origin)
162+ connections_to_close = connection_pool._connections_for_origin(origin)
163+ for connection in connections_to_close:
164+ await connection_pool._remove_from_pool(connection)
165+ try:
166+ await connection.aclose()
167+ except httpcore.NetworkError as e:
168+ logger.warning('Error closing an existing connection', exc_info=e)
169+
170+
171+class AsyncHTTPTransportNoHttp(httpcore.AsyncHTTPTransport):
172+ """Block HTTP request"""
173+
174+ async def arequest(self, method, url, headers=None, stream=None, ext=None):
175+ raise httpcore.UnsupportedProtocol("HTTP protocol is disabled")
176+
177+
178+class AsyncProxyTransportFixed(AsyncProxyTransport):
179+ """Fix httpx_socks.AsyncProxyTransport
180+
181+ Map python_socks exceptions to httpcore.ProxyError
182+
183+ Map socket.gaierror to httpcore.ConnectError
184+
185+ Note: keepalive_expiry is ignored, AsyncProxyTransport should call:
186+ * self._keepalive_sweep()
187+ * self._response_closed(self, connection)
188+
189+ Note: AsyncProxyTransport inherit from AsyncConnectionPool
190+
191+ Note: the API is going to change on httpx 0.18.0
192+ see https://github.com/encode/httpx/pull/1522
193+ """
194+
195+ async def arequest(self, method, url, headers=None, stream=None, ext=None):
196+ retry = 2
197+ while retry > 0:
198+ retry -= 1
199+ try:
200+ return await super().arequest(method, url, headers, stream, ext)
201+ except (python_socks._errors.ProxyConnectionError,
202+ python_socks._errors.ProxyTimeoutError,
203+ python_socks._errors.ProxyError) as e:
204+ raise httpcore.ProxyError(e)
205+ except OSError as e:
206+ # socket.gaierror when DNS resolution fails
207+ raise httpcore.NetworkError(e)
208+ except httpcore.RemoteProtocolError as e:
209+ # in case of httpcore.RemoteProtocolError: Server disconnected
210+ await close_connections_for_url(self, url)
211+ logger.warning('httpcore.RemoteProtocolError: retry', exc_info=e)
212+ # retry
213+ except (httpcore.NetworkError, httpcore.ProtocolError) as e:
214+ # httpcore.WriteError on HTTP/2 connection leaves a new opened stream
215+ # then each new request creates a new stream and raise the same WriteError
216+ await close_connections_for_url(self, url)
217+ raise e
218+
219+
220+class AsyncHTTPTransportFixed(httpx.AsyncHTTPTransport):
221+ """Fix httpx.AsyncHTTPTransport"""
222+
223+ async def arequest(self, method, url, headers=None, stream=None, ext=None):
224+ retry = 2
225+ while retry > 0:
226+ retry -= 1
227+ try:
228+ return await super().arequest(method, url, headers, stream, ext)
229+ except OSError as e:
230+ # socket.gaierror when DNS resolution fails
231+ raise httpcore.ConnectError(e)
232+ except httpcore.CloseError as e:
233+ # httpcore.CloseError: [Errno 104] Connection reset by peer
234+ # raised by _keepalive_sweep()
235+ # from https://github.com/encode/httpcore/blob/4b662b5c42378a61e54d673b4c949420102379f5/httpcore/_backends/asyncio.py#L198 # noqa
236+ await close_connections_for_url(self._pool, url)
237+ logger.warning('httpcore.CloseError: retry', exc_info=e)
238+ # retry
239+ except httpcore.RemoteProtocolError as e:
240+ # in case of httpcore.RemoteProtocolError: Server disconnected
241+ await close_connections_for_url(self._pool, url)
242+ logger.warning('httpcore.RemoteProtocolError: retry', exc_info=e)
243+ # retry
244+ except (httpcore.ProtocolError, httpcore.NetworkError) as e:
245+ await close_connections_for_url(self._pool, url)
246+ raise e
247+
248+
249+def get_transport_for_socks_proxy(verify, local_address, proxy_url):
250+ global LOOP, LIMITS, TRANSPORT_KWARGS
251+ # support socks5h (requests compatibility):
252+ # https://requests.readthedocs.io/en/master/user/advanced/#socks
253+ # socks5:// hostname is resolved on client side
254+ # socks5h:// hostname is resolved on proxy side
255+ rdns = False
256+ socks5h = 'socks5h://'
257+ if proxy_url.startswith(socks5h):
258+ proxy_url = 'socks5://' + proxy_url[len(socks5h):]
259+ rdns = True
260+
261+ proxy_type, proxy_host, proxy_port, proxy_username, proxy_password = parse_proxy_url(proxy_url)
262+
263+ return AsyncProxyTransportFixed(proxy_type=proxy_type, proxy_host=proxy_host, proxy_port=proxy_port,
264+ username=proxy_username, password=proxy_password,
265+ rdns=rdns,
266+ loop=LOOP,
267+ verify=verify,
268+ local_address=local_address,
269+ max_connections=LIMITS.max_connections,
270+ max_keepalive_connections=LIMITS.max_keepalive_connections,
271+ keepalive_expiry=LIMITS.keepalive_expiry,
272+ **TRANSPORT_KWARGS)
273+
274+
275+def get_transport(verify, local_address, proxy_url):
276+ global LIMITS
277+ return AsyncHTTPTransportFixed(verify=verify,
278+ local_address=local_address,
279+ limits=LIMITS,
280+ proxy=httpx._config.Proxy(proxy_url) if proxy_url else None,
281+ **TRANSPORT_KWARGS)
282+
283+
284+def iter_proxies(proxies):
285+ # https://www.python-httpx.org/compatibility/#proxy-keys
286+ if isinstance(proxies, str):
287+ yield 'all://', proxies
288+ elif isinstance(proxies, dict):
289+ for pattern, proxy_url in proxies.items():
290+ pattern = PROXY_PATTERN_MAPPING.get(pattern, pattern)
291+ yield pattern, proxy_url
292+
293+
294+def new_client(verify, local_address, proxies, max_redirects, enable_http):
295+ # See https://www.python-httpx.org/advanced/#routing
296+ mounts = {}
297+ for pattern, proxy_url in iter_proxies(proxies):
298+ if not enable_http and (pattern == 'http' or pattern.startswith('http://')):
299+ continue
300+ if proxy_url.startswith('socks4://') \
301+ or proxy_url.startswith('socks5://') \
302+ or proxy_url.startswith('socks5h://'):
303+ mounts[pattern] = get_transport_for_socks_proxy(verify, local_address, proxy_url)
304+ else:
305+ mounts[pattern] = get_transport(verify, local_address, proxy_url)
306+
307+ if not enable_http:
308+ mounts['http://'] = AsyncHTTPTransportNoHttp()
309+
310+ transport = get_transport(verify, local_address, None)
311+ return httpx.AsyncClient(transport=transport, mounts=mounts, max_redirects=max_redirects)
312+
313+
314+def get_client(verify, local_address, proxies, max_redirects, allow_http):
315+ global CLIENTS
316+ key = (verify, local_address, repr(proxies), max_redirects, allow_http)
317+ if key not in CLIENTS:
318+ CLIENTS[key] = new_client(verify, local_address, proxies, max_redirects, allow_http)
319+ return CLIENTS[key]
320+
321+
322+async def send_request(method, url, enable_http, kwargs):
323+ if isinstance(url, bytes):
324+ url = url.decode()
325+
326+ verify = kwargs.pop('verify', True)
327+ local_address = next(LOCAL_ADDRESS_CYCLE)
328+ proxies = kwargs.pop('proxies', None) or get_global_proxies()
329+ max_redirects = kwargs.pop('max_redirects', DEFAULT_REDIRECT_LIMIT)
330+
331+ client = get_client(verify, local_address, proxies, max_redirects, enable_http)
332+ response = await client.request(method.upper(), url, **kwargs)
333+
334+ # requests compatibility
335+ # see also https://www.python-httpx.org/compatibility/#checking-for-4xx5xx-responses
336+ response.ok = not response.is_error
337+
338+ return response
339+
340+
155341def request(method, url, **kwargs):
156342 """same as requests/requests/api.py request(...)"""
157343 time_before_request = time()
158344
159− # session start
160− session = SessionSinglePool()
161−
162− # proxies
163− if not kwargs.get('proxies'):
164− kwargs['proxies'] = get_global_proxies()
165−
166345 # timeout
167346 if 'timeout' in kwargs:
168347 timeout = kwargs['timeout']
169348 else:
170− timeout = getattr(threadLocal, 'timeout', None)
349+ timeout = getattr(THREADLOCAL, 'timeout', None)
171350 if timeout is not None:
172351 kwargs['timeout'] = timeout
173352
@@ -178,24 +357,23 @@ def request(method, url, **kwargs):
178357 del kwargs['raise_for_httperror']
179358
180359 # do request
181− response = session.request(method=method, url=url, **kwargs)
182−
183− time_after_request = time()
184−
185− # is there a timeout for this engine ?
186− if timeout is not None:
187− timeout_overhead = 0.2 # seconds
188− # start_time = when the user request started
189− start_time = getattr(threadLocal, 'start_time', time_before_request)
190− search_duration = time_after_request - start_time
191− if search_duration > timeout + timeout_overhead:
192− raise requests.exceptions.Timeout(response=response)
193−
194− # session end
195− session.close()
196−
197− if hasattr(threadLocal, 'total_time'):
198− threadLocal.total_time += time_after_request - time_before_request
360+ future = asyncio.run_coroutine_threadsafe(send_request(method, url, get_enable_http_protocol(), kwargs), LOOP)
361+ try:
362+ if timeout:
363+ timeout += 0.2 # overhead
364+ start_time = getattr(THREADLOCAL, 'start_time', time_before_request)
365+ if start_time:
366+ timeout -= time() - start_time
367+
368+ response = future.result(timeout or 120)
369+ except concurrent.futures.TimeoutError as e:
370+ raise httpx.TimeoutException('Timeout', request=None) from e
371+
372+ # update total_time.
373+ # See get_time_for_thread() and reset_time_for_thread()
374+ if hasattr(THREADLOCAL, 'total_time'):
375+ time_after_request = time()
376+ THREADLOCAL.total_time += time_after_request - time_before_request
199377
200378 # raise an exception
201379 if check_for_httperror:
@@ -204,6 +382,49 @@ def request(method, url, **kwargs):
204382 return response
205383
206384
385+async def stream_chunk_to_queue(method, url, q, **kwargs):
386+ verify = kwargs.pop('verify', True)
387+ local_address = next(LOCAL_ADDRESS_CYCLE)
388+ proxies = kwargs.pop('proxies', None) or get_global_proxies()
389+ # "30" from requests:
390+ # https://github.com/psf/requests/blob/8c211a96cdbe9fe320d63d9e1ae15c5c07e179f8/requests/models.py#L55
391+ max_redirects = kwargs.pop('max_redirects', 30)
392+ client = get_client(verify, local_address, proxies, max_redirects, True)
393+ try:
394+ async with client.stream(method, url, **kwargs) as response:
395+ q.put(response)
396+ async for chunk in response.aiter_bytes(65536):
397+ if len(chunk) > 0:
398+ q.put(chunk)
399+ except (httpx.HTTPError, OSError, h2.exceptions.ProtocolError) as e:
400+ q.put(e)
401+ finally:
402+ q.put(None)
403+
404+
405+def stream(method, url, **kwargs):
406+ """Replace httpx.stream.
407+
408+ Usage:
409+ stream = poolrequests.stream(...)
410+ response = next(stream)
411+ for chunk in stream:
412+ ...
413+
414+ httpx.Client.stream requires to write the httpx.HTTPTransport version of the
415+ the httpx.AsyncHTTPTransport declared above.
416+ """
417+ q = SimpleQueue()
418+ future = asyncio.run_coroutine_threadsafe(stream_chunk_to_queue(method, url, q, **kwargs), LOOP)
419+ chunk_or_exception = q.get(timeout=60)
420+ while chunk_or_exception is not None:
421+ if isinstance(chunk_or_exception, Exception):
422+ raise chunk_or_exception
423+ yield chunk_or_exception
424+ chunk_or_exception = q.get(timeout=60)
425+ return future.result()
426+
427+
207428def get(url, **kwargs):
208429 kwargs.setdefault('allow_redirects', True)
209430 return request('get', url, **kwargs)
@@ -233,3 +454,97 @@ def patch(url, data=None, **kwargs):
233454
234455def delete(url, **kwargs):
235456 return request('delete', url, **kwargs)
457+
458+
459+def init():
460+ # log
461+ for logger_name in ('hpack.hpack', 'hpack.table'):
462+ logging.getLogger(logger_name).setLevel(logging.WARNING)
463+
464+ # loop
465+ def loop_thread():
466+ global LOOP
467+ LOOP = asyncio.new_event_loop()
468+ LOOP.run_forever()
469+
470+ th = threading.Thread(
471+ target=loop_thread,
472+ name='asyncio_loop',
473+ daemon=True,
474+ )
475+ th.start()
476+
477+
478+@atexit.register
479+def done():
480+ """Close all HTTP client
481+
482+ Avoid a warning at exit
483+ see https://github.com/encode/httpx/blob/1a6e254f72d9fd5694a1c10a28927e193ab4f76b/httpx/_client.py#L1785
484+ """
485+ global LOOP
486+
487+ async def close_client(client):
488+ try:
489+ await client.aclose()
490+ except httpx.HTTPError:
491+ pass
492+
493+ async def close_clients():
494+ await asyncio.gather(*[close_client(client) for client in CLIENTS.values()], return_exceptions=False)
495+ future = asyncio.run_coroutine_threadsafe(close_clients(), LOOP)
496+ # wait 3 seconds to close the HTTP clients
497+ future.result(3)
498+
499+
500+init()
501+
502+
503+# ## TEMPORARY DEBUG ##
504+
505+
506+def debug_connection(connection):
507+ now = LOOP.time()
508+ expired = (connection.state == httpcore._async.base.ConnectionState.IDLE
509+ and connection.expires_at is not None
510+ and now >= connection.expires_at)
511+ return connection.info()\
512+ + (', connect_failed' if connection.connect_failed else '')\
513+ + (', expired' if expired else '')
514+
515+
516+def debug_origin(origin):
517+ return origin[0].decode() + '://' + origin[1].decode() + ':' + str(origin[2])
518+
519+
520+def debug_transport(transport):
521+ result = {
522+ '__class__': str(transport.__class__.__name__)
523+ }
524+ if isinstance(transport, (httpx.AsyncHTTPTransport, AsyncHTTPTransportFixed)):
525+ pool = transport._pool
526+ result['__pool_class__'] = str(pool.__class__.__name__)
527+ if isinstance(pool, httpcore.AsyncConnectionPool):
528+ for origin, connections in pool._connections.items():
529+ result[debug_origin(origin)] = [debug_connection(connection) for connection in connections]
530+ return result
531+ elif isinstance(transport, AsyncProxyTransportFixed):
532+ for origin, connections in transport._connections.items():
533+ result[debug_origin(origin)] = [debug_connection(connection) for connection in connections]
534+ return result
535+ return result
536+
537+
538+def debug_asyncclient(client, key=None):
539+ result = {}
540+ if key:
541+ result['__key__'] = [k if isinstance(k, (str, int, float, bool, type(None))) else repr(k) for k in key]
542+ result['__default__'] = debug_transport(client._transport)
543+ for urlpattern, transport in client._mounts.items():
544+ result[urlpattern.pattern] = debug_transport(transport)
545+ return result
546+
547+
548+def debug_asyncclients():
549+ global CLIENTS
550+ return [debug_asyncclient(client, key) for key, client in CLIENTS.items()]