Cloud Native Computing Foundation KubeCon + CloudNativeCon Europe 2020 - Virtual, 11 Sep 2020

Previous Meeting Next Meeting

⏯

youtube image

►

From YouTube: Production Multi-node Jobs with Gang Scheduling, K8s, GPUs... Madhukar Korupolu & Sanjay Chatterjee

Description

Don’t miss out! Join us at our upcoming events: EnvoyCon Virtual on October 15 and KubeCon + CloudNativeCon North America 2020 Virtual from November 17-20. Learn more at https://kubecon.io. The conferences feature presentations from developers and end users of Kubernetes, Prometheus, Envoy, and all of the other CNCF-hosted projects.

Production Multi-node Jobs with Gang Scheduling, K8s, GPUs and RDMA - Madhukar Korupolu & Sanjay Chatterjee, NVIDIA

With the growing scale of DL and ML applications, distributed execution of jobs across multiple nodes becomes increasingly critical -- to solve bigger problems faster -- as illustrated by the recent MLperf results. However running such workloads in a production K8s cluster shared by multiple jobs/users has several challenges. In this talk, we’ll give an overview of this area -- including distributed Tensorflow, Pytorch, Horovod, MPI -- and the use of GPU nodes with NCCL and RDMA for accelerated performance. We’ll describe our end-to-end flow for multi-node jobs in K8s including gang scheduling, quotas, fairness and backfilling implemented in our custom scheduler for GPUs. Our cluster includes high-speed networking through RoCE and SR-IOV / Multus CNI. We’ll share our design choices, learnings and operational experience including failure handling, performance and telemetry.

https://sched.co/ZejQ

A

Hello, everyone welcome to our virtual cubecon session. We have an interesting topic today. This is on multi-node jobs with kubernetes, I'm madukar, and this will be a joint presentation with my colleague, sanjay chatterjee.

A

A

Our work is motivated by applications of deep learning.

A

As you know, deep learning is becoming popular in several areas: example in transportation, self-driving cars, medical imaging, smart cities, robotics and so on.

A

In each of these areas, deep learning is enabling more automation and also adding new value by uncovering more insights.

A

Now, if you look into the ai deep learning stack, there are several layers: the top layer. We have the different applications like computer vision, with image classification, natural language processing, recommendation engines, languages and so on.

A

At the bottom layer we have the hardware components like gpus cpus, the various systems that are built using these components.

A

Above them are the architecture specific libraries like nickel and so on about those are the deep learning frameworks like tensorflow, pytorch, etc now for an application to work. Well, all these layers must function well together.

A

There's a lot of innovation happening in each of the layers.

A

So, in this talk, we'll see how to put these things together to work well,.

A

So a trend we see in all these applications is that the data is becoming big and models are also increasingly complex and larger as an example for natural language. Processing, bert, is a recent popular model.

A

It began with around 110 million parameters and burt large grew to around 330 million parameters.

A

More recent ones are like 8 billion and 17 billion parameters, so so for such large models, training them takes a long time even with a single gpu.

A

Because of this there's a demand for multi gpu jobs. It's an example. If you look at the graph here, it shows the training time for bird large using different numbers of gpus x axis is the number of nvidia v100, gpus and y-axis is the training time in minutes.

A

So we can see that 16 gpus it takes around 60 hours just around 2.5 days, 1000 gpus. It comes down to about an hour.

A

So this speed up is significant, because the data scientists can then have a faster turnaround, debug faster iterate, faster and improve the results. They can also handle larger problems.

A

Similar trend can also be seen in the mlp of training results. So mlp is a competition where several organizations submit their best training results for popular benchmarks, and even there we see that more gpus feels faster running times.

A

Now, given the demand for these multi gpu systems, how do we build them? So here's a sample, multi gpu node from nvidia called dgx1, so this has eight v100 gpus shown in green here and they're connected by nvi link within the node.

A

