Tighten up IterativeFinder async close behavior (DHT iterator continues after consumer breaks out of it)

[moodyjon]

See: #3504

Script:
scratch.py.txt

Script output:
scratch_1.txt

Initially there is a lot of background activity from IterativeNodeFinder during what I presume is startup.
In steady state there is little activity after each "PAUSE---". Usually just stuff related to cancellation, if anything.

Eventually, I hope the "async with aclosing(...)" feature can be used:
https://docs.python.org/3.10/library/contextlib.html?highlight=aclosing#contextlib.aclosing

[shyba]

Hello, thank you for looking into it!

Since aclosing is introduced in python 3.10, I think it would be better to add the snippet that they show as equivalent to utils.py:

from contextlib import asynccontextmanager

@asynccontextmanager
async def aclosing(thing):
    try:
        yield thing
    finally:
        await thing.aclose()

Then, we can use it from utils until we support 3.10, changing the import to the built-in one.

I modified the snippet to test that:

import asyncio
import inspect
from contextlib import asynccontextmanager

@asynccontextmanager
async def aclosing(thing):
    try:
        yield thing
    finally:
        close = thing.aclose()
        if inspect.isawaitable(close):
            await close

from lbry.dht.constants import generate_id
from lbry.dht.node import Node
from lbry.dht.peer import PeerManager
from lbry.extras.daemon.storage import SQLiteStorage
from lbry.conf import Config
import logging
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s %(levelname)-4s %(name)s:%(lineno)d: %(message)s")
log = logging.getLogger(__name__)


async def do_it():
    loop = asyncio.get_event_loop()
    #loop.set_debug(True)
    conf = Config()
    print(str(conf))
    storage = SQLiteStorage(conf, ":memory:", loop, loop.time)
    await storage.open()
    node = Node(
        loop, PeerManager(loop), generate_id(), 4444, 4444, 4444, None,
        storage=storage
    )
    node.start(interface='0.0.0.0', known_node_urls=conf.known_dht_nodes)
    await asyncio.sleep(10)
    key = bytes.fromhex('2ecce30223bc2808bd10cf8e18f1cccecedf9fd35fcc370b9d15ad7473a59d3dc16e85a8f08a1891e668608d1660b8c7')
    while True:
        async with aclosing(node.get_iterative_value_finder(key=key, max_results=1)) as values:
            i = 0
            print('SEARCH---')
            async for _ in values:
                print(f'{i}: {_}')
                i += 1
                break
        print('PAUSE---')
        await asyncio.sleep(10)
    await asyncio.wait(asyncio.all_tasks())

if __name__ == '__main__':
    asyncio.run(do_it())

Doing it this way, the pause and search messages clean, like

PAUSE---
SEARCH---

This looks cleaner than implementing the finally->aclose and other changes on every call and we would be able to use the built in version when it is out.

[moodyjon]

Implemented an aclosing() in lbry/utils.py as suggested.

[shyba]

Thank you for the quick reply!
Aside from trapping the CancelledError, I think this looks great but we need to separate the change in behavior from the iterator bugfix to ease testing and changelog.

[moodyjon]

I think I have addressed the 3 comments. The IterativeFinder is mostly back to the original logic with no generator function stuff or output limiting based on max_results. Let me know if there are more comments or a request to squash changes.

[moodyjon]

I've seen this failure in my runs as well. Seems like the upper bound needs a bump analogous to:

fd73412

File "/home/runner/work/lbry-sdk/lbry-sdk/tests/integration/other/test_exchange_rate_manager.py", line 30, in test_exchange_rate_manager
123
self.assertLessEqual(lbc, 20_000)
124
AssertionError: Decimal('20004.00080016') not less than or equal to 20000
125

[moodyjon]

Argggggh :-/

test_exchange_rate_manager.py still failing, but due to server being down:

Hotbit is under maintenance...
...
Dear Respected Hotbit Users, Hotbit will be under maintenance from 1:30 AM to 1:30 PM UTC on 17th May to optimize the servers' performance. During the maintenance, the service will be unavailable. We apologize for any inconvenience caused!

[eukreign]

@moodyjon @shyba what's the status of this PR? (it needs a rebase to get some test run fixes from master)

[shyba]

@moodyjon Can you please rebase on master instead of merging? Using rebase you can also remove the merge commits, making it a cleaner history

[coveralls]

Coverage increased (+0.01%) to 57.379% when pulling e5e9873 on moodyjon:async-for-pr3504 into 5c708e1 on lbryio:master.

[moodyjon]

@moodyjon @shyba what's the status of this PR? (it needs a rebase to get some test run fixes from master)

I had been observing/merging the test changes. Added one more test change yesterday: f91bf8f

@moodyjon Can you please rebase on master instead of merging? Using rebase you can also remove the merge commits, making it a cleaner history

OK, I think I have rebased correctly.

[moodyjon]

Removed exchange rate test bump of upper limit. It was changed in 5c708e1

[shyba]

@moodyjon thank you very much!