fix

Author: DevAlone

main.py

@@ -54,7 +54,7 @@ async def proxy_counter():
 
     loop.run_until_complete(proxy_provider_server.start(loop))
 
-    loop.run_until_complete(asyncio.wait([
+    loop.run_until_complete(asyncio.gather(*[
         proxy_processor.exec(),
         proxy_counter(),
     ]))

processor.py

@@ -55,7 +55,7 @@ def __init__(self):
         self.queue = asyncio.Queue(maxsize=settings.PROXY_QUEUE_SIZE)
 
     async def exec(self):
-        await asyncio.wait([
+        await asyncio.gather(*[
             self.producer(),
             self.consumer(),
         ])
@@ -74,7 +74,7 @@ async def consumer(self):
                     i += 1
 
                 if tasks:
-                    await asyncio.wait(tasks)
+                    await asyncio.gather(*tasks)
                     tasks.clear()
             except KeyboardInterrupt:
                 raise
@@ -91,7 +91,7 @@ async def producer(self):
                 collector_states = await db.execute(
                     CollectorState.select().where(
                         CollectorState.last_processing_time < time.time() - CollectorState.processing_period
-                    )
+                    ).limit(settings.CONCURRENT_TASKS_COUNT)
                 )
 
                 tasks = [
@@ -100,7 +100,7 @@ async def producer(self):
                 ]
 
                 if tasks:
-                    await asyncio.wait(tasks)
+                    await asyncio.gather(*tasks)
                     tasks.clear()
 
                 # check proxies
@@ -193,11 +193,11 @@ async def process_raw_proxies(self, proxies, collector_id):
         for proxy in proxies:
             tasks.append(self.process_raw_proxy(proxy, collector_id))
             if len(tasks) > settings.CONCURRENT_TASKS_COUNT:
-                await asyncio.wait(tasks)
+                await asyncio.gather(*tasks)
                 tasks.clear()
 
         if tasks:
-            await asyncio.wait(tasks)
+            await asyncio.gather(*tasks)
 
     async def process_raw_proxy(self, proxy, collector_id):
         self.logger.debug("adding raw proxy \"{}\" to queue".format(proxy))