Rewrite the asyncio polling of nodes to use a configurable count of worker tasks to process nodes, rather than creating a task for each node and relying on aiohttp to rate limit.
The nodes to be pulled should be added to a Queue that the workers will pull from, with results pushed onto a deque (maybe two deques, one for successful nodes and one for errros).
The current design is causing issues with troubleshooting collector issues (particularly when the process just hangs, consuming all the CPU, e.g. #99), because the tasks were all created at the beginning so it isn't clear which node is causing the issue (assuming that is where the issue lies). The gather()-ing of all the results doesn't conform to what appears to be best practice, based on the Python examples and new features that are coming out (e.g. TaskGroup). I kind of realized that to begin with but didn't understand at that time the right way to do it, but I think I do now.
Rewrite the
asynciopolling of nodes to use a configurable count of worker tasks to process nodes, rather than creating a task for each node and relying onaiohttpto rate limit.The nodes to be pulled should be added to a
Queuethat the workers will pull from, with results pushed onto adeque(maybe twodeques, one for successful nodes and one for errros).The current design is causing issues with troubleshooting collector issues (particularly when the process just hangs, consuming all the CPU, e.g. #99), because the tasks were all created at the beginning so it isn't clear which node is causing the issue (assuming that is where the issue lies). The
gather()-ing of all the results doesn't conform to what appears to be best practice, based on the Python examples and new features that are coming out (e.g.TaskGroup). I kind of realized that to begin with but didn't understand at that time the right way to do it, but I think I do now.