The node has dual socket cpus and four nicks. These are melanox edr mix capable of about 100 gbps, so we can run up to an 8 gpu job inside this node using fast and willing, and if you want to go beyond a gpus, then we can connect multiple nodes. Together we can connect them using infiniband or rocky, which is rdma over ethernet.

A

And we do that by connecting these four next to a connected fabric using different topologies like factory dragonfly or other things based on our requirements. So that's the hardware part.

A

Now, let's look at the software side. So how do we train these multi-gpu multi-node systems?

A

So there are two popular approaches: data parallelism and model parallelism, and you can also have a hybrid of these combining the two. So in this talk, we'll focus on data parallelism.

A

Another important concept with deep learning is that of sjd stochastic gradient descent. This is where, in the backward path of the neural network, we compute gradients and apply them to the weights of the neurons.

A

There are two popular hdd approaches: async and sync, so async sjd is the parameter, support approach shown here and sync sjd is the already approach shown in the figure below so here the different workers come gpu nodes. They do a computation and at the end of the cycle they exchange the gradients in all reduced fashion and that's implemented through nickel or mpi.

A

Nickel is nvidia collective communications library and it internally handles unveiling and also infiniband rocky across nodes. So does it makes use of the fast interconnects?

A

The recent horror world library from uber makes it easy to run the all reduced approach with tensorflow and phi torch.

A

So while multi-node is gaining traction, kubernetes is also becoming increasingly popular to orchestrate these clusters, mainly because of its advantages in ease of management operations, cicd and then other aspects over running, multi-node and kubernetes has multiple challenges. So we divide this into two parts.

A

The left side, part one, is how to get a single multi-node job running end-to-end and then the right side is part two, where we look at running multiple multi-node jobs at the same time in a cluster shared cluster.

A

So let's focus on the left side, and here we will look into the end-to-end flow, like array, jobs, mpi, gang scheduling and multi-rail rdma towards that here's, a big picture overview of the flow user submits a multi-node job using a cli or ui, so that goes to the api server.

A

It's a custom, job controller which watches for that and creates an mpi job, and we use the upstream mpi operator from kubeflow that watches for the mpi jobs and then creates corresponding pods for the mpa job.

A

Those parts are then scheduled by the scheduler, using a gang scheduling approach once scheduled. The cubelet takes up the pod and tries to run it, so it pulls the container and the container image has the necessary libraries, it'll have framework tensorflow, pytorch, nickel, mpa and so on.

A

Once these pods are up and running, they communicate using nickel on the back end. So that's how the system works.

A

Now, let's look at a sample by dot job, a bird job with two nodes, how we would do that, so on the left side, we see a command on node 0, node 1, which is rank 0 and then the command on node 2, which is rank 1..

A

So on both these nodes we run torch.distribute.launch and then that starts up eight processes on the node one on each gpu and on this gpu we run the word: train dot, py script, that's a single gpu script, so we run that eight times on node, zero and then mode one can. The master address is set to local host on rank zero and ip address of rank 0 and rank 1.. So that's how they communicate.

A

So, on the right side, we see how we launch the same job using mpi run. So we use mpi run is the launcher. We specify the number of replicas which is array size and then it creates that many instances of the torch distributed launch.

A

Okay, so rest all is same as before, so note that we use mpi run as the launcher here. The back end is still nickel.

A

So we manage these mpi jobs using the mpi operator and for that I'll hand it over to sanjay.

B

Thanks mother girl, hi, I'm sanjay chatterjee and it's great to be presenting at kubecon europe, 2020 virtual, all right, so we have seen how a multi-node job can be submitted by the user. Now let us look inside the kubernetes cluster in slightly more detail. Now, typically, clusters have many different implementations, different hardware, different stack, etc.

B

So we use the term array jobs as an abstraction for all kinds of multi-node jobs and users can configure the type and size of the job they want. After an array, job is submitted to the cluster.

B

A an internal job spec for the cluster is created, and this spec is converted by the control plane components to create the actual k8s pods for a multi-node job.

B

