[fix] [python client] Better Python garbage collection management for C++-owned objects

pulsar-client-cpp/lib/checksum/crc32c_arm.h

@@ -37,11 +37,11 @@
 #define crc32c_u16(crc, v) __crc32ch(crc, v)
 #define crc32c_u32(crc, v) __crc32cw(crc, v)
 #define crc32c_u64(crc, v) __crc32cd(crc, v)
-#define PREF4X64L1(buffer, PREF_OFFSET, ITR)                                                                \
-    __asm__("PRFM PLDL1KEEP, [%x[v],%[c]]" ::[v] "r"(buffer), [ c ] "I"((PREF_OFFSET) + ((ITR) + 0) * 64)); \
-    __asm__("PRFM PLDL1KEEP, [%x[v],%[c]]" ::[v] "r"(buffer), [ c ] "I"((PREF_OFFSET) + ((ITR) + 1) * 64)); \
-    __asm__("PRFM PLDL1KEEP, [%x[v],%[c]]" ::[v] "r"(buffer), [ c ] "I"((PREF_OFFSET) + ((ITR) + 2) * 64)); \
-    __asm__("PRFM PLDL1KEEP, [%x[v],%[c]]" ::[v] "r"(buffer), [ c ] "I"((PREF_OFFSET) + ((ITR) + 3) * 64));
+#define PREF4X64L1(buffer, PREF_OFFSET, ITR)                                                              \
+    __asm__("PRFM PLDL1KEEP, [%x[v],%[c]]" ::[v] "r"(buffer), [c] "I"((PREF_OFFSET) + ((ITR) + 0) * 64)); \
+    __asm__("PRFM PLDL1KEEP, [%x[v],%[c]]" ::[v] "r"(buffer), [c] "I"((PREF_OFFSET) + ((ITR) + 1) * 64)); \
+    __asm__("PRFM PLDL1KEEP, [%x[v],%[c]]" ::[v] "r"(buffer), [c] "I"((PREF_OFFSET) + ((ITR) + 2) * 64)); \
+    __asm__("PRFM PLDL1KEEP, [%x[v],%[c]]" ::[v] "r"(buffer), [c] "I"((PREF_OFFSET) + ((ITR) + 3) * 64));
 
 #define PREF1KL1(buffer, PREF_OFFSET)    \
     PREF4X64L1(buffer, (PREF_OFFSET), 0) \

pulsar-client-cpp/python/pulsar_test.py

@@ -19,6 +19,8 @@
 #
 
 
+import threading
+import logging
 from unittest import TestCase, main
 import time
 import os
@@ -1238,6 +1240,35 @@ def test_json_schema_encode(self):
         second_encode = schema.encode(record)
         self.assertEqual(first_encode, second_encode)
 
+    def test_logger_thread_leaks(self):
+        def _do_connect(close):
+            logger = logging.getLogger(str(threading.currentThread().ident))
+            logger.setLevel(logging.INFO)
+            client = pulsar.Client(
+                service_url="pulsar://localhost:6650",
+                io_threads=4,
+                message_listener_threads=4,
+                operation_timeout_seconds=1,
+                log_conf_file_path=None,
+                authentication=None,
+                logger=logger,
+            )
+            client.get_topic_partitions("persistent://public/default/partitioned_topic_name_test")
+            if close:
+                client.close()
+
+        for should_close in (True, False):
+            assert threading.active_count() == 1, "Explicit close: {}; baseline is 1 thread".format(should_close)
+            _do_connect(should_close)
+            assert threading.active_count() == 1, "Explicit close: {}; synchronous connect doesn't leak threads".format(should_close)
+            threads = []
+            for _ in range(10):
+                threads.append(threading.Thread(target=_do_connect))
+                threads[-1].start()
+            for thread in threads:
+                thread.join()
+            assert threading.active_count() == 1, "Explicit close: {}; threaded connect in parallel doesn't leak threads".format(should_close)
+
     def _check_value_error(self, fun):
         with self.assertRaises(ValueError):
             fun()

pulsar-client-cpp/python/src/config.cc

@@ -93,70 +93,76 @@ static ReaderConfiguration& ReaderConfiguration_setCryptoKeyReader(ReaderConfigu
     return conf;
 }
 
