Patent Description:
In collective communication, processing of collective operations usually requires participation of a plurality of processes. Each process may receive data from several other different processes and need to perform corresponding processing, and send processed data to the other processes.

For communication between processes running on different computer devices, data transmission between different processes needs to be implemented by using network adapters on the different computer devices. Due to network impact, data packets transmitted between the processes disorderly arrive at a receive end computer device, and time of arrival is uncertain. An interrupt is triggered when a process running on the receive end computer device receives data. When an operating system of the receive end computer device is processing a computing task by using all cores, due to reception of data transmitted between the processes, the operating system stops a computing task being processed by using a part of the cores, and turns to processing the interrupt. However, time for the interrupt and context switching causes time consumption that occurs when the operating system of the receive end computer device processes the task, affecting performance of the receive end computer device.

To resolve a problem of overall performance deterioration caused by the interrupt, one manner is to control data transmission between the processes by using a queue when the network adapter sends the data. However, this manner has problems of high latency and resource consumption of the collective communication.

Document <CIT> generally discloses a Network Interface (NI) that includes a host interface, which is configured to receive from a host processor of a node one or more work requests that are derived from an operation to be executed by the node. The NI maintains a plurality of work queues for carrying out transport channels to one or more peer nodes over a network. The NI further includes control circuitry, which is configured to accept the work requests via the host interface, and to execute the work requests using the work queues by controlling an advance of at least a given work queue according to an advancing condition, which depends on a completion status of one or more other work queues, so as to carry out the operation. Document <CIT> generally discloses a system and method for collective send operations on a system area network. The mechanisms of the illustrative embodiments provide for the creation, modification, and removal of collective send queues (CSQs) that allow the upper level protocol (ULP) used by a consumer to send the same message to a collective set of queue pairs (QPs). In order to use the transport services of a CSQ, a consumer process posts a write work request (WR) to the CSQ. The write WR causes a write work queue element (WQE) to be generated and placed in the CSQ. A channel interface (CI) is provided that effectively copies the write WQE to all of the send queues (SQs) of the QPs in the QP set associated with the CSQ. When all the QPs complete processing of their respective write WQEs, the HCA releases all data segments referenced by the write WR. Document <CIT> generally discloses a task scheduling method and device. The task scheduling method comprises: determining, according to operands corresponding to multiple operating tasks in an operating task queue, dependence relationships among the multiple operating tasks); and scheduling the multiple operating tasks in the operating task queue on the basis of the dependence relationships among the multiple operating tasks.

Embodiments of this application provide a method for implementing collective communication, a computer device, and a communication system, to resolve problems of high communication latency and resource consumption in the conventional technology. In particular, this application provides a computer device, and a method, having the features of respective independent claims. The dependent claims relate to preferred embodiments.

According to a first aspect, an embodiment of this application provides a computer device, according to independent claim <NUM>. Further embodiments of the first aspect are provided in dependent claims <NUM>-<NUM>.

According to a second aspect, an embodiment of this application provides a method for implementing communication, according to independent claim <NUM>. Further embodiments of the second aspect are provided in collective dependent claims <NUM>-<NUM>.

The following briefly describes the accompanying drawings for describing embodiments. It is clear that the accompanying drawings in the following descriptions show merely some embodiments of the present invention, and a person of ordinary skill in the art may derive other drawings from these accompanying drawings without creative efforts.

The following describes embodiments of the present invention with reference to the accompanying drawings.

In the specification, claims, and accompanying drawings of this application, the terms "first", "second", and so on are intended to distinguish between similar objects but do not necessarily indicate a specific order or sequence. It should be understood that the data termed in such a way are interchangeable in proper circumstances, so that embodiments of the present invention described herein can be implemented in other orders than the order illustrated or described herein. In addition, the terms "first" and "second" are merely intended for a purpose of description, and shall not be understood as an indication or implication of relative importance or implicit indication of a quantity of indicated technical features. Therefore, a feature limited by "first" or "second" may explicitly or implicitly include one or more features.

In the specification and claims of this application, the terms "include", "contain" and any other variants mean to cover a non-exclusive inclusion, for example, a process, method, system, product, or device that includes a series of steps or modules is not necessarily limited to those expressly listed steps or modules, but may include other steps or modules not expressly listed or inherent to such a process, method, product, or device. Names or numbers of steps in this application do not mean that the steps in the method procedure need to be performed in a time/logical sequence indicated by the names or numbers. An execution sequence of the steps in the procedure that have been named or numbered can be changed based on a technical objective to be achieved, provided that same or similar technical effects can be achieved. Division into units in this application is logical division. During actual application, there may be another division manner. For example, a plurality of units may be combined or integrated into another system, or some features may be ignored or not performed. In addition, the displayed or discussed mutual couplings or direct couplings or communication connections may be implemented through some interfaces. The indirect couplings or communication connections between the units may be implemented in electrical or other similar forms. This is not limited in this application. In addition, units or subunits described as separate parts may or may not be physically separate, may or may not be physical units, or may be distributed into a plurality of circuit units. All or a part of the units may be selected based on actual requirements to achieve the objectives of the solutions of this application.

It should be understood that the terms used in the descriptions of the various examples in the specification and claims of this application are merely intended to describe specific examples, but are not intended to limit the examples. Terms "one" ("a" or "an") and "the" of singular forms used in the descriptions of the various examples and the appended claims are also intended to include plural forms, unless otherwise specified in the context clearly.

It should also be understood that the term " and/or" used in the specification and claims of this application indicates and includes any or all possible combinations of one or more items in associated listed items. The term "and/or" describes an association relationship between associated objects and represents that three relationships may exist. In addition, the character "/" in this application usually indicates an "or" relationship between the associated objects.

It should be understood that determining B based on A does not mean that B is determined based on only A, and B may alternatively be determined based on A and/or other information.

It should be further understood that the term "include" (also referred to as "includes", "including", "comprises", and/or "comprising") used in this specification specifies presence of the stated features, integers, steps, operations, elements, and/or components, with presence or addition of one or more other features, integers, steps, operations, elements, components, and/or their components not excluded.

It should be further understood that the term "if" may be interpreted as a meaning "when" ("when" or "upon"), "in response to determining", or "in response to detecting". Similarly, according to the context, the phrase "if it is determined that" or "if (a stated condition or event) is detected" may be interpreted as a meaning of "when it is determined that", "in response to determining", "when (a stated condition or event) is detected", or "in response to detecting (a stated condition or event)".

It should be understood that "one embodiment", "an embodiment", and "a possible implementation" mentioned in the entire specification mean that particular features, structures, or characteristics related to the embodiment or the implementation are included in at least one embodiment of this application. Therefore, "in one embodiment", "in an embodiment", or "in a possible implementation" appearing throughout this specification does not necessarily mean a same embodiment. In addition, these particular features, structures, or characteristics may be combined in one or more embodiments by using any appropriate manner.

First, some terms and related technologies in this application is explained to facilitate understanding.

Parallel computing is based on an idea: A big problem may be divided into some small problems that can be resolved simultaneously (concurrently) by using an existing resource capability. The big problem is resolved by resolving the small problems. The parallel computing is relative to serial computing, and a feature of the serial computing is that a process sequentially runs computational algorithms according to instructions. The parallel computing is divided into two types: temporal parallelism and spatial parallelism. The temporal parallelism refers to a pipeline technology used in a central processing unit of a computer. Each instruction is divided into a plurality of steps, and the steps may be overlapped and executed in time. The spatial parallelism refers to concurrent execution of computer instructions by a plurality of processors, accelerating troubleshooting. An advantage of the parallel computing is to break a limitation of a computing capability of the serial computing, improve a computing speed, complete a computing task in a shorter period of time, better exert a computing capability of hardware, and save computing costs.

HPC refers to a complete set of computer systems with a specified level of computing capability. Because it is difficult for a single processor to implement such a powerful computing capability, the HPC needs to be jointly implemented by a plurality of central processing units (central processing units, CPUs) or a plurality of hosts (for example, a plurality of computer devices). A main purpose of constructing a high-performance computing system is to improve a computing speed. To achieve a computing speed of trillion times per second, requirements on a system processor, memory bandwidth, a computing mode, an input/output (input/output, I/O) of a system, and storage are very high. Each of these links will directly affect the computing speed of the system. The HPC is mainly used to quickly complete data-intensive, compute-intensive, and I/O-intensive computing in fields such as scientific research, engineering design, finance, industry, and social management. Typical applications include bioengineering, new drug development, petroleum geophysical exploration, vehicle design (aerospace, ship, and automobile), material engineering, nuclear explosion simulation, high-tech weapon manufacturing, cryptographic research, various types of large-scale information processing, and the like. Goals of the high performance computing are to minimize computing time for completing a special computing problem, maximize a scale of a problem that can be completed within specified time, handle a large quantity of complex problems that cannot be achieved before, improve cost-effectiveness, resolve a medium-scale problem and a budget in an expansive manner, and the like.

An MPI is a message passing interface for developing a parallel program based on message passing. A purpose of the MPI is to provide a user with a practical, portable, efficient, and flexible message passing interface. The MPI may be used in a plurality of types of system architectures, such as a distributed/shared memory multi-core processor, a high-performance network, and a combination of these architectures. The MPI is also a parallel programming function library, and compilation and running of the MPI need to be combined with a specific programming language. The MPI is implemented on mainstream operating systems, including Windows and Linux systems. The MPI may be a process-level parallel software middleware. An MPI framework manages all computing processes to form a system and provides various functions for inter-process communication. A process is a running instance of a program. In addition to a program code, the process also includes an execution environment (a memory, a register, a program counter, and the like) of the process, and is an independent executable basic program unit in the operating system. The MPI may support a plurality of different communication protocols, for example, InfiniBand or a transmission control protocol (transmission control protocol, TCP). The MPI encapsulates these protocols and provides a set of unified communication interfaces to shield communication details of a bottom layer. The MPI management framework assigns a rank number (rank number) to each process. The rank number sequentially starts from <NUM>. Specific work completed by each process of an MPI program is determined based on a rank number of the process. MPI processes need to communicate within a communication domain. The communication domain is a communication environment between processes, and includes a process group, a context, virtual topology, and the like. When the MPI is started, the system establishes a global communication domain. Each process is in the global communication domain. A parameter of the communication domain needs to be specified for the inter-process communication.

Collective communication is also called group communication. An important difference between the collective communication and point-to-point communication is that a plurality of processes participate in communication at the same time. This is different from the point-to-point communication in which only two processes of a sender and a receiver are involved. Specific processes participating in the collective communication and a context of the collective communication are limited by a communication domain invoked by the collective communication. The collective communication usually includes three functions: communication, synchronization, and computing. The communication function mainly implements data transmission in a set, the synchronization function implements consistent execution progress of all processes in the set at a specific point, and the computing function is an operation on specific data. An MPI collective communication is one type of common collective communication.

InfiniBand (IB for short) is also referred to as "infinite bandwidth" technology, is a computer network communication standard for high performance computing, has an extremely high throughput and extremely low latency, and is used for data interconnection between computers. The InfiniBand is also used as a direct or switched interconnection between a server and a storage system, and an interconnection between storage systems.

A grid includes a plurality of independent computers, to provide an online computing and storage capability. These computer resources are distributed within a wide range. By using idle computing resources in the grid, a virtual and powerful computing platform may be created. This high-performance computer provides a possibility to deal with large-scale computing problems in fields such as biology, mathematics, and chemistry. The grid organizes interconnected computers, and integrates all types of resources and services connected in a network into a virtual computer having a great capability. For the user, the grid provides an infrastructure including various services and resources, and the user faces a resource far beyond a capacity of any single supercomputer. A powerful computing capability of the grid is used to resolve a problem. In addition, a service provided by any node in the grid may be used, regardless of a physical location of the node.

