RE: IB atomic operations

["Bharath Ramesh" <bramesh-PjAqaU27lzQ@public.gmane.org>]

[-- Attachment #1.1: Type: text/plain, Size: 404 bytes --]

> -----Original Message-----
> From: frank zago [mailto:fzago-klaOcWyJdxkshyMvu7JE4pqQE7yCjDx5@public.gmane.org]
> 
> Sorry, I don't see what's wrong.

Could there be a reason why it would fail on certain nodes but run on other
nodes. I picked two other nodes and ran the attached test code. It runs fine
there without any issues, the hardware is same across all nodes of the
cluster.

Regards,

Bharath

[-- Attachment #1.2: fetch_add.c --]
[-- Type: application/octet-stream, Size: 13005 bytes --]

#include <getopt.h>
#include <infiniband/verbs.h>
#include <netdb.h>
#include <netinet/in.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>

#define IB_MAX_CQE 1024
#define IB_MAX_RD_ATOMIC 8
#define IB_MAX_WR 1024
#define IB_MAX_SGE 8
#define IB_PORT 1
#define IB_RETRY_CNT 7
#define IB_RNR_RETRY 7
#define IB_RNR_TIMER 12
#define IB_TIMEOUT 14
#define TCP_MSG_SIZE 48

void client (char *);
void server (void);
static int connect_ibverbs_qp (struct ibv_qp *, int, int);
static struct ibv_qp *create_ibverbs_qp (void);
static void query_ibverbs_qp (struct ibv_qp *);
static void ibverbs_exit (void);
static int ibverbs_init (void);
static void usage (const char *);

static int count = 100;
static uint64_t data = 0;
static struct ibv_cq *ib_cq = NULL;
static struct ibv_context *ib_ctx = NULL;
static uint16_t ib_lid = 0;
static struct ibv_mr *ib_mr = NULL;
static enum ibv_mtu ib_mtu = 0;
static struct ibv_pd *ib_pd = NULL;
static struct ibv_qp *ib_qp = NULL;

int
main (int argc, char *argv[])
{
	int c = 0;
	struct option opts[] = {
		{.name = "count", .has_arg = 1, .val = 'c'},
		{.name = "help", .has_arg = 0, .val = 'h'},
		{.name = NULL, .has_arg = 0, .val = 0}
	};
	char *host = NULL;

	while (c != EOF) {
		c = getopt_long (argc, argv, "c:h", opts, NULL);
		switch (c) {
			case EOF:
				break;

			case 'c':
				count = atoi (optarg);
				break;

			case '?':
			case 'h':
			default:
				usage (argv[0]);
		}
	}

	if (optind == (argc - 1)) {
		host = strdup (argv[optind]);
	} else if (optind < argc) {
		usage (argv[0]);
	}

	if (ibverbs_init () != 0) {
		printf ("ERROR: ibverbs_init failed.\n");
		return EXIT_FAILURE;
	}

	if (host == NULL)
		server ();
	else
		client (host);

	ibverbs_exit ();
	if (host != NULL)
		free (host);

	return EXIT_SUCCESS;
}

void
client (char *host)
{
	struct hostent *he;
	int lid, qpn, sock;
	char msg[TCP_MSG_SIZE];
	struct sockaddr_in sock_addr;

	ib_qp = create_ibverbs_qp ();
	if (ib_qp == NULL) {
		printf ("ERROR: create_ibverbs_qp failed.\n");
		return;
	}

	if ((sock = socket (PF_INET, SOCK_STREAM, 0)) == -1) {
		printf ("ERROR: socket failed.\n");
		return;
	}

	if ((he = gethostbyname (host)) == NULL) {
		printf ("ERROR: gethostbyname failed.\n");
		close (sock);
		return;
	}

	memset (&sock_addr, 0, sizeof (struct sockaddr_in));
	sock_addr.sin_family = AF_INET;
	sock_addr.sin_addr = *((struct in_addr *) he->h_addr);
	sock_addr.sin_port = htons (23380);
	if (connect (sock, (struct sockaddr *) &sock_addr,
				sizeof (struct sockaddr_in)) == -1) {
		printf ("ERROR: connect failed.\n");
		close (sock);
		return;
	}

	memset (msg, 0, sizeof (char) * TCP_MSG_SIZE);
	sprintf (msg, "%08x:%08x:%08x:%016lx", ib_lid, ib_qp->qp_num,
			ib_mr->rkey, (uint64_t) &data);
	if (write (sock, msg, sizeof (char) * TCP_MSG_SIZE)
			!=  sizeof (char) * TCP_MSG_SIZE) {
		printf ("ERROR: write failed.\n");
		close (sock);
		return;
	}

	if (read (sock, msg, sizeof (char) * TCP_MSG_SIZE)
			!=  sizeof (char) * TCP_MSG_SIZE) {
		printf ("ERROR: read failed.\n");
		close (sock);
		return;
	}

	sscanf (msg, "%x:%x", &lid, &qpn);
	if (connect_ibverbs_qp (ib_qp, lid, qpn) != 0) {
		printf ("ERROR: connect_ibverbs_qp failed.\n");
		close (sock);
		return;
	}

	if (write (sock, msg, sizeof (char) * TCP_MSG_SIZE)
			!=  sizeof (char) * TCP_MSG_SIZE) {
		printf ("ERROR: write failed.\n");
		close (sock);
		return;
	}

	if (read (sock, msg, sizeof (char) * TCP_MSG_SIZE)
			!=  sizeof (char) * TCP_MSG_SIZE) {
		printf ("ERROR: read failed.\n");
		close (sock);
		return;
	}

	printf ("data: %lu\n", data);
	if (sock != -1)
		close (sock);

	return;
}

void
server (void)
{
	int c, cli, i, j, k, lid, qpn, ret, sock;
	char msg[TCP_MSG_SIZE];
	uint32_t rkey;
	struct ibv_sge sge;
	struct sockaddr_in sock_addr;
	uint64_t vaddr;
	int val = 1;
	struct ibv_wc wc;
	struct ibv_send_wr wr, *wr_bad;

	ib_qp = create_ibverbs_qp ();
	if (ib_qp == NULL) {
		printf ("ERROR: create_ibverbs_qp failed.\n");
		return;
	}

	if ((sock = socket (PF_INET, SOCK_STREAM, 0)) == -1) {
		printf ("ERROR: socket failed.\n");
		return;
	}

	if (setsockopt (sock, SOL_SOCKET, SO_REUSEADDR, &val, sizeof (int))
			== -1) {
		printf ("ERROR: setsockopt failed.\n");
		close (sock);
		return;
	}

	memset (&sock_addr, 0, sizeof (struct sockaddr_in));
	sock_addr.sin_family = AF_INET;
	sock_addr.sin_addr.s_addr = htonl (INADDR_ANY);
	sock_addr.sin_port = htons (23380);
	if (bind (sock, (struct sockaddr *) &sock_addr,
				sizeof (struct sockaddr_in)) == -1) {
		printf ("ERROR: bind failed.\n");
		close (sock);
		return;
	}

	if (listen (sock, 2) == -1) {
		printf ("ERROR: listen failed.\n");
		close (sock);
		return;
	}

	if ((cli = accept (sock, NULL, NULL)) == -1) {
		printf ("ERROR: accept failed.\n");
		close (sock);
		return;
	}

	if (read (cli, msg, sizeof (char) * TCP_MSG_SIZE)
			!=  sizeof (char) * TCP_MSG_SIZE) {
		printf ("ERROR: read failed.\n");
		close (cli);
		close (sock);
		return;
	}

	sscanf (msg, "%x:%x:%x:%lx", &lid, &qpn, &rkey, &vaddr);
	memset (msg, 0, sizeof (char) * TCP_MSG_SIZE);
	sprintf (msg, "%08x:%08x", ib_lid, ib_qp->qp_num);
	if (write (cli, msg, sizeof (char) * TCP_MSG_SIZE)
			!= sizeof (char) * TCP_MSG_SIZE) {
		printf ("ERROR: write failed.\n");
		close (cli);
		close (sock);
		return;
	}

	if (connect_ibverbs_qp (ib_qp, lid, qpn) != 0) {
		printf ("ERROR: connect_ibverbs_qp failed.\n");
		close (sock);
		return;
	}

	if (read (cli, msg, sizeof (char) * TCP_MSG_SIZE)
			!=  sizeof (char) * TCP_MSG_SIZE) {
		printf ("ERROR: read failed.\n");
		close (cli);
		close (sock);
		return;
	}

	sge.addr = (uintptr_t) &data;
	sge.length = sizeof (uint64_t);
	sge.lkey = ib_mr->lkey;
	wr.opcode = IBV_WR_ATOMIC_FETCH_AND_ADD;
	wr.send_flags = IBV_SEND_SIGNALED;
	wr.next = NULL;
	wr.num_sge = 1;
	wr.sg_list = &sge;
	wr.wr.atomic.remote_addr = vaddr;
	wr.wr.atomic.compare_add = 1;
	//wr.wr.atomic.swap = 0;
	wr.wr.atomic.rkey = rkey;
	for (i = 0, j = 0; i < count; i++) {
		wr.wr_id = (uint64_t) i + 1;
		ret = ibv_post_send (ib_qp, &wr, &wr_bad);
		if (ret != 0) {
			printf ("ERROR: ibv_post_send failed.\n");
			printf ("i: %d\n", i);
			return;
		}

		++j;
		if ((i % IB_MAX_RD_ATOMIC) == 0) {
			for (k = 0; k < j; k++) {
				do {
					c = ibv_poll_cq (ib_cq, 1, &wc);
				} while (c == 0);

				if (c < 0) {
					printf ("ERROR: ibv_poll_cq failed.\n");
					return;
				}

				if (wc.status != IBV_WC_SUCCESS) {
					printf ("ERROR: work request completion"
							" failed.\n");
					printf ("wc status: %d, wrid: %lu, "
							"vendor err: %d\n",
							wc.status, wc.wr_id,
							wc.vendor_err);
					printf ("i: %d, j: %d\n", i, j);
					query_ibverbs_qp (ib_qp);
					return;
				}
			}

			printf ("j: %d\n", j);
			j = 0;
		}
	}

	if (j != 0) {
		for (k = 0; k < j; k++) {
			do {
				c = ibv_poll_cq (ib_cq, 1, &wc);
			} while (c == 0);

			if (c < 0) {
				printf ("ERROR: ibv_poll_cq failed.\n");
				return;
			}

			if (wc.status != IBV_WC_SUCCESS) {
				printf ("ERROR: work request completion "
						"failed.\n");
				printf ("wc status: %d, wrid: %lu, vendor err: "
						"%d\n", wc.status, wc.wr_id,
						wc.vendor_err);
				printf ("i: %d, j: %d\n", i, j);
				query_ibverbs_qp (ib_qp);
				return;
			}
		}

		printf ("j: %d\n", j);
		j = 0;
	}

	if (write (cli, msg, sizeof (char) * TCP_MSG_SIZE)
			!= sizeof (char) * TCP_MSG_SIZE) {
		printf ("ERROR: write failed.\n");
		close (cli);
		close (sock);
		return;
	}

	printf ("data: %lu\n", data);
	if (cli != -1)
		close (cli);

	if (sock != -1)
		close (sock);

	return;
}

static int
connect_ibverbs_qp (struct ibv_qp *qp, int lid, int qpn)
{
       struct ibv_qp_attr attr;

        // Transition the QP to RTR.
        memset (&attr, 0, sizeof (struct ibv_qp_attr));
        attr.qp_state = IBV_QPS_RTR;
        attr.path_mtu = ib_mtu;
        attr.dest_qp_num = qpn;
        attr.rq_psn = 0;
        attr.max_dest_rd_atomic = IB_MAX_RD_ATOMIC;
        attr.min_rnr_timer = IB_RNR_TIMER;
        attr.ah_attr.is_global = 0;
        attr.ah_attr.dlid = lid;
        attr.ah_attr.sl = 0;
        attr.ah_attr.src_path_bits = 0;
        attr.ah_attr.port_num = IB_PORT;
        if (ibv_modify_qp (qp, &attr, IBV_QP_STATE | IBV_QP_AV |
                                IBV_QP_PATH_MTU | IBV_QP_DEST_QPN |
                                IBV_QP_RQ_PSN | IBV_QP_MAX_DEST_RD_ATOMIC |
                                IBV_QP_MIN_RNR_TIMER) != 0) {
                printf ("ERROR: Unable to modify QP to RTR state.\n");
                return -1;
        }

        // Transition the QP to RTS.
        attr.timeout = IB_TIMEOUT;
        attr.retry_cnt = IB_RETRY_CNT;
        attr.rnr_retry = IB_RNR_RETRY;
        attr.qp_state = IBV_QPS_RTS;
        attr.sq_psn = 0;
        attr.max_rd_atomic = IB_MAX_RD_ATOMIC;
        if (ibv_modify_qp (qp, &attr, IBV_QP_STATE | IBV_QP_SQ_PSN |
                                IBV_QP_MAX_QP_RD_ATOMIC | IBV_QP_RETRY_CNT |
                                IBV_QP_RNR_RETRY | IBV_QP_TIMEOUT) != 0) {
                printf ("ERROR: Unable to modify QP to RTS state.\n");
                return -1;
        }

        return 0;
}

static struct ibv_qp *
create_ibverbs_qp (void)
{
	struct ibv_qp_attr attr;
        struct ibv_qp_init_attr init_attr;
        struct ibv_qp *qp;

        memset (&init_attr, 0, sizeof (struct ibv_qp_init_attr));
        init_attr.recv_cq = ib_cq;
        init_attr.send_cq = ib_cq;
        init_attr.cap.max_recv_wr = IB_MAX_WR;
        init_attr.cap.max_send_wr = IB_MAX_WR;
        init_attr.cap.max_recv_sge = IB_MAX_SGE;
        init_attr.cap.max_send_sge = IB_MAX_SGE;
        init_attr.qp_type = IBV_QPT_RC;
        qp = ibv_create_qp (ib_pd, &init_attr);
        if (qp == NULL) {
                printf ("ERROR: Unable to create IB verbs QP.\n");
                return NULL;
        }

        // Transition QP to INIT state.
        attr.qp_state = IBV_QPS_INIT;
        attr.pkey_index = 0;
        attr.port_num = IB_PORT;
        attr.qp_access_flags = IBV_ACCESS_LOCAL_WRITE
		| IBV_ACCESS_REMOTE_ATOMIC | IBV_ACCESS_REMOTE_READ
		| IBV_ACCESS_REMOTE_WRITE;
        if (ibv_modify_qp (qp, &attr, IBV_QP_STATE | IBV_QP_PKEY_INDEX |
                                IBV_QP_PORT | IBV_QP_ACCESS_FLAGS) != 0) {
                printf ("ERROR: Unable to modify QP to INIT state.\n");
                if (ibv_destroy_qp (qp) != 0)
                        printf ("ERROR: Unable to destroy IB verbs QP.\n");

                return NULL;
        }

        return qp;
}

static void
query_ibverbs_qp (struct ibv_qp *qp)
{
	struct ibv_qp_attr attr;
	struct ibv_qp_init_attr init_attr;
	enum ibv_qp_attr_mask mask;
	int ret;

	mask = IBV_QP_STATE | IBV_QP_MAX_QP_RD_ATOMIC |
		IBV_QP_MAX_DEST_RD_ATOMIC;
	ret = ibv_query_qp (qp, &attr, mask, &init_attr);
	if (ret != 0) {
		printf ("ERROR: Unable to query QP.\n");
		printf ("Info: retval: %d\n", ret);
		return;
	}

	printf ("Info: QP state: %d\n", attr.qp_state);
	printf ("Info: Max QP RD Atomic: %d\n", attr.max_rd_atomic);
	printf ("Info: Max QP Dest RD Atomic: %d\n", attr.max_dest_rd_atomic);

	return;
}

static void
ibverbs_exit (void)
{
	if (ib_qp != NULL)
		ibv_destroy_qp (ib_qp);
	
	if (ib_mr != NULL)
		ibv_dereg_mr (ib_mr);

	if (ib_cq != NULL)
		ibv_destroy_cq (ib_cq);

	if (ib_pd != NULL)
		ibv_dealloc_pd (ib_pd);

	if (ib_ctx != NULL)
		ibv_close_device (ib_ctx);

	return;
}

static int
ibverbs_init (void)
{
	struct ibv_device *ib_dev, **ib_dev_list;
	int ndevs;
	struct ibv_port_attr port_attr;

	ib_dev_list = ibv_get_device_list (&ndevs);
	if (ndevs == 0) {
		printf ("ERROR: No IB device(s) found.\n");
		return -1;
	}

	ib_dev = ib_dev_list[0];
	ib_ctx = ibv_open_device (ib_dev);
	if (ib_ctx == NULL) {
		printf ("ERROR: ibv_open_device failed.\n");
		return -1;
	}

	if (ibv_query_port (ib_ctx, IB_PORT, &port_attr) != 0) {
		printf ("ERROR: ibv_query_device failed.\n");
		return -1;
	}

	ib_lid = port_attr.lid;
	ib_mtu = port_attr.active_mtu;
	ib_pd = ibv_alloc_pd (ib_ctx);
	if (ib_pd == NULL) {
		printf ("ERROR: ibv_alloc_pd failed.\n");
		return -1;
	}

	ib_cq = ibv_create_cq (ib_ctx, IB_MAX_CQE, NULL, NULL, 0);
	if (ib_cq == NULL) {
		printf ("ERROR: ibv_create_cq failed.\n");
		return -1;
	}

	ib_mr = ibv_reg_mr (ib_pd, &data, sizeof (uint64_t),
			IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_ATOMIC
			| IBV_ACCESS_REMOTE_READ | IBV_ACCESS_REMOTE_WRITE);
	if (ib_mr == NULL) {
		printf ("ERROR: ibv_reg_mr failed.\n");
		return -1;
	}

	if (ib_dev_list != NULL)
		ibv_free_device_list (ib_dev_list);

	return 0;
}

static void
usage (const char *name)
{
	printf ("usage:\n");
	printf ("\t%s [options]\t\tstart a server and wait for connection.\n",
			name);
	printf ("\t%s [options] <host>\tconnect to server at <host>.\n",
			name);
	printf ("\noptions:\n");
	printf ("\t-c, --count=<count>\tatomic counts up to <count>(default: "
			"100).\n");
	printf ("\t-h, --help\t\tprint this message and quit.\n");
	exit (EXIT_SUCCESS);
}

[-- Attachment #2: smime.p7s --]
[-- Type: application/x-pkcs7-signature, Size: 4985 bytes --]