kSuccess,

kInterrupt,

kTimeout,

public:

ScopedEvent() {

sigint_event = _PyOS_SigintEvent();

assert(sigint_event != nullptr);

handle = ::CreateEventA(/*lpEventAttributes=*/nullptr,

/*bManualReset=*/TRUE,

/*bInitialState=*/FALSE,

/*lpName=*/nullptr);

assert(handle != nullptr);

}

~ScopedEvent() { ::CloseHandle(handle); }

void Set() { ::SetEvent(handle); }

ScopedEventWaitResult Wait(absl::Time deadline) {

const HANDLE handles[2] = {handle, sigint_event};

DWORD timeout;

if (deadline == absl::InfiniteFuture()) {

timeout = INFINITE;

} else {

int64_t ms = absl::ToInt64Milliseconds(deadline - absl::Now());

ms = std::max(int64_t(0), ms);

timeout =

static_cast<DWORD>(std::min(ms, static_cast<int64_t>(INFINITE)));

}

DWORD res = ::WaitForMultipleObjectsEx(2, handles, /*bWaitAll=*/FALSE,

/*dwMilliseconds=*/timeout,

/*bAlertable=*/FALSE);

if (res == WAIT_OBJECT_0 + 1) {

::ResetEvent(sigint_event);

return ScopedEventWaitResult::kInterrupt;

} else if (res == WAIT_OBJECT_0) {

return ScopedEventWaitResult::kSuccess;

} else {

assert(res == WAIT_TIMEOUT);

return ScopedEventWaitResult::kTimeout;

}

}

HANDLE handle;

HANDLE sigint_event;

public:

ScopedEvent() {

{

[[maybe_unused]] int err = ::pthread_mutex_init(&mutex, nullptr);

assert(err == 0);

}

{

[[maybe_unused]] int err = ::pthread_cond_init(&cond, nullptr);

assert(err == 0);

}

}

~ScopedEvent() {

{

[[maybe_unused]] int err = ::pthread_cond_destroy(&cond);

assert(err == 0);

}

{

[[maybe_unused]] int err = ::pthread_mutex_destroy(&mutex);

assert(err == 0);

}

}

void Set() {

{

[[maybe_unused]] int err = ::pthread_mutex_lock(&mutex);

assert(err == 0);

}

set = true;

{

[[maybe_unused]] int err = ::pthread_mutex_unlock(&mutex);

assert(err == 0);

}

::pthread_cond_signal(&cond);

}

ScopedEventWaitResult Wait(absl::Time deadline) {

{

[[maybe_unused]] int err = ::pthread_mutex_lock(&mutex);

assert(err == 0);

}

bool set_value = set;

bool timeout = false;

if (!set_value) {

if (deadline == absl::InfiniteFuture()) {

::pthread_cond_wait(&cond, &mutex);

} else {

const auto tspec = ToTimespec(deadline);

timeout = ::pthread_cond_timedwait(&cond, &mutex, &tspec) == ETIMEDOUT;

}

set_value = set;

}

{

[[maybe_unused]] int err = ::pthread_mutex_unlock(&mutex);

assert(err == 0);

}

return set_value ? ScopedEventWaitResult::kSuccess

: (timeout ? ScopedEventWaitResult::kTimeout

: ScopedEventWaitResult::kInterrupt);

}

bool set{false};

::pthread_mutex_t mutex;

::pthread_cond_t cond;

public:

ScopedEvent() {

[[maybe_unused]] int err = ::sem_init(&sem, /*pshared=*/0, 0);

assert(err == 0);

}

~ScopedEvent() {

[[maybe_unused]] int err = ::sem_destroy(&sem);

assert(err == 0);

}

void Set() {

[[maybe_unused]] int err = ::sem_post(&sem);

assert(err == 0);

}

ScopedEventWaitResult Wait(absl::Time deadline) {

if (deadline == absl::InfiniteFuture()) {

if (::sem_wait(&sem) == 0) return ScopedEventWaitResult::kSuccess;

assert(errno == EINTR);

} else {

const auto tspec = absl::ToTimespec(deadline);

if (::sem_timedwait(&sem, &tspec) == 0) {

return ScopedEventWaitResult::kSuccess;

}

assert(errno == EINTR || errno == ETIMEDOUT);

if (errno == ETIMEDOUT) return ScopedEventWaitResult::kTimeout;

}

return ScopedEventWaitResult::kInterrupt;

}

