23552450315

Committed 25 Mar 2026 04:35PM UTC coverage: 64.539% (+0.4%) from 64.115%

Build # 23552450315

Build Type

Pull #503

github

Committed by

web-flow

Commit Message

Merge 4f0691d9f into e255622a0

Pull Request Pull Request #503: Added pass and pipeline statistics

Coverage Stats

19 of 226 new or added lines in 5 files covered. (8.41%)

491 existing lines in 15 files now uncovered.

27003 of 41840 relevant lines covered (64.54%)

405.96 hits per line

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

73.81

/python/docc/compiler/docc_program.py

from abc import ABC, abstractmethod
from typing import Any, Optional
import os

from docc.sdfg import StructuredSDFG
from docc.sdfg._sdfg import (
    _enable_statistics,
    _statistics_enabled_by_env,
    _statistics_summary,
)
from docc.compiler.compiled_sdfg import CompiledSDFG
from docc.compiler.target_registry import (
    get_target_schedule_fn,
    get_target_compile_fn,
    get_target_expand_fn,
)


def _is_debug_dump() -> bool:
    return bool(os.environ.get("DOCC_DEBUG"))


class DoccProgram(ABC):

    def __init__(
        self,
        name: str,
        target: str = "none",
        category: str = "server",
        instrumentation_mode: Optional[str] = None,
        capture_args: Optional[bool] = None,
        remote_tuning: bool = False,
    ):
        self.name = name
        self.target = target
        self.category = category
        self.remote_tuning = remote_tuning
        self.last_sdfg: Optional[StructuredSDFG] = None
        self.cache: dict = {}
        self.debug_dump: bool = _is_debug_dump()

        # Check environment variable DOCC_CI
        docc_ci = os.environ.get("DOCC_CI", "")
        if docc_ci:
            if docc_ci == "regions":
                if instrumentation_mode is None:
                    instrumentation_mode = "ols"
            elif docc_ci == "arg-capture":
                if capture_args is None:
                    capture_args = True
            else:
                # Full mode (or unknown value treated as full)
                if instrumentation_mode is None:
                    instrumentation_mode = "ols"
                if capture_args is None:
                    capture_args = True

        self.instrumentation_mode = instrumentation_mode
        self.capture_args = capture_args

    @abstractmethod
    def __call__(self, *args: Any) -> Any:
        pass

    @abstractmethod
    def compile(self, *args: Any, output_folder: Optional[str] = None) -> CompiledSDFG:
        pass

    def sdfg_pipe(
        self,
        sdfg: StructuredSDFG,
        output_folder: Optional[str],
        instrumentation_mode: str,
        capture_args: bool,
    ) -> str:

        if self.debug_dump:
            sdfg.dump(output_folder, "py0.parsed", dump_dot=True)

        # Enable statistics if envvar is set
        if _statistics_enabled_by_env():
            _enable_statistics()

        sdfg.validate()

        # Tensor targets keep tensor nodes
        if self.target != "onnx":
            custom_expand_fn = get_target_expand_fn(self.target)
            if custom_expand_fn is not None:
                custom_expand_fn(sdfg, self.category, {})
            else:
                sdfg.expand()
            if self.debug_dump:
                sdfg.dump(output_folder, "py1.expanded", dump_dot=True)

        # Simplify pipelines
        sdfg.simplify()
        if self.debug_dump:
            sdfg.dump(output_folder, "py2.opt", dump_dot=True)

        # Normalization for scheduling
        if self.target != "none":
            sdfg.normalize()

        if self.debug_dump or instrumentation_mode or capture_args:
            sdfg.dump(
                output_folder,
                "py3.norm",
                dump_dot=self.debug_dump,
                dump_json=True,
                record_for_instrumentation=True,
            )

        # Schedule if target is specified

        if self.target != "none":
            # Check for custom registered target first
            custom_schedule_fn = get_target_schedule_fn(self.target)
            if custom_schedule_fn is not None:
                custom_schedule_fn(
                    sdfg, self.category, {"remote_tuning": self.remote_tuning}
                )
            else:
                sdfg.schedule(self.target, self.category, self.remote_tuning)

            if self.debug_dump:
                sdfg.dump(output_folder, "py4.post_sched", dump_dot=True)

        self.last_sdfg = sdfg

        # Dump statistics before compile
        if _statistics_enabled_by_env():
            print(_statistics_summary())

        custom_compile_fn = get_target_compile_fn(self.target)
        if custom_compile_fn is not None:
            lib_path = custom_compile_fn(
                sdfg, output_folder, instrumentation_mode, capture_args, {}
            )
        else:
            lib_path = sdfg._compile(
                output_folder=output_folder,
                target=self.target,
                instrumentation_mode=instrumentation_mode,
                capture_args=capture_args,
            )

        return lib_path

    @abstractmethod
    def to_sdfg(self, *args: Any) -> StructuredSDFG:
        pass

    @abstractmethod
    def _convert_inputs(self, args: tuple) -> tuple:
        pass

    @abstractmethod
    def _convert_outputs(self, result: Any, original_args: tuple) -> Any:
        pass

    def _get_cache_key(self, *args: Any) -> str:
        return ""

1	from abc import ABC, abstractmethod	4✔
2	from typing import Any, Optional	4✔
3	import os	4✔
4
5	from docc.sdfg import StructuredSDFG	4✔
6	from docc.sdfg._sdfg import (	4✔
7	_enable_statistics,
8	_statistics_enabled_by_env,
9	_statistics_summary,
10	)
11	from docc.compiler.compiled_sdfg import CompiledSDFG	4✔
12	from docc.compiler.target_registry import (	4✔
13	get_target_schedule_fn,
14	get_target_compile_fn,
15	get_target_expand_fn,
16	)
17
18
19	def _is_debug_dump() -> bool:	4✔
20	return bool(os.environ.get("DOCC_DEBUG"))	4✔
21
22
23	class DoccProgram(ABC):	4✔
24
25	def __init__(	4✔
26	self,
27	name: str,
28	target: str = "none",
29	category: str = "server",
30	instrumentation_mode: Optional[str] = None,
31	capture_args: Optional[bool] = None,
32	remote_tuning: bool = False,
33	):
34	self.name = name	4✔
35	self.target = target	4✔
36	self.category = category	4✔
37	self.remote_tuning = remote_tuning	4✔
38	self.last_sdfg: Optional[StructuredSDFG] = None	4✔
39	self.cache: dict = {}	4✔
40	self.debug_dump: bool = _is_debug_dump()	4✔
41
42	# Check environment variable DOCC_CI
43	docc_ci = os.environ.get("DOCC_CI", "")	4✔
44	if docc_ci:	4✔
45	if docc_ci == "regions":	×
46	if instrumentation_mode is None:	×
47	instrumentation_mode = "ols"	×
48	elif docc_ci == "arg-capture":	×
49	if capture_args is None:	×
50	capture_args = True	×
51	else:
52	# Full mode (or unknown value treated as full)
53	if instrumentation_mode is None:	×
54	instrumentation_mode = "ols"	×
55	if capture_args is None:	×
56	capture_args = True	×
57
58	self.instrumentation_mode = instrumentation_mode	4✔
59	self.capture_args = capture_args	4✔
60
61	@abstractmethod	4✔
62	def __call__(self, *args: Any) -> Any:	4✔
63	pass	×
64
65	@abstractmethod	4✔
66	def compile(self, *args: Any, output_folder: Optional[str] = None) -> CompiledSDFG:	4✔
67	pass	×
68
69	def sdfg_pipe(	4✔
70	self,
71	sdfg: StructuredSDFG,
72	output_folder: Optional[str],
73	instrumentation_mode: str,
74	capture_args: bool,
75	) -> str:
76
77	if self.debug_dump:	4✔
78	sdfg.dump(output_folder, "py0.parsed", dump_dot=True)	×
79
80	# Enable statistics if envvar is set
81	if _statistics_enabled_by_env():	4✔
NEW 82	_enable_statistics()	×
83
84	sdfg.validate()	4✔
85
86	# Tensor targets keep tensor nodes
87	if self.target != "onnx":	4✔
88	custom_expand_fn = get_target_expand_fn(self.target)	4✔
89	if custom_expand_fn is not None:	4✔
90	custom_expand_fn(sdfg, self.category, {})	4✔
91	else:
92	sdfg.expand()	4✔
93	if self.debug_dump:	4✔
94	sdfg.dump(output_folder, "py1.expanded", dump_dot=True)	×
95
96	# Simplify pipelines
97	sdfg.simplify()	4✔
98	if self.debug_dump:	4✔
99	sdfg.dump(output_folder, "py2.opt", dump_dot=True)	×
100
101	# Normalization for scheduling
102	if self.target != "none":	4✔
103	sdfg.normalize()	4✔
104
105	if self.debug_dump or instrumentation_mode or capture_args:	4✔
106	sdfg.dump(	4✔
107	output_folder,
108	"py3.norm",
109	dump_dot=self.debug_dump,
110	dump_json=True,
111	record_for_instrumentation=True,
112	)
113
114	# Schedule if target is specified
115
116	if self.target != "none":	4✔
117	# Check for custom registered target first
118	custom_schedule_fn = get_target_schedule_fn(self.target)	4✔
119	if custom_schedule_fn is not None:	4✔
120	custom_schedule_fn(	4✔
121	sdfg, self.category, {"remote_tuning": self.remote_tuning}
122	)
123	else:
124	sdfg.schedule(self.target, self.category, self.remote_tuning)	4✔
125
126	if self.debug_dump:	4✔
127	sdfg.dump(output_folder, "py4.post_sched", dump_dot=True)	×
128
129	self.last_sdfg = sdfg	4✔
130
131	# Dump statistics before compile
132	if _statistics_enabled_by_env():	4✔
NEW 133	print(_statistics_summary())	×
134
135	custom_compile_fn = get_target_compile_fn(self.target)	4✔
136	if custom_compile_fn is not None:	4✔
137	lib_path = custom_compile_fn(	4✔
138	sdfg, output_folder, instrumentation_mode, capture_args, {}
139	)
140	else:
141	lib_path = sdfg._compile(	4✔
142	output_folder=output_folder,
143	target=self.target,
144	instrumentation_mode=instrumentation_mode,
145	capture_args=capture_args,
146	)
147
148	return lib_path	4✔
149
150	@abstractmethod	4✔
151	def to_sdfg(self, *args: Any) -> StructuredSDFG:	4✔
152	pass	×
153
154	@abstractmethod	4✔
155	def _convert_inputs(self, args: tuple) -> tuple:	4✔
156	pass	×
157
158	@abstractmethod	4✔
159	def _convert_outputs(self, result: Any, original_args: tuple) -> Any:	4✔
160	pass	×
161
162	def _get_cache_key(self, *args: Any) -> str:	4✔
163	return ""	×

daisytuner / docc / 23552450315

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