<FIG> is a schematic diagram of a structure for implementing collective communication. As shown in <FIG>, four processes (for example, a process <NUM>, a process <NUM>, a process <NUM>, and a process <NUM>, which are not shown in the figure) are run in a computer device <NUM>, and separately send data to a computer device <NUM> via a network. The computer device <NUM> receives, by using a queue, the data sent by the four processes in the computer device <NUM>, and writes a payload (payload) sent by each process into a receive queue. After receiving the payloads of the four processes, a process (for example, a process <NUM>, which is not shown in the figure) in the computer device <NUM> performs related processing (for example, summing, obtaining a maximum value, or obtaining a minimum value) and sends processed data.

When the foregoing operations are performed, due to impact of the network, the data sent by each process running on the computer device <NUM> disorderly arrives at the computer device <NUM>, and time of arrival is also uncertain. The computer device <NUM> cannot predict generation time of each interrupt, and the interrupt is an interrupt generated when data transmitted by any one of the four processes running on the computer device <NUM> via the network arrives at the computer device <NUM>. Each time the computer device <NUM> receives a payload sent by a process, the computer device <NUM> triggers an interrupt task to an operating system. When the operating system of the computer device <NUM> is executing a computing task by using all cores, the generated interrupt causes the operating system to stop a computing task being executed by using a part of the cores, turn to processing the interrupt, and return to executing the computing task after processing the interrupt task. When processing the interrupt, the operating system of the computer device <NUM> needs to save a context of a current computing task. Overall overheads are high. In this way, a scheduling program of the operating system of the computer device <NUM> is interfered, and generates an "operating system noise", that is, interrupt overheads generated when a message is received. The system noise causes waits of a large quantity of processes and losses of a large quantity of processor cycles (for example, CPU cycles). In most cases, processing of the data is simple, but time overheads for the interrupt and context switching are greater than time overheads for processing the data by the computer device <NUM>. Therefore, when such an operation is implemented, execution efficiency of the computer device <NUM> is very low.

A manner to resolve the foregoing problem is to introduce a queue for management, and data sent by different processes is managed by using the queue. For example, when the data of the four different processes in <FIG> is managed by using the queue, interrupts are all triggered after completion information of four receive queues all arrives. In this way, generation of the "operating system noise" can be avoided.

However, when management is implemented by using the queue, because the different processes are not distinguished, a process that does not need to be managed is also included in a scope of queue management, resulting in low communication efficiency and resource occupation.

Broadcast communication in collective communication is used as an example. A data flow between eight processes is shown in <FIG> is a schematic logical diagram of a data flow during communication between the eight processes. In <FIG>, a digit in each circle represents a rank number, and a line between the circles represents a communication relationship between the processes.

When communication is implemented without queue management, a structure of queue pairs for the communication between the eight processes in <FIG> is shown in Table <NUM>.

P0 in Table <NUM> represents a process <NUM> in <FIG>, and so on, and P7 represents a process <NUM> in <FIG>. A queue pair (queue pair, QP) is a label for a queue pair. For example, the process <NUM> sends data to a process <NUM>. "Send QP <NUM>" indicates that the process <NUM> sends the data to a QP of the process <NUM>, and "Send disable" indicates that send enable is canceled.

When the inter-process communication shown in <FIG> is managed by using the queue, a structure of queue pairs for the communication between the eight processes shown in <FIG> is shown in Table <NUM>.

It can be learned from Table <NUM> that, when the inter-process communication is managed by using the queue, inter-process communication having no communication dependency is also included in management. The inter-process communication may not need to be managed by using the queue, resulting in problems of prolonged communication duration and waste of resources. For example, data sent by the process <NUM> to the process <NUM>, a process <NUM>, and a process <NUM> in <FIG> may be directly sent. The data may be sent without controlling "send disable" in Table <NUM> by using "send enable" in queue management. To implement management in a queue manner, communication between the process <NUM> and the process <NUM>, the process <NUM>, and the process <NUM> needs to be loaded into the queue, and a sending function is enabled. This not only increases latency of the communication between the process <NUM> and the process <NUM>, the process <NUM>, and the process <NUM>, but also consumes processing resources used when a network adapter executes such management.

During actual implementation, communication between some processes that perform collective communication have no dependency relationship with each other. Communication between the processes that have no dependency relationship is managed by using the queue. This not only increases the communication latency, but also consumes processor performance due to occupation of processor resources in the network adapter due to processing of the management.

An embodiment of this application provides a method for implementing collective communication. When an operation of collective communication is performed, a process that has no communication dependency may not be managed by using a queue. In this way, performance of inter-process communication having no communication dependency in the collective communication can be improved, overall latency of the collective communication can be reduced, and resource occupation and consumption caused by management of the inter-process communication having no communication dependency can be avoided. Collective communication by using an MPI is used as an example. When an upper-layer application (an HPC application, a big data application, or the like) uses the MPI as a communication component, overall performance of the upper-layer application can be accelerated in an end-to-end manner. For example, in a typical application in the molecular dynamics field of the HPC industry, time for MPI collective communication accounts for <NUM>% of entire end-to-end running time. If performance of the MPI collective communication is accelerated, overall end-to-end performance of the application can be accelerated.

First, a device that implements the method for implementing collective communication provided in embodiments of this application is described.

<FIG> is a schematic diagram of a structure of a computer device <NUM> according to an embodiment of this application. As shown in <FIG>, the computer device <NUM> includes a processor <NUM>, a memory <NUM>, a host channel adapter (host channel adapter, HCA) <NUM>, and a bus <NUM>. The processor <NUM>, the memory <NUM>, and the host channel adapter <NUM> communicate with each other through the bus <NUM>. The bus <NUM> may be a peripheral component interconnect (peripheral component interconnect, PCI) bus, a peripheral component interconnect express (peripheral component interconnect express, PCIe) bus, an extended industry standard architecture (extended industry standard architecture, EISA) bus, or the like. The bus may be classified into an address bus, a data bus, a control bus, and the like. For ease of representation, only one bold line is used to represent the bus in <FIG>, but this does not mean that there is only one bus or only one type of bus. The host channel adapter <NUM> is connected to another computer device via a network.

