CN115336236B - Method implemented by first computing node, first computing node and readable medium - Google Patents

Abstract

In distributed training, in order to avoid network congestion, the first computing node may be based at least in part on whether the first network interface controller associated with the first process and the second network interface controller associated with the second process are located in the same A computing node is in or linked to the same switch to determine a routing identifier for routing data from a first process to a second process, the first process and the second process belonging to a specific inter-node ring connecting a plurality of different nodes under the network topology. The first computing node may then route the data from the first process to the second process based on the routing identifier.

Description

Method implemented by a first computing node, first computing node, and readable medium

Technical field

The present application relates to the field of computer technology, and in particular to a method implemented by a first computing node, the first computing node, and a readable medium.

Background technique

With the rapid development of neural networks such as Deep Neural Network (DNN), various application fields (such as computer vision, natural language processing, speech recognition, etc.) have been developed and will benefit from the inherent diversity of neural networks. Benefit from functionality and flexibility. However, due to the increasing complexity of neural network applications and increasingly stringent accuracy requirements, the size of neural network models and the size of training data required to train the models have also increased significantly, which will inevitably lead to increasingly longer training times. The longer it is, thus adversely affecting the effectiveness and timeliness of the training model to meet changing application environments.

To reduce the time it takes to train a neural network model, a distributed training system using parallel training can be used. Generally speaking, a distributed training system may include a large number of computing nodes or servers distributed over a network, and a subset of computing tasks may be assigned to the computing nodes or servers for performing computations using parallel training. However, data communication between computing nodes or servers in a distributed training system creates a lower limit or bottleneck on the amount of reduction in training time that can occur in a distributed training system. This is especially true when a distributed training system includes various types of heterogeneous connections or interconnections within and between computing nodes or servers that exhibit different performance in terms of latency, bandwidth, topology, etc. characteristic. This heterogeneity of connections or interconnections increases the difficulty and complexity of designing data communication networks for computing nodes or servers in a distributed training system.

Additionally, network congestion may occur due to excessive data flow through specific network switches or connections between compute nodes or servers in a distributed training system, which may result in extended training times due to delays in processing training results. Excessive data flow through a particular network switch or connection may be due to an out-of-control routing of data being sent between compute nodes or servers.

Contents of the invention

The present application provides a method implemented by a first computing node, the first computing node and a readable medium to solve the problem of network congestion.

In a first aspect, a method implemented by a first computing node is provided, comprising: based at least in part on a first network interface controller associated with a first process and a second network interface controller associated with a second process. Determine the routing identifier for routing data from the first process to the second process whether it is located in the same computing node or linked to the same switch, and the first process and the second process belong to the connection under the network topology A specific inter-node ring of a plurality of different nodes; and routing the data from the first process to the second process based on the routing identifier.

In a second aspect, one or more machine-readable media are provided having machine-readable instructions stored thereon that, when executed by a first computing node, cause the first computing node to perform an action, the Actions include determining to transfer the data based at least in part on whether a first network interface controller associated with the first process and a second network interface controller associated with the second process are located in the same compute node or linked to the same switch. A routing identifier for routing from the first process to the second process, the first process and the second process belonging to a specific inter-node ring connecting multiple different nodes under a network topology; and according to the routing identifier The operator routes the data from the first process to the second process.

In a third aspect, a first computing node is provided, including: one or more processing units; and a memory that stores machine-executable instructions, which when executed by the one or more processing units, causes The one or more processing units perform actions that include based, at least in part, on whether a first network interface controller associated with the first process and a second network interface controller associated with the second process are co-located. The computing node or connected to the same switch determines the routing identifier for routing data from the first process to the second process. The first process and the second process belong to a network topology that connects multiple different processes. a specific inter-node ring of nodes; and routing the data from the first process to the second process based on the routing identifier.

In a fourth aspect, a first computing node is provided, comprising: a determining module for determining based, at least in part, on a first network interface controller associated with the first process and a second network interface controller associated with the second process. Whether the processor is located in the same computing node or linked to the same switch to determine the routing identifier for routing data from the first process to the second process, the first process and the second process belong to the network topology a specific inter-node ring connecting a plurality of different nodes; and a routing module for routing the data from the first process to the second process based on the routing identifier.

In a fifth aspect, a computer program product is provided, including a computer program, wherein when the computer program is run in a computer, the computer program is used to implement the method described in the first aspect or any possible implementation manner of the first aspect.

Description of drawings

Detailed description is given with reference to the accompanying drawings. In the drawings, the leftmost digit of a reference number indicates the figure in which the reference number first appears. The use of the same reference numbers in different drawings indicates similar or identical references.

Figure 1 shows an example environment in which a distributed training system may be applied.

Figure 2 shows an example compute node in more detail.

Figure 3A shows a ring configuration interconnecting a preset number of nodes.

FIG. 3B shows a halving and doubling configuration in which a preset number of nodes are connected to each other.

Figure 4 shows a schematic diagram of an exemplary cluster communications library.

Figure 5 illustrates an exemplary topology-aware multi-stage algorithm.

Figure 6 illustrates an exemplary ring-based algorithm for computing the intra-node reduction spread phase of a node.

Figure 7 illustrates an exemplary halving-doubling algorithm for computing the intra-node reduction spread phase of a node.

Figure 8 shows an exemplary halve-double algorithm for the inter-node global reduction phase.

Figure 9 shows an exemplary halving-doubling algorithm for the inter-node global reduction phase in more detail.

Figure 10 illustrates an exemplary ring-based cluster communication algorithm.

Figure 11 shows an example scenario of executing the intra-node reduction spreading phase, the inter-node global reduction phase, and the intra-node global aggregation phase in a parallel or overlapping manner.

FIG. 12 illustrates an exemplary fat-tree network topology.

Figure 13 shows an example scenario using the first congestion avoidance method.

FIG. 14 shows an example scenario using the second congestion avoidance method.

Figure 15 illustrates an exemplary topology-aware multi-stage approach.

Figure 16 illustrates a first exemplary network congestion avoidance method.

Figure 17 illustrates a second exemplary network congestion avoidance method.

Figure 18 shows an exemplary parallel approach based on hybrid architecture in distributed training.

Detailed ways

Overview

As mentioned above, existing distributed training systems create performance bottlenecks for good scalability due to data communication between computing nodes in the distributed training system. In addition, due to the diversity of network structures (including, for example, Ethernet, InfiniBand, PCIe, NVLink, NVSwitch, QPI/UPI, etc.) and the high differences in network characteristics (such as latency, bandwidth, topology, etc.), distributed training Systems often do not make good use of this heterogeneous type of connection or interconnection to perform cluster data operations in and between compute nodes and to transfer data between compute nodes. Additionally, network congestion occurs because the path selection for routing data sent between compute nodes can get out of control, causing excessive data flow through specific network switches or connections between compute nodes in a distributed training system and resulting in Delay in processing training results results in extended training time. Furthermore, existing distributed training systems fail to differentiate between algorithms for different types of underlying structures for cluster operations, thus resulting in poor performance.

This disclosure describes an exemplary distributed training system. In implementation, an exemplary distributed training system may employ a structure-aware cluster communication library that enables the distributed training system to scale linearly. In an implementation, the cluster communication library may tailor communication algorithms to achieve desired or maximum efficiency based at least in part on analysis of the underlying structure and supporting network architecture. In implementation, a distributed training system can split a basic operation into multiple sub-operations, each using one type of structure.

In implementation, an exemplary distributed training system may implement a hybrid algorithm that allows multiple algorithms to coexist in a single cluster operation and selectively employs algorithms for specific structures to improve or maximize the efficiency of the entire communication path. . In implementation, a distributed training system can employ a dual-process parallel algorithm that launches two concurrent processes and pipelines the use of intra-node and inter-node connections, thereby improving performance by overlapping intra-node and inter-node communication. Communication efficiency.

In an implementation, an exemplary distributed training system may employ a probe-based routing control mechanism that generates a mapping from connections to paths, thereby reordering participants or processes in a cluster operation and routing them across the distributed training system. Data flows are mapped to specific physical links to distribute or spread connections to different aggregation or intermediate switches in the communications network to avoid network congestion.

This application describes a number of different embodiments and implementations. The following section describes an exemplary framework suitable for practicing various implementations. Next, this application describes an exemplary system, device, and process for implementing a distributed training system.

Example Environment

1 shows an exemplary environment 100 that can be used to implement a distributed training system. The environment 100 may include a distributed training system 102. In this example, the distributed training system 102 may include a plurality of computing nodes or servers 104-1, 104-2, ..., 104-K (hereinafter collectively referred to as computing nodes 104), where K is a positive integer greater than 1. In implementation, the plurality of computing nodes 104 may communicate data with each other via a communication network 106.

Computing node 104 may be implemented as any variety of computing devices with computing/processing and communication capabilities, including but not limited to servers, desktop computers, notebook or portable computers, handheld devices, netbooks, Internet devices, tablets, mobile devices (e.g. Mobile phones, personal digital assistants, smart phones, etc.), etc., or combinations thereof.

Communication network 106 may be a wireless or wired network, or a combination thereof. Network 106 may be a collection of independent networks that are interconnected with each other and serve as a single large network (eg, the Internet or an intranet). Examples of such independent networks include, but are not limited to, telephone networks, cable networks, Local Area Networks (LANs), Wide Area Networks (WANs), and Metropolitan Area Networks (MANs). Additionally, a stand-alone network may be a wireless or wired network, or a combination thereof. A wired network may include electrical carrier connections (eg, communications cables, etc.) and/or optical carriers or connections (eg, fiber optic connections, etc.). Wireless networks may include, for example, WiFi networks, other radio frequency networks (e.g., Bluetooth Zifeng (Zigbee, etc.) etc. In an implementation, communication network 106 may include a plurality of inter-node interconnects or switches 108-1, 108-2, ..., 108-L (hereinafter collectively referred to as inter-node switches 108) for providing connectivity between computing nodes 104. , where L is a positive integer greater than 1.

In an implementation, environment 100 may further include client device 110 . A user may instruct distributed training system 102 to perform training on a specific learning model (eg, a deep neural network model) based on data sent to distributed training system 102 from client device 110 .

Example compute node

Figure 2 shows the compute node 104 in greater detail. In an implementation, the computing node 104 may include, but is not limited to, one or more processing units 202 , input/output (I/O) interfaces 204 and/or one or more network interfaces 206 , and memory 208 . In an implementation, the compute node 104 may further include one or more intra-node interconnects or switches 210 .

In implementations, processing unit 202 may be configured to execute instructions stored in memory 208 and/or received from input/output interface 204 and/or network interface 206 . In an implementation, the processing unit 202 may be implemented as one or more hardware processors, including, for example, a microprocessor, a dedicated instruction set processor, a physical processing unit (Physics Processing Unit, PPU), a central processing unit (Central Processing Unit, CPU). ), graphics processing unit, digital signal processor, tensor processing unit, etc. Additionally or alternatively, the functions described herein may be performed, at least in part, by one or more hardware logic components. For example, but not limited to, example types of hardware logic components that can be used include field programmable gate arrays (FPGAs), application-specific integrated circuits (Application-Specific Integrated Circuits, ASICs), application-specific standard products (Application-Specific Standard Product, ASSP), System-on-a-Chip System (SOC), Complex Programmable Logic Device (CPLD), etc.

Memory 208 may include machine-readable media in the form of volatile memory, such as random access memory (Random Access Memory, RAM) and/or non-volatile memory, such as read only memory (Read Only Memory, ROM) or flash RAM. Memory 208 is an example of a machine-readable medium.

Machine-readable media may include volatile or nonvolatile types, removable or non-removable media that may utilize any method or technology for storage of information. This information may include machine-readable instructions, data structures, program modules or other data. Examples of machine-readable media include, but are not limited to, phase-change memory (Phase-Change Memory, PRAM), static random access memory (Static Random Access Memory, SRAM), dynamic random access memory (Dynamic Random Access Memory, DRAM), Other types of random access memory (Random-AccessMemory, RAM), read-only memory (Read-Only Memory, ROM), electrically erasable programmable read-only memory (Electronically Erasable Programmable Read-Only Memory, EEPROM), fast flash memory or other internal storage technology, Compact Disk Read-Only Memory (CD-ROM), Digital Versatile Disc (DVD) or other optical storage, tape cassette, magnetic disk storage or other magnetic storage device, or any other non-transmission medium that can be used to store information that can be accessed by a computing node. As defined herein, machine-readable media does not include any transient media, such as modulated data signals and carrier waves.

In an implementation, network interface 206 may be configured to connect computing node 104 to other computing nodes through communication network 106 . In an implementation, the network interface 206 may be established through a Network Interface Controller (NIC), which may use hardware and software to connect the computing node 104 to the communication network 106 . In implementation, each type of NIC may use a different type of fabric or connector to connect to the physical medium associated with the communications network 106 . Examples of various types of structures or connectors can be found in the IEEE 802 specification and can include, for example, Ethernet (defined in 802.3), Token Ring (defined in 802.5), Wireless Network (defined in 802.11) , unlimited bandwidth (InfiniBand), etc.

In implementations, intra-node switch 210 may include various types of interconnects or switches, which may include, but are not limited to, high-speed serial computer expansion buses (e.g., PCIe, etc.), serial multi-channel short-range communications links (e.g., Nolan, It is a wired-based communication protocol (serial multi-channel short-range communication link), a switch chip with multiple ports (such as NVSwitch, etc.), point-to-point processor interconnect (such as Intel QPI/UPI, etc.), etc.

Although in this example, only hardware components are described in compute node 104 , in other examples, compute node 104 may further include other hardware components and/or other software components, such as to execute instructions stored in memory 208 to perform Program modules 212 for various operations, and program data 214 for storing data received for training, intermediate and final results calculated during training, etc.

Example cluster communication algorithm

3A and 3B illustrate exemplary cluster communication algorithms that may be used in distributed training system 102. In implementation, cluster communication algorithms may include, but are not limited to, ring-based communication algorithms, halve-double communication algorithms, etc.

3A illustrates a ring configuration that interconnects a predetermined number of nodes (eg, N nodes, where N is a positive integer greater than 1) with a plurality of connections (ie, N connections) and connects data (eg, , data packet or message) is divided into multiple data blocks (i.e., N data blocks) for transmission, and communication of multiple steps (N-1 steps in this example) is required to complete the cluster operation. At each step, a node may receive data from one of its neighboring nodes, perform specific operations on the received data to obtain local results, and forward the received data to another of its neighboring nodes. After N-1 steps, each node in the ring has data from other nodes in the ring, and the final result is spread to all nodes, which requires an additional N-1 steps to broadcast the respective local results. For each node, the total data size forwarded is 2S, where S represents the data size or message size.

Figure 3B illustrates a halved-double configuration interconnecting a predetermined number of nodes (eg, N nodes, where N is a positive integer greater than 1). In this halved configuration, nodes communicate with each other in pairs, and each step of communication requires only N/2 connections. In the first step, adjacent nodes are paired together, sending half of the messages or data to their respective paired nodes and receiving the other half for processing. Therefore, intermediate results can be spread to paired nodes. In subsequent steps, new pairs are formed with increased or doubled distances, and the data size for processing is halved. After log ₂ N steps of communication, the results are spread across all nodes in the halved-double configuration. The local results in the node are then broadcast to other nodes via additional log ₂ N steps of communication.

Example cluster communication library

FIG. 4 shows a schematic diagram depicting an example cluster communication library 400 that may be used by the distributed training system 102. In implementation, the cluster communication library is a communication library designed to provide high performance, high scalability and strong availability, and can be configured not only for operations such as global reduction (Allreduce) and global aggregation (Allgather). It provides support for standard cluster operations as well as other custom operations for custom applications. In an implementation, the cluster communication library 400 may employ different types of interconnects or switches with different characteristics (e.g., latency, bandwidth, topology) and provide a mechanism to collect information about the underlying hardware in the network and compute nodes so that The design of a topology-aware algorithm is developed based on one or more pieces of collected information.

In implementation, the cluster communication library 400 may provide flexibility to allow multiple algorithms to be executed in a single operation and improve performance (e.g., communication and training performance) by exploiting parallelism between intra-node and inter-node communication. wait). Additionally, the cluster communication library 400 can utilize multiple NICs in compute nodes with legacy or new mapping algorithms and eliminate network congestion through topology-aware placement of connections.

In implementation, cluster communication library 400 may include a software stack 402. In implementation, software stack 402 may include multiple components, which may include but are not limited to transport component 404, operation component 406, communicator component 408, and library context component 410. In implementation, software stack 402 may be designed in a modular manner to allow generality and extensibility.

In implementation, the transmission component 404 may be responsible for the transfer or transmission of peer-to-peer (P2P) data in intra-node and inter-node communications. As an example and not a limitation, the cluster communication library 400 may support TCP (Transmission Control Protocol) and RDMA (Remote Direct Memory Access) for inter-node communication, as well as P2P structures for intra-node communication, such as PCIe (Peripheral Component Interconnect Express), NVLink/NVSwitch, and QPI/UPI (Quick Path Interconnect/Ultra Path Interconnect), etc. For RDMA communication, the transmission component 404 may also be configured to manage memory regions (MR) and corresponding memory buffers in a processing unit (such as a graphics processing unit (GPU) device) and a host memory.

In an implementation, the operations component 406 may provide a set of basic operations and various network algorithms. For example, the basic operations may be configured with algorithms supported by the cluster communication library 400. In addition, the operation component 406 can allow users to define new operations based on these basic operations to implement heterogeneity-aware operations that can employ optimal or better algorithms for each type of structure.

In an implementation, communicator component 408 may be associated with a software process and may be configured to perform manipulation and processing on a processing unit, such as a GPU device. The communicator component 408 may save or log information about other peer components (eg, sort IDs, IP addresses, etc.) and maintain connections with peer components. In an implementation, the communicator component 408 may further collect intra-node and inter-node topology information and use this information to guide algorithm design. In an implementation, intra-node information may include, but is not limited to, types of interconnects, distances between locations of processing units, distances between processing units and network interface controllers, and the like. In an implementation, the inter-node information may include, but is not limited to, for example, the number of available network interface controllers, the topology of the cluster or compute nodes, and the location of the compute nodes in the cluster.

In an implementation, library context component 410 may be configured to expose one or more application interfaces for setting system configuration (eg, environment variables), managing communicator component 408, and providing other functionality such as logging.

Additionally, in some cases, the cluster communications library 400 may further include or provide a number of tools and utilities 412 for topology-aware design, testing and evaluation, and usability improvements. By way of example, and not limitation, tools and utilities 412 may include performance testing tools for transport components 404 to assist in algorithm design and evaluation, probe-based routing mechanisms to ensure system availability, and other functionality, such as scalable to support in addition to Device management capabilities for devices other than GPUs.

Exemplary topology-aware multi-stage algorithm for cluster communication

In an implementation, cluster communication may be defined as communication involving a group of processing units or processes, and operations of the cluster communication may be performed together by all processing units or processes included in the group. Examples of cluster communication operations may include, but are not limited to, global reduction operations, global aggregation operations, reduce-scatter operations, etc. In implementation, global reduction operations are one of many important foundations for cluster communication in distributed training and involve performing reductions on data across processes in the group. Examples of reductions may include, but are not limited to, summation operations, averaging operations, maximization operations, minimization operations, and the like.

As an example and not a limitation, the global reduction operation is used here to illustrate how to divide the cluster operation into multiple micro-operations or sub-operations. In implementation, the distributed training system 102 may adopt a topology-aware multi-stage algorithm that divides the global reduction operation into multiple micro-operations or sub-operations, and selectively selects one or more micro-operations or sub-operations as needed. , thereby reducing the amount of data transferred by eliminating micro-operations or sub-operations that may not be needed. In implementations, the distributed training system 102 may decouple cluster communication algorithms from micro-operations or sub-operations and allow independent or individual matching between algorithms and micro-operations or sub-operations based on underlying structural information, thereby reducing transmission Data volume, maximize or optimize bandwidth utilization.

FIG. 5 illustrates an exemplary topology-aware multi-stage algorithm 500 that may be used with distributed training system 102. In an implementation, the topology-aware multi-stage algorithm 500 may include multiple stages, such as an intra-node reduction spreading stage 502, an inter-node global reduction stage 504, and an intra-node global aggregation stage 506.

In an implementation, the distributed training system 102 may first allocate portions of the data to be processed for training to multiple computing nodes 104 such that each computing node 104 of the multiple computing nodes 104 can receive the corresponding portion of the data. In an implementation, each computing node 104 may divide the corresponding portion of the data into multiple data slices (eg, N data slices, where N is a positive integer), and allocate the multiple data slices to the data included in the corresponding computing node 104 Multiple local processing units or processes in (for example, N local processing units or processes).

In an implementation, during the intra-node reduction spreading phase 502, each local processing unit or process included in each compute node 104 may divide its assigned data slice into a plurality of data chunks (e.g., M chunks ). Local processing units or processes included in each compute node 104 may then cooperatively perform intra-node reduction spread sub-operations to obtain multiple values in the corresponding compute node 104 in multiple steps or iterations according to a particular cluster communication algorithm. All reduction results of the data block. At the end of the intra-node reduction spread phase 502 , a local processing unit or process included in a compute node 104 may have the reduction results (or referred to as a reduction) of all processing units or processes included in the compute node 104 in different data blocks. about dispersion results).

By way of example and not limitation, specific cluster communication algorithms are described using two exemplary cluster communication algorithms, a ring-based algorithm and a halving-doubling algorithm, to illustrate specific mechanisms or operations in the intra-node reduction spread phase 502 . However, other cluster communication algorithms may be used in the intra-node reduction spreading phase 502. For example, the distributed training system 102 may select a particular cluster communication algorithm for use in the intra-node reduction dispersion phase 502 based on information collected by the cluster communication library 400 on multiple factors. In an implementation, the plurality of factors may include, but are not limited to, the type of interconnections between processing units (or processes) in the computing node, the number of interconnections in the computing node, and the like.

