28521653886

Committed 01 Jul 2026 01:36PM UTC coverage: 50.92% (-1.4%) from 52.278%

Build # 28521653886

Build Type

Pull #83

github

Committed by

web-flow

Commit Message

Merge 9bdedb1e9 into 2efed6649

Pull Request Pull Request #83: refactor and improve code QoL

Coverage Stats

31893 of 80049 branches covered (39.84%)

Branch coverage included in aggregate %.

789 of 1613 new or added lines in 87 files covered. (48.92%)

5007 existing lines in 181 files now uncovered.

32812 of 47024 relevant lines covered (69.78%)

9905.42 hits per line

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

46.31

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

#define PY_SSIZE_T_CLEAN
#include <Python.h>
#include <dftracer/utils/python/py_dict_helpers.h>
#include <dftracer/utils/python/py_errors.h>
#include <dftracer/utils/python/py_type_helpers.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", "io_threads", NULL};
    Py_ssize_t threads = 0;
    Py_ssize_t io_threads = 0;

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

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

    try {
        dftracer::utils::ExecutorConfig config;
        config.num_threads = static_cast<std::size_t>(threads);
        config.io_pool_size = static_cast<std::size_t>(io_threads);
        self->runtime =
            std::make_shared<dftracer::utils::Runtime>(config, true);
    } catch (const std::exception &e) {
        set_typed_py_error(e);
        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) {
    return dict_set_size(d, key, val) == 0;
}

static bool set_double(PyObject *d, const char *key, double val) {
    return dict_set_f64(d, key, val) == 0;
}

static bool set_str(PyObject *d, const char *key, const std::string &val) {
    return dict_set_str(d, key, val.c_str()) == 0;
}

static bool set_bool(PyObject *d, const char *key, bool val) {
    return dict_set_bool(d, key, val) == 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) {
        set_typed_py_error(e);
        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 PyObject *Runtime_get_io_threads(RuntimeObject *self, void *closure) {
    if (!self->runtime) {
        PyErr_SetString(PyExc_RuntimeError, "Runtime not initialized");
        return NULL;
    }
    return PyLong_FromSize_t(self->runtime->io_threads());
}

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

llnl / dftracer-utils / 28521653886

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