Commit 0045e8bf authored by Pierre Augier's avatar Pierre Augier
Browse files

Debug MPI. Tests pass!

--HG--
branch : dev
parent 630b5cc5db24
......@@ -8,12 +8,16 @@ black:
tests:
pytest fluidpythran data_tests/ipynb
tests_mpi:
mpirun -np 2 pytest fluidpythran
tests_nbval:
pytest --nbval data_tests/ipynb
tests_coverage:
mkdir -p .coverage
coverage run -p -m pytest
mpirun -np 2 coverage run -p -m pytest fluidpythran
report_coverage:
coverage combine
......
......@@ -25,8 +25,8 @@ Internal API
import inspect
import time
from pathlib import Path
import subprocess
import os
from .util import (
get_module_name,
......@@ -44,6 +44,12 @@ from .annotation import (
)
from .log import logger
from . import mpi
from .mpi import Path
if mpi.nb_proc == 1:
mpi.has_to_build = has_to_build
mpi.modification_date = modification_date
is_transpiling = False
modules = {}
......@@ -78,7 +84,7 @@ def _get_fluidpythran_calling_module(index_frame: int = 2):
!= has_to_pythranize_at_import()
or hasattr(fp, "path_mod")
and fp.path_mod.exists()
and has_to_build(fp.path_pythran, fp.path_mod)
and mpi.has_to_build(fp.path_pythran, fp.path_mod)
):
fp = FluidPythran(frame=frame, reuse=False)
else:
......@@ -222,31 +228,39 @@ class FluidPythran:
path_ext = None
if has_to_pythranize_at_import() and path_mod.exists():
if has_to_build(path_pythran, path_mod):
if mpi.has_to_build(path_pythran, path_mod):
if path_pythran.exists():
time_pythran = modification_date(path_pythran)
time_pythran = mpi.modification_date(path_pythran)
else:
time_pythran = 0
print(f"Running fluidpythran on file {path_mod}... ", end="")
# better to do this in another process because the file is already run...
returncode = subprocess.call(
["fluidpythran", "-np", str(path_mod)]
)
returncode = None
if mpi.rank == 0:
print(f"Running fluidpythran on file {path_mod}... ", end="")
# better to do this in another process because the file is already run...
os.environ["FLUIDPYTHRAN_NO_MPI"] = "1"
returncode = subprocess.call(
["fluidpythran", "-np", str(path_mod)]
)
del os.environ["FLUIDPYTHRAN_NO_MPI"]
returncode = mpi.bcast(returncode)
if returncode != 0:
raise RuntimeError(
"fluidpythran does not manage to produce the Pythran "
f"file for {path_mod}"
)
print("Done!")
if mpi.rank == 0:
print("Done!")
path_ext = path_pythran.with_name(
name_ext_from_path_pythran(path_pythran)
)
time_pythran_after = modification_date(path_pythran)
time_pythran_after = mpi.modification_date(path_pythran)
# We have to touch the files to signal that they are up-to-date
if time_pythran_after == time_pythran:
if time_pythran_after == time_pythran and mpi.rank == 0:
if not has_to_build(path_ext, path_pythran):
path_pythran.touch()
if path_ext.exists():
......@@ -265,7 +279,8 @@ class FluidPythran:
and path_mod.exists()
and not self.path_extension.exists()
):
print("Launching Pythran to compile a new extension...")
if mpi.rank == 0:
print("Launching Pythran to compile a new extension...")
self.process = compile_pythran_file(
path_pythran, name_ext_file=self.path_extension.name
)
......
from pathlib import Path
import numpy as np
# pythran import numpy as np
from fluidpythran import cachedjit, used_by_cachedjit
from .mpi import Path
@used_by_cachedjit("func1")
......@@ -18,7 +18,7 @@ def func():
@cachedjit
def func1(a: "int[][] or float[]", l: list):
def func1(a: "int[][] or float[]", l: "int list"):
tmp = np.exp(sum(l))
result = tmp * a * func0(a) + func()
return result
......
......@@ -87,10 +87,8 @@ from .log import logger
modules = {}
path_cachedjit = path_root / "__cachedjit__"
if mpi.rank == 0:
path_cachedjit.mkdir(parents=True, exist_ok=True)
mpi.barrier()
path_cachedjit = mpi.Path(path_root) / "__cachedjit__"
path_cachedjit.mkdir(parents=True, exist_ok=True)
_COMPILE_CACHEDJIT = strtobool(os.environ.get("FLUID_COMPILE_CACHEDJIT", "True"))
......@@ -239,8 +237,6 @@ class CachedJIT:
def __call__(self, func):
# FIXME: MPI?
if not pythran:
return func
......@@ -256,8 +252,7 @@ class CachedJIT:
module_name = mod.module_name
path_pythran = path_cachedjit / module_name.replace(".", os.path.sep)
if mpi.rank == 0:
path_pythran.mkdir(parents=True, exist_ok=True)
path_pythran.mkdir(parents=True, exist_ok=True)
path_pythran = (path_pythran / func_name).with_suffix(".py")
path_pythran_header = path_pythran.with_suffix(".pythran")
......@@ -287,7 +282,7 @@ class CachedJIT:
for function in functions:
src += "\n" + get_source_without_decorator(function)
if path_pythran.exists():
if path_pythran.exists() and mpi.rank == 0:
with open(path_pythran) as file:
src_old = file.read()
if src_old == src:
......@@ -334,7 +329,6 @@ class CachedJIT:
if not exports:
return
mpi.barrier()
if path_pythran_header.exists():
# get the old signature(s)
......@@ -349,8 +343,9 @@ class CachedJIT:
# FIXME: what do we do with the old signatures?
exports.update(exports_old)
header = "\n".join(exports) + "\n"
header = "\n".join(sorted(exports)) + "\n"
mpi.barrier()
if mpi.rank == 0:
logger.info(
f"write Pythran signature in file {path_pythran_header} with types\n{arg_types}"
......@@ -361,8 +356,9 @@ class CachedJIT:
# compute the new path of the extension
hex_header = make_hex(header)
# to be sure (possible bug...)
hex_header = mpi.bcast(hex_header)
# if mpi.nb_proc > 1:
# hex_header0 = mpi.bcast(hex_header)
# assert hex_header0 == hex_header
name_ext_file = (
func_name + "_" + hex_src + "_" + hex_header + ext_suffix
)
......
......@@ -4,6 +4,7 @@
"""
import os
from pathlib import Path
if "FLUIDPYTHRAN_NO_MPI" in os.environ:
nb_proc = 1
......@@ -57,3 +58,47 @@ class ShellProcessMPI:
def is_alive_root(self):
return self.process.poll() is None
PathSeq = Path
if nb_proc > 1:
class PathMPI(type(Path())):
def exists(self):
respons = None
if rank == 0:
respons = super().exists()
return comm.bcast(respons)
def unlink(self):
if rank == 0:
super().unlink()
comm.barrier()
def mkdir(self, *args, **kwargs):
if rank == 0:
super().mkdir(*args, **kwargs)
comm.barrier()
Path = PathMPI
def has_to_build(output_file: Path, input_file: Path):
from . import util
ret = None
if rank == 0:
ret = util.has_to_build(output_file, input_file)
return comm.bcast(ret)
def modification_date(filename):
from . import util
ret = None
if rank == 0:
ret = util.modification_date(filename)
return comm.bcast(ret)
if __name__ == "__main__":
p = Path.home() / "Dev"
print(p.exists(), type(p))
p = PathSeq(p)
print(p, type(p))
......@@ -26,14 +26,14 @@ Internal API
import multiprocessing
import subprocess
import time
from pathlib import Path
from typing import Union, Iterable, Optional
import sysconfig
import hashlib
import os
from .compat import open, implementation
from . import mpi
from .mpi import Path, PathSeq
ext_suffix = sysconfig.get_config_var("EXT_SUFFIX") or ".so"
......@@ -43,6 +43,23 @@ def make_hex(src):
return hashlib.md5(src.encode("utf8")).hexdigest()
def name_ext_from_path_pythran(path_pythran):
"""Return an extension name given the path of a Pythran file"""
name = None
if mpi.rank == 0:
path_pythran = PathSeq(path_pythran)
if path_pythran.exists():
with open(path_pythran) as file:
src = file.read()
else:
src = ""
name = path_pythran.stem + "_" + make_hex(src) + ext_suffix
return mpi.bcast(name)
class SchedulerPopen:
"""Limit the number of Pythran compilations performed in parallel
......@@ -154,22 +171,6 @@ def wait_for_all_extensions():
scheduler.wait_for_all_extensions()
def name_ext_from_path_pythran(path_pythran):
"""Return an extension name given the path of a Pythran file"""
name = None
if mpi.rank == 0:
if path_pythran.exists():
with open(path_pythran) as file:
src = file.read()
else:
src = ""
name = path_pythran.stem + "_" + make_hex(src) + ext_suffix
return mpi.bcast(name)
def compile_pythran_files(
paths: Iterable[Path], str_pythran_flags: str, parallel=True
):
......
......@@ -144,37 +144,3 @@ def parse_args():
)
return parser.parse_args()
def pythran_fluid():
"""Minimal layer above the Pythran commandline"""
if not pythran:
logger.error("Pythran does not seem to be importable")
assert sys.argv[0].endswith("_pythran-fluid")
args = sys.argv[1:]
path = Path.cwd() / args[0]
logger.info(f"Pythranize {path}")
if "-o" in args:
index_output = args.index("-o") + 1
name_out = args[index_output]
name_out_base = name_out.split(".", 1)[0]
name_tmp = name_out_base + ".tmp"
args[index_output] = name_tmp
args.insert(0, "pythran")
subprocess.call(args)
path_tmp = Path(name_tmp)
if path_tmp.exists():
path_out = path_tmp.with_suffix(ext_suffix)
path_tmp.rename(path_out)
logger.info(f"file {Path.cwd() / path_out.name} created")
else:
logger.error(
f"file {Path.cwd() / path_tmp.name} has not been created by Pythran"
)
from pathlib import Path
import importlib
from . import FluidPythran
from . import FluidPythran, mpi
from .compat import rmtree
from .mpi import Path
def test_not_fluidpythranized():
......@@ -13,8 +13,9 @@ def test_not_fluidpythranized():
path_output = path_for_test.parent / "__pythran__"
if path_output.exists():
if path_output.exists() and mpi.rank == 0:
rmtree(path_output)
mpi.barrier()
from . import for_test_init
......
from pathlib import Path
import importlib
import unittest
import os
......@@ -17,6 +17,7 @@ from .util import (
name_ext_from_path_pythran,
)
from .aheadoftime import modules
from .mpi import Path
module_name = "fluidpythran.for_test_init"
......
......@@ -10,12 +10,15 @@ except ImportError:
from .justintime import path_cachedjit, modules
from .pythranizer import scheduler, wait_for_all_extensions
from . import mpi
scheduler.nb_cpus = 2
module_name = "fluidpythran.for_test_justintime"
path_pythran_dir = path_cachedjit / module_name.replace(".", os.path.sep)
path_pythran_dir = mpi.PathSeq(path_cachedjit) / module_name.replace(
".", os.path.sep
)
def delete_pythran_files(func_name):
......@@ -24,9 +27,12 @@ def delete_pythran_files(func_name):
path_pythran_file.unlink()
delete_pythran_files("func1")
delete_pythran_files("func2")
delete_pythran_files("func_dict")
if mpi.rank == 0:
delete_pythran_files("func1")
delete_pythran_files("func2")
delete_pythran_files("func_dict")
mpi.barrier()
def test_cachedjit():
......
......@@ -2,17 +2,18 @@ import sys
import time
from . import path_data_tests
import pytest
from . import path_data_tests
from .run import run
from . import util
from .compat import rmtree
from .mpi import nb_proc
path_dir_out = path_data_tests / "__pythran__"
@pytest.mark.skipif(nb_proc > 1, reason="No commandline in MPI")
def test_create_pythran_files():
if path_dir_out.exists():
rmtree(path_dir_out)
......@@ -28,6 +29,7 @@ def test_create_pythran_files():
run()
@pytest.mark.skipif(nb_proc > 1, reason="No commandline in MPI")
def test_create_pythran_simple():
sys.argv = "fluidpythran --version".split()
......@@ -37,6 +39,7 @@ def test_create_pythran_simple():
run()
@pytest.mark.skipif(nb_proc > 1, reason="No commandline in MPI")
def test_create_pythran_classic():
util.input = lambda: "y"
......
......@@ -32,7 +32,6 @@ from logging import DEBUG
from token import tok_name
from io import BytesIO
from pathlib import Path
from runpy import run_module, run_path
import sys
from unittest.mock import MagicMock as Mock
......@@ -49,6 +48,7 @@ from .log import logger, set_log_level
from .annotation import compute_pythran_types_from_valued_types
from .util import has_to_build, get_source_without_decorator
from .compat import open
from .mpi import Path
import fluidpythran
......
......@@ -25,8 +25,6 @@ def main():
assert sys.argv[0].endswith("_pythran-fluid")
print("NEW")
args = sys.argv[1:]
path = Path.cwd() / args[0]
......
......@@ -21,7 +21,9 @@ deps =
coverage
pytest
pytest-ipynb
mpi4py
py36-pythran: pythran
whitelist_externals = make
commands =
make tests_coverage
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment