RDMA编程实践-SEND-RECEICVE原语应用

RDMA编程实践

本文描述了RDMA编程过程中的SEND-RECEIVE双边原语的代码实现。包含多个版本，1、client向server发送消息，server回复client收到消息(ACK)，然后两边断开连接。2、server端循环等待客户端建立连接，client发送一次消息后，双方断开连接。3、server端循环等待客户端建立连接，一旦建立，client端可以一直向server端发送消息，直到发送消息为disconnect，server和client断开链接，但是server此时仍然可以等待别的client发送消息。
代码基于代码基于send-receive样例实现。关于代码注释，可以参考代码解释：
Makefile文件、会编译当前目录下的所有.c文件：

.PHONY: all clean

CC := gcc
CFLAGS := -Wall -g
LDLIBS := -lrdmacm -libverbs -lpthread -g

SRCS := $(wildcard *.c)
APPS := $(SRCS:.c=)

all: $(APPS)

%: %.c
	$(CC) $(CFLAGS) $< -o $@ $(LDLIBS)

clean:
	rm -f $(APPS)

version1 客户端-服务端消息一次传递

在这个阶段，我们希望能实现下面这样一个场景。client与server端相连接，client端能够发送一条消息给server，server收到该条消息之后恢复一条消息给client端表示我已经确认收到。之后两者断开连接。

代码：

// client1.c
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <netdb.h>
#include <errno.h>
#include <getopt.h>
#include <rdma/rdma_cma.h>
#include <rdma/rdma_verbs.h>

static const char *server = "10.10.10.1";
static const char *port = "7471";

static struct rdma_cm_id *id;
static struct ibv_mr *mr, *send_mr;
static int send_flags;
static uint8_t send_msg[16];
static uint8_t recv_msg[16];

static int run(void)
{
	struct rdma_addrinfo hints, *res;
	struct ibv_qp_init_attr attr;
	struct ibv_wc wc;
	int ret;

	memset(&hints, 0, sizeof hints);
	hints.ai_port_space = RDMA_PS_TCP;
	ret = rdma_getaddrinfo(server, port, &hints, &res);
	if (ret) {
		printf("rdma_getaddrinfo: %s\n", gai_strerror(ret));
		goto out;
	}

	memset(&attr, 0, sizeof attr);
	attr.cap.max_send_wr = attr.cap.max_recv_wr = 1;
	attr.cap.max_send_sge = attr.cap.max_recv_sge = 1;
	attr.cap.max_inline_data = 16;
	attr.qp_context = id;
	attr.sq_sig_all = 1;
	ret = rdma_create_ep(&id, res, NULL, &attr);
	// Check to see if we got inline data allowed or not
	if (attr.cap.max_inline_data >= 16)
		send_flags = IBV_SEND_INLINE;
	else
		printf("rdma_client: device doesn't support IBV_SEND_INLINE, "
		       "using sge sends\n");

	if (ret) {
		perror("rdma_create_ep");
		goto out_free_addrinfo;
	}

	mr = rdma_reg_msgs(id, recv_msg, 16);
	if (!mr) {
		perror("rdma_reg_msgs for recv_msg");
		ret = -1;
		goto out_destroy_ep;
	}
	if ((send_flags & IBV_SEND_INLINE) == 0) {
		send_mr = rdma_reg_msgs(id, send_msg, 16);
		if (!send_mr) {
			perror("rdma_reg_msgs for send_msg");
			ret = -1;
			goto out_dereg_recv;
		}
	}

	ret = rdma_post_recv(id, NULL, recv_msg, 16, mr);
	if (ret) {
		perror("rdma_post_recv");
		goto out_dereg_send;
	}

	ret = rdma_connect(id, NULL);
	if (ret) {
		perror("rdma_connect");
		goto out_dereg_send;
	}


	printf("client send: %s\n", (char *)send_msg);
	ret = rdma_post_send(id, NULL, send_msg, 16, send_mr, send_flags);
	if (ret) {
		perror("rdma_post_send");
		goto out_disconnect;
	}

	while ((ret = rdma_get_send_comp(id, &wc)) == 0);
	if (ret < 0) {
		perror("rdma_get_send_comp");
		goto out_disconnect;
	}

	while ((ret = rdma_get_recv_comp(id, &wc)) == 0);
	if (ret < 0)
		perror("rdma_get_recv_comp");
	else
		ret = 0;
	
	printf("client received: %s\n", (char *) recv_msg);

out_disconnect:
	rdma_disconnect(id);
out_dereg_send:
	if ((send_flags & IBV_SEND_INLINE) == 0)
		rdma_dereg_mr(send_mr);
out_dereg_recv:
	rdma_dereg_mr(mr);
out_destroy_ep:
	rdma_destroy_ep(id);
out_free_addrinfo:
	rdma_freeaddrinfo(res);
out:
	return ret;
}

