Explicit copy memset nodes

dace/codegen/codegen.py

@@ -204,6 +204,16 @@ def generate_code(sdfg: SDFG, validate=True) -> List[CodeObject]:
     # Set default storage/schedule types in SDFG
     infer_types.set_default_schedule_and_storage_types(sdfg, None)
 
+    # Give every implicit copy a node of its own, before the expansion below lowers it. An implicit
+    # copy is a write no node performs, so an empty memlet ordering that write has nothing to point
+    # at and the copy is free to move ahead of a write it must follow.
+    if config.Config.get_bool('compiler', 'cpu', 'explicit_copy'):
+        from dace.transformation.passes.insert_explicit_copies import InsertExplicitCopies
+        InsertExplicitCopies().apply_pass(sdfg, {})
+        # The nodes just inserted carry no inferred connector types, and the expansion below reads
+        # them to decide pointer vs. value. Storage defaults above already hold.
+        infer_types.infer_connector_types(sdfg)
+
     # Recursively expand library nodes that have not yet been expanded
     sdfg.expand_library_nodes()
 

dace/codegen/instrumentation/gpu_tx_markers.py

@@ -152,6 +152,24 @@ def on_copy_end(self, sdfg: SDFG, cfg: ControlFlowRegion, state: SDFGState, src_
             return
         self.print_range_pop(local_stream)
 
+    def on_node_begin(self, sdfg: SDFG, cfg: ControlFlowRegion, state: SDFGState, node: nodes.Node,
+                      outer_stream: CodeIOStream, inner_stream: CodeIOStream, global_stream: CodeIOStream) -> None:
+        if not isinstance(node, nodes.CodeNode) or node.instrument != dtypes.InstrumentationType.GPU_TX_MARKERS:
+            return
+        if is_devicelevel_gpu_kernel(sdfg, state, node):
+            # Don't instrument device code
+            return
+        self.print_range_push(node.label, sdfg, outer_stream)
+
+    def on_node_end(self, sdfg: SDFG, cfg: ControlFlowRegion, state: SDFGState, node: nodes.Node,
+                    outer_stream: CodeIOStream, inner_stream: CodeIOStream, global_stream: CodeIOStream) -> None:
+        if not isinstance(node, nodes.CodeNode) or node.instrument != dtypes.InstrumentationType.GPU_TX_MARKERS:
+            return
+        if is_devicelevel_gpu_kernel(sdfg, state, node):
+            # Don't instrument device code
+            return
+        self.print_range_pop(outer_stream)
+
     def on_scope_entry(self, sdfg: SDFG, cfg: ControlFlowRegion, state: SDFGState, node: nodes.EntryNode,
                        outer_stream: CodeIOStream, inner_stream: CodeIOStream, global_stream: CodeIOStream) -> None:
         if node.map.instrument != dtypes.InstrumentationType.GPU_TX_MARKERS:

dace/codegen/targets/cpp.py

@@ -813,11 +813,13 @@ def unparse_cr(sdfg, wcr_ast, dtype):
 
 
 def connected_to_gpu_memory(node: nodes.Node, state: SDFGState, sdfg: SDFG):
+    # Both ends of the path count: a host tasklet that only WRITES GPU memory needs the stream just
+    # as much as one that reads it. Same rule as the stream-retention walk in ``cuda.py``.
     for e in state.all_edges(node):
         path = state.memlet_path(e)
-        if ((isinstance(path[0].src, nodes.AccessNode)
-             and path[0].src.desc(sdfg).storage is dtypes.StorageType.GPU_Global)):
-            return True
+        for endpoint in (path[0].src, path[-1].dst):
+            if isinstance(endpoint, nodes.AccessNode) and endpoint.desc(sdfg).storage is dtypes.StorageType.GPU_Global:
+                return True
     return False
 
 
@@ -915,7 +917,17 @@ def unparse_tasklet(sdfg, cfg, state_id, dfg, node, function_stream, callsite_st
                 gpu_codegen = next(cg for cg in codegen._dispatcher.used_targets if isinstance(cg, cuda.CUDACodeGen))
             except StopIteration:
                 return
-            synchronize_streams(sdfg, cfg, state_dfg, state_id, node, node, callsite_stream, gpu_codegen)
+            # The tasklet's own code names the stream through the local defined above, so the
+            # synchronization it may need must name the same expression.
+            synchronize_streams(sdfg,
+                                cfg,
+                                state_dfg,
+                                state_id,
+                                node,
+                                node,
+                                callsite_stream,
+                                gpu_codegen,
+                                stream_expr='__dace_current_stream')
         return
 
     body = node.code.code
@@ -1346,10 +1358,12 @@ def presynchronize_streams(sdfg: SDFG, cfg: ControlFlowRegion, dfg: StateSubgrap
 
 
 # TODO: This should be in the CUDA code generator. Add appropriate conditions to node dispatch predicate
-def synchronize_streams(sdfg, cfg, dfg, state_id, node, scope_exit, callsite_stream, codegen):
+def synchronize_streams(sdfg, cfg, dfg, state_id, node, scope_exit, callsite_stream, codegen, stream_expr=None):
     # Post-kernel stream synchronization (with host or other streams)
     max_streams = int(Config.get("compiler", "cuda", "max_concurrent_streams"))
-    if max_streams >= 0:
+    if stream_expr is not None:
+        cudastream = stream_expr
+    elif max_streams >= 0:
         cudastream = common.gpu_stream_expr(node._cuda_stream)
     else:  # Only default stream is used
         cudastream = 'nullptr'
@@ -1393,8 +1407,30 @@ def synchronize_streams(sdfg, cfg, dfg, state_id, node, scope_exit, callsite_str
     if max_streams >= 0 and hasattr(node, "_cuda_stream"):
         backend = common.get_gpu_backend()
 
+        synced_host = False
         for edge in dfg.out_edges(scope_exit):
 
+            # A host-located destination is read by plain host code as soon as the asynchronous
+            # work is issued -- a kernel launch packing a by-value argument counts -- and neither
+            # events nor consumer stream stamps order the host. Wait on the issuing stream once
+            # (copy-edge analog: ``_emit_copy`` in cuda.py).
+            hostnode = edge.dst
+            while (isinstance(hostnode, nodes.AccessNode) and hostnode.data is not None
+                   and isinstance(sdfg.arrays[hostnode.data], data.View)):
+                hostnode = dfg.out_edges(hostnode)[0].dst
+            if (isinstance(hostnode, nodes.AccessNode) and hostnode.data is not None
+                    and sdfg.arrays[hostnode.data].storage
+                    not in (dtypes.StorageType.GPU_Global, dtypes.StorageType.GPU_Shared)):
+                if not synced_host:
+                    callsite_stream.write(
+                        "DACE_GPU_CHECK(%sStreamSynchronize(%s));" % (backend, cudastream),
+                        cfg,
+                        state_id,
+                        [edge.src, edge.dst],
+                    )
+                    synced_host = True
+                continue
+
             if (isinstance(edge.dst, nodes.AccessNode) and hasattr(edge.dst, '_cuda_stream')
                     and edge.dst._cuda_stream != node._cuda_stream):
                 # Stream assignment gives a cross-stream edge its own event. Event 0 belongs to some

dace/codegen/targets/cpu.py

@@ -1305,6 +1305,17 @@ def memlet_definition(self,
                         # constexpr arrays
                         if memlet.data in self._frame.symbols_and_constants(sdfg):
                             result += "const {} {} = {};".format(memlet_type, local_name, expr)
+                        elif (var_type == DefinedType.Scalar and isinstance(conntype, dtypes.pointer)
+                              and not isinstance(desc.dtype, dtypes.opaque)):
+                            # Scalar source feeding a pointer-typed connector (e.g. CopyLibraryNode
+                            # -> cudaMemcpyAsync from a host scalar argument). The connector's
+                            # pointer type wins over the source's scalar ctypedef, and the address
+                            # of the variable is what the callee wants; `define_out_memlet` already
+                            # does this on the write side. Skip opaque dtypes (MPI_Comm /
+                            # MPI_Request / GPU handles) -- the value is already a pointer-like
+                            # handle, so address-of adds an indirection the callee rejects
+                            # (``MPI_Bcast`` expects ``MPI_Comm``, not ``MPI_Comm *``).
+                            result += "{}* {} = &{};".format(ctypedef, local_name, expr)
                         else:
                             # Pointer reference
                             result += "{} {} = {};".format(ctypedef, local_name, expr)

dace/codegen/targets/cuda.py

@@ -99,6 +99,9 @@ def __init__(self, frame_codegen: 'DaCeCodeGenerator', sdfg: SDFG):
         self._exitcode = CodeIOStream()
         self._global_sdfg: SDFG = sdfg
         self._toplevel_schedule = None
+        # True while generating an SDFG nested below the one whose schedule established the current
+        # device-level scope, i.e. no longer the kernel's own state machine.
+        self._below_toplevel_sdfg = False
         self._arglists: Dict[nodes.MapEntry, Dict[str, dt.Data]] = {}
         """
         # Keep track of which kernels got a threadBlock map inserted
@@ -1633,7 +1636,15 @@ def generate_devicelevel_state(self, sdfg: SDFG, cfg: ControlFlowRegion, state:
 
                 callsite_stream.write("}  // subgraph end", cfg, state.block_id)
 
-            callsite_stream.write('__gbar.Sync();', cfg, state.block_id)
+            # A state machine needs a barrier between its states wherever it runs: a nested SDFG
+            # with several states (or control flow) below the kernel still synchronizes between
+            # them. An SDFG that is one lone state has no state transition to order, so below the
+            # kernel's own SDFG it emits no barrier -- it may run inside a single-thread-guarded
+            # component, where a barrier is reached by one thread and never releases. Its writes
+            # are ordered by the enclosing state's own barrier instead.
+            lone_state = sdfg.number_of_nodes() == 1 and isinstance(sdfg.nodes()[0], SDFGState)
+            if not (self._below_toplevel_sdfg and lone_state):
+                callsite_stream.write('__gbar.Sync();', cfg, state.block_id)
 
             # done here, code is generated
             return
@@ -1722,9 +1733,12 @@ def generate_scope(self, sdfg: SDFG, cfg: ControlFlowRegion, dfg_scope: StateSub
                     self._in_device_code = oldval
 
                     self.extra_nsdfg_args.append((desc.as_arg(name=''), inner_name, outer_name))
+                    # A DefinedType.Pointer registers the POINTER ctype, as every other allocation
+                    # site does; the element type here makes consumers that adopt the defined ctype
+                    # (``emit_memlet_reference``) declare the parameter one indirection short.
                     self._dispatcher.defined_vars.add(inner_name,
                                                       DefinedType.Pointer,
-                                                      desc.dtype.ctype,
+                                                      dtypes.pointer(desc.dtype).ctype,
                                                       allow_shadowing=True)
                     extra_call_args.append(outer_name)
                     extra_call_args_typed.append(desc.as_arg(name=inner_name))
@@ -2936,16 +2950,21 @@ def _generate_NestedSDFG(self, sdfg: SDFG, cfg: ControlFlowRegion, dfg: StateSub
                              node: nodes.NestedSDFG, function_stream: CodeIOStream,
                              callsite_stream: CodeIOStream) -> None:
         old_schedule = self._toplevel_schedule
+        old_below_toplevel = self._below_toplevel_sdfg
         nested_schedule = get_node_schedule(sdfg, dfg, node)
         if nested_schedule != dtypes.ScheduleType.Default:
             self._toplevel_schedule = nested_schedule
+        # A device-level scope is already open, so this SDFG sits inside one of its components
+        # rather than being the state machine that scope is made of.
+        self._below_toplevel_sdfg = old_schedule in dtypes.GPU_SCHEDULES
         old_codegen = self._cpu_codegen.calling_codegen
         self._cpu_codegen.calling_codegen = self
 
         self._cpu_codegen._generate_NestedSDFG(sdfg, cfg, dfg, state_id, node, function_stream, callsite_stream)
 
         self._cpu_codegen.calling_codegen = old_codegen
         self._toplevel_schedule = old_schedule
+        self._below_toplevel_sdfg = old_below_toplevel
 
     def _generate_MapExit(self, sdfg: SDFG, cfg: ControlFlowRegion, dfg: StateSubgraphView, state_id: int,
                           node: nodes.MapExit, function_stream: CodeIOStream, callsite_stream: CodeIOStream) -> None:

dace/config_schema.yml

@@ -349,6 +349,24 @@ required:
                             generate "#pragma omp parallel sections" code around
                             them.
 
+                    parallel_transfer_min_elements:
+                        type: int
+                        default: 262144
+                        title: Parallel copy/fill element threshold
+                        description: >
+                            Minimum element count (symbolic size counts as large) for a
+                            copy/fill to lower to an OpenMP element map instead of memcpy/memset.
+
+                    explicit_copy:
+                        type: bool
+                        default: true
+                        title: Lower implicit copies to explicit copy nodes
+                        description: >
+                            If set to true, codegen lifts every implicit copy edge to a
+                            CopyLibraryNode before expansion. An implicit copy is a write no node
+                            performs, so an empty memlet ordering that write has nothing to point
+                            at; giving the copy a node is what makes it orderable.
+
             #############################################
             # GPU (CUDA/HIP) compiler
             cuda:

dace/libraries/standard/environments/__init__.py

@@ -1,2 +1,3 @@
 # Copyright 2019-2026 ETH Zurich and the DaCe authors. All rights reserved.
+from .cpu import CPU
 from .cuda import CUDA

dace/libraries/standard/environments/cpu.py

@@ -0,0 +1,23 @@
+# Copyright 2019-2026 ETH Zurich and the DaCe authors. All rights reserved.
+"""DaCe library environment exposing the C++ standard headers used by CPU-side libnode expansions."""
+import dace.library
+
+
+@dace.library.environment
+class CPU:
+    """Minimal library environment that pulls in ``<cstring>`` for plain CPU expansions."""
+
+    cmake_minimum_version = None
+    cmake_packages = []
+    cmake_variables = {}
+    cmake_includes = []
+    cmake_libraries = []
+    cmake_compile_flags = []
+    cmake_link_flags = []
+    cmake_files = []
+
+    headers = {'frame': ["cstring"]}
+    state_fields = []
+    init_code = ""
+    finalize_code = ""
+    dependencies = []

dace/libraries/standard/helper.py

@@ -0,0 +1,96 @@
+# Copyright 2019-2026 ETH Zurich and the DaCe authors. All rights reserved.
+"""Shared helpers for CopyLibraryNode and FillLibraryNode expansions."""
+from typing import Callable, List, Tuple
+
+import dace
+from dace import dtypes
+from dace.sdfg import nodes
+from dace.sdfg.scope import is_in_scope
+
+# Both legacy and experimental codegens consume this exact name for stream wiring.
+CURRENT_STREAM_NAME = "__dace_current_stream"
+
+# Register is intentionally in neither set: resolves by scope (GPU register vs. host stack slot).
+GPU_RESIDENT_STORAGES = frozenset({
+    dtypes.StorageType.GPU_Global,
+    dtypes.StorageType.GPU_Shared,
+})
+CPU_RESIDENT_STORAGES = frozenset({
+    dtypes.StorageType.CPU_Heap,
+    dtypes.StorageType.CPU_Pinned,
+    dtypes.StorageType.CPU_ThreadLocal,
+})
+
+
+def collapse_shape_and_strides(
+        subset: dace.subsets.Range,
+        strides: List[dace.symbolic.SymExpr]) -> Tuple[List[dace.symbolic.SymExpr], List[dace.symbolic.SymExpr]]:
+    """Drop length-1 dims from a (subset, strides) pair; surviving strides scale by the subset step.
+
+    A tiled dimension (``b:e:step:tile``) addresses ``tile`` contiguous elements per step, which no
+    single (length, stride) pair expresses -- it expands into two dims: the step count at
+    ``stride * step``, then the tile at ``stride``.
+
+    :param subset: The access range, one ``(begin, end, step)`` per dimension.
+    :param strides: The parent array strides, aligned with ``subset``.
+    :returns: ``(collapsed_shape, collapsed_strides)`` with singletons removed.
+    """
+    collapsed_shape = []
+    collapsed_strides = []
+    # ``Range.size()`` already folds the tile in (``tile * ceiling((e + 1 - b) / step)``); dividing
+    # it back out is exact and avoids re-deriving a per-dim count formula that could drift from it.
+    for (_, _, s), stride, tile, dim_size in zip(subset, strides, subset.tile_sizes, subset.size()):
+        length = dim_size / tile
+        if length != 1:
+            collapsed_shape.append(length)
+            collapsed_strides.append(stride * s)
+        if tile != 1:
+            collapsed_shape.append(tile)
+            collapsed_strides.append(stride)
+    return collapsed_shape, collapsed_strides
+
+
+def is_parallel_cpu_transfer_size(num_elements: dace.symbolic.SymbolicType) -> bool:
+    """False only when ``num_elements`` is a compile-time constant below
+    ``compiler.cpu.parallel_transfer_min_elements``; a symbolic (unknown-at-compile-time) size
+    is assumed large and takes the parallel path too.
+
+    :param num_elements: total contiguous element count (constant or symbolic).
+    :returns: ``True`` to route to the mapped expansion, ``False`` to keep the single libc call.
+    """
+    threshold = int(dace.Config.get('compiler', 'cpu', 'parallel_transfer_min_elements'))
+    try:
+        return int(dace.symbolic.simplify(num_elements)) >= threshold
+    except (TypeError, ValueError):
+        return True
+
+
+def is_in_parallel_scope(node: nodes.LibraryNode, parent_state: dace.SDFGState) -> bool:
+    """True when a multi-threaded map encloses this transfer, so the mapped form would open one
+    OpenMP region per entry instead of one for the whole transfer.
+
+    ``Default`` counts: an unresolved enclosing map becomes ``CPU_Multicore`` at the top level.
+
+    :param node: the transfer library node.
+    :param parent_state: state containing ``node``.
+    :returns: ``True`` if a parallel map scope encloses the node, at any nesting depth.
+    """
+    return is_in_scope(parent_state.sdfg, parent_state, node,
+                       [dtypes.ScheduleType.CPU_Multicore, dtypes.ScheduleType.Default])
+
+
+def auto_dispatch(node: nodes.LibraryNode, parent_state: dace.SDFGState,
+                  select_fn: Callable[[nodes.LibraryNode, dace.SDFGState], str], library_cls: type):
+    """Dispatch a library node's ``'Auto'`` implementation to the one ``select_fn`` picks, setting
+    ``node.implementation`` so introspection reflects what was chosen.
+
+    :param node: the library node being expanded.
+    :param parent_state: state containing ``node`` (owning SDFG is ``parent_state.sdfg``).
+    :param select_fn: callable returning a concrete implementation name (not ``'Auto'``).
+    :param library_cls: the library node class with the ``implementations`` dict.
+    :returns: whatever the resolved expansion returns.
+    """
+    impl_name = select_fn(node, parent_state)
+    assert impl_name != 'Auto', f"{select_fn.__name__} must not return 'Auto'."
+    node.implementation = impl_name
+    return library_cls.implementations[impl_name].expansion(node, parent_state, parent_state.sdfg)

dace/libraries/standard/nodes/__init__.py

@@ -1,4 +1,6 @@
 # Copyright 2019-2026 ETH Zurich and the DaCe authors. All rights reserved.
 from .code import CodeLibraryNode
+from .copy import CopyLibraryNode
+from .fill import FillLibraryNode
 from .gearbox import Gearbox
 from .reduce import Reduce