Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

gh-110693: Add a Timeout to _PyEval_AddPendingCall() #118329

Draft
wants to merge 3 commits into
base: main
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
4 changes: 3 additions & 1 deletion Include/internal/pycore_ceval.h
Expand Up @@ -51,13 +51,15 @@ extern void _PyEval_SignalReceived(void);
typedef int _Py_add_pending_call_result;
#define _Py_ADD_PENDING_SUCCESS 0
#define _Py_ADD_PENDING_FULL -1
#define _Py_ADD_PENDING_TIMED_OUT -2

// Export for '_testinternalcapi' shared extension
PyAPI_FUNC(_Py_add_pending_call_result) _PyEval_AddPendingCall(
PyInterpreterState *interp,
_Py_pending_call_func func,
void *arg,
int flags);
int flags,
PY_TIMEOUT_T timeout);

#ifdef HAVE_FORK
extern PyStatus _PyEval_ReInitThreads(PyThreadState *tstate);
Expand Down
12 changes: 7 additions & 5 deletions Modules/_testinternalcapi.c
Expand Up @@ -1090,13 +1090,16 @@ pending_threadfunc(PyObject *self, PyObject *args, PyObject *kwargs)
if (ensure_added) {
_Py_add_pending_call_result r;
do {
r = _PyEval_AddPendingCall(interp, &_pending_callback, callable, 0);
r = _PyEval_AddPendingCall(
interp, &_pending_callback, callable, 0, 0);
assert(r == _Py_ADD_PENDING_SUCCESS
|| r == _Py_ADD_PENDING_FULL);
} while (r == _Py_ADD_PENDING_FULL);
}
else {
if (_PyEval_AddPendingCall(interp, &_pending_callback, callable, 0) < 0) {
if (_PyEval_AddPendingCall(
interp, &_pending_callback, callable, 0, 0) < 0)
{
break;
}
}
Expand Down Expand Up @@ -1157,9 +1160,8 @@ pending_identify(PyObject *self, PyObject *args)
_Py_add_pending_call_result r;
do {
Py_BEGIN_ALLOW_THREADS
r = _PyEval_AddPendingCall(interp,
&_pending_identify_callback, (void *)mutex,
0);
r = _PyEval_AddPendingCall(
interp, &_pending_identify_callback, (void *)mutex, 0, 0);
Py_END_ALLOW_THREADS
assert(r == _Py_ADD_PENDING_SUCCESS
|| r == _Py_ADD_PENDING_FULL);
Expand Down
6 changes: 4 additions & 2 deletions Modules/signalmodule.c
Expand Up @@ -316,7 +316,8 @@ trip_signal(int sig_num)
_PyEval_AddPendingCall(interp,
report_wakeup_send_error,
(void *)(intptr_t) last_error,
_Py_PENDING_MAINTHREADONLY);
_Py_PENDING_MAINTHREADONLY,
0 /* timeout */);
}
}
}
Expand All @@ -336,7 +337,8 @@ trip_signal(int sig_num)
_PyEval_AddPendingCall(interp,
report_wakeup_write_error,
(void *)(intptr_t)errno,
_Py_PENDING_MAINTHREADONLY);
_Py_PENDING_MAINTHREADONLY,
0 /* timeout */);
}
}
}
Expand Down
52 changes: 44 additions & 8 deletions Python/ceval_gil.c
Expand Up @@ -7,6 +7,7 @@
#include "pycore_pylifecycle.h" // _PyErr_Print()
#include "pycore_pymem.h" // _PyMem_IsPtrFreed()
#include "pycore_pystats.h" // _Py_PrintSpecializationStats()
#include "pycore_time.h" // _PyDeadline_Init()