For example, in the intra-node reduction dispersion phase 502, the distributed training system 102 may employ a first cluster communication algorithm for a first computing node and a second cluster communication algorithm for a second computing node, where the second computing node has the same A computing node may have the same or different processing and connection capabilities, and the first cluster communication algorithm may be the same as or different from the second cluster communication algorithm. By way of example, and not limitation, for compute nodes interconnected using NVSwitch or PCIe and including a power of two number of processing units or processes for training, the distributed training system 102 may employ a halve-double algorithm, and Distributed training system 102 employs a ring-based algorithm for another compute node interconnected using NVLink or other and using a non-power-of-two number of processing units or processes for training, and so on.

Figure 6 illustrates an exemplary ring-based algorithm 600 for computing nodes in the intra-node reduction spread phase 502. For brevity and description purposes, the exemplary ring-based algorithm only includes a one-ring configuration. However, any ring-based algorithm may be employed that includes a configuration of more than one ring, eg, each ring processing a portion of a data block.

In this example, the computing node described includes M processing units or processes (with ordering identifiers or numbers 1, 2, ..., M), and the data assigned to each processing unit or process is divided into M data blocks. In the first step, the processing unit or process (such as P1) can send one of its M data blocks to the next processing unit or process (such as P2) in the ring, receive another data block from the previous processing unit or process (such as PM) in the ring, and reduce the received data block with the corresponding local data block to obtain a partial reduction result. In each subsequent step (such as the kth step), the processing unit or process (such as P1) can send a partial reduction result (in this example, the partial reduction result obtained by P1 in the k-1th step) to the next processing unit or process (such as P2) in the ring, receive a partial reduction result from the previous processing unit or process (such as PM) (in this example, the partial reduction result obtained by PM in the k-1th step), and reduce the received partial reduction result with another local data block that was not previously sent or reduced with other data.

As shown in Figure 6, in each step, different data blocks can be received and reduced or sent by different processing units or processes in the computing node. Additionally, each processing unit or process can send or receive and reduce different chunks of data (or partial results) at different steps. At the end of the intra-node reduction spread phase 502 (i.e., after M-1 steps), each processing unit or process may include a result data block that stores the results of the M processing units or processes in the compute node. The reduction results of M corresponding data blocks. For example, after M-1 steps, the data block of P1 at the "top position" stores the reduction results of all data blocks corresponding to the M processing units or processes of the "top position", as shown in Figure 6.

Figure 7 illustrates an exemplary halving-doubling algorithm 700 for computing nodes in the intra-node reduction spread phase 502. In this example, the described computing node includes M processing units or processes (in this example, M is set to 8 for description). In the first step, a processing unit or process (e.g. P1) can send half of the data allocated to it to another nearby processing unit or process (e.g. P2) and receive the half of the data allocated to that other processing unit or process (e.g. P2). P2) half of the data, and use the other half of the data allocated to this processing unit or process (such as P1) to reduce the received data to obtain a partial reduction result. At each subsequent step, the processing unit or process (i.e., P1) can send half of the partial reduction result obtained locally in the previous step to a different location located further and further away from the processing unit or process (i.e., P1). processing unit or process, and reduces the received partial reduction result with the other half of the partial reduction result obtained locally in the previous step to obtain a new partial reduction result for the processing unit or process (i.e., P1). At the end of the intra-node reduction spread phase 502 (i.e., after log ₂ M steps, ie, after 3 steps in this example as shown in Figure 7), each processing unit or process may include result data block, the result data block stores the reduction results of M corresponding data blocks of M (in this example, 8 as shown in Figure 7) processing units or processes in the computing node. For example, after log ₂ M steps, the data block storage of P1 at the "bottom position" corresponds to M (in this example, 8 as shown in Figure 7 ) Processes the reduction results of all data blocks of a unit or process.

In an implementation, in the inter-node global reduction stage 504, the inter-node global reduction sub-operations are node-based (i.e., between different computing nodes), and may be performed on processing units (or processes) included in different computing nodes. ). In an implementation, the processing units (or processes) of different computing nodes holding data blocks with the same reduction result (or reduction spread result) form the same group, and transfer their respective results to each other in the same group to execute the nodes between global reduction suboperations. At the end of the inter-node global reduction phase 504, each processing unit or process of each compute node in a specific group may have a specific data block of the reduction results of all processing units or processes in the same group, processing of different groups The units or processes have different data blocks of the reduction results of the corresponding processing units or processes in different groups.

In implementation, the distributed training system 102 may select a specific cluster communication algorithm based on one or more selection criteria, and may implement the inter-node global reduction sub-operation based on the selected cluster communication algorithm. Examples of specific cluster communication algorithms may include, but are not limited to, ring-based algorithms (e.g., hierarchical ring algorithms, multi-ring algorithms, etc.), halving and doubling algorithms, etc. In implementation, one or more selection criteria may include, but are not limited to, the topology of the communication network (e.g., communication network 206) connecting the computing nodes, the number of switches used in the communication network, the type of switches used in the communication network, the network type of the communication network, etc.

By way of example and not limitation, specific cluster communication algorithms are described using two exemplary cluster communication algorithms, a ring-based algorithm and a halving-doubling algorithm, to illustrate specific mechanisms or operations in the inter-node global reduction stage 504 . However, other cluster communication algorithms may be used in the inter-node global reduction stage 504 based on one or more of the selection criteria described above.

Figures 8 and 9 illustrate an exemplary halve-double algorithm for the inter-node global reduction stage 504. In this example, as shown in FIG. 8 , for simplicity and description purposes, the distributed training system 102 is described as including a plurality of computing nodes (i.e., Node 0, Node 1, Node 2, ... Node N-1, in order to To illustrate, N is shown as 4 in Figure 8), where each compute node includes eight processing units or processes, with corresponding sequence numbers (i.e., sequence 0, sequence 1, sequence 2, ... sequence M-1, for the purpose of illustration, M is shown as 8) in Figure 8. As shown in FIG. 8 , processing units or processes with the same sorting number in corresponding computing nodes include data blocks of the same reduction result (or reduction spread result), and form the same group. For example, a processing unit or process with order number 0 in the corresponding computing node includes the data block of the reduction result at the first position among the local data blocks, and they form the same group (eg, group 0). In an implementation, processing units or processes in different groups may not communicate with each other.

In an implementation, the inter-node global reduction sub-operation can be performed individually among the processing units (or processes) in each group, so that each processing unit (or process) in the group can obtain all processing units in the same group ( or process) all reduction results of the same data block. Similar to the mechanism of the halve-double algorithm described above for the intra-node reduction spread phase, processing units or processes in each group can iteratively send the local reduction results of the corresponding data block with other processing units or processes in the corresponding group. , receiving each local reduction result of the corresponding data block from other processing units or processes in doubled or increased distances, and performing a reduction operation on the received reduction result using the local reduction result.

FIG9 shows an example scenario of applying the halving and doubling algorithm to eight computing nodes. In this example, as shown in FIG9 , the number of steps executed in the inter-node global reduction phase 504 using the halving and doubling algorithm is log ₂ N=log ₂ 8=3, where N is the number of computing nodes. In the first step, a first processing unit or process (e.g., a processing unit or process with a sorting number of 0) of a group in a first computing node (e.g., node 0) can send its local reduction result to a second processing unit or process of the same group in a second computing node (e.g., node 1), receive the local reduction result from the second processing unit or process of the same group in the second computing node, and perform a reduction operation on its local reduction result and the received local reduction result to obtain a new local reduction result.

In the second step, the first processing unit or process (e.g., the processing unit or process with order number 0) in the first computing node (e.g., node 0) may send its new local reduction result to the third A third processing unit or process of the same set (e.g., sequence number 0) in the compute node (i.e., Node 2 in this example) receives the local reduction result, and performs a reduction operation on its new local reduction result and the received local reduction result to obtain another new local reduction result.

In the third (or final) step, the same operation is performed on the first processing unit or process, but this time on the same set of fourth processing units in the fourth computing node (i.e., node 4 in this example) or process performs the same operation.

At the end of the inter-node global reduction phase 504, each processing unit or process of each compute node in a specific group may have a specific data block of the reduction results of all processing units or processes in the same group, processing of different groups The units or processes have different data blocks of the reduction results of the corresponding processing units or processes in different groups.

Similar to the halve-double algorithm, the inter-node global reduction sub-operation can be performed individually among the processing units (or processes) in each group of multiple computing nodes (such as N computing nodes) using a ring-based algorithm, such that Each processing unit (or process) in a group can obtain all reduction results of the same data block for all processing units (or processes) in the same group. Similar to the mechanism of the ring-based algorithm described above for the intra-node reduction spread phase, each group of processing units or processes in a compute node can iteratively send the local reduction results of the corresponding data block to the next compute node. The processing unit or process of the corresponding group receives the local reduction result of the corresponding data block from the processing unit or process of the corresponding group in the previous computing node, and uses its local reduction result to perform the reduction operation on the received reduction result. . At the end of the inter-node global reduction phase 504 (i.e. after N-1 steps), each processing unit or process of each compute node in a particular group may have the reduction results for all processing units or processes in that same group A specific data block, different groups of processing units or processes have different data blocks of the reduction results of the corresponding processing units or processes in different groups.

In an implementation, similar to the intra-node reduction spreading stage 502 , in the intra-node global aggregation stage 506 , the global aggregation sub-operation may span local processing units in each of the plurality of computing nodes of the distributed training system 102 or Processes are executed to locally broadcast the corresponding reduction results obtained in the inter-node global reduction stage 504 to each other in the same computing node. At the end of the intra-node global aggregation phase 506, each processing unit or process in each computing node of the distributed training system 102 may have the reduction results of the entire data distributed among the multiple computing nodes.

By way of example and not limitation, a ring-based algorithm is used herein to illustrate how reduction results obtained by local processing units or processes in the computing nodes of the distributed training system 102 (in the inter-node global reduction stage 504) are broadcast. However, the distributed training system 102 may adopt different or the same cluster communication algorithm (such as the halving and doubling algorithm, etc.) for different computing nodes. For example, distributed training system 102 may employ different or the same cluster communication algorithms for different computing nodes based on multiple factors associated with each individual computing node. In an implementation, the plurality of factors may include, but are not limited to, the type of interconnections between processing units (or processes) in the computing node, the number of interconnections in the computing node, and the like.

Figure 10 illustrates an exemplary ring-based cluster communication algorithm 1000 for broadcasting individual reduction results of processing units or processes to each other within computing nodes of a distributed training system 102. As shown in Figure 10, in the first step, each of the M processing units or processes (eg, P1) in the computing node can be used in the inter-node global reduction stage 504 according to the ring configuration. The obtained reduction results are sent to one of the two adjacent processing units or processes (e.g., P2 in this case) and are retrieved from the other of the two adjacent processing units or processes (e.g., PM in this case). ) receives the reduction result. At each subsequent step, each processing unit or process (e.g., P1) can send the newly received reduction result to one of the two adjacent processing units or processes (e.g., P2 in this example) according to the ring configuration , and receives another reduction result from another one of the two adjacent processing units or processes (e.g., PM in this case). At the end of the intra-node global aggregation phase 506 (ie, after M-1 steps), each processing unit or process in the compute node may have the reduction result of the reduction results for all processing units or processes in the compute node.

Example parallel algorithm

In an implementation, the distributed training system 102 may execute multiple stages included in the topology-aware multi-stage algorithm, namely, an intra-node reduction dispersion stage 502 , an inter-node global reduction stage 504 and an intra-node global aggregation stage 506 in sequence. wait. In an implementation, the distributed training system 102 may alternatively partially or substantially overlap some of the intra-node reduction spreading phase 502 , the inter-node global reduction phase 504 , and the intra-node global aggregation phase 506 and execute them in parallel. Some of these stages.

For example, since the intra-node reduction dispersion phase 502 and the intra-node global aggregation phase 506 involve intra-node data communication or transmission (i.e., data communication or transmission within the computing node), and the inter-node global reduction phase 504 involves inter-node data communication or transmission (i.e., data communication or transmission between computing nodes). In an implementation, the distributed training system 102 may allow at least a portion of the intra-node reduction spreading phase 502 and the inter-node global reduction phase 504 to be executed in parallel, and the nodes part of the inter-node global reduction phase 504 and the intra-node global aggregation phase 506, thereby improving the utilization of intra-node and inter-node links (or connections) and preventing intra-node links from being idle when inter-node links are in use, and vice versa. Likewise.

Figure 11 shows an example scenario of executing the intra-node reduction spreading phase, the inter-node global reduction phase, and the intra-node global aggregation phase in a parallel or overlapping manner. As shown in Figure 11, the processing unit or process of the compute node can divide the data block into multiple chunks (in this example, 4 chunks as shown in Figure 11) and distribute these chunks to at least two concurrent threads (eg first thread 1102 and second thread 1104). In this manner, a processing unit or process can pipeline intra-node and inter-node sub-operations for execution by at least two concurrent threads (in this example, first thread 1102 and second thread 1104).

By way of example and not limitation, the first thread 1102 may perform an inter-node global reduce sub-operation (i.e., the operation in the inter-node global reduce stage 504) on a first data block (e.g., data block 1106), while the second thread 1104 performs an intra-node reduce-scatter sub-operation (i.e., the operation in the intra-node reduce-scatter stage 502) on a second data block (e.g., data block 1108). For example, the first thread 1102 may perform an intra-node global gather sub-operation (i.e., the operation in the intra-node global gather stage 506) on a third data block (e.g., data block 1110), while the second thread 1104 performs an inter-node global reduce sub-operation on a fourth data block (e.g., data block 1112).

By way of example and not limitation, another operation involved in distributed neural network training can be further used as an example. In implementation, the distributed training system 102 may divide the global aggregation operation involved in distributed neural network training into multiple sub-operations, namely, the inter-node global aggregation sub-operation, the intra-node global aggregation sub-operation and the data copy sub-operation. In implementation, the inter-node global aggregation sub-operation may be similar to the inter-node global reduction sub-operation as described above, but broadcast data (e.g., the result of the reduction) rather than the reduction operation (e.g., with the local result The result received by the reduction), and the inter-node global aggregation sub-operation may be similar to or equivalent to the inter-node global aggregation sub-operation as described above. In an implementation, the data replicator operation may include an operation that replicates the result data (eg, the final reduction result) as an output parameter.

In an implementation, the processing unit or process of the computing node may divide the block of data into multiple chunks (e.g., four chunks) and distribute the chunks to at least two concurrent threads (e.g., a first thread and a second thread) , and pipeline intra-node and inter-node sub-operations for execution by at least two concurrent threads.

For example, the first thread may perform an inter-node global aggregation sub-operation on the first data block, while the second thread may perform an intra-node global aggregation sub-operation on the second data block. In addition, the first thread may perform a data replication sub-operation on the third data block, while the second thread may perform an inter-node global aggregation sub-operation on the fourth data block.

Example congestion avoidance methods

In an implementation, data or traffic congestion may occur at some switches or links in the communication network 206 due to data transmission between multiple computing nodes in the distributed training system 102 . To avoid congestion, the distributed training system 102 may employ predetermined congestion avoidance strategies to distribute or transfer data traffic among various switches or links in the communication network 206 to avoid excessive data flow during training (e.g., between nodes). The global reduction sub-operation or stage, or the global aggregation sub-operation or stage between nodes) passes through a certain switch or link in the communication network 206.

In implementation, the distributed training system 102 may employ a first congestion avoidance method that includes a ring generation strategy, and subsequent routing management of network flows. Additionally or alternatively, the distributed training system 102 may employ a second congestion avoidance method that includes a strategy for reordering node identities and subsequent routing management of network flows. Depending on the type of network topology of the communication network 206 and the processing and communication capabilities of the plurality of computing nodes 104, etc., the distributed training system 102 may select one or more of the first congestion avoidance method or the second congestion avoidance method for Data flows are routed between all or portions of the plurality of computing nodes in the distributed training system 102. In addition, the distributed training system 102 can selectively combine parts of the first congestion avoidance method and the second congestion avoidance method to implement a new congestion avoidance method. In an implementation, both the first congestion avoidance method and the second congestion avoidance method may be designed to designate dedicated network paths for each direction of inter-node data flow in such a manner that the inter-node data flows have little or no conflict with each other.

In implementation, the distributed training system 102 may obtain or establish a mapping relationship between a communication connection and a routing path (e.g., a physical link) in advance. In implementation, a connection path data structure in the form of a table, a linked list, etc. may be created and used to store information about the mapping relationship. In implementation, the distributed training system 102 may selectively or strategically use a specific path to establish a connection between any two computing nodes based on the connection path data structure.

In an implementation, the distributed training system 102 may determine communication connections and routing paths by enabling each computing node of the distributed training system 102 to send probe packets to other compute nodes through the source/destination ports of the change probe packets. The mapping relationship between them is to exhaust the possible communication connections between the computing nodes of the distributed training system 102. Obviously, the distributed training system 102 can use other methods to explore the mapping relationship between communication connections and routing paths, which are not limited here.

By way of example, and not limitation, the first computing node may send a plurality of probe packets to the second computing node, each probe packet having a different combination of source port and destination port, and the source address and destination address are respectively the first computing node and the address of the second computing node. Each probe packet may record the switches through which the corresponding probe packet passes, so when the corresponding probe packet returns to the first computing node, the first computing node may know the entire routing path of the corresponding probe packet for mapping. Accordingly, a connection path data structure (eg, a connection path data structure) may be established between the first computing node and the second computing node. Similarly, mapping relationships (and connection path data structures) between communication connections and routing paths of other pairs of computing nodes in the distributed training system 102 can be established accordingly.

For purposes of simplicity and illustration, an exemplary network topology, namely a fat tree network (or specifically a two-layer Clos network architecture in a full mesh topology) is used herein as the communication network 206 associated with the distributed training system 102 Example network topology. However, the exemplary congestion avoidance strategies described here may be applicable to other network topologies as well.

Figure 12 illustrates an exemplary fat tree network topology 1200. In this example, the exemplary fat-tree network topology is a two-layer Clos network architecture in a full mesh topology. One layer corresponds to a layer of slice switches 1202 directly connected to compute nodes 1204, with each slice switch 1202 connected to one or more compute nodes 1204. In an implementation, compute node 1204 may include one or more network interface controllers (eg, four network interface controllers) connected to one or more ports (eg, four ports) of slice switch 1202 . In an implementation, the number of network interface controllers for each compute node 1204 may be the same or different. Another layer corresponds to a layer of aggregation switches 1206 (or spine switches 1206) connected to one or more shard switches 1202.

In implementation, if two processing units or processes included in different computing nodes are connected under the same switch, the data packets transmitted between the two processing units or processes will pass through the same switch instead of Any aggregation switch. Alternatively, if two processing units or processes included in different computing nodes are connected under different slice switches, data packets transmitted between the two processing units or processes will pass through one of the aggregation switches. Using the connection path data structure as described above, packets traveling between two processing units or processes can be caused to flow through a designated aggregation switch by setting the appropriate combination of source and destination ports in the packet. In an implementation, the route management of the first congestion avoidance method and/or the second congestion avoidance method may be designed to enable data flows from the same slice switch to different destination slice switches to pass through different aggregation switches, and/or enable data flows from different destination slice switches to pass through different aggregation switches. Data flows from the source switch to the same destination switch can pass through different aggregation switches, thereby avoiding conflicts between data flows and eliminating network congestion at the aggregation switches.

In an implementation, as described in the previous description, the first congestion avoidance method may include a ring generation strategy, and subsequent routing management of network flows. The first congestion avoidance method can support various ring-based algorithms, including but not limited to ring algorithms, ring blocking algorithms, multi-ring algorithms, hierarchical ring algorithms, algorithms involving multiple hierarchical rings, node-aware ring algorithms, etc.

In an implementation, the ring-generated policy may include a ring-generated topology-aware policy. As an example and limitation, the ring-generated topology-aware policy may include multiple rules to establish a ring or ring configuration of a processing unit or process. In an implementation, a processing unit or process in a computing node may send/receive data to/from a processing unit or process in another computing node via a network interface controller. In an implementation, a processing unit or process in a computing node may be associated with a single network interface controller or multiple network interface controllers to transmit data to processing units or processes in other computing nodes. Additionally or alternatively, multiple processing units or processes may be associated with a single network interface controller and use the network interface controller to transmit data to processing units or processes in other computing nodes.

In an implementation, the plurality of rules may include, but are not limited to, a priority for a processing unit or process in the first computing node to select an adjacent processing unit or process, a condition for the network interface controller in the first computing node to send or receive data, Conditions for the network interface controller in the first computing node to route data to/from the network interface controller in the second computing node, and the like.

In an implementation, selecting the priority of an adjacent processing unit or process by a processing unit or process in a first computing node may include, in descending order of priority, selecting a processing unit or process in the first computing node and using inter-process communication ( If applicable), select a processing unit or process in the second compute node that is connected to the same slice switch as the first compute node to which the first compute node is connected. A processing unit or process is selected in a third computing node of the switch, where the first computing node is different from the second computing node and the third computing node.

In an implementation, the conditions under which a network interface controller in a first computing node sends or receives data may include, for example, the network interface controller is able to send data only to a network interface controller in a second computing node, and/or the network interface controller is able to receive data only from a network interface controller in a third computing node, wherein the first computing node is different from the second computing node and the third computing node, and the second computing node may be the same as or different from the third computing node.

In an implementation, the conditions under which the network interface controller in the first computing node routes data to/from the network interface controller in the second computing node may include, for example, if the data is routed through the network interface controller in the first computing node data sent by processing units or processes belonging to multiple rings is routed to the network interface controller in the second computing node. In an implementation, the conditions for the network interface controller in the first computing node to route data to/from the network interface controller in the second computing node may further include if the data is passed by a processing unit or process belonging to multiple rings through the first computing node. The data sent by the network interface controller in the second computing node is received by the network interface controller in the first computing node.

In implementation, the routing management of the first congestion avoidance method may assign a network interface controller (NIC) identifier to each network interface controller connected or linked to the same switch. The routing management of the first congestion avoidance method may also assign an aggregation identifier to each aggregation switch in the communication network 206. For a processing unit or process in a certain ring, the routing management may determine a routing identifier for routing a data packet from the processing unit or process.

For example, if the network interface controller of a processing unit or process and the network interface controller of the next processing unit or process in the ring are in the same compute node, or are directly connected or linked to the same slice switch, the route identifier can Is determined to be the default value or default identifier. This default route identifier indicates that the data is routed either within the compute node or through the slice switch, and not through any aggregation switch in the communications network. Otherwise, the route identifier may be determined to be equal to the NIC identifier of the processing unit or process or other predefined value. Based on the mapping relationship between the routing identifier and the aggregation identifier, the aggregation identifier may be determined based on the determined routing identifier. In an implementation, for example, the mapping relationship between routing identifiers and rendezvous identifiers may be predetermined using a probe-based routing mechanism (eg, sending probe packets between computing nodes as described in the preceding description).

In other words, data flows between processing units (or processes) included in the same computing node or network interface controllers of the same switch will not pass through any aggregation switch in the communication network. On the other hand, the data flow between the processing units (or processes) included in different computing nodes and the network interface controllers of different chip switches will pass through the designated aggregation switch based on the predetermined mapping relationship, thereby achieving routing control and control of the data flow. Manage and distribute data flows to different aggregation switches to avoid network congestion.

Figure 13 shows an example scenario using the first congestion avoidance method. In this example, four inter-node rings (or ring configurations, R0, R1, R2, and R3) containing eight compute nodes (Node 0, Node 1, ..., Node 7) are generated, and each ring uses a different aggregation switch to send and receive data (e.g., during the inter-node global reduction phase 504). Therefore, there is no conflict between these four rings. In addition, only one data flow enters and one data flow leaves each chip switch of any ring, thus avoiding the occurrence of network congestion.

In implementation, as mentioned above, the second congestion avoidance method may include a strategy regarding the reordering of node identities, and subsequent routing management of network flows. In an implementation, in order to minimize communication costs, the second congestion avoidance method may reorder identifiers of computing nodes and processing units (or processes) according to a network topology connecting computing nodes and processing units (or processes) based on multiple rules.

In an implementation, the plurality of rules may include, for example, grouping compute nodes by each slice switch. For example, compute nodes connected to the same switch (eg, compute nodes having network interface controllers linked to the same switch) are grouped into a group, and each compute node is assigned a node identifier. Since the compute nodes are connected to the same switch, these compute nodes are (physically) adjacent to each other.

In an implementation, the plurality of rules may also include assigning a sort identifier (or sort number) to each processing unit or process in the compute node using the same sequential sequence. For example, k processing units (or processes) in a first computing node may be assigned ranking identifiers 0, 1, ..., k-1, while k processing units (or processes) in a second computing node may be assigned Sorting identifiers k, k+1, ..., 2k-1, etc. are assigned, and the same is true for other computing nodes. Processing units (or processes) in a compute node can be ordered according to the corresponding network interface controller used by the processing unit (or process), and processing units (or processes) using the same network interface controller are (physically) connected to each other. adjacent.

In this case, in the first log ₂ L steps, the data flow between the processing units (or processes) in the computing nodes can be restricted to pass through the corresponding slice switches, which have better latency than the aggregation switches, and thus will not cause network congestion. In implementation, L is the number of computing nodes per slice switch for the node-aware halving and doubling algorithm as described above. In implementation, for the traditional halving and doubling algorithm, L is the product of the number of computing nodes per slice switch and the number of processing units (or processes) per computing node.

In an implementation, routing management of the second congestion avoidance method may include determining whether to send data from a first processing unit (or process) with a first ordering identifier in a first computing node with a first node identifier to a node with a second node identifier. An aggregate identifier of a data stream or packet of a second processing unit (or process) having a second ordering identifier in a second computing node of the symbol, where the first computing node may be the same as or different from the second computing node.

In an implementation, the aggregation identifier may be determined based at least in part on at least some of the ordering identifier, the node identifier, the number of network interface controllers per compute node, and the maximum number of compute nodes at each switch slice. By way of example, and not limitation, the aggregation identifier may be determined as a first ordering identifier of a first processing unit (or process) sending a data stream or packet + a first computing node having the first processing unit (or process) The first node identifier) % the maximum number of compute nodes at each slice switch) × the number of network interface controllers per compute node, where % represents the modulo operator. Obviously, other calculation methods for aggregate identifiers are also suitable as long as consistent results are obtained. For example, the aggregation identifier may be determined based on a preset mapping relationship between the aggregation identifier and a combination of the sorting identifier and the node identifier, or the like.

In an implementation, route management of the second congestion avoidance method may include pre-assigning aggregation identifiers to each aggregation switch in the communication network 206 associated with the distributed training system 102 . If the first processing unit (or process) and the second processing unit (or process) are linked to or under the same switch (e.g., through respective network controllers), then the data flow or packet will pass through the switch switch without going through any aggregation switch in the communications network 206. If the first processing unit (or process) and the second processing unit (or process) are not linked to the same switch or are not under the same switch, the first processing unit (or process) sends the message to the second processing unit (or process). ) will pass through the aggregation switch with the determined aggregation identifier. In this example, the described number of network interface controllers included in each computing node is four.

Figure 14 shows an example scenario using the second congestion avoidance method. In this example, all computing nodes include the same number of processing units (or processes) and the same number of network interface controllers, each network interface controller having the same number of processing units (or processes) to be associated. Additionally, the number of network interface controllers linked to slice switches is less than the number of aggregation switches in the network. In this example, the number of network interface controllers per compute node is 4, and the maximum number of compute nodes at each slice switch is 2. In implementation, for the node-aware halves and doubling algorithm, the number of computing nodes under the same switch can be a power of 2, and the number of network interface controllers included in the computing nodes under the same switch can be a power of 2. , for the traditional halve-double algorithm, the number of processing units (or processes) using the same network interface controller can be a power of 2. In this example, the described number of network interface controllers included in each computing node is four.

In this example, in the node-to-node global reduction phase in the node-aware halving and doubling algorithm, the processing units (or processes) of the computing nodes (node 0, node 2, node 4, and node 6) will use aggregation switches with aggregation identifiers (e.g., A1, A2, A3, and A4), and the processing units (or processes) of the computing nodes (node 1, node 3, node 5, and node 7) will use aggregation switches with aggregation identifiers (e.g., A5, A6, A7, and A8). Accordingly, there is no conflict in the data flow between the computing nodes, thereby avoiding network congestion at any aggregation switch in the network.

In implementation, at each step of the inter-node global reduction phase, a processing unit (or process) may communicate data with a new processing unit (or process). In implementation, synchronization may be performed to ensure that the data flow performed by a processing unit (or process) using a network interface controller at the current step does not overlap with the data flow performed by an adjacent processing unit (or process) using the same network interface controller at the previous step, so as to avoid microbursts (incast) and thus avoid microburst congestion.

Example methods

Figure 15 shows a schematic diagram of an exemplary topology-aware multi-stage method. Figure 16 shows a schematic diagram of a first exemplary network congestion avoidance method. Figure 17 shows a schematic diagram of a second exemplary network congestion avoidance method. Figure 18 shows a schematic diagram of an exemplary parallel method based on hybrid architecture in distributed training. The methods of Figures 15-18 may (but are not necessarily) implemented in the environment shown in Figure 1 using the computing nodes shown in Figure 2 with the aid of the methods and scenarios shown in Figures 3-14. For ease of description, methods 1500-1800 are described with reference to Figures 1-14. However, methods 1500-1800 may alternatively be implemented in other environments and/or utilizing other systems.

Methods 1500-1800 are described in the general context of machine-executable instructions. Generally, machine-executable instructions may include routines, programs, objects, components, data structures, procedures, modules, functions, etc., that perform specific functions or implement specific abstract data types. Additionally, each example method is represented as a collection of boxes in a logic flow diagram that represents a sequence of operations that may be implemented in hardware, software, firmware, or a combination thereof. The order in which the method is described is not intended to be construed as limiting, and any number of the described method blocks may be combined in any order to implement the method or alternative methods. Additionally, individual blocks may be omitted from the method without departing from the spirit and scope of the subject matter described herein. In the context of software, blocks represent computer instructions that perform the operations when executed by one or more processors. In the context of hardware, some or all of the blocks may represent application specific integrated circuits (ASICs) or other physical components that perform the operations described.

Referring to Figure 15, in block 1502, a first computing node (eg, computing node 104) may perform a reduction spread sub-operation among a first set of processing units in the first computing node according to a first cluster communication algorithm.

In an implementation, prior to performing the reduce spread sub-operation, the first computing node may select the first cluster communication algorithm based at least in part on the type or bandwidth of intra-node connections between the first set of processing units in the first computing node . In an implementation, the first cluster communication algorithm may include, but is not limited to, a ring-based algorithm or a halve-double algorithm.

In an implementation, performing the reduction spread sub-operation among the first set of processing units in the first computing node according to the first cluster communication algorithm may include dividing the data into a plurality of data blocks; assigning the plurality of data blocks to the first a set of processing units; receiving a data block at a first processing unit of the first set of processing units from a second processing unit of the first set of processing units according to a first cluster communication algorithm; and homing at the first processing unit with the local data block About the data block received.

In block 1504, the first computing node may perform a global reduction sub-operation between the first set of processing units in the first computing node and the second set of processing units in the second computing node according to the second cluster communication algorithm.

In an implementation, before performing the global reduce sub-operation, the first compute node may be based at least in part on the type or bandwidth of the inter-node connection between the first compute node and other compute nodes, and/or the first compute node and other compute nodes. The connection topology of the nodes is calculated to select the second cluster communication algorithm. In implementation, the first cluster communication algorithm may include but is not limited to a ring-based algorithm, or a halving and doubling algorithm (such as a node-aware halving and doubling algorithm), etc.

In an implementation, performing the global reduction sub-operation between the first set of processing units in the first computing node and the second set of processing units in the second computing node according to the second cluster algorithm may include: the first set of processing units receiving a second set of processing units in the second computing node dispersing portions of the reduction results obtained by the second clustering algorithm, each processing unit of the first set of processing units forming a group with a corresponding processing unit of the second set of processing units, and receiving corresponding portions of the reduction spread results from corresponding processing units; the first set of processing units pairs the reduction spread results by corresponding local portions of the reduction spread results obtained after performing the reduction spread sub-operations among the first set of processing units Each part performs the reduction.

In block 1506 , the first computing node may perform a global aggregation sub-operation among a first set of processing units in the first computing node according to a first cluster communication algorithm.

In an implementation, performing the global aggregation sub-operation between the first set of processing units in the first computing node according to the first cluster communication algorithm may include: according to the first cluster communication algorithm, between the first processing unit of the first set of processing units receiving a data block from a second processing unit of the first set of processing units; and reducing the received data block with a local data block at the first processing unit.

Referring to FIG. 16 , in block 1602 , a first computing node (eg, computing node 104 ) or a first process may be based, at least in part, on a network interface controller associated with the first process and a network interface associated with the second process. Whether the controller is located in the same compute node or linked to the same switch determines the routing identifier for routing data from the first process to the second process.

In an implementation, the first process and the second process may belong to a specific inter-node ring connecting multiple different nodes under a specific network topology. By way of example, and not limitation, certain network topologies may include fat tree topologies.

In an implementation, the network interface controller associated with the first process is configured to only send data to or receive data from a second computing node in the ring topology, the second computing node being different than the first computing node.

In an implementation, the network interface controller associated with the first process is also associated with one or more processes, wherein all data sent from the first process and the one or more processes is sent through the network interface controller.

In implementations, in response to determining that the network interface controller associated with the first process and the network interface controller associated with the second process are located in the same computing node or linked to the same blade switch, the routing identifier may be set or determined as a default identifier.

In an implementation, in response to determining that the network interface controller associated with the first process and the network interface controller associated with the second process are located in different compute nodes or linked to different slice switches, the routing identifier may be set or determined to be equal to the identifier of the network interface controller associated with the first process.

In block 1604, the first computing node or the first process may route data from the first process to the second process according to the routing identifier.

In an implementation, routing data from the first process to the second process based on the routing identifier may include at least passing through a slice switch connected to the network interface controller and an aggregation having an identifier that has a corresponding relationship with the identifier of the network interface controller. The switch routes data from the first process to the second process, and the network interface controller is associated with the first process.

17 , in box 1702, a first computing node (e.g., computing node 104) or a first process may determine an aggregation identifier for sending a data packet from the first process to a second process based on a node-aware halving and doubling algorithm, wherein the first process and the second process belong to different nodes connected to different slice switches under a specific network topology.

In an implementation, the first computing node may assign different aggregation identifiers to data packets directed to computing nodes connected to different slice switches, so that the data packets can be routed to nodes connected to different slice switches through different aggregation switches.

In an implementation, the first computing node may allocate the source port and the destination port corresponding to the aggregation switch associated with the aggregation identifier based at least in part on the predetermined correspondence. In an implementation, the correspondence relationship may record the relationship between aggregation identifiers of multiple aggregation switches and corresponding source port and destination port pairs. In implementations, the specific network topology may include a fat tree topology.

In block 1704, the first computing node may send the data packet from the first process to the second process through the aggregation switch corresponding to the aggregation identifier.

In an implementation, the first computing node may also transfer data from the first process set included in the first computing node to the process set included in the second computing node through multiple different aggregation switches corresponding to multiple different aggregation identifiers assigned to each data packet. The second set of processes sends each data packet.

In an implementation, the first computing node may also use multiple different aggregation switches corresponding to multiple different aggregation identifiers assigned to each data packet. A second set of processes receives the packet.

Referring to Figure 18, in block 1802, a first computing node (eg, computing node 104) or processing unit may divide a data block allocated to the processing unit into a plurality of data segments, the plurality of data segments including at least first data segment and the second data segment.

In block 1804, the first computing node or processing unit may allocate the plurality of data segments to the plurality of threads, including at least a first thread and a second thread.

In block 1806, the first compute node or processing unit may perform an intra-node sub-operation on a portion of the first data segment using a first thread, and in parallel use a second thread to perform an inter-node sub-operation on a portion of the second data segment.

In an implementation, performing an intra-node sub-operation on a portion of the first data segment using the first thread may include transmitting the portion of the first data segment between a processing unit included in the first computing node and another processing unit via an intra-node connection. .

In an implementation, performing an inter-node sub-operation on a portion of the second data segment using the second thread may include an inter-node connection between the processing unit and another processing unit included in a second computing node that is different from the first computing node. Transmit a portion of the second data segment described above.

In an implementation, the intra-node sub-operation may include a reduce spread sub-operation or a global aggregation sub-operation performed within a first compute node, and the inter-node sub-operation may include a first compute node and a second compute node different from the first compute node. Compute global reduction sub-operations performed across nodes.

In an implementation, the intra-node sub-operation may include a global aggregation sub-operation or a replication sub-operation executed in a first computing node, and the inter-node sub-operation may include a global aggregation sub-operation executed between the first computing node and a second computing node different from the first computing node.

In an implementation, the first computing node or processing unit may use a first thread to perform another inter-node sub-operation on a portion of the first data segment, and in parallel use a second thread to perform another inter-node sub-operation on a portion of the second data segment. Suboperations within a node.

In an implementation, a first thread is used to perform an intra-node sub-operation on the portion of the first data segment, and a second thread is used in parallel to perform an inter-node sub-operation on the portion of the second data segment, such that the first is combined using an intra-node join. transmitting a portion of the data segment to another processing unit included in the first computing node, and concurrently using an inter-node connection to transmit a portion of the second data segment to another processing unit included in a second computing node different from the first computing node .

Although the method blocks described above are executed in a certain order, in some implementations, some or all of the method blocks may be executed in another order or in parallel.

Summarize

Although embodiments have been described in language specific to structural features and/or methodological acts, it is to be understood that the claims are not necessarily limited to the specific features or acts described. Rather, the specific features and acts are disclosed as example forms of implementing the claimed subject matter. Additionally or alternatively, some or all operations may be implemented by one or more ASICS, FPGAs, or other hardware.

This disclosure may further be understood using the following terms:

Clause 1: A method implemented by a first computing node, the method comprising: performing a reduction spread sub-operation between a first set of processing units in the first computing node according to a first cluster communication algorithm; according to a second cluster The communication algorithm performs a global reduction sub-operation between the first set of processing units in the first computing node and the second set of processing units in the second computing node; and performs a global reduction sub-operation on the first set of processing units in the first computing node according to the first cluster communication algorithm. Global aggregation sub-operations are performed between a collection of processing units.

Clause 2: The method of Clause 1, further comprising selecting the first cluster communication algorithm based at least in part on a type or bandwidth of an intra-node connection between the first set of processing units in the first computing node.

Clause 3: The method of Clause 1, further comprising: based at least in part on the type or bandwidth of the inter-node connection between the first computing node and the other computing nodes, and/or the connection of the first computing node and the other computing nodes. topology to select the second cluster communication algorithm.

Clause 4: The method of Clause 1, wherein the first cluster communication algorithm includes a ring-based algorithm, or a halve-double algorithm.

Clause 5: The method of clause 1, wherein performing the reduction spread sub-operation among the first set of processing units in the first computing node according to the first cluster communication algorithm includes: dividing the data into a plurality of data blocks; allocating a plurality of data blocks to the first set of processing units; receiving the data blocks at a first processing unit of the first set of processing units from a second processing unit of the first set of processing units according to a first cluster communication algorithm; and at the first The received data blocks are reduced using local data blocks at the processing unit.

Clause 6: The method of Clause 1, wherein performing the global reduction sub-operation between the first set of processing units in the first computing node and the second set of processing units in the second computing node according to the second clustering algorithm includes : The first set of processing units receives portions of the reduction dispersion results obtained by the second set of processing units in the second computing node according to the second cluster algorithm, wherein each processing unit of the first set of processing units is identical to the second set of processing units. The corresponding processing units of the set form a group and receive corresponding portions of the reduction spread results from the corresponding processing units; the first set of processing units obtains the reduction spread results obtained by performing the reduction spread sub-operations among the first set of processing units. The reduction is performed on each part of the reduction spread result corresponding to the local part.

Clause 7: The method of Clause 1, wherein performing the global aggregation sub-operation between the first set of processing units in the first computing node according to the first cluster communication algorithm comprises: according to the first cluster communication algorithm, on the first processing unit A data block is received at a first processing unit of the set of units from a second processing unit of the first set of processing units; and the received data block is reduced with a local data block at the first processing unit.

Item 8: One or more machine-readable media storing machine-readable instructions that, when executed by a first computing node, cause the first computing node to perform actions, the actions comprising: performing a reduce-scatter sub-operation between a first set of processing units in the first computing node according to a first cluster communication algorithm; performing a global reduce sub-operation between a first set of processing units in the first computing node and a second set of processing units in a second computing node according to a second cluster communication algorithm; and performing a global gather sub-operation between a first set of processing units in the first computing node according to the first cluster communication algorithm.

Clause 9: The one or more machine-readable media of Clause 8, the actions further comprising: selecting the first computing node based at least in part on a type or bandwidth of an intra-node connection between the first set of processing units in the first computing node. Cluster communication algorithm.

Clause 10: According to one or more machine-readable media of clause 8, the action further comprises: selecting a second cluster communication algorithm based at least in part on a type or bandwidth of an inter-node connection between the first computing node and the other computing nodes, and/or a connection topology between the first computing node and the other computing nodes.

Clause 11: The one or more machine-readable media of Clause 8, wherein the first cluster communication algorithm includes a ring-based algorithm, or a halve-double algorithm.

Clause 12: The one or more machine-readable media of Clause 8, wherein performing the reduction spread sub-operation among the first set of processing units in the first computing node according to the first cluster communication algorithm includes: partitioning the data into a plurality of data blocks; allocating the plurality of data blocks to a first set of processing units; receiving at a first processing unit of the first set of processing units from a second processing unit of the first set of processing units according to a first cluster communication algorithm data blocks; and reducing the received data blocks with local data blocks at the first processing unit.

Clause 13: One or more machine-readable media as described in clause 8, wherein between a first set of processing units in a first computing node and a second set of processing units in a second computing node according to a second clustering algorithm Executing the global reduction sub-operation includes: the first set of processing units receiving each part of the reduction spread result obtained by the second set of processing units in the second computing node according to the second cluster algorithm, and each processing of the first set of processing units The units form groups with corresponding processing units of the second set of processing units and receive corresponding portions of the reduction spread results from the corresponding processing units; the first set of processing units is obtained by performing the reduction spread sub-operations among the first set of processing units The corresponding local part of the reduce spread result performs the reduction on each part of the reduce spread result.

Clause 14: The one or more of clause 8 and the readable medium thereof, wherein performing the global aggregation sub-operation between the first set of processing units in the first computing node according to the first cluster communication algorithm includes: according to the first A cluster communication algorithm for receiving a data block at a first processing unit of the first set of processing units from a second processing unit of the first set of processing units; and reducing the received data block with a local data block at the first processing unit.

Clause 15: A first computing node, comprising: a first set of processing units; and a memory storing machine-executable instructions that, when executed by the first set of processing units, cause the first set of processing units to execute Actions, the actions comprising: performing a reduction spread sub-operation between a first set of processing units in a first computing node according to a first cluster communication algorithm; performing first processing in the first computing node according to a second cluster communication algorithm A global reduction sub-operation is performed between the unit set and the second processing unit set in the second computing node; and a global aggregation sub-operation is performed between the first processing unit set in the first computing node according to the first cluster communication algorithm.

Clause 16: The first computing node of clause 15, the actions further comprising: selecting the first cluster communication based at least in part on a type or bandwidth of an intra-node connection between the first set of processing units of the first computing node; The second cluster communication algorithm is selected based on the type or bandwidth of the inter-node connection between the first computing node and the other computing nodes, and/or, the topology of the connection between the first computing node and the other computing nodes.

Clause 17: The first computing node of clause 15, wherein the first cluster communication algorithm comprises a ring-based algorithm, or a halving-doubling algorithm.

Clause 18: The first computing node of Clause 15, wherein performing the reduction spread sub-operation among the first set of processing units in the first computing node according to the first cluster communication algorithm includes: partitioning the data into a plurality of data blocks; allocating the plurality of data blocks to the first set of processing units; receiving the data blocks at a first processing unit of the first set of processing units from a second processing unit of the first set of processing units according to a first cluster communication algorithm; and The received data blocks are reduced with local data blocks at the first processing unit.