int main(int argc, char **argv)
{
	int ret;

	char *s = "hello world";
	// printf("client send: %s\n", s);
	memcpy(send_msg, s , strlen(s));

	printf("rdma_client: start\n");
	ret = run();
	printf("rdma_client: end %d\n", ret);
	return ret;
}

server端代码

// server1.c
/*
 * Copyright (c) 2005-2009 Intel Corporation.  All rights reserved.
 *
 * This software is available to you under the OpenIB.org BSD license
 * below:
 *
 *     Redistribution and use in source and binary forms, with or
 *     without modification, are permitted provided that the following
 *     conditions are met:
 *
 *      - Redistributions of source code must retain the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer.
 *
 *      - Redistributions in binary form must reproduce the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer in the documentation and/or other materials
 *        provided with the distribution.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AWV
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <getopt.h>
#include <netdb.h>
#include <rdma/rdma_cma.h>
#include <rdma/rdma_verbs.h>

static const char *server = "0.0.0.0";
static const char *port = "7471";

static struct rdma_cm_id *listen_id, *id;
static struct ibv_mr *mr, *send_mr;
static int send_flags;
static uint8_t send_msg[16];
static uint8_t recv_msg[16];

static int run(void)
{
	struct rdma_addrinfo hints, *res;
	struct ibv_qp_init_attr init_attr;
	struct ibv_qp_attr qp_attr;
	struct ibv_wc wc;
	int ret;

	memset(&hints, 0, sizeof hints);
	hints.ai_flags = RAI_PASSIVE;
	hints.ai_port_space = RDMA_PS_TCP;
	ret = rdma_getaddrinfo(server, port, &hints, &res);
	if (ret) {
		printf("rdma_getaddrinfo: %s\n", gai_strerror(ret));
		return ret;
	}

	memset(&init_attr, 0, sizeof init_attr);
	init_attr.cap.max_send_wr = init_attr.cap.max_recv_wr = 1;
	init_attr.cap.max_send_sge = init_attr.cap.max_recv_sge = 1;
	init_attr.cap.max_inline_data = 16;
	init_attr.sq_sig_all = 1;
	ret = rdma_create_ep(&listen_id, res, NULL, &init_attr);
	if (ret) {
		perror("rdma_create_ep");
		goto out_free_addrinfo;
	}

	ret = rdma_listen(listen_id, 0);
	if (ret) {
		perror("rdma_listen");
		goto out_destroy_listen_ep;
	}

	ret = rdma_get_request(listen_id, &id);
	if (ret) {
		perror("rdma_get_request");
		goto out_destroy_listen_ep;
	}

	memset(&qp_attr, 0, sizeof qp_attr);
	memset(&init_attr, 0, sizeof init_attr);
	ret = ibv_query_qp(id->qp, &qp_attr, IBV_QP_CAP,
			   &init_attr);
	if (ret) {
		perror("ibv_query_qp");
		goto out_destroy_accept_ep;
	}
	if (init_attr.cap.max_inline_data >= 16)
		send_flags = IBV_SEND_INLINE;
	else
		printf("rdma_server: device doesn't support IBV_SEND_INLINE, "
		       "using sge sends\n");

	mr = rdma_reg_msgs(id, recv_msg, 16);
	if (!mr) {
		ret = -1;
		perror("rdma_reg_msgs for recv_msg");
		goto out_destroy_accept_ep;
	}
	if ((send_flags & IBV_SEND_INLINE) == 0) {
		send_mr = rdma_reg_msgs(id, send_msg, 16);
		if (!send_mr) {
			ret = -1;
			perror("rdma_reg_msgs for send_msg");
			goto out_dereg_recv;
		}
	}

	ret = rdma_post_recv(id, NULL, recv_msg, 16, mr);
	if (ret) {
		perror("rdma_post_recv");
		goto out_dereg_send;
	}

	ret = rdma_accept(id, NULL);
	if (ret) {
		perror("rdma_accept");
		goto out_dereg_send;
	}

	while ((ret = rdma_get_recv_comp(id, &wc)) == 0);
	if (ret < 0) {
		perror("rdma_get_recv_comp");
		goto out_disconnect;
	}

    printf("server received: %s\n" , (char *)recv_msg);
    
    char *s = "ACK";
    memcpy(send_msg, s, strlen(s));
    printf("server send: %s\n", (char *)send_msg);

	ret = rdma_post_send(id, NULL, send_msg, 16, send_mr, send_flags);
	if (ret) {
		perror("rdma_post_send");
		goto out_disconnect;
	}

	while ((ret = rdma_get_send_comp(id, &wc)) == 0);
	if (ret < 0)
		perror("rdma_get_send_comp");
	else
		ret = 0;

out_disconnect:
	rdma_disconnect(id);
out_dereg_send:
	if ((send_flags & IBV_SEND_INLINE) == 0)
		rdma_dereg_mr(send_mr);
out_dereg_recv:
	rdma_dereg_mr(mr);
out_destroy_accept_ep:
	rdma_destroy_ep(id);
out_destroy_listen_ep:
	rdma_destroy_ep(listen_id);
out_free_addrinfo:
	rdma_freeaddrinfo(res);
	return ret;
}

int main(int argc, char **argv)
{
	int ret;

	printf("rdma_server: start\n");
	ret = run();
	printf("rdma_server: end %d\n", ret);
	return ret;
}

首先make编译完之后，在server端执行 ./server1，然后在客户端执行./client1
运行结果：
在这里插入图片描述

可以看到 client向server发送了hello world，server收到之后打印出来并回复给client端ACK消息，client收到之后并打印。最后双方断开连接，完成！

version2-客户端发送一次，服务端循环等待

client2的代码跟上面一样，server2代码不一样。
server2的逻辑：在run函数进来之后记录一个connect点，当远程客户端发送完信息后，释放连接的资源，跳转到connect阶段准备让下一个client连接。

/*
 * Copyright (c) 2005-2009 Intel Corporation.  All rights reserved.
 *
 * This software is available to you under the OpenIB.org BSD license
 * below:
 *
 *     Redistribution and use in source and binary forms, with or
 *     without modification, are permitted provided that the following
 *     conditions are met:
 *
 *      - Redistributions of source code must retain the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer.
 *
 *      - Redistributions in binary form must reproduce the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer in the documentation and/or other materials
 *        provided with the distribution.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AWV
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <getopt.h>
#include <netdb.h>
#include <rdma/rdma_cma.h>
#include <rdma/rdma_verbs.h>


#define N 100
#define MAX_CAP 32

static const char *server = "0.0.0.0";
static const char *port = "7471";

static struct rdma_cm_id *listen_id, *id;
static struct ibv_mr *mr, *send_mr;
static int send_flags;
static uint8_t send_msg[MAX_CAP];
static uint8_t recv_msg[MAX_CAP];

static int run(void)
{
	struct rdma_addrinfo hints, *res;
	struct ibv_qp_init_attr init_attr;
	struct ibv_qp_attr qp_attr;
	struct ibv_wc wc;
	int ret;

    while(1)
    {

        memset(&hints, 0, sizeof hints);
        hints.ai_flags = RAI_PASSIVE;
        hints.ai_port_space = RDMA_PS_TCP;
        ret = rdma_getaddrinfo(server, port, &hints, &res);
        if (ret) {
            printf("rdma_getaddrinfo: %s\n", gai_strerror(ret));
            return ret;
        }

        memset(&init_attr, 0, sizeof init_attr);
        init_attr.cap.max_send_wr = init_attr.cap.max_recv_wr = N;
        init_attr.cap.max_send_sge = init_attr.cap.max_recv_sge = 1;
        init_attr.cap.max_inline_data = MAX_CAP;
        init_attr.sq_sig_all = 1;

        ret = rdma_create_ep(&listen_id, res, NULL, &init_attr);
        if (ret) {
            perror("rdma_create_ep");
            goto out_free_addrinfo;
        }

        ret = rdma_listen(listen_id, 0);
        if (ret) {
            perror("rdma_listen");
            goto out_destroy_listen_ep;
        }

        ret = rdma_get_request(listen_id, &id);
        if (ret) {
            perror("rdma_get_request");
            goto out_destroy_listen_ep;
        }

        memset(&qp_attr, 0, sizeof qp_attr);
        memset(&init_attr, 0, sizeof init_attr);
        ret = ibv_query_qp(id->qp, &qp_attr, IBV_QP_CAP,
                &init_attr);
        if (ret) {
            perror("ibv_query_qp");
            goto out_destroy_accept_ep;
        }
        if (init_attr.cap.max_inline_data >= MAX_CAP)
            send_flags = IBV_SEND_INLINE;
        else
            printf("rdma_server: device doesn't support IBV_SEND_INLINE, "
                "using sge sends\n");
        mr = rdma_reg_msgs(id, recv_msg, N);
        if (!mr) {
            ret = -1;
            perror("rdma_reg_msgs for recv_msg");
            goto out_destroy_accept_ep;
        }
        if ((send_flags & IBV_SEND_INLINE) == 0) {
            send_mr = rdma_reg_msgs(id, send_msg, MAX_CAP);
            if (!send_mr) {
                ret = -1;
                perror("rdma_reg_msgs for send_msg");
                goto out_dereg_recv;
            }
        }   
        

        ret = rdma_accept(id, NULL);
        if (ret) {
            perror("rdma_accept");
            goto out_dereg_send;
        }

        
        memset(recv_msg, 0 , sizeof recv_msg);
        memset(send_msg, 0 , sizeof send_msg);
        ret = rdma_post_recv(id, NULL, recv_msg, MAX_CAP, mr);
        if (ret) {
            perror("rdma_post_recv");
            goto out_dereg_send;
        }
        while ((ret = rdma_get_recv_comp(id, &wc)) == 0);
        if (ret < 0) {
            perror("rdma_get_recv_comp");
            goto out_disconnect;
        }
        
        printf("server received: %s\n", (char *)recv_msg);
        memcpy(send_msg, recv_msg, sizeof(recv_msg));
        
        ret = rdma_post_send(id, NULL, send_msg, MAX_CAP, send_mr, send_flags);
        if (ret) {
            perror("rdma_post_send");
            goto out_disconnect;
        }
        while ((ret = rdma_get_send_comp(id, &wc)) == 0); // 确认对方已经收到 对方会发送ack
        if (ret < 0)
            perror("rdma_get_send_comp");
        else
            ret = 0;
        
        rdma_disconnect(id);
        if ((send_flags & IBV_SEND_INLINE) == 0)
            rdma_dereg_mr(send_mr);
        rdma_dereg_mr(mr);
        rdma_destroy_ep(id);
        rdma_destroy_ep(listen_id);
        rdma_freeaddrinfo(res);  
    }
         
    

out_disconnect:
    rdma_disconnect(id);
out_dereg_send:
    if ((send_flags & IBV_SEND_INLINE) == 0)
        rdma_dereg_mr(send_mr);
out_dereg_recv:
    rdma_dereg_mr(mr);
out_destroy_accept_ep:
    rdma_destroy_ep(id);
out_destroy_listen_ep:
    rdma_destroy_ep(listen_id);
out_free_addrinfo:
    rdma_freeaddrinfo(res);  

	return ret;
}

int main(int argc, char **argv)
{
	int ret;
	printf("rdma_server: start\n");
	ret = run();
	printf("rdma_server: end %d\n", ret);
	return ret;
}

运行结果：
在这里插入图片描述

在这里插入图片描述

可以看到客户端发送一次消息之后便结束了，服务端却一直等待连接，直到按下ctrl+c。

version3-客户端循环发送，服务端循环等待，一次连接

和上述版本2不同的时候，这里client和server只连接一次，然后可以多次发送消息。直到client发送的消息为disconnect

// client3.c
/*
 * Copyright (c) 2010 Intel Corporation.  All rights reserved.
 *
 * This software is available to you under the OpenIB.org BSD license
 * below:
 *
 *     Redistribution and use in source and binary forms, with or
 *     without modification, are permitted provided that the following
 *     conditions are met:
 *
 *      - Redistributions of source code must retain the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer.
 *
 *      - Redistributions in binary form must reproduce the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer in the documentation and/or other materials
 *        provided with the distribution.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AWV
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <netdb.h>
#include <errno.h>
#include <getopt.h>
#include <rdma/rdma_cma.h>
#include <rdma/rdma_verbs.h>

#define N 100
#define MAX_CAP 32

static const char *server = "10.10.10.1";
static const char *port = "7471";

static struct rdma_cm_id *id;
static struct ibv_mr *mr, *send_mr;
static int send_flags;
static uint8_t send_msg[MAX_CAP];
static uint8_t recv_msg[MAX_CAP];

static int run(void)
{
	struct rdma_addrinfo hints, *res;
	struct ibv_qp_init_attr attr;
	struct ibv_wc wc;
	int ret;

	memset(&hints, 0, sizeof hints);
	hints.ai_port_space = RDMA_PS_TCP;
	ret = rdma_getaddrinfo(server, port, &hints, &res);
	if (ret) {
		printf("rdma_getaddrinfo: %s\n", gai_strerror(ret));
		goto out;
	}
    memset(&attr, 0, sizeof attr);
    attr.cap.max_send_wr = attr.cap.max_recv_wr = 5;
    attr.cap.max_send_sge = attr.cap.max_recv_sge = 1;
    attr.cap.max_inline_data = MAX_CAP;
    attr.qp_context = id;
    attr.sq_sig_all = 1;
    ret = rdma_create_ep(&id, res, NULL, &attr);
    // Check to see if we got inline data allowed or not
    if (attr.cap.max_inline_data >= MAX_CAP)
        send_flags = IBV_SEND_INLINE;
    else
        printf("rdma_client: device doesn't support IBV_SEND_INLINE, "
            "using sge sends\n");

    if (ret) {
        perror("rdma_create_ep");
        goto out_free_addrinfo;
    }

    mr = rdma_reg_msgs(id, recv_msg, MAX_CAP);
    if (!mr) {
        perror("rdma_reg_msgs for recv_msg");
        ret = -1;
        goto out_destroy_ep;
    }
    if ((send_flags & IBV_SEND_INLINE) == 0) {
        send_mr = rdma_reg_msgs(id, send_msg, MAX_CAP);
        if (!send_mr) {
            perror("rdma_reg_msgs for send_msg");
            ret = -1;
            goto out_dereg_recv;
        }
    }

    // ret = rdma_post_recv(id, NULL, recv_msg, 16, mr);
    // if (ret) {
    //     perror("rdma_post_recv");
    //     goto out_dereg_send;
    // }

    // printf("123\n");

    ret = rdma_connect(id, NULL);
    if (ret) {
        perror("rdma_connect");
        goto out_dereg_send;
    }

    while(1)
    {
        // sleep(5);
        memset(recv_msg, 0 , sizeof recv_msg);
        memset(send_msg, 0 , sizeof send_msg);
        printf("input send message: ");
        scanf("%s", send_msg);
        getchar();
        
        ret = rdma_post_recv(id, NULL, recv_msg, MAX_CAP, mr);
        if (ret) {
            perror("rdma_post_recv");
            goto out_dereg_send;
        }

        ret = rdma_post_send(id, NULL, send_msg, MAX_CAP, send_mr, send_flags);
        if (ret) {
            perror("rdma_post_send");
            goto out_disconnect;
        }

        while ((ret = rdma_get_send_comp(id, &wc)) == 0);
        if (ret < 0) {
            perror("rdma_get_send_comp");
            goto out_disconnect;
        }

        while ((ret = rdma_get_recv_comp(id, &wc)) == 0);
        if (ret < 0)
            perror("rdma_get_recv_comp");
        else
            ret = 0;
        
        if(strcmp((char*)send_msg,"disconnect") == 0)
        {
            printf("disconnect\n");
            goto out_disconnect;
        }
        else
        {
            printf("%s\n", recv_msg);
        }
            
    }

out_disconnect:
    rdma_disconnect(id);
out_dereg_send:
    if ((send_flags & IBV_SEND_INLINE) == 0)
            rdma_dereg_mr(send_mr);
out_dereg_recv:
    rdma_dereg_mr(mr);
out_destroy_ep:
    rdma_destroy_ep(id);
out_free_addrinfo:
    rdma_freeaddrinfo(res);
out:
    return ret;

}



int main(int argc, char **argv)
{
	int ret;
    //memcpy(send_msg, argv[1], 50);
	// while ((op = getopt(argc, argv, "s:p:")) != -1) {
	// 	switch (op) {
	// 	case 's':
	// 		server = optarg;
	// 		break;
	// 	case 'p':
	// 		port = optarg;
	// 		break;
	// 	default:
	// 		printf("usage: %s\n", argv[0]);
	// 		printf("\t[-s server_address]\n");
	// 		printf("\t[-p port_number]\n");
	// 		exit(1);
	// 	}
	// }

	printf("rdma_client: start\n");
	ret = run();
	printf("rdma_client: end %d\n", ret);
    
	return ret;
}

// server3.c
/*
 * Copyright (c) 2005-2009 Intel Corporation.  All rights reserved.
 *
 * This software is available to you under the OpenIB.org BSD license
 * below:
 *
 *     Redistribution and use in source and binary forms, with or
 *     without modification, are permitted provided that the following
 *     conditions are met:
 *
 *      - Redistributions of source code must retain the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer.
 *
 *      - Redistributions in binary form must reproduce the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer in the documentation and/or other materials
 *        provided with the distribution.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AWV
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <getopt.h>
#include <netdb.h>
#include <rdma/rdma_cma.h>
#include <rdma/rdma_verbs.h>


#define N 100
#define MAX_CAP 32

static const char *server = "0.0.0.0";
static const char *port = "7471";

static struct rdma_cm_id *listen_id, *id;
static struct ibv_mr *mr, *send_mr;
static int send_flags;
static uint8_t send_msg[MAX_CAP];
static uint8_t recv_msg[MAX_CAP];

static int run(void)
{
	struct rdma_addrinfo hints, *res;
	struct ibv_qp_init_attr init_attr;
	struct ibv_qp_attr qp_attr;
	struct ibv_wc wc;
	int ret;

connect:
	memset(&hints, 0, sizeof hints);
	hints.ai_flags = RAI_PASSIVE;
	hints.ai_port_space = RDMA_PS_TCP;
	ret = rdma_getaddrinfo(server, port, &hints, &res);
	if (ret) {
		printf("rdma_getaddrinfo: %s\n", gai_strerror(ret));
		return ret;
	}

	memset(&init_attr, 0, sizeof init_attr);
	init_attr.cap.max_send_wr = init_attr.cap.max_recv_wr = N;
	init_attr.cap.max_send_sge = init_attr.cap.max_recv_sge = 1;
	init_attr.cap.max_inline_data = MAX_CAP;
	init_attr.sq_sig_all = 1;

	ret = rdma_create_ep(&listen_id, res, NULL, &init_attr);
	if (ret) {
		perror("rdma_create_ep");
		goto out_free_addrinfo;
	}

	ret = rdma_listen(listen_id, 0);
	if (ret) {
		perror("rdma_listen");
		goto out_destroy_listen_ep;
	}

    ret = rdma_get_request(listen_id, &id);
    if (ret) {
        perror("rdma_get_request");
        goto out_destroy_listen_ep;
    }

    memset(&qp_attr, 0, sizeof qp_attr);
    memset(&init_attr, 0, sizeof init_attr);
    ret = ibv_query_qp(id->qp, &qp_attr, IBV_QP_CAP,
            &init_attr);
    if (ret) {
        perror("ibv_query_qp");
        goto out_destroy_accept_ep;
    }
    if (init_attr.cap.max_inline_data >= MAX_CAP)
        send_flags = IBV_SEND_INLINE;
    else
        printf("rdma_server: device doesn't support IBV_SEND_INLINE, "
            "using sge sends\n");
    mr = rdma_reg_msgs(id, recv_msg, N);
    if (!mr) {
        ret = -1;
        perror("rdma_reg_msgs for recv_msg");
        goto out_destroy_accept_ep;
    }
    if ((send_flags & IBV_SEND_INLINE) == 0) {
        send_mr = rdma_reg_msgs(id, send_msg, MAX_CAP);
        if (!send_mr) {
            ret = -1;
            perror("rdma_reg_msgs for send_msg");
            goto out_dereg_recv;
        }
    }   
    

    ret = rdma_accept(id, NULL);
    if (ret) {
        perror("rdma_accept");
        goto out_dereg_send;
    }

    while (1) {
        memset(recv_msg, 0 , sizeof recv_msg);
        memset(send_msg, 0 , sizeof send_msg);
        ret = rdma_post_recv(id, NULL, recv_msg, MAX_CAP, mr);
        if (ret) {
            perror("rdma_post_recv");
            goto out_dereg_send;
        }
        while ((ret = rdma_get_recv_comp(id, &wc)) == 0);
        if (ret < 0) {
            perror("rdma_get_recv_comp");
            goto out_disconnect;
        }
        
        char *s = (char *)recv_msg;
        int total_length = strlen("server get ") + strlen(s); // 加1是为了存储字符串结束符'\0'
        char *recv_str = (char *)malloc(total_length);  // 分配足够的空间
        strcpy(recv_str, "server get ");
        strcat(recv_str, s);
        //printf("%s\n", recv_str);
        memcpy(send_msg, recv_str, strlen(recv_str));
        
        ret = rdma_post_send(id, NULL, send_msg, MAX_CAP, send_mr, send_flags);
        if (ret) {
            perror("rdma_post_send");
            goto out_disconnect;
        }

        if(strcmp((char*)recv_msg,"disconnect") == 0)
        {
            //printf("%s\n",recv_msg);
            printf("client disconnect\n");
            rdma_disconnect(id);
            if ((send_flags & IBV_SEND_INLINE) == 0)
                rdma_dereg_mr(send_mr);
            rdma_dereg_mr(mr);
            rdma_destroy_ep(id);
            rdma_destroy_ep(listen_id);
            rdma_freeaddrinfo(res);  
            //goto out_disconnect;
            goto connect;
            
        }
        else
        {
            printf("%s\n", recv_msg);
        }
        
        // while ((ret = rdma_get_send_comp(id, &wc)) == 0); // 确认对方已经收到 对方发送ack
        // printf("after send\n");
        // if (ret < 0)
        //     perror("rdma_get_send_comp");
        // else
		//     ret = 0;
         
    }

out_disconnect:
    rdma_disconnect(id);
out_dereg_send:
    if ((send_flags & IBV_SEND_INLINE) == 0)
        rdma_dereg_mr(send_mr);
out_dereg_recv:
    rdma_dereg_mr(mr);
out_destroy_accept_ep:
    rdma_destroy_ep(id);
out_destroy_listen_ep:
    rdma_destroy_ep(listen_id);
out_free_addrinfo:
    rdma_freeaddrinfo(res);  

	return ret;
}

int main(int argc, char **argv)
{
	int op, ret;

	while ((op = getopt(argc, argv, "s:p:")) != -1) {
		switch (op) {
		case 's':
			server = optarg;
			break;
		case 'p':
			port = optarg;
			break;
		default:
			printf("usage: %s\n", argv[0]);
			printf("\t[-s server_address]\n");
			printf("\t[-p port_number]\n");
			exit(1);
		}
	}

	printf("rdma_server: start\n");
	ret = run();
	printf("rdma_server: end %d\n", ret);
	return ret;
}

运行结果：

在这里插入图片描述

总结：

本文实现了rdma中send-receive双边原语的三种需求版本，从单次发送到两者都能多次发送。理解其中的代码逻辑，想要发送消息之前对端得创建一个recv队列用来接收消息。发送完了有一个发送完成队列，接收完了也有一个接收完成队列。最后双方断开连接需要一起断开，不能某一方执行disconnect另一方不执行。本次实验有一个关键点：

while ((ret = rdma_get_send_comp(id, &wc)) == 0)

这一行代码是等待发送成功，发送成功之后，对方会给一个隐式信息表示我已经收到。这里耗费得时间比较长一点，在版本3中，如果不注释掉，在server端的receive队列还没有建立好，这就导致client发送了消息，server还没有收到，双方就陷入了死循环中。