23529483807

Committed 25 Mar 2026 07:17AM UTC coverage: 48.515% (-1.6%) from 50.098%

Build # 23529483807

Build Type

Pull #57

github

Committed by

web-flow

Commit Message

Merge 5b1e117ad into 38f9f3616

Pull Request Pull Request #57: feat(comparator): add pairwise traces comparator

Coverage Stats

18829 of 49412 branches covered (38.11%)

Branch coverage included in aggregate %.

1584 of 1933 new or added lines in 14 files covered. (81.95%)

3552 existing lines in 135 files now uncovered.

18474 of 27477 relevant lines covered (67.23%)

241072.53 hits per line

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

48.36

/src/dftracer/utils/python/runtime.cpp

#define PY_SSIZE_T_CLEAN
#include <Python.h>
#include <dftracer/utils/python/runtime.h>

#include <chrono>
#include <memory>

static std::shared_ptr<dftracer::utils::Runtime> g_default_runtime;

dftracer::utils::Runtime *get_default_runtime() {
    if (!g_default_runtime) {
        g_default_runtime = std::make_shared<dftracer::utils::Runtime>(0);
    }
    return g_default_runtime.get();
}

static void Runtime_dealloc(RuntimeObject *self) {
    self->runtime.reset();
    Py_TYPE(self)->tp_free((PyObject *)self);
}

static PyObject *Runtime_new(PyTypeObject *type, PyObject *args,
                             PyObject *kwds) {
    RuntimeObject *self = (RuntimeObject *)type->tp_alloc(type, 0);
    if (self) {
        // Placement-new the shared_ptr (tp_alloc gives raw memory)
        new (&self->runtime) std::shared_ptr<dftracer::utils::Runtime>(nullptr);
    }
    return (PyObject *)self;
}

static int Runtime_init(RuntimeObject *self, PyObject *args, PyObject *kwds) {
    static const char *kwlist[] = {"threads", NULL};
    Py_ssize_t threads = 0;

    if (!PyArg_ParseTupleAndKeywords(args, kwds, "|n", (char **)kwlist,
                                     &threads)) {
        return -1;
    }

    if (threads < 0) {
        PyErr_SetString(PyExc_ValueError, "threads must be >= 0");
        return -1;
    }

    try {
        self->runtime = std::make_shared<dftracer::utils::Runtime>(
            static_cast<std::size_t>(threads));
    } catch (const std::exception &e) {
        PyErr_SetString(PyExc_RuntimeError, e.what());
        return -1;
    }

    return 0;
}

static PyObject *Runtime_shutdown(RuntimeObject *self,
                                  PyObject *Py_UNUSED(ignored)) {
    if (!self->runtime) {
        PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");
        return NULL;
    }
    Py_BEGIN_ALLOW_THREADS self->runtime->shutdown();
    Py_END_ALLOW_THREADS Py_RETURN_NONE;
}

static bool set_size(PyObject *d, const char *key, std::size_t val) {
    PyObject *v = PyLong_FromSize_t(val);
    if (!v) return false;
    int rc = PyDict_SetItemString(d, key, v);
    Py_DECREF(v);
    return rc == 0;
}

static bool set_double(PyObject *d, const char *key, double val) {
    PyObject *v = PyFloat_FromDouble(val);
    if (!v) return false;
    int rc = PyDict_SetItemString(d, key, v);
    Py_DECREF(v);
    return rc == 0;
}

static bool set_str(PyObject *d, const char *key, const std::string &val) {
    PyObject *v = PyUnicode_FromStringAndSize(
        val.data(), static_cast<Py_ssize_t>(val.size()));
    if (!v) return false;
    int rc = PyDict_SetItemString(d, key, v);
    Py_DECREF(v);
    return rc == 0;
}

static bool set_bool(PyObject *d, const char *key, bool val) {
    PyObject *v = val ? Py_True : Py_False;
    Py_INCREF(v);
    int rc = PyDict_SetItemString(d, key, v);
    Py_DECREF(v);
    return rc == 0;
}

static PyObject *build_task_progress(const dftracer::utils::TaskProgress &tp) {
    PyObject *td = PyDict_New();
    if (!td) return NULL;

    if (!set_str(td, "name", tp.name) || !set_str(td, "state", tp.state) ||
        !set_double(td, "queued_duration_ms", tp.queued_duration_ms) ||
        !set_double(td, "execution_duration_ms", tp.execution_duration_ms) ||
        !set_size(td, "total_subtasks", tp.total_subtasks) ||
        !set_size(td, "completed_subtasks", tp.completed_subtasks) ||
        !set_double(td, "progress_pct", tp.progress_percentage) ||
        !set_str(td, "location", tp.location)) {
        Py_DECREF(td);
        return NULL;
    }

    PyObject *children =
        PyList_New(static_cast<Py_ssize_t>(tp.children.size()));
    if (!children) {
        Py_DECREF(td);
        return NULL;
    }
    for (std::size_t i = 0; i < tp.children.size(); ++i) {
        PyObject *child = build_task_progress(tp.children[i]);
        if (!child) {
            Py_DECREF(children);
            Py_DECREF(td);
            return NULL;
        }
        PyList_SET_ITEM(children, static_cast<Py_ssize_t>(i), child);
    }
    if (PyDict_SetItemString(td, "children", children) < 0) {
        Py_DECREF(children);
        Py_DECREF(td);
        return NULL;
    }
    Py_DECREF(children);
    return td;
}

static PyObject *Runtime_get_progress(RuntimeObject *self,
                                      PyObject *Py_UNUSED(ignored)) {
    if (!self->runtime) {
        PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");
        return NULL;
    }

    dftracer::utils::ExecutorProgress prog;
    Py_BEGIN_ALLOW_THREADS prog = self->runtime->get_progress();
    Py_END_ALLOW_THREADS

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

    if (!set_size(d, "total", prog.total_tasks_submitted) ||
        !set_size(d, "completed", prog.tasks_completed) ||
        !set_size(d, "running", prog.tasks_running) ||
        !set_size(d, "queued", prog.tasks_queued) ||
        !set_size(d, "failed", prog.tasks_failed)) {
        Py_DECREF(d);
        return NULL;
    }

    // Workers
    PyObject *workers =
        PyList_New(static_cast<Py_ssize_t>(prog.workers.size()));
    if (!workers) {
        Py_DECREF(d);
        return NULL;
    }
    for (std::size_t i = 0; i < prog.workers.size(); ++i) {
        const auto &w = prog.workers[i];
        PyObject *wd = PyDict_New();
        if (!wd || !set_size(wd, "id", w.worker_id) ||
            !set_bool(wd, "idle", w.is_idle) ||
            !set_str(wd, "task", w.current_task_name) ||
            !set_size(wd, "queue_depth", w.local_queue_depth)) {
            Py_XDECREF(wd);
            Py_DECREF(workers);
            Py_DECREF(d);
            return NULL;
        }
        PyList_SET_ITEM(workers, static_cast<Py_ssize_t>(i), wd);
    }
    if (PyDict_SetItemString(d, "workers", workers) < 0) {
        Py_DECREF(workers);
        Py_DECREF(d);
        return NULL;
    }
    Py_DECREF(workers);

    // Tasks
    PyObject *tasks =
        PyList_New(static_cast<Py_ssize_t>(prog.root_tasks.size()));
    if (!tasks) {
        Py_DECREF(d);
        return NULL;
    }
    for (std::size_t i = 0; i < prog.root_tasks.size(); ++i) {
        PyObject *tp = build_task_progress(prog.root_tasks[i]);
        if (!tp) {
            Py_DECREF(tasks);
            Py_DECREF(d);
            return NULL;
        }
        PyList_SET_ITEM(tasks, static_cast<Py_ssize_t>(i), tp);
    }
    if (PyDict_SetItemString(d, "tasks", tasks) < 0) {
        Py_DECREF(tasks);
        Py_DECREF(d);
        return NULL;
    }
    Py_DECREF(tasks);

    // Errors
    PyObject *errors =
        PyList_New(static_cast<Py_ssize_t>(prog.recent_errors.size()));
    if (!errors) {
        Py_DECREF(d);
        return NULL;
    }
    for (std::size_t i = 0; i < prog.recent_errors.size(); ++i) {
        const auto &[tid, msg] = prog.recent_errors[i];
        PyObject *ed = PyDict_New();
        if (!ed || !set_size(ed, "task_id", static_cast<std::size_t>(tid)) ||
            !set_str(ed, "message", msg)) {
            Py_XDECREF(ed);
            Py_DECREF(errors);
            Py_DECREF(d);
            return NULL;
        }
        PyList_SET_ITEM(errors, static_cast<Py_ssize_t>(i), ed);
    }
    if (PyDict_SetItemString(d, "errors", errors) < 0) {
        Py_DECREF(errors);
        Py_DECREF(d);
        return NULL;
    }
    Py_DECREF(errors);

    return d;
}

static PyObject *Runtime_is_responsive(RuntimeObject *self,
                                       PyObject *Py_UNUSED(ignored)) {
    if (!self->runtime) {
        PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");
        return NULL;
    }
    bool resp;
    Py_BEGIN_ALLOW_THREADS resp = self->runtime->is_responsive();
    Py_END_ALLOW_THREADS return PyBool_FromLong(resp ? 1 : 0);
}

static PyObject *Runtime_set_timeout(RuntimeObject *self, PyObject *args,
                                     PyObject *kwds) {
    static const char *kwlist[] = {"global_ms", NULL};
    Py_ssize_t ms = 0;

    if (!PyArg_ParseTupleAndKeywords(args, kwds, "|n", (char **)kwlist, &ms)) {
        return NULL;
    }

    if (!self->runtime) {
        PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");
        return NULL;
    }

    Py_BEGIN_ALLOW_THREADS self->runtime->set_global_timeout(
        std::chrono::milliseconds(ms));
    Py_END_ALLOW_THREADS Py_RETURN_NONE;
}