Clause 19: A first computing node according to clause 15, wherein a global reduction is performed between a first set of processing units in the first computing node and a second set of processing units in the second computing node according to a second clustering algorithm The sub-operations include: the first set of processing units receiving each part of the reduction spread result obtained by the second set of processing units in the second computing node according to the second cluster algorithm, each processing unit of the first set of processing units communicating with the second set of processing units. Corresponding processing units of the set of processing units form a group and receive corresponding parts of the reduction spread results from the corresponding processing units; the second set of processing units passes the reduction spread obtained after performing the reduction spread sub-operations among the first set of processing units The corresponding local part of the result performs the reduction on each part of the reduction spread result.

Clause 20: The first computing node of clause 15, wherein performing the global aggregation sub-operation between the first set of processing units in the first computing node according to the first cluster communication algorithm includes: according to the first cluster communication algorithm, in A data block is received at a first processing unit of the first set of processing units from a second processing unit of the first set of processing units; and the received data block is reduced with a local data block at the first processing unit.

Clause 21: A method implemented by a first computing node, the method comprising: based, at least in part, on whether a network interface controller associated with the first process and a network interface controller associated with the second process are located on the same compute node. in a node or linked to the same switch to determine a routing identifier for routing data from a first process to a second process, the first process and the second process belonging to a specific inter-node ring connecting multiple different nodes under a specific network topology; and Route data from the first process to the second process based on the routing identifier.

Clause 22: The method of clause 21, wherein the network interface controller associated with the first process is configured to send data to or receive data from only a second computing node in the ring topology, the second computing node being different from the first computing node.

Clause 23: The method of clause 21, wherein the network interface controller associated with the first process is also associated with one or more processes, and wherein all data sent from the first process and the one or more processes passes through Sent by the network interface controller.

Clause 24: The method of clause 21, wherein the specific network topology includes a fat tree topology.

Clause 25: The method of Clause 21, further comprising: in response to determining that the network interface controller associated with the first process and the network interface controller associated with the second process are located in the same computing node or linked to the same slice. switch, sets the routing identifier to the default identifier.

Clause 26: The method according to Clause 21 further includes: in response to determining that the network interface controller associated with the first process and the network interface controller associated with the second process are located on different computing nodes or linked to different slice switches, setting the routing identifier to be equal to the identifier of the network interface controller associated with the first process.

Clause 27: The method of Clause 26, wherein routing data from the first process to the second process based on the routing identifier includes: at least through a slice switch connected to the network interface controller and having an identifier with the network interface controller An aggregation switch having a corresponding identifier routes data from a first process to a second process, the network interface controller being associated with the first process.

Clause 28: One or more machine-readable media having machine-readable instructions stored thereon that, when executed by a first computing node, cause the first computing node to perform an action, the action comprising: based at least in part on: Whether the network interface controller associated with the first process and the network interface controller associated with the second process are located in the same compute node or linked to the same switch to determine the route for routing data from the first process to the second process The identifier, the first process and the second process belong to a specific inter-node ring connecting multiple different nodes under a specific network topology; and routing data from the first process to the second process according to the routing identifier.

Clause 29: One or more machine-readable media as described in clause 28, wherein the network interface controller associated with the first process is configured to send data to or receive data from only a second computing node in the ring topology, the second computing node being different from the first computing node.

Clause 30: The one or more machine-readable media of Clause 28, wherein the network interface controller associated with the first process is also associated with one or more processes, wherein from the first process and the one or more All data sent by the process is sent through the network interface controller.

Clause 31: The one or more machine-readable media of Clause 28, wherein the particular network topology includes a fat tree topology.

Clause 32: The one or more machine-readable media of Clause 28, wherein the actions further comprise: in response to determining a network interface controller associated with the first process and a network interface control associated with the second process If the router is located in the same compute node or linked to the same switch, set the routing identifier to the default identifier.

Clause 33: One or more machine-readable media according to clause 28, wherein the action further comprises: in response to determining that the network interface controller associated with the first process and the network interface controller associated with the second process are located on different computing nodes or linked to different slice switches, setting the routing identifier to be equal to the identifier of the network interface controller associated with the first process.

Clause 34: The one or more machine-readable media of clause 33, wherein routing data from the first process to the second process based on the routing identifier includes: at least through a slice switch connected to the network interface controller and having An aggregation switch having an identifier of a corresponding relationship routes data from a first process to a second process, the network interface controller being associated with the first process.

Clause 35: A first computing node comprising: one or more processing units; and a memory storing machine-executable instructions that, when executed by the one or more processing units, cause one or more processes The unit performs actions that include determining based at least in part on whether a network interface controller associated with the first process and a network interface controller associated with the second process are located in the same compute node or linked to the same switch. Routing data from a first process to a routing identifier of a second process, the first process and the second process belonging to a specific inter-node ring connecting multiple different nodes under a specific network topology; and routing data from the first process according to the routing identifier Route to the second process.

Clause 36: The first computing node of Clause 35, wherein the network interface controller associated with the first process is configured to send data only to or receive data from the second computing node in the ring topology, The second computing node is different from the first computing node.

Clause 37: The first computing node of Clause 35, wherein the network interface controller associated with the first process is also associated with one or more processes, and wherein all Data is sent through the network interface controller.

Clause 38: The first computing node of Clause 35, wherein the actions further comprise: in response to determining that the network interface controller associated with the first process and the network interface controller associated with the second process are located on the same compute node. In the node or connected to the same switch, set the routing identifier as the default identifier.

Clause 39: The first computing node of Clause 35, wherein the actions further comprise: in response to determining that a network interface controller associated with the first process and a network interface controller associated with the second process are located on different The compute node or link to a different slice switch sets the route identifier equal to the identifier of the network interface controller associated with the first process.

Clause 40: The first compute node of Clause 39, wherein routing data from the first process to the second process based on the routing identifier includes: at least through a slice switch connected to the network interface controller and having a connection to the network interface controller An aggregation switch having an identifier corresponding to an identifier routes data from a first process to a second process, the network interface controller being associated with the first process.

Clause 41: A method implemented by a first computing node, the method comprising:

Determining a rendezvous identifier for sending packets from a first process to a second process, the first process and the second process belonging to different nodes connected to different slice switches under a specific network topology according to a node-aware halving and doubling algorithm; and by The aggregation switch corresponding to the aggregation identifier sends the data packet from the first process to the second process.

Clause 42: The method of Clause 41, further comprising: assigning different aggregation identifiers to data packets directed to nodes connected to different slice switches to enable routing of data packets to nodes connected to different shard switches through the different aggregation switches. The node of the chip switch.

Clause 43: The method of Clause 41, further comprising allocating a source port and a destination port corresponding to the aggregation switch associated with the aggregation identifier based at least in part on the predetermined correspondence.

Clause 44: The method according to Clause 43, wherein the correspondence relationship records a relationship between aggregation identifiers of a plurality of aggregation switches and corresponding source port and destination port pairs.

Clause 45: The method of clause 41, wherein the particular network topology comprises a fat tree topology.

Clause 46: The method according to Clause 41 further includes: sending each data packet from a first set of processes included in a first computing node to a second set of processes included in a second computing node through multiple different aggregation switches corresponding to multiple different aggregation identifiers assigned to each data packet.

Clause 47: The method according to Clause 41 further includes: receiving data packets from a second set of processes included in a second computing node by a first set of processes included in a first computing node through multiple different aggregation switches corresponding to multiple different aggregation identifiers assigned to each data packet.

Item 48: One or more machine-readable media storing machine-readable instructions that, when executed by a first computing node, cause the first computing node to perform actions, the actions comprising: determining an aggregation identifier for sending a data packet from a first process to a second process based on a node-aware halving and doubling algorithm, the first process and the second process belonging to different nodes connected to different slice switches under a specific network topology; and sending a data packet from the first process to the second process through an aggregation switch corresponding to the aggregation identifier.

Clause 49: The one or more machine-readable media of Clause 48, wherein the actions further comprise assigning different rendezvous identifiers to packets directed to nodes connected to different slice switches to enable different rendezvous identifiers to be The aggregation switches route packets to nodes connected to different slice switches.

Clause 50: The one or more machine-readable media of clause 48, wherein the actions further comprise: allocating a source port and a destination port corresponding to an aggregation switch associated with the aggregation identifier based at least in part on a predetermined correspondence.

Clause 51: One or more machine-readable media according to clause 50, wherein the correspondence records the relationship between aggregation identifiers of a plurality of aggregation switches and corresponding source port and destination port pairs.

Clause 52: The one or more machine-readable media of Clause 48, wherein the particular network topology includes a fat-tree topology.

Clause 53: The one or more machine-readable media of Clause 48, wherein the actions further comprise: transmitting data from a first packet to a packet via a plurality of different aggregation switches corresponding to a plurality of different aggregation identifiers assigned to each data packet. A first set of processes included in the computing node sends each data packet to a second set of processes included in the second computing node.

Clause 54: The one or more machine-readable media of clause 48, wherein the actions further comprise: through a plurality of different aggregation switches corresponding to a plurality of different aggregation identifiers assigned to each data packet, by the first A first set of processes included in a computing node receives data packets from a second set of processes included in a second computing node.

Clause 55: A first computing node comprising: one or more processing units; and a memory storing machine-executable instructions that, when executed by the one or more processing units, cause one or more The processing unit performs actions, the actions including: determining a convergence identifier for sending a data packet from a first process to a second process according to a node-aware halving and doubling algorithm, the first process and the second process belonging to a connection under a specific network topology to different nodes on different slice switches; and sending the data packet from the first process to the second process through the aggregation switch corresponding to the aggregation identifier.

Clause 56: The first compute node of Clause 55, wherein the actions further comprise assigning different aggregation identifiers to packets directed to nodes connected to different shard switches to enable routing through the different aggregation switches. Packets are routed to nodes connected to different slice switches.

Clause 57: The first computing node of clause 55, wherein the actions further comprise allocating source and destination ports corresponding to the aggregation switch associated with the aggregation identifier based at least in part on the predetermined correspondence.

Clause 58: The first computing node of Clause 57, wherein the correspondence records a relationship between aggregation identifiers of a plurality of aggregation switches and corresponding source port and destination port pairs.

Clause 59: The first computing node of Clause 55, wherein the actions further comprise: transferring data from the first computing node to a plurality of different aggregation switches corresponding to a plurality of different aggregation identifiers assigned to each data packet. The first set of processes sends each data packet to a second set of processes included in the second computing node.

Clause 60: The first computing node of Clause 55, wherein the actions further comprise: through a plurality of different aggregation switches corresponding to a plurality of different aggregation identifiers assigned to each data packet. The first set of processes receives data packets from a second set of processes included in the second computing node.

Clause 61: A method implemented by a first computing node, the method comprising: dividing a data packet allocated to a processing unit into a plurality of data segments, the plurality of data segments including at least a first data segment and a second data segment; Allocating multiple data segments to multiple threads, the plurality of threads including at least a first thread and a second thread; using the first thread to perform an intra-node sub-operation on a part of the first data segment, and using the second thread in parallel to perform an intra-node sub-operation on a part of the first data segment. Part of the second data segment performs inter-node sub-operations.

Clause 62: The method of Clause 61, wherein performing an intra-node sub-operation on a portion of the first data segment using the first thread comprises: via an intra-node connection between a processing unit included in the first compute node and another processing unit Transmit this portion of the first data segment.

