Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add flux executor support for PsijWorker #710

Open
wants to merge 9 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions .github/workflows/testpsijflux.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
name: PSI/J-Flux

on:
push:
branches:
- master
pull_request:

jobs:
test:
runs-on: ubuntu-latest
permissions:
packages: read

container:
image: fluxrm/flux-sched:focal-v0.28.0
options: "--platform=linux/amd64 --user root -it --init"

steps:
- name: Make Space
run: |
rm -rf /usr/share/dotnet
rm -rf /opt/ghc
- name: Checkout
uses: actions/checkout@v4

- name: Setup Python
run: |
apt-get update && apt-get install -y python3-venv && apt-get install less
export PATH=$PWD/bin:$PATH
ln -s /usr/bin/python3 /usr/bin/python
python -m pip install --upgrade pip && pip install -e ".[test]" && python -c 'import pydra; print(pydra.__version__)'
pip install -e "git+https://github.com/ExaWorks/psij-python.git@main#egg=psij-python"
- name: Run pytest
run: |
export PATH=$PWD/bin:$PATH
flux start python -V
flux start pytest --psij=flux --color=yes -vs --cov pydra --cov-config .coveragerc --cov-report xml:cov.xml --doctest-modules pydra/
2 changes: 1 addition & 1 deletion pydra/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ def pytest_addoption(parser):
"--psij",
action="store",
help="run with psij subtype plugin",
choices=["local", "slurm"],
choices=["local", "slurm", "flux"],
)


Expand Down
12 changes: 6 additions & 6 deletions pydra/engine/workers.py
Original file line number Diff line number Diff line change
Expand Up @@ -913,7 +913,7 @@ def __init__(self, subtype, **kwargs):
self.psij = psij

# Check if the provided subtype is valid
valid_subtypes = ["local", "slurm"]
valid_subtypes = ["local", "slurm", "flux"]
if subtype not in valid_subtypes:
raise ValueError(
f"Invalid 'subtype' provided. Available options: {', '.join(valid_subtypes)}"
Expand Down Expand Up @@ -992,7 +992,7 @@ async def exec_psij(self, runnable, rerun=False):
with open(file_path, "wb") as file:
pickle.dump(runnable._run, file)
func_path = absolute_path / "run_pickled.py"
spec = self.make_spec("python", [func_path, file_path])
spec = self.make_spec("python", [str(func_path), str(file_path)])
else: # it could be tuple that includes pickle files with tasks and inputs
cache_dir = runnable[-1].cache_dir
file_path_1 = cache_dir / "taskmain.pkl"
Expand All @@ -1006,9 +1006,9 @@ async def exec_psij(self, runnable, rerun=False):
spec = self.make_spec(
"python",
[
func_path,
file_path_1,
file_path_2,
str(func_path),
str(file_path_1),
str(file_path_2),
],
)

Expand Down Expand Up @@ -1044,6 +1044,6 @@ def close(self):
"sge": SGEWorker,
**{
"psij-" + subtype: lambda subtype=subtype: PsijWorker(subtype=subtype)
for subtype in ["local", "slurm"]
for subtype in ["local", "slurm", "flux"]
},
}