[WIP] Use anyio for structured concurrency

[Graeme22]

Description

Begins adding structured concurrency via anyio, a new dependency which replaces async_timeout.

Related issue(s)

Fixes #292

API Changes

• Most objects (Redis clients, connection pools, pubsub, pipelines) are now streamlined for an async-first design. This means they require async context managers and don't provide other ways to use them.
• Pipelines auto-execute when the async context manager exits. This makes the syntax very clean and makes type safety straightforward. If you need to abort a pipeline for some reason you can still call clear.
• Connections pools now use two separate pools. One pool is for multiplexing normal commands, the other pool is for dedicated connections. Pools are also blocking by default (in fact, non-blocking pools no longer exist).

Pre-merge checklist

• Add cluster support
• Add sentinel support
• Passing tests
• Passing lints
• Docs update

[Graeme22]

Leaving this PR here for reference as well: redis/redis-py#3647.

This looks like a really nice implementation. It would allow us to potentially not break the API at all, since this implementation actually doesn't require the async context manager afaict.

[Graeme22]

Small update on this! This code is working for me with this PR:

from trio import run
from coredis import Redis

redis = Redis.from_url("redis://localhost:6379", decode_responses=True)

async def main():
    async with redis:
        print(await redis.ping())
        async with redis.pubsub(channels=["mychannel"]) as ps:
            await redis.publish("mychannel", "test message!")
            async for msg in ps:
                print(msg)
                if msg["type"] == "message":
                    break
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr("tmpkey")
            val = pipe.get("tmpkey")
            pipe.delete(["tmpkey"])
        print(await val)

run(main)

"""
PONG
{'type': 'subscribe', 'pattern': None, 'channel': 'mychannel', 'data': 1}
{'type': 'message', 'pattern': None, 'channel': 'mychannel', 'data': 'test message!'}
1
"""

So base functionality is working, and notice I'm using Trio here!

[Graeme22]

I've also updated the connection pool logic. It works like this: When a client requests a connection from the pool, it gets an existing connection that has less than MAX_REQUESTS_PER_CONNECTION pending requests (currently 32, will have to play around to find optimal value). Otherwise, we create a new connection if we're below the max number of connections. If all connections are maxed out, we either throw an error or wait depending on whether it's a blocking pool or not. Also, requests like XREAD that can be blocking set the whole connection to blocked regardless of whether it's full. So new connections are only created when there are a lot of pending requests. This should provide better performance but will have to be tested of course.

[hyperlint-ai]

PR Change Summary

Introduced structured concurrency using anyio, enhancing the async capabilities of the library.

• Added structured concurrency via anyio, replacing async_timeout.
• Streamlined Redis clients and connection pools for async-first design.
• Pipelines now auto-execute upon exiting the async context manager.
• Default connection settings allow for 32 concurrent requests.

Modified Files

• HISTORY.rst
• docs/source/handbook/development.rst

---

How can I customize these reviews?

Check out the Hyperlint AI Reviewer docs for more information on how to customize the review.

If you just want to ignore it on this PR, you can add the hyperlint-ignore label to the PR. Future changes won't trigger a Hyperlint review.

Note specifically for link checks, we only check the first 30 links in a file and we cache the results for several hours (for instance, if you just added a page, you might experience this). Our recommendation is to add hyperlint-ignore to the PR to ignore the link check for this PR.

[alisaifee]

@Graeme22 perhaps this will fix the merge issues: 8f63204