[Plasma] Segfault when retrieving RecordBatch from plasma store

[asfimport]

It can be reproduced with the following script:

import pyarrow as pa
import pyarrow.plasma as plasma

def retrieve1():
    client = plasma.connect('test', "", 0)

    key = "keynumber1keynumber1"
    pid = plasma.ObjectID(bytearray(key,'UTF-8'))

    [buff] = client .get_buffers([pid])
    batch = pa.RecordBatchStreamReader(buff).read_next_batch()

    print(batch)
    print(batch.schema)
    print(batch[0])

    return batch

client = plasma.connect('test', "", 0)

test1 = [1, 12, 23, 3, 21, 34]
test1 = pa.array(test1, pa.int32())

batch = pa.RecordBatch.from_arrays([test1], ['FIELD1'])

key = "keynumber1keynumber1"
pid = plasma.ObjectID(bytearray(key,'UTF-8'))
sink = pa.MockOutputStream()
stream_writer = pa.RecordBatchStreamWriter(sink, batch.schema)
stream_writer.write_batch(batch)
stream_writer.close()

bff = client.create(pid, sink.size())

stream = pa.FixedSizeBufferWriter(bff)
writer = pa.RecordBatchStreamWriter(stream, batch.schema)
writer.write_batch(batch)
client.seal(pid)

batch = retrieve1()
print(batch)
print(batch.schema)
print(batch[0])

 

Preliminary backtrace:

 

CESS (code=1, address=0x111138158)

    frame #0: 0x000000010e6457fc lib.so`__pyx_pw_7pyarrow_3lib_10Int32Value_1as_py(_object*, _object*) + 28

lib.so`__pyx_pw_7pyarrow_3lib_10Int32Value_1as_py:

->  0x10e6457fc <+28>: movslq (%rdx,%rcx,4), %rdi

    0x10e645800 <+32>: callq  0x10e698170               ; symbol stub for: PyInt_FromLong

    0x10e645805 <+37>: testq  %rax, %rax

    0x10e645808 <+40>: je     0x10e64580c               ; <+44>

(lldb) bt
 * thread #1: tid = 0xf1378e, 0x000000010e6457fc lib.so`__pyx_pw_7pyarrow_3lib_10Int32Value_1as_py(_object*, _object*) + 28, queue = 'com.apple.main-thread', stop reason = EXC_BAD_ACCESS (code=1, address=0x111138158)

  * frame #0: 0x000000010e6457fc lib.so`__pyx_pw_7pyarrow_3lib_10Int32Value_1as_py(_object*, _object*) + 28

    frame #1: 0x000000010e5ccd35 lib.so`__Pyx_PyObject_CallNoArg(_object*) + 133

    frame #2: 0x000000010e613b25 lib.so`__pyx_pw_7pyarrow_3lib_10ArrayValue_3__repr__(_object*) + 933

    frame #3: 0x000000010c2f83bc libpython2.7.dylib`PyObject_Repr + 60

    frame #4: 0x000000010c35f651 libpython2.7.dylib`PyEval_EvalFrameEx + 22305

Reporter: Philipp Moritz / @pcmoritz
Assignee: Antoine Pitrou / @pitrou

Related issues:

• [Plasma] PlasmaClient::Get() could take vector of object ids (supercedes)

PRs and other links:

• GitHub Pull Request #1807

_{Note: This issue was originally created as ARROW-2195. Please see the migration documentation for further details.}

[asfimport]

Antoine Pitrou / @pitrou:
The returned BatchRecord doesn't keep the PlasmaBuffer alive. In turn, PlasmaBuffer.__dealloc__ calls self.client.release(self.object_id). If you remove that line, the program runs fine.

[asfimport]

Antoine Pitrou / @pitrou:
The detailed explanation here is that the C++ shared_ptr is passed around, not the Python wrapper. So PlasmaBuffer.__dealloc__ is called even though there are still C++ exports of the underlying data, particularly in the BatchRecord object.

