| 711 | self._condition.release() |
| 712 | |
| 713 | def release(self, tag, acquire_token): |
| 714 | sequence_number = acquire_token |
| 715 | logger.debug("Releasing acquire %s/%s", tag, sequence_number) |
| 716 | self._condition.acquire() |
| 717 | try: |
| 718 | if tag not in self._tag_sequences: |
| 719 | raise ValueError(f"Attempted to release unknown tag: {tag}") |
| 720 | max_sequence = self._tag_sequences[tag] |
| 721 | if self._lowest_sequence[tag] == sequence_number: |
| 722 | # We can immediately process this request and free up |
| 723 | # resources. |
| 724 | self._lowest_sequence[tag] += 1 |
| 725 | self._count += 1 |
| 726 | self._condition.notify() |
| 727 | queued = self._pending_release.get(tag, []) |
| 728 | while queued: |
| 729 | if self._lowest_sequence[tag] == queued[-1]: |
| 730 | queued.pop() |
| 731 | self._lowest_sequence[tag] += 1 |
| 732 | self._count += 1 |
| 733 | else: |
| 734 | break |
| 735 | elif self._lowest_sequence[tag] < sequence_number < max_sequence: |
| 736 | # We can't do anything right now because we're still waiting |
| 737 | # for the min sequence for the tag to be released. We have |
| 738 | # to queue this for pending release. |
| 739 | self._pending_release.setdefault(tag, []).append( |
| 740 | sequence_number |
| 741 | ) |
| 742 | self._pending_release[tag].sort(reverse=True) |
| 743 | else: |
| 744 | raise ValueError( |
| 745 | "Attempted to release unknown sequence number " |
| 746 | f"{sequence_number} for tag: {tag}" |
| 747 | ) |
| 748 | finally: |
| 749 | self._condition.release() |
| 750 | |
| 751 | |
| 752 | class ChunksizeAdjuster: |