Alternative proposal for concurrent.futures support in effectful

[kiranandcode]

In draft mode as it's very WIP, but following discussions with @jfeser, this branch proposes an alternative implementation for async support for effectful through concurrent.futures.

• main design complexities (so far):
• for thread safety, we need to make the interpretation thread local, but new threads do not inherit the value of thread local variables, they see fresh instances, so we need to save and restore the runtime context at the time of submission inside the submitted task.

from concurrent.futures import Future, ThreadPoolExecutor

from effectful.handlers.futures import ThreadPoolFuturesInterpretation, Executor
from effectful.ops.semantics import NotHandled, defop, handler


@defop
def add(n: int, k: int) -> int:
    raise NotHandled


@defop
def a_mul(n: int, k: int) -> Future[int]:
    raise NotHandled


def a_mul_interp(n, k):
    def mul(a: int, b: int):
        a, b = (-a, -b) if a < 0 else (a, b)
        res: int = 0
        while a > 0:
            a, res = a - 1, add(res, b)
        return res

    return Executor.submit(mul, n, k)


with handler(ThreadPoolFuturesInterpretation()), handler({a_mul: a_mul_interp}):
    fut: Future[int] = a_mul(2, 10)
    # with handler({add: lambda a,b: a + b}): # wouldn't change result
    res = fut.result(None)
    assert res == add(add(0, 10), 10)
    with handler({add: lambda a, b: a + b}):
        fut: Future[int] = a_mul(2, 10)
        assert fut.result() == 20

[jfeser]

I think this looks like it's moving in good directions. The simple API is a good sign.

I don't like the look of release_handler_lock and friends. I'm surprised that nothing in the LLM provider code seems to call it, which suggests that it must be called by users of that code? I definitely don't think we want this to be part of the public API. My suggestion is to leave this extra API out until we have a clear need for it.

[jfeser]

Isn't this defined in the concurrent.futures library? Do we need our own dataclass definition?

[kiranandcode]

it's a namedtuple in the library, I added this dataclass because @defop ... using the namedtuple type was not allowing .done iirc (or there could have been another issue that was causing this to fail and I misdiagnosed the cause)

[eb8680]

There shouldn't be a behavioral difference in effectful between the dataclass and the namedtuple for this simple case, so it's probably worth creating a self-contained test that passes with the dataclass and fails with the namedtuple so that we can try to fix it.

[kiranandcode]

Oh yep will do!

[kiranandcode]

I don't like the look of release_handler_lock and friends. I'm surprised that nothing in the LLM provider code seems to call it, which suggests that it must be called by users of that code? I definitely don't think we want this to be part of the public API. My suggestion is to leave this extra API out until we have a clear need for it.

Oh, yep, that was enabled by the realisation that Executor.submit's function is evaluated in a new thread, so the code doesn't hold the lock at all. So likely release_handler_lock can probably be removed.

[kiranandcode]

Proposed interface is now:

if __name__ == "__main__":
    from openai import OpenAI
    from effectful.ops.syntax import implements
    from effectful.ops.semantics import fwd

    openai_provider = OpenAIAPIProvider(client=OpenAI())
    user_request: str = "write a fibonnaci function"
    def safe_synthesis_get[T](future: Future[T]) -> T | None:
        try:
            return future.result()
        except SynthesisError:
            return None

    with handler(ThreadPoolFuturesInterpretation()),  handler(openai_provider), handler(ProgramSynthesis()):
        # Launch multiple LLM calls concurrently
        tasks = [Executor.submit(generate_program, user_request + f" {i}") for i in range(3)]
        results_as_completed = (safe_synthesis_get(f) for f in futures.as_completed(tasks))
        valid_results = list(result for result in results_as_completed if result)
        best_program = next(result for result in valid_results)

    print(f"programs were {valid_results}")
    print(f"best_program was {getattr(best_program, '__src__', None)}")