Incidentally, it seems ForeignBuffer will have the same problem: when the C++ shared_ptr is passed around, the original Python object won't be kept alive. Instead, perhaps we need to be able to add destructors to an arbitrary C++ Buffer object, for example:

Status Buffer::AddDestructor(std::function<void()>&& destructor);

Then it will be possible to keep a Python object alive as long as the corresponding C++ buffer is alive:

Status AddBufferPayload(std::shared_ptr<Buffer> buffer, PyObject* ref)
{
  RETURN_NOT_OK(buffer->AddDestructor([&ref]() { Py_DECREF(ref); }));
  Py_INCREF(ref);
  return Status::OK();
}

def foreign_buffer(addr, size, base):
    cdef shared_ptr[CBuffer] buffer = make_shared[CBuffer](
        <uint8_t*>c_addr, c_size)
    AddBufferPayload(buffer, base)
    return pyarrow_wrap_buffer(buffer)

cdef class _PlasmaObjectKeepAlive:
    cdef:
        ObjectID object_id
        PlasmaClient client

    def __cinit__(self, ObjectID object_id, PlasmaClient client):
        self.object_id = object_id
        self.client = client

    def __dealloc__(self):
        """
        Notify Plasma that the object is no longer needed.

        If the plasma client has been shut down, then don't do anything.
        """
        self.client.release(self.object_id)

cdef class PlasmaClient:

    cdef _make_plasma_buffer(self, ObjectID object_id,
                             shared_ptr[CBuffer] buffer, int64_t size):
        AddBufferPaylod(buffer, _PlasmaObjectKeepAlive(object_id, self)
        return pyarrow_wrap_buffer(buffer)

cc @xhochy

[asfimport]

Wes McKinney / @wesm:
Instead of adding a dtor to the base Buffer we can instead create an arrow::ForeignBuffer subclass with the extra dtor member

[asfimport]

Antoine Pitrou / @pitrou:
That would indeed solve the ForeignBuffer issue, but not the Plasma issue, right? That is, the solution only applies when the buffer is created from Python...

[asfimport]

Wes McKinney / @wesm:
Right probably need to make Plasma buffers work like the MemoryMap in https://github.com/apache/arrow/blob/master/cpp/src/arrow/io/file.cc#L593 that manages the lifetime of the mapped memory created by the MemoryMappedFile class

[asfimport]

Antoine Pitrou / @pitrou:

Right probably need to make Plasma buffers work like the MemoryMap in https://github.com/apache/arrow/blob/master/cpp/src/arrow/io/file.cc#L593 that manages the lifetime of the mapped memory created by the MemoryMappedFile class
Right now it seems C++ users are supposed to call PlasmaClient::Release explicitly when they are done with an object. Doing it automatically risks introducing double (or early) releases, am I wrong?

[asfimport]

Wes McKinney / @wesm:
adding @robertnishihara also

[asfimport]

Wes McKinney / @wesm:
Moving to 0.10.0

[asfimport]

Antoine Pitrou / @pitrou:
I think the right thing to do here is to add a new PlasmaClient::GetAuto method that returns automatically-releasing buffers.

The one caveat is that the buffers must keep a pointer to the plasma client, so the caller must guarantee that the client stays alive long enough. Unfortunately, the PlasmaClient constructor/factory currently doesn't return a shared_ptr...

[asfimport]

Antoine Pitrou / @pitrou:
By the way, does anyone know how to build the GPU extensions for Plasma?

[asfimport]

Antoine Pitrou / @pitrou:
Ok, apparently I just had to pass ARROW_GPU=ON to cmake after installing the CUDA toolkit :)

[asfimport]

Antoine Pitrou / @pitrou:
The Create() behavior is specific enough to warrant its own issue.

[asfimport]

Antoine Pitrou / @pitrou:
Opened ARROW-2363 for Create().

[asfimport]

Philipp Moritz / @pcmoritz:
Issue resolved by pull request 1807
#1807