Ceph

Call aio_execute of custom class op.

python client code:

  mypool.aio_execute(obj_name, 'xxx', 'yyy', data, length=16384, oncomplete=my_oncomplete())

Cython code allocates PyBytes and provide raw pointer and length to librados rados_aio_exec.

rados.pyx:

  def aio_execute:
    Completion completion
    ...
    completion = self.__get_completion(oncomplete_ if oncomplete else None, onsafe_ if onsafe else None)
    completion.buf = PyBytes_FromStringAndSize(NULL, length)
    ret_buf = PyBytes_AsString(completion.buf)
    ...
    rados_aio_exec(..., completion.rados_comp, ..., ret_buf, length)

IoCtxImpl::aio_exec() initialize bufferlist with raw pointer and provided length,
and set same raw pointer to out_buf.

IoCtxImpl.cc:

  librados::IoCtxImpl::aio_exec(..., c, ..., buf, out_len):
    ...
    c->bl.clear();
    c->bl.push_back(buffer::create_static(out_len, buf));
    c->blp = &c->bl;
    c->out_buf = buf;
    ...

    (gdb) p c->out_buf
    $6 = 0x10c0b80 "\344\061\\xfb\\x" 

    (gdb) p c->bl
    $7 = {
      _buffers = std::__cxx11::list = {
        [0] = {
          _raw = 0x10b5530,
          _off = 0,
          _len = 16384
        }
      },
      _len = 16384,
      _memcopy_count = 0,
      append_buffer = {
        _raw = 0x0,
        _off = 0,
        _len = 0
      },
      last_p = {
        <ceph::buffer::list::iterator_impl<false>> = {
          <std::iterator<std::forward_iterator_tag, char, long, char*, char&>> = {<No data fields>},
          members of ceph::buffer::list::iterator_impl<false>:
          bl = 0xd8eb58,
          ls = 0xd8eb58,
          off = 0,
          p = {
      _raw = 0x1,
      _off = 16384,
      _len = 0
    },
          p_off = 0
        }, <No data fields>},
      static CLAIM_DEFAULT = 0,
      static CLAIM_ALLOW_NONSHAREABLE = 1
    }

    (gdb) p *((ceph::buffer::raw*)0x10b5530)
    $5 = (ceph::buffer::raw_static) {
      <ceph::buffer::raw> = {
        _vptr.raw = 0x7ffff6539b60 <vtable for ceph::buffer::raw_static+16>,
        data = 0x10c0b80 "\344\061\\xfb\\x",
        len = 16384,
        nref = {
          <std::__atomic_base<unsigned int>> = {
            static _S_alignment = 4,
            _M_i = 1
          }, <No data fields>},
        mempool = 10,
        crc_spinlock = {
          <std::__atomic_flag_base> = {
            _M_i = false
          }, <No data fields>},
        crc_map = std::map with 0 elements
      }, <No data fields>}

C_aio_Complete::finish() checks if bufferlist contains out_buf pointer,
but is_provided_buffer also checks if bufferlist is contiguous.
So if operation returns more data than length of buffer - than bufferlist will be extended
and is_provided_buffer() will return false.
So copy() will be called and overwrites malloc metadata.
And return value will be set to greater than length of provided buffer.

IoCtxImpl.c:

  librados::IoCtxImpl::C_aio_Complete::finish(...):
    ...
    if (r == 0 && c->blp && c->blp->length() > 0) {
      if (c->out_buf && !c->blp->is_provided_buffer(c->out_buf))
        c->blp->copy(0, c->blp->length(), c->out_buf);
      c->rval = c->blp->length();
    }
    ...
    c->io->client->finisher.queue(new C_AioComplete(c));
    ...

    (gdb) p c->bl
        $1 = {
          _buffers = std::__cxx11::list = {
            [0] = {
              _raw = 0x10b5530,
              _off = 0,
              _len = 16384
            },
            [1] = {
              _raw = 0x7fffcc00a150,
              _off = 0,
              _len = 3680
            }
          },
          _len = 20064,
          _memcopy_count = 0,
          append_buffer = {
            _raw = 0x0,
            _off = 0,
            _len = 0
          },
          last_p = {
            <ceph::buffer::list::iterator_impl<false>> = {
              <std::iterator<std::forward_iterator_tag, char, long, char*, char&>> = {<No data fields>},
              members of ceph::buffer::list::iterator_impl<false>:
              bl = 0xd8eb58,
              ls = 0xd8eb58,
              off = 0,
              p = {
          _raw = 0x2,
          _off = 20064,
          _len = 0
        },
              p_off = 0
            }, <No data fields>},
          static CLAIM_DEFAULT = 0,
          static CLAIM_ALLOW_NONSHAREABLE = 1
        }

_PyBytes_Resize() tries to resize buffer that is already broken.

rados.pyx:

  def aio_execute:
    def oncomplete_:
    ...
      if return_value > 0 and return_value != length:
          _PyBytes_Resize(&_completion_v.buf, return_value)
    ...