Lets imagine a simple web server that serves both fast-loading pages and also performs some computation on slower loading pages. In our case this will be a simple Fibonnaci serving application, but you can imagine replacing the fib
function for running a machine learning model on some input data, fetching results from a database, etc..
import tornado.ioloop
import tornado.web
def fib(n):
if n < 2:
return n
else:
return fib(n - 1) + fib(n - 2)
class FibHandler(tornado.web.RequestHandler):
def get(self, n):
result = fib(int(n))
self.write(str(result))
class FastHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello!")
def make_app():
return tornado.web.Application([
(r"/fast", FastHandler),
(r"/fib/(\d+)", FibHandler),
])
app = make_app()
app.listen(8000)
We know that users associate fast response time to authoritative content and trust, so we want to measure how fast our pages load. We're particularly interested in doing this during many simultaneous loads, simulating how our web server will respond when serving many users
import asyncio
import tornado.httpclient
client = tornado.httpclient.AsyncHTTPClient()
from time import time
async def measure(url, n=100):
""" Get url n times concurrently. Print duration. """
start = time()
futures = [client.fetch(url) for i in range(n)]
results = await asyncio.gather(*futures)
end = time()
print(url, ', %d simultaneous requests, ' % n, 'total time: ', (end - start))
We see that
await measure('http://localhost:8000/fast', n=1)
await measure('http://localhost:8000/fast', n=100)
await measure('http://localhost:8000/fib/28', n=1)
await measure('http://localhost:8000/fib/28', n=100)
In the example below we see that one call to the slow fib/
route will unfortunately block other much faster requests:
a = asyncio.ensure_future(measure('http://localhost:8000/fib/35', n=1))
b = asyncio.ensure_future(measure('http://localhost:8000/fast', n=1))
await b
There are two problems/opportunities here:
fib
calls are independent, we would like to run these computations in parallel with multiple cores or a nearby cluster.fib
requests can get in the way of our fast requests. One slow user can affect everyone else.To resolve both of these problems we will offload computation to other processes or computers using Dask. Because Dask is an async framework it can integrate nicely with Tornado or Asyncio.
from dask.distributed import Client
dask_client = await Client(asynchronous=True) # use local processes for now
dask_client
def fib(n):
if n < 2:
return n
else:
return fib(n - 1) + fib(n - 2)
class FibHandler(tornado.web.RequestHandler):
async def get(self, n):
future = dask_client.submit(fib, int(n)) # submit work to happen elsewhere
result = await future
self.write(str(result))
class MainHandler(tornado.web.RequestHandler):
async def get(self):
self.write("Hello, world")
def make_app():
return tornado.web.Application([
(r"/fast", MainHandler),
(r"/fib/(\d+)", FibHandler),
])
app = make_app()
app.listen(9000)
By offloading the fib computation to Dask we acheive two things:
We can now support more requests in less time. The following experiment asks for fib(28)
simultaneously from 20 requests. In the old version we computed these sequentially over a few seconds (the last person to request waited for a few seconds while their browser finished). In the new one many of these may be computed in parallel and so everyone gets an answer in a few hundred milliseconds.
# Before parallelism
await measure('http://localhost:8000/fib/28', n=20)
# After parallelism
await measure('http://localhost:9000/fib/28', n=20)
Previously while one request was busy evaluating fib(...)
Tornado was blocked. It was unable to handle any other request. This is particularly problematic when our server provides both expensive computations and cheap ones. The cheap requests get hung up needlessly.
Because Dask is able to integrate with asynchronous systems like Tornado or Asyncio, our web server can freely jump between many requests, even while computation is going on in the background. In the example below we see that even though the slow computation started first, the fast computation returned in just a few milliseconds.
# Before async
a = asyncio.ensure_future(measure('http://localhost:8000/fib/35', n=1))
b = asyncio.ensure_future(measure('http://localhost:8000/fast', n=1))
await b
await a
# After async
a = asyncio.ensure_future(measure('http://localhost:9000/fib/35', n=1))
b = asyncio.ensure_future(measure('http://localhost:9000/fast', n=1))
await b
await a
In these situations people today tend to use concurrent.futures or Celery.
In this context Dask provides some of the benefits of both. It is easy to set up and use in the common single-machine case, but can also scale out to a cluster. It integrates nicely with async frameworks and adds only very small latencies.
async def f():
start = time()
result = await dask_client.submit(lambda x: x + 1, 10)
end = time()
print('Roundtrip latency: %.2f ms' % ((end - start) * 1000))
await f()