Cache Pipelining: Reducing Round-Trip Latency in Redis Operations

import redis
import time

r = redis.Redis(decode_responses=True)

class PipelineBenchmarker:
    def __init__(self, redis_client):
        self.r = redis_client

    def benchmark_sequential(self, operations=1000):
        """Measure time for N sequential operations."""
        start = time.perf_counter()
        for i in range(operations):
            self.r.set(f"seq:{i}", f"val_{i}", ex=3600)
        seq_time = time.perf_counter() - start

        start = time.perf_counter()
        for i in range(operations):
            self.r.get(f"seq:{i}")
        seq_read_time = time.perf_counter() - start

        return {"write": round(seq_time, 4), "read": round(seq_read_time, 4)}

    def benchmark_pipelined(self, operations=1000):
        """Measure time for N pipelined operations."""
        pipe = self.r.pipeline()

        start = time.perf_counter()
        for i in range(operations):
            pipe.set(f"pipe:{i}", f"val_{i}", ex=3600)
        pipe.execute()
        pipe_time = time.perf_counter() - start

        pipe = self.r.pipeline()
        start = time.perf_counter()
        for i in range(operations):
            pipe.get(f"pipe:{i}")
        pipe.execute()
        pipe_read_time = time.perf_counter() - start

        return {"write": round(pipe_time, 4), "read": round(pipe_read_time, 4)}

    def compare(self, operations=1000):
        """Compare sequential vs pipelined performance."""
        seq = self.benchmark_sequential(operations)
        pipe = self.benchmark_pipelined(operations)

        print(f"Operations: {operations}")
        print(f"{'Method':12s} {'Write(s)':10s} {'Read(s)':10s} {'Speedup':10s}")
        print("-" * 42)
        print(f"{'Sequential':12s} {seq['write']:<10.4f} {seq['read']:<10.4f} {'1.0x':10s}")
        print(f"{'Pipelined':12s} {pipe['write']:<10.4f} {pipe['read']:<10.4f} "
              f"{seq['write']/pipe['write']:.1f}x")

bench = PipelineBenchmarker(r)
bench.compare(1000)

import redis
import time

r = redis.Redis(decode_responses=True)

class PipelineTuner:
    def __init__(self, redis_client):
        self.r = redis_client

    def test_batch_sizes(self, sizes=None):
        """Test throughput for different pipeline batch sizes."""
        if sizes is None:
            sizes = [1, 10, 50, 100, 200, 500, 1000, 2000, 5000]

        results = []
        total_ops = 10000

        for batch_size in sizes:
            batches = total_ops // batch_size
            pipe = self.r.pipeline()

            start = time.perf_counter()
            for batch in range(batches):
                for i in range(batch_size):
                    pipe.set(f"tune:{batch}:{i}", f"val", ex=3600)
                pipe.execute()
                pipe = self.r.pipeline()

            elapsed = time.perf_counter() - start
            throughput = total_ops / elapsed

            results.append({
                "batch_size": batch_size,
                "elapsed": round(elapsed, 3),
                "throughput": round(throughput),
            })

        return results

    def report(self):
        """Generate tuning report."""
        results = self.test_batch_sizes()
        print(f"{'Batch Size':12s} {'Time(s)':10s} {'Ops/sec':12s} {'Efficiency':12s}")
        print("-" * 46)
        best = max(results, key=lambda r: r["throughput"])
        for r in results:
            eff = r["throughput"] / best["throughput"] * 100
            print(f"{r['batch_size']:<12d} {r['elapsed']:<10.3f} "
                  f"{r['throughput']:<12,d} {eff:<11.1f}%")
        return best

tuner = PipelineTuner(r)
best = tuner.report()
print(f"\nOptimal batch size: {best['batch_size']} "
      f"({best['throughput']:,} ops/sec)")

import redis
import time

r = redis.Redis(decode_responses=True)

class MixedPipeline:
    def __init__(self, redis_client):
        self.r = redis_client

    def execute_mixed(self, operations):
        """Execute a mixed set of read/write operations in one pipeline."""
        pipe = self.r.pipeline()
        results_map = {}

        for i, op in enumerate(operations):
            tag = op.get("tag", f"op_{i}")
            cmd = op["command"]

            if cmd == "get":
                pipe.get(op["key"])
                results_map[tag] = ("result", op["key"])
            elif cmd == "set":
                pipe.setex(op["key"], op.get("ttl", 3600), op["value"])
                results_map[tag] = ("status", op["key"])
            elif cmd == "delete":
                pipe.delete(op["key"])
                results_map[tag] = ("deleted", op["key"])
            elif cmd == "incr":
                pipe.incr(op["key"])
                results_map[tag] = ("counter", op["key"])

        results = pipe.execute()
        output = {}
        result_idx = 0

        for tag, (result_type, key) in results_map.items():
            output[tag] = {result_type: results[result_idx], "key": key}
            result_idx += 1

        return output

    def check_and_set_with_pipeline(self, key, expected, new_value, ttl=3600):
        """Check-then-set pattern using pipelining."""
        pipe = self.r.pipeline()
        pipe.get(key)
        pipe.setex(key, ttl, new_value)
        results = pipe.execute()

        old_value = results[0]
        return {
            "key": key,
            "old_value": old_value,
            "new_value": new_value,
            "was_overwritten": old_value is not None and old_value == expected,
        }

pipeline = MixedPipeline(r)

operations = [
    {"tag": "get_user", "command": "get", "key": "mixed:user:1"},
    {"tag": "set_config", "command": "set", "key": "mixed:config:theme", "value": "dark"},
    {"tag": "incr_counter", "command": "incr", "key": "mixed:visitors"},
    {"tag": "get_config", "command": "get", "key": "mixed:config:theme"},
]

results = pipeline.execute_mixed(operations)
for tag, result in results.items():
    print(f"{tag}: {result}")

result = pipeline.check_and_set_with_pipeline(
    "mixed:flag", "old_value", "new_value"
)
print(f"\nCheck-and-set: {result}")