* RE: IB atomic operations
[not found] ` <4D67E006.8000705-klaOcWyJdxkshyMvu7JE4pqQE7yCjDx5@public.gmane.org>
2011-02-25 17:29 ` Bharath Ramesh
@ 2011-02-25 18:30 ` Bharath Ramesh
1 sibling, 0 replies; 8+ messages in thread
From: Bharath Ramesh @ 2011-02-25 18:30 UTC (permalink / raw)
To: 'frank zago'; +Cc: linux-rdma-u79uwXL29TY76Z2rM5mHXA
[-- 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 --]
^ permalink raw reply [flat|nested] 8+ messages in thread