Commit 4abd8445 authored by Pierre Augier's avatar Pierre Augier
Browse files

_pythran-fluid + more MPI (broken in MPI)

--HG--
branch : dev
parent b24a9c434536
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)
------------------
......
__version__ = "0.1.1"
__version__ = "0.1.2"
try:
from pyfiglet import figlet_format
......
......@@ -136,7 +136,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(
......@@ -366,7 +366,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(
......
......@@ -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,13 +82,15 @@ 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.mkdir(parents=True, exist_ok=True)
if mpi.rank == 0:
path_cachedjit.mkdir(parents=True, exist_ok=True)
mpi.barrier()
_COMPILE_CACHEDJIT = strtobool(os.environ.get("FLUID_COMPILE_CACHEDJIT", "True"))
......@@ -154,8 +156,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)
......@@ -268,7 +272,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")
......@@ -289,24 +293,30 @@ class CachedJIT:
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"):
......@@ -334,17 +344,18 @@ class CachedJIT:
header = "\n".join(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()
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)
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,7 +369,7 @@ class CachedJIT:
openmp=self.openmp,
)
glob_name_ext_file = func_name + "_" + hex_src + "_*" + ext_suffix_short
glob_name_ext_file = func_name + "_" + hex_src + "_*" + ext_suffix
ext_files = list(path_pythran.parent.glob(glob_name_ext_file))
if not ext_files:
......@@ -373,7 +384,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(
......
......@@ -33,18 +33,21 @@ def bcast(value, root=0):
return value
class ShellThreadMPI:
def __init__(self, thread, root=0):
class ShellProcessMPI:
def __init__(self, process, root=0):
if rank != root:
assert thread is None
assert process is None
self.thread = thread
self.process = process
self.root = root
def is_alive(self):
if rank == self.root:
is_alive = self.thread.is_alive()
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
......@@ -24,7 +24,6 @@ Internal API
"""
import multiprocessing
import threading
import subprocess
import time
from pathlib import Path
......@@ -37,13 +36,6 @@ from . import mpi
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]
def make_hex(src):
"""Produce a hash from a sting"""
......@@ -60,7 +52,7 @@ class SchedulerPopen:
def __init__(self, parallel=True):
if mpi.rank > 0:
return
self.threads = []
self.processes = []
if parallel:
self.limit_nb_processes = max(1, multiprocessing.cpu_count() // 2)
else:
......@@ -68,48 +60,29 @@ class SchedulerPopen:
def block_until_avail(self, parallel=True):
if parallel:
limit = self.limit_nb_processes
else:
limit = 1
while len(self.threads) >= limit:
time.sleep(self.deltat)
self.threads = [
thread for thread in self.threads if thread.is_alive()
]
# def launch_popen(self, words_command, cwd=None, parallel=True):
# """Launch a program (blocking if too many processes launched)"""
# 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"""
if mpi.rank == 0:
while self.threads:
while self.processes:
time.sleep(self.deltat)
self.threads = [
thread
for thread in self.threads
if thread.is_alive()
self.processes = [
process
for process in self.processes
if process.is_alive_root()
]
mpi.barrier()
......@@ -117,13 +90,15 @@ class SchedulerPopen:
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.split()
flags = str_pythran_flags.strip().split()
else:
flags = []
......@@ -140,32 +115,30 @@ class SchedulerPopen:
if openmp:
update_flags("-fopenmp")
def create_extension():
words_command = ["_pythran-fluid", path.name]
words_command = [".pythran-fluid", path.name]
words_command.extend(flags)
if name_ext_file is None:
name_ext_file = name_ext_from_path_pythran(path)
words_command.extend(("-o", name_ext_file))
cwd = path.parent
if implementation == "PyPy":
cwd = str(cwd)
words_command = [str(word) for word in words_command]
words_command.extend(flags)
# we don't use subprocess.call on purpose
process = subprocess.Popen(words_command, cwd=cwd)
cwd = path.parent
if implementation == "PyPy":
cwd = str(cwd)
words_command = [str(word) for word in words_command]
while process.poll() is None:
time.sleep(0.2)
# FIXME lock file ?
# FIXME lock file...
self.block_until_avail(parallel)
thread = None
process = None
if mpi.rank == 0:
thread = threading.Thread(target=create_extension, daemon=True)
if mpi.nb_proc > 1:
thread = mpi.ShellThreadMPI(thread)
process = subprocess.Popen(words_command, cwd=cwd)
self.threads.append(thread)
process = mpi.ShellProcessMPI(process)
self.processes.append(process)
return process
scheduler = SchedulerPopen()
......@@ -187,7 +160,7 @@ def name_ext_from_path_pythran(path_pythran):
else:
src = ""
name = path_pythran.stem + "_" + make_hex(src) + ext_suffix_short
name = path_pythran.stem + "_" + make_hex(src) + ext_suffix
return mpi.bcast(name)
......@@ -195,15 +168,10 @@ def name_ext_from_path_pythran(path_pythran):
def compile_pythran_files(
paths: Iterable[Path], str_pythran_flags: str, parallel=True
):
pythran_flags = str_pythran_flags.strip().split()
for path in paths:
name_ext = name_ext_from_path_pythran(path)
words_command = ["pythran", path.name, "-o", name_ext]
words_command.extend(pythran_flags)
print("pythranize file", path)
scheduler.launch_popen(
words_command, cwd=str(path.parent), parallel=parallel
scheduler.compile_with_pythran(
path, str_pythran_flags=str_pythran_flags, parallel=parallel
)
......@@ -214,25 +182,10 @@ def compile_pythran_file(
xsimd=True,
openmp=False,
):
if mpi.rank > 0:
return
if not isinstance(path, Path):
path = Path(path)
words_command = ["pythran", "-v", path.name]
if name_ext_file is not None:
words_command.extend(("-o", name_ext_file))
if native:
words_command.append("-march=native")
if xsimd:
words_command.append("-DUSE_XSIMD")
if openmp:
words_command.append("-fopenmp")
# return the process
return scheduler.launch_popen(words_command, cwd=str(path.parent))
return scheduler.compile_with_pythran(
path, name_ext_file, native=native, xsimd=xsimd, openmp=openmp
)
......@@ -8,16 +8,20 @@ Internal API
.. autofunction:: parse_args
.. autofunction:: pythran_fluid
"""
import argparse
from pathlib import Path
from glob import glob
import subprocess
import sys
from . import __version__
from .transpiler import make_pythran_files
from .log import logger, set_log_level
from .pythranizer import compile_pythran_files, ext_suffix_short
from .pythranizer import compile_pythran_files, ext_suffix
from .util import has_to_build, clear_cached_extensions
try:
......@@ -84,7 +88,7 @@ def run():
for path in paths:
path = Path(path)
pythran_path = path.parent / "__pythran__" / ("_" + path.name)
ext_path = pythran_path.with_suffix(ext_suffix_short)
ext_path = pythran_path.with_suffix(ext_suffix)
if has_to_build(ext_path, pythran_path):
pythran_paths.append(pythran_path)
......@@ -140,3 +144,35 @@ 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")
\ No newline at end of file
import shutil
import pytest
from . import dist
from .mpi import nb_proc
from .dist import detect_pythran_extensions, modification_date, make_pythran_files
from . import path_data_tests
......@@ -9,7 +13,7 @@ dist.can_import_pythran = True
shutil.rmtree(path_data_tests / "__pythran__", ignore_errors=True)
@pytest.mark.skipif(nb_proc > 1, reason="No dist in MPI")
def test_detect_pythran_extensions():
names = [
......@@ -30,7 +34,7 @@ def test_detect_pythran_extensions():
ext_names = detect_pythran_extensions(path_data_tests)
assert len(ext_names) == len(names) - 1
@pytest.mark.skipif(nb_proc > 1, reason="No dist in MPI")
def test_modification_date():
modification_date(path_data_tests / "no_pythran_.py")
......@@ -59,7 +59,6 @@ except ImportError:
from .compat import implementation
from .pythranizer import (
ext_suffix,
ext_suffix_short,
name_ext_from_path_pythran,
make_hex,
)
......@@ -177,12 +176,6 @@ def import_from_path(path: Path, module_name: str):
f"[path.name for path in path.parent.glob('*')]:\n{[path.name for path in path.parent.glob('*')]}\n"
)
if implementation == "PyPy" and path.suffix == ext_suffix_short:
new_path = path.with_suffix(ext_suffix)
if has_to_build(new_path, path):
shutil.copy(str(path), str(new_path))
path = new_path
if "." in module_name:
package_name, mod_name = module_name.rsplit(".", 1)
name_file = path.name.split(".", 1)[0]
......@@ -195,7 +188,7 @@ def import_from_path(path: Path, module_name: str):
module = sys.modules[module_name]
if (
module.__file__.endswith(ext_suffix_short)
module.__file__.endswith(ext_suffix)
and Path(module.__file__) == path
):
return module
......
......@@ -31,6 +31,7 @@ setup(
version=__version__,
packages=find_packages(exclude=["doc"]),
entry_points={
"console_scripts": ["fluidpythran = fluidpythran.run:run"]
"console_scripts": ["fluidpythran = fluidpythran.run:run",
"_pythran-fluid = fluidpythran.run:pythran_fluid"]
},
)
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