/*
Notes about the implementation:
Expand Down Expand Up @@ -739,7 +740,8 @@ _pop_pending_call(struct _pending_calls *pending,

_Py_add_pending_call_result
_PyEval_AddPendingCall(PyInterpreterState *interp,
_Py_pending_call_func func, void *arg, int flags)
_Py_pending_call_func func, void *arg, int flags,
PY_TIMEOUT_T timeout)
{
struct _pending_calls *pending = &interp->ceval.pending;
int main_only = (flags & _Py_PENDING_MAINTHREADONLY) != 0;
Expand All @@ -749,11 +751,41 @@ _PyEval_AddPendingCall(PyInterpreterState *interp,
pending = &_PyRuntime.ceval.pending_mainthread;
}

PyTime_t endtime = 0;
if (timeout > 0) {
endtime = _PyDeadline_Init(timeout);
}

PyMutex_Lock(&pending->mutex);
_Py_add_pending_call_result result =
_push_pending_call(pending, func, arg, flags);
_push_pending_call(pending, func, arg, flags);
PyMutex_Unlock(&pending->mutex);

if (timeout > 0) {
/* We use this lock only to sleep. */
// XXX Use time.sleep()? Don't sleep at all?
PyMutex timeout_sleeper = (PyMutex){0};
PyMutex_Lock(&timeout_sleeper);
while (result == _Py_ADD_PENDING_FULL) {
if (timeout <= 0) {
result = _Py_ADD_PENDING_TIMED_OUT;
break;
}
// XXX Use a smaller sleep interval?
#define SLEEP_NS 1000 /* 1 millisecond */
(void)_PyMutex_LockTimed(
&timeout_sleeper, SLEEP_NS, _Py_LOCK_DONT_DETACH);

PyMutex_Lock(&pending->mutex);
result = _push_pending_call(pending, func, arg, flags);
PyMutex_Unlock(&pending->mutex);

timeout = _PyDeadline_Get(endtime);
}
PyMutex_Unlock(&timeout_sleeper);
}

// XXX Do not update the eval breaker if not _Py_ADD_PENDING_SUCCESS.
if (main_only) {
_Py_set_eval_breaker_bit(_PyRuntime.main_tstate, _PY_CALLS_TO_DO_BIT);
}
Expand All @@ -774,14 +806,18 @@ Py_AddPendingCall(_Py_pending_call_func func, void *arg)
/* Legacy users of this API will continue to target the main thread
(of the main interpreter). */
PyInterpreterState *interp = _PyInterpreterState_Main();
_Py_add_pending_call_result r =
_PyEval_AddPendingCall(interp, func, arg, _Py_PENDING_MAINTHREADONLY);
if (r == _Py_ADD_PENDING_FULL) {
_Py_add_pending_call_result r = _PyEval_AddPendingCall(
interp, func, arg, _Py_PENDING_MAINTHREADONLY, 0);
switch (r) {
case _Py_ADD_PENDING_TIMED_OUT: // fall through
case _Py_ADD_PENDING_FULL:
return -1;
}
else {
assert(r == _Py_ADD_PENDING_SUCCESS);
case _Py_ADD_PENDING_SUCCESS:
return 0;
default:
// We added a new result kind but forgot to handle it here.
assert(0);
return -1;
}
}

Expand Down
8 changes: 6 additions & 2 deletions Python/crossinterp.c
Expand Up @@ -34,7 +34,9 @@ _Py_CallInInterpreter(PyInterpreterState *interp,
return func(arg);
}
// XXX Emit a warning if this fails?
_PyEval_AddPendingCall(interp, (_Py_pending_call_func)func, arg, 0);
// XXX Block until added?
_PyEval_AddPendingCall(
interp, (_Py_pending_call_func)func, arg, 0, 0 /* timeout */);
return 0;
}

Expand All @@ -48,7 +50,9 @@ _Py_CallInInterpreterAndRawFree(PyInterpreterState *interp,
return res;
}
// XXX Emit a warning if this fails?
_PyEval_AddPendingCall(interp, func, arg, _Py_PENDING_RAWFREE);
// XXX Block until added?
_PyEval_AddPendingCall(
interp, func, arg, _Py_PENDING_RAWFREE, 0 /* timeout */);
return 0;
}

Expand Down