coffea.compute backend interface design

[nsmith-]

When I originally heard the term coffea.compute, I thought it's going to be a function (and the namespace would have a different name).
I was imagining something like coffea.compute(some computable, executor=something) sort of like what dask.compute does where you tell it what to compute and some executor/scheduler.
Is this out of the question? Is there a strong reason why you preferred coffea.compute be a namespace and have backend.compute do the execution?

I thought about that originally, but context managers are very nice things to have. I see bugs in clearing the progress bar, for example, that are probably as a result of the lack of these currently that would be nice to get past once and for all.

Originally posted by @nsmith- in #1470 (comment)

A possible compromise to have both is as follows:

No context manager approach:

backend = Backend()
result = backend.compute(blah)  # blocking

Context manager approach:

with Backend() as backend:
    task = backend.compute(blah)  # non-blocking
    out = task.result() # blocking

Internally, the context manager just does with self: to implement.

[nsmith-]

@ikrommyd FYI

[nsmith-]

Another design aspect is how the retries would be handled. With the current interface, a pretty typical user code might look like:

rest = None

with Backend(retries=5, send_metrics=True) as backend:
    proc = Processor()

    task = backend.compute(prepared.map_events(proc))
    try:
        out = task.result()
    except Exception as e:
        print(f"Computation failed: {e}")
        out, rest = task.partial_result()
        print("Saving partial result to output.coffea")
        coffea.util.save(out, "output.coffea")

if rest:
    with Backend() as better_backend:
        print("Resuming computation with a better backend")
        task = better_backend.compute(rest)
        out = out + task.result()
        coffea.util.save(out, "output.coffea")

which is a lot of boilerplate. How might we simplify this?

[ikrommyd]

I'd honestly be in favor of having both approaches. I'm personally not a big fan of such context managers. I get the benefits though. For example I have never used a dask client as a context manager in my own code.

[NJManganelli]

For power users, one can always directly call ctx.__enter__() and exit at the end, so do we need to actually make a public interface? I think if the benefits of context managers are clear enough, we should make that the official way of doing things... not a strong feeling, but we've discussed how we have too many methods / patterns for doing things (e.g. installation recommendations for coffea). I guess one question on that is, how does this work with a dask cluster? Is the context manager just connecting to a scheduler (already established, running continuously for a while), rather than creating one? Building/tearing down dask clusters via HTCondor is probably not something you wanna do inside it every time you wanna compute a few tasks.

As far as boilerplate, can we just make coffea.compute be essentially the above? Have a primary backend, give a default fallback backend (iterative?), and perhaps include the save command?

This then serves also as the "intended pattern" we want users to mimic and expand upon. First time users will just use the basic function. Then they eventually need to compute faster, so they swap to taskvine or dask backend. Then they go, "I want more control" or "I need to compute multiple things and maybe abandon things entirely if there's X kinds of failures," and start copying and modifying the minimal info in this function.

def compute(computable, saveloc, primary=ThreadedBackend, fallback=IterativeBackend, primary_retries=5, fallback_retries=2):
    with primary(retries=primary_retries, send_metrics=True) as backend:
        task = backend.compute(computable)
        try:
            out = task.result()
        except Exception as e:
            print(f"Computation failed: {e}")
            out, rest = task.partial_result()
            print(f"Saving partial result to {saveloc}.coffea")
            coffea.util.save(out, "{saveloc}.coffea")
    
    if rest:
        with fallback(retries=fallback_retries, send_metrics=True) as better_backend:
            print("Resuming computation with a better backend")
            task = better_backend.compute(rest)
            out = out + task.result()
            coffea.util.save(out, "{saveloc}.coffea")