DISPATCH-848, DISPATCH-1962 Fix leak of IoAdapter_init

.github/workflows/build.yaml

@@ -32,6 +32,7 @@ jobs:
         buildType: [Debug]
         runtimeCheck: [asan]
         protonGitRef: [main, 0.36.0]
+        python: [/usr/bin/python3-dbg]
     env:
       BuildType: ${{matrix.buildType}}
       ProtonBuildDir: ${{github.workspace}}/qpid-proton/build
@@ -47,6 +48,7 @@ jobs:
         -DBUILD_TESTING=OFF
         -DENABLE_FUZZ_TESTING=OFF
         -DRUNTIME_CHECK=${{matrix.runtimeCheck}}
+        -DPython_EXECUTABLE=${{matrix.python}}
       DispatchCMakeExtraArgs: >
         -DCMAKE_C_COMPILER_LAUNCHER=ccache
         -DCMAKE_CXX_COMPILER_LAUNCHER=ccache
@@ -56,6 +58,7 @@ jobs:
         -DRUNTIME_CHECK=${{matrix.runtimeCheck}}
         -DSANITIZE_3RD_PARTY=ON
         -DBUILD_BENCHMARKS=ON
+        -DPython_EXECUTABLE=${{matrix.python}}
 
       CCACHE_BASEDIR: ${{github.workspace}}
       CCACHE_DIR: ${{github.workspace}}/.ccache
@@ -109,7 +112,10 @@ jobs:
       - name: Install Linux build dependencies
         if: ${{ runner.os == 'Linux' }}
         run: |
-          sudo apt update; sudo apt install -y swig libpython3-dev libsasl2-dev libjsoncpp-dev libwebsockets-dev ccache ninja-build pixz libbenchmark-dev
+          sudo apt update; sudo apt install -y swig python3-dbg libpython3-dbg libsasl2-dev libjsoncpp-dev libwebsockets-dev ccache ninja-build pixz libbenchmark-dev
+
+      - name: Install Python build dependencies
+        run: ${{matrix.python}} -m pip install setuptools wheel tox
 
       - name: Zero ccache stats
         run: ccache -z
@@ -177,6 +183,7 @@ jobs:
         buildType: [Debug]
         runtimeCheck: [asan]
         protonGitRef: [main, 0.36.0]
+        python: [/usr/bin/python3-dbg]
         shard: [1, 2]
         shards: [2]
     env:
@@ -189,6 +196,11 @@ jobs:
       LD_LIBRARY_PATH: ${{github.workspace}}/install/lib
       QPID_SYSTEM_TEST_TIMEOUT: 300
       QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST: True
+      # the PyMalloc mechanism is incompatible with Valgrind, different mechanism must be set here
+      #  https://docs.python.org/3/using/cmdline.html#envvar-PYTHONMALLOC
+      #  https://pythonextensionpatterns.readthedocs.io/en/latest/debugging/debug_python.html#debug-version-of-python-memory-alloc-label
+      PYTHONMALLOC: malloc_debug
+      PYTHONTRACEMALLOC: 5
     steps:
 
       - name: Show environment (Linux)
@@ -207,18 +219,24 @@ jobs:
           architecture: x64
 
       - name: Install Python runtime/test dependencies
-        run: python -m pip install tox websockets pytest
+        run: python -m pip install tox quart selectors h2 grpcio protobuf websockets pytest
 
       - name: Install Linux runtime/test dependencies
         if: ${{ runner.os == 'Linux' }}
         run: |
-          sudo apt update; sudo apt install -y libsasl2-2 libsasl2-modules sasl2-bin libjsoncpp1 libwebsockets15 libbenchmark1 pixz bubblewrap curl
+          sudo apt update; sudo apt install -y python3-dbg libsasl2-2 libsasl2-modules sasl2-bin libjsoncpp1 libwebsockets15 libbenchmark1 pixz bubblewrap curl
 
       - name: Unpack archive
         run: tar -I pixz -xf archive.tar.xz
 
-      - name: install qpid-proton python wheel
-        run: python -m pip install $(find ${ProtonBuildDir}/python/ -name 'python_qpid_proton*.whl')
+      - name: Install Python runtime/test dependencies
+        run: ${{matrix.python}} -m pip install tox quart selectors h2 grpcio protobuf websockets pytest
+
+      - name: Replace /usr/bin/python3 with ${{matrix.python}}, for tools such as qdmanage
+        run: sudo ln -sf ${{matrix.python}} /usr/bin/python3
+
+      - name: Install qpid-proton python wheel (python3-dbg)
+        run: ${{matrix.python}} -m pip install $(find ${ProtonBuildDir}/python/ -name 'python_qpid_proton*.whl')
 
       - name: CTest
         working-directory: ${{env.DispatchBuildDir}}