In <FIG>, the processor <NUM> may be a chip having a computing capability, for example, a CPU, a graphics processing unit (Graphics Processing Unit, GPU), a general-purpose graphics processing unit (general-purpose GPU, GPGPU), a tensor processing unit (Tensor Processing Unit, TPU), a data processing unit (Data Processing Unit, DPU), a microprocessor (microprocessor, MP), or a digital signal processor (digital signal processor, DSP).

The memory <NUM> may include a volatile memory (volatile memory), for example, a random access memory (random access memory, RAM), or may include a non-volatile memory (non-volatile memory), for example, a read-only memory (read-only memory, ROM), a flash memory, an HDD, or an SSD. The memory <NUM> stores a program or instructions, for example, including a program code of at least one process or a program code of at least one thread. Certainly, the memory <NUM> may further store data, including but not limited to data that needs to be stored by the computer device <NUM>.

The host channel adapter <NUM> includes a control unit <NUM>, a first interface <NUM>, and a second interface <NUM>. The first interface <NUM> is an interface through which the host channel adapter <NUM> communicates with the processor <NUM>. The first interface <NUM> receives, through the bus <NUM>, a request sent by the processor <NUM>, and converts the received request into a format that can be recognized by the host channel adapter <NUM>, or converts data or a packet that is sent by the host channel adapter <NUM> to the processor <NUM> into a format that can be recognized by the processor <NUM>. The control unit <NUM> receives, through the first interface <NUM>, the request sent by the processor <NUM>, and performs corresponding processing on the received request. The second interface <NUM> is an interface for connecting the computer device <NUM> to a network. The host channel adapter <NUM> receives, through the second interface <NUM>, a request or data sent by another computer device via the network, or sends a request or data to another computer device via the network. Optionally, the second interface <NUM> may include a plurality of ports and be connected to the network through the plurality of ports, to implement synchronous transmission on a plurality of paths.

In an implementation, the host channel adapter <NUM> may be implemented by using a NIC. In another implementation, the host channel adapter <NUM> may alternatively be implemented by using a chipset or an independent chip.

It should be noted that <FIG> shows merely some hardware components and software components for ease of describing embodiments of this application. During specific implementation, the computer device <NUM> may further include another hardware component, for example, a hard disk, or may include other software, for example, an application program or an operating system. A structural composition shown in <FIG> should not be used as a limitation on embodiments of this application.

<FIG> is a schematic diagram of a structure of a system according to an embodiment of this application. As shown in <FIG>, the system includes a computer device <NUM> and a computer device <NUM>. The computer device <NUM> is connected to the computer device <NUM> via a network N100. The network N100 between the computer device <NUM> and the computer device <NUM> may be an InfiniBand-based network. For example, an InfiniBand architecture is used as an interconnection solution between the computer device <NUM> and the computer device <NUM>. Optionally, the network N100 between the computer device <NUM> and the computer device <NUM> may alternatively be an Ethernet network (Ethernet), a remote direct memory access over converged Ethernet (Remote Direct Memory Access over Converged Ethernet, RoCE) network, or the like.

Composition of the computer device <NUM> is similar to that of the computer device <NUM>. The computer device <NUM> includes a processor <NUM>, a memory <NUM>, a host channel adapter <NUM>, and a bus <NUM>. The processor <NUM>, the memory <NUM>, and the host channel adapter <NUM> communicate with each other through the bus <NUM>. The memory <NUM> stores a program or instructions, for example, including a program code of at least one process. The host channel adapter <NUM> includes a control unit <NUM>, a first interface <NUM>, and a second interface <NUM>.

A request initiated by a process running on the computer device <NUM>, for example, a request for sending data to the process running on the computer device <NUM>, is transmitted to the computer device <NUM> via the network N100. A related process running on the computer device <NUM> performs corresponding processing based on the received request. In an implementation, one or more processes in the computer device <NUM> implements MPI collective communication with one or more processes in the computer device <NUM> via the network N100.

The following further describes the collective communication method provided in embodiments of this application by using an example in which an application program running on the computer device <NUM> initiates MPI collective operations to implement communication with the computer device <NUM>.

The processor <NUM> in the computer device <NUM> reads the program or instructions in the memory <NUM>, to implement a corresponding function of the MPI collective communication. <FIG> is a schematic diagram of a logical structure of the program or instructions that need to be executed by the processor <NUM>. As shown in <FIG>, the memory <NUM> includes an application program <NUM>, an application interface module <NUM>, a control module <NUM>, a transmission module <NUM>, and a forwarding module <NUM>. The application interface module <NUM>, the control module <NUM>, and the transmission module <NUM> constitute an MPI layer, and the MPI layer is a unified communication framework implemented based on an MPI standard.

The application program <NUM> may be any application that implements collective communication, including but not limited to an HPC industry application, an HPC-AI industry application, a big data industry application, and the like. These applications usually need a large quantity of computing tasks. To execute the large quantity of computing tasks, a plurality of processes or threads are usually started. Therefore, a collective communication interface of the MPI needs to be invoked for data computing and inter-process information exchange. For example, the application program <NUM> may be an application WRF (Weather Research and Forecasting) in the meteorological field, an application OpenFoam in computational fluid dynamics, an application VASP (Vienna Ab initio Simulation Package) in the molecular dynamics field, or the like.

The application interface module <NUM> is an interface between an MPI application layer and the application program <NUM>, and is configured to receive, from the application program <NUM>, a task that needs to be executed. For example, the application interface module <NUM> may receive an operation request of the collective communication triggered by the application program <NUM>.

The control module <NUM> is configured to convert, based on the operation request delivered by the application program <NUM>, the operation request into a work request (work request, WR). For example, the conversion by the control module <NUM> includes but is not limited to: performing grid division based on a computing instance to be computed, determining a process that needs to be executed, a task that needs to be executed by each process, a communication mode of communication between processes, and the like. For example, the control module <NUM> may be a unified communication group (unified communication group, UCG).

The transmission module <NUM> is configured to: abstract a difference between architectures of different hardware (for example, different network adapters), and provide a low-level application programming interface (application programming interface, API). The low-level API is configured to implement the collective communication. For example, the transmission module <NUM> may be unified communication transport (unified communication transport, UCT).

