BLPOP Randomly Drops First Element

[dave-ccs]

The following Python code demonstrates an issue with BLPOP randomly dropping the first element pushed to a list. It is inconsistent, sometimes working correctly and sometimes not. I've tried it with multiple versions of Garnet on multiple platforms (Windows, Docker, macOS in .NET Core). All demonstrate the same random behavior. The issue does not occur with any version of Redis that I've tested.

I've tried to identify whether it's an issue in Garnet or in the Python Redis library, but I wasn't able to follow the Garnet code well enough to figure out where the issue lies. I only know that the behavior changes depending on the BLPOP timeout value given. A value of 0 (zero) fails frequently. Values of 0.25 or less but greater than zero don't fail or don't fail often. Values of 1 second or greater fail similar to 0 (zero).

To confirm, run the following Python code. You should see a pop for every push. You don't have to let it run too long. It's always the first element that disappears. Running it a few times should show the problem.

import asyncio
import logging
import redis.asyncio as redis_asyncio
import time

logging.basicConfig(
    level=logging.INFO,
    format="{asctime}.{msecs:03.0f} {module}.{funcName}({lineno}) {levelname} {message}",
    datefmt="%H:%M:%S",
    style="{")

_pool = redis_asyncio.ConnectionPool(host="localhost", port=6379, db=0, decode_responses=True)


async def display_list():
    async with redis_asyncio.Redis(connection_pool=_pool) as redis:
        while True:
            value = await redis.blpop("key", 0)  # <- change timeout value here (0, 0.01, 0.1, 0.25, 1.0, etc.)
            if value:
                logging.info(f"pop {value}")
                print()


async def main_async():
    asyncio.create_task(display_list())
    await asyncio.sleep(0.1)

    async with redis_asyncio.Redis(connection_pool=_pool) as redis:
        while True:
            value = time.time()
            logging.info(f"push {value}")
            response = await redis.rpush("key", value)
            assert response == 1
            await asyncio.sleep(1)


if __name__ == "__main__":
    try:
        asyncio.run(main_async())
    except KeyboardInterrupt:
        pass