fiorix/txredisapi — pull request #56 by mjpizz (2013-12-21 06:02:58+00:00)
Title: Added support for pipelining

The [redis pipelining docs](http://redis.io/topics/pipelining) describe how to serially write multiple commands at once to improve performance. As far as I can tell, txredisapi doesn't support this yet, so I put together an implementation that works pretty similarly to transactions.

It works like this:

```python
db = yield txredisapi.Connection(...)

pipe = db.pipeline()
pipe.set("foo", 123)
pipe.set("bar", 987)
results = yield pipe.execute_pipeline()
```

To benchmark this, I compared two approaches:

* 1000 `SET` commands in parallel
* 1000 `SET` commands in a pipeline

The pipeline approach is 6X faster for this use case.

---
Benchmark code:

```python
import time, txredisapi
from twisted.internet import defer, reactor

redis_host = "localhost"
redis_port = 6379
numkeys = 1000

@defer.inlineCallbacks
def benchmark_parallel():
db = yield txredisapi.Connection(redis_host, redis_port, reconnect=False)
deferreds = []
keys_and_values = [("parallel%s" % i, i) for i in range(numkeys)]
start = time.time()
for key, value in keys_and_values:
deferreds.append(db.set(key, value))
yield defer.DeferredList(deferreds)
print "PARALLEL:", time.time() - start
yield sanitycheck(db, keys_and_values)

@defer.inlineCallbacks
def benchmark_pipeline():
db = yield txredisapi.Connection(redis_host, redis_port, reconnect=False)
keys_and_values = [("pipeline%s" % i, i) for i in range(numkeys)]
start = time.time()
pipeline = yield db.pipeline()
for key, value in keys_and_values:
pipeline.set(key, value)
yield pipeline.execute_pipeline()
print "PIPELINE:", time.time() - start
yield sanitycheck(db, keys_and_values)

@defer.inlineCallbacks
def sanitycheck(db, keys_and_values):
results = yield db.mget([key for key, value in keys_and_values])
for result, (key, value) in zip(results, keys_and_values):
if result != value:
print "FAILED SANITY CHECK"

@defer.inlineCallbacks
def run():
yield benchmark_parallel()
yield benchmark_pipeline()
reactor.stop()

if __name__ == "__main__":
reactor.callLater(0, run)
reactor.run()
```

Benchmark results:
```
PARALLEL: 0.397115945816
PIPELINE: 0.0628750324249
```