Now we use a component called uh mpi operator from upstream and queue flow, which creates the pods and also manages their life cycle as a group or gang of pods.

B

Now the pod specs include uh the setup to handle all the multi-part job launch, as well as all the node specific resource requests, as shown here and finally, all the parts belonging to a specific job are gang, scheduled by the scheduler to run on the clusters nodes.

B

Okay, next we'll focus on a few key areas in this flow and see. What's going on under the hood.

B

First, let us look at how we enable high-speed networking for multi-node jobs in a kubernetes cluster.

B

Now this is an absolutely key requirement for high performance distributed deep learning applications uh we've seen earlier, a single nvidia, dgx1 node comes equipped with eight volta v100 gpus and four 100 gig melanox rdma nx, along with dual 10 gig ethernet nics.

B

So when the keraspot spec requests all these resources, as shown here with this four sri vrdma resources and a gpus, we need to make all these available to the pod when it lands on the node.

B

Now the cubelet is the kubernetes node agent and it manages all the allocatable resources on the node for a pod, so to first include the cube rdm nx as cubelet managed resources. We use a sriv network device plugin, which will discover the rdma nx and register them with the cubelet all right. So after a pod is bound to a node. The cubelet will start the process of moving the requested network resources from the host namespace to the pods namespace. And to do this, it invokes the associated cni, plugin cni, meaning the container network interface.

B

Since uh the dgx1 node has multiple network interfaces, we use a meta, plugin called multis, which can delegate to multiple other plugins. Now. All of these other plugins can be specific to a different kind of network interface. So we have a sriv cni plug-in to configure the rdma virtual interfaces and a component called flannel which configures the default. Zero interface and both of these are managed by multis.

B

Okay, the next thing we are going to focus on is the scheduling of array jobs, specifically the gang scheduling aspect of it. um We've seen earlier that an array job in kubernetes will create multiple ks pods. Now, a further requirement of these pods is to run as a gang, that is, a all-or-none policy, meaning either all the pods belonging to the same multi-node job will get scheduled to run or none of them can run. Similarly, one fails, then all of them should fail.

B

Now. This is primarily so that deep learning applications which use the synchronous sgd model we talked about earlier now they can make progress when they use collectives, such as all reduce broadcast, etc.

B

Now the default cube scheduler schedule spots, one by one, and so, if parts from different array jobs get scheduled in an interleaved manner, it can lead to deadlocks.

B

So gang scheduling is currently actively discussed in the community, and one proposal is to repeatedly try scheduling the pods, and if that fails, then back off and repeat again and with this approach there might be a potential for live lock. So we are experimenting with a different approach, namely using a reservation system for full node bonds right. So in our approach to gang scheduling, we use something called as a pod groups to be the unit of scheduling for multi-node jobs.

B

Now, pod groups are essentially a logical grouping of pods belonging to this, a single multi-node job, and currently we only support scheduling, full node pods in a multi-node job, so to schedule board groups. We made uh some extensions to the default kubernetes schedule, which include a new scheduling queue that holds bot groups. That is, in addition to the regular pods.

B

We are also experimenting with a new module called the pod group manager, which can reserve full nodes for the pods.

B

Now the node reservations are actually added as a label on the board, and this allows the main scheduling, control loop to bypass the regular uh pod scheduling and directly bind the pods to its reserve nodes.

B

Okay, now we're going to look at a demo that will actually show the scheduling and end-to-end flow of the multi-node jobs. So here I'm submitting a two node by bert pytorch, multi-node training, job and we'll see that the job will be creating uh kubernetes pods. That will be scheduled and will be expected to run on the notes.

B

So uh once the job is created, we can look into the cluster and see that the pods are created and we see that two parts are created: worker, zero and worker, one with the same job id and they are currently in the pending state. So we can look into the pods annotations to look at acute reason to see why it's spending and we can see that the reason is- resources are unavailable.

B

