Mirrors
/
cpython
mirror of https://github.com/python/cpython
3
0
Fork 0
Code Releases Activity
Browse Source

bpo-32348: Optimize asyncio.Future schedule/add/remove callback. (#4907)

pull/4914/head
Yury Selivanov 9 years ago
committed by GitHub
parent
4c72bc4a38
commit
1b7c11ff0e
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 499 additions and 102 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 159
      Lib/test/test_asyncio/test_futures.py
  2. 43
      Lib/test/test_asyncio/test_tasks.py
  3. 2
      Misc/NEWS.d/next/Library/2017-12-16-18-50-57.bpo-32348.5j__he.rst
  4. 397
      Modules/_asynciomodule.c

159
Lib/test/test_asyncio/test_futures.py
View File

@ -145,37 +145,60 @@ class BaseFutureTests:
self.assertRaises(TypeError, self._new_future, 42)
def test_uninitialized(self):
# Test that C Future doesn't crash when Future.__init__()
# call was skipped.
fut = self.cls.__new__(self.cls, loop=self.loop)
self.assertRaises(asyncio.InvalidStateError, fut.result)
fut = self.cls.__new__(self.cls, loop=self.loop)
self.assertRaises(asyncio.InvalidStateError, fut.exception)
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut.set_result(None)
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut.set_exception(Exception)
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut.cancel()
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut.add_done_callback(lambda f: None)
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut.remove_done_callback(lambda f: None)
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut._schedule_callbacks()
fut = self.cls.__new__(self.cls, loop=self.loop)
try:
repr(fut)
except AttributeError:
except (RuntimeError, AttributeError):
pass
fut = self.cls.__new__(self.cls, loop=self.loop)
try:
fut.__await__()
except RuntimeError:
pass
fut = self.cls.__new__(self.cls, loop=self.loop)
try:
iter(fut)
except RuntimeError:
pass
fut = self.cls.__new__(self.cls, loop=self.loop)
fut.cancelled()
fut.done()
iter(fut)
self.assertFalse(fut.cancelled())
self.assertFalse(fut.done())
def test_cancel(self):
f = self._new_future(loop=self.loop)
@ -246,30 +269,32 @@ class BaseFutureTests:
self.loop.set_debug(True)
f_pending_debug = self._new_future(loop=self.loop)
frame = f_pending_debug._source_traceback[-1]
self.assertEqual(repr(f_pending_debug),
'<Future pending created at %s:%s>'
% (frame[0], frame[1]))
self.assertEqual(
repr(f_pending_debug),
f'<{self.cls.__name__} pending created at {frame[0]}:{frame[1]}>')
f_pending_debug.cancel()
self.loop.set_debug(False)
f_pending = self._new_future(loop=self.loop)
self.assertEqual(repr(f_pending), '<Future pending>')
self.assertEqual(repr(f_pending), f'<{self.cls.__name__} pending>')
f_pending.cancel()
f_cancelled = self._new_future(loop=self.loop)
f_cancelled.cancel()
self.assertEqual(repr(f_cancelled), '<Future cancelled>')
self.assertEqual(repr(f_cancelled), f'<{self.cls.__name__} cancelled>')
f_result = self._new_future(loop=self.loop)
f_result.set_result(4)
self.assertEqual(repr(f_result), '<Future finished result=4>')
self.assertEqual(
repr(f_result), f'<{self.cls.__name__} finished result=4>')
self.assertEqual(f_result.result(), 4)
exc = RuntimeError()
f_exception = self._new_future(loop=self.loop)
f_exception.set_exception(exc)
self.assertEqual(repr(f_exception),
'<Future finished exception=RuntimeError()>')
self.assertEqual(
repr(f_exception),
f'<{self.cls.__name__} finished exception=RuntimeError()>')
self.assertIs(f_exception.exception(), exc)
def func_repr(func):
@ -280,11 +305,12 @@ class BaseFutureTests:
f_one_callbacks = self._new_future(loop=self.loop)
f_one_callbacks.add_done_callback(_fakefunc)
fake_repr = func_repr(_fakefunc)
self.assertRegex(repr(f_one_callbacks),
r'<Future pending cb=\[%s\]>' % fake_repr)
self.assertRegex(
repr(f_one_callbacks),
r'<' + self.cls.__name__ + r' pending cb=\[%s\]>' % fake_repr)
f_one_callbacks.cancel()
self.assertEqual(repr(f_one_callbacks),
'<Future cancelled>')
f'<{self.cls.__name__} cancelled>')
f_two_callbacks = self._new_future(loop=self.loop)
f_two_callbacks.add_done_callback(first_cb)
@ -292,7 +318,7 @@ class BaseFutureTests:
first_repr = func_repr(first_cb)
last_repr = func_repr(last_cb)
self.assertRegex(repr(f_two_callbacks),
r'<Future pending cb=\[%s, %s\]>'
r'<' + self.cls.__name__ + r' pending cb=\[%s, %s\]>'
% (first_repr, last_repr))
f_many_callbacks = self._new_future(loop=self.loop)
@ -301,11 +327,12 @@ class BaseFutureTests:
f_many_callbacks.add_done_callback(_fakefunc)
f_many_callbacks.add_done_callback(last_cb)
cb_regex = r'%s, <8 more>, %s' % (first_repr, last_repr)
self.assertRegex(repr(f_many_callbacks),
r'<Future pending cb=\[%s\]>' % cb_regex)
self.assertRegex(
repr(f_many_callbacks),
r'<' + self.cls.__name__ + r' pending cb=\[%s\]>' % cb_regex)
f_many_callbacks.cancel()
self.assertEqual(repr(f_many_callbacks),
'<Future cancelled>')
f'<{self.cls.__name__} cancelled>')
def test_copy_state(self):
from asyncio.futures import _copy_future_state
@ -475,7 +502,7 @@ class BaseFutureTests:
support.gc_collect()
if sys.version_info >= (3, 4):
regex = r'^Future exception was never retrieved\n'
regex = f'^{self.cls.__name__} exception was never retrieved\n'
exc_info = (type(exc), exc, exc.__traceback__)
m_log.error.assert_called_once_with(mock.ANY, exc_info=exc_info)
else:
@ -531,7 +558,16 @@ class BaseFutureTests:
@unittest.skipUnless(hasattr(futures, '_CFuture'),
'requires the C _asyncio module')
class CFutureTests(BaseFutureTests, test_utils.TestCase):
cls = getattr(futures, '_CFuture')
cls = futures._CFuture
@unittest.skipUnless(hasattr(futures, '_CFuture'),
'requires the C _asyncio module')
class CSubFutureTests(BaseFutureTests, test_utils.TestCase):
class CSubFuture(futures._CFuture):
pass
cls = CSubFuture
class PyFutureTests(BaseFutureTests, test_utils.TestCase):
@ -556,6 +592,76 @@ class BaseFutureDoneCallbackTests():
def _new_future(self):
raise NotImplementedError
def test_callbacks_remove_first_callback(self):
bag = []
f = self._new_future()
cb1 = self._make_callback(bag, 42)
cb2 = self._make_callback(bag, 17)
cb3 = self._make_callback(bag, 100)
f.add_done_callback(cb1)
f.add_done_callback(cb2)
f.add_done_callback(cb3)
f.remove_done_callback(cb1)
f.remove_done_callback(cb1)
self.assertEqual(bag, [])
f.set_result('foo')
self.run_briefly()
self.assertEqual(bag, [17, 100])
self.assertEqual(f.result(), 'foo')
def test_callbacks_remove_first_and_second_callback(self):
bag = []
f = self._new_future()
cb1 = self._make_callback(bag, 42)
cb2 = self._make_callback(bag, 17)
cb3 = self._make_callback(bag, 100)
f.add_done_callback(cb1)
f.add_done_callback(cb2)
f.add_done_callback(cb3)
f.remove_done_callback(cb1)
f.remove_done_callback(cb2)
f.remove_done_callback(cb1)
self.assertEqual(bag, [])
f.set_result('foo')
self.run_briefly()
self.assertEqual(bag, [100])
self.assertEqual(f.result(), 'foo')
def test_callbacks_remove_third_callback(self):
bag = []
f = self._new_future()
cb1 = self._make_callback(bag, 42)
cb2 = self._make_callback(bag, 17)
cb3 = self._make_callback(bag, 100)
f.add_done_callback(cb1)
f.add_done_callback(cb2)
f.add_done_callback(cb3)
f.remove_done_callback(cb3)
f.remove_done_callback(cb3)
self.assertEqual(bag, [])
f.set_result('foo')
self.run_briefly()
self.assertEqual(bag, [42, 17])
self.assertEqual(f.result(), 'foo')
def test_callbacks_invoked_on_set_result(self):
bag = []
f = self._new_future()
@ -678,6 +784,17 @@ class CFutureDoneCallbackTests(BaseFutureDoneCallbackTests,
return futures._CFuture(loop=self.loop)
@unittest.skipUnless(hasattr(futures, '_CFuture'),
'requires the C _asyncio module')
class CSubFutureDoneCallbackTests(BaseFutureDoneCallbackTests,
test_utils.TestCase):
def _new_future(self):
class CSubFuture(futures._CFuture):
pass
return CSubFuture(loop=self.loop)
class PyFutureDoneCallbackTests(BaseFutureDoneCallbackTests,
test_utils.TestCase):

43
Lib/test/test_asyncio/test_tasks.py
View File

@ -2187,23 +2187,51 @@ def add_subclass_tests(cls):
return cls
@unittest.skipUnless(hasattr(futures, '_CFuture'),
@unittest.skipUnless(hasattr(futures, '_CFuture') and
hasattr(tasks, '_CTask'),
'requires the C _asyncio module')
class CTask_CFuture_Tests(BaseTaskTests, test_utils.TestCase):
Task = getattr(tasks, '_CTask', None)
Future = getattr(futures, '_CFuture', None)
@unittest.skipUnless(hasattr(futures, '_CFuture'),
@unittest.skipUnless(hasattr(futures, '_CFuture') and
hasattr(tasks, '_CTask'),
'requires the C _asyncio module')
@add_subclass_tests
class CTask_CFuture_SubclassTests(BaseTaskTests, test_utils.TestCase):
Task = getattr(tasks, '_CTask', None)
Future = getattr(futures, '_CFuture', None)
class Task(tasks._CTask):
pass
class Future(futures._CFuture):
pass
@unittest.skipUnless(hasattr(tasks, '_CTask'),
'requires the C _asyncio module')
@add_subclass_tests
class CTaskSubclass_PyFuture_Tests(BaseTaskTests, test_utils.TestCase):
class Task(tasks._CTask):
pass
Future = futures._PyFuture
@unittest.skipUnless(hasattr(futures, '_CFuture'),
'requires the C _asyncio module')
@add_subclass_tests
class PyTask_CFutureSubclass_Tests(BaseTaskTests, test_utils.TestCase):
class Future(futures._CFuture):
pass
Task = tasks._PyTask
@unittest.skipUnless(hasattr(tasks, '_CTask'),
'requires the C _asyncio module')
class CTask_PyFuture_Tests(BaseTaskTests, test_utils.TestCase):
Task = getattr(tasks, '_CTask', None)
Future = futures._PyFuture
@ -2223,8 +2251,11 @@ class PyTask_PyFuture_Tests(BaseTaskTests, test_utils.TestCase):
@add_subclass_tests
class PyTask_PyFuture_SubclassTests(BaseTaskTests, test_utils.TestCase):
Task = tasks._PyTask
Future = futures._PyFuture
class Task(tasks._PyTask):
pass
class Future(futures._PyFuture):
pass
class BaseTaskIntrospectionTests:

2
Misc/NEWS.d/next/Library/2017-12-16-18-50-57.bpo-32348.5j__he.rst
View File

@ -0,0 +1,2 @@
Optimize asyncio.Future schedule/add/remove callback. The optimization
shows 3-6% performance improvements of async/await code.

397
Modules/_asynciomodule.c
View File

@ -59,6 +59,7 @@ typedef enum {
#define FutureObj_HEAD(prefix) \
PyObject_HEAD \
PyObject *prefix##_loop; \
PyObject *prefix##_callback0; \
PyObject *prefix##_callbacks; \
PyObject *prefix##_exception; \
PyObject *prefix##_result; \
@ -93,6 +94,16 @@ typedef struct {
} TaskWakeupMethWrapper;
static PyTypeObject FutureType;
static PyTypeObject TaskType;
#define Future_CheckExact(obj) (Py_TYPE(obj) == &FutureType)
#define Task_CheckExact(obj) (Py_TYPE(obj) == &TaskType)
#define Future_Check(obj) PyObject_TypeCheck(obj, &FutureType)
#define Task_Check(obj) PyObject_TypeCheck(obj, &TaskType)
#include "clinic/_asynciomodule.c.h"
@ -101,6 +112,7 @@ class _asyncio.Future "FutureObj *" "&Future_Type"
[clinic start generated code]*/
/*[clinic end generated code: output=da39a3ee5e6b4b0d input=00d3e4abca711e0f]*/
/* Get FutureIter from Future */
static PyObject* future_new_iter(PyObject *);
static inline int future_call_schedule_callbacks(FutureObj *);
@ -233,47 +245,95 @@ get_event_loop(void)
}
static int
call_soon(PyObject *loop, PyObject *func, PyObject *arg)
{
PyObject *handle;
handle = _PyObject_CallMethodIdObjArgs(
loop, &PyId_call_soon, func, arg, NULL);
if (handle == NULL) {
return -1;
}
Py_DECREF(handle);
return 0;
}
static inline int
future_is_alive(FutureObj *fut)
{
return fut->fut_loop != NULL;
}
static inline int
future_ensure_alive(FutureObj *fut)
{
if (!future_is_alive(fut)) {
PyErr_SetString(PyExc_RuntimeError,
"Future object is not initialized.");
return -1;
}
return 0;
}
#define ENSURE_FUTURE_ALIVE(fut) \
do { \
assert(Future_Check(fut) || Task_Check(fut)); \
if (future_ensure_alive((FutureObj*)fut)) { \
return NULL; \
} \
} while(0);
static int
future_schedule_callbacks(FutureObj *fut)
{
Py_ssize_t len;
PyObject *callbacks;
int i;
Py_ssize_t i;
if (fut->fut_callback0 != NULL) {
/* There's a 1st callback */
int ret = call_soon(
fut->fut_loop, fut->fut_callback0, (PyObject *)fut);
Py_CLEAR(fut->fut_callback0);
if (ret) {
/* If an error occurs in pure-Python implementation,
all callbacks are cleared. */
Py_CLEAR(fut->fut_callbacks);
return ret;
}
/* we called the first callback, now try calling
callbacks from the 'fut_callbacks' list. */
}
if (fut->fut_callbacks == NULL) {
PyErr_SetString(PyExc_RuntimeError, "uninitialized Future object");
return -1;
/* No more callbacks, return. */
return 0;
}
len = PyList_GET_SIZE(fut->fut_callbacks);
if (len == 0) {
/* The list of callbacks was empty; clear it and return. */
Py_CLEAR(fut->fut_callbacks);
return 0;
}
callbacks = PyList_GetSlice(fut->fut_callbacks, 0, len);
if (callbacks == NULL) {
return -1;
}
if (PyList_SetSlice(fut->fut_callbacks, 0, len, NULL) < 0) {
Py_DECREF(callbacks);
return -1;
}
for (i = 0; i < len; i++) {
PyObject *handle;
PyObject *cb = PyList_GET_ITEM(callbacks, i);
PyObject *cb = PyList_GET_ITEM(fut->fut_callbacks, i);
handle = _PyObject_CallMethodIdObjArgs(fut->fut_loop, &PyId_call_soon,
cb, fut, NULL);
if (handle == NULL) {
Py_DECREF(callbacks);
if (call_soon(fut->fut_loop, cb, (PyObject *)fut)) {
/* If an error occurs in pure-Python implementation,
all callbacks are cleared. */
Py_CLEAR(fut->fut_callbacks);
return -1;
}
Py_DECREF(handle);
}
Py_DECREF(callbacks);
Py_CLEAR(fut->fut_callbacks);
return 0;
}
@ -311,10 +371,8 @@ future_init(FutureObj *fut, PyObject *loop)
}
}
Py_XSETREF(fut->fut_callbacks, PyList_New(0));
if (fut->fut_callbacks == NULL) {
return -1;
}
fut->fut_callback0 = NULL;
fut->fut_callbacks = NULL;
return 0;
}
@ -322,6 +380,10 @@ future_init(FutureObj *fut, PyObject *loop)
static PyObject *
future_set_result(FutureObj *fut, PyObject *res)
{
if (future_ensure_alive(fut)) {
return NULL;
}
if (fut->fut_state != STATE_PENDING) {
PyErr_SetString(asyncio_InvalidStateError, "invalid state");
return NULL;
@ -416,25 +478,61 @@ future_get_result(FutureObj *fut, PyObject **result)
static PyObject *
future_add_done_callback(FutureObj *fut, PyObject *arg)
{
if (!future_is_alive(fut)) {
PyErr_SetString(PyExc_RuntimeError, "uninitialized Future object");
return NULL;
}
if (fut->fut_state != STATE_PENDING) {
PyObject *handle = _PyObject_CallMethodIdObjArgs(fut->fut_loop,
&PyId_call_soon,
arg, fut, NULL);
if (handle == NULL) {
/* The future is done/cancelled, so schedule the callback
right away. */
if (call_soon(fut->fut_loop, arg, (PyObject*) fut)) {
return NULL;
}
Py_DECREF(handle);
}
else {
if (fut->fut_callbacks == NULL) {
PyErr_SetString(PyExc_RuntimeError, "uninitialized Future object");
return NULL;
/* The future is pending, add a callback.
Callbacks in the future object are stored as follows:
callback0 -- a pointer to the first callback
callbacks -- a list of 2nd, 3rd, ... callbacks
Invariants:
* callbacks != NULL:
There are some callbacks in in the list. Just
add the new callback to it.
* callbacks == NULL and callback0 == NULL:
This is the first callback. Set it to callback0.
* callbacks == NULL and callback0 != NULL:
This is a second callback. Initialize callbacks
with a new list and add the new callback to it.
*/
if (fut->fut_callbacks != NULL) {
int err = PyList_Append(fut->fut_callbacks, arg);
if (err != 0) {
return NULL;
}
}
int err = PyList_Append(fut->fut_callbacks, arg);
if (err != 0) {
return NULL;
else if (fut->fut_callback0 == NULL) {
Py_INCREF(arg);
fut->fut_callback0 = arg;
}
else {
fut->fut_callbacks = PyList_New(1);
if (fut->fut_callbacks == NULL) {
return NULL;
}
Py_INCREF(arg);
PyList_SET_ITEM(fut->fut_callbacks, 0, arg);
}
}
Py_RETURN_NONE;
}
@ -487,6 +585,7 @@ static int
FutureObj_clear(FutureObj *fut)
{
Py_CLEAR(fut->fut_loop);
Py_CLEAR(fut->fut_callback0);
Py_CLEAR(fut->fut_callbacks);
Py_CLEAR(fut->fut_result);
Py_CLEAR(fut->fut_exception);
@ -499,6 +598,7 @@ static int
FutureObj_traverse(FutureObj *fut, visitproc visit, void *arg)
{
Py_VISIT(fut->fut_loop);
Py_VISIT(fut->fut_callback0);
Py_VISIT(fut->fut_callbacks);
Py_VISIT(fut->fut_result);
Py_VISIT(fut->fut_exception);
@ -522,6 +622,13 @@ _asyncio_Future_result_impl(FutureObj *self)
/*[clinic end generated code: output=f35f940936a4b1e5 input=49ecf9cf5ec50dc5]*/
{
PyObject *result;
if (!future_is_alive(self)) {
PyErr_SetString(asyncio_InvalidStateError,
"Future object is not initialized.");
return NULL;
}
int res = future_get_result(self, &result);
if (res == -1) {
@ -554,6 +661,12 @@ static PyObject *
_asyncio_Future_exception_impl(FutureObj *self)
/*[clinic end generated code: output=88b20d4f855e0710 input=733547a70c841c68]*/
{
if (!future_is_alive(self)) {
PyErr_SetString(asyncio_InvalidStateError,
"Future object is not initialized.");
return NULL;
}
if (self->fut_state == STATE_CANCELLED) {
PyErr_SetNone(asyncio_CancelledError);
return NULL;
@ -589,6 +702,7 @@ static PyObject *
_asyncio_Future_set_result(FutureObj *self, PyObject *res)
/*[clinic end generated code: output=a620abfc2796bfb6 input=5b9dc180f1baa56d]*/
{
ENSURE_FUTURE_ALIVE(self)
return future_set_result(self, res);
}
@ -608,6 +722,7 @@ static PyObject *
_asyncio_Future_set_exception(FutureObj *self, PyObject *exception)
/*[clinic end generated code: output=f1c1b0cd321be360 input=e45b7d7aa71cc66d]*/
{
ENSURE_FUTURE_ALIVE(self)
return future_set_exception(self, exception);
}
@ -648,15 +763,45 @@ _asyncio_Future_remove_done_callback(FutureObj *self, PyObject *fn)
{
PyObject *newlist;
Py_ssize_t len, i, j=0;
Py_ssize_t cleared_callback0 = 0;
ENSURE_FUTURE_ALIVE(self)
if (self->fut_callback0 != NULL) {
int cmp = PyObject_RichCompareBool(fn, self->fut_callback0, Py_EQ);
if (cmp == -1) {
return NULL;
}
if (cmp == 1) {
/* callback0 == fn */
Py_CLEAR(self->fut_callback0);
cleared_callback0 = 1;
}
}
if (self->fut_callbacks == NULL) {
PyErr_SetString(PyExc_RuntimeError, "uninitialized Future object");
return NULL;
return PyLong_FromSsize_t(cleared_callback0);
}
len = PyList_GET_SIZE(self->fut_callbacks);
if (len == 0) {
return PyLong_FromSsize_t(0);
Py_CLEAR(self->fut_callbacks);
return PyLong_FromSsize_t(cleared_callback0);
}
if (len == 1) {
int cmp = PyObject_RichCompareBool(
fn, PyList_GET_ITEM(self->fut_callbacks, 0), Py_EQ);
if (cmp == -1) {
return NULL;
}
if (cmp == 1) {
/* callbacks[0] == fn */
Py_CLEAR(self->fut_callbacks);
return PyLong_FromSsize_t(1 + cleared_callback0);
}
/* callbacks[0] != fn and len(callbacks) == 1 */
return PyLong_FromSsize_t(cleared_callback0);
}
newlist = PyList_New(len);
@ -683,6 +828,12 @@ _asyncio_Future_remove_done_callback(FutureObj *self, PyObject *fn)
}
}
if (j == 0) {
Py_CLEAR(self->fut_callbacks);
Py_DECREF(newlist);
return PyLong_FromSsize_t(len + cleared_callback0);
}
if (j < len) {
Py_SIZE(newlist) = j;
}
@ -694,7 +845,7 @@ _asyncio_Future_remove_done_callback(FutureObj *self, PyObject *fn)
}
}
Py_DECREF(newlist);
return PyLong_FromSsize_t(len - j);
return PyLong_FromSsize_t(len - j + cleared_callback0);
fail:
Py_DECREF(newlist);
@ -715,6 +866,7 @@ static PyObject *
_asyncio_Future_cancel_impl(FutureObj *self)
/*[clinic end generated code: output=e45b932ba8bd68a1 input=515709a127995109]*/
{
ENSURE_FUTURE_ALIVE(self)
return future_cancel(self);
}
@ -728,7 +880,7 @@ static PyObject *
_asyncio_Future_cancelled_impl(FutureObj *self)
/*[clinic end generated code: output=145197ced586357d input=943ab8b7b7b17e45]*/
{
if (self->fut_state == STATE_CANCELLED) {
if (future_is_alive(self) && self->fut_state == STATE_CANCELLED) {
Py_RETURN_TRUE;
}
else {
@ -749,7 +901,7 @@ static PyObject *
_asyncio_Future_done_impl(FutureObj *self)
/*[clinic end generated code: output=244c5ac351145096 input=28d7b23fdb65d2ac]*/
{
if (self->fut_state == STATE_PENDING) {
if (!future_is_alive(self) || self->fut_state == STATE_PENDING) {
Py_RETURN_FALSE;
}
else {
@ -760,7 +912,7 @@ _asyncio_Future_done_impl(FutureObj *self)
static PyObject *
FutureObj_get_blocking(FutureObj *fut)
{
if (fut->fut_blocking) {
if (future_is_alive(fut) && fut->fut_blocking) {
Py_RETURN_TRUE;
}
else {
@ -771,6 +923,10 @@ FutureObj_get_blocking(FutureObj *fut)
static int
FutureObj_set_blocking(FutureObj *fut, PyObject *val)
{
if (future_ensure_alive(fut)) {
return -1;
}
int is_true = PyObject_IsTrue(val);
if (is_true < 0) {
return -1;
@ -782,6 +938,7 @@ FutureObj_set_blocking(FutureObj *fut, PyObject *val)
static PyObject *
FutureObj_get_log_traceback(FutureObj *fut)
{
ENSURE_FUTURE_ALIVE(fut)
if (fut->fut_log_tb) {
Py_RETURN_TRUE;
}
@ -804,7 +961,7 @@ FutureObj_set_log_traceback(FutureObj *fut, PyObject *val)
static PyObject *
FutureObj_get_loop(FutureObj *fut)
{
if (fut->fut_loop == NULL) {
if (!future_is_alive(fut)) {
Py_RETURN_NONE;
}
Py_INCREF(fut->fut_loop);
@ -814,16 +971,57 @@ FutureObj_get_loop(FutureObj *fut)
static PyObject *
FutureObj_get_callbacks(FutureObj *fut)
{
Py_ssize_t i;
Py_ssize_t len;
PyObject *new_list;
ENSURE_FUTURE_ALIVE(fut)
if (fut->fut_callbacks == NULL) {
Py_RETURN_NONE;
if (fut->fut_callback0 == NULL) {
Py_RETURN_NONE;
}
else {
new_list = PyList_New(1);
if (new_list == NULL) {
return NULL;
}
Py_INCREF(fut->fut_callback0);
PyList_SET_ITEM(new_list, 0, fut->fut_callback0);
return new_list;
}
}
assert(fut->fut_callbacks != NULL);
if (fut->fut_callback0 == NULL) {
Py_INCREF(fut->fut_callbacks);
return fut->fut_callbacks;
}
assert(fut->fut_callback0 != NULL);
len = PyList_GET_SIZE(fut->fut_callbacks);
new_list = PyList_New(len + 1);
if (new_list == NULL) {
return NULL;
}
Py_INCREF(fut->fut_callback0);
PyList_SET_ITEM(new_list, 0, fut->fut_callback0);
for (i = 0; i < len; i++) {
PyObject *cb = PyList_GET_ITEM(fut->fut_callbacks, i);
Py_INCREF(cb);
PyList_SET_ITEM(new_list, i + 1, cb);
}
Py_INCREF(fut->fut_callbacks);
return fut->fut_callbacks;
return new_list;
}
static PyObject *
FutureObj_get_result(FutureObj *fut)
{
ENSURE_FUTURE_ALIVE(fut)
if (fut->fut_result == NULL) {
Py_RETURN_NONE;
}
@ -834,6 +1032,7 @@ FutureObj_get_result(FutureObj *fut)
static PyObject *
FutureObj_get_exception(FutureObj *fut)
{
ENSURE_FUTURE_ALIVE(fut)
if (fut->fut_exception == NULL) {
Py_RETURN_NONE;
}
@ -844,7 +1043,7 @@ FutureObj_get_exception(FutureObj *fut)
static PyObject *
FutureObj_get_source_traceback(FutureObj *fut)
{
if (fut->fut_source_tb == NULL) {
if (!future_is_alive(fut) || fut->fut_source_tb == NULL) {
Py_RETURN_NONE;
}
Py_INCREF(fut->fut_source_tb);
@ -859,6 +1058,8 @@ FutureObj_get_state(FutureObj *fut)
_Py_IDENTIFIER(FINISHED);
PyObject *ret = NULL;
ENSURE_FUTURE_ALIVE(fut)
switch (fut->fut_state) {
case STATE_PENDING:
ret = _PyUnicode_FromId(&PyId_PENDING);
@ -896,6 +1097,8 @@ static PyObject *
_asyncio_Future__schedule_callbacks_impl(FutureObj *self)
/*[clinic end generated code: output=5e8958d89ea1c5dc input=4f5f295f263f4a88]*/
{
ENSURE_FUTURE_ALIVE(self)
int ret = future_schedule_callbacks(self);
if (ret == -1) {
return NULL;
@ -908,6 +1111,8 @@ FutureObj_repr(FutureObj *fut)
{
_Py_IDENTIFIER(_repr_info);
ENSURE_FUTURE_ALIVE(fut)
PyObject *rinfo = _PyObject_CallMethodIdObjArgs((PyObject*)fut,
&PyId__repr_info,
NULL);
@ -1068,12 +1273,10 @@ static PyTypeObject FutureType = {
.tp_finalize = (destructor)FutureObj_finalize,
};
#define Future_CheckExact(obj) (Py_TYPE(obj) == &FutureType)
static inline int
future_call_schedule_callbacks(FutureObj *fut)
{
if (Future_CheckExact(fut)) {
if (Future_CheckExact(fut) || Task_CheckExact(fut)) {
return future_schedule_callbacks(fut);
}
else {
@ -1122,12 +1325,26 @@ typedef struct {
FutureObj *future;
} futureiterobject;
#define FI_FREELIST_MAXLEN 255
static futureiterobject *fi_freelist = NULL;
static Py_ssize_t fi_freelist_len = 0;
static void
FutureIter_dealloc(futureiterobject *it)
{
PyObject_GC_UnTrack(it);
Py_XDECREF(it->future);
PyObject_GC_Del(it);
Py_CLEAR(it->future);
if (fi_freelist_len < FI_FREELIST_MAXLEN) {
fi_freelist_len++;
it->future = (FutureObj*) fi_freelist;
fi_freelist = it;
}
else {
PyObject_GC_Del(it);
}
}
static PyObject *
@ -1272,10 +1489,23 @@ future_new_iter(PyObject *fut)
PyErr_BadInternalCall();
return NULL;
}
it = PyObject_GC_New(futureiterobject, &FutureIterType);
if (it == NULL) {
return NULL;
ENSURE_FUTURE_ALIVE(fut)
if (fi_freelist_len) {
fi_freelist_len--;
it = fi_freelist;
fi_freelist = (futureiterobject*) it->future;
it->future = NULL;
_Py_NewReference((PyObject*) it);
}
else {
it = PyObject_GC_New(futureiterobject, &FutureIterType);
if (it == NULL) {
return NULL;
}
}
Py_INCREF(fut);
it->future = (FutureObj*)fut;
PyObject_GC_Track(it);
@ -1549,20 +1779,25 @@ _asyncio_Task___init___impl(TaskObj *self, PyObject *coro, PyObject *loop)
/*[clinic end generated code: output=9f24774c2287fc2f input=8d132974b049593e]*/
{
PyObject *res;
int tmp;
if (future_init((FutureObj*)self, loop)) {
return -1;
}
if (!PyCoro_CheckExact(coro)) {
// fastpath failed, perfom slow check
// raise after Future.__init__(), attrs are required for __del__
res = PyObject_CallFunctionObjArgs(asyncio_iscoroutine_func,
coro, NULL);
/* 'coro' is not a native coroutine, call asyncio.iscoroutine()
to check if it's another coroutine flavour.
Do this check after 'future_init()'; in case we need to raise
an error, __del__ needs a properly initialized object.
*/
res = PyObject_CallFunctionObjArgs(
asyncio_iscoroutine_func, coro, NULL);
if (res == NULL) {
return -1;
}
tmp = PyObject_Not(res);
int tmp = PyObject_Not(res);
Py_DECREF(res);
if (tmp < 0) {
return -1;
@ -2023,8 +2258,6 @@ static PyTypeObject TaskType = {
.tp_finalize = (destructor)TaskObj_finalize,
};
#define Task_CheckExact(obj) (Py_TYPE(obj) == &TaskType)
static void
TaskObj_dealloc(PyObject *self)
{
@ -2079,22 +2312,14 @@ task_call_step(TaskObj *task, PyObject *arg)
static int
task_call_step_soon(TaskObj *task, PyObject *arg)
{
PyObject *handle;
PyObject *cb = TaskStepMethWrapper_new(task, arg);
if (cb == NULL) {
return -1;
}
handle = _PyObject_CallMethodIdObjArgs(task->task_loop, &PyId_call_soon,
cb, NULL);
int ret = call_soon(task->task_loop, cb, NULL);
Py_DECREF(cb);
if (handle == NULL) {
return -1;
}
Py_DECREF(handle);
return 0;
return ret;
}
static PyObject *
@ -2746,6 +2971,26 @@ _asyncio__leave_task_impl(PyObject *module, PyObject *loop, PyObject *task)
/*********************** Module **************************/
static void
module_free_freelists()
{
PyObject *next;
PyObject *current;
next = (PyObject*) fi_freelist;
while (next != NULL) {
assert(fi_freelist_len > 0);
fi_freelist_len--;
current = next;
next = (PyObject*) ((futureiterobject*) current)->future;
PyObject_GC_Del(current);
}
assert(fi_freelist_len == 0);
fi_freelist = NULL;
}
static void
module_free(void *m)
{
@ -2764,6 +3009,8 @@ module_free(void *m)
Py_CLEAR(current_tasks);
Py_CLEAR(all_tasks);
module_free_freelists();
}
static int

Write Preview
Loading…
Cancel
Save