Skip to content

Commit bdbf630

Browse files
xhochywesm
authored andcommitted
ARROW-4582: [Python/C++] Acquire the GIL on Py_INCREF
Minimal reproducing example: ``` import dask import pandas as pd import pyarrow as pa import numpy as np def segfault_me(df): pa.Table.from_pandas(df, nthreads=1) while True: df = pd.DataFrame( {"P": np.arange(0, 10), "L": np.arange(0, 10), "TARGET": np.arange(10, 20)} ) dask.compute([ dask.delayed(segfault_me)(df), dask.delayed(segfault_me)(df), dask.delayed(segfault_me)(df), dask.delayed(segfault_me)(df), dask.delayed(segfault_me)(df), ]) ``` Segfaults are more likely when run in AddressSanitizer or otherwise slow system with many cores. It is important that always the same df is passed into the functions. The issue was that the reference count of the underlying NumPy array was increased at the same time by multiple threads. The decrease happend then with a GIL, so the array was sometimes destroyed while still used. Author: Korn, Uwe <Uwe.Korn@blue-yonder.com> Closes #3655 from xhochy/ARROW-4582 and squashes the following commits: 7f9838d <Korn, Uwe> docker-compose run clang-format 3d6e5ee <Korn, Uwe> ARROW-4582:  Acquire the GIL on Py_INCREF
1 parent 09cb71c commit bdbf630

File tree

1 file changed

+18
-15
lines changed

1 file changed

+18
-15
lines changed

cpp/src/arrow/python/numpy_convert.cc

Lines changed: 18 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@ bool is_contiguous(PyObject* array) {
4646
}
4747

4848
NumPyBuffer::NumPyBuffer(PyObject* ao) : Buffer(nullptr, 0) {
49+
PyAcquireGIL lock;
4950
arr_ = ao;
5051
Py_INCREF(ao);
5152

@@ -187,8 +188,6 @@ Status NumPyDtypeToArrow(PyArray_Descr* descr, std::shared_ptr<DataType>* out) {
187188
#undef TO_ARROW_TYPE_CASE
188189

189190
Status NdarrayToTensor(MemoryPool* pool, PyObject* ao, std::shared_ptr<Tensor>* out) {
190-
PyAcquireGIL lock;
191-
192191
if (!PyArray_Check(ao)) {
193192
return Status::TypeError("Did not pass ndarray object");
194193
}
@@ -199,25 +198,29 @@ Status NdarrayToTensor(MemoryPool* pool, PyObject* ao, std::shared_ptr<Tensor>*
199198

200199
int ndim = PyArray_NDIM(ndarray);
201200

201+
// This is also holding the GIL, so don't already draw it.
202202
std::shared_ptr<Buffer> data = std::make_shared<NumPyBuffer>(ao);
203203
std::vector<int64_t> shape(ndim);
204204
std::vector<int64_t> strides(ndim);
205205

206-
npy_intp* array_strides = PyArray_STRIDES(ndarray);
207-
npy_intp* array_shape = PyArray_SHAPE(ndarray);
208-
for (int i = 0; i < ndim; ++i) {
209-
if (array_strides[i] < 0) {
210-
return Status::Invalid("Negative ndarray strides not supported");
206+
{
207+
PyAcquireGIL lock;
208+
npy_intp* array_strides = PyArray_STRIDES(ndarray);
209+
npy_intp* array_shape = PyArray_SHAPE(ndarray);
210+
for (int i = 0; i < ndim; ++i) {
211+
if (array_strides[i] < 0) {
212+
return Status::Invalid("Negative ndarray strides not supported");
213+
}
214+
shape[i] = array_shape[i];
215+
strides[i] = array_strides[i];
211216
}
212-
shape[i] = array_shape[i];
213-
strides[i] = array_strides[i];
214-
}
215217

216-
std::shared_ptr<DataType> type;
217-
RETURN_NOT_OK(
218-
GetTensorType(reinterpret_cast<PyObject*>(PyArray_DESCR(ndarray)), &type));
219-
*out = std::make_shared<Tensor>(type, data, shape, strides);
220-
return Status::OK();
218+
std::shared_ptr<DataType> type;
219+
RETURN_NOT_OK(
220+
GetTensorType(reinterpret_cast<PyObject*>(PyArray_DESCR(ndarray)), &type));
221+
*out = std::make_shared<Tensor>(type, data, shape, strides);
222+
return Status::OK();
223+
}
221224
}
222225

223226
Status TensorToNdarray(const std::shared_ptr<Tensor>& tensor, PyObject* base,

0 commit comments

Comments
 (0)