ProactorEventLoop - ValueError: loop argument must agree with Future

Question

I am using this asynchronous project (called Broker - see git) with the following code:

   proxies = asyncio.Queue()
   broker = Broker(proxies)
   tasks = asyncio.gather(broker.find(types = ['HTTPS'], strict = True, limit = 10),
                               self.storeProxies(proxies))
   loop = asyncio.get_event_loop()
   loop.run_until_complete(tasks)

where self.storeProxies is an async function that contains

while True:
    proxy = await proxies.get()
    if proxy is None: break
    self.proxies['http://{0}:{1}'.format(proxy.host, proxy.port)] = 1

However, I do not think that it is related to the Broker part (not sure of course).

Everytime I run this code, after a quite random number of successes, it fails with [WinError 10038] An operation was attempted on something that is not a socket. I then tried to do some research and ended up at this answer. But when trying with this code, I get this error:

ValueError: loop argument must agree with Future

Any ideas on how to deal with this?

Details that might be useful:

OS: Windows 10 (version 1809) Python version: 3.7.1

Based on the answer from Mikhail Gerasimov

async def storeProxies(self, proxies):
    logger.info("IN IT!")
    while True:
        proxy = await proxies.get()
        if proxy is None: break
        self.proxies['http://{0}:{1}'.format(proxy.host, proxy.port)] = 1

async def getfetching(self, amount):
    from proxybroker import Broker

    proxies = asyncio.Queue()
    broker = Broker(proxies)
    return await asyncio.gather(
        broker.find(
            types = ['HTTP', 'HTTPS'],
            strict = True,
            limit = 10
        ),
        self.storeProxies(proxies)
    )
def fetchProxies(self, amount):
    if os.name == 'nt':
        loop = asyncio.SelectorEventLoop() # for subprocess' pipes on Windows
        asyncio.set_event_loop(loop)
    else:
        loop = asyncio.get_event_loop()
    loop.run_until_complete(self.getfetching(amount))
    logger.info("FETCHING COMPLETE!!!!!!")
    logger.info('Proxies: {}'.format(self.proxies))

where fetchProxies is called at certain intervals from another place. This works perfectly the first time, but then "fails" with the warnings:

2019-04-06 21:04:21 [py.warnings] WARNING: d:\users\me\anaconda3\lib\site-packages\proxybroker\providers.py:78: DeprecationWarning: The object should be created from async function
  headers=get_headers(), cookies=self._cookies, loop=self._loop

2019-04-06 21:04:21 [py.warnings] WARNING: d:\users\me\anaconda3\lib\site-packages\aiohttp\connector.py:730: DeprecationWarning: The object should be created from async function
  loop=loop)

2019-04-06 21:04:21 [py.warnings] WARNING: d:\users\me\anaconda3\lib\site-packages\aiohttp\cookiejar.py:55: DeprecationWarning: The object should be created from async function
  super().__init__(loop=loop)

followed by a behavior that seems like an infinite loop (hard stuck outputting nothing). Noteworthy, this also happened to me in the example from Gerasimov (in the comments) when importing Broker globally. Finally, I am starting to see the light of the tunnel.

Answer 1

Take a look at this part:

proxies = asyncio.Queue()

# ...

loop.run_until_complete(tasks)

asyncio.Queue() when created is bound to current event loop (default event loop or one been set to current with asyncio.set_event_loop()). Usually only loop object is bound to can manage it. If you change current loop, you should recreate object. Same is true for many other asyncio objects.

To make sure each object is bound to new event loop, it's better to create asyncio-related objects already after you set and ran new event loop. It'll look like this:

async def main():
    proxies = asyncio.Queue()
    broker = Broker(proxies)
    return await asyncio.gather(
        broker.find(
            types = ['HTTPS'], 
            strict = True, 
            limit = amount
        ),
        self.storeProxies(proxies)
    )


if os.name == 'nt':
    loop = asyncio.ProactorEventLoop() # for subprocess' pipes on Windows
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Sometimes (rarely) you'll have to place some imports inside main() either.

---

Upd:

Problem happens because you change event loop between fetchProxies calls, but Broker imported only once (Python caches imported modules).

Reloading Broker didn't work for me thus I couldn't find a better way than to reuse event loop you set once.

Replace this code

if os.name == 'nt':
    loop = asyncio.SelectorEventLoop() # for subprocess' pipes on Windows
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()

with this

if os.name == 'nt':
    loop = asyncio.get_event_loop()
    if not isinstance(loop, asyncio.SelectorEventLoop):
        loop = asyncio.SelectorEventLoop() # for subprocess' pipes on Windows
        asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()

P.S.

BTW you don't have to do this in the first place:

if os.name == 'nt':
    loop = asyncio.SelectorEventLoop()
    asyncio.set_event_loop(loop)

Windows already use SelectorEventLoop as default and it doesn't support pipes (doc).