fix too many open files

Author: DevAlone

async_requests.py

@@ -31,7 +31,9 @@ async def request(method, url, **kwargs):
 
     # headers={'User-Agent': get_random_user_agent()}
     if 'headers' not in kwargs:
-        kwargs['headers'] = {'User-Agent': get_random_user_agent()}
+        kwargs['headers'] = {
+            'User-Agent': get_random_user_agent()
+        }
     elif 'User-Agent' not in kwargs['headers']:
             kwargs['headers']['User-Agent'] = get_random_user_agent()
 

processor.py

@@ -1,3 +1,5 @@
+import aiohttp
+
 import collectors_list
 import proxy_utils
 
@@ -7,13 +9,16 @@
 from models import session
 from models import get_or_create
 
+
 import sqlalchemy
 import sqlalchemy.exc
 import asyncio
 import time
 import re
 import logging
 
+from aiosocks.connector import ProxyConnector, ProxyClientRequest
+
 
 # TODO: add ipv6 addresses
 PROXY_VALIDATE_REGEX = r"^((?P<protocol>(http|socks4|socks5))://)?" \
@@ -72,14 +77,18 @@ async def consumer(self):
         while True:
             await asyncio.sleep(0.1)
             i = 0
-            while not self.queue.empty() and i <= settings.CONCURRENT_TASKS_COUNT:
-                proxy_data = self.queue.get_nowait()
-                tasks.append(self.process_proxy(*proxy_data))
-                self.queue.task_done()
+            async with aiohttp.ClientSession(
+                    connector=ProxyConnector(remote_resolve=False),
+                    request_class=ProxyClientRequest
+            ) as aiohttp_proxy_check_session:
+                while not self.queue.empty() and i <= settings.CONCURRENT_TASKS_COUNT:
+                    proxy_data = self.queue.get_nowait()
+                    tasks.append(self.process_proxy(*proxy_data, aiohttp_proxy_check_session))
+                    self.queue.task_done()
 
-            if tasks:
-                await asyncio.wait(tasks)
-                tasks.clear()
+                if tasks:
+                    await asyncio.wait(tasks)
+                    tasks.clear()
 
     async def producer(self):
         while True:
@@ -190,7 +199,8 @@ async def process_raw_proxies(self, proxies, collector_id):
                         collector_id,
                     ))
 
-    async def process_proxy(self, raw_protocol: int, auth_data: str, domain: str, port: int, collector_id: int):
+    async def process_proxy(self, raw_protocol: int, auth_data: str, domain: str, port: int, collector_id: int,
+                            aiohttp_proxy_check_session):
         self.logger.debug("start processing proxy {}://{}@{}:{} with collector id {}".format(
             raw_protocol, auth_data, domain, port, collector_id))
 
@@ -205,7 +215,7 @@ async def process_proxy(self, raw_protocol: int, auth_data: str, domain: str, po
 
         try:
             start_checking_time = time.time()
-            check_result = await proxy_utils.check_proxy(proxy_url)
+            check_result = await proxy_utils.check_proxy(proxy_url, aiohttp_proxy_check_session)
             end_checking_time = time.time()
 
             if check_result:

proxy_py/_settings.py

@@ -12,8 +12,8 @@
 
 # fetcher settings
 
-CONCURRENT_TASKS_COUNT = 128 
-PROXY_QUEUE_SIZE = 512
+CONCURRENT_TASKS_COUNT = 4096
+PROXY_QUEUE_SIZE = 4096
 
 MIN_PROXY_CHECKING_PERIOD = 5 * 60
 MAX_PROXY_CHECKING_PERIOD = 45 * 60

proxy_utils.py

@@ -11,13 +11,18 @@
 
 
 # TODO: add multiple checks with several sites
-async def check_proxy(proxy_url: str):
+async def check_proxy(proxy_url: str, session):
     try:
-        res = await async_requests.get(
+        res = await _request(
+            'get',
             'https://pikagraphs.d3d.info/OK/',
-            proxy=proxy_url,
-            timeout=settings.PROXY_CHECKING_TIMEOUT,
+            proxy_url,
+            settings.PROXY_CHECKING_TIMEOUT,
+            session
         )
+        res2 = await _request('GET', 'https://wtfismyip.com/text', proxy_url, settings.PROXY_CHECKING_TIMEOUT, session)
+        res = async_requests.Response(200, "OK")
+        print(res2)
         if res.status == 200 and res.text == "OK":
             return True
     except (aiohttp.client_exceptions.ServerDisconnectedError,
@@ -29,18 +34,19 @@ async def check_proxy(proxy_url: str):
             asyncio.TimeoutError,
             aiohttp.client_exceptions.ClientOSError,
             ) as ex:
+        message = str(ex)
+        if "file" in message:
+            print(message)
         # TODO: check to "Too many open files"
         return False
-#
-# async def get_working_proxies(domain, port, auth_data=""):
-#     proxies = []
-#
-#     for protocol in Proxy.PROTOCOLS:
-#         proxy = Proxy(protocol=protocol, domain=domain, port=port, auth_data=auth_data)
-#
-#         start_checking_time = time.time()
-#         if await check_proxy(proxy):
-#             end_checking_time = time.time()
-#             proxies.append((proxy, start_checking_time, end_checking_time))
-#
-#     return proxies
+
+
+async def _request(method, url, proxy_url, timeout, session : aiohttp.ClientSession):
+    headers = {
+        'User-Agent': async_requests.get_random_user_agent()
+    }
+
+    async with session.request(method, url, timeout=timeout, headers=headers, proxy=proxy_url) as response:
+        status = response.status
+        text = await response.text()
+        return async_requests.Response(status, text)