System with hybrid communication strategy for large-scale distributed deep learning

Abstract

A computer in a distributed computing system is disclosed. The computer includes: a graphics processing unit (GPU) memory; a central processing unit (CPU) memory comprising a Key-Value Store (KVS) module; an execution engine module configured to run a deep learning (DL) program to create a plurality of operator graph layers in the graphics processing unit memory; a client library module configured to create a GPU-CPU synchronization (GCS) module for each of the plurality of operator graph layers; a coordination service module configured to compute network cost of a first and a second communication scheme and select, based on the network cost, one of the first and second communication scheme for transmitting data associated with one of the plurality of operator graph layers from a corresponding GCS module.

1 . A distributed computing system comprising a computer comprising:
 a graphics processing unit (GPU) memory;   a central processing unit (CPU) memory comprising a Key-Value Store (KVS) module;   an execution engine module configured to run a deep learning (DL) program to create a plurality of operator graph layers in the graphics processing unit memory;   a client library module configured to create a GPU-CPU synchronization (GCS) module for each of the plurality of operator graph layers;   a coordination service module configured to compute network cost of a first and a second communication scheme and select, based on the network cost, one of the first and second communication scheme for transmitting data associated with one of the plurality of operator graph layers from a corresponding GCS module; and   wherein the client library module is further configured to initiate a data transfer from the GCS module using the selected communication scheme.   
     
     
         2 . The system of  claim 1 , wherein the first communication scheme comprises broadcasting data associated with the one of the plurality of operator graph layers from the corresponding GCS module to one or more GCS modules directly. 
     
     
         3 . The system of  claim 2 , wherein the network cost associated with the first communication scheme can be computed as P 2 B (M+N), wherein P is a number of computers in the distributed system, B is a batch size, M and N are dimensions of a matrix associated with the operator graph layer. 
     
     
         4 . The system of  claim 1 , wherein the second communication scheme comprises using the KVS module as an intermediary to transmit data from one GCS to another GCS. 
     
     
         5 . The system of  claim 4 , wherein the network cost associated with the second communication scheme can be computed as PMN, wherein P is a number of computers in the distributed system, M and N are dimensions of a matrix associated with the operator graph layer. 
     
     
         6 . The system of  claim 1 , wherein the client library module is further configured to create send and receive ports for each of the plurality of GCS modules. 
     
     
         7 . The system of  claim 1 , wherein the execution engine module running the DL program comprising populating two operator graphs' model parameters and intermediate values according to input datum. 
     
     
         8 . The system of  claim 7 , wherein the execution engine module is configured to populate the model parameters and intermediate values according to back propagation algorithm. 
     
     
         9 . The system of  claim 1 , wherein at least one of the GCS modules is in communication with the KVS module. 
     
     
         10 . The system of  claim 1 , wherein at least one of the GCS modules is configured to receive data from another GCS module directly. 
     
     
         11 . The system of  claim 1 , wherein at least one of the GCS modules is configured to receive data from a KVS module. 
     
     
         12 . A method of running a Deep Learning (DL) program comprising:
 parsing DL program code;   constructing a plurality of operator graph layers in a GPU memory;   creating a GCS module for each of the operator graph layers;   activating a KVS module in a CPU memory;   computing the network cost of a first and a second communication schemes for transmitting data;   for each GCS module, selecting one of the communication schemes based on the network cost; and   transmitting data from each GCS module using the selected communication scheme;   wherein at least one GCS module uses the first communication scheme and at least one GCS module uses the second communication scheme.   
     
     
         13 . The method of  claim 12 , where transmitting data using the first communication scheme comprises broadcasting data associated with the one of the plurality of operator graph layers from the corresponding GCS module to one or more other GCS modules directly. 
     
     
         14 . The method of  claim 13 , wherein the network cost associated with the first communication scheme is computed as P 2 B (M+N), wherein P is a number of computers in the distributed system, B is a batch size, M and N are dimensions of a matrix associated with the operator graph layer. 
     
     
         15 . The method of  claim 12 , wherein transmitting data using the second communication scheme comprises using the KVS module as an intermediary to transmit data from one GCS to another GCS. 
     
     
         16 . The method of  claim 15 , wherein the network cost associated with the second communication scheme is computed as PMN, wherein P is a number of computers in the distributed system, M and N are dimensions of a matrix associated with the operator graph layer. 
     
     
         17 . The method of  claim 12 , further comprising creating send and receive ports for each of the plurality of GCS modules. 
     
     
         18 . The method of  claim 12 , wherein parsing the DL code comprises populating two operator graphs' model parameters and intermediate values according to input datum. 
     
     
         19 . The method of  claim 12 , further comprising at least one of the GCS modules receiving data from another GCS module directly. 
     
     
         20 . The method of  claim 12 , further comprising at least one of the GCS modules receiving data from a KVS module.