static PyObject *Runtime_set_default_task_timeout(RuntimeObject *self,
                                                  PyObject *args,
                                                  PyObject *kwds) {
    static const char *kwlist[] = {"ms", NULL};
    Py_ssize_t ms = 0;

    if (!PyArg_ParseTupleAndKeywords(args, kwds, "|n", (char **)kwlist, &ms)) {
        return NULL;
    }

    if (!self->runtime) {
        PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");
        return NULL;
    }

    Py_BEGIN_ALLOW_THREADS self->runtime->set_default_task_timeout(
        std::chrono::milliseconds(ms));
    Py_END_ALLOW_THREADS Py_RETURN_NONE;
}

static PyObject *Runtime_wait_all(RuntimeObject *self,
                                  PyObject *Py_UNUSED(ignored)) {
    if (!self->runtime) {
        PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");
        return NULL;
    }
    try {
        Py_BEGIN_ALLOW_THREADS self->runtime->wait_all();
        Py_END_ALLOW_THREADS
    } catch (const std::exception &e) {
        PyErr_SetString(PyExc_RuntimeError, e.what());
        return NULL;
    }
    Py_RETURN_NONE;
}

static PyObject *Runtime_enter(RuntimeObject *self,
                               PyObject *Py_UNUSED(ignored)) {
    Py_INCREF(self);
    return (PyObject *)self;
}

static PyObject *Runtime_exit(RuntimeObject *self, PyObject *args) {
    if (self->runtime) {
        Py_BEGIN_ALLOW_THREADS self->runtime->shutdown();
        Py_END_ALLOW_THREADS
    }
    Py_RETURN_NONE;
}

static PyObject *Runtime_get_threads(RuntimeObject *self, void *closure) {
    if (!self->runtime) {
        PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");
        return NULL;
    }
    return PyLong_FromSize_t(self->runtime->threads());
}

static PyObject *get_default_runtime_py(PyObject *Py_UNUSED(module),
                                        PyObject *Py_UNUSED(ignored)) {
    dftracer::utils::Runtime *rt = get_default_runtime();
    if (!rt) {
        PyErr_SetString(PyExc_RuntimeError, "Failed to create default runtime");
        return NULL;
    }

    RuntimeObject *obj = (RuntimeObject *)RuntimeType.tp_alloc(&RuntimeType, 0);
    if (!obj) return NULL;

    new (&obj->runtime)
        std::shared_ptr<dftracer::utils::Runtime>(g_default_runtime);
    return (PyObject *)obj;
}

static PyObject *set_default_runtime_py(PyObject *Py_UNUSED(module),
                                        PyObject *args) {
    PyObject *arg;
    if (!PyArg_ParseTuple(args, "O", &arg)) return NULL;

    if (arg == Py_None) {
        g_default_runtime.reset();
        Py_RETURN_NONE;
    }

    if (!PyObject_TypeCheck(arg, &RuntimeType)) {
        PyErr_SetString(PyExc_TypeError, "Expected Runtime or None");
        return NULL;
    }

    g_default_runtime = ((RuntimeObject *)arg)->runtime;
    Py_RETURN_NONE;
}

static PyMethodDef Runtime_methods[] = {
    {"shutdown", (PyCFunction)Runtime_shutdown, METH_NOARGS,
     "shutdown()\n"
     "--\n"
     "\n"
     "Shut down the runtime.\n"},
    {"get_progress", (PyCFunction)Runtime_get_progress, METH_NOARGS,
     "Return progress dict with keys: total, completed, running,\n"
     "queued, failed."},
    {"is_responsive", (PyCFunction)Runtime_is_responsive, METH_NOARGS,
     "Return True if the runtime is making progress."},
    {"set_timeout", (PyCFunction)Runtime_set_timeout,
     METH_VARARGS | METH_KEYWORDS,
     "Set global timeout in milliseconds.\n"
     "\n"
     "Args:\n"
     "    global_ms (int): Timeout in milliseconds (0 = no timeout).\n"},
    {"set_default_task_timeout", (PyCFunction)Runtime_set_default_task_timeout,
     METH_VARARGS | METH_KEYWORDS,
     "Set default per-task timeout in milliseconds.\n"
     "\n"
     "Args:\n"
     "    ms (int): Timeout in milliseconds (0 = no timeout).\n"},
    {"wait_all", (PyCFunction)Runtime_wait_all, METH_NOARGS,
     "Wait for all outstanding submitted tasks to complete."},
    {"__enter__", (PyCFunction)Runtime_enter, METH_NOARGS,
     "Enter context manager."},
    {"__exit__", (PyCFunction)Runtime_exit, METH_VARARGS,
     "Exit context manager (calls shutdown)."},
    {NULL}};

