27052412546

Committed 06 Jun 2026 04:20AM UTC coverage: 50.862% (+1.0%) from 49.905%

Build # 27052412546

Build Type

Pull #73

github

Committed by

web-flow

Commit Message

Merge 734572730 into 88a3c8457

Pull Request Pull Request #73: add portable dependencies wheel support

Coverage Stats

31801 of 79859 branches covered (39.82%)

Branch coverage included in aggregate %.

32491 of 46545 relevant lines covered (69.81%)

9947.11 hits per line

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

47.96

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

llnl / dftracer-utils / 27052412546

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