Fixes #412, DISPATCH-1962 - Python shutdown leaks

python/skupper_router_internal/dispatch.py

@@ -81,9 +81,8 @@ def __init__(self, handle: int) -> None:
 
         self._prototype(self.qd_dispatch_configure_policy, None, [self.qd_dispatch_p, py_object])
         self._prototype(self.qd_dispatch_register_policy_manager, None, [self.qd_dispatch_p, py_object])
-        self._prototype(self.qd_dispatch_policy_c_counts_alloc, c_long, [], check=False)
-        self._prototype(self.qd_dispatch_policy_c_counts_free, None, [c_long], check=False)
-        self._prototype(self.qd_dispatch_policy_c_counts_refresh, None, [c_long, py_object])
+        self._prototype(self.qd_dispatch_policy_c_counts_alloc, py_object, [], check=False)
+        self._prototype(self.qd_dispatch_policy_c_counts_refresh, None, [py_object, py_object])
         self._prototype(self.qd_dispatch_policy_host_pattern_add, ctypes.c_bool, [self.qd_dispatch_p, py_object])
         self._prototype(self.qd_dispatch_policy_host_pattern_remove, None, [self.qd_dispatch_p, py_object])
         self._prototype(self.qd_dispatch_policy_host_pattern_lookup, c_char_p, [self.qd_dispatch_p, py_object])

python/skupper_router_internal/policy/policy_local.py

@@ -18,7 +18,7 @@
 #
 
 """Entity implementing the business logic of user connection/access policy."""
-
+import ctypes
 import json
 from typing import Any, Dict, List, Union, TYPE_CHECKING
 
@@ -590,7 +590,7 @@ def disconnect(self, conn_id, user, host):
     def count_other_denial(self) -> None:
         self.conn_mgr.count_other_denial()
 
-    def get_cstats(self) -> int:
+    def get_cstats(self) -> ctypes.py_object:
         return self._cstats
 
 #
