| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -351,22 +351,19 @@ def _set_concurrent_future_state(concurrent, source): | |||
| 351 | 351 | def _copy_future_state(source, dest): | |
| 352 | 352 | """Internal helper to copy state from another Future. | |
| 353 | 353 | ||
| 354 | - The other Future may be a concurrent.futures.Future. | ||
| 354 | + The other Future must be a concurrent.futures.Future. | ||
| 355 | 355 | """ | |
| 356 | - assert source.done() | ||
| 357 | 356 | if dest.cancelled(): | |
| 358 | 357 | return | |
| 359 | 358 | assert not dest.done() | |
| 360 | - if source.cancelled(): | ||
| 359 | + done, cancelled, result, exception = source._get_snapshot() | ||
| 360 | + assert done | ||
| 361 | + if cancelled: | ||
| 361 | 362 | dest.cancel() | |
| 363 | + elif exception is not None: | ||
| 364 | + dest.set_exception(_convert_future_exc(exception)) | ||
| 362 | 365 | else: | |
| 363 | - exception = source.exception() | ||
| 364 | - if exception is not None: | ||
| 365 | - dest.set_exception(_convert_future_exc(exception)) | ||
| 366 | - else: | ||
| 367 | - result = source.result() | ||
| 368 | - dest.set_result(result) | ||
| 369 | - | ||
| 366 | + dest.set_result(result) | ||
| 370 | 367 | ||
| 371 | 368 | def _chain_future(source, destination): | |
| 372 | 369 | """Chain two futures so that when one completes, so does the other. | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -558,6 +558,33 @@ def set_exception(self, exception): | |||
| 558 | 558 | self._condition.notify_all() | |
| 559 | 559 | self._invoke_callbacks() | |
| 560 | 560 | ||
| 561 | + def _get_snapshot(self): | ||
| 562 | + """Get a snapshot of the future's current state. | ||
| 563 | + | ||
| 564 | + This method atomically retrieves the state in one lock acquisition, | ||
| 565 | + which is significantly faster than multiple method calls. | ||
| 566 | + | ||
| 567 | + Returns: | ||
| 568 | + Tuple of (done, cancelled, result, exception) | ||
| 569 | + - done: True if the future is done (cancelled or finished) | ||
| 570 | + - cancelled: True if the future was cancelled | ||
| 571 | + - result: The result if available and not cancelled | ||
| 572 | + - exception: The exception if available and not cancelled | ||
| 573 | + """ | ||
| 574 | + # Fast path: check if already finished without lock | ||
| 575 | + if self._state == FINISHED: | ||
| 576 | + return True, False, self._result, self._exception | ||
| 577 | + | ||
| 578 | + # Need lock for other states since they can change | ||
| 579 | + with self._condition: | ||
| 580 | + # We have to check the state again after acquiring the lock | ||
| 581 | + # because it may have changed in the meantime. | ||
| 582 | + if self._state == FINISHED: | ||
| 583 | + return True, False, self._result, self._exception | ||
| 584 | + if self._state in {CANCELLED, CANCELLED_AND_NOTIFIED}: | ||
| 585 | + return True, True, None, None | ||
| 586 | + return False, False, None, None | ||
| 587 | + | ||
| 561 | 588 | __class_getitem__ = classmethod(types.GenericAlias) | |
| 562 | 589 | ||
| 563 | 590 | class Executor(object): | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -413,23 +413,23 @@ def func_repr(func): | |||
| 413 | 413 | def test_copy_state(self): | |
| 414 | 414 | from asyncio.futures import _copy_future_state | |
| 415 | 415 | ||
| 416 | - f = self._new_future(loop=self.loop) | ||
| 416 | + f = concurrent.futures.Future() | ||
| 417 | 417 | f.set_result(10) | |
| 418 | 418 | ||
| 419 | 419 | newf = self._new_future(loop=self.loop) | |
| 420 | 420 | _copy_future_state(f, newf) | |
| 421 | 421 | self.assertTrue(newf.done()) | |
| 422 | 422 | self.assertEqual(newf.result(), 10) | |
| 423 | 423 | ||
| 424 | - f_exception = self._new_future(loop=self.loop) | ||
| 424 | + f_exception = concurrent.futures.Future() | ||
| 425 | 425 | f_exception.set_exception(RuntimeError()) | |
| 426 | 426 | ||
| 427 | 427 | newf_exception = self._new_future(loop=self.loop) | |
| 428 | 428 | _copy_future_state(f_exception, newf_exception) | |
| 429 | 429 | self.assertTrue(newf_exception.done()) | |
| 430 | 430 | self.assertRaises(RuntimeError, newf_exception.result) | |
| 431 | 431 | ||
| 432 | - f_cancelled = self._new_future(loop=self.loop) | ||
| 432 | + f_cancelled = concurrent.futures.Future() | ||
| 433 | 433 | f_cancelled.cancel() | |
| 434 | 434 | ||
| 435 | 435 | newf_cancelled = self._new_future(loop=self.loop) | |
@@ -441,7 +441,7 @@ def test_copy_state(self): | |||
| 441 | 441 | except BaseException as e: | |
| 442 | 442 | f_exc = e | |
| 443 | 443 | ||
| 444 | - f_conexc = self._new_future(loop=self.loop) | ||
| 444 | + f_conexc = concurrent.futures.Future() | ||
| 445 | 445 | f_conexc.set_exception(f_exc) | |
| 446 | 446 | ||
| 447 | 447 | newf_conexc = self._new_future(loop=self.loop) | |
@@ -454,6 +454,56 @@ def test_copy_state(self): | |||
| 454 | 454 | newf_tb = ''.join(traceback.format_tb(newf_exc.__traceback__)) | |
| 455 | 455 | self.assertEqual(newf_tb.count('raise concurrent.futures.InvalidStateError'), 1) | |
| 456 | 456 | ||
| 457 | + def test_copy_state_from_concurrent_futures(self): | ||
| 458 | + """Test _copy_future_state from concurrent.futures.Future. | ||
| 459 | + | ||
| 460 | + This tests the optimized path using _get_snapshot when available. | ||
| 461 | + """ | ||
| 462 | + from asyncio.futures import _copy_future_state | ||
| 463 | + | ||
| 464 | + # Test with a result | ||
| 465 | + f_concurrent = concurrent.futures.Future() | ||
| 466 | + f_concurrent.set_result(42) | ||
| 467 | + f_asyncio = self._new_future(loop=self.loop) | ||
| 468 | + _copy_future_state(f_concurrent, f_asyncio) | ||
| 469 | + self.assertTrue(f_asyncio.done()) | ||
| 470 | + self.assertEqual(f_asyncio.result(), 42) | ||
| 471 | + | ||
| 472 | + # Test with an exception | ||
| 473 | + f_concurrent_exc = concurrent.futures.Future() | ||
| 474 | + f_concurrent_exc.set_exception(ValueError("test exception")) | ||
| 475 | + f_asyncio_exc = self._new_future(loop=self.loop) | ||
| 476 | + _copy_future_state(f_concurrent_exc, f_asyncio_exc) | ||
| 477 | + self.assertTrue(f_asyncio_exc.done()) | ||
| 478 | + with self.assertRaises(ValueError) as cm: | ||
| 479 | + f_asyncio_exc.result() | ||
| 480 | + self.assertEqual(str(cm.exception), "test exception") | ||
| 481 | + | ||
| 482 | + # Test with cancelled state | ||
| 483 | + f_concurrent_cancelled = concurrent.futures.Future() | ||
| 484 | + f_concurrent_cancelled.cancel() | ||
| 485 | + f_asyncio_cancelled = self._new_future(loop=self.loop) | ||
| 486 | + _copy_future_state(f_concurrent_cancelled, f_asyncio_cancelled) | ||
| 487 | + self.assertTrue(f_asyncio_cancelled.cancelled()) | ||
| 488 | + | ||
| 489 | + # Test that destination already cancelled prevents copy | ||
| 490 | + f_concurrent_result = concurrent.futures.Future() | ||
| 491 | + f_concurrent_result.set_result(10) | ||
| 492 | + f_asyncio_precancelled = self._new_future(loop=self.loop) | ||
| 493 | + f_asyncio_precancelled.cancel() | ||
| 494 | + _copy_future_state(f_concurrent_result, f_asyncio_precancelled) | ||
| 495 | + self.assertTrue(f_asyncio_precancelled.cancelled()) | ||
| 496 | + | ||
| 497 | + # Test exception type conversion | ||
| 498 | + f_concurrent_invalid = concurrent.futures.Future() | ||
| 499 | + f_concurrent_invalid.set_exception(concurrent.futures.InvalidStateError("invalid")) | ||
| 500 | + f_asyncio_invalid = self._new_future(loop=self.loop) | ||
| 501 | + _copy_future_state(f_concurrent_invalid, f_asyncio_invalid) | ||
| 502 | + self.assertTrue(f_asyncio_invalid.done()) | ||
| 503 | + with self.assertRaises(asyncio.exceptions.InvalidStateError) as cm: | ||
| 504 | + f_asyncio_invalid.result() | ||
| 505 | + self.assertEqual(str(cm.exception), "invalid") | ||
| 506 | + | ||
| 457 | 507 | def test_iter(self): | |
| 458 | 508 | fut = self._new_future(loop=self.loop) | |
| 459 | 509 | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -6,6 +6,7 @@ | |||
| 6 | 6 | PENDING, RUNNING, CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED, Future) | |
| 7 | 7 | ||
| 8 | 8 | from test import support | |
| 9 | + from test.support import threading_helper | ||
| 9 | 10 | ||
| 10 | 11 | from .util import ( | |
| 11 | 12 | PENDING_FUTURE, RUNNING_FUTURE, CANCELLED_FUTURE, | |
@@ -282,6 +283,62 @@ def test_multiple_set_exception(self): | |||
| 282 | 283 | ||
| 283 | 284 | self.assertEqual(f.exception(), e) | |
| 284 | 285 | ||
| 286 | + def test_get_snapshot(self): | ||
| 287 | + """Test the _get_snapshot method for atomic state retrieval.""" | ||
| 288 | + # Test with a pending future | ||
| 289 | + f = Future() | ||
| 290 | + done, cancelled, result, exception = f._get_snapshot() | ||
| 291 | + self.assertFalse(done) | ||
| 292 | + self.assertFalse(cancelled) | ||
| 293 | + self.assertIsNone(result) | ||
| 294 | + self.assertIsNone(exception) | ||
| 295 | + | ||
| 296 | + # Test with a finished future (successful result) | ||
| 297 | + f = Future() | ||
| 298 | + f.set_result(42) | ||
| 299 | + done, cancelled, result, exception = f._get_snapshot() | ||
| 300 | + self.assertTrue(done) | ||
| 301 | + self.assertFalse(cancelled) | ||
| 302 | + self.assertEqual(result, 42) | ||
| 303 | + self.assertIsNone(exception) | ||
| 304 | + | ||
| 305 | + # Test with a finished future (exception) | ||
| 306 | + f = Future() | ||
| 307 | + exc = ValueError("test error") | ||
| 308 | + f.set_exception(exc) | ||
| 309 | + done, cancelled, result, exception = f._get_snapshot() | ||
| 310 | + self.assertTrue(done) | ||
| 311 | + self.assertFalse(cancelled) | ||
| 312 | + self.assertIsNone(result) | ||
| 313 | + self.assertIs(exception, exc) | ||
| 314 | + | ||
| 315 | + # Test with a cancelled future | ||
| 316 | + f = Future() | ||
| 317 | + f.cancel() | ||
| 318 | + done, cancelled, result, exception = f._get_snapshot() | ||
| 319 | + self.assertTrue(done) | ||
| 320 | + self.assertTrue(cancelled) | ||
| 321 | + self.assertIsNone(result) | ||
| 322 | + self.assertIsNone(exception) | ||
| 323 | + | ||
| 324 | + # Test concurrent access (basic thread safety check) | ||
| 325 | + f = Future() | ||
| 326 | + f.set_result(100) | ||
| 327 | + results = [] | ||
| 328 | + | ||
| 329 | + def get_snapshot(): | ||
| 330 | + for _ in range(1000): | ||
| 331 | + snapshot = f._get_snapshot() | ||
| 332 | + results.append(snapshot) | ||
| 333 | + | ||
| 334 | + threads = [threading.Thread(target=get_snapshot) for _ in range(4)] | ||
| 335 | + with threading_helper.start_threads(threads): | ||
| 336 | + pass | ||
| 337 | + # All snapshots should be identical for a finished future | ||
| 338 | + expected = (True, False, 100, None) | ||
| 339 | + for result in results: | ||
| 340 | + self.assertEqual(result, expected) | ||
| 341 | + | ||
| 285 | 342 | ||
| 286 | 343 | def setUpModule(): | |
| 287 | 344 | setup_module() | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -0,0 +1,3 @@ | |||
| 1 | + Speed up :mod:`asyncio` performance of transferring state from thread | ||
| 2 | + pool :class:`concurrent.futures.Future` by up to 4.4x. Patch by J. Nick | ||
| 3 | + Koston. | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments