You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
When executing a dynamic workflow in the Flyte backend, compiling the future.pb file before sub-node execution takes some time. If an out-of-memory (OOM) error occurs during sub-node execution, re-running the dynamic workflow requires recompiling the future.pb file, consuming additional time and computational resources. To address this, we propose caching the compiled future.pb file.
The future cache mechanism in this PR stores the compiled file as artifact data in Flyte storage via the datacatalog. The cache key, TagName, is determined by the hash of the input values. The diagrams below illustrate the data flow in the backend system:
What changes were proposed in this pull request?
Please refer to bellow diagram for understanding the procedure of how future read/write:
key changes:
1.Added Mode to TaskTemplate Metadata in flyteidl to indicate whether a task is a DynamicTask.
2.Implemented logic in flytepropeller to send cache read/write requests to datacatalog.
3.Added cache read/write functionality to the storage layer in datacatalog.
How was this patch tested?
To test how much time it saved when future file cache hit, we modify this block of code to count the time spend to run taskNodeHandler.Handle for test purpose.
startTime := time.Now()
if !isDynamic || cacheStatus == nil || cacheStatus.GetCacheStatus() != core.CatalogCacheStatus_CACHE_HIT {
var err error
taskTrns, err := d.TaskNodeHandler.Handle(ctx, nCtx)
trns = &taskTrns
if err != nil {
logger.Debug(ctx, "Failed to compile dynamic workflow")
return *trns, prevState, err
}
}
if isDynamic {
duration := time.Since(startTime)
logger.Infof(ctx, "Dynamic node compilation took %v", duration)
}
To mimic OOM, we will throw panic in dynamicNodeHandler. So the Dynamic workflow will be aborted after future file compiled
then we run bellow test dynamic workflow for the first time
import typing
from flytekit import dynamic, task, workflow
@task
def return_index(character: str) -> int:
"""
Computes the character index (which needs to fit into the 26 characters list)
"""
if character.islower():
return ord(character) - ord("a")
else:
return ord(character) - ord("A")
@task
def update_list(freq_list: typing.List[int], list_index: int) -> typing.List[int]:
"""
Notes the frequency of characters
"""
freq_list[list_index] += 1
return freq_list
@task
def derive_count(freq1: typing.List[int], freq2: typing.List[int]) -> int:
"""
Derives the number of common characters
"""
count = 0
for i in range(26):
count += min(freq1[i], freq2[i])
return count
@dynamic(cache_version="4", cache=True)
def count_characters(s1: str, s2: str) -> int:
"""
Calls the required tasks and returns the final result
"""
# s1 and s2 are accessible
# initiliaze an empty list consisting of 26 empty slots corresponding to every alphabet (lower and upper case)
freq1 = [0] * 26
freq2 = [0] * 26
# looping through the string s1
for i in range(len(s1)):
# index and freq1 are not accesible as they are promises
index = return_index(character=s1[i])
freq1 = update_list(freq_list=freq1, list_index=index)
# looping through the string s2
for i in range(len(s2)):
# index and freq2 are not accesible as they are promises
index = return_index(character=s2[i])
freq2 = update_list(freq_list=freq2, list_index=index)
# counting the common characters
return derive_count(freq1=freq1, freq2=freq2)
@workflow
def wf(s1: str, s2: str) -> int:
"""
Calls the dynamic workflow and returns the result
"""
# sending two strings to the workflow
return count_characters(s1=s1, s2=s2)
We use the command pyflyte run --remote dynamic.py wf --s1="dynamic" --s2="test"
The first run took around 17 ms to compile future file
Then we remove the panic throwing and re-execute the same workflow with the same input. The future cache hit and it only took 83ns to go through the taskNodeHandler.Handle code block, which means the taskNodeHandler.Handle is skipped.
Moreover, unit tests has been added to make sure cache behaviors.
Labels
Please add one or more of the following labels to categorize your PR:
added: For new features.
This is important to improve the readability of release notes.
Limitations and other issues: ❌ Failure - Bito Code Review Agent didn't review this pull request automatically because it exceeded the size limit. No action is needed if you didn't intend for the agent to review it. Otherwise, you can initiate the review by typing /review in a comment below.
Limitations and other issues: ❌ Failure - Bito Code Review Agent didn't review this pull request automatically because it exceeded the size limit. No action is needed if you didn't intend for the agent to review it. Otherwise, you can initiate the review by typing /review in a comment below.
Limitations and other issues: ❌ Failure - Bito Code Review Agent didn't review this pull request automatically because it exceeded the size limit. No action is needed if you didn't intend for the agent to review it. Otherwise, you can initiate the review by typing /review in a comment below.
Limitations and other issues: ❌ Failure - Bito Code Review Agent didn't review this pull request automatically because it exceeded the size limit. No action is needed if you didn't intend for the agent to review it. Otherwise, you can initiate the review by typing /review in a comment below.
This an interesting change @popojk. I have been working in this area lately and I'm trying to fully understand where this provides values.
When you say OOM happens during sub node execution are you referring to OOM in Propeller or OOM in the task?
I think during normal operation compiling the workflow only happens once so I'm skeptical if the change here is worth it.
Hi @Sovietaced . This PR is aiming to add a cache for future.pb that compiled by dynamic workflow. As far as I know, when a task in dynamic workflow OOM during execution, propeller will re-compile future.pb while re-executing the same dynamic workflow, leading to duplicated computation. Please let me know if you have any other suggestions. 😎
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Tracking issue
Closes #5543
Why are the changes needed?
When executing a dynamic workflow in the Flyte backend, compiling the future.pb file before sub-node execution takes some time. If an out-of-memory (OOM) error occurs during sub-node execution, re-running the dynamic workflow requires recompiling the future.pb file, consuming additional time and computational resources. To address this, we propose caching the compiled future.pb file.
The future cache mechanism in this PR stores the compiled file as artifact data in Flyte storage via the datacatalog. The cache key, TagName, is determined by the hash of the input values. The diagrams below illustrate the data flow in the backend system:
What changes were proposed in this pull request?
Please refer to bellow diagram for understanding the procedure of how future read/write:
key changes:
1.Added Mode to TaskTemplate Metadata in flyteidl to indicate whether a task is a DynamicTask.
2.Implemented logic in flytepropeller to send cache read/write requests to datacatalog.
3.Added cache read/write functionality to the storage layer in datacatalog.
How was this patch tested?
To test how much time it saved when future file cache hit, we modify this block of code to count the time spend to run taskNodeHandler.Handle for test purpose.
To mimic OOM, we will throw panic in dynamicNodeHandler. So the Dynamic workflow will be aborted after future file compiled
then we run bellow test dynamic workflow for the first time
We use the command
pyflyte run --remote dynamic.py wf --s1="dynamic" --s2="test"The first run took around 17 ms to compile future file
Then we remove the panic throwing and re-execute the same workflow with the same input. The future cache hit and it only took 83ns to go through the taskNodeHandler.Handle code block, which means the taskNodeHandler.Handle is skipped.
Moreover, unit tests has been added to make sure cache behaviors.
Labels
Please add one or more of the following labels to categorize your PR:
This is important to improve the readability of release notes.
Check all the applicable boxes
Related PRs
flytekit PR