some refactoring

Author: DevAlone

TODO

@@ -13,3 +13,4 @@ p2p
 log memory consumption
 stop using ipinfo.io because of rate limits
 fix leak of something, again :(
+remove processor proxies queue model

processor.py

@@ -66,40 +66,15 @@ def __init__(self):
 
         self.logger.debug("processor initialization...")
 
-        self.queue = asyncio.Queue(maxsize=settings.PROXY_QUEUE_SIZE)
         self.proxies_semaphore = asyncio.BoundedSemaphore(settings.NUMBER_OF_CONCURRENT_TASKS)
         self.good_proxies_are_processed = False
 
     async def worker(self):
         await asyncio.gather(*[
-            self.producer(),
-            self.consumer(),
+            self.process_proxies(),
+            self.process_collectors(),
         ])
 
-    async def consumer(self):
-        while True:
-            await asyncio.sleep(0.00001)
-
-            try:
-                if not self.proxies_semaphore.locked():
-                    asyncio.ensure_future(self.process_proxy(
-                        *(await self.queue.get())
-                    ))
-            except KeyboardInterrupt:
-                raise
-            except BaseException as ex:
-                self.logger.exception(ex)
-                if settings.DEBUG:
-                    raise ex
-                await asyncio.sleep(settings.SLEEP_AFTER_ERROR_PERIOD)
-
-    async def producer(self):
-        while True:
-            await asyncio.gather(*[
-                self.process_proxies(),
-                self.process_collectors(),
-            ])
-
     async def process_proxies(self):
         while True:
             await asyncio.sleep(0.00001)
@@ -127,7 +102,7 @@ async def process_proxies(self):
                         Proxy.number_of_bad_checks > 0,
                         Proxy.number_of_bad_checks < settings.DEAD_PROXY_THRESHOLD,
                         Proxy.last_check_time < time.time() - settings.BAD_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)
                 )
 
                 await self.add_proxies_to_queue(proxies)
@@ -141,7 +116,7 @@ async def process_proxies(self):
                         Proxy.number_of_bad_checks >= settings.DEAD_PROXY_THRESHOLD,
                         Proxy.number_of_bad_checks < settings.DO_NOT_CHECK_ON_N_BAD_CHECKS,
                         Proxy.last_check_time < time.time() - settings.DEAD_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)
                 )
 
                 await self.add_proxies_to_queue(proxies)
@@ -179,16 +154,16 @@ async def process_collectors(self):
 
                 await asyncio.sleep(settings.SLEEP_AFTER_ERROR_PERIOD)
 
-    def is_queue_free(self):
-        return self.queue.qsize() < settings.NUMBER_OF_CONCURRENT_TASKS
+    async def add_proxy_to_queue(self, proxy: Proxy, collector_id=None):
+        while self.proxies_semaphore.locked():
+            await asyncio.sleep(0.001)
 
-    async def add_proxy_to_queue(self, proxy: Proxy):
-        await self.queue.put((
+        asyncio.ensure_future(self.process_proxy(
             proxy.get_raw_protocol(),
             proxy.auth_data,
             proxy.domain,
             proxy.port,
-            None
+            collector_id,
         ))
 
     async def add_proxies_to_queue(self, proxies: list):
@@ -285,13 +260,13 @@ async def process_raw_proxy(self, proxy, collector_id):
                 while not self.good_proxies_are_processed:
                     await asyncio.sleep(0.01)
 
-                await self.queue.put((
-                    raw_protocol,
-                    auth_data,
-                    domain,
-                    port,
-                    collector_id,
-                ))
+                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)
 
     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,14 +32,13 @@
     # 'local/collectors',  # use to add your own collectors
 ]
 
-NUMBER_OF_CONCURRENT_TASKS = 512
+NUMBER_OF_CONCURRENT_TASKS = 128
 # makes aiohttp to not send more
 # than this number of simultaneous requests
 # works by common connector
-NUMBER_OF_SIMULTANEOUS_REQUESTS = 256
+NUMBER_OF_SIMULTANEOUS_REQUESTS = 64
 # the same, but per host
 NUMBER_OF_SIMULTANEOUS_REQUESTS_PER_HOST = NUMBER_OF_SIMULTANEOUS_REQUESTS
-PROXY_QUEUE_SIZE = NUMBER_OF_CONCURRENT_TASKS * 2
 
 MIN_PROXY_CHECKING_PERIOD = 10 * 60
 MAX_PROXY_CHECKING_PERIOD = 60 * 60

statistics/statistics.py

@@ -11,7 +11,6 @@ async def worker():
         await process_graph(ProxyCountItem, 60, create_proxy_count_item)
         await process_graph(NumberOfProxiesToProcess, 60, number_of_proxies_to_process)
         await process_graph(NumberOfCollectorsToProcess, 60, number_of_collectors_to_process)
-        await process_graph(ProcessorProxiesQueueSize, 60, processor_proxies_queue_size)
         await asyncio.sleep(10)
 
 
@@ -98,11 +97,3 @@ async def number_of_collectors_to_process(timestamp):
         timestamp=timestamp,
         value=number_of_collectors,
     )
-
-
-async def processor_proxies_queue_size(timestamp):
-    await db.create(
-        ProcessorProxiesQueueSize,
-        timestamp=timestamp,
-        value=Processor.get_instance().queue.qsize(),
-    )