Refactor interactive argument handling #585
Conversation
WalkthroughThis pull request refactors how future objects and exceptions are handled within the executor library. In Changes
Sequence Diagram(s)sequenceDiagram
participant C as Client
participant E as Task Executor
participant F as Future Handler
C->>E: Initiate task execution
E->>F: get_future_objects_from_input(args, kwargs)
F-->>E: Return list of futures and flag
E->>F: check_exception_was_raised(future)
F-->>E: Return exception status
E->>F: update_futures_in_input(args, kwargs)
F-->>E: Return updated arguments
E-->>C: Complete task execution with resolved futures
Possibly related PRs
Poem
✨ Finishing Touches
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
Codecov ReportAll modified and coverable lines are covered by tests ✅
Additional details and impacted files@@ Coverage Diff @@
## main #585 +/- ##
==========================================
+ Coverage 95.91% 95.92% +0.01%
==========================================
Files 25 26 +1
Lines 1174 1177 +3
==========================================
+ Hits 1126 1129 +3
Misses 48 48 ☔ View full report in Codecov by Sentry. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (4)
tests/test_interactive_future_arguments.py (1)
53-58: Consider adding timeout test case.The test verifies exception detection but doesn't cover the timeout scenario handled in the implementation.
def test_check_exception_was_raised(self): f_with_exception = Future() f_with_exception.set_exception(ValueError()) f_without_exception = Future() + f_not_done = Future() # Add test for timeout case self.assertTrue(check_exception_was_raised(future_obj=f_with_exception)) self.assertFalse(check_exception_was_raised(future_obj=f_without_exception)) + self.assertFalse(check_exception_was_raised(future_obj=f_not_done))executorlib/standalone/interactive/arguments.py (3)
6-33: Consider optimizing future object detection.The implementation traverses the input twice - once for args and once for kwargs. Consider combining them into a single traversal.
def get_future_objects_from_input(args: tuple, kwargs: dict): future_lst = [] def find_future_in_list(lst): for el in lst: if isinstance(el, Future): future_lst.append(el) elif isinstance(el, list): find_future_in_list(lst=el) elif isinstance(el, dict): find_future_in_list(lst=el.values()) - find_future_in_list(lst=args) - find_future_in_list(lst=kwargs.values()) + find_future_in_list(lst=list(args) + list(kwargs.values())) boolean_flag = len([future for future in future_lst if future.done()]) == len( future_lst ) return future_lst, boolean_flag
52-66: Consider making timeout configurable.The timeout value is hardcoded to 10^-10 seconds. Consider making it configurable through a parameter with a reasonable default.
-def check_exception_was_raised(future_obj: Future) -> bool: +def check_exception_was_raised(future_obj: Future, timeout: float = 1e-10) -> bool: """ Check if exception was raised by future object Args: future_obj (Future): future object + timeout (float): timeout in seconds for checking exception Returns: bool: True if exception was raised, False otherwise """ try: - excp = future_obj.exception(timeout=10**-10) + excp = future_obj.exception(timeout=timeout) return excp is not None and not isinstance(excp, CancelledError) except TimeoutError: return False
81-89: Enhance type hints for better code understanding.The type hints could be more specific to better document the expected types.
-def get_result(arg: Union[list[Future], Future]) -> Any: +def get_result(arg: Union[list[Any], dict[str, Any], Future, Any]) -> Any:
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
executorlib/interactive/shared.py(3 hunks)executorlib/standalone/__init__.py(1 hunks)executorlib/standalone/interactive/arguments.py(1 hunks)tests/test_interactive_future_arguments.py(1 hunks)
✅ Files skipped from review due to trivial changes (1)
- executorlib/standalone/init.py
⏰ Context from checks skipped due to timeout of 90000ms (17)
- GitHub Check: unittest_openmpi (ubuntu-latest, 3.10)
- GitHub Check: unittest_openmpi (ubuntu-latest, 3.11)
- GitHub Check: unittest_openmpi (ubuntu-latest, 3.12)
- GitHub Check: unittest_mpich (ubuntu-latest, 3.10)
- GitHub Check: unittest_win
- GitHub Check: unittest_mpich (ubuntu-latest, 3.11)
- GitHub Check: unittest_openmpi (ubuntu-latest, 3.13)
- GitHub Check: unittest_mpich (ubuntu-latest, 3.12)
- GitHub Check: unittest_openmpi (macos-latest, 3.13)
- GitHub Check: unittest_old
- GitHub Check: unittest_mpich (ubuntu-latest, 3.13)
- GitHub Check: unittest_mpich (macos-latest, 3.13)
- GitHub Check: unittest_flux_openmpi
- GitHub Check: unittest_flux_mpich
- GitHub Check: notebooks
- GitHub Check: benchmark (ubuntu-latest, 3.13, .ci_support/environment-mpich.yml)
- GitHub Check: benchmark (ubuntu-latest, 3.13, .ci_support/environment-openmpi.yml)
🔇 Additional comments (3)
tests/test_interactive_future_arguments.py (2)
13-18: LGTM! Test validates future object detection in various data structures.The test thoroughly checks future object detection in positional args, keyword args, lists, and dictionaries.
27-44: LGTM! Test verifies future resolution in complex data structures.The test effectively validates that future objects are correctly resolved and their values are properly propagated through nested data structures.
executorlib/interactive/shared.py (1)
469-480: Potential race condition in future object handling.The function checks future.done() after getting exceptions. This could lead to a race condition if the future completes between these operations.
Consider using a more atomic approach:
for task_wait_dict in wait_lst: - exception_lst = get_exception_lst(future_lst=task_wait_dict["future_lst"]) - if len(exception_lst) > 0: - task_wait_dict["future"].set_exception(exception_lst[0]) - elif all(future.done() for future in task_wait_dict["future_lst"]): + all_done = all(future.done() for future in task_wait_dict["future_lst"]) + if all_done: + exception_lst = get_exception_lst(future_lst=task_wait_dict["future_lst"]) + if len(exception_lst) > 0: + task_wait_dict["future"].set_exception(exception_lst[0]) + else: + del task_wait_dict["future_lst"] + task_wait_dict["args"], task_wait_dict["kwargs"] = update_futures_in_input( + args=task_wait_dict["args"], kwargs=task_wait_dict["kwargs"] + ) + executor_queue.put(task_wait_dict) + else: + wait_tmp_lst.append(task_wait_dict)
| future_lst, ready_flag = get_future_objects_from_input( | ||
| args=task_dict["args"], kwargs=task_dict["kwargs"] | ||
| ) | ||
| exception_lst = get_exception_lst(future_lst=future_lst) | ||
| if not check_exception_was_raised(future_obj=task_dict["future"]): |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add error handling for get_future_objects_from_input.
The function call could fail if args/kwargs contain invalid types. Consider adding try-except block.
- future_lst, ready_flag = get_future_objects_from_input(
- args=task_dict["args"], kwargs=task_dict["kwargs"]
- )
- exception_lst = get_exception_lst(future_lst=future_lst)
- if not check_exception_was_raised(future_obj=task_dict["future"]):
+ try:
+ future_lst, ready_flag = get_future_objects_from_input(
+ args=task_dict["args"], kwargs=task_dict["kwargs"]
+ )
+ exception_lst = get_exception_lst(future_lst=future_lst)
+ if not check_exception_was_raised(future_obj=task_dict["future"]):
+ except Exception as e:
+ task_dict["future"].set_exception(e)
+ future_queue.task_done()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| future_lst, ready_flag = get_future_objects_from_input( | |
| args=task_dict["args"], kwargs=task_dict["kwargs"] | |
| ) | |
| exception_lst = get_exception_lst(future_lst=future_lst) | |
| if not check_exception_was_raised(future_obj=task_dict["future"]): | |
| try: | |
| future_lst, ready_flag = get_future_objects_from_input( | |
| args=task_dict["args"], kwargs=task_dict["kwargs"] | |
| ) | |
| exception_lst = get_exception_lst(future_lst=future_lst) | |
| if not check_exception_was_raised(future_obj=task_dict["future"]): | |
| except Exception as e: | |
| task_dict["future"].set_exception(e) | |
| future_queue.task_done() |
Summary by CodeRabbit
New Features
Refactor
Documentation
Tests