@@ -939,7 +939,7 @@ def lookup_settings(
             self,
             vhost_in: str,
             groupname: str,
-            upolicy: Dict[Any, Any]
+            upolicy: Dict[str, Any]
     ) -> bool:
         """
         Given a settings name, return the aggregated policy blob.

src/dispatch.c

@@ -58,6 +58,7 @@ void            qd_router_free(qd_router_t *router);
 void            qd_error_initialize();
 static void qd_dispatch_set_router_id(qd_dispatch_t *qd, char *_id);
 static void qd_dispatch_set_router_area(qd_dispatch_t *qd, char *_area);
+static void qd_dispatch_policy_c_counts_free(PyObject *capsule);
 
 const char     *CLOSEST_DISTRIBUTION   = "closest";
 const char     *MULTICAST_DISTRIBUTION = "multicast";
@@ -270,20 +271,24 @@ QD_EXPORT qd_error_t qd_dispatch_register_display_name_service(qd_dispatch_t *qd
     return qd_register_display_name_service(qd, object);
 }
 
-
-QD_EXPORT long qd_dispatch_policy_c_counts_alloc()
+QD_EXPORT PyObject* qd_dispatch_policy_c_counts_alloc()
 {
-    return qd_policy_c_counts_alloc();
+    return PyCapsule_New(qd_policy_c_counts_alloc(), "qd_policy_c_counts", qd_dispatch_policy_c_counts_free);
 }
 
-
-QD_EXPORT void qd_dispatch_policy_c_counts_free(long ccounts)
+static void qd_dispatch_policy_c_counts_free(PyObject *capsule)
 {
+    void *ccounts = PyCapsule_GetPointer(capsule, "qd_policy_c_counts");
     qd_policy_c_counts_free(ccounts);
 }
 
-QD_EXPORT void qd_dispatch_policy_c_counts_refresh(long ccounts, qd_entity_t *entity)
+QD_EXPORT void qd_dispatch_policy_c_counts_refresh(PyObject *ccounts_capsule, qd_entity_t *entity)
 {
+    assert(PyCapsule_CheckExact(ccounts_capsule));
+    const char * name = PyCapsule_GetName(ccounts_capsule);
+    assert(PyCapsule_IsValid(ccounts_capsule, name));
+    void* ccounts = PyCapsule_GetPointer(ccounts_capsule, name);
+    qd_error_py();
     qd_policy_c_counts_refresh(ccounts, entity);
 }
 
@@ -324,6 +329,7 @@ qd_error_t qd_dispatch_prepare(qd_dispatch_t *qd)
 void qd_dispatch_set_agent(qd_dispatch_t *qd, void *agent) {
     assert(agent);
     assert(!qd->agent);
+    Py_IncRef(agent);
     qd->agent = agent;
 }
 

src/entity.c

@@ -70,6 +70,18 @@ long qd_entity_get_long(qd_entity_t *entity, const char* attribute) {
     return result;
 }
 
+void *qd_entity_get_pointer_from_capsule(qd_entity_t *entity, const char *attribute) {
+    qd_error_clear();
+    PyObject *py_obj = qd_entity_get_py(entity, attribute);
+    assert(PyCapsule_CheckExact(py_obj));
+    const char * name = PyCapsule_GetName(py_obj);
+    assert(PyCapsule_IsValid(py_obj, name));
+    void* result = PyCapsule_GetPointer(py_obj, name);
+    Py_XDECREF(py_obj);
+    qd_error_py();
+    return result;
+}
+
 bool qd_entity_get_bool(qd_entity_t *entity, const char* attribute) {
     qd_error_clear();
     PyObject *py_obj = qd_entity_get_py(entity, attribute);

src/entity.h

@@ -44,6 +44,9 @@ char *qd_entity_get_string(qd_entity_t *entity, const char* attribute);
 /** Get an integer valued attribute. Return -1 and set qd_error if there is an error. */
 long qd_entity_get_long(qd_entity_t *entity, const char* attribute);
 
+/** Get a void* valued attribute stored in a PyCapsule. Return NULL and set qd_error if there is an error. */
+void *qd_entity_get_pointer_from_capsule(qd_entity_t *entity, const char *attribute);
+
 /** Get a boolean valued attribute. Return false and set qd_error if there is an error. */
 bool qd_entity_get_bool(qd_entity_t *entity, const char *attribute);
 

src/policy.c

@@ -191,26 +191,23 @@ qd_error_t qd_register_policy_manager(qd_policy_t *policy, void *policy_manager)
 }
 
 
-long qd_policy_c_counts_alloc()
+void *qd_policy_c_counts_alloc()
 {
-    qd_policy_denial_counts_t * dc = NEW(qd_policy_denial_counts_t);
+    qd_policy_denial_counts_t *dc = NEW(qd_policy_denial_counts_t);
     assert(dc);
     ZERO(dc);
-    return (long)dc;
+    return dc;
 }
 
-
-void qd_policy_c_counts_free(long ccounts)
+void qd_policy_c_counts_free(void *dc)
 {
-    void *dc = (void *)ccounts;
     assert(dc);
     free(dc);
 }
 
 
-qd_error_t qd_policy_c_counts_refresh(long ccounts, qd_entity_t *entity)
+qd_error_t qd_policy_c_counts_refresh(qd_policy_denial_counts_t* dc, qd_entity_t *entity)
 {
-    qd_policy_denial_counts_t *dc = (qd_policy_denial_counts_t*)ccounts;
     if (!qd_entity_set_long(entity, "sessionDenied", dc->sessionDenied) &&
         !qd_entity_set_long(entity, "senderDenied", dc->senderDenied) &&
         !qd_entity_set_long(entity, "receiverDenied", dc->receiverDenied) &&
@@ -573,7 +570,7 @@ bool qd_policy_open_fetch_settings(
                         settings->sourceParseTree      = qd_policy_parse_tree(settings->sourcePattern);
                         settings->targetParseTree      = qd_policy_parse_tree(settings->targetPattern);
                         settings->denialCounts         = (qd_policy_denial_counts_t*)
-                                                        qd_entity_get_long((qd_entity_t*)upolicy, "denialCounts");
+                                                        qd_entity_get_pointer_from_capsule((qd_entity_t*)upolicy, "denialCounts");
                         res = true; // named settings content returned
                     } else {
                         // lookup failed: object did not exist in python database

src/policy.h

@@ -85,17 +85,17 @@ qd_error_t qd_register_policy_manager(qd_policy_t *policy, void *policy_manager)
 /** Allocate counts statistics block.
  * Called from Python
  */
-long qd_policy_c_counts_alloc();
+void* qd_policy_c_counts_alloc();
 
 /** Free counts statistics block.
  * Called from Python
  */
-void qd_policy_c_counts_free(long ccounts);
+void qd_policy_c_counts_free(void* dc);
 
 /** Refresh a counts statistics block
  * Called from Python
  */
-qd_error_t qd_policy_c_counts_refresh(long ccounts, qd_entity_t*entity);
+qd_error_t qd_policy_c_counts_refresh(qd_policy_denial_counts_t* dc, qd_entity_t*entity);
 
 
 /** Allow or deny an incoming connection based on connection count(s).

src/python_embedded.c

@@ -66,6 +66,7 @@ void qd_python_finalize(void)
 {
     (void) qd_python_lock();
 
+    Py_DECREF(message_type);
     Py_DECREF(dispatch_module);
     dispatch_module = 0;
     PyGC_Collect();
@@ -565,7 +566,8 @@ static PyTypeObject LogAdapterType = {
     .tp_dealloc   = (destructor)LogAdapter_dealloc,
     .tp_flags     = Py_TPFLAGS_DEFAULT,
     .tp_methods   = LogAdapter_methods,
-    .tp_init      = (initproc)LogAdapter_init
+    .tp_init      = (initproc)LogAdapter_init,
+    .tp_new       = PyType_GenericNew,
 };
 
 
@@ -710,10 +712,24 @@ static int IoAdapter_init(IoAdapter *self, PyObject *args, PyObject *kwds)
     return 0;
 }
 
+// visit all members which may conceivably participate in reference cycles
+static int IoAdapter_traverse(IoAdapter* self, visitproc visit, void *arg)
+{
+    Py_VISIT(self->handler);
+    return 0;
+}
+
+static int IoAdapter_clear(IoAdapter* self)
+{
+    Py_CLEAR(self->handler);
+    return 0;
+}
+
 static void IoAdapter_dealloc(IoAdapter* self)
 {
     qdr_core_unsubscribe(self->sub);
-    Py_DECREF(self->handler);
+    PyObject_GC_UnTrack(self);
+    IoAdapter_clear(self);
     Py_TYPE(self)->tp_free((PyObject*)self);
 }
 
@@ -795,10 +811,13 @@ static PyTypeObject IoAdapterType = {
     .tp_name      = DISPATCH_MODULE ".IoAdapter",
     .tp_doc       = "Dispatch IO Adapter",
     .tp_basicsize = sizeof(IoAdapter),
+    .tp_traverse  = (traverseproc)IoAdapter_traverse,
+    .tp_clear     = (inquiry)IoAdapter_clear,
     .tp_dealloc   = (destructor)IoAdapter_dealloc,
-    .tp_flags     = Py_TPFLAGS_DEFAULT,
+    .tp_flags     = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC,
     .tp_methods   = IoAdapter_methods,
     .tp_init      = (initproc)IoAdapter_init,
+    .tp_new       = PyType_GenericNew,
 };
 
 
@@ -814,8 +833,6 @@ static void qd_register_constant(PyObject *module, const char *name, uint32_t va
 
 static void qd_python_setup(void)
 {
-    LogAdapterType.tp_new = PyType_GenericNew;
-    IoAdapterType.tp_new  = PyType_GenericNew;
     if ((PyType_Ready(&LogAdapterType) < 0) || (PyType_Ready(&IoAdapterType) < 0)) {
         qd_error_py();
         qd_log(log_source, QD_LOG_CRITICAL, "Unable to initialize Adapters");

src/router_node.c

@@ -2092,12 +2092,12 @@ void qd_router_free(qd_router_t *router)
 
     qd_container_set_default_node_type(router->qd, 0, 0, QD_DIST_BOTH);
 
+    qd_router_python_free(router);
     qdr_core_free(router->router_core);
     qd_tracemask_free(router->tracemask);
     qd_timer_free(router->timer);
     sys_mutex_free(router->lock);
     qd_router_configure_free(router);
-    qd_router_python_free(router);
 
     free(router);
     qd_router_id_finalize();

src/router_pynode.c

@@ -443,6 +443,7 @@ qd_error_t qd_router_python_setup(qd_router_t *router)
     // Instantiate the router
     //
     pyRouter = PyObject_CallObject(pClass, pArgs);
+    Py_DECREF(pClass);
     Py_DECREF(pArgs);
     Py_DECREF(adapterType);
     QD_ERROR_PY_RET();
@@ -455,7 +456,14 @@ qd_error_t qd_router_python_setup(qd_router_t *router)
 }
 
 void qd_router_python_free(qd_router_t *router) {
-    // empty
+    qd_python_lock_state_t ls = qd_python_lock();
+    Py_XDECREF(pyRouter);
+    Py_CLEAR(pyTick);
+    Py_CLEAR(pySetMobileSeq);
+    Py_CLEAR(pySetMyMobileSeq);
+    Py_CLEAR(pyLinkLost);
+    PyGC_Collect();
+    qd_python_unlock(ls);
 }
 
 

tests/lsan.supp

@@ -2,9 +2,6 @@
 # found by AddressSanitizer (ASAN)
 #
 
-# to be triaged; pretty much all tests
-leak:^IoAdapter_init$
-
 # to be triaged; pretty much all tests
 leak:^load_server_config$
 
@@ -14,9 +11,6 @@ leak:^qd_policy_open_fetch_settings$
 # to be triaged; system_tests_handle_failover
 leak:^parse_failover_property_list$
 
-# to be triaged; system_tests_policy, system_tests_policy_oversize_basic
-leak:^qd_policy_c_counts_alloc$
-
 # to be triaged; system_tests_http
 leak:^callback_healthz$
 leak:^callback_metrics$