Coalescing duplicate in-flight requests: one shared result for concurrent callers

import time
import threading
from collections import defaultdict


class CoalescingExecutor:
    def __init__(self):
        self._locks = defaultdict(threading.Lock)
        self._in_flight = {}

    def execute(self, key, func):
        with self._locks[key]:
            if key in self._in_flight:
                return self._in_flight[key]

        future = threading.Event()
        result_holder = {}

        def run():
            try:
                result_holder["value"] = func()
            finally:
                future.set()
                with self._locks[key]:
                    self._in_flight.pop(key, None)

        with self._locks[key]:
            if key in self._in_flight:
                return self._in_flight[key]
            self._in_flight[key] = future
            threading.Thread(target=run, daemon=True).start()

        future.wait()
        return result_holder["value"]


if __name__ == "__main__":
    executor = CoalescingExecutor()
    calls = []

    def slow_operation():
        calls.append("start")
        time.sleep(0.1)
        calls.append("end")
        return 42

    results = []
    threads = [threading.Thread(target=lambda: results.append(executor.execute("key", slow_operation))) for _ in range(5)]

    for t in threads:
        t.start()
    for t in threads:
        t.join()

    print("results:", sorted(results))
    print("calls:", calls)