29187058583

Committed 12 Jul 2026 09:11AM UTC coverage: 51.324% (-1.4%) from 52.754%

Build # 29187058583

Build Type

Pull #96

github

Committed by

web-flow

Commit Message

Merge 7e90b76dd into 056c79287

Pull Request Pull Request #96: fix: consolidate stale-index rebuilds across consumers and fix multi-run breaks

Coverage Stats

33807 of 84432 branches covered (40.04%)

Branch coverage included in aggregate %.

145 of 152 new or added lines in 16 files covered. (95.39%)

5186 existing lines in 197 files now uncovered.

34558 of 48770 relevant lines covered (70.86%)

10797.15 hits per line

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

59.76

/src/dftracer/utils/python/utilities/statistics_aggregator.cpp

#define PY_SSIZE_T_CLEAN
#include <dftracer/utils/core/coro/task.h>
#include <dftracer/utils/core/runtime.h>
#include <dftracer/utils/core/tasks/coro_scope.h>
#include <dftracer/utils/python/py_dict_helpers.h>
#include <dftracer/utils/python/py_runtime_mixin.h>
#include <dftracer/utils/python/py_type_helpers.h>
#include <dftracer/utils/python/runtime.h>
#include <dftracer/utils/python/utilities/statistics_aggregator.h>
#include <dftracer/utils/utilities/composites/dft/indexing/resolve_and_build.h>
#include <dftracer/utils/utilities/composites/dft/internal/utils.h>
#include <dftracer/utils/utilities/composites/dft/statistics/statistics_aggregator_utility.h>
#include <dftracer/utils/utilities/composites/dft/statistics/trace_statistics.h>

#include <string>

using dftracer::utils::CoroScope;
using dftracer::utils::run_coro_scope;
using dftracer::utils::Runtime;
using dftracer::utils::coro::CoroTask;
using namespace dftracer::utils::utilities::composites::dft::statistics;
namespace indexing = dftracer::utils::utilities::composites::dft::indexing;

DFTRACER_UTILS_RUNTIME_BACKED_SLOTS(StatisticsAggregator,
                                    StatisticsAggregatorObject)

static PyObject *StatisticsAggregator_compute(StatisticsAggregatorObject *self,
                                              PyObject *args, PyObject *kwds) {
    static const char *kwlist[] = {"file_path", "index_dir", NULL};
    const char *file_path;
    const char *index_dir = "";
    if (!PyArg_ParseTupleAndKeywords(args, kwds, "s|s", (char **)kwlist,
                                     &file_path, &index_dir))
        return NULL;

    std::string file_path_str(file_path);
    std::string index_dir_str(index_dir);
    TraceStatistics stats;

    if (!run_blocking([&] {
            Runtime *rt = resolve_runtime(self);

            StatisticsAggregatorInput input;
            input.file_path = file_path_str;
            input.index_dir = index_dir_str;
            input.index_path = dftracer::utils::utilities::composites::dft::
                internal::determine_index_path(file_path_str, index_dir_str);

            rt->submit(
                  run_coro_scope(rt->executor(),
                                 [file_path_str, index_dir_str](
                                     CoroScope &scope) -> CoroTask<void> {
                                     co_await indexing::ensure_indexes_fresh(
                                         &scope, "", {file_path_str},
                                         index_dir_str, false);
                                 }),
                  "stats-ensure-fresh")
                .get();

            auto *stats_p = &stats;
            auto input_copy = input;
            auto task = [stats_p, input_copy]() -> CoroTask<void> {
                StatisticsAggregatorUtility util;
                *stats_p = co_await util.process(input_copy);
            };
            rt->submit(task(), "stats-aggregator").get();
        })) {
        return NULL;
    }

    PyObject *d = PyDict_New();
    if (!d) return NULL;

    int rc = 0;
    rc |= dict_set_str(d, "file_path", stats.file_path.c_str());
    rc |= dict_set_u64(d, "total_events", stats.total_events());
    rc |= dict_set_u64(d, "num_chunks", stats.num_chunks);
    rc |= dict_set_bool(d, "success", stats.success);
    rc |= dict_set_str(d, "error_message", stats.error_message.c_str());
    rc |= dict_set_f64(d, "time_span_seconds", stats.time_span_seconds());
    rc |= dict_set_f64(d, "duration_mean_us", stats.duration_mean_us());
    rc |= dict_set_f64(d, "duration_stddev_us", stats.duration_stddev_us());
    rc |= dict_set_size(d, "num_categories", stats.num_categories());
    rc |= dict_set_size(d, "num_unique_names", stats.num_unique_names());
    rc |= dict_set_size(d, "num_pid_tids", stats.num_pid_tids());
    rc |= dict_set_u64(d, "min_timestamp_us", stats.merged.min_timestamp_us);
    rc |= dict_set_u64(d, "max_timestamp_us", stats.merged.max_timestamp_us);

    if (rc != 0) {
        Py_DECREF(d);
        return NULL;
    }

    return d;
}

