Efficient non-contiguous I/O vector and strided data transfer in one sided communication on multiprocessor computers

[0002] Not applicable

[0003] A portion of the disclosure of this patent document contains material which is subject to copyright protection. The copyright owner has no objection to the facsimile reproduction by anyone of the patent document or the patent disclosure as it appears in the Patent and Trademark Office patent file or records, but otherwise reserves all copyright rights whatsoever.

[0004] 1. Field of the Invention

[0005] The invention disclosed broadly relates to the field of high speed computers, and more particularly relates to the transfer of noncontiguous data blocks during a one-side communications between two or more computational nodes in distributed parallel computing machines.

[0006] 2. Description of the Related Art

[0007] The introduction of highly parallel distributed multiprocessor systems such as the IBM RISC System/6000 Scalable POWERparallel (SP) systems provide high reliability and availability. These systems in their simplest form can be viewed as a plurality of uniprocessor and multiprocessor computer systems coupled together to function as one coordinated system through a local area network (LAN).

[0008] Data transfer between nodes of highly parallel distributed multiprocessor systems is necessary to enable truly scalable and efficient computing. Data transfer between nodes is broadly divided into two groups, contiguous and noncontiguous data transfer. Contiguous data that is stored in adjacent locations in a computer memory. In contrast, noncontiguous data is data that is not stored is collection of adjacent locations in a computer memory device. It is well known that the transfer of noncontiguous data requires more pipeline and supporting processor overhead than the transfer of contiguous data. The transfer of noncontiguous data block is also referred to as a transfer of I/O vectors.

[0009] Typically, there are two types of I/O vectors (i) general I/O vectors where each data block (or vector) can be a different length and (ii) strided I/O vectors where each data block (or vector) is a uniform length. Referring now to FIG. 1, show is the general I/O vector transfer. Shown are four data blocks 100 in strided I/O vector 110. It is important to note that the starting addresses of the data blocks may not be symmetrically spaced as shown. Each of the four data blocks has a starting address a0, a1, a2, a3 and a length 10, 11, 12, 13. The transfer of an I/O vector 110 with four data blocks 100 from an origin task 106 to a target task 108.

[0010] Turning now to FIG. 2 there is shown a block diagram of a strided I/O vector transfer. There are three data blocks 200 (or vector) are shown. Notice that the length or block size 204 of each data block 200 is uniform. Moreover, the stride size 202 or the distance in bytes between the beginning of one block (or vector) and the beginning of the next block (or vector) is uniform. The transfer of an I/O vector 210 with data blocks 200 from a source or origin task 206 to a target task 208 with the same block size and stride size is represented. In the general vector transfer, a number, N, of vectors on the source are transferred to a corresponding number of vectors on the target, in this example 3, where the length 204 of each vector transferred is the same as the length of the corresponding vector on the target task 208. During a strided I/O vector transfer the following parameters are specified, the block size, the stride size, the number of vectors or blocks and the starting addresses of the first block on the source and the target.

[0011] The teaching of a centralized multiprocessor system, such as the system disclosed in the U.S. Pat. No. 5,640,534 issued on Jun. 18, 1997, assigned to Cray Research, with name inventors Douglas R. Beard et al. for a “Method and Apparatus for Chaining Vector Instructions,” does not address the problem with vector transfer on highly parallel distributed multiprocessor systems, such as the IBM SP. More specifically the teachings of the centralized multiprocessor systems do not address the problem on highly parallel distributed multiprocessor systems of the transfer of vector data during a one-side communications between two or more computational nodes (where each node itself can comprise two or more processors). A one-sided communications is a communications where the receiver is not expecting or waiting to receive vector data communications. This data transfer is not efficient and a need exists for optimized noncontiguous data transfer on distributed multiprocessor machines like the IBM SP. These systems allows users to write application programs that run on one or more processing node to transfer vector data in a one-sided communications style. These applications programs make use of a library of APIs (Application Programming Interfaces). An API is a functional interface that allows an application program written in a high level program such as C/C++ or Fortran to use these specified data transfer functions of I/O vectors without understanding the underlying detail. Therefore a need exists for a method and a system to provide I/O vector data transfer during a one-sided communications in a highly parallel distributed multiprocessor system.

[0012] If noncontiguous I/O vector data transfer capability is not available on a distributed multiprocessor machines an application requiring noncontiguous I/O vector data transfer incurs one of two overheads: (I) pipeling and (ii) copying. To transfer non-contiguous data, user in the application program must issue of series of API data transfers. However, the use of successive API data transfer results in LAN pipelining overhead. Alternatively, the application program can be designed to copy all the noncontiguous vector data into a contiguous data buffer before initiating a data transfer. This approach results in copy overheads. Those skilled in the art would know that for efficient noncontiguous data transfer the pipeline costs and the copy costs both must be avoided. An efficient trade-off is needed between the reduction of the number of data packets that are transferred over the network and a reduction of the copy overhead is required. Accordingly, a need exists to overcome these problems by providing an efficient transfer noncontiguous data during one-sided communications.

[0013] Still, another problem with noncontiguous data transfer during a one-side communications in a highly parallel distributed multiprocessor system is that efficient packaging of noncontiguous data into fixed packet sizes must be addressed. The packaging of noncontiguous data reduces the number of data packets that must be sent across the network. Typically, minimum state information of the I/O vector data should be maintained during the node-to-node transfer over the LAN. A spillover state is created during the packing of data into packets when the data not fitting into a predefined packet size is placed into spillover state. The creation and maintenance of a spillover state when packing data into packets is inefficient and should be avoided. Therefore a need exists for a method and apparatus to provide efficient noncontiguous data transfer in a one-sided communications while maintaining minimum state information without producing a spillover state. The spillover state becomes especially difficult to handle if the packet with spillover data is to be re-transmitted.

[0014] Still, another problem with noncontiguous data transfer during a one-side communications in a highly parallel distributed multiprocessor system is that a request to transfer data from a target node to a source node, in a get operation, must include a description of the source data layout to the target. The description of the source data layout is the list of address and length of data for each vector and the number vectors in the transmission. This need to send a description of the layout of source data to a target process includes control information that needs to be sent to the target and back to the source. Accordingly, a need exists to transfer noncontiguous data while avoiding the sending of a description of the source data layout to the target.

[0015] Yet, still another problem with noncontiguous data transfer during a one-side communications in a highly parallel distributed multiprocessor system is that any method to reduce the inefficiencies of data vector into data packets must not be too time-consuming so as to offset any saving in time due to the possible reduction in the number of packets sent. Accordingly, a need exists for a method and apparatus that provides noncontiguous data transfer during a one-side communications that is less costly than the savings in time in reducing the number of packets sent.

[0016] Briefly, in accordance with the present invention, a method for grouping I/O vectors to be transferred across a distributed computing environment comprising a plurality of processing nodes coupled together over a network. The method reduces the total number of packets transmitted over a network between two nodes. The method includes the grouping of two or more I/O vectors into a single message, consisting of one packet with a predetermined maximum size, provided the sum of the sizes of the vectors are small enough to be placed into a single packet. The grouping method finds an efficient collection of vectors to form groups that fit inside a single packet. If two or more of the vectors can be combined so that the resulting single packet size does not exceed the predetermined maximum size, the vectors are grouped accordingly. Vectors whose size is greater than the predetermined maximum packet size are sent as a separate message. This results in a method to efficiently transfer strided vectors such that the total number of packets to be sent is minimized while ensuring that the amount of state information that needs to maintained is the same.

[0017] In accordance with another embodiment of the present invention, a computer readable medium is disclosed corresponding to the above method.

[0018] The subject matter which is regarded as the invention is particularly pointed out and distinctly claimed in the claims at the conclusion of the specification. The foregoing and other objects, features, and advantages of the invention will be apparent from the following detailed description taken in conjunction with the accompanying drawings.

[0019] FIG. 1 is an illustration of a general I/O transfer of a data vector according to the present invention.

[0020] FIG. 2 is an illustration of a strided I/O transfer of a data vector according to the present invention.

[0021] FIG. 3 depicts one example of a highly parallel distributed multiprocessor systems computing environment incorporating the principles of the present invention.

[0022] FIG. 4 depicts one example of an expanded view of a number of the processing nodes of the distributed computing environment of FIG. 3, in accordance with the principles of the present invention.

[0023] FIG. 5 is an illustration of the grouping for general I/O transfer of a data vector according to the present invention.

[0024] FIG. 6 is an illustration of the grouping for strided I/O transfer of a data vector according to the present invention.

