cas_attempt_1.py

# Attempt 1 at implementing compare-and-swap (CAS) using Redis transactions.
# This attempt fails since pipeline.get can't be evaluated within the scope
# of a transaction. Its return value cannot be used within a transaction,
# so we cannot conditionally invoke pipeline.set based on what is returned by pipeline.get.
# Below, current_value isn't actually None, but a reference to a pipeline.get operation
# to be queued into the transaction.

import redis
from multiprocessing import Process
import time


def check_and_set(key, value):
    cache = redis.Redis(host="localhost", port=6379)
    txn = cache.pipeline()
    txn.watch(key)
    try:
        txn.multi()
        current_value = txn.get(key)
        if current_value is None:
            txn.set(key, value)
        txn.execute()
    except redis.exceptions.WatchError:
        pass


if __name__ == "__main__":
    num_processes = 8
    key = f"key{time.time()}"

    processes = []
    for i in range(num_processes):
        p = Process(target=check_and_set, args=(key, i))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()

cas_attempt_2.py

# Attempt 2 at implementing compare-and-swap (CAS) using Redis transactions
# This attempt fails since pulling pipeline.get our of the transaction results
# in the attempted CAS implementation being non-atomic.
# We end up with torn CASes and therefore races conditions.

import redis
from multiprocessing import Process
import time


def check_and_set(key, value):
    cache = redis.Redis(host="localhost", port=6379)
    txn = cache.pipeline()
    txn.watch(key)
    current_value = txn.get(key)
    if current_value is None:
        try:
            txn.multi()
            txn.set(key, value)
            txn.execute()
        except redis.exceptions.WatchError:
            pass


if __name__ == "__main__":
    num_processes = 8
    key = f"key{time.time()}"

    processes = []
    for i in range(num_processes):
        p = Process(target=check_and_set, args=(key, i))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()

cas_attempt_3.py

# Attempt 3 at implementing compare-and-swap (CAS) using Lua scripts.
# This one was successful. This is because, for whatever reason, Redis supports
# the execution of arbitrary Lua scripts, and these scripts are executed *atomically*
# by the server. Since Lua script execution is atomic, we can group the necessary read
# and write transactions into the same script to execute a CAS.

import redis
import time
from multiprocessing import Process


def check_and_set(key, value):
    cache = redis.Redis(host="localhost", port=6379)
    success = cache.eval("""
        if redis.call('EXISTS', KEYS[1]) == 0 then
            redis.call('SET', KEYS[1], ARGV[1]);
            return 1;
            end;
        return 0;
    """, 1, key, value)
    if success:
        print("success!")


if __name__ == "__main__":
    num_processes = 8
    key = f"key{time.time()}"

    processes = []
    for i in range(num_processes):
        p = Process(target=check_and_set, args=(key, i))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()