static PyObject *StatisticsAggregator_call(PyObject *self, PyObject *args,
                                           PyObject *kwds) {
    return StatisticsAggregator_compute((StatisticsAggregatorObject *)self,
                                        args, kwds);
}

static PyMethodDef StatisticsAggregator_methods[] = {
    {"process", (PyCFunction)StatisticsAggregator_compute,
     METH_VARARGS | METH_KEYWORDS,
     "process(file_path, index_dir='')\n"
     "--\n"
     "\n"
     "Compute aggregated statistics from a trace file.\n"
     "\n"
     "Args:\n"
     "    file_path (str): Path to the trace file.\n"
     "    index_dir (str): Directory for .dftindex stores (default '').\n"
     "\n"
     "Returns:\n"
     "    dict: Aggregated statistics.\n"},
    {NULL}};

PyTypeObject StatisticsAggregatorType = {
    PyVarObject_HEAD_INIT(
        NULL, 0) "dftracer_utils_ext.StatisticsAggregatorUtility", /* tp_name */
    sizeof(StatisticsAggregatorObject),       /* tp_basicsize */
    0,                                        /* tp_itemsize */
    (destructor)StatisticsAggregator_dealloc, /* tp_dealloc */
    0,                                        /* tp_vectorcall_offset */
    0,                                        /* tp_getattr */
    0,                                        /* tp_setattr */
    0,                                        /* tp_as_async */
    0,                                        /* tp_repr */
    0,                                        /* tp_as_number */
    0,                                        /* tp_as_sequence */
    0,                                        /* tp_as_mapping */
    0,                                        /* tp_hash */
    StatisticsAggregator_call,                /* tp_call */
    0,                                        /* tp_str */
    0,                                        /* tp_getattro */
    0,                                        /* tp_setattro */
    0,                                        /* tp_as_buffer */
    Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE, /* tp_flags */
    "StatisticsAggregatorUtility(runtime: Runtime | None = None)\n"
    "--\n\n"
    "Aggregate statistics from an indexed trace file.\n\n"
    "Args:\n"
    "    runtime (Runtime or None): Runtime for thread pool control.\n",
    0,                                   /* tp_traverse */
    0,                                   /* tp_clear */
    0,                                   /* tp_richcompare */
    0,                                   /* tp_weaklistoffset */
    0,                                   /* tp_iter */
    0,                                   /* tp_iternext */
    StatisticsAggregator_methods,        /* tp_methods */
    0,                                   /* tp_members */
    0,                                   /* tp_getset */
    0,                                   /* tp_base */
    0,                                   /* tp_dict */
    0,                                   /* tp_descr_get */
    0,                                   /* tp_descr_set */
    0,                                   /* tp_dictoffset */
    (initproc)StatisticsAggregator_init, /* tp_init */
    0,                                   /* tp_alloc */
    StatisticsAggregator_new,            /* tp_new */
};

int init_statistics_aggregator(PyObject *m) {
    if (register_type(m, &StatisticsAggregatorType,
                      "StatisticsAggregatorUtility") < 0)
        return -1;

    return 0;
}

