Skip to content

[BugFix] openbb-cboe: Fix cache lock so that concurrent callers in separate processes don't hang - #7796

Open
dlee-bq wants to merge 5 commits into
developfrom
bugfix/fix-cboe-cache-lock
Open

dlee-bq wants to merge 5 commits into
developfrom
bugfix/fix-cboe-cache-lock

Conversation

@dlee-bq

@dlee-bq dlee-bq commented Oct 14, 2026

Copy link
Copy Markdown
Contributor

This PR fixes a bug in the caching layer of openbb-cboe where hangs can happen when the endpoints are hit concurrently via ProcessPool. The error thrown is: RuntimeError: <asyncio.locks.Lock …> is bound to a different event loop

Below is a script to test on this branch vs. develop:

import asyncio
import sys
import tempfile
import threading

from aiohttp import web
from openbb_cboe.utils import helpers
from openbb_core.provider.utils.helpers import run_async

THREADS, ROUNDS, PER_LOOP = 6, 20, 3


def start_server() -> str:
    """Serve JSON on a local port, standing in for Cboe."""
    ready, box = threading.Event(), {}

    async def main():
        async def handler(request):
            return web.json_response({"path": request.path})

        app = web.Application()
        app.router.add_get("/{tail:.*}", handler)
        runner = web.AppRunner(app)
        await runner.setup()
        await web.TCPSite(runner, "127.0.0.1", 0).start()
        box["url"] = f"http://127.0.0.1:{runner.addresses[0][1]}"
        ready.set()
        await asyncio.Event().wait()

    threading.Thread(target=lambda: asyncio.run(main()), daemon=True).start()
    ready.wait()

    return box["url"]


def main() -> int:
    url = start_server()
    cache_file = sys.argv[1] if len(sys.argv) > 1 else f"{tempfile.mkdtemp()}/cboe"
    helpers.cache_path = lambda: cache_file

    for round_number in range(ROUNDS):
        barrier, results = threading.Barrier(THREADS), {}

        def caller(index):
            async def burst():
                barrier.wait(10)
                urls = (f"{url}/{round_number}/{n}.json" for n in range(PER_LOOP))
                return await asyncio.wait_for(
                    asyncio.gather(*(helpers.get_cboe_data(u) for u in urls)), 10
                )

            try:
                results[index] = run_async(burst)
            except BaseException as error:  # noqa: BLE001
                results[index] = error

        threads = [threading.Thread(target=caller, args=(i,)) for i in range(THREADS)]
        [t.start() for t in threads]
        [t.join(30) for t in threads]
        errors = [f"{type(r).__name__}: {r}" for r in results.values() if isinstance(r, BaseException)]

        if errors or len(results) < THREADS:
            print(f"FAIL in round {round_number}: {len(errors)}/{THREADS} callers failed")
            print("  e.g.", errors[0] if errors else "a caller never returned")
            return 1

    print(f"PASS: {THREADS} threads x {ROUNDS} rounds x {PER_LOOP} requests, no failures")
    return 0


if __name__ == "__main__":
    sys.exit(main())

@dlee-bq dlee-bq added bug Bugs and bug fixes extensions Extension-related labels Oct 14, 2026

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

bug Bugs and bug fixes extensions Extension-related

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant