Parameter Server

[mrocklin]

I'm restarting the discussion on parameter servers from dask/dask-glm#57

When doing distributed training it is often useful to rapidly update and share parameters between distributed workers. The frequency with which these updates can be shared seems to strongly affect the rate of convergence. The common practice today seems to be to have dedicated nodes within the network that serve as parameter servers, accepting and aggregating updates from workers, and periodically publishing those back out to workers.

Dask's existing task scheduling model is not ideal for this. Every computation or communication checks in with the scheduler, which adds unnecessary network roundtrips and provides a central bottleneck for metadata tracking. To do parameter servers well we probably need to break out of this model and engage more peer-to-peer coordination, without frequent checking in with the scheduler.

Building this infrastructure is not hard, but if possible it would be good to build general machinery that can be used to solve both this problem, and potentially others. I would appreciate having a conversation with a few people to identify what programming and networking constructs might suffice for this.

cc @stsievert @fabianp @MLnick

[mrocklin]

One general mechanism that might fit this need well is Pub/Sub. Under this construct any worker can join a topic as either a publisher or subscriber (or both). It learns from the scheduler who all of its peers are on that topic. When it publishes data on that topic it sends this data directly to all subscribers (or, in the future, to some subscribers that then forward it along to others). Under this model the parameter subscribes to updates and publishes parameters.

An implementation of a single parameter server might look like the following:

def parameter_server(...):
    """ This runs on one special worker """
    parameters = np.random.random(...)

    updates_sub = Sub('updates')     # subscribe to updates from workers
    parameters_pub = Pub('parameters')  # publish new parameters to workers

    async def publish_parameters():
        while not should_stop():
            parameters_pub.put(parameters)    # maybe send only to a few round-robin instead?
            await gen.sleep(0.01)  # maybe control this based on network traffic?

    get_worker().loop.add_callback(publish_parameters)

    for new_data in updates_sub:
        update(parameters, new_data)
        if should_stop():
            return parameters


def worker(data: List[np.ndarray], # these correspond to chunks of data from a dask array
           labels: List[np.ndarray]):
    """ This runs on each worker and gets the arrays present on that worker """
    data = np.concat(data)
    labels = np.concat(labels)

    updates_pub = Pub('updates')
    parameters_sub = Sub('parameters')

    async def update_parameters():
        async for params in parameters_sub:
            parameters = params

    get_worker().loop.add_callback(update_parameters)

    while not should_stop():
        batch = data[np.random.randint(0, len(data))]
        result = compute_update(parameters, batch)
        updates_pub.put(result)

[mrocklin]

To be clear Pub and Sub don't exist today, but they could exist tomorrow if this is an appropriate model. If it isn't then we can build something else. Other ideas are welcome.

[stsievert]

The common practice today seems to be to have dedicated nodes within the network that serve as parameter servers,

That is common practice, and is what distributed Tensorflow uses.

accepting and aggregating updates from workers, and periodically publishing those back out to workers.

In the vanilla case, we only want to update parameters from the parameters we sent to other workers.

In the most simple use case for distributed tensorflow, each worker passes around some parameter global_step that identifies when parameters was pulled from the parameter server, and gathers updates from a certain number of workers. The update is discarded if the global_step a worker sends back is stale. A good reference for this is "Scaling Distributed Machine Learning with the Parameter Server".

This pub/sub framework LGTM. Is there any way to have a worker subscribe to a topic published by another worker in particular? Something like Sub('foo', worker=1) or Pub('foo', worker='nearest_neighbor'). This could enable some of the distributed algorithms (e.g., where nodes are connected to their neighbors and form a ring).

Two extensions:

• Async read/writes could be useful, and enable some new algorithms (e.g., Hogwild!).
  • Async read/writes := having some mechanism so two workers can write to parameters at the same time, and read partially written results. This is probably best accomplished by using threads.
• Gradient coding, which codes the gradient ndarray to some object (typically a dictionary).
  • Two function are defined: encode: ndarray -> dict and decode: dict -> ndarray so decode(encode(x)) == x (well, or maybe only approximately). This would involve the worker running updates_pub.put(encode(result), and the master running update(parameters, decode(new_data)).

I'm inclined to believe these extensions are best suited for a class interface, at least privately.

[mrocklin]

In the most simple use case for distributed tensorflow, each worker passes around some parameter global_step that identifies when parameters was pulled from the parameter server, and gathers updates from a certain number of workers. The update is discarded if the global_step a worker sends back is stale.

I encourage people to start thinking aspirationally here, a bit beyond the simplest case.

I was looking at Downpour and also this paper which had different algorithms that I felt were both probably decently well approximated with PubSub. It may be though that my approximations are significant.

Is there any way to have a worker subscribe to a topic published by another worker in particular?

This would be outside of the paradigm of pub/sub but is certainly something that is doable. I expect pubsub to be something like 200-400 lines of code. We can do more things if they are important.

I am not personally familiar with algorithms that are based on ring-like topologies. Is this common?

Async read/writes could be useful

The pseudocode above does async reads/writes. The pub.put commands return immediately. The subscribe commands would block, but are happening in an async block.

Gradient coding, which codes the gradient ndarray to some object

I'm hoping that this is independent of communication patterns and is something that folks like you would feel comfortable writing after pubsub, or whatever network machinery we need, is put in place.

I'm inclined to believe these extensions are best suited for a class interface, at least privately.

This work is likely to happen in two levels:

1. core dask stuff, like pubsub, which has nothing at all to do with parameter servers
2. ML stuff, which can be as specialized as is necessary

[stsievert]

Got it. This makes more sense now.

am not personally familiar with algorithms that are based on ring-like topologies. Is this common?

At least one algorithm depends on a ring-structure, Hogwild++. This is suited for HPC applications, and shows much better speedups than Hogwild as the number of workers grow. I don't think I'd say this is relevant or that distributed ring-like topologies are super common, but they could be useful.

core dask stuff, like pubsub, which has nothing at all to do with parameter servers ... ML stuff, which can be as specialized as is necessary

Pub/sub looks sufficient for parameter servers. Almost everything I said was ML specific.

[mrocklin]

A draft pubsub implementation is available here: dask/distributed#1999 I suspect that it is still buggy.

I'll work on setting up a mocked out example of the system above and then hopefully others can start playing with things.

My initial experiments show that we have about a 1ms latency when sending messages between workers. I am curious to from people who have an understanding of what kinds of latencies we need to get decent performance in these applications.

[stsievert]

Moderate sized deep learning models have latency of at least 10ms for feeding 1 image through the network for prediction (source: https://dawn.cs.stanford.edu/benchmark/). The time required for optimization of these networks is (likely) no more than 5-6 times as slow, so probably at least 50-100ms for one optimization step.

[mrocklin]

Well then I guess we're in the clear?

…

On Wed, May 23, 2018 at 1:01 PM, Scott Sievert ***@***.***> wrote: Moderate sized deep learning models have latency of at least 10ms for feeding 1 image through the network for prediction (source: https://dawn.cs.stanford.edu/benchmark/). The time required for optimization of these networks is (likely) no more than 5-6 times as slow, so probably at least 50-100ms for one optimization step. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#171 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszIZbuArkTpsK4RmGFvAomjh16NdGks5t1ZXtgaJpZM4UG6C2> .

[stsievert]

1ms latency seems reasonable – latency between cluster nodes for me is (on avg) 0.3ms. I think I'd say it a good first draft, and enough to start experimenting. I can implement a parameter server and look for any bottlenecks.

[mrocklin]

PubSub is merged. I have a tiny toy example here: https://gist.github.com/mrocklin/0d906828544ddeb8e6e1d3d193172ae9

[stsievert]

@mrocklin could I have your timing script to measure the latency? I'd like to compare on the same machine. I'm measuring the latency of PyTorch's distributed module to have a latency of 0.029ms with https://gist.github.com/stsievert/1a93cf732d66f22a3080fcd0729364d6

[mrocklin]

See https://github.com/dask/distributed/blob/12ddc080d1d876b0fce6fe4b2e863fe7c7b31543/distributed/tests/test_pubsub.py#L13

…

On Fri, Jun 8, 2018 at 6:15 PM, Scott Sievert ***@***.***> wrote: @mrocklin <https://github.com/mrocklin> could I have your timing script to measure the latency? I'd like to compare on the same machine. I'm measuring the latency of PyTorch's distributed module to have a latency of 0.029ms with https://gist.github.com/stsievert/ 1a93cf732d66f22a3080fcd0729364d6 — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#171 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszCosKvJ7JX3OxKrLwXmztvOoFe5Aks5t6vd5gaJpZM4UG6C2> .

[stsievert]

Side by side on my machine I measure PyTorch 0.4.0 to have a latency of 0.031ms and dask.distributed master to have a latency of 1.398ms, both between two processes with Python 3.6.5. I measure with the same gist, https://gist.github.com/stsievert/1a93cf732d66f22a3080fcd0729364d6