static PyGetSetDef Runtime_getsetters[] = {
    {"threads", (getter)Runtime_get_threads, NULL, "Number of worker threads",
     NULL},
    {NULL}};

PyTypeObject RuntimeType = {
    PyVarObject_HEAD_INIT(NULL, 0) "dftracer_utils_ext.Runtime",
    sizeof(RuntimeObject),                    /* tp_basicsize */
    0,                                        /* tp_itemsize */
    (destructor)Runtime_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 */
    0,                                        /* tp_call */
    0,                                        /* tp_str */
    0,                                        /* tp_getattro */
    0,                                        /* tp_setattro */
    0,                                        /* tp_as_buffer */
    Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE, /* tp_flags */
    "Runtime(threads: int = 0)\n"
    "--\n"
    "\n"
    "Coroutine runtime backed by a thread pool.\n"
    "\n"
    "Args:\n"
    "    threads (int): Number of worker threads. 0 (default) uses\n"
    "        the hardware concurrency.\n", /* tp_doc */
    0,                                     /* tp_traverse */
    0,                                     /* tp_clear */
    0,                                     /* tp_richcompare */
    0,                                     /* tp_weaklistoffset */
    0,                                     /* tp_iter */
    0,                                     /* tp_iternext */
    Runtime_methods,                       /* tp_methods */
    0,                                     /* tp_members */
    Runtime_getsetters,                    /* tp_getset */
    0,                                     /* tp_base */
    0,                                     /* tp_dict */
    0,                                     /* tp_descr_get */
    0,                                     /* tp_descr_set */
    0,                                     /* tp_dictoffset */
    (initproc)Runtime_init,                /* tp_init */
    0,                                     /* tp_alloc */
    Runtime_new,                           /* tp_new */
};

// Module-level function table (registered via PyModule_AddFunctions or
// appended to the module's method table in init_runtime).
static PyMethodDef runtime_module_methods[] = {
    {"get_default_runtime", get_default_runtime_py, METH_NOARGS,
     "Return the module-level default Runtime (lazy-created)."},
    {"set_default_runtime", set_default_runtime_py, METH_VARARGS,
     "Replace the module-level default Runtime (pass None to clear).\n"
     "\n"
     "Args:\n"
     "    runtime (Runtime or None): New default runtime.\n"},
    {NULL}};

int init_runtime(PyObject *m) {
    if (PyType_Ready(&RuntimeType) < 0) return -1;

    Py_INCREF(&RuntimeType);
    if (PyModule_AddObject(m, "Runtime", (PyObject *)&RuntimeType) < 0) {
        Py_DECREF(&RuntimeType);
        Py_DECREF(m);
        return -1;
    }

    for (PyMethodDef *def = runtime_module_methods; def->ml_name; ++def) {
        PyObject *fn = PyCFunction_New(def, NULL);
        if (!fn) return -1;
        if (PyModule_AddObject(m, def->ml_name, fn) < 0) {
            Py_DECREF(fn);
            return -1;
        }
    }

    return 0;
}

