11642291501

Committed 02 Nov 2024 11:16AM UTC coverage: 81.168% (+0.5%) from 80.657%

Build # 11642291501

Build Type

push

github

Committed by

web-flow

Commit Message

Upload to coveralls and docs from CI job running against python 3.12 (#726)

Run Details

3474 of 4280 relevant lines covered (81.17%)

0.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.96

/ffcx/codegeneration/jit.py

# Copyright (C) 2004-2019 Garth N. Wells
#
# This file is part of FFCx.(https://www.fenicsproject.org)
#
# SPDX-License-Identifier:    LGPL-3.0-or-later
"""Just-in-time compilation."""

from __future__ import annotations

import importlib
import io
import logging
import os
import re
import sys
import sysconfig
import tempfile
import time
from contextlib import redirect_stdout
from pathlib import Path

import cffi
import numpy as np
import numpy.typing as npt
import ufl

import ffcx
import ffcx.naming
from ffcx.codegeneration.C.file_template import libraries as _libraries

logger = logging.getLogger("ffcx")
root_logger = logging.getLogger()

# Get declarations directly from ufcx.h
file_dir = os.path.dirname(os.path.abspath(__file__))
with open(file_dir + "/ufcx.h") as f:
    ufcx_h = "".join(f.readlines())

# Emulate C preprocessor on __STDC_NO_COMPLEX__
if sys.platform.startswith("win32"):
    # Remove macro statements and content
    ufcx_h = re.sub(
        r"\#ifndef __STDC_NO_COMPLEX__.*?\#endif // __STDC_NO_COMPLEX__",
        "",
        ufcx_h,
        flags=re.DOTALL,
    )
else:
    # Remove only macros keeping content
    ufcx_h = ufcx_h.replace("#ifndef __STDC_NO_COMPLEX__", "")
    ufcx_h = ufcx_h.replace("#endif // __STDC_NO_COMPLEX__", "")

header = ufcx_h.split("<HEADER_DECL>")[1].split("</HEADER_DECL>")[0].strip(" /\n")
header = header.replace("{", "{{").replace("}", "}}")
UFC_HEADER_DECL = header + "\n"

UFC_FORM_DECL = "\n".join(re.findall("typedef struct ufcx_form.*?ufcx_form;", ufcx_h, re.DOTALL))

UFC_INTEGRAL_DECL = "\n".join(
    re.findall(r"typedef void ?\(ufcx_tabulate_tensor_float32\).*?\);", ufcx_h, re.DOTALL)
)
UFC_INTEGRAL_DECL += "\n".join(
    re.findall(r"typedef void ?\(ufcx_tabulate_tensor_float64\).*?\);", ufcx_h, re.DOTALL)
)
UFC_INTEGRAL_DECL += "\n".join(
    re.findall(r"typedef void ?\(ufcx_tabulate_tensor_complex64\).*?\);", ufcx_h, re.DOTALL)
)
UFC_INTEGRAL_DECL += "\n".join(
    re.findall(r"typedef void ?\(ufcx_tabulate_tensor_complex128\).*?\);", ufcx_h, re.DOTALL)
)

UFC_INTEGRAL_DECL += "\n".join(
    re.findall("typedef struct ufcx_integral.*?ufcx_integral;", ufcx_h, re.DOTALL)
)

UFC_EXPRESSION_DECL = "\n".join(
    re.findall("typedef struct ufcx_expression.*?ufcx_expression;", ufcx_h, re.DOTALL)
)


def _compute_option_signature(options):
    """Return options signature (some options should not affect signature)."""
    return str(sorted(options.items()))


def get_cached_module(module_name, object_names, cache_dir, timeout):
    """Look for an existing C file and wait for compilation, or if it does not exist, create it."""
    cache_dir = Path(cache_dir)
    c_filename = cache_dir.joinpath(module_name).with_suffix(".c")
    ready_name = c_filename.with_suffix(".c.cached")

    # Ensure cache dir exists
    cache_dir.mkdir(exist_ok=True, parents=True)

    try:
        # Create C file with exclusive access
        with open(c_filename, "x"):
            pass
        return None, None
    except FileExistsError:
        logger.info("Cached C file already exists: " + str(c_filename))
        finder = importlib.machinery.FileFinder(
            str(cache_dir),
            (importlib.machinery.ExtensionFileLoader, importlib.machinery.EXTENSION_SUFFIXES),
        )
        finder.invalidate_caches()

        # Now, wait for ready
        for i in range(timeout):
            if os.path.exists(ready_name):
                spec = finder.find_spec(module_name)
                if spec is None:
                    raise ModuleNotFoundError("Unable to find JIT module.")
                compiled_module = importlib.util.module_from_spec(spec)
                spec.loader.exec_module(compiled_module)

                compiled_objects = [getattr(compiled_module.lib, name) for name in object_names]
                return compiled_objects, compiled_module

            logger.info(f"Waiting for {ready_name} to appear.")
            time.sleep(1)
        raise TimeoutError(
            "JIT compilation timed out, probably due to a failed previous compile. "
            f"Try cleaning cache (e.g. remove {c_filename}) or increase timeout option."
        )


def _compilation_signature(cffi_extra_compile_args, cffi_debug):
    """Compute the compilation-inputs part of the signature.

    Used to avoid cache conflicts across Python versions, architectures, installs.

    - SOABI includes platform, Python version, debug flags
    - CFLAGS includes prefixes, arch targets
    """
    if sys.platform.startswith("win32"):
        # NOTE: SOABI not defined on win32, EXT_SUFFIX contains e.g. '.cp312-win_amd64.pyd'
        return (
            str(cffi_extra_compile_args)
            + str(cffi_debug)
            + str(sysconfig.get_config_var("EXT_SUFFIX"))
        )
    else:
        return (
            str(cffi_extra_compile_args)
            + str(cffi_debug)
            + str(sysconfig.get_config_var("CFLAGS"))
            + str(sysconfig.get_config_var("SOABI"))
        )


def compile_forms(
    forms: list[ufl.Form],
    options: dict = {},
    cache_dir: Path | None = None,
    timeout: int = 10,
    cffi_extra_compile_args: list[str] = [],
    cffi_verbose: bool = False,
    cffi_debug: bool = False,
    cffi_libraries: list[str] = [],
    visualise: bool = False,
):
    """Compile a list of UFL forms into UFC Python objects.

    Args:
        forms: List of ufl.form to compile.
        options: Options
        cache_dir: Cache directory
        timeout: Timeout
        cffi_extra_compile_args: Extra compilation args for CFFI
        cffi_verbose: Use verbose compile
        cffi_debug: Use compiler debug mode
        cffi_libraries: libraries to use with compiler
        visualise: Toggle visualisation
    """
    p = ffcx.options.get_options(options)

    # Get a signature for these forms
    module_name = "libffcx_forms_" + ffcx.naming.compute_signature(
        forms,
        _compute_option_signature(p) + _compilation_signature(cffi_extra_compile_args, cffi_debug),
    )

    form_names = [ffcx.naming.form_name(form, i, module_name) for i, form in enumerate(forms)]

    if cache_dir is not None:
        cache_dir = Path(cache_dir)
        obj, mod = get_cached_module(module_name, form_names, cache_dir, timeout)
        if obj is not None:
            return obj, mod, (None, None)
    else:
        cache_dir = Path(tempfile.mkdtemp())

    try:
        decl = (
            UFC_HEADER_DECL.format(np.dtype(p["scalar_type"]).name)  # type: ignore
            + UFC_INTEGRAL_DECL
            + UFC_FORM_DECL
        )

        form_template = "extern ufcx_form {name};\n"
        for name in form_names:
            decl += form_template.format(name=name)

        impl = _compile_objects(
            decl,
            forms,
            form_names,
            module_name,
            p,
            cache_dir,
            cffi_extra_compile_args,
            cffi_verbose,
            cffi_debug,
            cffi_libraries,
            visualise=visualise,
        )
    except Exception as e:
        try:
            # remove c file so that it will not timeout next time
            c_filename = cache_dir.joinpath(module_name + ".c")
            os.replace(c_filename, c_filename.with_suffix(".c.failed"))
        except Exception:
            pass
        raise e

    obj, module = _load_objects(cache_dir, module_name, form_names)
    return obj, module, (decl, impl)


def compile_expressions(
    expressions: list[tuple[ufl.Expr, npt.NDArray[np.floating]]],
    options: dict = {},
    cache_dir: Path | None = None,
    timeout: int = 10,
    cffi_extra_compile_args: list[str] = [],
    cffi_verbose: bool = False,
    cffi_debug: bool = False,
    cffi_libraries: list[str] = [],
    visualise: bool = False,
):
    """Compile a list of UFL expressions into UFC Python objects.

    Args:
        expressions: List of (UFL expression, evaluation points).
        options: Options
        cache_dir: Cache directory
        timeout: Timeout
        cffi_extra_compile_args: Extra compilation args for CFFI
        cffi_verbose: Use verbose compile
        cffi_debug: Use compiler debug mode
        cffi_libraries: libraries to use with compiler
        visualise: Toggle visualisation
    """
    p = ffcx.options.get_options(options)

    module_name = "libffcx_expressions_" + ffcx.naming.compute_signature(
        expressions,
        _compute_option_signature(p) + _compilation_signature(cffi_extra_compile_args, cffi_debug),
    )
    expr_names = [
        ffcx.naming.expression_name(expression, module_name) for expression in expressions
    ]

    if cache_dir is not None:
        cache_dir = Path(cache_dir)
        obj, mod = get_cached_module(module_name, expr_names, cache_dir, timeout)
        if obj is not None:
            return obj, mod, (None, None)
    else:
        cache_dir = Path(tempfile.mkdtemp())

    try:
        decl = (
            UFC_HEADER_DECL.format(np.dtype(p["scalar_type"]).name)  # type: ignore
            + UFC_INTEGRAL_DECL
            + UFC_FORM_DECL
            + UFC_EXPRESSION_DECL
        )

        expression_template = "extern ufcx_expression {name};\n"
        for name in expr_names:
            decl += expression_template.format(name=name)

        impl = _compile_objects(
            decl,
            expressions,
            expr_names,
            module_name,
            p,
            cache_dir,
            cffi_extra_compile_args,
            cffi_verbose,
            cffi_debug,
            cffi_libraries,
            visualise=visualise,
        )
    except Exception as e:
        try:
            # remove c file so that it will not timeout next time
            c_filename = cache_dir.joinpath(module_name + ".c")
            os.replace(c_filename, c_filename.with_suffix(".c.failed"))
        except Exception:
            pass
        raise e

    obj, module = _load_objects(cache_dir, module_name, expr_names)
    return obj, module, (decl, impl)


def _compile_objects(
    decl,
    ufl_objects,
    object_names,
    module_name,
    options,
    cache_dir,
    cffi_extra_compile_args,
    cffi_verbose,
    cffi_debug,
    cffi_libraries,
    visualise: bool = False,
):
    import ffcx.compiler

    libraries = _libraries + cffi_libraries if cffi_libraries is not None else _libraries

    # JIT uses module_name as prefix, which is needed to make names of all struct/function
    # unique across modules
    _, code_body = ffcx.compiler.compile_ufl_objects(
        ufl_objects, prefix=module_name, options=options, visualise=visualise
    )

    # Raise error immediately prior to compilation if no support for C99
    # _Complex. Doing this here allows FFCx to be used for complex codegen on
    # Windows.
    if sys.platform.startswith("win32"):
        if np.issubdtype(options["scalar_type"], np.complexfloating):
            raise NotImplementedError("win32 platform does not support C99 _Complex numbers")
        elif isinstance(options["scalar_type"], str) and "complex" in options["scalar_type"]:
            raise NotImplementedError("win32 platform does not support C99 _Complex numbers")

    # Compile in C17 mode
    if sys.platform.startswith("win32"):
        cffi_base_compile_args = ["-std:c17"]
    else:
        cffi_base_compile_args = ["-std=c17"]

    cffi_final_compile_args = cffi_base_compile_args + cffi_extra_compile_args

    ffibuilder = cffi.FFI()

    ffibuilder.set_source(
        module_name,
        code_body,
        include_dirs=[ffcx.codegeneration.get_include_path()],
        extra_compile_args=cffi_final_compile_args,
        libraries=libraries,
    )

    ffibuilder.cdef(decl)

    c_filename = cache_dir.joinpath(module_name + ".c")
    ready_name = c_filename.with_suffix(".c.cached")

    # Compile (ensuring that compile dir exists)
    cache_dir.mkdir(exist_ok=True, parents=True)

    logger.info(79 * "#")
    logger.info("Calling JIT C compiler")
    logger.info(79 * "#")

    t0 = time.time()
    f = io.StringIO()
    # Temporarily set root logger handlers to string buffer only
    # since CFFI logs into root logger
    old_handlers = root_logger.handlers.copy()
    root_logger.handlers = [logging.StreamHandler(f)]
    with redirect_stdout(f):
        ffibuilder.compile(tmpdir=cache_dir, verbose=True, debug=cffi_debug)
    s = f.getvalue()
    if cffi_verbose:
        print(s)

    logger.info(f"JIT C compiler finished in {time.time() - t0:.4f}")

    # Create a "status ready" file. If this fails, it is an error,
    # because it should not exist yet.
    # Copy the stdout verbose output of the build into the ready file
    fd = open(ready_name, "x")
    fd.write(s)
    fd.close()

    # Copy back the original handlers (in case someone is logging into
    # root logger and has custom handlers)
    root_logger.handlers = old_handlers

    return code_body


def _load_objects(cache_dir, module_name, object_names):
    # Create module finder that searches the compile path
    finder = importlib.machinery.FileFinder(
        str(cache_dir),
        (importlib.machinery.ExtensionFileLoader, importlib.machinery.EXTENSION_SUFFIXES),
    )

    # Find module. Clear search cache to be sure dynamically created
    # (new) modules are found
    finder.invalidate_caches()
    spec = finder.find_spec(module_name)
    if spec is None:
        raise ModuleNotFoundError("Unable to find JIT module.")

    # Load module
    compiled_module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(compiled_module)

    compiled_objects = []
    for name in object_names:
        obj = getattr(compiled_module.lib, name)
        compiled_objects.append(obj)

    return compiled_objects, compiled_module

1	# Copyright (C) 2004-2019 Garth N. Wells
2	#
3	# This file is part of FFCx.(https://www.fenicsproject.org)
4	#
5	# SPDX-License-Identifier: LGPL-3.0-or-later
6	"""Just-in-time compilation."""
7
8	from __future__ import annotations	1✔
9
10	import importlib	1✔
11	import io	1✔
12	import logging	1✔
13	import os	1✔
14	import re	1✔
15	import sys	1✔
16	import sysconfig	1✔
17	import tempfile	1✔
18	import time	1✔
19	from contextlib import redirect_stdout	1✔
20	from pathlib import Path	1✔
21
22	import cffi	1✔
23	import numpy as np	1✔
24	import numpy.typing as npt	1✔
25	import ufl	1✔
26
27	import ffcx	1✔
28	import ffcx.naming	1✔
29	from ffcx.codegeneration.C.file_template import libraries as _libraries	1✔
30
31	logger = logging.getLogger("ffcx")	1✔
32	root_logger = logging.getLogger()	1✔
33
34	# Get declarations directly from ufcx.h
35	file_dir = os.path.dirname(os.path.abspath(__file__))	1✔
36	with open(file_dir + "/ufcx.h") as f:	1✔
37	ufcx_h = "".join(f.readlines())	1✔
38
39	# Emulate C preprocessor on __STDC_NO_COMPLEX__
40	if sys.platform.startswith("win32"):	1✔
41	# Remove macro statements and content
42	ufcx_h = re.sub(	×
43	r"\#ifndef __STDC_NO_COMPLEX__.*?\#endif // __STDC_NO_COMPLEX__",
44	"",
45	ufcx_h,
46	flags=re.DOTALL,
47	)
48	else:
49	# Remove only macros keeping content
50	ufcx_h = ufcx_h.replace("#ifndef __STDC_NO_COMPLEX__", "")	1✔
51	ufcx_h = ufcx_h.replace("#endif // __STDC_NO_COMPLEX__", "")	1✔
52
53	header = ufcx_h.split("<HEADER_DECL>")[1].split("</HEADER_DECL>")[0].strip(" /\n")	1✔
54	header = header.replace("{", "{{").replace("}", "}}")	1✔
55	UFC_HEADER_DECL = header + "\n"	1✔
56
57	UFC_FORM_DECL = "\n".join(re.findall("typedef struct ufcx_form.*?ufcx_form;", ufcx_h, re.DOTALL))	1✔
58
59	UFC_INTEGRAL_DECL = "\n".join(	1✔
60	re.findall(r"typedef void ?\(ufcx_tabulate_tensor_float32\).*?\);", ufcx_h, re.DOTALL)
61	)
62	UFC_INTEGRAL_DECL += "\n".join(	1✔
63	re.findall(r"typedef void ?\(ufcx_tabulate_tensor_float64\).*?\);", ufcx_h, re.DOTALL)
64	)
65	UFC_INTEGRAL_DECL += "\n".join(	1✔
66	re.findall(r"typedef void ?\(ufcx_tabulate_tensor_complex64\).*?\);", ufcx_h, re.DOTALL)
67	)
68	UFC_INTEGRAL_DECL += "\n".join(	1✔
69	re.findall(r"typedef void ?\(ufcx_tabulate_tensor_complex128\).*?\);", ufcx_h, re.DOTALL)
70	)
71
72	UFC_INTEGRAL_DECL += "\n".join(	1✔
73	re.findall("typedef struct ufcx_integral.*?ufcx_integral;", ufcx_h, re.DOTALL)
74	)
75
76	UFC_EXPRESSION_DECL = "\n".join(	1✔
77	re.findall("typedef struct ufcx_expression.*?ufcx_expression;", ufcx_h, re.DOTALL)
78	)
79
80
81	def _compute_option_signature(options):	1✔
82	"""Return options signature (some options should not affect signature)."""
83	return str(sorted(options.items()))	1✔
84
85
86	def get_cached_module(module_name, object_names, cache_dir, timeout):	1✔
87	"""Look for an existing C file and wait for compilation, or if it does not exist, create it."""
88	cache_dir = Path(cache_dir)	1✔
89	c_filename = cache_dir.joinpath(module_name).with_suffix(".c")	1✔
90	ready_name = c_filename.with_suffix(".c.cached")	1✔
91
92	# Ensure cache dir exists
93	cache_dir.mkdir(exist_ok=True, parents=True)	1✔
94
95	try:	1✔
96	# Create C file with exclusive access
97	with open(c_filename, "x"):	1✔
98	pass	1✔
99	return None, None	1✔
100	except FileExistsError:	×
101	logger.info("Cached C file already exists: " + str(c_filename))	×
102	finder = importlib.machinery.FileFinder(	×
103	str(cache_dir),
104	(importlib.machinery.ExtensionFileLoader, importlib.machinery.EXTENSION_SUFFIXES),
105	)
106	finder.invalidate_caches()	×
107
108	# Now, wait for ready
109	for i in range(timeout):	×
110	if os.path.exists(ready_name):	×
111	spec = finder.find_spec(module_name)	×
112	if spec is None:	×
113	raise ModuleNotFoundError("Unable to find JIT module.")	×
114	compiled_module = importlib.util.module_from_spec(spec)	×
115	spec.loader.exec_module(compiled_module)	×
116
117	compiled_objects = [getattr(compiled_module.lib, name) for name in object_names]	×
118	return compiled_objects, compiled_module	×
119
120	logger.info(f"Waiting for {ready_name} to appear.")	×
121	time.sleep(1)	×
122	raise TimeoutError(	×
123	"JIT compilation timed out, probably due to a failed previous compile. "
124	f"Try cleaning cache (e.g. remove {c_filename}) or increase timeout option."
125	)
126
127
128	def _compilation_signature(cffi_extra_compile_args, cffi_debug):	1✔
129	"""Compute the compilation-inputs part of the signature.
130
131	Used to avoid cache conflicts across Python versions, architectures, installs.
132
133	- SOABI includes platform, Python version, debug flags
134	- CFLAGS includes prefixes, arch targets
135	"""
136	if sys.platform.startswith("win32"):	1✔
137	# NOTE: SOABI not defined on win32, EXT_SUFFIX contains e.g. '.cp312-win_amd64.pyd'
138	return (	×
139	str(cffi_extra_compile_args)
140	+ str(cffi_debug)
141	+ str(sysconfig.get_config_var("EXT_SUFFIX"))
142	)
143	else:
144	return (	1✔
145	str(cffi_extra_compile_args)
146	+ str(cffi_debug)
147	+ str(sysconfig.get_config_var("CFLAGS"))
148	+ str(sysconfig.get_config_var("SOABI"))
149	)
150
151
152	def compile_forms(	1✔
153	forms: list[ufl.Form],
154	options: dict = {},
155	cache_dir: Path \| None = None,
156	timeout: int = 10,
157	cffi_extra_compile_args: list[str] = [],
158	cffi_verbose: bool = False,
159	cffi_debug: bool = False,
160	cffi_libraries: list[str] = [],
161	visualise: bool = False,
162	):
163	"""Compile a list of UFL forms into UFC Python objects.
164
165	Args:
166	forms: List of ufl.form to compile.
167	options: Options
168	cache_dir: Cache directory
169	timeout: Timeout
170	cffi_extra_compile_args: Extra compilation args for CFFI
171	cffi_verbose: Use verbose compile
172	cffi_debug: Use compiler debug mode
173	cffi_libraries: libraries to use with compiler
174	visualise: Toggle visualisation
175	"""
176	p = ffcx.options.get_options(options)	1✔
177
178	# Get a signature for these forms
179	module_name = "libffcx_forms_" + ffcx.naming.compute_signature(	1✔
180	forms,
181	_compute_option_signature(p) + _compilation_signature(cffi_extra_compile_args, cffi_debug),
182	)
183
184	form_names = [ffcx.naming.form_name(form, i, module_name) for i, form in enumerate(forms)]	1✔
185
186	if cache_dir is not None:	1✔
187	cache_dir = Path(cache_dir)	1✔
188	obj, mod = get_cached_module(module_name, form_names, cache_dir, timeout)	1✔
189	if obj is not None:	1✔
190	return obj, mod, (None, None)	×
191	else:
192	cache_dir = Path(tempfile.mkdtemp())	1✔
193
194	try:	1✔
195	decl = (	1✔
196	UFC_HEADER_DECL.format(np.dtype(p["scalar_type"]).name) # type: ignore
197	+ UFC_INTEGRAL_DECL
198	+ UFC_FORM_DECL
199	)
200
201	form_template = "extern ufcx_form {name};\n"	1✔
202	for name in form_names:	1✔
203	decl += form_template.format(name=name)	1✔
204
205	impl = _compile_objects(	1✔
206	decl,
207	forms,
208	form_names,
209	module_name,
210	p,
211	cache_dir,
212	cffi_extra_compile_args,
213	cffi_verbose,
214	cffi_debug,
215	cffi_libraries,
216	visualise=visualise,
217	)
218	except Exception as e:	×
219	try:	×
220	# remove c file so that it will not timeout next time
221	c_filename = cache_dir.joinpath(module_name + ".c")	×
222	os.replace(c_filename, c_filename.with_suffix(".c.failed"))	×
223	except Exception:	×
224	pass	×
225	raise e	×
226
227	obj, module = _load_objects(cache_dir, module_name, form_names)	1✔
228	return obj, module, (decl, impl)	1✔
229
230
231	def compile_expressions(	1✔
232	expressions: list[tuple[ufl.Expr, npt.NDArray[np.floating]]],
233	options: dict = {},
234	cache_dir: Path \| None = None,
235	timeout: int = 10,
236	cffi_extra_compile_args: list[str] = [],
237	cffi_verbose: bool = False,
238	cffi_debug: bool = False,
239	cffi_libraries: list[str] = [],
240	visualise: bool = False,
241	):
242	"""Compile a list of UFL expressions into UFC Python objects.
243
244	Args:
245	expressions: List of (UFL expression, evaluation points).
246	options: Options
247	cache_dir: Cache directory
248	timeout: Timeout
249	cffi_extra_compile_args: Extra compilation args for CFFI
250	cffi_verbose: Use verbose compile
251	cffi_debug: Use compiler debug mode
252	cffi_libraries: libraries to use with compiler
253	visualise: Toggle visualisation
254	"""
255	p = ffcx.options.get_options(options)	1✔
256
257	module_name = "libffcx_expressions_" + ffcx.naming.compute_signature(	1✔
258	expressions,
259	_compute_option_signature(p) + _compilation_signature(cffi_extra_compile_args, cffi_debug),
260	)
261	expr_names = [	1✔
262	ffcx.naming.expression_name(expression, module_name) for expression in expressions
263	]
264
265	if cache_dir is not None:	1✔
266	cache_dir = Path(cache_dir)	×
267	obj, mod = get_cached_module(module_name, expr_names, cache_dir, timeout)	×
268	if obj is not None:	×
269	return obj, mod, (None, None)	×
270	else:
271	cache_dir = Path(tempfile.mkdtemp())	1✔
272
273	try:	1✔
274	decl = (	1✔
275	UFC_HEADER_DECL.format(np.dtype(p["scalar_type"]).name) # type: ignore
276	+ UFC_INTEGRAL_DECL
277	+ UFC_FORM_DECL
278	+ UFC_EXPRESSION_DECL
279	)
280
281	expression_template = "extern ufcx_expression {name};\n"	1✔
282	for name in expr_names:	1✔
283	decl += expression_template.format(name=name)	1✔
284
285	impl = _compile_objects(	1✔
286	decl,
287	expressions,
288	expr_names,
289	module_name,
290	p,
291	cache_dir,
292	cffi_extra_compile_args,
293	cffi_verbose,
294	cffi_debug,
295	cffi_libraries,
296	visualise=visualise,
297	)
298	except Exception as e:	×
299	try:	×
300	# remove c file so that it will not timeout next time
301	c_filename = cache_dir.joinpath(module_name + ".c")	×
302	os.replace(c_filename, c_filename.with_suffix(".c.failed"))	×
303	except Exception:	×
304	pass	×
305	raise e	×
306
307	obj, module = _load_objects(cache_dir, module_name, expr_names)	1✔
308	return obj, module, (decl, impl)	1✔
309
310
311	def _compile_objects(	1✔
312	decl,
313	ufl_objects,
314	object_names,
315	module_name,
316	options,
317	cache_dir,
318	cffi_extra_compile_args,
319	cffi_verbose,
320	cffi_debug,
321	cffi_libraries,
322	visualise: bool = False,
323	):
324	import ffcx.compiler	1✔
325
326	libraries = _libraries + cffi_libraries if cffi_libraries is not None else _libraries	1✔
327
328	# JIT uses module_name as prefix, which is needed to make names of all struct/function
329	# unique across modules
330	_, code_body = ffcx.compiler.compile_ufl_objects(	1✔
331	ufl_objects, prefix=module_name, options=options, visualise=visualise
332	)
333
334	# Raise error immediately prior to compilation if no support for C99
335	# _Complex. Doing this here allows FFCx to be used for complex codegen on
336	# Windows.
337	if sys.platform.startswith("win32"):	1✔
338	if np.issubdtype(options["scalar_type"], np.complexfloating):	×
339	raise NotImplementedError("win32 platform does not support C99 _Complex numbers")	×
340	elif isinstance(options["scalar_type"], str) and "complex" in options["scalar_type"]:	×
341	raise NotImplementedError("win32 platform does not support C99 _Complex numbers")	×
342
343	# Compile in C17 mode
344	if sys.platform.startswith("win32"):	1✔
345	cffi_base_compile_args = ["-std:c17"]	×
346	else:
347	cffi_base_compile_args = ["-std=c17"]	1✔
348
349	cffi_final_compile_args = cffi_base_compile_args + cffi_extra_compile_args	1✔
350
351	ffibuilder = cffi.FFI()	1✔
352
353	ffibuilder.set_source(	1✔
354	module_name,
355	code_body,
356	include_dirs=[ffcx.codegeneration.get_include_path()],
357	extra_compile_args=cffi_final_compile_args,
358	libraries=libraries,
359	)
360
361	ffibuilder.cdef(decl)	1✔
362
363	c_filename = cache_dir.joinpath(module_name + ".c")	1✔
364	ready_name = c_filename.with_suffix(".c.cached")	1✔
365
366	# Compile (ensuring that compile dir exists)
367	cache_dir.mkdir(exist_ok=True, parents=True)	1✔
368
369	logger.info(79 * "#")	1✔
370	logger.info("Calling JIT C compiler")	1✔
371	logger.info(79 * "#")	1✔
372
373	t0 = time.time()	1✔
374	f = io.StringIO()	1✔
375	# Temporarily set root logger handlers to string buffer only
376	# since CFFI logs into root logger
377	old_handlers = root_logger.handlers.copy()	1✔
378	root_logger.handlers = [logging.StreamHandler(f)]	1✔
379	with redirect_stdout(f):	1✔
380	ffibuilder.compile(tmpdir=cache_dir, verbose=True, debug=cffi_debug)	1✔
381	s = f.getvalue()	1✔
382	if cffi_verbose:	1✔
383	print(s)	×
384
385	logger.info(f"JIT C compiler finished in {time.time() - t0:.4f}")	1✔
386
387	# Create a "status ready" file. If this fails, it is an error,
388	# because it should not exist yet.
389	# Copy the stdout verbose output of the build into the ready file
390	fd = open(ready_name, "x")	1✔
391	fd.write(s)	1✔
392	fd.close()	1✔
393
394	# Copy back the original handlers (in case someone is logging into
395	# root logger and has custom handlers)
396	root_logger.handlers = old_handlers	1✔
397
398	return code_body	1✔
399
400
401	def _load_objects(cache_dir, module_name, object_names):	1✔
402	# Create module finder that searches the compile path
403	finder = importlib.machinery.FileFinder(	1✔
404	str(cache_dir),
405	(importlib.machinery.ExtensionFileLoader, importlib.machinery.EXTENSION_SUFFIXES),
406	)
407
408	# Find module. Clear search cache to be sure dynamically created
409	# (new) modules are found
410	finder.invalidate_caches()	1✔
411	spec = finder.find_spec(module_name)	1✔
412	if spec is None:	1✔
413	raise ModuleNotFoundError("Unable to find JIT module.")	×
414
415	# Load module
416	compiled_module = importlib.util.module_from_spec(spec)	1✔
417	spec.loader.exec_module(compiled_module)	1✔
418
419	compiled_objects = []	1✔
420	for name in object_names:	1✔
421	obj = getattr(compiled_module.lib, name)	1✔
422	compiled_objects.append(obj)	1✔
423
424	return compiled_objects, compiled_module	1✔

FEniCS / ffcx / 11642291501

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous