add.py

from operator import add

from toolz import sliding_window
from tornado import gen
from tornado.ioloop import IOLoop

from distributed import Worker, Scheduler, Client, Nanny

loop = IOLoop.current()

s = Scheduler(loop=loop)
s.start(0)
workers = [Nanny(s.ip, s.port, loop=loop) for i in range(4)]

from dask import delayed
@delayed
def inc(x):
    return x + 1

L = [inc(i) for i in range(5000)]
for i in range(4):
    L = [add(x, y) for x, y in sliding_window(2, L)]
total = delayed(sum, pure=True)(L)

@gen.coroutine
def f():
    yield [w._start(0) for w in workers]
    c = Client(s.address, loop=loop, start=False)
    yield c._start()

    future = c.compute(total)

    result = yield future._result()
    print(result)

loop.run_sync(f)


# pip install snakeviz
# python -m cProfile -o prof-add.out add.py
# snakeviz prof.out

inc.py

from distributed import Worker, Scheduler, Client, Nanny

from tornado import gen
from tornado.ioloop import IOLoop

loop = IOLoop.current()

s = Scheduler(loop=loop)
s.start(0)
workers = [Nanny(s.ip, s.port, loop=loop) for i in range(4)]

from dask import delayed
@delayed
def inc(x):
    return x + 1

L = [inc(i) for i in range(5000)]
for i in range(4):
    L = [inc(i) for i in L]
total = delayed(sum, pure=True)(L)

@gen.coroutine
def f():
    yield [w._start(0) for w in workers]
    c = Client(s.address, loop=loop, start=False)
    yield c._start()

    future = c.compute(total)

    result = yield future._result()
    print(result)

loop.run_sync(f)


# pip install snakeviz
# python -m cProfile -o prof-inc.out inc.py
# snakeviz prof.out