And if you look at the current state of the queue we can see, there's a number of q and pods that are in queue. We can look into the pod spec in more detail to see it's the same command.

B

Eventually, these pods will get to the running state and then the user can actually look at some real-time telemetry, uh some utilization numbers and uh with these at real time they can also look at the status history and some logging as well.

B

B

So now, let's take a look at the job telemetry in slightly more detail.

B

This is an experimental ui created for some internal users and it's a very useful tool for them to monitor their jobs in real time now, as we can see here, uh they can see some key metrics from their jobs, such as utilization numbers for gpus, tensor, ports, cpus memory, pci links and so on, and we've built a monitoring and logging pipeline that exports all these node level, performance, counters and other information to the users, so that they can view how their jobs are progressing in real time.

B

Okay, if you remember, we had submitted a bird multi-node training job in the demo, so here's some throughput scaling numbers from a similar bird job and uh they have two different configurations: phase one and phase two, uh and they are using some sample training data sets.

B

Now we know all the performance uh depends on a number of system and application dependent factors such as batch size for gpu.

B

We have seen that the throughput scaling results, for example, here the sequences per second have been consistently high in terms of their parallel efficiency, and in this case we can see that the application can scale uh close to ideal, even when running on a very large number of gpus. For example, here on 32 nodes, uh the job was uh actually using 256 gpus concurrently and mind you. These results are from a shared cluster, where multiple other jobs were also running and sharing the cluster network all right.

B

So now I'm going to hand it back over to madhukar for the rest of the slides. Thank you.

A

Thank you sanjay, so we now looked at part one which shows how to run a single multi-node job end-to-end using the bird example.

A

Now, let's look at part two here: how do we run multiple multi-node jobs at the same time in a shared cluster, and this needs additional features like quotas, time limits, backfilling, etc? So, let's see how we handle this.

A

So towards this we have a we're building a production, k8s cluster for multi-node jobs.

A

This is a shared on-prem cluster inside a data centers, and it's currently comprised of about 100 dgx nodes arranged in a single hop fashion and it's being tried out by early internal users. So these are data scientists who are interested in training large models so if to enable these users to share the cluster efficiently, we enforce quotas.

A

Each user has a default gpu quota of about 32 gpus, and that determines how many they can use. Concurrently, it's configurable. We also enforce time limits on jobs, a wall, time limit and also a node r limit. Let's say 128 hours, which means a 16 knot. Job can run about eight hours.

A

One key angle with multi-node jobs is that of starvation and bad filling. What this means is because gang scheduling is not blocking. There is a large multi-node job set that will do no job upfront.

A

It can start for a long time.

A

So to work around that we enforces threshold and beyond the threshold we start blocking for that job. This can cause low utilization because of ideal nodes. So then we backfill the nodes using smaller jobs, shorter jobs that helps us to improve utilization. So this is a work in progress. We have some initial things in place, but we're looking for more input and feedback to improve that.

A

We also enforce fairness via the drx score. So drf stands for dominant resource fairness, it's a popular measure. We also use starvation and age as factors for the job and we combine all these using a weighted function. So this gives us a dynamic priority for the job so that the scheduler can use that to decide which job to pick next.

A

For a production cluster, we also need good operations, srd ci, cd and also dashboards and alerts. So here's an example.

A

This is scheduler and mpi operator dashboard. It shows number of jobs in the queue for money over quota. How many under quota it says, scheduling, latency, starvation, duration and so on. Even for the mpa operator, it shows latency and other angles. So this is a regular grafana chart built using prometheus matrix and one can build different dashboards like this easily using grafana.

A

These are you. This can be used by the art, admins and also engineers.

A

We are also trying to enable alerts so that sre can react to changes quickly.

A

So, in summary, we have looked at multi-node clusters with kubernetes how to enable that using gang scheduling, mpi, rdm, etc. So this is a work in progress as next steps we are looking to harden these production deployments also need to improve performance, using storage, caching and so on. Multi-Node clusters also looking to add additional array types like mpi spark.

A

