Skip to content

Commit

Permalink
Add async.h
Browse files Browse the repository at this point in the history
  • Loading branch information
lidavidm committed Aug 18, 2023
1 parent 0bd8c44 commit 0cc328e
Show file tree
Hide file tree
Showing 3 changed files with 54 additions and 20 deletions.
6 changes: 4 additions & 2 deletions python/pyarrow/includes/common.pxd
Original file line number Diff line number Diff line change
Expand Up @@ -157,12 +157,14 @@ cdef extern from "arrow/util/future.h" namespace "arrow" nogil:
ctypedef object PyWrapper(void*)


cdef extern from "arrow/python/async.h" namespace "arrow::py" nogil:
void BindFuture[T](CFuture[T], object cb, PyWrapper wrapper)


cdef extern from "arrow/python/common.h" namespace "arrow::py" nogil:
T GetResultValue[T](CResult[T]) except *
cdef function[F] BindFunction[F](void* unbound, object bound, ...)

void BindFuture[T](CFuture[T], object cb, PyWrapper wrapper)


cdef inline object PyObject_to_object(PyObject* o):
# Cast to "object" increments reference count
Expand Down
50 changes: 50 additions & 0 deletions python/pyarrow/src/arrow/python/async.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.

#pragma once

#include <utility>

#include "arrow/python/common.h"
#include "arrow/status.h"
#include "arrow/util/future.h"

namespace arrow::py {
/// \brief Bind a Python callback to an arrow::Future.
///
/// \param future The future to bind to.
/// \param py_cb The Python callback function. Will be passed the result of
/// py_wrapper, or a Python exception if the future failed or one was
/// raised by py_wrapper.
/// \param py_wrapper A function (likely defined in Cython) to convert the C++
/// result of the future to a Python object.
template <typename T, typename Wrapper = PyObject* (*)(void*)>
void BindFuture(Future<T> future, PyObject* py_cb, Wrapper py_wrapper) {
Py_INCREF(py_cb);
OwnedRefNoGIL cb_ref(py_cb);

auto future_cb = [cb_ref = std::move(cb_ref), py_wrapper](Result<T> result) {
SafeCallIntoPythonVoid([&]() {
OwnedRef py_value_or_exc{WrapResult(std::move(result), std::move(py_wrapper))};
Py_XDECREF(
PyObject_CallFunctionObjArgs(cb_ref.obj(), py_value_or_exc.obj(), NULLPTR));
ARROW_WARN_NOT_OK(CheckPyError(), "Internal error in async call");
});
};
future.AddCallback(std::move(future_cb));
}
} // namespace arrow::py
18 changes: 0 additions & 18 deletions python/pyarrow/src/arrow/python/common.h
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
#include "arrow/python/pyarrow.h"
#include "arrow/python/visibility.h"
#include "arrow/result.h"
#include "arrow/util/future.h"
#include "arrow/util/macros.h"

namespace arrow {
Expand Down Expand Up @@ -288,23 +287,6 @@ std::function<OutFn> BindFunction(Return (*unbound)(PyObject*, Args...),
[bound_fn](Args... args) { return bound_fn->Invoke(std::forward<Args>(args)...); };
}

// XXX Put this in arrow/python/async.h to avoid adding more random stuff here?
template <typename T, typename Wrapper = PyObject* (*)(void*)>
void BindFuture(Future<T> future, PyObject* py_cb, Wrapper py_wrapper) {
Py_INCREF(py_cb);
OwnedRefNoGIL cb_ref(py_cb);

auto future_cb = [cb_ref = std::move(cb_ref), py_wrapper](Result<T> result) {
SafeCallIntoPythonVoid([&]() {
OwnedRef py_value_or_exc{WrapResult(std::move(result), std::move(py_wrapper))};
Py_XDECREF(
PyObject_CallFunctionObjArgs(cb_ref.obj(), py_value_or_exc.obj(), NULLPTR));
ARROW_WARN_NOT_OK(CheckPyError(), "Internal error in async call");
});
};
future.AddCallback(std::move(future_cb));
}

// A temporary conversion of a Python object to a bytes area.
struct PyBytesView {
const char* bytes;
Expand Down

0 comments on commit 0cc328e

Please sign in to comment.