1	#define PY_SSIZE_T_CLEAN
2	#include <dftracer/utils/core/coro/task.h>
3	#include <dftracer/utils/core/runtime.h>
4	#include <dftracer/utils/core/tasks/coro_scope.h>
5	#include <dftracer/utils/python/py_dict_helpers.h>
6	#include <dftracer/utils/python/py_runtime_mixin.h>
7	#include <dftracer/utils/python/py_type_helpers.h>
8	#include <dftracer/utils/python/runtime.h>
9	#include <dftracer/utils/python/utilities/statistics_aggregator.h>
10	#include <dftracer/utils/utilities/composites/dft/indexing/resolve_and_build.h>
11	#include <dftracer/utils/utilities/composites/dft/internal/utils.h>
12	#include <dftracer/utils/utilities/composites/dft/statistics/statistics_aggregator_utility.h>
13	#include <dftracer/utils/utilities/composites/dft/statistics/trace_statistics.h>
14
15	#include <string>
16
17	using dftracer::utils::CoroScope;
18	using dftracer::utils::run_coro_scope;
19	using dftracer::utils::Runtime;
20	using dftracer::utils::coro::CoroTask;
21	using namespace dftracer::utils::utilities::composites::dft::statistics;
22	namespace indexing = dftracer::utils::utilities::composites::dft::indexing;
23
24	DFTRACER_UTILS_RUNTIME_BACKED_SLOTS(StatisticsAggregator,	21✔
25	StatisticsAggregatorObject)
26
27	static PyObject StatisticsAggregator_compute(StatisticsAggregatorObject self,	7✔
28	PyObject args, PyObject kwds) {
29	static const char *kwlist[] = {"file_path", "index_dir", NULL};
30	const char *file_path;
31	const char *index_dir = "";	7✔
32	if (!PyArg_ParseTupleAndKeywords(args, kwds, "s\|s", (char **)kwlist,	7!
33	&file_path, &index_dir))
34	return NULL;	×
35
36	std::string file_path_str(file_path);	7✔
37	std::string index_dir_str(index_dir);	7!
38	TraceStatistics stats;	7!
39
40	if (!run_blocking([&] {	14!
41	Runtime *rt = resolve_runtime(self);	7✔
42
43	StatisticsAggregatorInput input;	7✔
44	input.file_path = file_path_str;	7!
45	input.index_dir = index_dir_str;	7!
46	input.index_path = dftracer::utils::utilities::composites::dft::	7!
47	internal::determine_index_path(file_path_str, index_dir_str);	7✔
48
49	rt->submit(	21!
50	run_coro_scope(rt->executor(),	14!
51	[file_path_str, index_dir_str](	56!
52	CoroScope &scope) -> CoroTask<void> {	7!
53	co_await indexing::ensure_indexes_fresh(	56!
54	&scope, "", {file_path_str},	21!
55	index_dir_str, false);	21✔
56	}),	14✔
57	"stats-ensure-fresh")	7!
58	.get();	7!
59
60	auto *stats_p = &stats;	7✔
61	auto input_copy = input;	7!
62	auto task = [stats_p, input_copy]() -> CoroTask<void> {	56!
63	StatisticsAggregatorUtility util;	21!
64	*stats_p = co_await util.process(input_copy);	28!
65	};	21!
66	rt->submit(task(), "stats-aggregator").get();	7!
67	})) {	7✔
68	return NULL;	×
69	}
70
71	PyObject *d = PyDict_New();	7!
72	if (!d) return NULL;	7!
73
74	int rc = 0;	7✔
75	rc \|= dict_set_str(d, "file_path", stats.file_path.c_str());	7!
76	rc \|= dict_set_u64(d, "total_events", stats.total_events());	7!
77	rc \|= dict_set_u64(d, "num_chunks", stats.num_chunks);	7!
78	rc \|= dict_set_bool(d, "success", stats.success);	7!
79	rc \|= dict_set_str(d, "error_message", stats.error_message.c_str());	7!
80	rc \|= dict_set_f64(d, "time_span_seconds", stats.time_span_seconds());	7!
81	rc \|= dict_set_f64(d, "duration_mean_us", stats.duration_mean_us());	7!
82	rc \|= dict_set_f64(d, "duration_stddev_us", stats.duration_stddev_us());	7!
83	rc \|= dict_set_size(d, "num_categories", stats.num_categories());	7!
84	rc \|= dict_set_size(d, "num_unique_names", stats.num_unique_names());	7!
85	rc \|= dict_set_size(d, "num_pid_tids", stats.num_pid_tids());	7!
86	rc \|= dict_set_u64(d, "min_timestamp_us", stats.merged.min_timestamp_us);	7!
87	rc \|= dict_set_u64(d, "max_timestamp_us", stats.merged.max_timestamp_us);	7!
88
89	if (rc != 0) {	7!
UNCOV 90	Py_DECREF(d);	×
91	return NULL;	×
92	}
93
94	return d;	7✔
95	}	7✔
96
97	static PyObject StatisticsAggregator_call(PyObject self, PyObject *args,	1✔
98	PyObject *kwds) {
99	return StatisticsAggregator_compute((StatisticsAggregatorObject *)self,	2✔
100	args, kwds);	1✔
101	}
102
103	static PyMethodDef StatisticsAggregator_methods[] = {
104	{"process", (PyCFunction)StatisticsAggregator_compute,
105	METH_VARARGS \| METH_KEYWORDS,
106	"process(file_path, index_dir='')\n"
107	"--\n"
108	"\n"
109	"Compute aggregated statistics from a trace file.\n"
110	"\n"
111	"Args:\n"
112	" file_path (str): Path to the trace file.\n"
113	" index_dir (str): Directory for .dftindex stores (default '').\n"
114	"\n"
115	"Returns:\n"
116	" dict: Aggregated statistics.\n"},
117	{NULL}};
118
119	PyTypeObject StatisticsAggregatorType = {
120	PyVarObject_HEAD_INIT(
121	NULL, 0) "dftracer_utils_ext.StatisticsAggregatorUtility", /* tp_name */
122	sizeof(StatisticsAggregatorObject), /* tp_basicsize */
123	0, /* tp_itemsize */
124	(destructor)StatisticsAggregator_dealloc, /* tp_dealloc */
125	0, /* tp_vectorcall_offset */
126	0, /* tp_getattr */
127	0, /* tp_setattr */
128	0, /* tp_as_async */
129	0, /* tp_repr */
130	0, /* tp_as_number */
131	0, /* tp_as_sequence */
132	0, /* tp_as_mapping */
133	0, /* tp_hash */
134	StatisticsAggregator_call, /* tp_call */
135	0, /* tp_str */
136	0, /* tp_getattro */
137	0, /* tp_setattro */
138	0, /* tp_as_buffer */
139	Py_TPFLAGS_DEFAULT \| Py_TPFLAGS_BASETYPE, /* tp_flags */
140	"StatisticsAggregatorUtility(runtime: Runtime \| None = None)\n"
141	"--\n\n"
142	"Aggregate statistics from an indexed trace file.\n\n"
143	"Args:\n"
144	" runtime (Runtime or None): Runtime for thread pool control.\n",
145	0, /* tp_traverse */
146	0, /* tp_clear */
147	0, /* tp_richcompare */
148	0, /* tp_weaklistoffset */
149	0, /* tp_iter */
150	0, /* tp_iternext */
151	StatisticsAggregator_methods, /* tp_methods */
152	0, /* tp_members */
153	0, /* tp_getset */
154	0, /* tp_base */
155	0, /* tp_dict */
156	0, /* tp_descr_get */
157	0, /* tp_descr_set */
158	0, /* tp_dictoffset */
159	(initproc)StatisticsAggregator_init, /* tp_init */
160	0, /* tp_alloc */
161	StatisticsAggregator_new, /* tp_new */
162	};
163
164	int init_statistics_aggregator(PyObject *m) {	1✔
165	if (register_type(m, &StatisticsAggregatorType,	1✔
166	"StatisticsAggregatorUtility") < 0)	1!
167	return -1;	×
168
169	return 0;	1✔
170	}	1✔

llnl / dftracer-utils / 29187058583

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