Skip to content

Ttl counter

TTLCounter

TTLCounter is a counter/list that expires items after a TTL period expires.

Source code in video_sampler/ttl_counter.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class TTLCounter:
    """TTLCounter is a counter/list that expires items after a TTL period expires."""

    def __init__(self, max_ttl: int) -> None:
        self.inner_counter = []
        self.max_ttl = max_ttl

    def __len__(self):
        """Return the number of items in the counter."""
        return len(self.inner_counter)

    def add_item(self, hash: str):
        """Add an item with the max TTL."""
        heapq.heappush(self.inner_counter, (self.max_ttl, hash))

    def tick(self):
        """Decrease the TTL of all items by 1."""
        for i, (ttl, hash) in enumerate(self.inner_counter):
            self.inner_counter[i] = (ttl - 1, hash)

    def expire_one(self):
        """Expire the first item if its TTL is 0. Expires AT MOST one item."""
        # peek the first item
        ttl, hash = self.inner_counter[0]
        if ttl <= 0:
            heapq.heappop(self.inner_counter)
            return hash
        return None

    def expire_all(self):
        """Expire all items."""
        for _, hash in self.inner_counter:
            yield hash
        self.inner_counter.clear()

__len__()

Return the number of items in the counter.

Source code in video_sampler/ttl_counter.py 
11
12
13
def __len__(self):
    """Return the number of items in the counter."""
    return len(self.inner_counter)

add_item(hash)

Add an item with the max TTL.

Source code in video_sampler/ttl_counter.py 
15
16
17
def add_item(self, hash: str):
    """Add an item with the max TTL."""
    heapq.heappush(self.inner_counter, (self.max_ttl, hash))

expire_all()

Expire all items.

Source code in video_sampler/ttl_counter.py 
33
34
35
36
37
def expire_all(self):
    """Expire all items."""
    for _, hash in self.inner_counter:
        yield hash
    self.inner_counter.clear()

expire_one()

Expire the first item if its TTL is 0. Expires AT MOST one item.

Source code in video_sampler/ttl_counter.py 
24
25
26
27
28
29
30
31
def expire_one(self):
    """Expire the first item if its TTL is 0. Expires AT MOST one item."""
    # peek the first item
    ttl, hash = self.inner_counter[0]
    if ttl <= 0:
        heapq.heappop(self.inner_counter)
        return hash
    return None

tick()

Decrease the TTL of all items by 1.

Source code in video_sampler/ttl_counter.py 
19
20
21
22
def tick(self):
    """Decrease the TTL of all items by 1."""
    for i, (ttl, hash) in enumerate(self.inner_counter):
        self.inner_counter[i] = (ttl - 1, hash)