diff --git a/.github/workflows/test-suite.yaml b/.github/workflows/test-suite.yaml index c7cb134b..423ddfbf 100644 --- a/.github/workflows/test-suite.yaml +++ b/.github/workflows/test-suite.yaml @@ -94,7 +94,7 @@ jobs: GLOBUS_COMPUTE_CLIENT_ID: ${{ secrets.GC_CLIENT_ID }} GLOBUS_COMPUTE_CLIENT_SECRET: ${{ secrets.GC_CLIENT_SECRET }} run: | - docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ matrix.python-version }}/bin/activate; pytest -s -v --assert=plain --cov=src/chiltepin --cov-report=term-missing --cov-report=markdown:coverage.md --cov-report=xml --cov-report=html --cov-fail-under=0 --config=tests/configs/docker.yaml" + docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ matrix.python-version }}/bin/activate; pytest -s -v --cov=src/chiltepin --cov-report=term-missing --cov-report=markdown:coverage.md --cov-report=xml --cov-report=html --cov-fail-under=0 --config=tests/configs/docker.yaml" - name: Copy coverage reports from container if: always() @@ -233,7 +233,7 @@ jobs: GLOBUS_COMPUTE_CLIENT_ID: ${{ secrets.GC_CLIENT_ID }} GLOBUS_COMPUTE_CLIENT_SECRET: ${{ secrets.GC_CLIENT_SECRET }} run: | - docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ env.LATEST_PYTHON_VERSION }}/bin/activate; pytest -s -v --assert=plain --config=tests/configs/docker.yaml" + docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ env.LATEST_PYTHON_VERSION }}/bin/activate; pytest -s -v --config=tests/configs/docker.yaml" - name: Debug session if: ${{ failure() }} diff --git a/docs/api.rst b/docs/api.rst index 319e403d..0830f700 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -11,6 +11,13 @@ Configuration Module :members: :show-inheritance: +Workflow Module +--------------- + +.. automodule:: chiltepin.workflow + :members: + :show-inheritance: + Endpoint Management Module --------------------------- diff --git a/docs/configuration.rst b/docs/configuration.rst index 302dcd79..7d112716 100644 --- a/docs/configuration.rst +++ b/docs/configuration.rst @@ -17,6 +17,45 @@ Resources can be: The configuration file defines named resources, where each resource represents a pool of nodes and/or cores on which tasks can be executed. +Default "local" Resource +^^^^^^^^^^^^^^^^^^^^^^^^ + +Chiltepin automatically provides a default resource named **"local"** that is always +available, even if you don't define any resources in your configuration file. This +default resource: + +- Uses the current machine (localhost provider) +- Has 1 core per worker and 1 maximum worker +- Starts with 0 blocks and can scale up to 1 block +- Is useful for testing and light computational tasks + +You can use this default resource without any configuration: + +.. code-block:: python + + from chiltepin import run_workflow + from chiltepin.tasks import python_task + + @python_task + def my_task(): + return "Hello from local!" + + # Works even with an empty config + with run_workflow({}): + result = my_task(executor=["local"]).result() + +You can override the default "local" resource by defining your own resource with +that name in your configuration file. This is useful if you need different settings +for local execution: + +.. code-block:: yaml + + local: + provider: "localhost" + max_blocks: 2 + max_workers_per_node: 4 + cores_per_worker: 2 + Basic Structure --------------- @@ -415,26 +454,70 @@ Loading Configurations Parse and Load ^^^^^^^^^^^^^^ +**Loading from a file:** + .. code-block:: python - import chiltepin.configure - import parsl + from chiltepin import run_workflow - # Parse YAML configuration file - config_dict = chiltepin.configure.parse_file("my_config.yaml") - - # Create Parsl configuration - parsl_config = chiltepin.configure.load( - config_dict, + # Load configuration from YAML file and run workflow + with run_workflow( + "my_config.yaml", include=["compute", "mpi"], # Only load specific resources run_dir="./runinfo" # Directory for Parsl runtime files - ) + ): + # Run your tasks here + result = my_task(executor=["compute"]).result() - # Initialize Parsl with configuration - parsl.load(parsl_config) +**Loading from a dict:** + +.. code-block:: python + + from chiltepin import run_workflow + + # Define configuration as a dictionary + config_dict = { + "compute": { + "provider": "slurm", + "account": "my-account", + "partition": "compute", + "nodes_per_block": 1, + "cores_per_node": 40, + "walltime": "01:00:00", + }, + "mpi": { + "provider": "slurm", + "account": "my-account", + "partition": "compute", + "nodes_per_block": 2, + "launcher": "mpi", + "walltime": "02:00:00", + } + } + + # Load configuration from dict and run workflow + with run_workflow( + config_dict, + include=["compute", "mpi"], # Only load specific resources + run_dir="./runinfo" + ): + # Run your tasks here + result = my_task(executor=["compute"]).result() The ``include`` parameter lets you selectively load only specific resources from your -configuration file. If omitted, all resources are loaded. +configuration. If omitted, all resources are loaded. + +.. note:: + The default **"local"** resource is always available, regardless of the ``include`` + parameter. You do not need to add "local" to the include list to use it. This ensures + you always have a fallback resource available for tasks. + + .. code-block:: python + + # "local" is available even though not in include list + with run_workflow("config.yaml", include=["compute"]): + local_result = my_task(executor=["local"]).result() # Works! + compute_result = my_task(executor=["compute"]).result() # Works! Configuration Best Practices ----------------------------- diff --git a/docs/container.rst b/docs/container.rst index 5d514008..58c3ed58 100644 --- a/docs/container.rst +++ b/docs/container.rst @@ -66,13 +66,13 @@ After installation, you can run the test suite using the Docker-specific configu .. code-block:: console - (container) $ pytest --assert=plain --config=tests/configs/docker.yaml + (container) $ pytest --config=tests/configs/docker.yaml For more verbose output: .. code-block:: console - (container) $ pytest -s -vvv --assert=plain --config=tests/configs/docker.yaml + (container) $ pytest -s -vvv --config=tests/configs/docker.yaml Adjusting Core Count -------------------- diff --git a/docs/data.rst b/docs/data.rst index 60c7e33e..5e6e44ca 100644 --- a/docs/data.rst +++ b/docs/data.rst @@ -45,7 +45,7 @@ Basic Usage dst_ep="hpc-scratch", src_path="/Users/me/data/input.dat", dst_path="/scratch/project/input.dat", - executor="local" + executor=["local"] ) # Wait for transfer to complete @@ -115,7 +115,7 @@ Transfer entire directories recursively: src_path="/Users/me/project/data/", dst_path="/scratch/project/data/", recursive=True, - executor="local" + executor=["local"] ) Endpoint Names vs UUIDs @@ -159,7 +159,7 @@ Basic Usage delete_future = delete_task( src_ep="hpc-scratch", src_path="/scratch/project/temp.dat", - executor="local" + executor=["local"] ) # Wait for deletion to complete @@ -219,7 +219,7 @@ Delete entire directories: src_ep="hpc-scratch", src_path="/scratch/project/temp_data/", recursive=True, - executor="local" + executor=["local"] ) .. warning:: @@ -238,8 +238,7 @@ A common pattern is to stage data, process it, then clean up: .. code-block:: python - import parsl - import chiltepin.configure + from chiltepin import run_workflow from chiltepin.tasks import python_task from chiltepin.data import transfer_task, delete_task @@ -251,24 +250,21 @@ A common pattern is to stage data, process it, then clean up: result = df.mean().to_dict() return result - # Load configuration and start Parsl - config_dict = chiltepin.configure.parse_file("config.yaml") - parsl_config = chiltepin.configure.load(config_dict) - - with parsl.load(parsl_config): + # Load configuration and start workflow + with run_workflow("config.yaml"): # Stage data to compute resource stage_in = transfer_task( src_ep="my-laptop", dst_ep="hpc-scratch", src_path="/Users/me/data/dataset.csv", dst_path="/scratch/project/dataset.csv", - executor="local" + executor=["local"] ) # Process the staged data (waits for transfer via inputs) analysis = analyze_data( "/scratch/project/dataset.csv", - executor="compute", + executor=["compute"], inputs=[stage_in] # Non-blocking dependency ) @@ -276,7 +272,7 @@ A common pattern is to stage data, process it, then clean up: cleanup = delete_task( src_ep="hpc-scratch", src_path="/scratch/project/dataset.csv", - executor="local", + executor=["local"], inputs=[analysis] # Non-blocking dependency ) @@ -306,7 +302,7 @@ Transfer multiple files in parallel: dst_ep="hpc-scratch", src_path=f"/Users/me/data/{filename}", dst_path=f"/scratch/project/{filename}", - executor="local" + executor=["local"] ) transfers.append(future) @@ -340,8 +336,8 @@ To run a transfer or deletion after multiple tasks complete, pass them via the return True # Generate files in parallel - config_ready = generate_config(executor="compute") - input_ready = generate_input(executor="compute") + config_ready = generate_config(executor=["compute"]) + input_ready = generate_input(executor=["compute"]) # Wait for both files before transferring (non-blocking dependency) transfer = transfer_task( @@ -350,7 +346,7 @@ To run a transfer or deletion after multiple tasks complete, pass them via the src_path="/scratch/", dst_path="/results/", recursive=True, - executor="local", + executor=["local"], inputs=[config_ready, input_ready] # Multiple dependencies ) @@ -386,21 +382,21 @@ dependencies: dst_ep="hpc-scratch", src_path="/data/raw.csv", dst_path="/scratch/raw.csv", - executor="local" + executor=["local"] ) # Preprocess (waits for transfer via inputs) preprocess_future = preprocess( "/scratch/raw.csv", "/scratch/clean.csv", - executor="compute", + executor=["compute"], inputs=[stage_raw] # Wait for transfer non-blocking ) # Analyze (waits for preprocess completing and passing output path) analysis_future = analyze( preprocess_future, # Parsl waits for this future and passes the output path - executor="compute" + executor=["compute"] ) # Stage results back (waits for analysis via inputs) @@ -409,7 +405,7 @@ dependencies: dst_ep="my-laptop", src_path="/scratch/clean.csv", dst_path="/data/cleaned_output.csv", - executor="local", + executor=["local"], inputs=[analysis_future] # Wait for analysis non-blocking ) @@ -418,7 +414,7 @@ dependencies: src_ep="hpc-scratch", src_path="/scratch/", recursive=True, - executor="local", + executor=["local"], inputs=[stage_out] # Wait for stage out non-blocking ) diff --git a/docs/quickstart.rst b/docs/quickstart.rst index a9e6c214..80fe39cc 100644 --- a/docs/quickstart.rst +++ b/docs/quickstart.rst @@ -89,8 +89,8 @@ Create ``my_config.yaml`` with your endpoint UUID: .. code-block:: yaml - # Local resource for small tasks - local: + # Laptop/workstation resource for small tasks + laptop: provider: "localhost" init_blocks: 1 max_blocks: 1 @@ -116,8 +116,7 @@ Create ``my_workflow.py``: .. code-block:: python - import parsl - import chiltepin.configure + from chiltepin import run_workflow from chiltepin.tasks import bash_task, python_task # Define tasks @@ -137,23 +136,16 @@ Create ``my_workflow.py``: return result if __name__ == "__main__": - # Load configuration - config_dict = chiltepin.configure.parse_file("my_config.yaml") - parsl_config = chiltepin.configure.load( - config_dict, - include=["local", "remote"], - run_dir="./runinfo" - ) - - with parsl.load(parsl_config): - # Run local task on "local" resource - local_future = hello_local(executor="local") + # Load configuration and run workflow + with run_workflow("my_config.yaml", include=["laptop", "remote"], run_dir="./runinfo"): + # Run local task on "laptop" resource + local_future = hello_local(executor=["laptop"]) # Run remote bash task on "remote" resource (returns exit code: 0 = success) - remote_future = hello_remote(executor="remote") + remote_future = hello_remote(executor=["remote"]) # Run multiple compute tasks on "remote" resource - futures = [compute_task(i, executor="remote") for i in range(1, 5)] + futures = [compute_task(i, executor=["remote"]) for i in range(1, 5)] # Get the results print(f"Local: {local_future.result()}") @@ -203,7 +195,7 @@ Configuration File (``local_config.yaml``) .. code-block:: yaml - local: + laptop: provider: "localhost" init_blocks: 1 max_blocks: 1 @@ -213,8 +205,7 @@ Simple Workflow (``simple_workflow.py``) .. code-block:: python - import parsl - import chiltepin.configure + from chiltepin import run_workflow from chiltepin.tasks import bash_task, python_task # Define tasks @@ -227,15 +218,12 @@ Simple Workflow (``simple_workflow.py``) return "echo 'Task completed successfully'" if __name__ == "__main__": - # Load configuration - config_dict = chiltepin.configure.parse_file("local_config.yaml") - parsl_config = chiltepin.configure.load(config_dict, run_dir="./runinfo") - - with parsl.load(parsl_config): - result = multiply(6, 7, executor="local").result() + # Load configuration and run workflow + with run_workflow("local_config.yaml", run_dir="./runinfo"): + result = multiply(6, 7, executor=["laptop"]).result() print(f"6 * 7 = {result}") - exit_code = system_info(executor="local").result() + exit_code = system_info(executor=["laptop"]).result() print(f"Bash task exit code: {exit_code}") Run it: @@ -274,8 +262,7 @@ MPI Workflow .. code-block:: python - import parsl - import chiltepin.configure + from chiltepin import run_workflow from chiltepin.tasks import bash_task @bash_task @@ -287,18 +274,15 @@ MPI Workflow return f"srun -n {ranks} ./mpi_app" if __name__ == "__main__": - config_dict = chiltepin.configure.parse_file("mpi_config.yaml") - parsl_config = chiltepin.configure.load(config_dict, run_dir="./runinfo") - - with parsl.load(parsl_config): + with run_workflow("mpi_config.yaml", run_dir="./runinfo"): # Compile MPI application on the MPI resource (returns exit code) - compile_result = compile_mpi(executor="mpi-resource-name").result() + compile_result = compile_mpi(executor=["mpi-resource-name"]).result() print(f"Compilation exit code: {compile_result}") # Run with different rank counts on the MPI resource results = [] for ranks in [4, 8, 16]: - future = run_mpi(ranks, executor="mpi-resource-name") + future = run_mpi(ranks, executor=["mpi-resource-name"]) results.append(future.result()) for i, result in enumerate(results, 1): @@ -336,7 +320,7 @@ When calling a task, use the ``executor`` parameter to specify which resource to return "result" # Specify which resource to use - result = my_task(executor="compute").result() + result = my_task(executor=["compute"]).result() The ``executor`` value must match a resource name from your configuration file. @@ -347,16 +331,46 @@ The ``executor`` value must match a resource name from your configuration file. Configuration Loading ^^^^^^^^^^^^^^^^^^^^^ -The ``include`` parameter selects specific resources to load from the configuration: +The ``include`` parameter selects specific resources to load from the configuration. + +**Loading from a file:** + +.. code-block:: python + + # Load only specific resources from YAML file + with run_workflow( + "my_config.yaml", + include=["laptop", "compute"], # Only these resources + run_dir="./runinfo" + ): + # Run tasks using selected resources + result = my_task(executor=["compute"]).result() + +**Loading from a dict:** .. code-block:: python - # Load only specific resources - parsl_config = chiltepin.configure.load( - config_dict, - include=["local", "compute"], # Only these resources + # Define configuration as a dictionary + config = { + "laptop": { + "provider": "localhost", + "cores_per_node": 4, + }, + "compute": { + "provider": "slurm", + "partition": "compute", + "nodes_per_block": 1, + } + } + + # Load only specific resources from dict + with run_workflow( + config, + include=["laptop", "compute"], # Only these resources run_dir="./runinfo" - ) + ): + # Run tasks using selected resources + result = my_task(executor=["compute"]).result() If ``include`` is omitted, all resources in the configuration are loaded. diff --git a/docs/tasks.rst b/docs/tasks.rst index 6b5b6c7c..aa3e57d6 100644 --- a/docs/tasks.rst +++ b/docs/tasks.rst @@ -47,7 +47,7 @@ Basic Usage return "Hello from a Chiltepin task!" # Call the task and specify where to run it - future = hello_world(executor="my-resource") + future = hello_world(executor=["my-resource"]) result = future.result() # Wait for completion and get result print(result) # "Hello from a Chiltepin task!" @@ -63,11 +63,11 @@ Python tasks can accept both positional and keyword arguments: return (a + b) * multiply_by # Use positional arguments - future1 = add_numbers(5, 3, executor="compute") + future1 = add_numbers(5, 3, executor=["compute"]) print(future1.result()) # 8 # Use keyword arguments - future2 = add_numbers(5, 3, multiply_by=2, executor="compute") + future2 = add_numbers(5, 3, multiply_by=2, executor=["compute"]) print(future2.result()) # 16 Importing Modules @@ -111,9 +111,9 @@ Python tasks can return any serializable Python object: import pandas as pd return pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]}) - list_result = get_list(5, executor="local").result() - dict_result = get_dict("temperature", 72.5, executor="local").result() - df_result = get_dataframe(executor="compute").result() + list_result = get_list(5, executor=["local"]).result() + dict_result = get_dict("temperature", 72.5, executor=["local"]).result() + df_result = get_dataframe(executor=["compute"]).result() Bash Tasks ---------- @@ -133,7 +133,7 @@ Basic Usage return "echo 'Hello from bash!'" # Bash tasks return the exit code (0 = success) - future = echo_hello(executor="my-resource") + future = echo_hello(executor=["my-resource"]) exit_code = future.result() print(exit_code) # 0 @@ -152,7 +152,7 @@ Use function arguments to dynamically construct commands: def run_simulation(config_file, num_steps): return f"./my_simulator --config {config_file} --steps {num_steps}" - exit_code = process_file("data.txt", "sorted.txt", executor="compute").result() + exit_code = process_file("data.txt", "sorted.txt", executor=["compute"]).result() .. warning:: Be careful with shell injection vulnerabilities. Validate and sanitize inputs @@ -172,7 +172,7 @@ By default, bash tasks return the exit code. To capture stdout or stderr, use th # Capture stdout to a file future = get_hostname( - executor="compute", + executor=["compute"], stdout="hostname_output.txt" ) exit_code = future.result() @@ -196,7 +196,7 @@ You can also capture stderr for debugging: return "some_command_that_might_fail" future = risky_command( - executor="compute", + executor=["compute"], stdout="output.txt", stderr="errors.txt" ) @@ -245,9 +245,9 @@ Join tasks call other tasks and return futures: @join_task def process_list(values, factor): # Launch multiple tasks in parallel - futures = [multiply(v, factor, executor="compute") for v in values] + futures = [multiply(v, factor, executor=["compute"]) for v in values] # Aggregate results with another task - return add_values(*futures, executor="compute") + return add_values(*futures, executor=["compute"]) # Process [1, 2, 3] with factor 2: (1*2) + (2*2) + (3*2) = 12 result = process_list([1, 2, 3], 2).result() @@ -277,9 +277,9 @@ Example - Processing Multiple Files: @join_task def process_all_files(file_list): # Process all files in parallel - futures = [process_file(f, executor="compute") for f in file_list] + futures = [process_file(f, executor=["compute"]) for f in file_list] # Check if all succeeded - return check_all_success(futures, executor="local") + return check_all_success(futures, executor=["local"]) files = ["data1.txt", "data2.txt", "data3.txt"] success = process_all_files(files).result() @@ -307,15 +307,15 @@ Join tasks can coordinate both python and bash tasks: @join_task def compile_and_run(input_file): # First compile - compile_future = compile_code(executor="compute") + compile_future = compile_code(executor=["compute"]) compile_future.result() # Wait for compilation # Then run - run_future = run_app(input_file, executor="compute", stdout="output.txt") + run_future = run_app(input_file, executor=["compute"], stdout="output.txt") run_future.result() # Wait for execution # Parse results - return parse_results("output.txt", executor="local") + return parse_results("output.txt", executor=["local"]) result = compile_and_run("input.dat").result() @@ -354,9 +354,9 @@ object-oriented workflow design: # Create instance and use tasks processor = DataProcessor({'normalize': True, 'export_format': 'json'}) - data = processor.load_data("input.csv", executor="compute").result() - transformed = processor.transform_data(data, executor="compute").result() - exit_code = processor.export_data("output.json", executor="compute").result() + data = processor.load_data("input.csv", executor=["compute"]).result() + transformed = processor.transform_data(data, executor=["compute"]).result() + exit_code = processor.export_data("output.json", executor=["compute"]).result() .. warning:: **Mutable Object State**: When using class methods as tasks, be aware that mutable @@ -389,7 +389,7 @@ Specify a single resource by name: return "result" # Run on the "compute" resource from your config - future = my_task(executor="compute") + future = my_task(executor=["compute"]) Multiple Resources ^^^^^^^^^^^^^^^^^^ @@ -433,7 +433,7 @@ Call ``.result()`` to wait for task completion and retrieve the result: time.sleep(2) return 42 - future = compute_value(executor="compute") + future = compute_value(executor=["compute"]) print("Task submitted, doing other work...") # This blocks until the task completes @@ -447,7 +447,7 @@ Check if a task is done without blocking: .. code-block:: python - future = compute_value(executor="compute") + future = compute_value(executor=["compute"]) if future.done(): print("Task completed!") @@ -463,7 +463,7 @@ Wait for multiple tasks efficiently: .. code-block:: python # Launch multiple tasks - futures = [compute_value(executor="compute") for _ in range(10)] + futures = [compute_value(executor=["compute"]) for _ in range(10)] # Wait for all to complete results = [f.result() for f in futures] @@ -480,7 +480,7 @@ Exceptions raised in tasks are re-raised when calling ``.result()``: def failing_task(): raise ValueError("Something went wrong!") - future = failing_task(executor="compute") + future = failing_task(executor=["compute"]) try: result = future.result() @@ -539,17 +539,17 @@ transfer and deletion tasks that can be incorporated into your workflows: dst_ep="my-dest-endpoint", src_path="/data/input.dat", dst_path="/scratch/input.dat", - executor="local" + executor=["local"] ) # Process the transferred data (waits for transfer by passing its future) - result = process_file(transfer, "/scratch/input.dat", executor="compute") + result = process_file(transfer, "/scratch/input.dat", executor=["compute"]) # Clean up after processing cleanup = delete_task( src_ep="my-dest-endpoint", src_path="/scratch/input.dat", - executor="local", + executor=["local"], inputs=[result] # Waits for processing to complete ) cleanup.result() @@ -574,8 +574,8 @@ Pass data directly through futures: return sum(data) # Data flows through futures - data_future = generate_data(100, executor="compute") - sum_future = sum_data(data_future, executor="compute") + data_future = generate_data(100, executor=["compute"]) + sum_future = sum_data(data_future, executor=["compute"]) result = sum_future.result() For large data, consider files or data staging strategies. @@ -625,9 +625,9 @@ The most common approach is to pass a future from one task as an argument to ano return f"final_{input_data}" # Chain tasks - data flows through futures - future1 = step1(executor="compute") - future2 = step2(future1, executor="compute") # Waits for future1 - future3 = step3(future2, executor="compute") # Waits for future2 + future1 = step1(executor=["compute"]) + future2 = step2(future1, executor=["compute"]) # Waits for future1 + future3 = step3(future2, executor=["compute"]) # Waits for future2 final_result = future3.result() @@ -649,7 +649,7 @@ parameter. This is automatically supported by all Chiltepin task decorators dst_ep="hpc-scratch", src_path="/data/input.dat", dst_path="/scratch/input.dat", - executor="local" + executor=["local"] ) # Process the data - waits for transfer without passing its result @@ -658,13 +658,13 @@ parameter. This is automatically supported by all Chiltepin task decorators with open(filepath) as f: return len(f.read()) - result = process_data("/scratch/input.dat", executor="compute", inputs=[stage]) + result = process_data("/scratch/input.dat", executor=["compute"], inputs=[stage]) # Clean up - waits for processing to complete cleanup = delete_task( src_ep="hpc-scratch", src_path="/scratch/input.dat", - executor="local", + executor=["local"], inputs=[result] ) @@ -700,12 +700,12 @@ You can combine both approaches and specify multiple dependencies: def combine(data1, data2): return f"{data1}_{data2}" - a = task_a(executor="compute") - b = task_b(executor="compute") - c = task_c(executor="compute") + a = task_a(executor=["compute"]) + b = task_b(executor=["compute"]) + c = task_c(executor=["compute"]) # Combine waits for a and b (via arguments) and c (via inputs) - result = combine(a, b, executor="compute", inputs=[c]) + result = combine(a, b, executor=["compute"], inputs=[c]) .. tip:: **Avoid premature .result() calls**: In this example, notice that ``.result()`` is only @@ -717,17 +717,17 @@ You can combine both approaches and specify multiple dependencies: .. code-block:: python - result1 = step1(executor="compute").result() # Blocks here - result2 = step2(result1, executor="compute").result() # Blocks here - result3 = step3(result2, executor="compute").result() # Blocks here + result1 = step1(executor=["compute"]).result() # Blocks here + result2 = step2(result1, executor=["compute"]).result() # Blocks here + result3 = step3(result2, executor=["compute"]).result() # Blocks here **Good practice** (maximizes parallelism): .. code-block:: python - future1 = step1(executor="compute") - future2 = step2(future1, executor="compute") # Scheduled, doesn't block - future3 = step3(future2, executor="compute") # Scheduled, doesn't block + future1 = step1(executor=["compute"]) + future2 = step2(future1, executor=["compute"]) # Scheduled, doesn't block + future3 = step3(future2, executor=["compute"]) # Scheduled, doesn't block result = future3.result() # Only block when you need the final result Timeout Handling @@ -745,7 +745,7 @@ Handle long-running tasks with timeouts: time.sleep(100) return "done" - future = long_task(executor="compute") + future = long_task(executor=["compute"]) try: result = future.result(timeout=10) # Wait max 10 seconds @@ -812,9 +812,9 @@ Map-Reduce @join_task def map_reduce(items): # Map phase - futures = [map_task(item, executor="compute") for item in items] + futures = [map_task(item, executor=["compute"]) for item in items] # Reduce phase - return reduce_task(futures, executor="compute") + return reduce_task(futures, executor=["compute"]) result = map_reduce([1, 2, 3, 4, 5]).result() # 55 @@ -836,9 +836,9 @@ Pipeline Processing return data ** 2 # Create pipeline - data1 = stage1(5, executor="compute") - data2 = stage2(data1, executor="compute") - result = stage3(data2, executor="compute").result() # ((5*2)+10)^2 = 400 + data1 = stage1(5, executor=["compute"]) + data2 = stage2(data1, executor=["compute"]) + result = stage3(data2, executor=["compute"]).result() # ((5*2)+10)^2 = 400 Parameter Sweep ^^^^^^^^^^^^^^^ @@ -855,7 +855,7 @@ Parameter Sweep futures = [] for p1 in [1, 2, 3]: for p2 in [10, 20, 30]: - future = run_experiment(p1, p2, executor="compute") + future = run_experiment(p1, p2, executor=["compute"]) futures.append(future) # Collect all results diff --git a/docs/testing.rst b/docs/testing.rst index 417731e4..627185fb 100644 --- a/docs/testing.rst +++ b/docs/testing.rst @@ -34,7 +34,7 @@ To run the full test suite: .. code-block:: console - $ pytest --assert=plain --config=tests/configs/.yaml + $ pytest --config=tests/configs/.yaml Where ```` is one of: @@ -53,7 +53,7 @@ For more detailed information during testing: .. code-block:: console - $ pytest -s -vvv --assert=plain --config=tests/configs/.yaml + $ pytest -s -vvv --config=tests/configs/.yaml Running Specific Tests ~~~~~~~~~~~~~~~~~~~~~~ diff --git a/src/chiltepin/__init__.py b/src/chiltepin/__init__.py index 98813136..c725beb8 100644 --- a/src/chiltepin/__init__.py +++ b/src/chiltepin/__init__.py @@ -1 +1,33 @@ # SPDX-License-Identifier: Apache-2.0 + +"""Chiltepin: Federated NWP Workflow Tools. + +This package provides tools for building scientific workflows that can execute +on distributed computing resources using Parsl and Globus services. +""" + +__all__ = [ + "run_workflow", + "run_workflow_from_file", + "run_workflow_from_dict", +] + + +def __getattr__(name): + """Lazy import of workflow functions to avoid loading Parsl unnecessarily. + + This allows users to import other chiltepin submodules (like configure, tasks) + without triggering the import of Parsl and its dependencies unless they + actually use the workflow context managers. + """ + if name in __all__: + from chiltepin.workflow import ( # noqa: F401 + run_workflow, + run_workflow_from_dict, + run_workflow_from_file, + ) + + globals()[name] = locals()[name] + return locals()[name] + + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/src/chiltepin/configure.py b/src/chiltepin/configure.py index 8fcc09d3..d7d98be2 100644 --- a/src/chiltepin/configure.py +++ b/src/chiltepin/configure.py @@ -41,7 +41,8 @@ def parse_file(filename: str) -> Dict[str, Any]: except yaml.YAMLError as e: print("Invalid yaml configuration") raise (e) - return yaml_config + # yaml.safe_load returns None for empty files; return empty dict instead + return yaml_config if yaml_config is not None else {} def create_provider(config: Dict[str, Any]) -> ExecutionProvider: @@ -381,41 +382,58 @@ def load( # Get project root directory for setting PYTHONPATH project_base = Path(__file__).parent.parent.parent.resolve() - # Define a default HTEX Executor with a local provider - # This includes adding project root directory to PYTHONPATH - executors = [ - HighThroughputExecutor( - label="local", - worker_debug=True, - cores_per_worker=1, - max_workers_per_node=1, - provider=LocalProvider( - init_blocks=0, - max_blocks=1, - worker_init=f"export PYTHONPATH=${{PYTHONPATH}}:{project_base}", - ), - ) - ] - - # Add an Executor for each resource + # Determine which resources to load if include is None: resources = config else: - # Validate that all requested resources exist - missing = [key for key in include if key not in config] + # Validate that all requested resources exist (except "local" which is always available) + missing = [key for key in include if key != "local" and key not in config] if missing: raise RuntimeError( f"Resources specified in include list not found in config: {missing}" ) - resources = {key: config[key] for key in include} - for resource_name, resource_config in resources.items(): + resources = {key: config[key] for key in include if key in config} + + # Create executors list + executors = [] + + # Add "local" executor - use user's definition if provided, otherwise use default + # This happens regardless of the include filter + if "local" in config: + # User defined their own "local", use it as the default executors.append( create_executor( - resource_name, - resource_config, + "local", + config["local"], client, ), ) + else: + # Use built-in default "local" + executors.append( + HighThroughputExecutor( + label="local", + worker_debug=True, + cores_per_worker=1, + max_workers_per_node=1, + provider=LocalProvider( + init_blocks=0, + max_blocks=1, + worker_init=f"export PYTHONPATH=${{PYTHONPATH}}:{project_base}", + ), + ) + ) + + # Add an Executor for each resource (skip "local" since we already added it) + for resource_name, resource_config in resources.items(): + if resource_name != "local": + executors.append( + create_executor( + resource_name, + resource_config, + client, + ), + ) config_kwargs = {"executors": executors} if run_dir is not None: diff --git a/src/chiltepin/data.py b/src/chiltepin/data.py index a853f12e..0e07ed4e 100644 --- a/src/chiltepin/data.py +++ b/src/chiltepin/data.py @@ -37,17 +37,17 @@ def process_data(input_path): dst_ep="hpc-scratch", src_path="/data/input.dat", dst_path="/scratch/input.dat", - executor="local" + executor=["local"] ) # Process the staged data (waits for stage via inputs parameter) - result = process_data("/scratch/input.dat", executor="compute", inputs=[stage]) + result = process_data("/scratch/input.dat", executor=["compute"], inputs=[stage]) # Clean up after processing completes (waits for result via inputs) cleanup = delete_task( src_ep="hpc-scratch", src_path="/scratch/input.dat", - executor="local", + executor=["local"], inputs=[result] ) diff --git a/src/chiltepin/tasks.py b/src/chiltepin/tasks.py index af4f9593..d1050db6 100644 --- a/src/chiltepin/tasks.py +++ b/src/chiltepin/tasks.py @@ -24,7 +24,7 @@ def add_numbers(a, b): return a + b # Execute on a specific resource - result = add_numbers(5, 3, executor="compute").result() + result = add_numbers(5, 3, executor=["compute"]).result() Define a bash task:: @@ -35,7 +35,7 @@ def list_files(directory): return f"ls -la {directory}" # Returns exit code (0 = success) - exit_code = list_files("/tmp", executor="compute").result() + exit_code = list_files("/tmp", executor=["compute"]).result() """ from functools import wraps diff --git a/src/chiltepin/workflow.py b/src/chiltepin/workflow.py new file mode 100644 index 00000000..bdc9c8e4 --- /dev/null +++ b/src/chiltepin/workflow.py @@ -0,0 +1,265 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Workflow context managers for Chiltepin. + +This module provides context managers that wrap Parsl configuration and lifecycle +management, eliminating the need for users to directly import or interact with Parsl. +""" + +import logging +import sys +from contextlib import contextmanager +from pathlib import Path +from typing import Any, Dict, List, Optional, Union + +import parsl +from globus_compute_sdk import Client + +from chiltepin import configure + +# Module-level logger for cleanup warnings +_logger = logging.getLogger(__name__) + + +@contextmanager +def run_workflow( + config: Union[str, Path, Dict[str, Any]], + *, + include: Optional[List[str]] = None, + run_dir: Optional[str] = None, + client: Optional[Client] = None, + log_file: Optional[str] = None, + log_level: Optional[int] = None, +): + """Context manager for Chiltepin workflows. + + This wraps Parsl configuration and lifecycle management, eliminating the need + for users to directly import or interact with Parsl. + + Parameters + ---------- + config : str, Path, or dict + Either a path to a YAML configuration file or a configuration dictionary + include : list of str, optional + List of resource labels to load. If None, all resources are loaded. + Note: The default "local" resource is always available, regardless of + this parameter. + run_dir : str, optional + Directory for Parsl runtime files. If None, uses Parsl's default. + client : globus_compute_sdk.Client, optional + Globus Compute client for Globus Compute resources. If None, one will + be created automatically if needed. + log_file : str, optional + Path to Parsl log file. If None, no file logging is configured. + log_level : int, optional + Logging level (e.g., logging.DEBUG). Only used if log_file is provided. + + Yields + ------ + None + The context manager doesn't yield anything. Tasks can be submitted + within the context. + + Examples + -------- + From a configuration file: + + >>> from chiltepin import run_workflow + >>> from chiltepin.tasks import python_task + >>> + >>> @python_task + >>> def my_task(): + ... return "Hello from workflow!" + >>> + >>> with run_workflow("config.yaml"): + ... result = my_task(executor=["compute"]) + ... print(result.result()) + + From a configuration dictionary: + + >>> config = {"laptop": {"provider": "localhost", "cores_per_node": 4}} + >>> with run_workflow(config): + ... result = my_task(executor=["laptop"]) + ... print(result.result()) + + With logging and selective resources: + + >>> import logging + >>> with run_workflow("config.yaml", include=["compute"], + ... log_file="workflow.log", log_level=logging.DEBUG): + ... # "compute" resource from config is available + ... result = my_task(executor=["compute"]) + ... # "local" resource is also always available + ... local_result = my_task(executor=["local"]) + """ + # Parse config if it's a file path + if isinstance(config, (str, Path)): + config_dict = configure.parse_file(str(config)) + else: + config_dict = config + + # Set up logging if requested + logger_handler = None + if log_file is not None: + import logging as log_module + + level = log_level if log_level is not None else log_module.INFO + logger_handler = parsl.set_file_logger(filename=log_file, level=level) + + # Initialize dfk to None before attempting to load + dfk = None + + try: + # Load configuration + parsl_config = configure.load( + config_dict, + include=include, + client=client, + run_dir=run_dir, + ) + + # Load Parsl with the configuration + dfk = parsl.load(parsl_config) + + yield + finally: + # Check if we're cleaning up during exception handling + # If so, don't mask the user's exception with cleanup exceptions + user_exception = sys.exc_info()[0] is not None + cleanup_exception = None + cleanup_tb = None # Preserve original traceback + + # Attempt all cleanup operations, catching exceptions + if dfk is not None: + try: + dfk.cleanup() + except Exception as e: + if user_exception: + _logger.warning( + "Exception during dfk.cleanup() while handling user exception", + exc_info=True, + ) + else: + cleanup_exception = e + cleanup_tb = sys.exc_info()[2] + + # Always call parsl.clear() + try: + parsl.clear() + except Exception as e: + if user_exception: + _logger.warning( + "Exception during parsl.clear() while handling user exception", + exc_info=True, + ) + elif cleanup_exception is None: + cleanup_exception = e + cleanup_tb = sys.exc_info()[2] + else: + e.__cause__ = cleanup_exception + cleanup_exception = e + cleanup_tb = sys.exc_info()[2] + + # Cleanup logger handler + if logger_handler is not None: + try: + logger_handler() + except Exception as e: + if user_exception: + _logger.warning( + "Exception during logger cleanup while handling user exception", + exc_info=True, + ) + elif cleanup_exception is None: + cleanup_exception = e + cleanup_tb = sys.exc_info()[2] + else: + e.__cause__ = cleanup_exception + cleanup_exception = e + cleanup_tb = sys.exc_info()[2] + + # Only raise cleanup exceptions if there wasn't a user exception + # Preserve the original traceback so the error location is clear + if cleanup_exception is not None and not user_exception: + raise cleanup_exception.with_traceback(cleanup_tb) + + +# Convenience aliases for clarity +@contextmanager +def run_workflow_from_file( + config_file: Union[str, Path], + **kwargs, +): + """Context manager for workflows using a YAML configuration file. + + This is an alias for run_workflow() that makes it explicit that a file + path is expected. + + Parameters + ---------- + config_file : str or Path + Path to YAML configuration file + **kwargs + Additional arguments passed to run_workflow() + + Examples + -------- + >>> from chiltepin import run_workflow_from_file + >>> from chiltepin.tasks import python_task + >>> + >>> @python_task + >>> def my_task(): + ... return 42 + >>> + >>> with run_workflow_from_file("my_config.yaml", include=["compute"]): + ... result = my_task(executor=["compute"]) + ... # "local" resource is also always available + ... local_result = my_task(executor=["local"]) + ... print(result.result()) + """ + with run_workflow(config_file, **kwargs): + yield + + +@contextmanager +def run_workflow_from_dict( + config_dict: Dict[str, Any], + **kwargs, +): + """Context manager for workflows using a configuration dictionary. + + This is an alias for run_workflow() that makes it explicit that a dict + is expected. + + Parameters + ---------- + config_dict : dict + Configuration dictionary + **kwargs + Additional arguments passed to run_workflow() + + Examples + -------- + >>> from chiltepin import run_workflow_from_dict + >>> from chiltepin.tasks import python_task + >>> + >>> @python_task + >>> def my_task(): + ... return 42 + >>> + >>> config = {"laptop": {"provider": "localhost"}} + >>> with run_workflow_from_dict(config): + ... result = my_task(executor=["laptop"]) + ... # "local" resource is also always available + ... local_result = my_task(executor=["local"]) + ... print(result.result()) + """ + with run_workflow(config_dict, **kwargs): + yield + + +__all__ = [ + "run_workflow", + "run_workflow_from_file", + "run_workflow_from_dict", +] diff --git a/tests/test_configure.py b/tests/test_configure.py index 4c5bd668..10e96347 100644 --- a/tests/test_configure.py +++ b/tests/test_configure.py @@ -64,6 +64,32 @@ def test_parse_invalid_yaml(self): finally: pathlib.Path(tmp_path).unlink() + def test_parse_empty_yaml(self): + """Test parsing an empty YAML file returns an empty dict.""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: + # Write nothing (empty file) + tmp_path = tmp.name + + try: + result = configure.parse_file(tmp_path) + assert result == {} + assert isinstance(result, dict) + finally: + pathlib.Path(tmp_path).unlink() + + def test_parse_yaml_comments_only(self): + """Test parsing a YAML file with only comments returns an empty dict.""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: + tmp.write("# This is just a comment\n# Another comment\n") + tmp_path = tmp.name + + try: + result = configure.parse_file(tmp_path) + assert result == {} + assert isinstance(result, dict) + finally: + pathlib.Path(tmp_path).unlink() + class TestCreateProvider: """Test create_provider() function for all provider types.""" @@ -612,6 +638,129 @@ def test_load_with_client(self): ] assert len(gc_executors) == 1 + def test_load_overrides_default_local(self): + """Test that user-defined 'local' resource overrides the default.""" + resources = { + "local": { + "provider": "localhost", + "cores_per_worker": 4, + "max_workers_per_node": 8, + } + } + config = configure.load(resources) + + # Should have only 1 executor (user's local, not default + user's) + assert len(config.executors) == 1 + assert config.executors[0].label == "local" + # Verify it's the user's config (4 cores per worker, not default 1) + assert config.executors[0].cores_per_worker == 4 + assert config.executors[0].max_workers_per_node == 8 + + def test_load_local_in_include_overrides_default(self): + """Test that including 'local' in config overrides default even with include filter.""" + resources = { + "local": { + "provider": "localhost", + "cores_per_worker": 2, + }, + "compute": { + "provider": "slurm", + "partition": "compute", + }, + } + config = configure.load(resources, include=["local"]) + + # Should have only user's local (not default) + assert len(config.executors) == 1 + assert config.executors[0].label == "local" + assert config.executors[0].cores_per_worker == 2 + + def test_load_include_without_local_uses_local_override(self): + """Test that user's local is used even when not in include list.""" + resources = { + "local": { + "provider": "localhost", + "cores_per_worker": 4, + }, + "compute": { + "provider": "slurm", + "partition": "compute", + }, + } + # Include only compute, not local + config = configure.load(resources, include=["compute"]) + + # Should have user's local + compute (user's local always used when defined) + assert len(config.executors) == 2 + labels = [ex.label for ex in config.executors] + assert "local" in labels + assert "compute" in labels + + # Verify it's the user's local (4 cores per worker, not default 1) + local_ex = [ex for ex in config.executors if ex.label == "local"][0] + assert local_ex.cores_per_worker == 4 + + def test_load_include_no_user_local_gets_default(self): + """Test that default local is used when user doesn't define one.""" + resources = { + "compute": { + "provider": "slurm", + "partition": "compute", + }, + "mpi": { + "provider": "slurm", + "mpi": True, + }, + } + # Include only compute (no local defined by user) + config = configure.load(resources, include=["compute"]) + + # Should have default local + compute + assert len(config.executors) == 2 + labels = [ex.label for ex in config.executors] + assert "local" in labels + assert "compute" in labels + + # Verify it's the default local (1 core per worker) + local_ex = [ex for ex in config.executors if ex.label == "local"][0] + assert local_ex.cores_per_worker == 1 + + def test_load_duplicate_in_include_is_harmless(self): + """Test that duplicate resource names in include list are harmless.""" + resources = { + "compute": {"provider": "slurm", "partition": "compute"}, + "mpi": {"provider": "slurm", "mpi": True}, + } + + # Duplicates in include list should just be ignored + config = configure.load(resources, include=["compute", "mpi", "compute"]) + + # Should have local + compute + mpi (compute only loaded once despite being in list twice) + assert len(config.executors) == 3 + labels = [ex.label for ex in config.executors] + assert "local" in labels + assert "compute" in labels + assert "mpi" in labels + + def test_load_include_local_without_defining_it(self): + """Test that including 'local' works even if user didn't define it in config.""" + resources = { + "compute": {"provider": "slurm", "partition": "compute"}, + } + + # Including "local" in the list should work even though it's not in config + config = configure.load(resources, include=["local", "compute"]) + + # Should have default local + compute + assert len(config.executors) == 2 + labels = [ex.label for ex in config.executors] + assert "local" in labels + assert "compute" in labels + + # Verify it's the default local (1 core per worker) + local_ex = [ex for ex in config.executors if ex.label == "local"][0] + assert local_ex.cores_per_worker == 1 + class TestIntegrationScenarios: """Integration tests for realistic configuration scenarios.""" diff --git a/tests/test_data.py b/tests/test_data.py index 5aacd1b7..9412812b 100644 --- a/tests/test_data.py +++ b/tests/test_data.py @@ -5,12 +5,11 @@ import uuid from unittest import mock -import parsl import pytest -import chiltepin.configure import chiltepin.data as data import chiltepin.endpoint as endpoint +from chiltepin import run_workflow # Set up fixture to initialize and cleanup Parsl @@ -30,30 +29,18 @@ def config(config_file): clients = endpoint.login() transfer_client = clients["transfer"] - # Set Parsl logging to DEBUG and redirect to a file in the output directory - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_data_parsl.log"), - level=logging.DEBUG, - ) - - # Load the default resource configuration - resources = chiltepin.configure.load( - {}, run_dir=str(output_dir / "test_data_runinfo") - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - # Generate unique destination filename to avoid conflicts in concurrent tests unique_dst = f"1MB.to_ursa.{uuid.uuid4()}" - # Run the tests with the loaded resources - yield {"client": transfer_client, "unique_dst": unique_dst} - - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + # Use workflow context manager with default (empty) resource configuration + with run_workflow( + {}, + run_dir=str(output_dir / "test_data_runinfo"), + log_file=str(output_dir / "test_data_parsl.log"), + log_level=logging.DEBUG, + ): + # Run the tests with the loaded resources + yield {"client": transfer_client, "unique_dst": unique_dst} def test_data_task_basic(config): diff --git a/tests/test_globus_compute_hello.py b/tests/test_globus_compute_hello.py index cc1aed4f..1a9b1cf9 100644 --- a/tests/test_globus_compute_hello.py +++ b/tests/test_globus_compute_hello.py @@ -4,11 +4,11 @@ import os.path import pathlib -import parsl import pytest import chiltepin.configure import chiltepin.endpoint as endpoint +from chiltepin import run_workflow from chiltepin.tasks import bash_task, python_task @@ -58,31 +58,17 @@ def config(config_file): assert len(endpoint_id) == 36 yaml_config["gc-service"]["endpoint"] = f"{endpoint_id}" - # Set Parsl logging to DEBUG and redirect to a file in the output directory - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_globus_compute_hello_parsl.log"), - level=logging.DEBUG, - ) - - # Load the finalized resource configuration - resources = chiltepin.configure.load( + # Use workflow context manager for Parsl lifecycle + with run_workflow( yaml_config, include=["gc-service"], client=compute_client, run_dir=str(output_dir / "test_globus_compute_hello_runinfo"), - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - - # Run the tests with the loaded resources - yield {"output_dir": output_dir} - - # Cleanup Parsl after tests are done - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + log_file=str(output_dir / "test_globus_compute_hello_parsl.log"), + log_level=logging.DEBUG, + ): + # Run the tests with the loaded resources + yield {"output_dir": output_dir} # Stop the test endpoint now that tests are done endpoint.stop("test", config_dir=f"{output_dir}/.globus_compute", timeout=15) diff --git a/tests/test_globus_compute_mpi.py b/tests/test_globus_compute_mpi.py index 3f8033e7..64247a57 100644 --- a/tests/test_globus_compute_mpi.py +++ b/tests/test_globus_compute_mpi.py @@ -6,11 +6,11 @@ import re from datetime import datetime as dt -import parsl import pytest import chiltepin.configure import chiltepin.endpoint as endpoint +from chiltepin import run_workflow from chiltepin.tasks import bash_task @@ -65,31 +65,17 @@ def config(config_file): yaml_config["gc-compute"]["endpoint"] = f"{endpoint_id}" yaml_config["gc-mpi"]["endpoint"] = f"{endpoint_id}" - # Set Parsl logging to DEBUG and redirect to a file in the output directory - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_globus_compute_mpi_parsl.log"), - level=logging.DEBUG, - ) - - # Load the finalized resource configuration - resources = chiltepin.configure.load( + # Use workflow context manager for Parsl lifecycle + with run_workflow( yaml_config, include=["gc-compute", "gc-mpi"], client=compute_client, run_dir=str(output_dir / "test_globus_compute_mpi_runinfo"), - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - - # Run the tests with the loaded resources - yield {"output_dir": output_dir} - - # Cleanup Parsl after tests are done - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + log_file=str(output_dir / "test_globus_compute_mpi_parsl.log"), + log_level=logging.DEBUG, + ): + # Run the tests with the loaded resources + yield {"output_dir": output_dir} # Stop the test endpoint now that tests are done endpoint.stop("test", config_dir=f"{output_dir}/.globus_compute", timeout=15) diff --git a/tests/test_parsl_hello.py b/tests/test_parsl_hello.py index 1578db93..79d3bf7c 100755 --- a/tests/test_parsl_hello.py +++ b/tests/test_parsl_hello.py @@ -4,10 +4,10 @@ import os.path import pathlib -import parsl import pytest import chiltepin.configure +from chiltepin import run_workflow from chiltepin.tasks import bash_task, python_task @@ -26,29 +26,16 @@ def config(config_file): f"export PYTHONPATH=${{PYTHONPATH}}:{pwd.parent.resolve()}" ) - # Set Parsl logging to DEBUG and redirect to a file in the output directory - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_parsl_hello_parsl.log"), - level=logging.DEBUG, - ) - - resources = chiltepin.configure.load( + # Use workflow context manager for Parsl lifecycle + with run_workflow( yaml_config, include=["service"], run_dir=str(output_dir / "test_parsl_hello_runinfo"), - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - - # Run the tests with the loaded resources - yield {"output_dir": output_dir} - - # Cleanup Parsl after tests are done - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + log_file=str(output_dir / "test_parsl_hello_parsl.log"), + log_level=logging.DEBUG, + ): + # Run the tests with the loaded resources + yield {"output_dir": output_dir} # Test bash hello world diff --git a/tests/test_parsl_mpi.py b/tests/test_parsl_mpi.py index 486fa4c9..099ec8d5 100644 --- a/tests/test_parsl_mpi.py +++ b/tests/test_parsl_mpi.py @@ -7,10 +7,10 @@ import re from datetime import datetime as dt -import parsl import pytest import chiltepin.configure +from chiltepin import run_workflow from chiltepin.tasks import bash_task @@ -33,29 +33,16 @@ def config(config_file): f"export PYTHONPATH=${{PYTHONPATH}}:{pwd.parent.resolve()}" ) - # Set Parsl logging to DEBUG and redirect to a file in the output directory - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_parsl_mpi_parsl.log"), - level=logging.DEBUG, - ) - - resources = chiltepin.configure.load( + # Use workflow context manager for Parsl lifecycle + with run_workflow( yaml_config, include=["compute", "mpi"], run_dir=str(output_dir / "test_parsl_mpi_runinfo"), - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - - # Run the tests with the loaded resources - yield {"output_dir": output_dir} - - # Cleanup Parsl after tests are done - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + log_file=str(output_dir / "test_parsl_mpi_parsl.log"), + log_level=logging.DEBUG, + ): + # Run the tests with the loaded resources + yield {"output_dir": output_dir} def test_parsl_hello_mpi(config): diff --git a/tests/test_tasks.py b/tests/test_tasks.py index 66a210c7..2ec39faa 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -14,7 +14,7 @@ import parsl import pytest -import chiltepin.configure +from chiltepin import run_workflow from chiltepin.tasks import bash_task, join_task, python_task @@ -28,12 +28,6 @@ def parsl_config(): output_dir = pwd / "test_output" output_dir.mkdir(exist_ok=True) - # Set Parsl logging to DEBUG and redirect to a file - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_tasks_parsl.log"), - level=logging.DEBUG, - ) - # Configure local executor with PYTHONPATH that includes the project root # This allows Parsl workers to import test modules project_root = pwd.parent.resolve() @@ -46,22 +40,14 @@ def parsl_config(): } } - # Load configuration with the environment setup - resources = chiltepin.configure.load( - local_config, run_dir=str(output_dir / "test_tasks_runinfo") - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - - # Run the tests - yield - - # Cleanup - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + # Use workflow context manager for Parsl lifecycle + with run_workflow( + local_config, + run_dir=str(output_dir / "test_tasks_runinfo"), + log_file=str(output_dir / "test_tasks_parsl.log"), + log_level=logging.DEBUG, + ): + yield # ===== Python Task Tests - Standalone Functions ===== @@ -603,7 +589,7 @@ def task2(): return 2 f1 = task1(executor=["test-local"]) - f2 = task2(executor="all") + f2 = task2(executor="all") # "all" is an alias for all executors assert f1.result() == 1 assert f2.result() == 2 diff --git a/tests/test_workflow.py b/tests/test_workflow.py new file mode 100644 index 00000000..b6e8577a --- /dev/null +++ b/tests/test_workflow.py @@ -0,0 +1,822 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for chiltepin.workflow module. + +This test suite validates that workflow context managers work correctly +for managing Parsl lifecycle without requiring users to directly interact +with Parsl. +""" + +import pathlib +import warnings +from unittest import mock + +import parsl +import parsl.errors +import pytest +import yaml + +import chiltepin.configure +from chiltepin import run_workflow, run_workflow_from_dict, run_workflow_from_file +from chiltepin.tasks import python_task + + +# Helper function to add PYTHONPATH to config +def add_pythonpath_to_config(config_dict, resource_name): + """Add project root to PYTHONPATH for a specific resource.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + if resource_name in config_dict: + env = config_dict[resource_name].get("environment", []) + # Make a copy to avoid modifying shared references + env = env.copy() if env else [] + env.append(f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}") + config_dict[resource_name]["environment"] = env + return config_dict + + +def _cleanup_parsl_state(phase, ignore_already_cleaned=False): + """Helper function to cleanup Parsl DFK state. + + Args: + phase: Description of when cleanup is happening (e.g., "pre-test", "post-test") + ignore_already_cleaned: If True, suppress warnings about already cleaned DFK + """ + try: + dfk = parsl.dfk() + if dfk: + try: + dfk.cleanup() + except Exception as e: + # Don't warn about double-cleanup attempts if requested + if ignore_already_cleaned and "already been cleaned-up" in str(e): + return + warnings.warn( + f"DFK cleanup failed in {phase} fixture cleanup: {e}", + RuntimeWarning, + stacklevel=3, + ) + try: + parsl.clear() + except Exception as e: + warnings.warn( + f"parsl.clear() failed in {phase} fixture cleanup: {e}", + RuntimeWarning, + stacklevel=3, + ) + except parsl.errors.NoDataFlowKernelError: + # Expected when no DFK loaded + pass + except Exception as e: + # Unexpected errors + warnings.warn( + f"Unexpected error in {phase} cleanup: {e}", + RuntimeWarning, + stacklevel=3, + ) + + +# Cleanup any existing Parsl state before tests +@pytest.fixture(scope="function", autouse=True) +def cleanup_parsl(): + """Ensure Parsl is cleaned up before and after each test.""" + # Cleanup before test + _cleanup_parsl_state("pre-test") + + yield + + # Cleanup after test + _cleanup_parsl_state("post-test", ignore_already_cleaned=True) + + +class TestWorkflowContextManager: + """Test run_workflow() context manager with different configurations.""" + + def test_workflow_with_dict_config(self, tmp_path): + """Test workflow context manager with a dictionary config.""" + + @python_task + def add_numbers(a, b): + """Simple task for testing.""" + return a + b + + @python_task + def multiply(x, y): + """Another simple task for testing.""" + return x * y + + # Get project root for PYTHONPATH + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-executor": { + "provider": "localhost", + "cores_per_node": 2, + "max_workers_per_node": 2, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + with run_workflow(config, run_dir=str(tmp_path / "runinfo1")): + # Submit tasks + future1 = add_numbers(10, 32, executor=["test-executor"]) + future2 = multiply(6, 7, executor=["test-executor"]) + + # Get results + result1 = future1.result() + result2 = future2.result() + + assert result1 == 42 + assert result2 == 42 + + def test_workflow_with_file_config(self, config_file, tmp_path): + """Test workflow context manager with a YAML config file.""" + + @python_task + def greet(name): + """Task that returns a greeting.""" + return f"Hello, {name}!" + + # Load config from file and add PYTHONPATH dynamically + config_dict = chiltepin.configure.parse_file(config_file) + add_pythonpath_to_config(config_dict, "service") + + # Write modified config to temp file + temp_config_file = tmp_path / "temp_config.yaml" + with open(temp_config_file, "w") as f: + yaml.dump(config_dict, f) + + # Test with file path argument + with run_workflow( + str(temp_config_file), + include=["service"], + run_dir=str(tmp_path / "runinfo2"), + ): + future = greet("Workflow", executor=["service"]) + result = future.result() + assert result == "Hello, Workflow!" + + +class TestWorkflowAliases: + """Test workflow_from_dict and workflow_from_file convenience aliases.""" + + def test_workflow_from_dict(self, tmp_path): + """Test workflow_from_dict convenience function.""" + + @python_task + def add_numbers(a, b): + return a + b + + # Get project root for PYTHONPATH + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "my-executor": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + with run_workflow_from_dict(config, run_dir=str(tmp_path / "runinfo5")): + future = add_numbers(100, 200, executor=["my-executor"]) + result = future.result() + assert result == 300 + + def test_workflow_from_file(self, config_file, tmp_path): + """Test workflow_from_file convenience function.""" + + @python_task + def multiply(x, y): + return x * y + + # Load config from file and add PYTHONPATH dynamically + config_dict = chiltepin.configure.parse_file(config_file) + add_pythonpath_to_config(config_dict, "service") + + # Write modified config to temp file + temp_config_file = tmp_path / "temp_config.yaml" + with open(temp_config_file, "w") as f: + yaml.dump(config_dict, f) + + # Test with file path argument using workflow_from_file + with run_workflow_from_file( + str(temp_config_file), + include=["service"], + run_dir=str(tmp_path / "runinfo6"), + ): + future = multiply(21, 2, executor=["service"]) + result = future.result() + assert result == 42 + + +class TestWorkflowCleanup: + """Test that workflow context managers properly cleanup resources.""" + + def test_sequential_workflows(self, tmp_path): + """Test that multiple workflows can be created sequentially.""" + + @python_task + def add_numbers(a, b): + return a + b + + @python_task + def multiply(x, y): + return x * y + + # Get project root for PYTHONPATH + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config1 = { + "exec1": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + config2 = { + "exec2": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # First workflow + with run_workflow(config1, run_dir=str(tmp_path / "runinfo7")): + future = add_numbers(1, 2, executor=["exec1"]) + assert future.result() == 3 + + # Second workflow should work without conflicts + with run_workflow(config2, run_dir=str(tmp_path / "runinfo8")): + future = multiply(3, 4, executor=["exec2"]) + assert future.result() == 12 + + def test_workflow_cleanup_on_exception(self, tmp_path): + """Test that workflow cleans up even if an exception occurs.""" + + @python_task + def add_numbers(a, b): + return a + b + + # Get project root for PYTHONPATH + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Workflow should cleanup even if exception occurs + with pytest.raises(ValueError): + with run_workflow(config, run_dir=str(tmp_path / "runinfo9")): + future = add_numbers(5, 5, executor=["test-exec"]) + result = future.result() + assert result == 10 + raise ValueError("Intentional test error") + + # Should be able to create another workflow after exception + with run_workflow(config, run_dir=str(tmp_path / "runinfo10")): + future = add_numbers(7, 8, executor=["test-exec"]) + assert future.result() == 15 + + +class TestWorkflowExceptionHandling: + """Test exception handling during workflow cleanup.""" + + def test_dfk_cleanup_exception(self, tmp_path): + """Test that exceptions during dfk.cleanup() are properly raised.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + + with pytest.raises(RuntimeError, match="Cleanup failed"): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_exc1")): + dfk_ref = parsl.dfk() # Capture dfk reference + pass + + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + + def test_parsl_clear_exception(self, tmp_path): + """Test that exceptions during parsl.clear() are properly raised.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + with mock.patch("parsl.clear") as mock_clear: + mock_clear.side_effect = RuntimeError("Clear failed") + + with pytest.raises(RuntimeError, match="Clear failed"): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_exc2")): + pass + + def test_logger_handler_exception(self, tmp_path): + """Test that exceptions during logger_handler() are properly raised. + + Note: Logger cleanup happens both inside dfk.cleanup() and in our explicit + logger_handler() call. When it fails during dfk.cleanup(), that exception + is what gets raised. + """ + # Ensure clean state before test + try: + parsl.clear() + except Exception: + pass + + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Mock the logger handler to raise an exception + def mock_set_file_logger(*args, **kwargs): + def failing_handler(): + raise RuntimeError("Logger cleanup failed") + + return failing_handler + + with mock.patch("parsl.set_file_logger", side_effect=mock_set_file_logger): + # The logger cleanup exception will be raised (either from dfk.cleanup or our call) + with pytest.raises(RuntimeError): + with run_workflow( + config, + run_dir=str(tmp_path / "runinfo_exc3"), + log_file=str(tmp_path / "test.log"), + ): + pass + + def test_chained_exceptions_cleanup_then_clear(self, tmp_path): + """Test exception chaining when dfk.cleanup() and parsl.clear() both fail.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + with mock.patch("parsl.clear") as mock_clear: + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + mock_clear.side_effect = RuntimeError("Clear failed") + + with pytest.raises(RuntimeError) as exc_info: + with run_workflow(config, run_dir=str(tmp_path / "runinfo_exc4")): + dfk_ref = parsl.dfk() # Capture dfk reference + pass + + # The last exception (clear) should be raised + assert "Clear failed" in str(exc_info.value) + # And the previous exception (cleanup) should be in the chain + assert exc_info.value.__cause__ is not None + assert "Cleanup failed" in str(exc_info.value.__cause__) + + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + + def test_chained_exceptions_all_three(self, tmp_path): + """Test exception chaining when all three cleanup operations fail.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + + # Mock the logger handler to raise an exception + def mock_set_file_logger(*args, **kwargs): + def failing_handler(): + raise RuntimeError("Logger cleanup failed") + + return failing_handler + + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + with mock.patch("parsl.clear") as mock_clear: + with mock.patch( + "parsl.set_file_logger", side_effect=mock_set_file_logger + ): + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + mock_clear.side_effect = RuntimeError("Clear failed") + + with pytest.raises(RuntimeError) as exc_info: + with run_workflow( + config, + run_dir=str(tmp_path / "runinfo_exc5"), + log_file=str(tmp_path / "test.log"), + ): + dfk_ref = parsl.dfk() # Capture dfk reference + pass + + # The last exception (logger) should be raised + assert "Logger cleanup failed" in str(exc_info.value) + # Check the exception chain + assert exc_info.value.__cause__ is not None + assert "Clear failed" in str(exc_info.value.__cause__) + assert exc_info.value.__cause__.__cause__ is not None + assert "Cleanup failed" in str(exc_info.value.__cause__.__cause__) + + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + + def test_parsl_clear_called_when_dfk_is_none(self, tmp_path): + """Test that parsl.clear() is called even when dfk is None.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Mock parsl.load to fail, so dfk stays None + with mock.patch("parsl.load") as mock_load: + with mock.patch("parsl.clear") as mock_clear: + mock_load.side_effect = RuntimeError("Load failed") + + with pytest.raises(RuntimeError, match="Load failed"): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_exc6")): + pass + + # parsl.clear() should still be called even though dfk is None + mock_clear.assert_called_once() + + def test_parsl_clear_exception_without_cleanup_exception(self, tmp_path): + """Test parsl.clear() exception when dfk.cleanup() succeeds.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # This tests the branch where parsl.clear() fails but dfk.cleanup() succeeds + # Explicitly ensure dfk.cleanup() succeeds so cleanup_exception stays None + original_clear = parsl.clear + dfk_ref = None + + def mock_cleanup_success(self): + """Mock dfk.cleanup() to succeed without doing anything.""" + # Don't actually clean up - we'll do it manually after + + call_count = [0] + + def mock_clear_selective(): + call_count[0] += 1 + if call_count[0] == 2: + # Second call (from workflow) - make it fail + raise RuntimeError("Clear failed") + else: + # First and subsequent calls should succeed + try: + original_clear() + except Exception: + pass + + with mock.patch.object(parsl.DataFlowKernel, "cleanup", mock_cleanup_success): + with mock.patch("parsl.clear", side_effect=mock_clear_selective): + # Explicitly make the first call to parsl.clear() to "use up" count==1 + try: + parsl.clear() + except Exception: + pass + + # Now the workflow's call will be count==2 and will raise + with pytest.raises(RuntimeError, match="Clear failed"): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_exc7")): + dfk_ref = parsl.dfk() # Capture for manual cleanup + pass + + # Manually clean up since we mocked cleanup + if dfk_ref: + try: + parsl.DataFlowKernel.cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + + +@pytest.fixture +def config_file_fixture(tmp_path): + """Create a temporary config file for testing file-based workflows.""" + import yaml + + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "service": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + config_file = tmp_path / "test_config.yaml" + with open(config_file, "w") as f: + yaml.dump(config, f) + + return str(config_file) + + +class TestWorkflowFromFileCoverage: + """Tests specifically for workflow_from_file function coverage.""" + + def test_workflow_from_file_with_local_fixture(self, config_file_fixture, tmp_path): + """Test workflow_from_file with a local config file fixture.""" + + @python_task + def simple_task(): + return "success" + + with run_workflow_from_file( + config_file_fixture, + include=["service"], + run_dir=str(tmp_path / "runinfo_file"), + ): + future = simple_task(executor=["service"]) + result = future.result() + assert result == "success" + + +class TestUserExceptionPrecedence: + """Test that user exceptions are not masked by cleanup exceptions.""" + + def test_user_exception_not_masked_by_cleanup_exception(self, tmp_path, caplog): + """Test that user exceptions take precedence over cleanup exceptions.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + + # Mock dfk.cleanup to fail + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + + # User exception should be raised, not cleanup exception + with pytest.raises(ValueError, match="User error"): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_user1")): + dfk_ref = parsl.dfk() # Capture dfk reference + raise ValueError("User error") + + # Cleanup exception should be logged as a warning + assert any( + "Exception during dfk.cleanup() while handling user exception" + in record.getMessage() + for record in caplog.records + ) + + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + + def test_all_cleanup_operations_attempted(self, tmp_path, caplog): + """Test that all cleanup operations are attempted even when some fail.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + + # Mock dfk.cleanup and parsl.clear to fail + original_clear = parsl.clear + call_count = [0] + + def mock_clear_selective(): + call_count[0] += 1 + # Fail on first call (our explicit call after dfk.cleanup fails) + if call_count[0] == 1: + raise RuntimeError("Clear failed") + else: + # Subsequent calls use original + original_clear() + + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + with mock.patch("parsl.clear", side_effect=mock_clear_selective): + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + + # User exception should be raised, not any cleanup exception + with pytest.raises(ValueError, match="User error"): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_user2")): + dfk_ref = parsl.dfk() # Capture dfk reference + raise ValueError("User error") + + # Both cleanup exceptions should be logged as warnings + assert any( + "Exception during dfk.cleanup() while handling user exception" + in record.getMessage() + for record in caplog.records + ) + assert any( + "Exception during parsl.clear() while handling user exception" + in record.getMessage() + for record in caplog.records + ) + + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + + def test_cleanup_exception_raised_when_no_user_exception(self, tmp_path): + """Test that cleanup exceptions are raised when there's no user exception.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + + # Mock dfk.cleanup to fail + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + + # Cleanup exception should be raised when there's no user exception + with pytest.raises(RuntimeError, match="Cleanup failed"): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_user3")): + dfk_ref = parsl.dfk() # Capture dfk reference + pass # No user exception + + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + + def test_user_exception_with_logger_cleanup_failure(self, tmp_path, caplog): + """Test that user exceptions take precedence when logger cleanup fails.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Mock the logger handler to raise an exception + def mock_set_file_logger(*args, **kwargs): + def failing_handler(): + raise RuntimeError("Logger cleanup failed") + + return failing_handler + + with mock.patch("parsl.set_file_logger", side_effect=mock_set_file_logger): + # User exception should be raised, not logger cleanup exception + with pytest.raises(ValueError, match="User error"): + with run_workflow( + config, + run_dir=str(tmp_path / "runinfo_user4"), + log_file=str(tmp_path / "test.log"), + ): + raise ValueError("User error") + + # Logger cleanup exception should be logged as a warning + assert any( + "Exception during logger cleanup while handling user exception" + in record.getMessage() + for record in caplog.records + ) + + def test_logger_handler_exception_standalone(self, tmp_path): + """Test logger cleanup failure when dfk.cleanup and parsl.clear succeed.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Create a handler that succeeds first time (dfk cleanup) but fails second time (our explicit call) + call_count = [0] + + def mock_set_file_logger(*args, **kwargs): + def conditional_failing_handler(): + call_count[0] += 1 + if call_count[0] == 2: # Fail on second call (our explicit cleanup) + raise RuntimeError("Logger cleanup failed") + + return conditional_failing_handler + + with mock.patch("parsl.set_file_logger", side_effect=mock_set_file_logger): + # Logger cleanup exception should be raised + with pytest.raises(RuntimeError, match="Logger cleanup failed"): + with run_workflow( + config, + run_dir=str(tmp_path / "runinfo_logger_standalone"), + log_file=str(tmp_path / "test.log"), + ): + pass