Clause 63: The method of clause 61, wherein using the second thread to perform an inter-node sub-operation on a portion of the second data segment comprises: in a processing unit and a second computing node different from the first computing node via an inter-node connection The portion of the second data segment is transmitted between another processing unit included.

Clause 64: The method of Clause 61, wherein the intra-node sub-operation includes a reduction spread sub-operation or a global aggregation sub-operation performed within a first compute node, and the inter-node sub-operation includes a first compute node and a different A global reduction sub-operation performed between the first computing node and the second computing node.

Clause 65: A method according to clause 61, wherein the intra-node sub-operation includes a global aggregation sub-operation or a replication sub-operation performed within a first computing node, and the inter-node sub-operation includes a global aggregation sub-operation performed between the first computing node and a second computing node different from the first computing node.

Clause 66: The method of Clause 66, further comprising: using the first thread to perform another inter-node sub-operation on a portion of the first data segment, and in parallel using the second thread to perform another on a portion of the second data segment. Suboperations within a node.

Clause 67: A method according to Clause 61, wherein an intra-node sub-operation is performed on the portion of the first data segment using a first thread, and an inter-node sub-operation is performed on the portion of the second data segment using a second thread in parallel, so that an intra-node connection is used to transfer a portion of the first data segment to another processing unit included in the first computing node, and an inter-node connection is used to concurrently transfer a portion of the second data segment to another processing unit included in a second computing node different from the first computing node.

Clause 68: One or more machine-readable media having machine-readable instructions stored thereon that, when executed by the first computing node, cause the first computing node to perform actions, the actions including: assigning to processing The data packet of the unit is divided into multiple data segments, the multiple data segments at least include a first data segment and a second data segment; the multiple data segments are allocated to multiple threads, the multiple threads include at least the first thread and the second data segment. Two threads: using the first thread to perform intra-node sub-operations on a portion of the first data segment, and using the second thread in parallel to perform inter-node sub-operations on a portion of the second data segment.

Clause 69: The one or more machine-readable media of Clause 68, wherein performing an intra-node sub-operation on a portion of the first data segment using the first thread includes another process included in the first compute node via an intra-node connection The portion of the first data segment is transferred between the unit and the processing unit.

Clause 70: One or more machine-readable media as described in clause 68, wherein performing an inter-node sub-operation on a portion of a second data segment using a second thread includes transmitting the portion of the second data segment between a processing unit and another processing unit included in a second computing node different from the first computing node via an inter-node connection.

Clause 71: One or more machine-readable media of Clause 68, wherein the intra-node sub-operation includes a reduce spread sub-operation or a global aggregation sub-operation performed within the first computing node, and the inter-node sub-operation includes a A global reduction sub-operation performed between a computing node and a second computing node different from the first computing node.

Clause 72: One or more machine-readable media according to clause 68, wherein the intra-node sub-operation includes a global aggregation sub-operation or a replication sub-operation performed within a first computing node, and the inter-node sub-operation includes a global aggregation sub-operation performed between the first computing node and a second computing node different from the first computing node.

Clause 73: The one or more machine-readable media of Clause 68, the actions further comprising: using the first thread to perform another inter-node sub-operation on the portion of the first data segment, and using the second thread in parallel Perform another in-node sub-operation on part of the second data segment.

Clause 74: One or more machine-readable media according to Clause 68, wherein a first thread is used to perform an intra-node sub-operation on a portion of the first data segment, and a second thread is used in parallel to perform an inter-node sub-operation on a portion of the second data segment. Sub-operations such that a portion of the first data segment is transmitted to another processing unit included in the first computing node using an intra-node connection, and concurrently transmitting a portion of the second data segment to another processing unit included in the first computing node using an inter-node connection. A different second computing node includes another processing unit.

Clause 75: A first computing node comprising: one or more processing units; and a memory storing machine-executable instructions that, when executed by the one or more processing units, cause one or more The processing unit performs actions, the actions include: dividing the data packet allocated to the processing unit into multiple data segments, the multiple data segments at least include a first data segment and a second data segment; allocating the multiple data segments to multiple data segments. Threads, the plurality of threads at least include a first thread and a second thread; using the first thread to perform intra-node sub-operations on a part of the first data segment, and using the second thread in parallel to perform inter-node sub-operations on a part of the second data segment suboperation.

Clause 76: The first compute node of clause 75, wherein performing the intra-node sub-operation on the portion of the first data segment using the first thread includes: another processing unit included in the first compute node and processing via an intra-node connection The portion of the first data segment is transmitted between units.

Clause 77: The first compute node of Clause 75, wherein performing the inter-node sub-operation on the portion of the second data segment using the second thread includes: The portion of the second data segment is transmitted between another processing unit included in the computing node.

Clause 78: The first compute node of Clause 75, wherein the intra-node sub-operations comprise a reduction spread sub-operation or a global aggregation sub-operation performed within the first compute node, and the inter-node sub-operations comprise the first compute node and A global reduction sub-operation performed between second computing nodes that are different from the first computing node.

Clause 79: The first compute node of clause 75, wherein the intra-node sub-operations comprise global aggregate sub-operations or replica sub-operations executed within the first compute node, and the inter-node sub-operations comprise the first compute node and are different from A global aggregation sub-operation performed between the first computing node and the second computing node.

Clause 80: The first compute node of Clause 75, wherein a first thread is used to perform intra-node sub-operations on the portion of the first data segment, and a second thread is used in parallel to perform inter-node sub-operations on the portion of the second data segment. Operate such that a portion of the first data segment is transmitted using an intra-node connection to another processing unit included in the first computing node, and concurrently using an inter-node connection to transmit a portion of the second data segment to a computing node different from the first computing node. The second computing node includes another processing unit.

Claims (15)

1. A method implemented by a first computing node, the method comprising: Determining to transfer the data from the first network interface controller associated with the first process to the second network interface controller associated with the second process is based at least in part on whether the first network interface controller associated with the second process is located in the same computing node or linked to the same switch. A route identifier for a first process to route to the second process, wherein in response to determining the first network interface controller associated with the first process and the third network interface controller associated with the second process Two network interface controllers are located in the same computing node or linked to the same switch, and set the routing identifier as a default identifier; in response to determining that the first network associated with the first process The interface controller and the second network interface controller associated with the second process are located on different computing nodes or linked to different slice switches, and the routing identifier is set equal to that associated with the first process The identifier of the first network interface controller connected; the first process and the second process belong to a specific inter-node ring connecting multiple different nodes under the network topology; and The data is routed from the first process to the second process based on the routing identifier. 2. The method of claim 1, wherein the first network interface controller associated with the first process is configured to send data to or from a second computing node in a ring topology. Two computing nodes receive data, the second computing node being different from the first computing node. 3. The method of claim 1, wherein the first network interface controller associated with the first process is also associated with one or more processes, wherein from the first process and the one Or the data sent by multiple processes is sent through the first network interface controller. 4. The method of claim 1, wherein the network topology includes a fat tree topology. 5. The method of claim 1, wherein routing the data from the first process to the second process based on the routing identifier includes at least one interface connected to the first network interface controller. A slice switch and an aggregation switch having an identifier corresponding to the identifier of the first network interface controller route the data from the first process to the second process, the first network An interface controller is associated with the first process. 6. One or more machine-readable media storing machine-readable instructions that, when executed by a first computing node, cause the first computing node to perform an action, the action including: Determining to transfer the data from the first network interface controller associated with the first process to the second network interface controller associated with the second process is based at least in part on whether the first network interface controller associated with the second process is located in the same computing node or linked to the same switch. A route identifier for a first process to route to the second process, wherein in response to determining the first network interface controller associated with the first process and the third network interface controller associated with the second process two network interface controllers located in the same computing node or linked to the same switch, setting the routing identifier as a default identifier; in response to determining that the first network associated with the first process The interface controller and the second network interface controller associated with the second process are located on different computing nodes or linked to different slice switches, and the routing identifier is set equal to that associated with the first process The identifier of the first network interface controller connected; the first process and the second process belong to a specific inter-node ring connecting multiple different nodes under the network topology; and The data is routed from the first process to the second process according to the routing identifier. 7. The one or more machine-readable media of claim 6, wherein the first network interface controller associated with the first process is configured to send data to a second computing node in a ring topology Or receiving data from a second computing node in a ring topology, the second computing node being different than the first computing node. 8. The one or more machine-readable media of claim 6, wherein the first network interface controller associated with the first process is further associated with one or more processes, wherein from the Data sent by the first process and the one or more processes are sent through the first network interface controller. 9. One or more machine-readable media according to claim 6, wherein the network topology includes a fat tree topology. 10. The one or more machine-readable media of claim 8, wherein routing the data from the first process to the second process based on the routing identifier includes at least communicating with the first process. A network interface controller connected slice switch and an aggregation switch having an identifier corresponding to the identifier of the first network interface controller route the data from the first process to the second process. A process, the first network interface controller is associated with the first process. 11. A first computing node, comprising: one or more processing units; and A memory that stores machine-executable instructions that, when executed by the one or more processing units, cause the one or more processing units to perform actions, the actions including: Determining to transfer the data from the first network interface controller associated with the first process to the second network interface controller associated with the second process is based at least in part on whether the first network interface controller associated with the second process is located in the same computing node or linked to the same switch. A route identifier for a first process to route to the second process, wherein in response to determining the first network interface controller associated with the first process and the third network interface controller associated with the second process Two network interface controllers are located in the same computing node or linked to the same switch, and set the routing identifier as a default identifier; in response to determining that the first network associated with the first process The interface controller and the second network interface controller associated with the second process are located on different computing nodes or linked to different slice switches, and the routing identifier is set equal to that associated with the first process The identifier of the first network interface controller connected; the first process and the second process belong to a specific inter-node ring connecting multiple different nodes under the network topology; and The data is routed from the first process to the second process based on the routing identifier. 12. The first computing node of claim 11, wherein the first network interface controller associated with the first process is configured to send data to or from a second computing node in a ring topology. A second computing node in , which is different from the first computing node, receives the data. 13. The first computing node of claim 11, wherein the first network interface controller associated with the first process is further associated with one or more processes, wherein from the first process and Data sent by the one or more processes is sent through the first network interface controller. 14. The first computing node of claim 11, wherein routing the data from the first process to the second process according to the routing identifier comprises: routing the data from the first process to the second process via at least a slice switch connected to the first network interface controller and an aggregation switch having an identifier that corresponds to the identifier of the first network interface controller, the first network interface controller being associated with the first process. 15. A first computing node, comprising: Determining means for determining based at least in part on whether a first network interface controller associated with the first process and a second network interface controller associated with the second process are located in the same computing node or linked to the same switch. A routing identifier for routing data from the first process to the second process, the first process and the second process belonging to a specific inter-node ring connecting a plurality of different nodes under a network topology; and a routing module for routing the data from the first process to the second process based on the routing identifier; The determination template is specifically used in response to determining that the first network interface controller associated with the first process and the second network interface controller associated with the second process are located in the same In a computing node or linked to the same switch, the routing identifier is set as a default identifier; in response to determining that the first network interface controller associated with the first process and the second network interface controller are The second network interface controller associated with the process is located on a different computing node or linked to a different chip switch, and the routing identifier is set equal to the first network interface controller associated with the first process. The identifier of the device.