::sem_t sem;

public:

ScopedFutureCallbackRegistration(FutureCallbackRegistration registration)

: registration_(std::move(registration)) {}

~ScopedFutureCallbackRegistration() { registration_.Unregister(); }

private:

FutureCallbackRegistration registration_;

using CancelCallbackBase = PythonFutureObject::CancelCallbackBase;

using Accessor =

internal::intrusive_linked_list::MemberAccessor<CancelCallbackBase>;

explicit CancelCallback(PythonFutureObject* base,

absl::FunctionRef<void()> callback)

: callback(callback) {

internal::intrusive_linked_list::InsertBefore(

Accessor{}, &base->cpp_data.cancel_callbacks, this);

}

~CancelCallback() {

internal::intrusive_linked_list::Remove(Accessor{}, this);

}

absl::FunctionRef<void()> callback;

PythonFutureObject* python_future) {

if (future.ready()) return;

{

GilScopedRelease gil_release;

future.Force();

}

ScopedEvent event;

const auto notify_done = [&event] { event.Set(); };

std::optional<CancelCallback> cancel_callback;

if (python_future) {

absl::MutexLock lock(python_future->cpp_data.mutex);

cancel_callback.emplace(python_future, notify_done);

}

ScopedFutureCallbackRegistration registration{

future.UntypedExecuteWhenReady([&event](auto future) { event.Set(); })};

while (true) {

ScopedEventWaitResult wait_result;

{

GilScopedRelease gil_release;

wait_result = event.Wait(deadline);

}

switch (wait_result) {

case ScopedEventWaitResult::kSuccess:

if (python_future) {

absl::MutexLock lock(python_future->cpp_data.mutex);

cancel_callback = std::nullopt;

if (StateIsCancelled(python_future->cpp_data.state)) {

ThrowCancelledError();

}

}

return;

case ScopedEventWaitResult::kTimeout:

if (python_future) {

absl::MutexLock lock(python_future->cpp_data.mutex);

cancel_callback = std::nullopt;

}

ThrowTimeoutError();

case ScopedEventWaitResult::kInterrupt:

break;

}

if (PyErr_CheckSignals() == -1) {

if (python_future) {

absl::MutexLock lock(python_future->cpp_data.mutex);

cancel_callback = std::nullopt;

}

throw py::error_already_set();

}

}

std::vector<pybind11::object> done_callbacks;

{

absl::MutexLock lock(cpp_data.mutex);

if (StateIsDone(cpp_data.state)) return false;

cpp_data.state = {};

cpp_data.registration.Unregister();

done_callbacks = std::move(cpp_data.callbacks);

// Run cancel callbacks. Note that the only user of cancel callbacks

// is InterruptibleWaitImpl, which is safe to run with locks held.

for (CancelCallbackBase* callback = cpp_data.cancel_callbacks.next;

callback != &cpp_data.cancel_callbacks;) {

auto* next = callback->next;

static_cast<CancelCallback*>(callback)->callback();

callback = next;

}

}

RunCallbacks(std::move(done_callbacks));

return true;

// Use copy of `state`, since `state` may be modified by another thread

// calling `Cancel` once GIL is released.

internal_future::FutureStatePointer state;

{

absl::MutexLock lock(cpp_data.mutex);

if (StateIsDone(cpp_data.state)) return;

state = cpp_data.state;

}

GilScopedRelease gil_release;

state->Force();

// Logically, `done_callback` needs to capture `awaitable_future`. However,

// lambda captures don't interoperate with Python garbage collection.

// Instead, we create it as a capture-less function and then use a

// `PyMethod` object to capture `awaitable_future` as the `self` argument.

auto done_callback = py::cpp_function([](py::handle awaitable_future,

py::handle source_future) {

awaitable_future.attr("get_loop")().attr("call_soon_threadsafe")(

py::cpp_function([](py::handle source_future,

py::object awaitable_future) {

if (awaitable_future.attr("done")().ptr() == Py_True) {

return;

}

if (source_future.attr("cancelled")().ptr() == Py_True) {

awaitable_future.attr("cancel")();

return;

}

auto exc = source_future.attr("exception")();

if (!exc.is_none()) {

awaitable_future.attr("set_exception")(std::move(exc));

} else {

awaitable_future.attr("set_result")(source_future.attr("result")());

}

}),

source_future, awaitable_future);

});

py::object awaitable_future =

python_imports.asyncio_get_event_loop_function().attr("create_future")();

// Ensure the PythonFutureObject is cancelled if the awaitable future is

// cancelled.

auto cancel_callback = py::cpp_function(

[](py::handle source_future, py::handle awaitable_future) {

reinterpret_cast<PythonFutureObject*>(source_future.ptr())->Cancel();

});

auto bound_cancel_callback = py::reinterpret_steal<py::object>(

PyMethod_New(cancel_callback.ptr(), reinterpret_cast<PyObject*>(this)));

if (!bound_cancel_callback) throw py::error_already_set();

awaitable_future.attr("add_done_callback")(bound_cancel_callback);

// Ensure the awaitable future is marked ready once the PythonFutureObject

// becomes ready.

auto bound_done_callback = py::reinterpret_steal<py::object>(

PyMethod_New(done_callback.ptr(), awaitable_future.ptr()));

if (!bound_done_callback) throw py::error_already_set();

AddDoneCallback(bound_done_callback);

return awaitable_future.attr("__await__")();

internal_future::FutureStatePointer state;

{

absl::MutexLock lock(cpp_data.mutex);

state = cpp_data.state;

}

if (StateIsCancelled(state)) ThrowCancelledError();

auto any_future = internal_future::FutureAccess::Construct<AnyFuture>(state);

InterruptibleWaitImpl(any_future, deadline, this);

return vtable->get_result(*state);

internal_future::FutureStatePointer state;

{

absl::MutexLock lock(cpp_data.mutex);

if (StateIsCancelled(cpp_data.state)) ThrowCancelledError();

state = cpp_data.state;

}

auto any_future = internal_future::FutureAccess::Construct<AnyFuture>(state);

InterruptibleWaitImpl(any_future, deadline, this);

return vtable->get_exception(*state);

internal_future::FutureStatePointer state;

{

absl::MutexLock lock(cpp_data.mutex);

if (StateIsCancelled(cpp_data.state)) return absl::CancelledError("");

state = cpp_data.state;

}

return vtable->get_python_value_future(*state);

absl::ReleasableMutexLock lock(cpp_data.mutex);

if (StateIsDone(cpp_data.state)) {

lock.Release();

callback(py::handle(reinterpret_cast<PyObject*>(this)));

return;

}

cpp_data.callbacks.push_back(py::reinterpret_borrow<py::object>(callback));

if (cpp_data.callbacks.size() == 1) {

Py_INCREF(reinterpret_cast<PyObject*>(this));

internal_future::FutureStatePointer state = cpp_data.state;

lock.Release();

// Minimal implementation of `Force`

GilScopedRelease gil_release;

state->Force();

}

bool do_decref = false;

size_t num_removed = 0;

{

absl::MutexLock lock(cpp_data.mutex);

auto& callbacks = cpp_data.callbacks;

// Since caller owns a reference to `callback`, we can be sure that removing

// `callback` from `callbacks` does not result in any reference counts

// reaching zero, and therefore we can be sure that `*this` is not modified.

auto it = std::remove_if(

callbacks.begin(), callbacks.end(),

[&](pybind11::handle h) { return h.ptr() == callback.ptr(); });

num_removed = callbacks.end() - it;

callbacks.erase(it, callbacks.end());

do_decref = num_removed && callbacks.empty();

}

if (do_decref) {

Py_DECREF(reinterpret_cast<PyObject*>(this));

}

return num_removed;

std::vector<pybind11::object> done_callbacks) {

if (done_callbacks.empty()) return;

// If this object has already been finalized, then it is not safe to call

// callbacks, because they may now be in an invalid state due to garbage

// collection cycle breaking.

if (!PyObject_GC_IsFinalized(reinterpret_cast<PyObject*>(this))) {

for (py::handle callback : done_callbacks) {

if (PyObject* callback_result = PyObject_CallFunctionObjArgs(

callback.ptr(), reinterpret_cast<PyObject*>(this), nullptr)) {

Py_DECREF(callback_result);

continue;

}

PyErr_WriteUnraisable(nullptr);

PyErr_Clear();

}

}

Py_DECREF(reinterpret_cast<PyObject*>(this));

std::optional<double> deadline) {

absl::Time deadline_time = absl::InfiniteFuture();

if (deadline) {

deadline_time = absl::UnixEpoch() + absl::Seconds(*deadline);

}

if (timeout) {

deadline_time =

std::min(deadline_time, absl::Now() + absl::Seconds(*timeout));

}

return deadline_time;

pybind11::handle loop) {

if (Py_TYPE(src.ptr()) == PythonFutureObject::python_type) {

return py::reinterpret_borrow<py::object>(src);

}

if (python_imports.asyncio_iscoroutine_function(src).ptr() != Py_True) {

return {};

}

if (loop.is_none()) {

throw py::value_error(

"no event loop specified and thread does not have a default event "

"loop");

}

auto asyncio_future =

python_imports.asyncio_run_coroutine_threadsafe_function(src, loop);

auto pair = PromiseFuturePair<GilSafePythonValueOrExceptionWeakRef>::Make();

// Create Python future wrapper before adding `done_callback`, to ensure that

// any references are retained by the `PythonObjectReferenceManager` of

// `future`.

auto future = PythonFutureObject::Make(std::move(pair.future));

// Register done callback.

py::object done_callback =

py::cpp_function([promise = pair.promise](py::object future_obj) {

py::object result;

if (py::object method = py::reinterpret_steal<py::object>(

PyObject_GetAttrString(future_obj.ptr(), "result"));

method.ptr()) {

result = py::reinterpret_steal<py::object>(

PyObject_CallFunctionObjArgs(method.ptr(), nullptr));

}

PythonValueOrException value =

result ? PythonValueOrException(std::move(result))

: PythonValueOrException::FromErrorIndicator();

PythonObjectReferenceManager manager;

PythonValueOrExceptionWeakRef value_weak_ref(manager, value);

// Release the GIL when invoking `promise.SetResult` in order to avoid

// blocking other Python threads while arbitrary C++ callbacks are run.

{

GilScopedRelease gil_release;

promise.SetResult(std::move(value_weak_ref));

}

});

asyncio_future.attr("add_done_callback")(done_callback);

// Register cancellation handler.

pair.promise.ExecuteWhenNotNeeded(

[asyncio_future = asyncio_future.release().ptr()] {

ExitSafeGilScopedAcquire gil;

if (!gil.acquired()) return;

// Invoke `cancel` method.

if (auto method = py::reinterpret_steal<py::object>(

PyObject_GetAttrString(asyncio_future, "cancel"));

!method.ptr()) {

// Ignore error obtaining `cancel` method.

PyErr_WriteUnraisable(nullptr);

PyErr_Clear();

} else if (!py::reinterpret_steal<py::object>(

PyObject_CallFunctionObjArgs(method.ptr(), nullptr))

.ptr()) {

// Ignore error calling `cancel` method.

PyErr_WriteUnraisable(nullptr);

PyErr_Clear();

}

Py_DECREF(asyncio_future);

});

return future;

auto& obj = *reinterpret_cast<PythonFutureObject*>(self);

auto& cpp_data = obj.cpp_data;

// Ensure object is not tracked by garbage collector before invalidating

// invariants during destruction.

PyObject_GC_UnTrack(self);

if (obj.weakrefs) PyObject_ClearWeakRefs(self);

// Clear `state`: this ensures that the callback corresponding to

// `registration` does not run with the reference count equal to 0.

FutureCallbackRegistration do_unregister;

PythonObjectReferenceManager refs;

{

absl::MutexLock lock(cpp_data.mutex);

cpp_data.state = {};

do_unregister = std::exchange(cpp_data.registration, {});

refs = std::move(cpp_data.reference_manager);

// There can be no live callbacks, otherwise the PyObject would be live.

assert(cpp_data.callbacks.empty());

}

if (cpp_data.weak_future_pointer) {

// Signal that the future is no longer live.

absl::MutexLock lock(cpp_data.weak_future_pointer->mutex);

cpp_data.weak_future_pointer->future = nullptr;

}

{

GilScopedRelease gil_release;

do_unregister.Unregister();

}

refs.Clear();

// Deallocate and free after clearing references.

cpp_data.~CppData();

PyTypeObject* type = Py_TYPE(self);

type->tp_free(self);

Py_DECREF(type);

auto& cpp_data = reinterpret_cast<PythonFutureObject*>(self)->cpp_data;

absl::MutexLock lock(cpp_data.mutex);

Py_VISIT(Py_TYPE(self));

for (const auto& obj : cpp_data.callbacks) {

Py_VISIT(obj.ptr());

}

return cpp_data.reference_manager.Traverse(visit, arg);

auto& cpp_data = reinterpret_cast<PythonFutureObject*>(self)->cpp_data;

FutureCallbackRegistration do_unregister;

PythonObjectReferenceManager refs;

std::vector<pybind11::object> done_callbacks;

{

absl::MutexLock lock(cpp_data.mutex);

cpp_data.state = {};

do_unregister = std::exchange(cpp_data.registration, {});

refs = std::move(cpp_data.reference_manager);

done_callbacks = std::move(cpp_data.callbacks);

}

// Signal that the future is no longer live.

if (cpp_data.weak_future_pointer) {

absl::MutexLock lock(cpp_data.weak_future_pointer->mutex);

cpp_data.weak_future_pointer->future = nullptr;

}

{

GilScopedRelease gil_release;

do_unregister.Unregister();

}

if (!done_callbacks.empty()) {

done_callbacks.clear();

Py_DECREF(self);

}

refs.Clear();

return 0;

const char* doc = R"(

)";