The forwarding module <NUM> is configured to forward a message or data between an API layer and the host channel adapter <NUM>. For example, the forwarding module <NUM> may be an open fabrics enterprise distribution (open fabrics enterprise distribution, OFED). After the control module <NUM> invokes an interface of the transmission module <NUM> to notify the transmission module <NUM> of a WR that needs to be executed, the transmission module <NUM> notifies the host channel adapter <NUM> by invoking an externally exposed interface of the OFED to knock on a doorbell of the hardware, and sends the WR to the host channel adapter <NUM> for parsing and processing. In an implementation, OFED may be an open source implementation of remote direct memory access (remote direct memory access, RDMA) and kernel bypass (kernel bypass).

<FIG> is a schematic flowchart of a method for identifying a work request to have no communication dependency according to an embodiment of this application. As shown in <FIG>, with reference to the software modules shown in <FIG>, the method includes the following steps.

Step <NUM>: The application program <NUM> initiates an operation request of collective communication.

In an implementation, the operation request of collective communication is a cross-node collective operation request, and different nodes communicate with each other via a network to implement the collective communication. For example, the cross-node operation request of collective communication may be an operation request that needs to be quickly forwarded between a plurality of nodes and/or needs to be accurately synchronized between the plurality of nodes. The node may be a computer device, including but not limited to a computer device that implements a computing function or a computer device that implements a storage function.

In another implementation, the operation request of collective communication is an intra-node operation request, to be specific, collective operations implemented between different processes or different threads in a same node.

Specifically, the application program <NUM> may initiate the operation request of collective communication by using a command for initiating collective operations. Optionally, the command for collective operations may be an MPI command or a shared memory command. The MPI command is used as an example. The MPI command includes but is not limited to MPI_max, MPI_min, MPI_sum, MPI_scatter, MPI_reduce, or the like.

Step <NUM>: The application interface module <NUM> receives the operation request initiated by the application program <NUM>, and forwards the operation request to the control module <NUM>.

After receiving the operation request initiated by the application program <NUM>, the application interface module <NUM> may perform general processing on the received operation request, and send obtained information to the control module <NUM>. The processing performed by the application interface module <NUM> on the operation request includes but is not limited to: obtaining information such as a quantity of processes in the collective communication, a task that needs to be executed by each process, a size of data transmitted between the processes, or a communication domain.

After obtaining the information, the application interface module <NUM> sends or transfers the information to the control module <NUM>. It may be understood that there may be another software module between the application interface module <NUM> and the control module <NUM>, for example, a software module that implements MPI communication based on an MPI communication protocol, and the application interface module <NUM> may send or transfer the obtained information to the control module <NUM> through the module. From a perspective of brief description, in this embodiment of this application, only sending or transmission of the obtained information to the control module <NUM> by the application interface module <NUM> is described.

Step <NUM>: The control module <NUM> converts the operation request of collective communication into a work request based on the information obtained from the application interface module <NUM>.

In an implementation, the control module <NUM> converts the operation request of collective communication into the work request based on the information obtained from the application interface module <NUM>.

In another implementation, the control module <NUM> may convert, based on the information obtained from the application interface module <NUM>, the collective operations into a work request and a control command for executing the work request. The control command is used to control the work request to implement collective operations.

During specific implementation, the control module <NUM> may convert, based on an MPI library stored in the memory <NUM> and in combination with the information obtained from the application interface module <NUM>, the collective operations into the work request, or convert the collective operations into the work request and the control command for executing the work request.

For example, the converting, based on the information obtained from the application interface module <NUM>, the collective operations into a work request and a control command for executing the work request includes the following steps.

Step S1: The control module <NUM> performs grid division based on the quantity of processes in the collective communication and the task that needs to be executed by each process, and allocates a computer device that runs the processes in the collective communication.

If the processes that implement the collective communication need to communicate via a network, different computer devices that communicate via the network need to be allocated to separately perform tasks of the collective communication. The system shown in <FIG> is used as an example. The computer device <NUM> and the computer device <NUM> are separately configured to run related processes, to implement the collective communication via the network. For example, the collective communication is communication between a root process <NUM> and three subprocesses (a subprocess <NUM>, a subprocess <NUM>, and a subprocess <NUM>). The root process <NUM> first needs to receive data from the subprocess <NUM>, the subprocess <NUM>, and the subprocess <NUM> and perform a reduction operation, and sends the data obtained after the reduction operation to the subprocess <NUM>, the subprocess <NUM>, and the subprocess <NUM>. The control module <NUM> may allocate the computer device <NUM> to run the root process <NUM>, and allocate the computer device <NUM> to run the subprocess <NUM>, the subprocess <NUM>, and the subprocess <NUM>. The root process <NUM> implements the collective communication with the subprocess <NUM>, the subprocess <NUM>, and the subprocess <NUM> via the network N100.

If the processes for implementing the collective communication do not need to implement communication via the network, only the computer device <NUM> may be allocated to perform the tasks of the collective communication. The communication between the root process <NUM> and the three subprocesses (the subprocess <NUM>, the subprocess <NUM>, and the subprocess <NUM>) is still used as an example. The root process <NUM>, the subprocess <NUM>, the subprocess <NUM>, and the subprocess <NUM> are all run on the computer device <NUM>. The root process <NUM> may implement communication with the subprocess <NUM>, the subprocess <NUM>, and the subprocess <NUM> via the host channel adapter <NUM>.

Step S2: The control module <NUM> determines a mode of implementing the collective communication between the processes.

Specifically, the control module <NUM> may first select, from the MPI library, an MPI collective communication interface used for the inter-process communication. The MPI collective communication interface includes but is not limited to MPI _Beast, MPI_Allreduce, MPI_Alltoall, or the like. Different MPI collective communication interfaces select different algorithms based on factors such as network topology, the quantity of processes, and the size of transmitted data. Common algorithms include but are not limited to Binomial Tree, K-nomial Tree, Recursive doubling, or the like. The algorithm is specifically used to determine the mode of the inter-process communication.

The control module <NUM> determines, by selecting different MPI collective communication interfaces and based on an algorithm applicable to each type of communication interface, a manner of implementing the collective communication between the processes.

