How to signal to a multithreaded function that it can stop?

user_2739654 0
Viewed 40

I have a list of paths to files, and I am running a function (called process_notebook()) against each path. The action that the function takes is not, I think, particularly important. What is important is that it has to defer if some dependencies (invariably the same function running against other files) has not been met, which it signals by returning False, or True if it runs without being deferred. Currently, I'm doing this:


last_count = len(deferred_notebooks)

while len(deferred_notebooks) > 0:

    for nb in deferred_notebooks:
        if process_notebook(nb, True):
            deferred_notebooks.remove(nb)

    # This should clear min one deferral each iteration. If it doesn't, then some dependency
    # is impossible to meet and we should give up.

    if len(deferred_notebooks) == last_count:
        break

That works fine, but of course processes the notebooks sequentially, and in many instances they'll not be ran for a couple of iterations (being down a chain of dependencies for example) and that means it's a bit slow. I'd like to make it faster by multiprocessing like so:

with ThreadPoolExecutor(max_workers=25) as executor:
    for nb in deferred_notebooks:
        executor.submit(process_notebook, nb, True)

In that scenario, process_notebook() should be able to just call executor.submit() and re-add itself to the queue if its dependencies aren't met.

What I can't figure out is how to replicate this part:

    # This should clear min one deferral each iteration. If it doesn't, then some dependency
    # is impossible to meet and we should give up.

    if len(deferred_notebooks) == last_count:
        break

How can I signal to the function that it needs to give up trying because there's some impossible dependency that's not clearing?

concurrent.futures multithreading python
1 Answers

If I understand correctly, you've changed your algorithm from batched serial processing to a continuous concurrent processing. Thus, you've lost your checkpoint at the end of every batch, where you can say, "abort if each and every processing failed."

One option would be to process concurrently in batches, something like so:

def process_item(item):
    # return True if processed successfully, False otherwise

with ThreadPoolExecutor(...) as executor:
    while len(items) > 0:
        failed = []
        for (item, ok) in zip(items, executor.map(process_item, items)):
            if not ok:
                failed.append(item)
        if len(failed) == len(items):  # all failed
            break
        items = failed
user_1323820
Related