[0025] FIG. 7 is an illustration of the grouping for I/O transfer of a data vector for recursive grouping according to the present invention.

[0026] FIG. 8 is an illustration of the grouping for the remaining vectors of FIG. 7 according to the present invention.

[0027] In one embodiment, the techniques of the present invention are used in distributed computing environments in order to provide multicomputer applications that are highly-available. Applications that are highly-available are able to continue to execute after a failure of some of its components. That is, the application is fault-tolerant and the integrity of customer data is preserved. One example of highly available distributed computing environment is the IBM RISC System/6000 Scalable POWERparallel systems, also known as the SP system.

[0028] Referring now in more detail to the drawings in which like numerals refer to like parts throughout several views, FIG. 3 is a block diagram of a distributed computing environment 300 that includes a plurality of nodes 302 coupled to one another via a plurality of network adapters 304. Each node 302 is an independent computer with their own operating system image 308, memory 310 and processor(s) 306 on a system memory bus 318, a system input/output bus 316 couples I/O adapters 312 and network adapter 304. Each network adapter is linked together via a network switch 320.

[0029] In one example, distributed computing environment 300 includes N nodes 302 with one or more processors 306. In one instance, each processing node is, a RISC/6000 computer running AIX, the IBM version of the UNIX operating system. The processing nodes do not have to be RISC/6000 computers running the AIX operating system. Some or all of the processing nodes 302 can include different types of computers and/or different operating systems 308. All of these variations are considered a part of the claimed invention.

[0030] In FIG. 4 shown is an expanded view of a number of processing nodes 302 of the distributed computing environment 300 of FIG. 3, according to the present invention. In one embodiment, an application program 402 that uses Low-Level APIs 404 (LAPI) to send and receive data vectors across two or more processing nodes 302. The operation of the LAPI structures and procedures are more fully described in the following section.

[0031] Summary of Grouping I/O Vectors

[0032] The method for efficient grouping is summarized as follows. First two or more vectors are grouped into a single message, consisting of one packet, provided the sum of the sizes of the vectors are small enough to fit in a single packet. The packing of two or more vectors into one package is defined as a packed packet. The grouping algorithm finds an efficient collection of vectors for each packed packet. If two or more vectors can be combined, the strategy groups accordingly. If a vector is large, i.e. the length of the vector is larger than a single packet size, it may be sent as a separate message. A separate message containing one vector is known as a non-packed vector. The use of packing results in a reduction in pipelining costs and a reduction in the total number of packets transmitted. FIG. 5 is an illustration of the grouping for general I/O transfer of data vector and FIG. 6 is an illustration for a strided I/O transfer of a data vector according to the present invention. In FIG. 5, it is important to notice that the grouping for the general case is according to relative vector size, where vectors 501 and 502 are collected together to form a packed packet 550, vector 503 is collected to form possible multiple non-packed packets 551 and vectors 506, 508 and 510 are grouped together to form a packed packet 552. FIG. 6 is an illustration of the groups in the strided vector transfer. Notice in the strided vector transfer that all of the relative sizes of the vectors 601, 603, 605, and 607 are identical in size. The vectors are partitioned to a predetermined maximum size as depicted by line 609. The vectors are partitioned usually to an even number of bytes, for example 1024 bytes, such that the full payload of the packet is utilized. The truncated portions of the vectors are sent as a separate strided vector as shown in FIGS. 7 and 8.

[0033] An Embodiment of the Grouping Strategy

[0034] In one embodiment, the grouping function for this present invention is embodied in a library of APIs (Application Programming Interfaces). An API is a functional interface that allows an application program 402 written in a high level program such as C/C++ or Fortran to use these specified data transfer functions of I/O vectors without understanding the underlying detail.

[0035] The current solution is to provide vector functions capability within the LAPI library so that pipelining overheads or copy overheads are not incurred. Many references are made to previous versions of LAPI. Refer to online URL www.ibm.com for more information on releases of LAPI. These structures and APIs can be shipped as part of a developer's toolkit on a CD, diskette, downloaded over the Internet or through other computer readable medium now known or later developed.

[0036] General Vector Transfer

[0037] Given a set of N vectors, that need to be transferred from the source to the target, a strategy to make the transfer efficient in terms of the number of data packets is disclosed. Let 1₁, 1₂, . . . , 1_nbe the number of bytes that are to be transferred for vectors 1, 2, . . . N. If the packet payload size is p, then the number of packets needed to transmit the vectors without any grouping, i.e. if they are sent as N separate messages, is:

