Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Adds circuit breaker when a task depends on itself #13882

Merged
merged 4 commits into from
Jun 11, 2024

Conversation

desertaxle
Copy link
Member

@desertaxle desertaxle commented Jun 7, 2024

Adds a circuit breaker to raise when a task depends on itself which would otherwise cause a flow to hang indefinitely.

Closes #13874

Example

This flow sneakily has a task that depends on itself by updating an array in place:

from prefect import flow, task

@task
def say_hello(name):
    return f"Hello {name}!"

@flow
def my_flow():
    greeting_queue = []
    for i in range(2):
        if greeting_queue:
            wait_for = greeting_queue
        else:
            wait_for = []
        future = say_hello.submit(name=f"Person {i}", wait_for=wait_for)
        greeting_queue.append(future)
    
    for fut in greeting_queue:
        print(fut.result())

Running this flow with this PR will raise an error like this:

ValueError: Discovered a task depending on itself. Raising to avoid a deadlock. Please inspect the inputs and dependencies of say_hello.

Checklist

  • This pull request references any related issue by including "closes <link to issue>"
    • If no issue exists and your change is not a small fix, please create an issue first.
  • If this pull request adds new functionality, it includes unit tests that cover the changes
  • This pull request includes a label categorizing the change e.g. maintenance, fix, feature, enhancement, docs.

For documentation changes:

  • This pull request includes redirect settings in mint.json for files that are removed or renamed.

For new functions or classes in the Python SDK:

  • This pull request includes helpful docstrings.
  • If a new Python file was added, this pull request contains a stub page in the Python SDK docs and an entry in docs/mint.json navigation.

@desertaxle desertaxle requested a review from a team as a code owner June 7, 2024 19:01
@desertaxle desertaxle added the fix A fix for a bug in an existing feature label Jun 7, 2024
tests/test_tasks.py Outdated Show resolved Hide resolved
else:
wait_for = []
future = say_hello.submit(name=f"Person {i}", wait_for=wait_for)
greeting_queue.append(future)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, so the only way this can happen is if the caller changes the wait_for iterable after submitting the task. If we just make sure to make our own copy of wait_for, doesn't this problem go away?

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@desertaxle pointed out copying could cause memory issues. Maybe better to start simple and copy if we need to later?

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It wouldn't be a copy of the futures, just a copy of the iterable the caller gave us so that it's kind of "causally locked in" at that point (I'm just thinking literally wait_for = list(wait_for). I can't help but think there are other classes of bugs lurking if we let this list of futures be mutable after it's passed to the engine.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This also happens if futures are passed in parameters which can be deeply nested and would require a deepcopy to prevent subsequent modification from causing deadlocks. This probably won't happen often, so I'd rather raise in the unlikely chance this happens then incur a memory penalty in all other cases.

@desertaxle desertaxle merged commit 35599d8 into main Jun 11, 2024
26 checks passed
@desertaxle desertaxle deleted the dont-wait-on-yourself branch June 11, 2024 13:49
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
fix A fix for a bug in an existing feature
Projects
None yet
Development

Successfully merging this pull request may close these issues.

Strange hanging behavior with ThreadPoolTaskRunner and wait_for
3 participants