Step <NUM>: The control module <NUM> identifies a work request having no communication dependency, and adds an identifier to the work request having no communication dependency.

For a process in which the control module <NUM> identifies the work request having no communication dependency, refer to a procedure shown in <FIG>.

After identifying the work request having no communication dependency, the control module <NUM> may add identification information to the work request having no communication dependency, to identify that the work request is the work request having no communication dependency. The added identification information may be in any form, and the added identification information may be at any location in the work request.

For example, an extended attribute may be added to Opcode in the work request, to add the identification information. For example, a composition of the work request may be shown in Table <NUM>:.

The control module <NUM> may add an identifier SEND _DIRECTLY to Opcode in the work request shown in Table <NUM>, to identify that the work request is a work request having no communication dependency.

Step <NUM>: Send the work request converted by the control module <NUM> to the host channel adapter <NUM>.

Specifically, the transmission module <NUM> transmits the work request converted by the control module <NUM> to the forwarding module <NUM>, and the forwarding module <NUM> sends the work request to the host channel adapter <NUM>. To be specific, the processor <NUM> transmits, to the forwarding module <NUM> by executing a program of the transmission module <NUM>, the work request converted by the control module <NUM>, and sends the work request to the host channel adapter <NUM> by executing a program of the forwarding module <NUM>. In an implementation, the processor <NUM> may convert, by executing the program of the forwarding module <NUM>, the work request into a format that can be recognized by the host channel adapter <NUM>, and send, through the first interface <NUM>, the work request whose format is converted to the host channel adapter <NUM>.

It may be understood that the control module <NUM> obtains a plurality of work requests after converting the operation request of collective communication. Some of the work requests are work requests having communication dependency, and some of the work requests are work requests having no communication dependency. The work request sent by the processor <NUM> to the host channel adapter includes the work request that has no communication dependency and to which the identifier is added, and also includes the work request having communication dependency. In this way, the host channel adapter can identify, based on the identifier, the work request having no communication dependency, and directly forward the work request.

For any work request, the control module <NUM> may determine, based on a mode of communication between processes in the work request, whether the work request has no communication dependency. <FIG> is a schematic flowchart of a method for identifying whether a work request is a work request having no communication dependency according to an embodiment of this application.

Step <NUM>: The control module <NUM> determines a work request with inter-process communication.

In step <NUM>, the control module <NUM> converts the collective operations into one or more work requests. The control module <NUM> needs to first determine a work request, and determine whether the work request is a work request having no communication dependency. Determining whether a work request is a work request having no communication dependency is determining whether communication between processes having a communication relationship in the work request needs to depend on another process.

An example in which a local process is a process A and a peer process is a process B in a work request is used below for description.

Step <NUM>: The control module <NUM> determines whether the local process (the process A) needs to send data to the peer process (the process B).

If the local process (the process A) needs to send data to the peer process (the process B), step <NUM> is performed. If the local process (the process A) does not need to send data to the peer process (the process B), it indicates that the local process (the process A) needs to receive data sent by the peer process (the process B), and step <NUM> is performed.

Step <NUM>: The control module <NUM> determines whether the data sent by the local process (the process A) to the peer process (the process B) is obtained from another process (for example, a process C, where the process C is used to represent the another process below for description). If the data is not obtained from the another process (the process C), step <NUM> is performed. If the data is obtained from the another process (the process C), step <NUM> is performed.

Step <NUM>: The control module <NUM> determines whether the data sent by the peer process (the process B) to the local process (the process A) is obtained from the another process (the process C). If the data is not obtained from the another process (the process C), step <NUM> is performed. If the data is obtained from the another process (the process C), step <NUM> is performed.

Step <NUM>: The control module <NUM> identifies that the work request is a process having no communication dependency.

If the data sent by the local process (the process A) to the peer process (the process B) is not obtained from the another process (the process C), it indicates that communication between the local process (the process A) and the peer process (the process B) does not depend on the another process. The work request in which the local process (the process A) sends the data to the peer process (the process B) is a work request having no communication dependency.

After identifying the work request having no communication dependency, the control module <NUM> identifies the work request having no communication dependency. In an implementation, a first identifier may be added to identify the work request having no communication dependency. The first identifier is used to indicate that the work request is a work request having no communication dependency.

Optionally, the first identifier may be added to an Opcode field in the work request. For example, the added first identifier is IBV_SEND_DIRECTLY.

Step <NUM>: The control module <NUM> identifies that the work request is a work request having communication dependency or does not perform identification processing.

If the data sent by the local process (the process A) to the peer process (the process B) is obtained from the another process (the process C), it indicates that communication between the local process (the process A) and the peer process (the process B) needs to depend on communication with the another process. The work request in which the local process (the process A) sends the data to the peer process (the process B) is a work request having communication dependency.

After identifying the work request having communication dependency, the control module <NUM> may identify the work request having communication dependency, or may not perform identification processing. If identification processing is not performed, it indicates that the work request is different from the work request having no communication dependency.

In an implementation, the work request having communication dependency may be identified by adding a second identifier to identify the work request having communication dependency. The second identifier is used to indicate that the work request is a work request having communication dependency.

Optionally, the second identifier may be added to the Opcode field in the work request. For example, the added second identifier may be IBV_SEND.

It should be noted that the software modules in <FIG> and <FIG> perform corresponding steps, but during specific implementation, the processor <NUM> implements corresponding functions by executing programs of the software modules. That is, the processor implements the method procedures shown in <FIG> and <FIG> by executing corresponding programs stored in the memory <NUM>.

<FIG> is a schematic diagram of a specific structure of a host channel adapter <NUM> according to an embodiment of this application. As shown in <FIG>, the host channel adapter <NUM> further includes a queue <NUM> and a storage unit <NUM>.

The storage unit <NUM> is connected to a control unit <NUM>, and is configured to: store a program or code for implementing a corresponding function of the control unit <NUM>, and store data that needs to be processed by the control unit <NUM>.

