Class SlidingWindowSemaphore

awscli/s3transfer/utils.py:654–749

A semaphore used to coordinate sequential resource access. This class is similar to the stdlib BoundedSemaphore: * It's initialized with a count. * Each call to ``acquire()`` decrements the counter. * If the count is at zero, then ``acquire()`` will either block until the cou

class SlidingWindowSemaphore(TaskSemaphore):
    """A semaphore used to coordinate sequential resource access.

    This class is similar to the stdlib BoundedSemaphore:

    * It's initialized with a count.
    * Each call to ``acquire()`` decrements the counter.
    * If the count is at zero, then ``acquire()`` will either block until the
      count increases, or if ``blocking=False``, then it will raise
      a NoResourcesAvailable exception indicating that it failed to acquire the
      semaphore.

    The main difference is that this semaphore is used to limit
    access to a resource that requires sequential access.  For example,
    if I want to access resource R that has 20 subresources R_0 - R_19,
    this semaphore can also enforce that you only have a max range of
    10 at any given point in time.  You must also specify a tag name
    when you acquire the semaphore.  The sliding window semantics apply
    on a per tag basis.  The internal count will only be incremented
    when the minimum sequence number for a tag is released.

    """

    def __init__(self, count):
        self._count = count
        # Dict[tag, next_sequence_number].
        self._tag_sequences = defaultdict(int)
        self._lowest_sequence = {}
        self._lock = threading.Lock()
        self._condition = threading.Condition(self._lock)
        # Dict[tag, List[sequence_number]]
        self._pending_release = {}

    def current_count(self):
        with self._lock:
            return self._count

    def acquire(self, tag, blocking=True):
        logger.debug("Acquiring %s", tag)
        self._condition.acquire()
        try:
            if self._count == 0:
                if not blocking:
                    raise NoResourcesAvailable(f"Cannot acquire tag '{tag}'")
                else:
                    while self._count == 0:
                        self._condition.wait()
            # self._count is no longer zero.
            # First, check if this is the first time we're seeing this tag.
            sequence_number = self._tag_sequences[tag]
            if sequence_number == 0:
                # First time seeing the tag, so record we're at 0.
                self._lowest_sequence[tag] = sequence_number
            self._tag_sequences[tag] += 1
            self._count -= 1
            return sequence_number
        finally:
            self._condition.release()