<commit_before>import asyncio
import heapq
import time

from .base import AbstractQueue


class TimestampQueue(AbstractQueue):
    async def init(self):
        self._future = None
        self._queue = []
        self._waiters = []

    def __len__(self):
        return len(self._queue)

    async def get(self):
        timestamp, value = heapq.nsmallest(1, self._queue)[0]
        if time.time() >= timestamp:
            ts, value = heapq.heappop(self._queue)
            return value
        waiter = self.loop.create_future()
        self._waiters.append(waiter)
        if not self._future or self._future.done():
            self._future = self.loop.create_task(self._timer())
        return await waiter

    async def put(self, value):
        heapq.heappush(self._queue, value)
        if self._future and not self._future.done():
            self._future.cancel()
        self._future = self.loop.create_task(self._timer())

    async def _timer(self):
        if not self._queue or not self._waiters:
            return

        ts = heapq.nsmallest(1, self._queue)[0][0]
        t = ts - time.time()
        if t < 0:
            f = self._waiters.pop(0)
            f.set_result(heapq.heappop(self._queue)[1])
        else:
            await asyncio.sleep(t, loop=self.loop)
            self._future = self.loop.create_task(self._timer())


class UniqueQueue(TimestampQueue):
    async def put(self, value):
        ts, val = value
        for i, (t, v) in enumerate(self._queue):
            if v == val:
                self._queue[i] = value
                heapq.heapify(self._queue)
                break
        else:<commit_msg>fix TimestampQueue<commit_after>import asyncio
import heapq
import time

from .base import AbstractQueue


class TimestampQueue(AbstractQueue):
    async def init(self):
        self._future = None
        self._queue = []
        self._waiters = []

    def __len__(self):
        return len(self._queue)

    async def get(self):
        if self._queue:
            timestamp, value = heapq.nsmallest(1, self._queue)[0]
            if time.time() >= timestamp:
                ts, value = heapq.heappop(self._queue)
                return value
            if not self._future or self._future.done():
                self._future = self.loop.create_task(self._timer())
        waiter = self.loop.create_future()
        self._waiters.append(waiter)
        return await waiter

    async def put(self, value):
        heapq.heappush(self._queue, value)
        if self._future and not self._future.done():
            self._future.cancel()
        self._future = self.loop.create_task(self._timer())

    async def _timer(self):
        if not self._queue or not self._waiters:
            return

        ts = heapq.nsmallest(1, self._queue)[0][0]
        t = ts - time.time()
        if t < 0:
            f = self._waiters.pop(0)
            f.set_result(heapq.heappop(self._queue)[1])
        else:
            await asyncio.sleep(t, loop=self.loop)
            self._future = self.loop.create_task(self._timer())


class UniqueQueue(TimestampQueue):
    async def put(self, value):
        ts, val = value
        for i, (t, v) in enumerate(self._queue):
            if v == val:
                self._queue[i] = value
                heapq.heapify(self._queue)
                break
        else:
{'fraction_non_alphanumeric': 0.11028315946348734, 'fraction_numerical': 0.004769001490312966, 'mean_word_length': 1.9106678230702516, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5688650', 'n_tokens_mistral': 1012, 'n_tokens_neox': 986, 'n_words': 234}