stanBienaives commented 3 months ago

Description:

I am encountering an issue with vcrpy when using it in a multithreaded context. The cassette file does not record all the requests when using ThreadPoolExecutor with more than one worker. The issue is not present when using single-threaded execution.

Test Code:

import vcr  # type: ignore
from concurrent.futures import ThreadPoolExecutor
import requests

vcr = vcr.VCR(record_mode='once', filter_headers=['authorization'], ignore_hosts=['api.smith.langchain.com'])

def test_multithreading():
    with vcr.use_cassette('./test_multithread.yaml', match_on=['method', 'scheme', 'host', 'port', 'path', 'query', 'body']):
        def get_google(num):
            return requests.get(f"https://www.google.com?q={num}")
        with ThreadPoolExecutor(max_workers=2) as executor:
            responses = executor.map(get_google, range(2))

    with open('./test_multithread.yaml', 'r') as f:
        data = f.read()
        assert data.count('https://www.google.com') == 2

def test_singlethreading():
    with vcr.use_cassette('./test_singlethread.yaml', match_on=['method', 'scheme', 'host', 'port', 'path', 'query', 'body']):
        def get_google(num):
            return requests.get(f"https://www.google.com?q={num}")
        with ThreadPoolExecutor(max_workers=1) as executor:
            responses = executor.map(get_google, range(2))

    with open('./test_singlethread.yaml', 'r') as f:
        data = f.read()
        assert data.count('https://www.google.com') == 2

Observed Behavior:

test_singlethreading passes as expected, with the cassette recording both requests. test_multithreading fails because the cassette file does not record both requests, resulting in a mismatch.

Environment:

Python 3.12.3
vcrpy==6.0.1
requests==2.31.0

Steps to Reproduce:

Run the provided test code. Observe that test_multithreading fails while test_singlethreading passes.

Expected Behavior:

Both test_multithreading and test_singlethreading should pass, with the cassette files correctly recording all the requests made during the tests.

stanBienaives commented 3 months ago

interesting enough:

Adding a delay makes the test pass:

def test_multithreading_record_with_delay():
    with vcr.use_cassette('./test_multithread_with_delay.yaml', match_on=['method', 'scheme', 'host', 'port', 'path', 'query', 'body']):
        def get_google(num):
            time.sleep(num)
            return requests.get(f"https://www.google.com?q={num}")
        with ThreadPoolExecutor(max_workers=2) as executor:
            responses  = executor.map(get_google, range(2))

    with open('./test_multithread_with_delay.yaml', 'r') as f:
        data = f.read()
        assert data.count('https://www.google.com') == 2

simon-weber commented 2 months ago

I suspect this has to do with race conditions involving force_reset. For example, vcr's getresponse will unpatch, call the underlying getresponse, then repatch. I'm guessing that's happening is a timeline like:

thread A tries getresponse
thread A needs to pass through the request and unpatches
thread B tries getresponse. the patches aren't currently applied, so the request goes through unnoticed.
thread A repatches

Afterwards, you're left with a recorded request from thread A and a missing one from thread B.

kevin1024 commented 2 months ago

@simon-weber I think that’s exactly what’s happening.

I guess we could put a mutex around the unpatch but you would lose some of the benefit of multithreading to begin with.

simon-weber commented 2 months ago

I'm not sure there's an easy locking fix inside vcrpy. Locking in force_reset doesn't work since it's not called if patches aren't applied.

What about if patches were left in place but dynamically disabled with a threadlocal? Wrapt has a way to do this and seems like it could pretty easily replace mock.patch.object.