Also aligning with upstream framework and so on so there's a work in progress, and we would like to hear from other folks who are trying similar things.

A

One thing I would like to draw attention to is the multi-node enabled containers so for frameworks like tensorflow pytorch, and also models like bird with multi-node enabled containers with all the necessary libraries and packages these are available from ngc.nbc.com, so we encourage all of you to download it from there and use them.

A

Finally, I would like to acknowledge the contributions of various team members within nvidia and also outside folks, who are working on different layers of the stack. The deep learning stack it all comes together would also like to thank kubecon for giving us this opportunity.

A

Thank you and we'll now take questions.

A

So we have a few questions. uh First, one is on whether the telemetry dashboard is open sourced, so the components we are using are open sourced, for example, dcgm, which is data center, gpu manager, it exports the different gpu metrics to prometheus and other components.

A

So that's upstream and using that in grafana, one can build different kinds of dashboards. The particular one we are using here is an experimental one that we built for some internal teams and we are trying to get feedback and see what makes sense, etc.

A

Also, looking at the uh slide here, several other components we are using are already upstream, for example, the rocky cni mpi operator on already upstream also the gpu specific things like device plug-in gpu operator upstream. So using these one can start putting together a different multi-node solutions.

A

Okay, second question is: if we can, it will be sharing the slides on sked.com sure yes, we'll upload them after this.

A

A

Yeah one thing I think.

A

Also, we'd like to add that if anybody has tried simple things, we would be interested to hear your experience and feedback. So we look forward to discussing with you either now or on slack afterlife.

B

Okay, um so I can see a question here, uh so the question is: is the gang scheduling code uh pgm, open sourced?

B

So, as we said earlier in the talk that we have, this experimental module called the pod group manager and we are looking at different aspects of it. uh Some of the work is currently in progress, things that we talked about earlier about starvation, handling, backfilling, etc.

B

We're trying out different things looking at seeing what it what works at scale uh and then. Finally, once we are confident about uh the code, then we'll go to the next steps um upstream it and we'll see how that works.

B

Okay, um there's another question here: uh this is about so the question is: what is the network bandwidth requirements for shared gpu experiments.

B

So a large majority of the multi-node jobs in our clusters distributed training applications and they use um what motherboard talked about earlier. The synchronous sgd model.

B

So these applications um heavily use uh collective primitives, like all reduce, broadcast, etc, and all of these collective primitives they require high speed high bandwidth networks.

B

So our job is to enable these applications to use the these high bandwidth networks as easily and transparently as possible, so nvidia gpus. They come equipped with envy link and- and it also supports.

B

The nodes they support, gpu, detect rdma across nodes and to integrate all of this there there's a library called nickel that we talked about earlier and so applications can transparently uh use nvi, link and gpu direct rdma using nickel by either using nickel as part of the application or using a framework like by torture tensorflow.

B

Overall, I think the network requirements or its utilization depends on many factors like how the application is written.

B

They can either use collectives through nickel or they can use some sort of communication and computation overlaps.

B

So applications are.

B

But we have seen that most applications have been able to scale well with very good parallel efficiency on these clusters.

B

So yeah, that's.

A

So we have another question which is on comparisons with volcano. So when we started few months back, volcano was still not in a production, ready state and also we had some.

A

Gpu requirements like working with high speed networks, rocky and so on, and also dgx specific ones like working with full node jobs. So our solution is kind of tailored towards gpu and djx specific requirements, and that's what we are demoing in production and yeah. I think so. We can now perhaps check with volcano and see what can be added there and so on.

A

Yeah one thing we would also add is that, as we see on the future work screen, different things need to come together uh for this to work in production like the cni mpi stuff, um and also like efficient storage being able to download the containers efficiently. So all these things need to work together and hardened in production forever to have a good cluster.

A

So our goal has been to kind of share our learnings from these things and also learn from the community.

A

Okay, I think we can now close the session and follow up on the slack channel for discussion. Thank you all for.

A