• OFI libfabric原理及应用解析

    Agenda 目录/议题

    • 编译通信软件
    • 硬件和软件带来的挑战
    • 为什么需要libfabric
    • libfabric架构
    • API分组
    • socket应用 VS libfabric应用区别
    • GPU数据传输示例

    编译通信软件

    • 可靠面向连接的TCP和无连接的数据报UDP协议
    • 高性能计算HPC或人工智能AI

    软硬件复杂性带来的挑战

    • 上千个节点的集群, 不同的网络类型(以太网, IB, 光纤等), 熊猫博士提到的CPU/GPU/XPU/IPU等等
    • xelink, nvlink, gpu
    • 软件库, 如nccl,intel mpi等
    • 所以咱们需要一个通用的通信库来桥接这些复杂的软硬件资源

    libfabric来解决上面的问题

    • 统一API, 让程序员更轻松
    • 高性能和高可扩展性
    • 核心组件: 众多网卡提供者的库, 核心服务, 测试程序等

    为什么需要libfabric

    • 承上启下, 桥接底层复杂多样的网络(socket, rdma, gpu, 共享内存等)和上层MPI, CCL, 共享内存等应用

    socket编程与libfabric编程对比

    • 语义类似, 如获取信息, bind, connect等, 但是libfabric底层支持多种网络类型

    支持GPU通信

    • gpu通信示例, 支持intel gpu, dpu, 或者其他供应商

    架构与四种服务

    • 控制类: 发现底层设备, 属性, 能力等
    • 通信接口: 建立连接, 初始资源等
    • 数据传输: 发送和接收数据
    • 完成服务: 报告发送或接收状态

    libfabric API分组

    • 整合底层提供者和上层开发者使用统一的API编程, 就像熊猫博士说的那样, 方便了提供者更方便的提供插件, 也方便了上层应用开发者

    tcp socket 代码截图

    socket收发数据示例

    server端启动:./example_socket客户端连接和发送:./example_socket 192.168.5.6Hello server this is client.

    example_socket.c 源码

    1. #include <stdio.h>
    2. #include <string.h>
    3. #include <sys/socket.h>
    4. #include <arpa/inet.h>
    5.  
    6. char *dst_addr = NULL;
    8. int main(int argc, char *argv[])
    9. {
    10.     int socket_desc, client_sock, client_size;
    11.     struct sockaddr_in server_addr, client_addr;
    12.     char server_message[2000], client_message[2000];
    13.  
    14.     dst_addr = inet_addr(argv[1]);
    15.     
    16.     // Clean buffers:
    17.     memset(server_message, '\0', sizeof(server_message));
    18.     memset(client_message, '\0', sizeof(client_message));
    19.     
    20.     // Create socket:
    21.     socket_desc = socket(AF_INET, SOCK_STREAM, 0);
    22.     
    23.     if(socket_desc < 0){
    24.         printf("Error while creating socket\n");
    25.         return -1;
    26.     }
    27.     printf("Socket created successfully\n");
    28.     
    29.     // Set port and IP:
    30.     server_addr.sin_family = AF_INET;
    31.     server_addr.sin_port = "43192";
    32.     server_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
    33.     if (!dst_addr) {
    34.     	// Bind to the set port and IP:
    35.     	if(bind(socket_desc, (struct sockaddr*)&server_addr, sizeof(server_addr))<0){
    36.     		printf("Couldn't bind to the port\n");
    37.     		return -1;
    38.     	}
    39.     	printf("Binding complete\n");
    40.     
    41.     	// Listen for clients:
    42.     	if(listen(socket_desc, 1) < 0){
    43.         	printf("Error while listening\n");
    44.         	return -1;
    45.     	}
    46.     	printf("Listening for incoming connections...\n");
    47.     
    48.     	// Accept an incoming connection:
    49.     	client_size = sizeof(client_addr);
    50.     	client_sock = accept(socket_desc, (struct sockaddr*)&client_addr, &client_size);
    51.     
    52.     	if (client_sock < 0){
    53.         	printf("Can't accept\n");
    54.         	return -1;
    55.     	}
    56.     	printf("Client connected at IP: %s and port: %i\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));
    57.     
    58.     	// Receive client's message:
    59.     	if (recv(client_sock, client_message, sizeof(client_message), 0) < 0){
    60.         	printf("Couldn't receive\n");
    61.         	return -1;
    62.     	}
    63.     	printf("Msg from client: %s\n", client_message);
    64.     
    65.     	// Respond to client:
    66.     	strcpy(server_message, "This is the server's message.");
    67.     
    68.     	if (send(client_sock, server_message, strlen(server_message), 0) < 0){
    69.         	printf("Can't send\n");
    70.         	return -1;
    71.     	}
    72.     }
    73.     
    74.     if (dst_addr) {
    75.        // Send connection request to server:
    76.        if(connect(socket_desc, (struct sockaddr*)&server_addr, sizeof(server_addr)) < 0){
    77. 	       printf("Unable to connect\n");
    78. 	       return -1;
    79.        }
    80.        printf("Connected with server successfully\n");
    82.        // Get input from the user:
    83.        printf("Enter message: ");
    84.        gets(client_message);
    86.        // Send the message to server:
    87.        if(send(socket_desc, client_message, strlen(client_message), 0) < 0){
    88. 	       printf("Unable to send message\n");
    89. 	       return -1;
    90.        }
    92.        // Receive the server's response:
    93.        if(recv(socket_desc, server_message, sizeof(server_message), 0) < 0){
    94. 	       printf("Error while receiving server's msg\n");
    95. 	       return -1;
    96.        }
    97.        printf("Server's response: %s\n",server_message);
    98.     }
    99.     
    100.     // Close the client socket:
    101.     close(client_sock);
    102.     // Closing the server socket:
    103.     close(socket_desc);
    105.     return 0;
    106. }

    libfabric收发数据示例截图

    1. 服务端启动: ./example_msg客户端连接发送数据: ./example_msg 192.168.5.6
    2. 客户端连接发送数据: ./example_msg 192.168.5.6

    参考代码

    1. https://github.com/ssbandjl/libfabric/blob/main/fabtests/functional/example_msg.c
    2. /*
    3.  *
    4.  * This software is available to you under the BSD license
    5.  * below:
    6.  *
    7.  *     Redistribution and use in source and binary forms, with or
    8.  *     without modification, are permitted provided that the following
    9.  *     conditions are met:
    10.  *
    11.  *      - Redistributions of source code must retain the above
    12.  *        copyright notice, this list of conditions and the following
    13.  *        disclaimer.
    14.  *
    15.  *      - Redistributions in binary form must reproduce the above
    16.  *        copyright notice, this list of conditions and the following
    17.  *        disclaimer in the documentation and/or other materials
    18.  *        provided with the distribution.
    19.  *
    20.  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
    21.  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
    22.  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
    23.  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
    24.  * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
    25.  * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
    26.  * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
    27.  * SOFTWARE.
    28.  */
    30. #include <stdio.h>
    31. #include <stdlib.h>
    32. #include <getopt.h>
    33. #include <netinet/in.h>
    34. #include <arpa/inet.h>
    35. #include <netdb.h>
    36. #include <rdma/fabric.h>
    37. #include <rdma/fi_domain.h>
    38. #include <rdma/fi_endpoint.h>
    39. #include <rdma/fi_cm.h>
    40. #include <shared.h>
    41.  
    42. //Build with
    43. //gcc -o example_msg example_msg.c -L<path to libfabric lib> -I<path to libfabric include> -lfabric
    44. //gcc -o example_msg example_msg.c -L/home/xb/project/libfabric/libfabric/build/lib -I/home/xb/project/libfabric/libfabric/build/include -I/home/xb/project/libfabric/libfabric/build/include -lfabric
    45.  
    46. #define BUF_SIZE 64
    47.  
    48. char *dst_addr = NULL;
    49. char *port = "9228";
    50. struct fi_info *hints, *info, *fi_pep;
    51. struct fid_fabric *fabric = NULL;
    52. struct fid_domain *domain = NULL;
    53. struct fid_ep *ep = NULL;
    54. struct fid_pep *pep = NULL;
    55. struct fid_cq *cq = NULL;
    56. struct fid_eq *eq = NULL;
    57. struct fi_cq_attr cq_attr = {0};
    58. struct fi_eq_attr eq_attr = {
    59. 	.wait_obj = FI_WAIT_UNSPEC
    60. };
    61. //const struct sockaddr_in *sin;
    62. char str_addr[INET_ADDRSTRLEN];
    63. int ret;
    64. char buffer[BUF_SIZE];
    65. fi_addr_t fi_addr = FI_ADDR_UNSPEC;
    66. struct fi_eq_cm_entry entry;
    67. uint32_t event;
    68. ssize_t rd;
    69.  
    70. /* Initializes all basic OFI resources to allow for a server/client to exchange a message */
    71. static int start_client(void)
    72. {
    73.  
    74. 	ret = fi_getinfo(FI_VERSION(1,9), dst_addr, port, dst_addr ? 0 : FI_SOURCE,
    75. 			hints, &info);
    76. 	if (ret) {
    77. 	        printf("fi_getinfo: %d\n", ret);
    78. 	        return ret;
    79. 	}
    80.  
    81. 	ret = fi_fabric(info->fabric_attr, &fabric, NULL);
    82. 	if (ret) {
    83. 		printf("fi_fabric: %d\n", ret);
    84.         	return ret;
    85. 	}
    86.  
    87. 	ret = fi_eq_open(fabric, &eq_attr, &eq, NULL);
    88. 	if (ret) {
    89. 	        printf("fi_eq_open: %d\n", ret);
    90. 	        return ret;
    91. 	}
    92.  
    93. 	ret = fi_domain(fabric, info, &domain, NULL);
    94. 	if (ret) {
    95.         	printf("fi_domain: %d\n", ret);
    96.         	return ret;
    97.         }
    98.  
    99. 	/* Initialize our completion queue. Completion queues are used to report events associated
    100. 	 * with data transfers. In this example, we use one CQ that tracks sends and receives, but
    101. 	 * often times there will be separate CQs for sends and receives.
    102. 	 */
    103. 	cq_attr.size = 128;
    104. 	cq_attr.format = FI_CQ_FORMAT_MSG;
    105. 	ret = fi_cq_open(domain, &cq_attr, &cq, NULL);
    106. 	if (ret) {
    107. 		printf("fi_cq_open error (%d)\n", ret);
    108. 		return ret;
    109. 	}
    110.  
    111. 	/* Bind our CQ to our endpoint to track any sends and receives that come in or out on that endpoint.
    112. 	 * A CQ can be bound to multiple endpoints but one EP can only have one send CQ and one receive CQ
    113. 	 * (which can be the same CQ).
    114. 	 */
    115.  
    116. 	ret = fi_endpoint(domain, info, &ep, NULL);
    117. 	if (ret) {
    118. 	        printf("fi_endpoint: %d\n", ret);
    119. 	        return ret;
    120. 	}
    121.  
    122. 	ret = fi_ep_bind(ep, &cq->fid, FI_SEND | FI_RECV);
    123.         if (ret) {
    124.             printf("fi_ep_bind cq error (%d)\n", ret);
    125.             return ret;
    126.         }
    127.  
    128. 	ret = fi_ep_bind((ep), &(eq)->fid, 0);
    129. 	if (ret) {
    130. 		printf("fi_ep_bind: %d\n", ret);
    131. 		return ret;
    132. 	}
    133.  
    134. 	ret = fi_enable(ep);
    135.         if (ret) {
    136. 		printf("fi_enable: %d\n", ret);
    137. 	        return ret;
    138. 	}
    139.  
    140. 	ret = fi_connect(ep, info->dest_addr, NULL, 0);
    141. 	if (ret) {
    142.  	        printf("fi_connect: %d\n", ret);
    143.  	        return ret;
    144. 	}
    145.  
    146.     rd = fi_eq_sread(eq, &event, &entry, sizeof(entry), -1, 0);
    147.     if (rd != sizeof(entry)) {
    148.         ret = (int) rd;
    149.         printf("fi_eq_sread: %d\n", ret);
    150.         return ret;
    151. 	}
    152.  
    153. 	return 0;
    154. }
    155.  
    156. static int start_server(void)
    157. {
    158. 	const struct sockaddr_in *sin;
    159.  
    160. 	/* The first OFI call to happen for initialization is fi_getinfo which queries libfabric
    161. 	 * and returns any appropriate providers that fulfill the hints requirements. Any applicable
    162. 	 * providers will be returned as a list of fi_info structs (&info). Any info can be selected.
    163. 	 * In this test we select the first fi_info struct. Assuming all hints were set appropriately,
    164. 	 * the first fi_info should be most appropriate.
    165. 	 * The flag FI_SOURCE is set for the server to indicate that the address/port refer to source
    166. 	 * information. This is not set for the client because the fields refer to the server, not
    167. 	 * the caller (client). */
    168. 	/* 初始化时发生的第一个 OFI 调用是 fi_getinfo，它查询 libfabric 并返回满足提示要求的任何适当的提供程序。 任何适用的提供程序都将作为 fi_info 结构 (&info) 列表返回。 可以选择任何信息。 在此测试中，我们选择第一个 fi_info 结构。 假设所有提示均已正确设置，第一个 fi_info 应该是最合适的。 为服务器设置标志FI_SOURCE以指示地址/端口引用源信息。 这不是为客户端设置的，因为这些字段引用服务器，而不是调用者（客户端） */
    169. 	ret = fi_getinfo(FI_VERSION(1,9), dst_addr, port, dst_addr ? 0 : FI_SOURCE,
    170. 			 		hints, &fi_pep);
    171. 	if (ret) {
    172. 		printf("fi_getinfo error (%d)\n", ret);
    173. 		return ret;
    174. 	}
    175.  
    176. 	/* Initialize our fabric. The fabric network represents a collection of hardware and software
    177. 	 * resources that access a single physical or virtual network. All network ports on a system
    178. 	 * that can communicate with each other through their attached networks belong to the same fabric.
    179. 	 */
    180. 	ret = fi_fabric(fi_pep->fabric_attr, &fabric, NULL); // 打开fabric, 初始化任何资源前需要打开fabric
    181. 	if (ret) {
    182. 		printf("fi_fabric error (%d)\n", ret);
    183. 		return ret;
    184. 	}
    185.  
    186. 	/* Initialize our endpoint. Endpoints are transport level communication portals which are used to
    187. 	 * initiate and drive communication. There are three main types of endpoints:
    188. 	 * FI_EP_MSG - connected, reliable
    189. 	 * FI_EP_RDM - unconnected, reliable
    190. 	 * FI_EP_DGRAM - unconnected, unreliable
    191. 	 * The type of endpoint will be requested in hints/fi_getinfo. Different providers support different
    192. 	 * types of endpoints. TCP supports only FI_EP_MSG but when used with RxM, can support FI_EP_RDM.
    193. 	 * In this application, we requested TCP and FI_EP_MSG.
    194. 	 */
    195.  
    196.     ret = fi_eq_open(fabric, &eq_attr, &eq, NULL); // 打开事件队列EQ, 一般用于建连, 收发数据产生的事件
    197.     if (ret) {
    198.             printf("fi_eq_open: %d\n", ret);
    199.             return ret;
    200.     }
    201.  
    202.     ret = fi_passive_ep(fabric, fi_pep, &pep, NULL); // 打开被动端点, 常用与服务端监听端口, 支持多个客户端domain连接进来
    203.     if (ret) {
    204.         printf("fi_passive_ep: %d\n", ret);
    205.         return ret;
    206.     }
    207.  
    208.     ret = fi_pep_bind(pep, &eq->fid, 0); // 为端点绑定事件队列
    209.     if (ret) {
    210.         printf("fi_pep_bind %d", ret);
    211.         return ret;
    212.     }
    213.  
    214.     ret = fi_listen(pep); // 监听端点, 等待客户端连接请求
    215.     if (ret) {
    216.         printf("fi_listen %d", ret);
    217.         return ret;
    218.     }
    219.  
    220. 	return 0;
    221. }
    222.  
    223. static int complete_connection(void)
    224. {
    225.  
    226. 	rd = fi_eq_sread(eq, &event, &entry, sizeof(entry), -1, 0); // 等待读取客户端触发的服务端事件, 读取事件, 推动进展(驱动程序运转)
    227.     if (rd != sizeof entry) {
    228. 	ret = (int) rd;
    229.             printf("fi_eq_sread: %d", ret);
    230. 		if (ret)
    231. 			goto err;
    232.     }
    233.  
    234. 	ret = fi_domain(fabric, info, &domain, NULL); // domain域用于将资源分组, 可基于域来做管理
    235. 	if (ret) {
    236. 		printf("fi_domain: %d\n", ret);
    237. 	        return ret;
    238.         }
    239.  
    240. 	ret = fi_domain_bind(domain, &eq->fid, 0);
    241. 	if (ret) {
    242. 	        printf("fi_domain_bind: %d\n", ret);
    243. 	        return ret;
    244. 	}
    245.  
    246. 	/* 
    247. 	 * Initialize our completion queue. Completion queues are used to report events associated
    248.          * with data transfers. In this example, we use one CQ that tracks sends and receives, but
    249.          * often times there will be separate CQs for sends and receives.
    250. 	 */
    251.     cq_attr.size = 128;
    252.     cq_attr.format = FI_CQ_FORMAT_MSG;
    253.     ret = fi_cq_open(domain, &cq_attr, &cq, NULL);
    254.     if (ret) {
    255.         printf("fi_cq_open error (%d)\n", ret);
    256.         return ret;
    257.     }
    258.  
    259.     /* Bind our CQ to our endpoint to track any sends and receives that 
    260. 	 * come in or out on that endpoint. A CQ can be bound to multiple
    261. 	 * endpoints but one EP can only have one send CQ and one receive CQ
    262.      * (which can be the same CQ).
    263. 	*/
    264.  
    265. 	ret = fi_endpoint(domain, info, &ep, NULL); // 用于客户端, 主动端点, 发起建连
    266.         if (ret) {
    267.                 printf("fi_endpoint: %d\n", ret);
    268.                 return ret;
    269.         }
    270.  
    271. 	ret = fi_ep_bind(ep, &cq->fid, FI_SEND | FI_RECV);
    272.         if (ret) {
    273.                 printf("fi_ep_bind cq error (%d)\n", ret);
    274.                 return ret;
    275.         }
    276.  
    277. 	ret = fi_ep_bind((ep), &(eq)->fid, 0);
    278.         if (ret) {
    279.                 printf("fi_ep_bind: %d\n", ret);
    280.                 return ret;
    281.         }
    282.  
    283. 	ret = fi_enable(ep);
    284.         if (ret) {
    285. 			printf("fi_enable: %d", ret);
    286.             return ret;
    287. 		}
    288.  
    289. 	ret = fi_accept(ep, NULL, 0);
    290. 	if (ret) {
    291. 	        printf("fi_accept: %d\n", ret);
    292. 	        return ret;
    293. 	}
    294.  
    295. 	rd = fi_eq_sread(eq, &event, &entry, sizeof(entry), -1, 0);
    296. 	if (rd != sizeof(entry)) {
    297. 		ret = (int) rd;
    298. 		printf("fi_eq_read: %d\n", ret);
    299. 		return ret;
    300. 	}
    301.     return 0;
    302.  
    303. err:
    304.     if (info)
    305.     	fi_reject(pep, info->handle, NULL, 0);
    306.     return ret;
    307.  
    308. }
    309.  
    310. static void cleanup(void)
    311. {
    312. 	int ret;
    313. 	
    314. 	/* All OFI resources are cleaned up using the same fi_close(fid) call. */
    315. 	if (ep) {
    316. 		ret = fi_close(&ep->fid);
    317. 		if (ret)
    318. 			printf("warning: error closing EP (%d)\n", ret);
    319. 	}
    320. 	if (pep) {
    321. 		ret = fi_close(&pep->fid);
    322. 		if (ret)
    323. 			printf("warning: error closing PEP (%d)\n", ret);
    324. 	}
    325.  
    326. 	ret = fi_close(&cq->fid);
    327. 	if (ret)
    328. 		printf("warning: error closing CQ (%d)\n", ret);
    329. 	
    330. 	ret = fi_close(&domain->fid);
    331. 	if (ret)
    332. 		printf("warning: error closing domain (%d)\n", ret);
    333.  
    334. 	ret = fi_close(&eq->fid);
    335.         if (ret)
    336.             printf("warning: error closing EQ (%d)\n", ret);
    337.  
    338. 	ret = fi_close(&fabric->fid);
    339.         if (ret)
    340.             printf("warning: error closing fabric (%d)\n", ret);
    341.  
    342. 	if (info)
    343. 		fi_freeinfo(info);
    344.  
    345. 	if (fi_pep)
    346.         fi_freeinfo(fi_pep);
    347. }
    348.  
    349. /* Post a receive buffer. This call does not ensure a message has been received, just
    350.  * that a buffer has been passed to OFI for the next message the provider receives.
    351.  * Receives may be directed or undirected using the address parameter. Here, we
    352.  * pass in the fi_addr but note that the server has not inserted the client's
    353.  * address into its AV, so the address is still FI_ADDR_UNSPEC, indicating that
    354.  * this buffer may receive incoming data from any address. An application may
    355.  * set this to a real fi_addr if the buffer should only receive data from a certain
    356.  * peer.
    357.  * When posting a buffer, if the provider is not ready to process messages (because
    358.  * of connection initialization for example), it may return -FI_EAGAIN. This does
    359.  * not indicate an error, but rather that the application should try again later.
    360.  * This is why we almost always wrap sends and receives in a do/while. Some providers
    361.  * may need the application to drive progress in order to get out of the -FI_EAGAIN
    362.  * loop. To drive progress, the application needs to call fi_cq_read (not necessarily
    363.  * reading any completion entries).
    364.  */
    365. static int post_recv(void)
    366. {
    367. 	int ret;
    368.  
    369. 	do {
    370. 		ret = fi_recv(ep, buffer, BUF_SIZE, NULL, fi_addr, NULL);
    371. 		if (ret && ret != -FI_EAGAIN) {
    372. 			printf("error posting recv buffer (%d\n", ret);
    373. 			return ret;
    374. 		}
    375. 		if (ret == -FI_EAGAIN)
    376. 			(void) fi_cq_read(cq, NULL, 0);
    377. 	} while (ret);
    378.  
    379. 	return 0;
    380. }
    381.  
    382. /* Post a send buffer. This call does not ensure a message has been sent, just that
    383.  * a buffer has been submitted to OFI to be sent. Unlike a receive buffer, a send
    384.  * needs a valid fi_addr as input to tell the provider where to send the message.
    385.  * Similar to the receive buffer posting porcess, when posting a send buffer, if the
    386.  * provider is not ready to process messages, it may return -FI_EAGAIN. This does not
    387.  * indicate an error, but rather that the application should try again later. Just like
    388.  * the receive, we drive progress with fi_cq_read if this is the case.
    389.  */
    390. static int post_send(void)
    391. {
    392. 	char *msg = "Hello, server! I am the client you've been waiting for!\0";
    393. 	int ret;
    394.  
    395. 	(void) snprintf(buffer, BUF_SIZE, "%s", msg);
    396.  
    397. 	do {
    398. 		ret = fi_send(ep, buffer, BUF_SIZE, NULL, fi_addr, NULL);
    399. 		if (ret && ret != -FI_EAGAIN) {
    400. 			printf("error posting send buffer (%d)\n", ret);
    401. 			return ret;
    402. 		}
    403. 		if (ret == -FI_EAGAIN)
    404. 			(void) fi_cq_read(cq, NULL, 0);
    405. 	} while (ret);
    406.  
    407. 	return 0;
    408. }
    409.  
    410. /* Wait for the message to be sent/received using the CQ. fi_cq_read not only drives progress
    411.  * but also returns any completed events to notify the application that it can reuse
    412.  * the send/recv buffer. The returned completion entry will have fields set to let the application
    413.  * know what operation completed. Not all fields will be valid. The fields set will be indicated
    414.  * by the cq format (when creating the CQ). In this example, we use FI_CQ_FORMAT_MSG in order to
    415.  * use the flags field.
    416.  */
    417. static int wait_cq(void)
    418. {
    419. 	struct fi_cq_err_entry comp;
    420. 	int ret;
    421.  
    422. 	do {
    423. 		ret = fi_cq_read(cq, &comp, 1);
    424. 		if (ret < 0 && ret != -FI_EAGAIN) {
    425. 			printf("error reading cq (%d)\n", ret);
    426. 			return ret;
    427. 		}
    428. 	} while (ret != 1);
    429.  
    430. 	if (comp.flags & FI_RECV)
    431. 		printf("I received a message!\n");
    432. 	else if (comp.flags & FI_SEND)
    433. 		printf("My message got sent!\n");
    434.  
    435. 	return 0;
    436. }
    437.  
    438. static int run(void)
    439. {
    440. 	int ret;
    441.  
    442. 	if (dst_addr) {
    443. 		printf("Client: send to server %s\n", dst_addr);
    444.  
    445. 		ret = post_send();
    446. 		if (ret)
    447. 			return ret;
    448.  
    449. 		ret = wait_cq();
    450. 		if (ret)
    451. 			return ret;
    452.  
    453. 	} else {
    454. 		printf("Server: post buffer and wait for message from client\n");
    455.  
    456. 		ret = post_recv();
    457. 		if (ret)
    458. 			return ret;
    459.  
    460. 		ret = wait_cq();
    461. 		if (ret)
    462. 			return ret;
    463.  
    464. 		printf("This is the message I received: %s\n", buffer);
    465. 	}
    466. 	return 1;
    467. }
    468.  
    469. int main(int argc, char **argv)
    470. {
    471. 	int ret;
    472.  
    473. 	/* Hints are used to request support for specific features from a provider */
    474. 	hints = fi_allocinfo(); // 
    475. 	if (!hints)
    476. 		return EXIT_FAILURE;
    477.  
    478. 	/* Server run with no args, client has server's address as an argument */
    479. 	dst_addr = argv[1];
    482. //Set anything in hints that the application needs
    484. 	/* Request FI_EP_MSG (reliable datagram) endpoint which will allow us
    485. 	 * to reliably send messages to peers without having to listen/connect/accept. */
    486. 	hints->ep_attr->type = FI_EP_MSG; // 可靠数据报端点, 类似socket, 但无须执行listen/connect/accept
    488. 	/* Request basic messaging capabilities from the provider (no tag matching,
    489. 	 * no RMA, no atomic operations) */
    490. 	hints->caps = FI_MSG;
    492. 	/* Specifically request the tcp provider for the simple test */
    493. 	// hints->fabric_attr->prov_name = "tcp"; // 类似socket的, 面向连接的消息类型端点
    494.  	hints->fabric_attr->prov_name = "ofi_rxm;verbs";
    496. 	/* Specifically request SOCKADDR_IN address format to simplify addressing for test */
    497. 	hints->addr_format = FI_SOCKADDR_IN;
    499. 	/* Default to FI_DELIVERY_COMPLETE which will make sure completions do not get generated
    500. 	 * until our message arrives at the destination. Otherwise, the client might get a completion
    501. 	 * and exit before the server receives the message. This is to make the test simpler */
    502. 	/* 默认为 FI_DELIVERY_COMPLETE，这将确保在我们的消息到达目的地之前不会生成完成(等待)。 否则，客户端可能会在服务器收到消息之前完成并退出。 这是为了让测试更简单 */
    503. 	hints->tx_attr->op_flags = FI_DELIVERY_COMPLETE;
    505. //Done setting hints
    507. 	if (!dst_addr) {
    508. 		ret = start_server();
    509. 		if (ret) {
    510. 			goto out;
    511. 			return ret;
    512. 		}
    513. 	}
    515. 	ret = dst_addr ? start_client() : complete_connection();
    516. 	if (ret) {
    517. 		goto out;
    518. 		return ret;
    519. 	}
    521. 	ret = run();
    522. out:
    523. 	cleanup();
    524. 	return ret;
    525. }

    socket vs libfabric消息类型示意图(两者都可完成建连和消息收发)

    GPU数据传输示例

    左边是内存直接访问DMA ibv verbs示例, 右边是DMA libfabric统一API的语义的示例

    1. verbs
    2. 服务端: ./rdmabw-xe -m host
    3. 客户端: ./rdmabw-xe -m host -S 1 -t write 192.168.5.6  #都是用主机内存, 完成1字节的远程内存写操作

    主机对主机GPU libfabric 内存直接访问DMA示例

    1. libfabric, host -> host DMA
    2. 服务端: ./fi-rdmabw-xe -m host
    3. 客户端: ./fi-rdmabw-xe -m host -S 1 -t write 192.168.5.6  #都是用主机内存, 完成1字节的远程内存写操作
    4. verbs 代码相对位置: fabtests/component/dmabuf-rdma/rdmabw-xe.c

    主机发给GPU设备 内存直接访问DMA libfabric示例

    1. libfabric GPU设备 -> host
    2. 服务端: ./fi-rdmabw-xe -m device #使用GPU设备的内存
    3. 客户端: ./fi-rdmabw-xe -m host -S 1 -t write 192.168.5.6  #用主机内存, 完成1字节的远程内存写操作
    4. libfabric 代码相对位置: fabtests/component/dmabuf-rdma/fi-rdmabw-xe.c

  • 相关阅读:
    《流畅的python》阅读笔记 - 第二章：数据结构
    羧甲基壳聚糖-丙硫菌唑水凝胶微球/壳聚糖/磷酸甘油盐温敏性水凝胶/石墨烯壳聚糖复合水凝胶
    Spring的IOC和AOP，学不会Spring的，你看我的文章
    [C++ 网络协议] 多线程服务器端
    【学习笔记】《Python深度学习》第一章：什么是深度学习
    Latent Topic-aware Multi-Label Classification
    BLIP-2小结
    一个月速刷leetcodeHOT100 day13 二叉树结构 以及相关简单题
    基于SpringSecurity的@PreAuthorize实现自定义权限校验方法
    碳中和&专利创新专题：各省市县专利面板(原始文件)、低碳专利授权数等多指标数据
  • 原文地址：https://blog.csdn.net/weixin_43778179/article/details/134524691