fixed leak, some refactoring, more debug logs

Author: DevAlone

main.py

@@ -11,6 +11,21 @@
 
 
 if __name__ == "__main__":
+    asyncio_logger = logging.getLogger("asyncio")
+    asyncio_logger_file_handler = logging.FileHandler("logs/asyncio.log")
+    asyncio_logger_file_handler.setLevel(logging.DEBUG)
+    asyncio_logger_file_handler.setFormatter(
+        logging.Formatter(
+            "%(levelname)s ~ %(asctime)s ~ %(funcName)30s() - %(message)s"
+        )
+    )
+    asyncio_logger.addHandler(asyncio_logger_file_handler)
+
+    if settings.DEBUG:
+        asyncio.get_event_loop().set_debug(True)
+
+        asyncio_logger.setLevel(logging.DEBUG)
+
     main_logger = logging.getLogger("proxy_py/main")
 
     if settings.DEBUG:
@@ -29,7 +44,6 @@
     main_logger.addHandler(logger_file_handler)
 
     loop = asyncio.get_event_loop()
-    # TODO: consider loop.set_debug
 
     proxy_processor = Processor.get_instance()
 

processor.py

@@ -10,6 +10,7 @@
 import logging
 import peewee
 
+
 # TODO: add ipv6 addresses, make domain checking better
 _0_TO_255_REGEX = r"([0-9]|[1-8][0-9]|9[0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])"
 DOMAIN_LETTER_REGEX = r"[a-zA-Z0-9_\-]"
@@ -77,14 +78,14 @@ async def worker(self):
 
     async def process_proxies(self):
         while True:
-            await asyncio.sleep(0.00001)
+            await asyncio.sleep(0.01)
             try:
                 # check good proxies
                 proxies = await db.execute(
                     Proxy.select().where(
                         Proxy.number_of_bad_checks == 0,
                         Proxy.last_check_time < time.time() - Proxy.checking_period,
-                        ).order_by(Proxy.last_check_time).limit(settings.NUMBER_OF_CONCURRENT_TASKS)
+                    ).order_by(Proxy.last_check_time).limit(settings.NUMBER_OF_CONCURRENT_TASKS)
                 )
                 if proxies:
                     self.good_proxies_are_processed = False
@@ -131,9 +132,8 @@ async def process_proxies(self):
 
     async def process_collectors(self):
         while True:
+            await asyncio.sleep(0.1)
             try:
-                await asyncio.sleep(0.000001)
-
                 # check collectors
                 collector_states = await db.execute(
                     CollectorState.select().where(
@@ -155,16 +155,14 @@ async def process_collectors(self):
                 await asyncio.sleep(settings.SLEEP_AFTER_ERROR_PERIOD)
 
     async def add_proxy_to_queue(self, proxy: Proxy, collector_id=None):
-        while self.proxies_semaphore.locked():
-            await asyncio.sleep(0.001)
-
-        asyncio.ensure_future(self.process_proxy(
-            proxy.get_raw_protocol(),
-            proxy.auth_data,
-            proxy.domain,
-            proxy.port,
-            collector_id,
-        ))
+        async with self.proxies_semaphore:
+            asyncio.ensure_future(self.process_proxy(
+                proxy.get_raw_protocol(),
+                proxy.auth_data,
+                proxy.domain,
+                proxy.port,
+                collector_id,
+            ))
 
     async def add_proxies_to_queue(self, proxies: list):
         for proxy in proxies:
@@ -258,15 +256,16 @@ async def process_raw_proxy(self, proxy, collector_id):
 
             for raw_protocol in range(len(Proxy.PROTOCOLS)):
                 while not self.good_proxies_are_processed:
-                    await asyncio.sleep(0.01)
+                    # TODO: find a better way
+                    await asyncio.sleep(0.1)
 
                 new_proxy = Proxy()
                 new_proxy.raw_protocol = raw_protocol
                 new_proxy.auth_data = auth_data
                 new_proxy.domain = domain
                 new_proxy.port = port
 
-                self.add_proxy_to_queue(new_proxy, collector_id)
+                await self.add_proxy_to_queue(new_proxy, collector_id)
 
     async def process_proxy(self, raw_protocol: int, auth_data: str, domain: str, port: int, collector_id):
         async with self.proxies_semaphore:

proxy_py/_settings.py

@@ -32,7 +32,7 @@
     # 'local/collectors',  # use to add your own collectors
 ]
 
-NUMBER_OF_CONCURRENT_TASKS = 128
+NUMBER_OF_CONCURRENT_TASKS = 64
 # makes aiohttp to not send more
 # than this number of simultaneous requests
 # works by common connector

proxy_utils.py

@@ -1,8 +1,9 @@
 import copy
+import random
+import asyncio
 
 from checkers.base_checker import CheckerResult
 from proxy_py import settings
-import random
 
 
 async def check_proxy(proxy_url: str, timeout=None) -> tuple:
@@ -14,6 +15,7 @@ async def check_proxy(proxy_url: str, timeout=None) -> tuple:
     results = []
 
     for checker, _ in zip(checkers, range(settings.MINIMUM_NUMBER_OF_CHECKERS_PER_PROXY)):
+        checker()
         result = await checker().check(proxy_url, timeout=timeout)
         if not result[0]:
             return False, None