python: fix listener ownership and isolate run config

python/opendht.pyx

@@ -539,7 +539,7 @@ cdef class TrustList(object):
 cdef class ListenToken(object):
     cdef cpp.InfoHash _h
     cdef cpp.shared_future[size_t] _t
-    _cb = dict()
+    cdef bool _cancelled
 
 cdef class Identity(object):
     cdef cpp.Identity _id
@@ -673,15 +673,18 @@ cdef class DhtRunner(_WithID):
         host_bytes = host.encode()
         port_bytes = port.encode() if port else b'4222'
         self.thisptr.get().bootstrap(<cpp.const_char*>host_bytes, <cpp.const_char*>port_bytes)
-    def run(self, Identity id=None, is_bootstrap=False, cpp.in_port_t port=0, str ipv4="", str ipv6="", DhtConfig config=DhtConfig()):
-        if id:
-            config.setIdentity(id)
+    def run(self, Identity id=None, is_bootstrap=False, cpp.in_port_t port=0, str ipv4="", str ipv6="", DhtConfig config=None):
+        cdef cpp.DhtRunnerConfig run_config = config._config if config is not None else cpp.DhtRunnerConfig()
+        if id is not None:
+            run_config.dht_config.id = id._id
+        if is_bootstrap:
+            run_config.dht_config.node_config.is_bootstrap = True
         if ipv4 or ipv6:
             bind4 = ipv4.encode() if ipv4 else b''
             bind6 = ipv6.encode() if ipv6 else b''
-            self.thisptr.get().run(bind4, bind6, str(port).encode(), config._config)
+            self.thisptr.get().run(bind4, bind6, str(port).encode(), run_config)
         else:
-            self.thisptr.get().run(port, config._config)
+            self.thisptr.get().run(port, run_config)
     def join(self):
         self.thisptr.get().join()
     def shutdown(self, shutdown_cb=None):
@@ -849,15 +852,14 @@ cdef class DhtRunner(_WithID):
         t = ListenToken()
         t._h = key._infohash
         cb_obj = {'valcb':value_cb}
-        t._cb['cb'] = cb_obj
-        # avoid the callback being destructed if the token is destroyed
-        ref.Py_INCREF(cb_obj)
-        t._t = self.thisptr.get().listen(t._h, cpp.bindValueCb(value_callback, <void*>cb_obj)).share()
+        t._t = self.thisptr.get().listen(t._h, cpp.bindPythonValueCb(value_callback, cb_obj)).share()
         return t
     def cancelListen(self, ListenToken token):
-        self.thisptr.get().cancelListen(token._h, token._t)
-        ref.Py_DECREF(<object>token._cb['cb'])
-        # fixme: not thread safe
+        if token is None:
+            raise TypeError("token must be a ListenToken")
+        if not token._cancelled and token._t.valid():
+            self.thisptr.get().cancelListen(token._h, token._t)
+            token._cancelled = True
 
 cdef class IndexValue(object):
     cdef cpp.shared_ptr[cpp.IndexValue] _value

python/opendht_cpp.pxd

@@ -243,9 +243,35 @@ cdef extern from "opendht/callbacks.h" namespace "dht":
         Config node_config
         Identity id
 
+cdef extern from *:
+    """
+    #include <Python.h>
+    #include <opendht/callbacks.h>
+
+    static dht::ValueCallback opendht_bind_python_value_cb(
+        dht::ValueCallbackRaw callback, PyObject* data)
+    {
+        Py_INCREF(data);
+        auto owner = std::shared_ptr<PyObject>(data, [](PyObject* object) {
+            auto state = PyGILState_Ensure();
+            Py_DECREF(object);
+            PyGILState_Release(state);
+        });
+        return [callback, owner = std::move(owner)](
+                   const std::vector<std::shared_ptr<dht::Value>>& values, bool expired) {
+            for (const auto& value : values)
+                if (!callback(value, expired, owner.get()))
+                    return false;
+            return true;
+        };
+    }
+    """
+    ValueCallback bindPythonValueCb "opendht_bind_python_value_cb"(ValueCallbackRaw callback, object data) except +
+
+ctypedef future[size_t] ListenToken
+ctypedef shared_future[size_t] SharedListenToken
+
 cdef extern from "opendht/dhtrunner.h" namespace "dht":
-    ctypedef future[size_t] ListenToken
-    ctypedef shared_future[size_t] SharedListenToken
     cdef cppclass DhtRunner:
         DhtRunner() except +
         cppclass Config:
@@ -285,8 +311,8 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
         void putEncrypted(InfoHash key, PkId to, shared_ptr[Value] val, DoneCallback done_cb, bool permanent)
         void putEncrypted(InfoHash key, shared_ptr[PublicKey] to, shared_ptr[Value] val, DoneCallback done_cb, bool permanent)
         void cancelPut(InfoHash key, shared_ptr[Value] val)
-        ListenToken listen(InfoHash key, ValueCallback get_cb)
-        void cancelListen(InfoHash key, SharedListenToken token)
+        ListenToken listen(InfoHash key, ValueCallback get_cb) except +
+        void cancelListen(InfoHash key, SharedListenToken token) except +
         vector[unsigned] getNodeMessageStats(bool i)
 
 ctypedef DhtRunner.Config DhtRunnerConfig