-
Notifications
You must be signed in to change notification settings - Fork 15
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #133 from macwilk/mac/sync-async-utils
Add Async Utilities to Mitigate Blocking Sync Functions
- Loading branch information
Showing
4 changed files
with
272 additions
and
3 deletions.
There are no files selected for viewing
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| import asyncio | ||
| import os | ||
| import time | ||
|
|
||
| from dotenv import load_dotenv | ||
|
|
||
| from hatchet_sdk import Context, sync_to_async | ||
| from hatchet_sdk.v2.hatchet import Hatchet | ||
|
|
||
| os.environ["PYTHONASYNCIODEBUG"] = "1" | ||
| load_dotenv() | ||
|
|
||
| hatchet = Hatchet(debug=True) | ||
|
|
||
|
|
||
| @hatchet.function() | ||
| async def fanout_sync_async(context: Context) -> dict: | ||
| print("spawning child") | ||
|
|
||
| context.put_stream("spawning...") | ||
| results = [] | ||
|
|
||
| n = context.workflow_input().get("n", 10) | ||
|
|
||
| start_time = time.time() | ||
| for i in range(n): | ||
| results.append( | ||
| ( | ||
| await context.aio.spawn_workflow( | ||
| "Child", | ||
| {"a": str(i)}, | ||
| key=f"child{i}", | ||
| options={"additional_metadata": {"hello": "earth"}}, | ||
| ) | ||
| ).result() | ||
| ) | ||
|
|
||
| result = await asyncio.gather(*results) | ||
|
|
||
| execution_time = time.time() - start_time | ||
| print(f"Completed in {execution_time:.2f} seconds") | ||
|
|
||
| return {"results": result} | ||
|
|
||
|
|
||
| @hatchet.workflow(on_events=["child:create"]) | ||
| class Child: | ||
| ###### Example Functions ###### | ||
| def sync_blocking_function(self): | ||
| time.sleep(5) | ||
| return {"type": "sync_blocking"} | ||
|
|
||
| @sync_to_async # this makes the function async safe! | ||
| def decorated_sync_blocking_function(self): | ||
| time.sleep(5) | ||
| return {"type": "decorated_sync_blocking"} | ||
|
|
||
| @sync_to_async # this makes the async function loop safe! | ||
| async def async_blocking_function(self): | ||
| time.sleep(5) | ||
| return {"type": "async_blocking"} | ||
|
|
||
| ###### Hatchet Steps ###### | ||
| @hatchet.step() | ||
| async def handle_blocking_sync_in_async(self, context: Context): | ||
| wrapped_blocking_function = sync_to_async(self.sync_blocking_function) | ||
|
|
||
| # This will now be async safe! | ||
| data = await wrapped_blocking_function() | ||
| return {"blocking_status": "success", "data": data} | ||
|
|
||
| @hatchet.step() | ||
| async def handle_decorated_blocking_sync_in_async(self, context: Context): | ||
| data = await self.decorated_sync_blocking_function() | ||
| return {"blocking_status": "success", "data": data} | ||
|
|
||
| @hatchet.step() | ||
| async def handle_blocking_async_in_async(self, context: Context): | ||
| data = await self.async_blocking_function() | ||
| return {"blocking_status": "success", "data": data} | ||
|
|
||
| @hatchet.step() | ||
| async def non_blocking_async(self, context: Context): | ||
| await asyncio.sleep(5) | ||
| return {"nonblocking_status": "success"} | ||
|
|
||
|
|
||
| def main(): | ||
| worker = hatchet.worker("fanout-worker", max_runs=50) | ||
| worker.register_workflow(Child()) | ||
| worker.start() | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| main() |
This file contains 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
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| import asyncio | ||
| import inspect | ||
| from functools import partial, wraps | ||
|
|
||
|
|
||
| def sync_to_async(func): | ||
| """ | ||
| A decorator to run a synchronous function or coroutine in an asynchronous context with added | ||
| asyncio loop safety. | ||
| This decorator allows you to safely call synchronous functions or coroutines from an | ||
| asynchronous function by running them in an executor. | ||
| Args: | ||
| func (callable): The synchronous function or coroutine to be run asynchronously. | ||
| Returns: | ||
| callable: An asynchronous wrapper function that runs the given function in an executor. | ||
| Example: | ||
| @sync_to_async | ||
| def sync_function(x, y): | ||
| return x + y | ||
| @sync_to_async | ||
| async def async_function(x, y): | ||
| return x + y | ||
| def undecorated_function(x, y): | ||
| return x + y | ||
| async def main(): | ||
| result1 = await sync_function(1, 2) | ||
| result2 = await async_function(3, 4) | ||
| result3 = await sync_to_async(undecorated_function)(5, 6) | ||
| print(result1, result2, result3) | ||
| asyncio.run(main()) | ||
| """ | ||
|
|
||
| @wraps(func) | ||
| async def run(*args, loop=None, executor=None, **kwargs): | ||
| """ | ||
| The asynchronous wrapper function that runs the given function in an executor. | ||
| Args: | ||
| *args: Positional arguments to pass to the function. | ||
| loop (asyncio.AbstractEventLoop, optional): The event loop to use. If None, the current running loop is used. | ||
| executor (concurrent.futures.Executor, optional): The executor to use. If None, the default executor is used. | ||
| **kwargs: Keyword arguments to pass to the function. | ||
| Returns: | ||
| The result of the function call. | ||
| """ | ||
| if loop is None: | ||
| loop = asyncio.get_running_loop() | ||
|
|
||
| if inspect.iscoroutinefunction(func): | ||
| # Wrap the coroutine to run it in an executor | ||
| async def wrapper(): | ||
| return await func(*args, **kwargs) | ||
|
|
||
| pfunc = partial(asyncio.run, wrapper()) | ||
| return await loop.run_in_executor(executor, pfunc) | ||
| else: | ||
| # Run the synchronous function in an executor | ||
| pfunc = partial(func, *args, **kwargs) | ||
| return await loop.run_in_executor(executor, pfunc) | ||
|
|
||
| return run |
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,6 @@ | ||
| [tool.poetry] | ||
| name = "hatchet-sdk" | ||
| version = "0.34.0a5" | ||
| version = "0.34.0a6" | ||
| description = "" | ||
| authors = ["Alexander Belanger <[email protected]>"] | ||
| readme = "README.md" | ||
|
|
@@ -42,7 +42,7 @@ known_third_party = [ | |
| "python_dotenv", | ||
| "python_dateutil", | ||
| "pyyaml", | ||
| "urllib3" | ||
| "urllib3", | ||
| ] | ||
|
|
||
| [tool.poetry.scripts] | ||
|
|
||