src/dispatch.c

@@ -340,6 +340,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/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,
 };
 
 
@@ -637,6 +639,11 @@ static uint64_t qd_io_rx_handler(void *context, qd_message_t *msg, int link_id,
     IoAdapter *self = (IoAdapter*) context;
     *error = 0;
 
+    if (self->handler == NULL) {
+        *error = qdr_error(QD_AMQP_COND_INTERNAL_ERROR, "Router is shutting down");
+        return PN_REJECTED;
+    }
+
     //
     // Parse the message through the body and exit if the message is not well formed.
     //
@@ -719,10 +726,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);
 }
 
@@ -802,10 +823,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,
 };
 
 
@@ -821,8 +845,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_core/router_core.c

@@ -255,6 +255,10 @@ void qdr_core_free(qdr_core_t *core)
     // this must happen after qdrc_endpoint_do_cleanup_CT calls
     qdr_modules_finalize(core);
 
+    // Drain the general work lists
+    //  (this can also generate actions, e.g. qdr_forward_on_message -> qd_io_rx_handler call chain does that)
+    qdr_general_handler(core);
+
     // discard any left over actions
 
     qdr_action_list_t  action_list;
@@ -268,9 +272,6 @@ void qdr_core_free(qdr_core_t *core)
         action = DEQ_HEAD(action_list);
     }
 
-    // Drain the general work lists
-    qdr_general_handler(core);
-
     sys_thread_free(core->thread);
     sys_cond_free(core->action_cond);
     sys_mutex_free(core->action_lock);

src/router_node.c

@@ -2169,12 +2169,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/http2_server.py

@@ -22,14 +22,14 @@
 
 from quart import Quart, request
 try:
-    from quart.static import send_file  # noqa  # mypy#8823  # type: ignore[attr-defined]
+    from quart.static import send_file  # type: ignore[attr-defined]  # mypy#8823
 except ImportError:
-    from quart.helpers import send_file  # noqa  # mypy#8823  # type: ignore[attr-defined, no-redef]  # mypy#1153
+    from quart.helpers import send_file  # type: ignore[attr-defined, no-redef]  # mypy#1153
 
 try:
-    from quart.exceptions import HTTPStatusException  # noqa  # mypy#8823  # type: ignore[attr-defined]
+    from quart.exceptions import HTTPStatusException  # type: ignore[attr-defined]
 except ImportError:
-    from werkzeug.exceptions import InternalServerError as HTTPStatusException  # noqa  # mypy#8823  # type: ignore[no-redef]  # mypy#1153
+    from werkzeug.exceptions import InternalServerError as HTTPStatusException  # type: ignore[no-redef]  # mypy#1153
 
 app = Quart(__name__)
 

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$
 
@@ -61,6 +58,8 @@ leak:^PyThread_allocate_lock$
 leak:^PyMem_Malloc$
 leak:^PyMem_Calloc$
 leak:^PyMem_Realloc$
+leak:^_PyMem_RawMalloc$
+leak:^_PyMem_RawRealloc$
 leak:^_PyObject_GC_Resize$
 # Python uses these alloc functions if you define PYTHONDEVMODE=1
 leak:^_PyMem_DebugRawAlloc$

tests/system_test.py

@@ -212,10 +212,12 @@ def port_available(port, protocol_family='IPv4'):
 def wait_port(port, protocol_family='IPv4', **retry_kwargs):
     """Wait up to timeout for port (on host) to be connectable.
     Takes same keyword arguments as retry to control the timeout"""
-    def check(e):
+
+    def check(e: Exception) -> None:
         """Only retry on connection refused"""
-        if not isinstance(e, socket.error) or not e.errno == errno.ECONNREFUSED:
-            raise
+        if isinstance(e, OSError) and e.errno == errno.ECONNREFUSED:
+            return
+        raise
 
     host = None
 

tests/system_tests_websockets.py

@@ -23,7 +23,7 @@
 try:
     import websockets
 except ImportError:
-    websockets = None  # noqa  # type: ignore[assignment]  # expression has type "None", variable has type Module
+    websockets = None  # type: ignore[assignment]  # expression has type "None", variable has type Module
 
 from system_test import Qdrouterd
 from system_test import main_module, TestCase, Process

tests/tox.ini.in

@@ -217,7 +217,8 @@ disable =
 
 [mypy]
 warn_redundant_casts = True
-warn_unused_ignores = True
+# enable when wanting to clean up ignores
+# warn_unused_ignores = True
 
 # mypy cannot handle overridden attributes
 # https://github.com/python/mypy/issues/7505