-class LoggerWrapper : public Logger {
-    PyObject* const _pyLogger;
+class LoggerWrapper : public Logger, public CaptivePythonObjectMixin {
     const int _pythonLogLevel;
     const std::unique_ptr<Logger> _fallbackLogger;
-
+    py::object _loggerModule;
     static constexpr int _getLogLevelValue(Level level) { return 10 + (level * 10); }
 
    public:
     LoggerWrapper(PyObject* pyLogger, int pythonLogLevel, Logger* fallbackLogger)
-        : _pyLogger(pyLogger), _pythonLogLevel(pythonLogLevel), _fallbackLogger(fallbackLogger) {
-        Py_XINCREF(_pyLogger);
+        : CaptivePythonObjectMixin(pyLogger),
+          _pythonLogLevel(pythonLogLevel),
+          _fallbackLogger(fallbackLogger) {
+        PyGILState_STATE state = PyGILState_Ensure();
+        _loggerModule = py::import("logging");
+        PyGILState_Release(state);
     }
 
     LoggerWrapper(const LoggerWrapper&) = delete;
     LoggerWrapper(LoggerWrapper&&) noexcept = delete;
     LoggerWrapper& operator=(const LoggerWrapper&) = delete;
     LoggerWrapper& operator=(LoggerWrapper&&) = delete;
 
-    virtual ~LoggerWrapper() { Py_XDECREF(_pyLogger); }
-
     bool isEnabled(Level level) { return _getLogLevelValue(level) >= _pythonLogLevel; }
 
     void log(Level level, int line, const std::string& message) {
+        switch (level) {
+            case Logger::LEVEL_DEBUG:
+                log(level, line, "debug", message);
+                break;
+            case Logger::LEVEL_INFO:
+                log(level, line, "info", message);
+                break;
+            case Logger::LEVEL_WARN:
+                log(level, line, "warning", message);
+                break;
+            case Logger::LEVEL_ERROR:
+                log(level, line, "error", message);
+                break;
+        }
+    }
+
+   private:
+    void log(Level level, int line, const char* method, const std::string& message) {
         if (!Py_IsInitialized()) {
             // Python logger is unavailable - fallback to console logger
             _fallbackLogger->log(level, line, message);
         } else {
             PyGILState_STATE state = PyGILState_Ensure();
-
+            py::object oldLogThreads = _loggerModule.attr("logThreads");
+            _loggerModule.attr("logThreads") = py::object(false);
             try {
-                switch (level) {
-                    case Logger::LEVEL_DEBUG:
-                        py::call_method<void>(_pyLogger, "debug", message.c_str());
-                        break;
-                    case Logger::LEVEL_INFO:
-                        py::call_method<void>(_pyLogger, "info", message.c_str());
-                        break;
-                    case Logger::LEVEL_WARN:
-                        py::call_method<void>(_pyLogger, "warning", message.c_str());
-                        break;
-                    case Logger::LEVEL_ERROR:
-                        py::call_method<void>(_pyLogger, "error", message.c_str());
-                        break;
-                }
-
+                py::call_method<void>(_captive, method, message.c_str());
+                _loggerModule.attr("logThreads") = oldLogThreads;
             } catch (const py::error_already_set& e) {
                 _fallbackLogger->log(level, line, message);
+                _loggerModule.attr("logThreads") = oldLogThreads;
             }
-
             PyGILState_Release(state);
         }
     }
 };
 
-class LoggerWrapperFactory : public LoggerFactory {
+class LoggerWrapperFactory : public LoggerFactory, public CaptivePythonObjectMixin {
     std::unique_ptr<LoggerFactory> _fallbackLoggerFactory{new ConsoleLoggerFactory};
-    PyObject* _pyLogger;
     Optional<int> _pythonLogLevel{Optional<int>::empty()};
 
     void initializePythonLogLevel() {
         PyGILState_STATE state = PyGILState_Ensure();
 
         try {
-            int level = py::call_method<int>(_pyLogger, "getEffectiveLevel");
+            int level = py::call_method<int>(_captive, "getEffectiveLevel");
             _pythonLogLevel = Optional<int>::of(level);
         } catch (const py::error_already_set& e) {
             // Failed to get log level from _pyLogger, set it to empty to fallback to _fallbackLogger
@@ -167,18 +173,14 @@ class LoggerWrapperFactory : public LoggerFactory {
     }
 
    public:
-    LoggerWrapperFactory(py::object pyLogger) {
-        _pyLogger = pyLogger.ptr();
-        Py_XINCREF(_pyLogger);
+    LoggerWrapperFactory(py::object pyLogger) : CaptivePythonObjectMixin(pyLogger) {
         initializePythonLogLevel();
     }
 
-    virtual ~LoggerWrapperFactory() { Py_XDECREF(_pyLogger); }
-
     Logger* getLogger(const std::string& fileName) {
         const auto fallbackLogger = _fallbackLoggerFactory->getLogger(fileName);
         if (_pythonLogLevel.is_present()) {
-            return new LoggerWrapper(_pyLogger, _pythonLogLevel.value(), fallbackLogger);
+            return new LoggerWrapper(_captive, _pythonLogLevel.value(), fallbackLogger);
         } else {
             return fallbackLogger;
         }
@@ -315,4 +317,4 @@ void export_config() {
         .def("read_compacted", &ReaderConfiguration::isReadCompacted)
         .def("read_compacted", &ReaderConfiguration::setReadCompacted)
         .def("crypto_key_reader", &ReaderConfiguration_setCryptoKeyReader, return_self<>());
-}
+}

pulsar-client-cpp/python/src/utils.h

@@ -82,3 +82,25 @@ struct CryptoKeyReaderWrapper {
     CryptoKeyReaderWrapper();
     CryptoKeyReaderWrapper(const std::string& publicKeyPath, const std::string& privateKeyPath);
 };
+
+class CaptivePythonObjectMixin {
+   protected:
+    PyObject* _captive;
+
+    CaptivePythonObjectMixin(PyObject* captive) {
+        _captive = captive;
+        PyGILState_STATE state = PyGILState_Ensure();
+        Py_XINCREF(_captive);
+        PyGILState_Release(state);
+    }
+
+    CaptivePythonObjectMixin(py::object captive) : CaptivePythonObjectMixin(captive.ptr()) {}
+
+    ~CaptivePythonObjectMixin() {
+        if (Py_IsInitialized()) {
+            PyGILState_STATE state = PyGILState_Ensure();
+            Py_XDECREF(_captive);
+            PyGILState_Release(state);
+        }
+    }
+};