static const PyMethodDef future_methods[] = {

{"__class_getitem__", Py_GenericAlias, METH_O | METH_CLASS, ""},

{nullptr},

};

PyType_Slot slots[] = {

{Py_tp_doc, const_cast<char*>(doc)},

{Py_tp_alloc, reinterpret_cast<void*>(&FutureAlloc)},

{Py_tp_dealloc, reinterpret_cast<void*>(&FutureDealloc)},

{Py_tp_traverse, reinterpret_cast<void*>(&FutureTraverse)},

{Py_tp_clear, reinterpret_cast<void*>(&FutureClear)},

{Py_tp_finalize, reinterpret_cast<void*>(&FutureFinalize)},

{Py_tp_methods,

const_cast<void*>(reinterpret_cast<const void*>(future_methods))},

{0, nullptr},

};

PyType_Spec spec = {};

spec.flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC;

spec.slots = slots;

auto cls = DefineHeapType<PythonFutureObject>(spec);

PythonFutureObject::python_type->tp_weaklistoffset =

offsetof(PythonFutureObject, weakrefs);

m.attr("Future") = cls;

return cls;

cls.def(

"__new__",

[](py::handle cls_unused, UntypedFutureLike python_future,

std::optional<AbstractEventLoopParameter> loop)

-> UntypedFutureWrapper {

if (!loop) loop.emplace().value = GetCurrentThreadAsyncioEventLoop();

if (py::object future =

TryConvertToFuture(python_future.value, loop->value)) {

return {future};

}

PythonObjectReferenceManager manager;

return {PythonFutureObject::Make(

Future<GilSafePythonValueOrExceptionWeakRef>(

GilSafePythonValueOrExceptionWeakRef{

PythonValueOrExceptionWeakRef(

manager, PythonValueOrException{

std::move(python_future.value)})}))};

},

R"(

)",

py::arg("future"), py::kw_only(), py::arg("loop") = std::nullopt);

cls.def("__await__", [](PythonFutureObject& self) -> AwaitResult<TypeVarT> {

return {self.GetAwaitable()};

});

cls.def(

"add_done_callback",

[](PythonFutureObject& self, Callable<void, Future<TypeVarT>> callback) {

self.AddDoneCallback(callback.value);

},

py::arg("callback"),

R"(

)");

cls.def(

"remove_done_callback",

[](PythonFutureObject& self, Callable<void, Future<TypeVarT>> callback) {

return self.RemoveDoneCallback(callback.value);

},

py::arg("callback"),

R"(

)");

cls.def(

"result",

[](PythonFutureObject& self, std::optional<double> timeout,

std::optional<double> deadline) -> TypeVarT {

return TypeVarT{self.GetResult(GetWaitDeadline(timeout, deadline))};

},

py::arg("timeout") = std::nullopt, py::arg("deadline") = std::nullopt,

R"(

)");

cls.def(

"exception",

[](PythonFutureObject& self, std::optional<double> timeout,

std::optional<double> deadline) -> py::object {