Commit e21acb25 authored by Pierre Augier's avatar Pierre Augier
Browse files

Merge MPI support

Future
------
- :code:`pythran_def` for simple methods (without assignation to attributes
and call of other methods)
0.1.2 (?)
---------
- Private command line :code:`_pythran-fluid` to call Pythran
- MPI aware (only process rank == 0 doing IO and compilation)
- Fix bug C-style `[][]`
- :code:`Array[float, "2d"]` supported
- :code:`NDim(0)` supported
- Function :code:`set_compile_cachedjit()` to disable compilation of
cachedjit functions
0.1.1 (2018-11-28)
------------------
......
......@@ -3,17 +3,21 @@ develop:
pip install -e .[test]
black:
black -l 82 fluidpythran
black -l 82 fluidpythran fluidpythran_cl
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
......
__version__ = "0.1.1"
__version__ = "0.1.2"
try:
from pyfiglet import figlet_format
......
......@@ -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:
......@@ -136,7 +142,7 @@ class CheckCompiling:
return self.func(*args, **kwargs)
fp = self.fp
if fp.is_compiling and fp.process.poll() is not None:
if fp.is_compiling and not fp.process.is_alive():
fp.is_compiling = False
time.sleep(0.1)
fp.module_pythran = import_from_path(
......@@ -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
)
......@@ -366,7 +381,7 @@ class FluidPythran:
"by `if fp.is_transpiled`"
)
if self.is_compiling and self.process.poll() is not None:
if self.is_compiling and not self.process.is_alive():
self.is_compiling = False
time.sleep(0.1)
self.module_pythran = import_from_path(
......
......@@ -301,9 +301,7 @@ def compute_pythran_types_from_types(types, **kwargs):
if "][" in _type:
# C style: we try to rewrite it in Cython style
base, dims = _type.split("[", 1)
dims = ", ".join(
[_ or ":" for _ in dims[:-1].split("][")]
)
dims = ", ".join([_ or ":" for _ in dims[:-1].split("][")])
_type = base + "[" + dims + "]"
elif _type.endswith("[]"):
_type = _type[:-2] + "[:]"
......
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
......
......@@ -65,7 +65,7 @@ try:
except ImportError:
np = None
from .pythranizer import compile_pythran_file, ext_suffix_short
from .pythranizer import compile_pythran_file, ext_suffix
from .util import (
get_module_name,
......@@ -82,14 +82,14 @@ from .annotation import make_signatures_from_typehinted_func
from .compat import open
from . import mpi
from .log import logger
modules = {}
path_cachedjit = path_root / "__cachedjit__"
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"))
......@@ -154,8 +154,10 @@ def _get_module_cachedjit(index_frame: int = 2):
try:
frame = inspect.stack()[index_frame]
except IndexError:
print("index_frame", index_frame)
print([frame[1] for frame in inspect.stack()])
logger.error(
f"index_frame {index_frame}"
f"{[frame[1] for frame in inspect.stack()]}"
)
raise
module_name = get_module_name(frame)
......@@ -235,8 +237,6 @@ class CachedJIT:
def __call__(self, func):
# FIXME: MPI?
if not pythran:
return func
......@@ -252,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")
......@@ -268,7 +267,7 @@ class CachedJIT:
src = None
if has_to_write and mpi.rank == 0:
if has_to_write:
import_lines = [
line.split("# pythran ")[1]
for line in mod.get_source().split("\n")
......@@ -283,30 +282,36 @@ 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:
has_to_write = False
if has_to_write:
print(f"write code in file {path_pythran}")
if has_to_write and mpi.rank == 0:
logger.info(f"write code in file {path_pythran}")
with open(path_pythran, "w") as file:
file.write(src)
file.flush()
if src is None:
if src is None and mpi.rank == 0:
with open(path_pythran) as file:
src = file.read()
# FIXME MPI all processes have to get these 2 variables
hex_src = None
name_mod = None
if mpi.rank == 0:
# hash from src (to produce the extension name)
hex_src = make_hex(src)
name_mod = ".".join(
path_pythran.absolute()
.relative_to(path_root)
.with_suffix("")
.parts
)
# hash from src (to produce the extension name)
hex_src = make_hex(src)
name_mod = ".".join(
path_pythran.absolute().relative_to(path_root).with_suffix("").parts
)
hex_src = mpi.bcast(hex_src)
name_mod = mpi.bcast(name_mod)
def pythranize_with_new_header(arg_types="no types"):
......@@ -326,25 +331,36 @@ class CachedJIT:
if path_pythran_header.exists():
# get the old signature(s)
with open(path_pythran_header) as file:
exports_old = [export.strip() for export in file.readlines()]
exports_old = None
if mpi.rank == 0:
with open(path_pythran_header) as file:
exports_old = [
export.strip() for export in file.readlines()
]
exports_old = mpi.bcast(exports_old)
# FIXME: what do we do with the old signatures?
exports.update(exports_old)
header = "\n".join(exports) + "\n"
header = "\n".join(sorted(exports)) + "\n"
print(
f"write Pythran signature in file {path_pythran_header} with types\n{arg_types}"
)
with open(path_pythran_header, "w") as file:
file.write(header)
file.flush()
mpi.barrier()
if mpi.rank == 0:
logger.info(
f"write Pythran signature in file {path_pythran_header} with types\n{arg_types}"
)
with open(path_pythran_header, "w") as file:
file.write(header)
file.flush()
# compute the new path of the extension
hex_header = make_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_short
func_name + "_" + hex_src + "_" + hex_header + ext_suffix
)
self.path_extension = path_pythran.with_name(name_ext_file)
......@@ -358,8 +374,11 @@ class CachedJIT:
openmp=self.openmp,
)
glob_name_ext_file = func_name + "_" + hex_src + "_*" + ext_suffix_short
ext_files = list(path_pythran.parent.glob(glob_name_ext_file))
ext_files = None
if mpi.rank == 0:
glob_name_ext_file = func_name + "_" + hex_src + "_*" + ext_suffix
ext_files = list(path_pythran.parent.glob(glob_name_ext_file))
ext_files = mpi.bcast(ext_files)
if not ext_files:
if has_to_pythranize_at_import() and _COMPILE_CACHEDJIT:
......@@ -373,7 +392,7 @@ class CachedJIT:
def type_collector(*args, **kwargs):
if self.compiling:
if self.process.poll() is not None:
if not self.process.is_alive():
self.compiling = False
time.sleep(0.1)
module_pythran = import_from_path(
......
"""Minimalist MPI module
========================
"""
try:
from fluiddyn.util import mpi as _mpi
except ImportError:
import os
from pathlib import Path
if "FLUIDPYTHRAN_NO_MPI" in os.environ:
nb_proc = 1
rank = 0
else:
try:
from mpi4py import MPI
from fluiddyn.util import mpi as _mpi
except ImportError:
nb_proc = 1
rank = 0
try:
from mpi4py import MPI
except ImportError:
nb_proc = 1
rank = 0
else:
comm = MPI.COMM_WORLD
nb_proc = comm.size
rank = comm.Get_rank()
else:
comm = MPI.COMM_WORLD
nb_proc = comm.size
rank = comm.Get_rank()
else:
rank = _mpi.rank
nb_proc = _mpi.nb_proc
if nb_proc > 1:
comm = _mpi.comm
rank = _mpi.rank
nb_proc = _mpi.nb_proc
if nb_proc > 1:
comm = _mpi.comm
def barrier():
if nb_proc > 1:
comm.barrier()
\ No newline at end of file
comm.barrier()
def bcast(value, root=0):
if nb_proc > 1:
value = comm.bcast(value, root=root)
return value
class ShellProcessMPI:
def __init__(self, process, root=0):
if rank != root:
assert process is None
self.process = process
self.root = root
def is_alive(self):
if rank == self.root:
is_alive = self.is_alive_root()
else:
is_alive = None
return bcast(is_alive, self.root)
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,22 +26,16 @@ Internal API
import multiprocessing
import subprocess
import time
from pathlib import Path
from typing import Union, Iterable, Optional
import sysconfig
import hashlib
from .compat import open, implementation
from . import mpi
from .mpi import Path, PathSeq
ext_suffix = sysconfig.get_config_var("EXT_SUFFIX") or ".so"
# if pythran and pythran.__version__ <= "0.9.0":
# avoid a Pythran bug with -o option
# it is bad because then we do not support using many Python versions
ext_suffix_short = "." + ext_suffix.rsplit(".", 1)[-1]
ext_suffix = sysconfig.get_config_var("EXT_SUFFIX") or ".so"
def make_hex(src):
......@@ -49,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
......@@ -65,29 +76,23 @@ class SchedulerPopen:
else:
self.limit_nb_processes = 1
def launch_popen(self, words_command, cwd=None, parallel=True):
"""Launch a program (blocking if too many processes launched)"""
if mpi.rank > 0:
return
def block_until_avail(self, parallel=True):
if parallel:
limit = self.limit_nb_processes
else:
limit = 1
while len(self.processes) >= limit:
time.sleep(self.deltat)
self.processes = [
process for process in self.processes if process.poll() is None
]
if mpi.rank == 0:
if parallel:
limit = self.limit_nb_processes
else:
limit = 1
if implementation == "PyPy":
cwd = str(cwd)
words_command = [str(word) for word in words_command]
while len(self.processes) >= limit:
time.sleep(self.deltat)
self.processes = [
process
for process in self.processes
if process.is_alive_root()
]
process = subprocess.Popen(words_command, cwd=cwd)
self.processes.append(process)
return process
mpi.barrier()
def wait_for_all_extensions(self):
"""Wait until all compilation processes are done"""
......@@ -95,54 +100,84 @@ class SchedulerPopen:
while self.processes:
time.sleep(self.deltat)
self.processes = [
process for process in self.processes if process.poll() is None
process
for process in self.processes
if process.is_alive_root()
]
mpi.barrier()
def compile_with_pythran(
self,
path: Path,
name_ext_file: Optional[str] = None,
native=True,
xsimd=True,
openmp=False,
str_pythran_flags: Optional[str] = None,
parallel=True,
):
if str_pythran_flags is not None:
flags = str_pythran_flags.strip().split()
else:
flags = []
scheduler = SchedulerPopen()