The queue <NUM> includes a plurality of work queues (work queues, WQs). Each work queue includes a plurality of work queue entries (work queue entries, WQEs). Each WQE includes information related to a network event, for example, may be information for sending a message to another node via a network or information for receiving a message from the another node via the network. The work queue is implemented by using at least one QP. Each QP includes a receive queue (receive queue, RQ) and a send queue (send queue, SQ). One QP is usually corresponding to one QP in a peer node. In this way, point-to-point transmission can be implemented. The receive queue is mainly used to receive the WQEs, and the send queue is mainly used to send related WQEs. The queue <NUM> further includes a completion queue (completion queue, CQ). The completion queue records a completion status of the WQE. Each entry in the completion queue is corresponding to one WQE. For example, the completion queue may be associated with a preset group of receive queues, and the group of receive queues are used to receive a message waiting to be received. A producer index (producer index, PI) is used to indicate a recently completed entry in the completion queue. Optionally, the PI may alternatively be used to indicate a recently processed WQE in the work queue. For example, when the completion queue is associated with the preset group of receive queues, and the group of receive queues are used to receive the message waiting to be received, the control unit <NUM> indicates that the message waiting to be received is received by identifying the PI in the completion queue. <FIG> shows only various queues for brevity, and shows only one SQ, RQ, and CQ. However, this does not mean that the queue <NUM> includes only these queues. During specific implementation, the queue <NUM> may further include a plurality of SQs, a plurality of RQs, or a plurality of CQs. Details are not described.

In another implementation, the control unit <NUM> may alternatively determine whether a work request loaded into the queue <NUM> is a data work request or a management work request. If the data work request is received, it is determined whether a management work request that triggers the data work request already exists in the queue <NUM>. When the management work request that triggers the data work request already exists in the queue <NUM>, the data work request is triggered based on the management work request. When there is no management work request that triggers the data work request in the queue <NUM>, the data work request is stored in the receive queue, and the management work request that triggers the data work request is waited for. If the management work request is received, it is determined whether a data work request to be triggered by the management work request already exists in the queue <NUM>. When the data work request to be triggered by the management work request already exists in the queue <NUM>, the data work request is triggered based on the management work request. When there is no data work request to be triggered by the management work request in the queue <NUM>, the management work request is stored in the receive queue, and the data work request to be triggered by the management work request is waited for.

In this embodiment of this application, the queue <NUM> and the storage unit <NUM> may be implemented by using a RAM, for example, may be implemented by using a static random access memory (static random access memory, SRAM) or a dynamic random access memory (dynamic random access memory, DRAM). The queue <NUM> and the storage unit <NUM> may be embedded in the control unit <NUM> or independent of the host channel adapter <NUM>.

Optionally, the control unit <NUM> may further include a computation subunit (not shown in <FIG>), configured to execute a related computing task in the work request. For example, the computation subunit may be an arithmetic logical unit (arithmetic logical unit, ALU). The computation subunit may be embedded in the control unit <NUM>, or may be a subunit independent of the control unit <NUM> in the host channel adapter <NUM>, and is controlled by the control unit <NUM>.

In an implementation, the control unit <NUM> and/or the computation subunit may be implemented by using a field programmable gate array (field programmable gate array, FPGA) and/or an application-specific integrated circuit (application-specific integrated circuit, ASIC).

As shown in <FIG>, after receiving a work request sent by the processor <NUM>, a first interface <NUM> sends the received work request to the control unit <NUM>. In this embodiment of this application, the control unit <NUM> determines whether the received work request including an identifier having no communication dependency.

In an implementation, the control unit <NUM> may determine, by determining whether the received work request includes a first identifier, whether the received work request is a work request having no communication dependency. When the received work request includes the first identifier, it is determined that the work request is a work request having no communication dependency; or when the received work request does not include the first identifier, it is determined that the work request is a work request having communication dependency. For example, after receiving a work request forwarded by the first interface <NUM>, the control unit <NUM> first parses whether an Opcode field in the received work request includes IBV_SEND_DIRECTLY. If IBV_SEND_DIRECTLY is included, it is determined that the work request is a work request having no communication dependency. If IBV_SEND_DIRECTLY is not included, it is determined that the work request is a work request having communication dependency.

In another implementation, the control unit <NUM> may determine, by determining whether the received work request includes a second identifier, whether the received work request is a work request having no communication dependency. When the received work request does not include the second identifier, it is determined that the work request is a work request having no communication dependency, or when the received work request includes the second identifier, it is determined that the work request is a work request having communication dependency.

The control unit <NUM> directly sends the work request having no communication dependency to a second interface <NUM>, to send the work request via the network. For the work request having communication dependency, the control unit <NUM> loads the work request into the queue <NUM>, controls sending of the work request by using the queue, and sends the work request via the network and through the second interface <NUM> when a condition for executing the work request is met.

For example, that the control unit <NUM> manages sending of the work request by using the queue may include the following manners:.

The control unit <NUM> first determines whether a condition for triggering a work request that has a communication dependency on the work request is met. If the condition is not met, the work request is stored in the receive queue for waiting. If the condition is met, sending of the work request is triggered, and the work request is sent via the network and through the second interface <NUM>. For example, in the foregoing example, the data sent by the process A to the process B needs to wait for the process C to send data to the process A. After receiving the work request that the process A sends the data to the process B, the control unit <NUM> loads the work request into the receive queue in the queue <NUM>, and determines whether the process A receives the data sent by the process C, that is, determines whether the completion queue of the queue <NUM> includes a completion record in which the process C sends the data to the process A. If the process A has not received the data sent by the process C, the control unit <NUM> stores, in the receive queue, the work request that the process A sends the data to the process B. After the work request that the process C sends the data to the process A is completed, a WQE of the completion is recorded in the completion queue. When an entry indicated by a PI indicates that the process C has sent the data to the process A, the condition for the work request that the process A sends the data to the process B is met. The control unit <NUM> extracts, from the receive queue, the work request that the process A sends the data to the process B, and sends the work request via the network and through the second interface <NUM>.