[0038] No. of packets=┌1₁/p┐+┌1₂/p+. . . +┌1_N/p┐

[0039] If, however, they are grouped into sets of vectors, L₁, L₂, . . . , L_G, where each set L_jcontains one or more vectors, then each set or group of vectors can be sent as a single message. In this case, the number of packets needed would be:

[0040] No of packets=┌L₁/p┐+┌L₂/p┐+. . . +┌L_G/p┐

[0041] One special case brings out the need for grouping. When each vector in 1₁, 1₂, . . . , 1_N, is such that 1₁<<p, 1₂<<p, . . . 1_N, <<p so that 1₁+1₂+. . . , 1_N≦p. In this case, sending each vector as a separate message would require 1 packet. The total number of packets would be N. On the other hand, grouping all the N vectors into one group would require a total of just 1 packet. Of course, the small space would be required for appropriate header information in a packet which contains such a grouped collection of vectors.

[0042] In the general case, we would like to group vectors which fit in a single packet (subject to a specified maximum number of such grouped vectors). If a vector is longer than the payload size, then it will be sent without any special packing to avoid spillover data state maintenance overheads.

[0043] Strided Vector Transfer Description

[0044] In the strided vector transfer, N vectors each of length 1 to be transferred. If each vector is sent as a separate message, then the number of packets is:

[0045] No of packets=┌1/p┐+┌1/p┐+. . . +┌1/p=N*┌1/p┐

[0046] Call this strategy 1.

[0047] If all the vectors were grouped together and sent as one message, then the number of packets sent is:

[0048] ┌N*1/p┐

[0049] Call this strategy 2.

[0050] We would like to find out the saving in number of packets, by using strategy 2 instead of strategy 1. Clearly, if the size 1 is such that it fits exactly into an even number of packets, i.e. IF 1=k*p, then there is no need for grouping, and we will use strategy 1.

[0051] Assuming 1=k*p+r, 0<r<p, we have the number of packets for strategy 1 as:

[0052] N*┌1/p┐=N*┌(k*p+r)/p┐=N*┌k+r/p┐=N*(k+1)

[0053] For strategy 2, the number of packets will be:

[0054] ┌N*1/p┐=N*(k*p+r)/p┐=┌(N*k*p+N*r)/p┐=┌N*k+/N*r/p┐=N*k+┌N*r/p┐

[0055] The saving in number of packets if we were to use strategy 2 is:

[0056] N*(k+1)−(N*k+┌N*r/p┐)=N−[N*r/p]

[0057] Expressed as a percentage, this will be $\frac{\left(N-\left[N*r/p\right]\right)}{\left(\underset{\_}{N}*\left(\underset{\_}{k}+1\right)\right)}$

[0058] If this last expression is smaller than a predetermined maximum specified value, then strategy 1 may be used. Else strategy 2 will be preferred.

[0059] If in fact strategy 2 turns out to be more efficient for a given case, the next problem is to find a grouping of the N vectors. One solution, which fits naturally with the existing scheme and the one proposed in the previous subsection on the strided vector transfers, is to split each vector into 2 parts. The first part, that is of length k*p, is sent as a separate message. This requires k packets. Sending the first parts of all the N vectors thus requires n*k packets. Next, the smaller second parts of the vectors, of length r, 0<r<p, are grouped such that ⊂p/r┘ such parts are sent per packet. Clearly, the total number of packets resulting from these r length parts is ┌N/└p/r┘┌. This modified version of strategy 2 will result in a total packet count of:

[0060] N*k+┌N/└p/r┘┐

[0061] Although, this increases the packet count somewhat compared to strategy 2, it will be apparent to those skilled in the art that this method has the advantage of reduced processing costs in terms of spillover data state maintenance.

[0062] Description of Strided I/O Vector Grouping with Recursion

