pybind/rados: Fix error handling and leaks in aio

aio_read:
The reference to ret_s begins existing at PyBytes_FromStringAndSize and
is handed over to the callback if rados_aio_read succeeds. This creates
a lot of subtle scenarios where it might not be XDECREFed (e.g. if
a KeyboardInterrupt arrives at the wrong time). Instead, store the pointer
to that buffer in the completion object, and hand over responsibility for
the XDECREF to it. This guarantees that the "special" reference to this
incomplete object will be released when the completion object is
deallocated.

Also make sure we don't try to _PyBytes_Resize with a negative length.

Add a failure case to the aio_read test in test_rados.py

Completion: the wrapper methods weren't being called, which prevents
the completion objects from being freed until the Ioctx is. Fix this
and add a refcount check to the aio_read test.

Signed-off-by: Hector Martin <marcan@marcan.st>

diff --git a/src/pybind/rados/rados.pyx b/src/pybind/rados/rados.pyx
index ad21368aa04ccc9ac77b7616aebf44e4cd12d019..54fe245aab67986d6e2d750b74e9d53e45f2713d 100644 (file)
--- a/src/pybind/rados/rados.pyx
+++ b/src/pybind/rados/rados.pyx
@@ -40,7 +40,7 @@ else:
 cdef extern from "Python.h":
     # These are in cpython/string.pxd, but use "object" types instead of
     # PyObject*, which invokes assumptions in cpython that we need to
-    # legitimately break to implement zero-copy string buffers in Image.read().
+    # legitimately break to implement zero-copy string buffers in Ioctx.read().
     # This is valid use of the Python API and documented as a special case.
     PyObject *PyBytes_FromStringAndSize(char *v, Py_ssize_t len) except NULL
     char* PyBytes_AsString(PyObject *string) except NULL
@@ -1524,6 +1524,7 @@ cdef class Completion(object):
          rados_callback_t complete_cb
          rados_callback_t safe_cb
          rados_completion_t rados_comp
+         PyObject* buf
 
     def __cinit__(self, Ioctx ioctx, object oncomplete, object onsafe):
         self.oncomplete = oncomplete
@@ -1611,8 +1612,12 @@ cdef class Completion(object):
         Call this when you no longer need the completion. It may not be
         freed immediately if the operation is not acked and committed.
         """
-        with nogil:
-            rados_aio_release(self.rados_comp)
+        ref.Py_XDECREF(self.buf)
+        self.buf = NULL
+        if self.rados_comp != NULL:
+            with nogil:
+                rados_aio_release(self.rados_comp)
+                self.rados_comp = NULL
 
     def _complete(self):
         self.oncomplete(self)
@@ -1681,7 +1686,7 @@ cdef int __aio_safe_cb(rados_completion_t completion, void *args) with gil:
     Callback to onsafe() for asynchronous operations
     """
     cdef object cb = <object>args
-    cb.onsafe(cb)
+    cb._safe()
     return 0
 
 
@@ -1690,7 +1695,7 @@ cdef int __aio_complete_cb(rados_completion_t completion, void *args) with gil:
     Callback to oncomplete() for asynchronous operations
     """
     cdef object cb = <object>args
-    cb.oncomplete(cb)
+    cb._complete()
     return 0
 
 
@@ -1930,34 +1935,24 @@ cdef class Ioctx(object):
 
             char *ref_buf
             size_t _length = length
-            PyObject* ret_s = NULL
 
         def oncomplete_(completion_v):
-            try:
-                return_value = completion_v.get_return_value()
-                if return_value != length:
-                    _PyBytes_Resize(&ret_s, return_value)
-                return oncomplete(completion_v, <object>ret_s if return_value >= 0 else None)
-            finally:
-                # We DECREF unconditionally: the cast to object above will have
-                # INCREFed if necessary. This also takes care of exceptions,
-                # including if _PyString_Resize fails (that will free the string
-                # itself and set ret_s to NULL, hence XDECREF).
-                ref.Py_XDECREF(ret_s)
+            cdef Completion _completion_v = completion_v
+            return_value = _completion_v.get_return_value()
+            if return_value > 0 and return_value != length:
+                _PyBytes_Resize(&_completion_v.buf, return_value)
+            return oncomplete(_completion_v, <object>_completion_v.buf if return_value >= 0 else None)
 
         completion = self.__get_completion(oncomplete_, None)
-        ret_s = PyBytes_FromStringAndSize(NULL, length)
-        try:
-            ret_buf = PyBytes_AsString(ret_s)
-            with nogil:
-                ret = rados_aio_read(self.io, _object_name, completion.rados_comp,
-                                    ret_buf, _length, _offset)
-            if ret < 0:
-                raise make_ex(ret, "error reading %s" % object_name)
-            self.__track_completion(completion)
-            return completion
-        except Exception:
-            ref.Py_XDECREF(ret_s)
+        completion.buf = PyBytes_FromStringAndSize(NULL, length)
+        ret_buf = PyBytes_AsString(completion.buf)
+        with nogil:
+            ret = rados_aio_read(self.io, _object_name, completion.rados_comp,
+                                ret_buf, _length, _offset)
+        if ret < 0:
+            raise make_ex(ret, "error reading %s" % object_name)
+        self.__track_completion(completion)
+        return completion
 
     def aio_remove(self, object_name, oncomplete=None, onsafe=None):
         """

diff --git a/src/test/pybind/test_rados.py b/src/test/pybind/test_rados.py
index 5549d695cce418fc53996c41f5e61a2a749f550e..c1ce5fc6ebced5998ac8123ee80e24da06a87616 100644 (file)
--- a/src/test/pybind/test_rados.py
+++ b/src/test/pybind/test_rados.py
@@ -649,6 +649,7 @@ class TestIoctx(object):
         assert(loops <= 10)
 
         eq(retval[0], payload)
+        eq(sys.getrefcount(comp), 2)
 
         # test2: use wait_for_complete_and_cb(), verify retval[0] is
         # set by the time we regain control
@@ -666,6 +667,25 @@ class TestIoctx(object):
         comp.wait_for_complete_and_cb()
         assert(retval[0] is not None)
         eq(retval[0], payload)
+        eq(sys.getrefcount(comp), 2)
+
+        # test3: error case, use wait_for_complete_and_cb(), verify retval[0] is
+        # set by the time we regain control
+
+        retval[0] = 1
+        self._take_down_acting_set('test_pool', 'bar')
+        comp = self.ioctx.aio_read("bar", len(payload), 0, cb)
+        eq(False, comp.is_complete())
+        time.sleep(3)
+        eq(False, comp.is_complete())
+        with lock:
+            eq(1, retval[0])
+        self._let_osds_back_up()
+
+        comp.wait_for_complete_and_cb()
+        eq(None, retval[0])
+        assert(comp.get_return_value() < 0)
+        eq(sys.getrefcount(comp), 2)
 
         [i.remove() for i in self.ioctx.list_objects()]