# -----------------------------------------------------------------------------
# SPDX-FileCopyrightText: Copyright (c) 2017-2026 Science and Technology
# Facilities Council
# SPDX-License-Identifier: BSD-3-Clause
# See the full LICENSE file in the project root for details.
# -----------------------------------------------------------------------------
''' This module provides the KernelModuleInlineTrans transformation.
TODO #2683 - rename this to {Privatise,Copy,Move}RoutineToLocalContainerTrans
and move it to psyir/transformations/.
'''
import logging
from typing import Any, Optional, Union
import warnings
from psyclone.psyGen import Transformation, CodedKern
from psyclone.psyir.transformations import TransformationError
from psyclone.psyir.symbols import (
ContainerSymbol, GenericInterfaceSymbol, RoutineSymbol, Symbol,
SymbolError)
from psyclone.psyir.nodes import (
Call, Container, FileContainer, IntrinsicCall, Reference, Routine,
ScopingNode)
from psyclone.utils import transformation_documentation_wrapper
[docs]
@transformation_documentation_wrapper
class KernelModuleInlineTrans(Transformation):
''' Brings the routine being called into the same Container as the call
site. For example:
>>> from psyclone.domain.common.transformations import \\
... KernelModuleInlineTrans
>>> from psyclone.psyGen import CodedKern
>>> from psyclone.psyir.frontend.fortran import FortranReader
>>> from psyclone.psyir.nodes import Call
>>>
>>> psyir = FortranReader().psyir_from_source("""
... module one
... contains
... subroutine my_subroutine()
... integer, dimension(10, 10) :: A
... A(:,:) = 0
... end subroutine my_subroutine
... end module one
... module two
... use one, only: my_subroutine
... contains
... subroutine call_it()
... call my_subroutine()
... call my_subroutine()
... end subroutine call_it
... end module two
... """)
>>> call = psyir.walk(Call)[-1]
>>> inline_trans = KernelModuleInlineTrans()
>>> inline_trans.apply(call)
>>>
>>> print(call.parent.parent.debug_string())
module two
use one, only : my_subroutine
implicit none
public
<BLANKLINE>
private :: my_subroutine_inlined_
<BLANKLINE>
contains
subroutine call_it()
<BLANKLINE>
call my_subroutine_inlined_()
call my_subroutine_inlined_()
<BLANKLINE>
end subroutine call_it
subroutine my_subroutine_inlined_()
integer, dimension(10,10) :: a
<BLANKLINE>
a(:,:) = 0
<BLANKLINE>
end subroutine my_subroutine_inlined_
<BLANKLINE>
end module two
<BLANKLINE>
.. warning ::
Not all Routines can be moved. This transformation will reject
attempts to move routines that access private data in the
original Container.
'''
def __str__(self):
return ("Copy the routine associated with a (Kernel) call into the "
"Container of the call site.")
# pylint: disable=too-many-branches
[docs]
def validate(self,
node: Union[CodedKern, Call],
options: Optional[dict[str, Any]] = None,
**kwargs) -> None:
'''
Checks that the supplied node is a Kernel or Call and that it is
possible to copy its PSyIR into the parent Container.
If the target of the supplied (Kernel) Call is already in local
scope then further checks are skipped.
:param node: the kernel or call which is the target of the
transformation.
:param options: a dictionary with options for transformations.
:raises TransformationError: if the target node is not a sub-class of
psyGen.CodedKern or psyir.nodes.Call or is an IntrinsicCall.
:raises TransformationError: if the call is to a polymorphic routine
and there's no Container at the call site to which to add the
interface definition.
:raises TransformationError: if the kernel cannot be safely inlined.
'''
if not options:
self.validate_options(**kwargs)
if isinstance(node, CodedKern):
kern_or_call = "Kernel"
elif isinstance(node, Call):
if isinstance(node, IntrinsicCall):
raise TransformationError(
f"Cannot module-inline a call to an intrinsic (got "
f"'{node.debug_string()}')")
kern_or_call = "routine"
else:
raise TransformationError(
f"Target of a {self.name} must be a sub-class of "
f"psyGen.CodedKern or psyir.nodes.Call but got "
f"'{type(node).__name__}'")
# Check that the PSyIR of the routine/kernel can be retrieved.
kname = node.routine.symbol.name
try:
kernels = node.get_callees()
except Exception as error:
raise TransformationError(
f"{self.name} failed to retrieve PSyIR for {kern_or_call} "
f"'{kname}' due to: {error}"
) from error
# Return early if the target of the (Kernel) Call is already local.
if self._target_is_local(node):
return
if len(kernels) > 1:
# We can't bring the target of a call to an interface into local
# scope if there's no Container in which to put the interface.
cntr = node
while cntr:
cntr = cntr.ancestor(Container)
if cntr and not isinstance(cntr, FileContainer):
break
else:
raise TransformationError(
f"Cannot copy the target of the call to '{kname}' since "
f"it is a polymorphic routine (i.e. an interface) and the "
f"call-site is not within a module.")
# Validate the PSyIR of each routine/kernel.
for kernel_schedule in kernels:
self._validate_schedule(node, kname, kern_or_call, kernel_schedule)
def _validate_schedule(self, node, kname, kern_or_call, kernel_schedule):
'''
Validates that the supplied schedule can be module-inlined.
:param node: the candidate kernel/routine call to inline.
:type node: :py:class:`psyclone.psyGen.CodedKern` |
:py:class:`psyclone.psyir.nodes.Call`
:param str kname: the name of the kernel/routine.
:param str kern_or_call: text for readable error messages.
:param kernel_schedule: the schedule of the routine to inline.
:type kernel_schedule: :py:class:`psyclone.psyir.nodes.Schedule`
:raises TransformationError: if the called routine contains accesses
to data declared in the same module scope or of unknown origin.
:raises TransformationError: if the called routine contains a local
Symbol that shadows a module name in its outer scope.
'''
# We do not support kernels that use symbols representing data
# declared in their own parent module (we would need to add new imports
# from this module at the call site, and we don't do this yet).
try:
kernel_schedule.check_outer_scope_accesses(node, kern_or_call)
except SymbolError as err:
raise TransformationError(
f"Cannot apply {self.name} to {kern_or_call} '{kname}' "
f"because it accesses data from its outer scope: "
f"{err.value}") from err
# We can't transform subroutines that shadow top-level symbol module
# names, because we won't be able to bring them into the subroutine.
# (We could attempt to rename the local symbol.)
symtab = kernel_schedule.ancestor(Container).symbol_table
ctr_names = [sym.name.lower() for sym in symtab.containersymbols]
for scope in kernel_schedule.walk(ScopingNode):
for symbol in scope.symbol_table.symbols:
if (symbol.name.lower() in ctr_names and
not isinstance(symbol, ContainerSymbol)):
raise TransformationError(
f"{kern_or_call} '{kname}' cannot be module-"
f"inlined because the subroutine contains a symbol "
f"'{symbol.name}' which shadows the name of a module "
f"in the outer scope.")
@staticmethod
def _prepare_code_to_inline(
routines_to_inline: list[Routine]) -> list[Routine]:
'''Prepare the PSyIR tree to inline by bringing in to the subroutine
all referenced symbols so that the implementation is self contained.
The provided routines are copied so that the original PSyIR is left
unmodified.
:param routines_to_inline: the routine(s) to module-inline.
:returns: the updated routine(s) to module-inline.
'''
# pylint: disable=too-many-branches
orig_container = routines_to_inline[0].ancestor(Container)
# Since we will be detaching Routines, we work with a copy of
# the Container that encapsulates them.
source_container = orig_container.copy()
new_routines = {}
for routine in source_container.walk(Routine):
new_routines[routine.name] = routine
copied_routines = []
for orig_routine in routines_to_inline:
code_to_inline = new_routines[orig_routine.name]
copied_routines.append(code_to_inline)
vam = code_to_inline.reference_accesses()
# First make a set with all symbols used inside the subroutine
all_symbols = set()
for sig in vam.all_signatures:
all_symbols.add(
code_to_inline.symbol_table.lookup(sig.var_name))
# Decide which symbols need to be brought inside the subroutine
symbols_to_bring_in = set()
for symbol in all_symbols:
if symbol.is_unresolved or symbol.is_import:
# This symbol may already be in the local symbol table,
# but adding it to the 'symbols_to_bring_in' will make the
# next step bring into the subroutine all modules that it
# could come from.
symbols_to_bring_in.add(symbol)
# Bring the selected symbols inside the subroutine
for symbol in symbols_to_bring_in:
if symbol.name not in code_to_inline.symbol_table:
if symbol.is_import:
# We must update its import interface (to ensure it
# references a ContainerSymbol in the correct scope)
# before it can be added to the table.
code_to_inline.symbol_table.\
localise_import_interface_of(symbol)
code_to_inline.symbol_table.add(symbol)
# And when necessary the modules where they come from
if symbol.is_unresolved:
# We don't know where this comes from, we need to bring
# in all top-level imports with wildcard imports
for mod in source_container.symbol_table.containersymbols:
if mod.wildcard_import:
if mod.name not in code_to_inline.symbol_table:
code_to_inline.symbol_table.add(mod)
else:
code_to_inline.symbol_table.lookup(mod.name).\
wildcard_import = True
elif symbol.is_import:
module_symbol = symbol.interface.container_symbol
if module_symbol.name not in code_to_inline.symbol_table:
code_to_inline.symbol_table.add(module_symbol)
else:
# If it already exists, we know it's a container (from
# the validation) so we just need to point to it
symbol.interface.container_symbol = \
code_to_inline.symbol_table.lookup(
module_symbol.name)
return copied_routines
def _target_is_local(self, node: Union[Call, CodedKern]) -> bool:
'''
:returns: whether or not the target of the supplied call/kernel
is already in local scope.
'''
kname = node.routine.symbol.name
routines = node.get_callees()
if len(routines) > 1:
iface_sym = node.scope.symbol_table.lookup(kname, otherwise=None)
if (not iface_sym or (iface_sym.is_import or
iface_sym.is_unresolved)):
return False
for kernel_schedule in routines:
rt_sym = node.scope.symbol_table.lookup(kernel_schedule.name,
otherwise=None)
if (not rt_sym or (rt_sym is not kernel_schedule.symbol) or
(node.ancestor(Container) is not
kernel_schedule.ancestor(Container)) or
(rt_sym.is_import or rt_sym.is_unresolved)):
return False
logger = logging.getLogger(__name__)
logger.info(
f"The target of '{node.debug_string().strip()}' is already "
f"present in the local scope.")
return True
[docs]
def apply(self,
node: Union[CodedKern, Call],
options: dict[str, Any] = None,
update_all: bool = True,
**kwargs):
''' Bring the implementation of this kernel/call into this Container.
NOTE: when applying this transformation to a Kernel in a PSyKAl invoke,
by default *all* calls to that Kernel are updated.
Similarly, when applied to a Call to a Routine in a particular scope,
*all* such Calls are updated. This behaviour may be changed using the
`update_all=False` option.
:param node: the Kernel or Call to module-inline.
:param options: a dictionary with options for transformations.
:param update_all: whether or not to update *all* (kernel) calls to the
target Kernel/routine within the program unit.
'''
if options:
# TODO 2668 - options dict is deprecated.
warnings.warn(self._deprecation_warning, DeprecationWarning, 2)
if not options:
options = {}
self.validate(node, options, **kwargs)
external_callee_name = None
if isinstance(node, CodedKern):
caller_name = node.name
else:
caller_name = node.routine.symbol.name
if (node.routine.symbol.is_import and
node.routine.symbol.interface.orig_name):
external_callee_name = node.routine.symbol.interface.orig_name
if not external_callee_name:
external_callee_name = caller_name
# Get the PSyIR of the routine to module inline as well as the name
# with which it is being called.
# Note that we use the resolved callee subroutine name and not the
# caller one; this is important because if it is an interface it will
# use the concrete implementation name. When this happens the new name
# may already be in use, but the equality check below guarantees
# that if it exists it is only valid when it references the exact same
# implementation.
codes_to_inline = node.get_callees()
interface_sym = None
if len(codes_to_inline) > 1:
interface_sym = codes_to_inline[0].symbol_table.lookup(
external_callee_name)
if self._target_is_local(node):
return
callsite_table = node.scope.symbol_table
if interface_sym:
called_sym = callsite_table.lookup(caller_name, otherwise=None)
else:
for routine in codes_to_inline:
# N.B. in a PSyKAl DSL, we won't have a RoutineSymbol for the
# Kernel that is being called, so we look it up instead of
# using node.symbol.
called_sym = callsite_table.lookup(caller_name,
otherwise=None)
if (not called_sym or called_sym is not routine.symbol or
(called_sym.is_import or called_sym.is_unresolved)):
# This routine is not module-inlined.
break
updated_routines = self._prepare_code_to_inline(codes_to_inline)
# The Container into which we will inline the Routine(s).
container = node.ancestor(Container)
# Mapping from original name to new RoutineSymbol.
name_map: dict[str, RoutineSymbol] = {}
for code_to_inline in updated_routines:
# Create a new name for the routine.
new_name = code_to_inline.name+"_inlined_"
new_sym = container.symbol_table.new_symbol(
new_name, symbol_type=RoutineSymbol)
new_sym.copy_properties(code_to_inline.symbol,
exclude_interface=True)
# Do not expose the new symbol externally to prevent unexpected
# collisions.
new_sym.visibility = Symbol.Visibility.PRIVATE
# Add the new symbol to the map.
name_map[code_to_inline.name] = new_sym
# Add the routine code into this Container
code_to_inline = code_to_inline.detach()
code_to_inline.symbol = new_sym
container.addchild(code_to_inline)
if interface_sym:
# Deal with the interface symbol - create a new, local
# private version.
new_sym = container.symbol_table.new_symbol(
interface_sym.name+"_inlined_",
symbol_type=GenericInterfaceSymbol,
routines=[(sym, True) for sym in name_map.values()],
visibility=Symbol.Visibility.PRIVATE)
name_map[interface_sym.name] = new_sym
if update_all:
# We will update all Calls/Kernels associated with the
# target routine.
all_calls = container.walk(type(node))
else:
# Only update the supplied Call/Kernel.
all_calls = [node]
target_sym = name_map.get(caller_name, None)
if not target_sym:
# If we haven't copied in a routine of 'caller_name' then it must
# be because the target of the call is renamed on import.
target_sym = name_map.get(external_callee_name)
for call in all_calls:
name = call.routine.symbol.name.lower()
if name == caller_name:
if isinstance(node, Call):
call.routine.symbol = target_sym
else:
# Otherwise node is a CodedKern.
call.routine = Reference(target_sym)
call._schedules = updated_routines
# In theory we could remove the import of the original routine but
# that is dangerous (e.g. if it's a subroutine-scoped or
# module-private symbol) and unnecessary so we don't.