[0063] Turning now to FIGS. 7 and 8 is an illustration of the grouping for the I/O transfer of a data vector for recursive grouping, according to the present invention. In the strided vector case, where the length of each vector is not an exact multiple of the packet payload, a form of recursive “cutting” of vectors may be employed. As the FIG. 7 shows, the original length of each vector is (L+L₁) where L is an exact multiple say “d” packets of packet payload and L₁is less than the payload. In this figure, we assume L is the same as the packet payload. So we first send (d*four) packets corresponding to the length L of each of the four vectors. Next, the length L₁that remains after length L has been transmitted, may itself be such that it is significantly smaller than the packet payload. If this happens, sending just a block of L₁in one packet would not be an optimal choice. It is more efficient to send several blocks of length L₂, where L₂<L₁, in one packet to make better use of the payload. Also, L₂is chosen so that n*L₂is the packet payload in the typical case, where n is the number of vectors. In the example shown in FIG. 8, we end up transmitting just one packet containing four blocks of length L₂using this approach, instead of the four packets that we would have sent otherwise. This process can be repeated until a stage is reached when the remaining block lengths will be small enough to package a multiple of them in one packet. In FIG. B, the small leftover blocks of length L₃are packaged into one packet and transmitted. Therefore, it takes a total of (4*d+1+1)=6 packets to transmit the four original vectors. On the other hand, without this optimization scheme, we would have sent the four original vectors in 8 packets. When d is equal to one (d=1), a saving of 25% in the number of packets transmitted will result in this case by using our optimization scheme. The saving in number of packets becomes more dramatic as the number of vectors increases and as the number of (small) blocks per packet increases.

[0064] An Embodiment of Structures for General I/O Transfer

[0065] In order to simplify the description of the noncontiguous I/O data vector transfer functions the following example data structure is defined:

[0066] Depending on the vec_type the remaining fields have the interpretations as will be described for FIGS. 1 and 2 below.

[0067] Turning now to FIG. 2, for the general I/O vector transfer case. In this the vec type of the structure defined above is set to a flag denoting general I/O vector transfer. The num vecs represents the number of vectors, the info array contains num vecs buffer addresses representing the starting address for each of the vectors. The len array contains the length in bytes_for each vector respectively. The len array contains num vecs entries. The num vecs field at the origin must be the same as the num vecs field at the target. The len[i] at the origin must also be the same as len[i] at the target for 0<i<num vecs.

[0068] In FIG. 1 the general strided data transfer case. In the strided data transfer case, the vec type must be set to a flag for strided I/O vector transfer. The num vecs is set to the number of vectors. The info[0] contains the starting address of the strided vectors, info[1] contains the size of each block in bytes and info[2] contains the stride size in bytes.

[0069] The len field is a don't care.

[0070] Description: This function transfers data from the origin process address space from locations and lengths described in the org vec to the target process address space in locations and lengths described in the tgt vec. Both structures, org vec and tgt vec, are located in the origin process address space, however the addresses of the actual vector locations in tgt vec refer to addresses in the target address space. This is a nonblocking call, in that, the calling program may not assume that the origin buffer can be changed, nor that contents of target buffers (described in tgt vec) on the target process is ready for use.

[0071] Description: This function transfers data from the target process address space from locations and lengths described in the tgt vec to the origin process address space in locations and lengths described in the org vec. Both structures, org vec and tgt vec, are located in the origin process address space, however the addresses of the actual vector locations in tgt vec refer to addresses in the target address space. This is a nonblocking call, in that, the calling program may not assume that the origin buffer can be changed, nor that contents of origin buffers (described in org vec) on the origin process is ready for use.

[0072] Description: This function is a generic version of the LAPI_Amsendv function. The number of vectors and the lengths of the vectors on the origin and target need not match. The effect of this function is to simply transfer a given number of bytes in noncontiguous buffers specified by the origin vector structure to another number of bytes in noncontiguous buffers specified by the vector structure returned by the header handler. If the total length of the noncontiguous buffers in the target, say N, is less than the total length of the noncontiguous buffers in the origin, say M, then only the first N bytes from the origin buffers will be transferred and the remaining bytes will be discarded. This function transfers hdr hdl function pointer along with the contents of uhdr and data described in org vec from the origin to the target process tgt. When the message arrives at the target process, the header-handler hdr hdl is invoked at the tgt with the pointer to uhdr as one of the parameters.

[0073] Although a specific embodiment of the invention has been disclosed, it will be understood by those having skill in the art that changes can be made to this specific embodiment without departing from the spirit and scope of the invention. The scope of the invention is not to be restricted, therefore, to the specific embodiment, and it is intended that the appended claims cover any and all such applications, modifications, and embodiments within the scope of the present invention.