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

46.91

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