Fix async coroutine limit not respected and add s3/gcs chunk size

[wild-endeavor]

Tracking issue

flyteorg/flyte#6188

Why are the changes needed?

Please see issue.

What changes were proposed in this pull request?

The moment the asyncio.Task is created, it is available for running. Once the await hits on the run coroutines in batches, all those tasks start running, rendering any batching of coroutines moot.

In testing, we also noticed that the s3fs and gcsfs default write chunk sizes were 50MB. We felt that was a bit high and in testing noticed no performance decrease (actually saw possibly non-statistically significant increase) in performance. We are therefore changing it by default to 25MB. Also this setting and the number of coroutine batches the list & dict transformers run will be configurable via environment variables.

The new env vars are:

• _F_P_WRITE_CHUNK_SIZE - This is the number of bytes and by default is set to "26214400" (25 * 2^20)
• _F_TE_MAX_COROS - This is the number of coroutines the type engine will use for lists/dicts and is set to "10".

How was this patch tested?

Tested on our byoc clusters, and by the OSS OP.

Setup process

Screenshots

Check all the applicable boxes

• I updated the documentation accordingly.
• All new and existing tests passed.
• All commits are signed-off.

Related PRs

Docs link

Summary by Bito

This pull request introduces improvements to the Flytekit library by adding configurable chunk sizes for S3 and GCS uploads and enhancing coroutine batching in the type engine. The chunk size for S3/GCS uploads is now set to 25MB by default and can be configured via an environment variable.

Unit tests added: False

Estimated effort to review (1-5, lower is better): 2

[flyte-bot]

Code Review Agent Run Status

• Limitations and other issues: ❌ Failure - The AI Code Review Agent skipped reviewing this change because it is configured to exclude certain pull requests based on the source/target branch or the pull request status. You can change the settings here, or contact the agent instance creator at [email protected].

[codecov]

Codecov Report

Attention: Patch coverage is 78.94737% with 4 lines in your changes missing coverage. Please review.

Project coverage is 51.83%. Comparing base (06db981) to head (aad5250).
Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
flytekit/core/type_engine.py	73.33%	4 Missing ⚠️

❗ There is a different number of reports uploaded between BASE (06db981) and HEAD (aad5250). Click for more details.

HEAD has 9 uploads less than BASE

Flag	BASE (06db981)	HEAD (aad5250)
	10	1

Additional details and impacted files

@@             Coverage Diff             @@
##           master    #3080       +/-   ##
===========================================
- Coverage   74.82%   51.83%   -23.00%     
===========================================
  Files         202      202               
  Lines       21446    21454        +8     
  Branches     2763     2763               
===========================================
- Hits        16048    11120     -4928     
- Misses       4626     9727     +5101     
+ Partials      772      607      -165     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[flyte-bot]

Code Review Agent Run Status

• Limitations and other issues: ❌ Failure - The AI Code Review Agent skipped reviewing this change because it is configured to exclude certain pull requests based on the source/target branch or the pull request status. You can change the settings here, or contact the agent instance creator at [email protected].

[flyte-bot]

Code Review Agent Run Status

• Limitations and other issues: ❌ Failure - The AI Code Review Agent skipped reviewing this change because it is configured to exclude certain pull requests based on the source/target branch or the pull request status. You can change the settings here, or contact the agent instance creator at [email protected].

[flyte-bot]

Code Review Agent Run Status

• Limitations and other issues: ❌ Failure - The AI Code Review Agent skipped reviewing this change because it is configured to exclude certain pull requests based on the source/target branch or the pull request status. You can change the settings here, or contact the agent instance creator at [email protected].

[eapolinario]

What about https://github.com/flyteorg/flytekit/blob/master/flytekit/core/base_task.py#L638 ?

Edit: this one is ok since it's just parallelizing outputs.

[flyte-bot]

Code Review Agent Run #c3e341

Actionable Suggestions - 0

Additional Suggestions - 1

• tests/flytekit/unit/core/test_list.py - 1

  • Improve error message clarity · Line 51-52

Review Details

• Files reviewed - 4 · Commit Range: 9c291f6..3fbb59a

  • flytekit/core/data_persistence.py
  • flytekit/core/type_engine.py
  • tests/flytekit/unit/core/test_data_persistence.py
  • tests/flytekit/unit/core/test_list.py
• Files skipped - 0
• Tools

  • Whispers (Secret Scanner) - ✔︎ Successful
  • Detect-secrets (Secret Scanner) - ✔︎ Successful
  • MyPy (Static Code Analysis) - ✔︎ Successful
  • Astral Ruff (Static Code Analysis) - ✔︎ Successful

---

AI Code Review powered by

[flyte-bot]

Changelist by Bito

This pull request implements the following key changes.

Key Change	Files Impacted
Feature Improvement - Add Configurable Chunk Size for S3/GCS	- data_persistence.py - Introduced configurable chunk size for S3/GCS uploads. - test_data_persistence.py - Added tests for configurable chunk size in data persistence.
Feature Improvement - Enhance Coroutine Batching in Type Engine	- type_engine.py - Improved coroutine batching with configurable batch size. - test_list.py - Added tests for coroutine batching in list transformer.