According to the foregoing implementation provided in this embodiment of this application, the processor <NUM> identifies the work request having no communication dependency. After receiving a work request of collective operations that is sent by the processor <NUM>, the host channel adapter <NUM> directly sends the work request having no communication dependency via the network. This avoids communication latency caused when the work request having no communication dependency is managed by using the queue, and can reduce resource consumption caused by performing related management y the host channel adapter <NUM>. The work request having no communication dependency is directly sent via the network, and more interrupts are triggered because queue management is not performed. However, latency and resource consumption caused by these interrupts are far less than those caused by the queue management. Therefore, overall communication performance of collective communication can be improved by using the implementation provided in this embodiment of this application.

In the foregoing embodiment, an example in which the collective operations are implemented by using the MPI is used to describe the solution provided in this embodiment of this application. Embodiments of this application are not limited thereto. For the collective operations implemented in another manner, reference may also be made to the foregoing implementation. Details are not described.

<FIG> is a schematic diagram of a structure of a computer device <NUM> according to an embodiment of this application. As shown in <FIG>, the computer device <NUM> includes a processor <NUM>, a memory <NUM>, and a host channel adapter <NUM>. The processor <NUM>, the memory <NUM>, and the host channel adapter <NUM> are connected to each other through a bus.

The memory <NUM> stores a computer executable program. The processor <NUM> is configured to execute the computer executable program, to implement the following operations:.

For a specific implementation of the computer device <NUM> shown in <FIG>, refer to the implementation of the computer device <NUM> shown in <FIG> and the implementations shown in <FIG> and <FIG>. For example, the foregoing communication submodule may be a process or the like described in <FIG> and <FIG>. Details are not described.

By using the computer device <NUM> shown in <FIG>, the processor <NUM> identifies the work request having no communication dependency. After receiving the work request of collective operations that is sent by the processor <NUM>, the host channel adapter <NUM> directly sends the work request having no communication dependency via a network. This avoids communication latency caused when the work request having no communication dependency is managed by using a queue, and can reduce resource consumption caused by performing related management by the host channel adapter <NUM>, to improve overall communication performance of the collective communication.

<FIG> is a schematic diagram of a structure of a communication system according to an embodiment of this application. As shown in <FIG>, the communication system includes at least one second computer device <NUM>, and the at least one second computer device <NUM> communicates with the computer device <NUM> in <FIG> via a network <NUM>.

The embodiment shown in <FIG> may be implemented with reference to the implementation of the system shown in <FIG>. Specifically, the second computer device <NUM> may be implemented with reference to the implementation of the computer device <NUM> in <FIG>. In <FIG>, there may be one or more second computer devices <NUM>, and the computer device <NUM> may communicate with the one or more second computer devices <NUM> via the network <NUM>.

An implementation of communication between the computer device <NUM> and the second computer device <NUM> in <FIG> may be implemented with reference to the implementations shown in <FIG>, <FIG>, and <FIG>. Details are not described again.

<FIG> is a schematic flowchart of a method for implementing collective communication according to an embodiment of this application.

Step <NUM>: Obtain an operation request of collective communication.

Step <NUM>: Convert the operation request of collective communication into a work request, and identify a work request having no communication dependency.

Step <NUM>: Directly forward the work request identified as having no communication dependency, and forward, after queue management, a work request that is not identified as having no communication dependency.

The method shown in <FIG> may be implemented by using a computer device, for example, may be implemented by using the computer device <NUM> in <FIG>. In addition, the method shown in <FIG> may alternatively be implemented with reference to the implementations shown in <FIG> and <FIG>. Details are not described again.

In the method shown in <FIG>, the work request having no communication dependency is directly forwarded via a network by identifying the work request having no communication dependency. This avoids communication latency caused when the work request having no communication dependency is managed by using a queue, and can reduce resource consumption caused by performing related management, to improve overall communication performance of the collective communication.

A person of ordinary skill in the art may be aware that, in combination with the examples described in embodiments disclosed in this specification, units and algorithm steps can be implemented by electronic hardware, computer software, or a combination thereof. To clearly describe the interchangeability between the hardware and the software, the foregoing has generally described compositions and steps of each example based on functions.

In the several embodiments provided in this application, it should be understood that the disclosed system, device, and method may be implemented in other manners. For example, division into the modules is merely logical function division and may be another division during actual implementation. For example, a plurality of modules or components may be combined or integrated. In addition, the displayed or discussed mutual couplings or direct couplings or communication connections may be indirect couplings or communication connections through some interfaces, apparatuses, or units.

When the integrated module is implemented in the form of a software functional unit and sold or used as an independent product, the integrated module may be stored in a computer-readable storage medium. Based on such an understanding, the technical solutions of the present invention essentially, or the part contributing to the prior art, or all or some of the technical solutions may be implemented in the form of a software product. The computer software product is stored in a storage medium and includes several instructions for instructing a computer device (which may be a personal computer, a server, or a network device) to perform all or some of the steps of the methods described in embodiments of the present invention. The foregoing storage medium includes: any medium that can store program code, such as a USB flash drive, a removable hard disk, a read-only memory (ROM, Read-Only Memory), a random access memory (RAM, Random Access Memory), a magnetic disk, or an optical disc.

Claim 1:
A computer device (<NUM>), comprising a processor (<NUM>), a memory (<NUM>), and a host channel adapter (<NUM>), wherein
the memory (<NUM>) stores a computer executable program;
the processor (<NUM>) is configured to execute the computer executable program, to implement the following operations:
obtaining an operation request of collective communication where a plurality of processes participate in communication at the same time;
converting (<NUM>) an operation request of collective communication into a work request, identifying (<NUM>) a work request having no communication dependency which involves determining whether communication between processes having a communication relationship in the work request needs to depend on another process; adding identification information to the work request having no communication dependency; and sending (<NUM>) the work request to the host channel adapter (<NUM>); and
the host channel adapter (<NUM>) is configured to: determine, by determining whether the received work request includes the identification information, whether the received work request is a work request having no communication dependency; directly forward a work request identified as having no communication dependency, and for a work request that is not identified as having no communication dependency: load the work request into a queue, determine whether a condition that is recorded in the queue and that triggers the work request that is not identified as having no communication dependency is met, and when the condition is met, send the work request that is not identified as having no communication dependency.