1	#define PY_SSIZE_T_CLEAN
2	#include <Python.h>
3	#include <dftracer/utils/python/runtime.h>
4
5	#include <chrono>
6	#include <memory>
7
8	static std::shared_ptr<dftracer::utils::Runtime> g_default_runtime;
9
10	dftracer::utils::Runtime *get_default_runtime() {	165✔
11	if (!g_default_runtime) {	165✔
12	g_default_runtime = std::make_shared<dftracer::utils::Runtime>(0);	1✔
13	}	1✔
14	return g_default_runtime.get();	165✔
15	}
16
17	static void Runtime_dealloc(RuntimeObject *self) {	57✔
18	self->runtime.reset();	57✔
19	Py_TYPE(self)->tp_free((PyObject *)self);	57✔
20	}	57✔
21
22	static PyObject Runtime_new(PyTypeObject type, PyObject *args,	56✔
23	PyObject *kwds) {
24	RuntimeObject self = (RuntimeObject )type->tp_alloc(type, 0);	56✔
25	if (self) {	56!
26	// Placement-new the shared_ptr (tp_alloc gives raw memory)
27	new (&self->runtime) std::shared_ptr<dftracer::utils::Runtime>(nullptr);	56✔
28	}	56✔
29	return (PyObject *)self;	56✔
30	}
31
32	static int Runtime_init(RuntimeObject self, PyObject args, PyObject *kwds) {	56✔
33	static const char *kwlist[] = {"threads", NULL};
34	Py_ssize_t threads = 0;	56✔
35
36	if (!PyArg_ParseTupleAndKeywords(args, kwds, "\|n", (char **)kwlist,	56!
37	&threads)) {
38	return -1;	×
39	}
40
41	if (threads < 0) {	56!
42	PyErr_SetString(PyExc_ValueError, "threads must be >= 0");	×
43	return -1;	×
44	}
45
46	try {
47	self->runtime = std::make_shared<dftracer::utils::Runtime>(	56!
48	static_cast<std::size_t>(threads));	56✔
49	} catch (const std::exception &e) {	56!
50	PyErr_SetString(PyExc_RuntimeError, e.what());	×
51	return -1;	×
52	}	×
53
54	return 0;	56✔
55	}	56✔
56
57	static PyObject Runtime_shutdown(RuntimeObject self,	55✔
58	PyObject *Py_UNUSED(ignored)) {
59	if (!self->runtime) {	55!
60	PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");	×
61	return NULL;	×
62	}
63	Py_BEGIN_ALLOW_THREADS self->runtime->shutdown();	55✔
64	Py_END_ALLOW_THREADS Py_RETURN_NONE;	55✔
65	}	55✔
66
67	static bool set_size(PyObject d, const char key, std::size_t val) {	75✔
68	PyObject *v = PyLong_FromSize_t(val);	75✔
69	if (!v) return false;	75!
70	int rc = PyDict_SetItemString(d, key, v);	75✔
71	Py_DECREF(v);	75✔
72	return rc == 0;	75✔
73	}	75✔
74
75	static bool set_double(PyObject d, const char key, double val) {	21✔
76	PyObject *v = PyFloat_FromDouble(val);	21✔
77	if (!v) return false;	21!
78	int rc = PyDict_SetItemString(d, key, v);	21✔
79	Py_DECREF(v);	21✔
80	return rc == 0;	21✔
81	}	21✔
82
83	static bool set_str(PyObject d, const char key, const std::string &val) {	29✔
84	PyObject *v = PyUnicode_FromStringAndSize(	29✔
85	val.data(), static_cast<Py_ssize_t>(val.size()));	29✔
86	if (!v) return false;	29!
87	int rc = PyDict_SetItemString(d, key, v);	29✔
88	Py_DECREF(v);	29✔
89	return rc == 0;	29✔
90	}	29✔
91
92	static bool set_bool(PyObject d, const char key, bool val) {	8✔
93	PyObject *v = val ? Py_True : Py_False;	8✔
94	Py_INCREF(v);	8✔
95	int rc = PyDict_SetItemString(d, key, v);	8✔
96	Py_DECREF(v);	8✔
97	return rc == 0;	8✔
98	}
99
100	static PyObject *build_task_progress(const dftracer::utils::TaskProgress &tp) {	7✔
101	PyObject *td = PyDict_New();	7✔
102	if (!td) return NULL;	7!
103
104	if (!set_str(td, "name", tp.name) \|\| !set_str(td, "state", tp.state) \|\|	14!
105	!set_double(td, "queued_duration_ms", tp.queued_duration_ms) \|\|	7!
106	!set_double(td, "execution_duration_ms", tp.execution_duration_ms) \|\|	7!
107	!set_size(td, "total_subtasks", tp.total_subtasks) \|\|	7!
108	!set_size(td, "completed_subtasks", tp.completed_subtasks) \|\|	7!
109	!set_double(td, "progress_pct", tp.progress_percentage) \|\|	7!
110	!set_str(td, "location", tp.location)) {	7✔
UNCOV 111	Py_DECREF(td);	×
112	return NULL;	×
113	}
114
115	PyObject *children =	7✔
116	PyList_New(static_cast<Py_ssize_t>(tp.children.size()));	7✔
117	if (!children) {	7!
UNCOV 118	Py_DECREF(td);	×
119	return NULL;	×
120	}
121	for (std::size_t i = 0; i < tp.children.size(); ++i) {	7!
122	PyObject *child = build_task_progress(tp.children[i]);	×
123	if (!child) {	×
UNCOV 124	Py_DECREF(children);	×
UNCOV 125	Py_DECREF(td);	×
126	return NULL;	×
127	}
128	PyList_SET_ITEM(children, static_cast<Py_ssize_t>(i), child);	×
UNCOV 129	}	×
130	if (PyDict_SetItemString(td, "children", children) < 0) {	7!
UNCOV 131	Py_DECREF(children);	×
UNCOV 132	Py_DECREF(td);	×
133	return NULL;	×
134	}
135	Py_DECREF(children);	7✔
136	return td;	7✔
137	}	7✔
138
139	static PyObject Runtime_get_progress(RuntimeObject self,	9✔
140	PyObject *Py_UNUSED(ignored)) {
141	if (!self->runtime) {	9!
142	PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");	×
143	return NULL;	×
144	}
145
146	dftracer::utils::ExecutorProgress prog;	9✔
147	Py_BEGIN_ALLOW_THREADS prog = self->runtime->get_progress();	9!
148	Py_END_ALLOW_THREADS	9!
149
150	PyObject *d = PyDict_New();	9!
151	if (!d) return NULL;	9!
152
153	if (!set_size(d, "total", prog.total_tasks_submitted) \|\|	18!
154	!set_size(d, "completed", prog.tasks_completed) \|\|	9!
155	!set_size(d, "running", prog.tasks_running) \|\|	9!
156	!set_size(d, "queued", prog.tasks_queued) \|\|	9!
157	!set_size(d, "failed", prog.tasks_failed)) {	9!
UNCOV 158	Py_DECREF(d);	×
159	return NULL;	×
160	}
161
162	// Workers
163	PyObject *workers =	9✔
164	PyList_New(static_cast<Py_ssize_t>(prog.workers.size()));	9!
165	if (!workers) {	9!
UNCOV 166	Py_DECREF(d);	×
167	return NULL;	×
168	}
169	for (std::size_t i = 0; i < prog.workers.size(); ++i) {	17✔
170	const auto &w = prog.workers[i];	8✔
171	PyObject *wd = PyDict_New();	8!
172	if (!wd \|\| !set_size(wd, "id", w.worker_id) \|\|	16!
173	!set_bool(wd, "idle", w.is_idle) \|\|	8!
174	!set_str(wd, "task", w.current_task_name) \|\|	8!
175	!set_size(wd, "queue_depth", w.local_queue_depth)) {	8!
176	Py_XDECREF(wd);	×
UNCOV 177	Py_DECREF(workers);	×
UNCOV 178	Py_DECREF(d);	×
179	return NULL;	×
180	}
181	PyList_SET_ITEM(workers, static_cast<Py_ssize_t>(i), wd);	8!
182	}	8✔
183	if (PyDict_SetItemString(d, "workers", workers) < 0) {	9!
UNCOV 184	Py_DECREF(workers);	×
UNCOV 185	Py_DECREF(d);	×
186	return NULL;	×
187	}
188	Py_DECREF(workers);	9!
189
190	// Tasks
191	PyObject *tasks =	9✔
192	PyList_New(static_cast<Py_ssize_t>(prog.root_tasks.size()));	9!
193	if (!tasks) {	9!
UNCOV 194	Py_DECREF(d);	×
195	return NULL;	×
196	}
197	for (std::size_t i = 0; i < prog.root_tasks.size(); ++i) {	16✔
198	PyObject *tp = build_task_progress(prog.root_tasks[i]);	7!
199	if (!tp) {	7!
UNCOV 200	Py_DECREF(tasks);	×
UNCOV 201	Py_DECREF(d);	×
202	return NULL;	×
203	}
204	PyList_SET_ITEM(tasks, static_cast<Py_ssize_t>(i), tp);	7!
205	}	7✔
206	if (PyDict_SetItemString(d, "tasks", tasks) < 0) {	9!
UNCOV 207	Py_DECREF(tasks);	×
UNCOV 208	Py_DECREF(d);	×
209	return NULL;	×
210	}
211	Py_DECREF(tasks);	9!
212
213	// Errors
214	PyObject *errors =	9✔
215	PyList_New(static_cast<Py_ssize_t>(prog.recent_errors.size()));	9!
216	if (!errors) {	9!
UNCOV 217	Py_DECREF(d);	×
218	return NULL;	×
219	}
220	for (std::size_t i = 0; i < prog.recent_errors.size(); ++i) {	9!
221	const auto &[tid, msg] = prog.recent_errors[i];	×
222	PyObject *ed = PyDict_New();	×
223	if (!ed \|\| !set_size(ed, "task_id", static_cast<std::size_t>(tid)) \|\|	×
224	!set_str(ed, "message", msg)) {	×
225	Py_XDECREF(ed);	×
UNCOV 226	Py_DECREF(errors);	×
UNCOV 227	Py_DECREF(d);	×
228	return NULL;	×
229	}
230	PyList_SET_ITEM(errors, static_cast<Py_ssize_t>(i), ed);	×
UNCOV 231	}	×
232	if (PyDict_SetItemString(d, "errors", errors) < 0) {	9!
UNCOV 233	Py_DECREF(errors);	×
UNCOV 234	Py_DECREF(d);	×
235	return NULL;	×
236	}
237	Py_DECREF(errors);	9!
238
239	return d;	9✔
240	}	9✔
241
242	static PyObject Runtime_is_responsive(RuntimeObject self,	1✔
243	PyObject *Py_UNUSED(ignored)) {
244	if (!self->runtime) {	1!
245	PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");	×
246	return NULL;	×
247	}
248	bool resp;
249	Py_BEGIN_ALLOW_THREADS resp = self->runtime->is_responsive();	1✔
250	Py_END_ALLOW_THREADS return PyBool_FromLong(resp ? 1 : 0);	1✔
251	}	1✔
252
253	static PyObject Runtime_set_timeout(RuntimeObject self, PyObject *args,	×
254	PyObject *kwds) {
255	static const char *kwlist[] = {"global_ms", NULL};
256	Py_ssize_t ms = 0;	×
257
258	if (!PyArg_ParseTupleAndKeywords(args, kwds, "\|n", (char **)kwlist, &ms)) {	×
259	return NULL;	×
260	}
261
262	if (!self->runtime) {	×
263	PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");	×
264	return NULL;	×
265	}
266
267	Py_BEGIN_ALLOW_THREADS self->runtime->set_global_timeout(	×
UNCOV 268	std::chrono::milliseconds(ms));	×
269	Py_END_ALLOW_THREADS Py_RETURN_NONE;	×
UNCOV 270	}	×
271
272	static PyObject Runtime_set_default_task_timeout(RuntimeObject self,	×
273	PyObject *args,
274	PyObject *kwds) {
275	static const char *kwlist[] = {"ms", NULL};
276	Py_ssize_t ms = 0;	×
277
278	if (!PyArg_ParseTupleAndKeywords(args, kwds, "\|n", (char **)kwlist, &ms)) {	×
279	return NULL;	×
280	}
281
282	if (!self->runtime) {	×
283	PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");	×
284	return NULL;	×
285	}
286
287	Py_BEGIN_ALLOW_THREADS self->runtime->set_default_task_timeout(	×
UNCOV 288	std::chrono::milliseconds(ms));	×
289	Py_END_ALLOW_THREADS Py_RETURN_NONE;	×
UNCOV 290	}	×
291
292	static PyObject Runtime_wait_all(RuntimeObject self,	71✔
293	PyObject *Py_UNUSED(ignored)) {
294	if (!self->runtime) {	71!
295	PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");	×
296	return NULL;	×
297	}
298	try {
299	Py_BEGIN_ALLOW_THREADS self->runtime->wait_all();	71!
300	Py_END_ALLOW_THREADS	71!
301	} catch (const std::exception &e) {	71!
302	PyErr_SetString(PyExc_RuntimeError, e.what());	×
303	return NULL;	×
304	}	×
305	Py_RETURN_NONE;	71✔
306	}	71✔
307
308	static PyObject Runtime_enter(RuntimeObject self,	×
309	PyObject *Py_UNUSED(ignored)) {
UNCOV 310	Py_INCREF(self);	×
311	return (PyObject *)self;	×
312	}
313
314	static PyObject Runtime_exit(RuntimeObject self, PyObject *args) {	×
315	if (self->runtime) {	×
316	Py_BEGIN_ALLOW_THREADS self->runtime->shutdown();	×
317	Py_END_ALLOW_THREADS	×
UNCOV 318	}	×
319	Py_RETURN_NONE;	×
320	}
321
322	static PyObject Runtime_get_threads(RuntimeObject self, void *closure) {	7✔
323	if (!self->runtime) {	7!
324	PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");	×
325	return NULL;	×
326	}
327	return PyLong_FromSize_t(self->runtime->threads());	7✔
328	}	7✔
329
330	static PyObject get_default_runtime_py(PyObject Py_UNUSED(module),	1✔
331	PyObject *Py_UNUSED(ignored)) {
332	dftracer::utils::Runtime *rt = get_default_runtime();	1✔
333	if (!rt) {	1!
334	PyErr_SetString(PyExc_RuntimeError, "Failed to create default runtime");	×
335	return NULL;	×
336	}
337
338	RuntimeObject obj = (RuntimeObject )RuntimeType.tp_alloc(&RuntimeType, 0);	1✔
339	if (!obj) return NULL;	1!
340
341	new (&obj->runtime)	1✔
342	std::shared_ptr<dftracer::utils::Runtime>(g_default_runtime);	1✔
343	return (PyObject *)obj;	1✔
344	}	1✔
345
346	static PyObject set_default_runtime_py(PyObject Py_UNUSED(module),	2✔
347	PyObject *args) {
348	PyObject *arg;
349	if (!PyArg_ParseTuple(args, "O", &arg)) return NULL;	2!
350
351	if (arg == Py_None) {	2!
352	g_default_runtime.reset();	×
353	Py_RETURN_NONE;	×
354	}
355
356	if (!PyObject_TypeCheck(arg, &RuntimeType)) {	2!
357	PyErr_SetString(PyExc_TypeError, "Expected Runtime or None");	×
358	return NULL;	×
359	}
360
361	g_default_runtime = ((RuntimeObject *)arg)->runtime;	2✔
362	Py_RETURN_NONE;	2✔
363	}	2✔
364
365	static PyMethodDef Runtime_methods[] = {
366	{"shutdown", (PyCFunction)Runtime_shutdown, METH_NOARGS,
367	"shutdown()\n"
368	"--\n"
369	"\n"
370	"Shut down the runtime.\n"},
371	{"get_progress", (PyCFunction)Runtime_get_progress, METH_NOARGS,
372	"Return progress dict with keys: total, completed, running,\n"
373	"queued, failed."},
374	{"is_responsive", (PyCFunction)Runtime_is_responsive, METH_NOARGS,
375	"Return True if the runtime is making progress."},
376	{"set_timeout", (PyCFunction)Runtime_set_timeout,
377	METH_VARARGS \| METH_KEYWORDS,
378	"Set global timeout in milliseconds.\n"
379	"\n"
380	"Args:\n"
381	" global_ms (int): Timeout in milliseconds (0 = no timeout).\n"},
382	{"set_default_task_timeout", (PyCFunction)Runtime_set_default_task_timeout,
383	METH_VARARGS \| METH_KEYWORDS,
384	"Set default per-task timeout in milliseconds.\n"
385	"\n"
386	"Args:\n"
387	" ms (int): Timeout in milliseconds (0 = no timeout).\n"},
388	{"wait_all", (PyCFunction)Runtime_wait_all, METH_NOARGS,
389	"Wait for all outstanding submitted tasks to complete."},
390	{"__enter__", (PyCFunction)Runtime_enter, METH_NOARGS,
391	"Enter context manager."},
392	{"__exit__", (PyCFunction)Runtime_exit, METH_VARARGS,
393	"Exit context manager (calls shutdown)."},
394	{NULL}};
395
396	static PyGetSetDef Runtime_getsetters[] = {
397	{"threads", (getter)Runtime_get_threads, NULL, "Number of worker threads",
398	NULL},
399	{NULL}};
400
401	PyTypeObject RuntimeType = {
402	PyVarObject_HEAD_INIT(NULL, 0) "dftracer_utils_ext.Runtime",
403	sizeof(RuntimeObject), /* tp_basicsize */
404	0, /* tp_itemsize */
405	(destructor)Runtime_dealloc, /* tp_dealloc */
406	0, /* tp_vectorcall_offset */
407	0, /* tp_getattr */
408	0, /* tp_setattr */
409	0, /* tp_as_async */
410	0, /* tp_repr */
411	0, /* tp_as_number */
412	0, /* tp_as_sequence */
413	0, /* tp_as_mapping */
414	0, /* tp_hash */
415	0, /* tp_call */
416	0, /* tp_str */
417	0, /* tp_getattro */
418	0, /* tp_setattro */
419	0, /* tp_as_buffer */
420	Py_TPFLAGS_DEFAULT \| Py_TPFLAGS_BASETYPE, /* tp_flags */
421	"Runtime(threads: int = 0)\n"
422	"--\n"
423	"\n"
424	"Coroutine runtime backed by a thread pool.\n"
425	"\n"
426	"Args:\n"
427	" threads (int): Number of worker threads. 0 (default) uses\n"
428	" the hardware concurrency.\n", /* tp_doc */
429	0, /* tp_traverse */
430	0, /* tp_clear */
431	0, /* tp_richcompare */
432	0, /* tp_weaklistoffset */
433	0, /* tp_iter */
434	0, /* tp_iternext */
435	Runtime_methods, /* tp_methods */
436	0, /* tp_members */
437	Runtime_getsetters, /* tp_getset */
438	0, /* tp_base */
439	0, /* tp_dict */
440	0, /* tp_descr_get */
441	0, /* tp_descr_set */
442	0, /* tp_dictoffset */
443	(initproc)Runtime_init, /* tp_init */
444	0, /* tp_alloc */
445	Runtime_new, /* tp_new */
446	};
447
448	// Module-level function table (registered via PyModule_AddFunctions or
449	// appended to the module's method table in init_runtime).
450	static PyMethodDef runtime_module_methods[] = {
451	{"get_default_runtime", get_default_runtime_py, METH_NOARGS,
452	"Return the module-level default Runtime (lazy-created)."},
453	{"set_default_runtime", set_default_runtime_py, METH_VARARGS,
454	"Replace the module-level default Runtime (pass None to clear).\n"
455	"\n"
456	"Args:\n"
457	" runtime (Runtime or None): New default runtime.\n"},
458	{NULL}};
459
460	int init_runtime(PyObject *m) {	1✔
461	if (PyType_Ready(&RuntimeType) < 0) return -1;	1!
462
463	Py_INCREF(&RuntimeType);	1✔
464	if (PyModule_AddObject(m, "Runtime", (PyObject *)&RuntimeType) < 0) {	1!
UNCOV 465	Py_DECREF(&RuntimeType);	×
UNCOV 466	Py_DECREF(m);	×
467	return -1;	×
468	}
469
470	for (PyMethodDef *def = runtime_module_methods; def->ml_name; ++def) {	3✔
471	PyObject *fn = PyCFunction_New(def, NULL);	2✔
472	if (!fn) return -1;	2!
473	if (PyModule_AddObject(m, def->ml_name, fn) < 0) {	2!
UNCOV 474	Py_DECREF(fn);	×
475	return -1;	×
476	}
477	}	2✔
478
479	return 0;	1✔
480	}	1✔

llnl / dftracer-utils / 23529483807

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