PYTHON-5215 Add an asyncio.Protocol implementation for KMS

[blink1073]

See benchmark gist.

Benchmark Results:

Before: 4.93s, 5.26s
After: 4.80s, 4.89s

Depends on mongodb-labs/drivers-evergreen-tools#679

[blink1073]

I'm debugging two failures:

test.asynchronous.test_connection_monitoring.AsyncTestCMAP.test_connection_monitoring_pool_clear_interrupting_pending_connections_clear_with_interruptInUseConnections___true_closes_pending_connections
test.asynchronous.test_connection_monitoring.AsyncTestCMAP.test_connection_monitoring_pool_create_min_size_error_error_during_minPoolSize_population_clears_pool

[blink1073]

I realized the benchmark test wasn't actually triggering the protocol -> I'm tweaking things locally

[blink1073]

This has a couple commits from #2467

[blink1073]

Okay this is ready for a look. We might consider switching to the base Protocol since the sizes of the responses are constrained and small, and we typically only have 2-3 reads per socket.

[blink1073]

I still need to make a PR for the fix to the kms mock server's 404 response.

[blink1073]

CSOT failure is unrelated: PYTHON-5492

[NoahStapp]

If sizehint = -1, which signals that the buffer size can be arbitrary, this check will always succeed, potentially returning an empty buffer, which is an error. We need to check that sizehint is a positive number as well.

[blink1073]

Updated

[NoahStapp]

To make sure I understand the intended flow here, is this example correct?

1. We call kms_request and enter the while kms_context.bytes_needed > 0: loop.
2. The first chunk of data, say 16 bytes worth is written into an existing buffer that still has space.
3. PyMongoKMSProtocol.read() is called and immediately returns those 10 bytes.
4. kms_context.bytes_needed updates to need 84 more bytes for a total of 100.
5. We call PyMongoKMSProtocol.read() again and wait on the _pending_listeners Future we create.
6. The second chunk of data, the remaining 84 bytes, requires a new buffer since the active buffer is full.
7. The Future is resolved with those bytes, which we return and feed into kms_context to complete the operation.

[blink1073]

• We call kms_request and enter the while kms_context.bytes_needed > 0: loop.
• The first chunk of data, say 16 bytes worth is written into an existing buffer that still has space.
• PyMongoKMSProtocol.read() is called and immediately returns those 10 bytes, pushing the start_index up by 10
• kms_context.bytes_needed updates to need 84 more bytes
• We call PyMongoKMSProtocol.read() again and wait on the _pending_listeners Future we created.
• The second chunk of data, the remaining 84 bytes, may require a new buffer
• If any bytes are available, we read in up to the newly requested 84 bytes from the active buffer(s), advancing start_index and exhausting buffers as appropriate
• Otherwise, we wait on the future to be resolved, which will contain up to the requested bytes.

[NoahStapp]

What happens if we need more bytes than we have? We've already popped the waiter and set it's result to data, which can only read up to self._bytes_ready bytes. Are we relying on the kms_context.bytes_needed loop to call the protocol read() method again and create a new waiter?

[blink1073]

Right, we give the partial result back to the kms context, and let it ask for more.

[NoahStapp]

Maybe we could get better performance by doing more of the looping inside the Protocol, but KMS requests won't be a significant part of runtime anyway so not worth spending more time on it. Can you add a comment to this effect somewhere saying that we rely on the looping behavior for this to function correctly?