(
self, opname, get_items, read, write, *, check_items=False
)
| 629 | |
| 630 | @requires_jit_disabled |
| 631 | def assert_races_do_not_crash( |
| 632 | self, opname, get_items, read, write, *, check_items=False |
| 633 | ): |
| 634 | # This might need a few dozen loops in some cases: |
| 635 | for _ in range(self.LOOPS): |
| 636 | items = get_items() |
| 637 | # Reset: |
| 638 | if check_items: |
| 639 | for item in items: |
| 640 | reset_code(item) |
| 641 | else: |
| 642 | reset_code(read) |
| 643 | # Specialize: |
| 644 | for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): |
| 645 | read(items) |
| 646 | if check_items: |
| 647 | for item in items: |
| 648 | self.assert_specialized(item, opname) |
| 649 | else: |
| 650 | self.assert_specialized(read, opname) |
| 651 | # Create writers: |
| 652 | writers = [] |
| 653 | for _ in range(self.WRITERS): |
| 654 | writer = threading.Thread(target=write, args=[items]) |
| 655 | writers.append(writer) |
| 656 | # Run: |
| 657 | for writer in writers: |
| 658 | writer.start() |
| 659 | read(items) # BOOM! |
| 660 | for writer in writers: |
| 661 | writer.join() |
| 662 | |
| 663 | @requires_specialization |
| 664 | def test_binary_subscr_getitem(self): |
no test coverage detected