#include "dapl_cr_util.h"
#include "dapl_name_service.h"
#include "dapl_ib_util.h"
-
-#include <stdio.h>
-#include <unistd.h>
-#include <fcntl.h>
-#include <netinet/tcp.h>
-#include <byteswap.h>
-#include <poll.h>
-
-#include <sys/socket.h>
-#include <netinet/in.h>
-#include <arpa/inet.h>
-
-#if __BYTE_ORDER == __LITTLE_ENDIAN
-static inline uint64_t cpu_to_be64(uint64_t x) {return bswap_64(x);}
-#elif __BYTE_ORDER == __BIG_ENDIAN
-static inline uint64_t cpu_to_be64(uint64_t x) {return x;}
-#endif
+#include "dapl_osd.h"
extern int g_scm_pipe[2];
+#if defined(_WIN32) || defined(_WIN64)
+enum DAPL_FD_EVENTS {
+ DAPL_FD_READ = 0x1,
+ DAPL_FD_WRITE = 0x2,
+ DAPL_FD_ERROR = 0x4
+};
+
+static int dapl_config_socket(DAPL_SOCKET s)
+{
+ unsigned long nonblocking = 1;
+ return ioctlsocket(s, FIONBIO, &nonblocking);
+}
+
+static int dapl_connect_socket(DAPL_SOCKET s, struct sockaddr *addr,
+ int addrlen)
+{
+ int err;
+
+ connect(s, addr, addrlen);
+ err = WSAGetLastError();
+ return (err == WSAEWOULDBLOCK) ? EAGAIN : err;
+}
+
+struct dapl_fd_set {
+ struct fd_set set[3];
+};
+
+static struct dapl_fd_set *dapl_alloc_fd_set(void)
+{
+ return dapl_os_alloc(sizeof(struct dapl_fd_set));
+}
+
+static void dapl_fd_zero(struct dapl_fd_set *set)
+{
+ FD_ZERO(&set->set[0]);
+ FD_ZERO(&set->set[1]);
+ FD_ZERO(&set->set[2]);
+}
+
+static int dapl_fd_set(DAPL_SOCKET s, struct dapl_fd_set *set,
+ enum DAPL_FD_EVENTS event)
+{
+ FD_SET(s, &set->set[(event == DAPL_FD_READ) ? 0 : 1]);
+ FD_SET(s, &set->set[2]);
+ return 0;
+}
+
+static enum DAPL_FD_EVENTS dapl_poll(DAPL_SOCKET s, enum DAPL_FD_EVENTS event)
+{
+ struct fd_set rw_fds;
+ struct fd_set err_fds;
+ struct timeval tv;
+ int ret;
+
+ FD_ZERO(&rw_fds);
+ FD_ZERO(&err_fds);
+ FD_SET(s, &rw_fds);
+ FD_SET(s, &err_fds);
+
+ tv.tv_sec = 0;
+ tv.tv_usec = 0;
+
+ if (event == DAPL_FD_READ)
+ ret = select(1, &rw_fds, NULL, &err_fds, &tv);
+ else
+ ret = select(1, NULL, &rw_fds, &err_fds, &tv);
+
+ if (ret == 0)
+ return 0;
+ else if (FD_ISSET(s, &rw_fds))
+ return event;
+ else if (FD_ISSET(s, &err_fds))
+ return DAPL_FD_ERROR;
+ else
+ return WSAGetLastError();
+}
+
+static int dapl_select(struct dapl_fd_set *set)
+{
+ return select(0, &set->set[0], &set->set[1], &set->set[2], NULL);
+}
+#else // _WIN32 || _WIN64
+enum DAPL_FD_EVENTS {
+ DAPL_FD_READ = POLLIN,
+ DAPL_FD_WRITE = POLLOUT,
+ DAPL_FD_ERROR = POLLERR
+};
+
+static int dapl_config_socket(DAPL_SOCKET s)
+{
+ int ret;
+
+ ret = fcntl(s, F_GETFL);
+ if (ret >= 0)
+ ret = fcntl(s, F_SETFL, ret | O_NONBLOCK);
+ return ret;
+}
+
+static int dapl_connect_socket(DAPL_SOCKET s, struct sockaddr *addr, int addrlen)
+{
+ int ret;
+
+ ret = connect(s, addr, addrlen);
+
+ return (errno == EINPROGRESS) ? EAGAIN : ret;
+}
+
+struct dapl_fd_set {
+ int index;
+ struct pollfd set[DAPL_FD_SETSIZE];
+};
+
+static struct dapl_fd_set *dapl_alloc_fd_set(void)
+{
+ return dapl_os_alloc(sizeof(struct dapl_fd_set));
+}
+
+static void dapl_fd_zero(struct dapl_fd_set *set)
+{
+ set->index = 0;
+}
+
+static int dapl_fd_set(DAPL_SOCKET s, struct dapl_fd_set *set,
+ enum DAPL_FD_EVENTS event)
+{
+ if (set->index == DAPL_FD_SETSIZE - 1) {
+ dapl_log(DAPL_DBG_TYPE_ERR,
+ "SCM ERR: cm_thread exceeded FD_SETSIZE %d\n",
+ set->index + 1);
+ return -1;
+ }
+
+ set->set[set->index].fd = s;
+ set->set[set->index].revents = 0;
+ set->set[set->index++].events = event;
+ return 0;
+}
+
+static enum DAPL_FD_EVENTS dapl_poll(DAPL_SOCKET s, enum DAPL_FD_EVENTS event)
+{
+ struct pollfd fds;
+ int ret;
+
+ fds.fd = s;
+ fds.events = event;
+ fds.revents = 0;
+ ret = poll(&fds, 1, 0);
+ if (ret <= 0)
+ return ret;
+
+ return fds.revents;
+}
+
+static int dapl_select(struct dapl_fd_set *set)
+{
+ return poll(set->set, set->index, -1);
+}
+#endif
+
static struct ib_cm_handle *dapli_cm_create(void)
{
struct ib_cm_handle *cm_ptr;
(void)dapl_os_memzero(cm_ptr, sizeof(*cm_ptr));
cm_ptr->dst.ver = htons(DSCM_VER);
- cm_ptr->socket = -1;
+ cm_ptr->socket = DAPL_INVALID_SOCKET;
return cm_ptr;
bail:
dapl_os_free(cm_ptr, sizeof(*cm_ptr));
/* cleanup, never made it to work queue */
if (cm_ptr->state == SCM_INIT) {
- if (cm_ptr->socket >= 0)
- close(cm_ptr->socket);
+ if (cm_ptr->socket != DAPL_INVALID_SOCKET)
+ closesocket(cm_ptr->socket);
dapl_os_free(cm_ptr, sizeof(*cm_ptr));
return;
}
cm_ptr->ep->cm_handle = IB_INVALID_HANDLE;
/* close socket if still active */
- if (cm_ptr->socket >= 0) {
- close(cm_ptr->socket);
- cm_ptr->socket = -1;
+ if (cm_ptr->socket != DAPL_INVALID_SOCKET) {
+ closesocket(cm_ptr->socket);
+ cm_ptr->socket = DAPL_INVALID_SOCKET;
}
dapl_os_unlock(&cm_ptr->lock);
return DAT_SUCCESS;
} else {
/* send disc date, close socket, schedule destroy */
- if (cm_ptr->socket >= 0) {
- if (write(cm_ptr->socket,
- &disc_data, sizeof(disc_data)) == -1)
+ if (cm_ptr->socket != DAPL_INVALID_SOCKET) {
+ if (send(cm_ptr->socket, (char *) &disc_data,
+ sizeof(disc_data), 0) == -1)
dapl_log(DAPL_DBG_TYPE_WARN,
" cm_disc: write error = %s\n",
strerror(errno));
- close(cm_ptr->socket);
- cm_ptr->socket = -1;
+ closesocket(cm_ptr->socket);
+ cm_ptr->socket = DAPL_INVALID_SOCKET;
}
cm_ptr->state = SCM_DISCONNECTED;
}
dapli_socket_connected(dp_ib_cm_handle_t cm_ptr, int err)
{
int len, opt = 1;
- struct iovec iovec[2];
+ struct iovec iov[2];
struct dapl_ep *ep_ptr = cm_ptr->ep;
if (err) {
" socket connected, write QP and private data\n");
/* no delay for small packets */
- setsockopt(cm_ptr->socket,IPPROTO_TCP,TCP_NODELAY,&opt,sizeof(opt));
+ setsockopt(cm_ptr->socket, IPPROTO_TCP, TCP_NODELAY,
+ (char *) &opt, sizeof(opt));
/* send qp info and pdata to remote peer */
- iovec[0].iov_base = &cm_ptr->dst;
- iovec[0].iov_len = sizeof(ib_qp_cm_t);
+ iov[0].iov_base = (void *) &cm_ptr->dst;
+ iov[0].iov_len = sizeof(ib_qp_cm_t);
if (cm_ptr->dst.p_size) {
- iovec[1].iov_base = cm_ptr->p_data;
- iovec[1].iov_len = ntohl(cm_ptr->dst.p_size);
+ iov[1].iov_base = cm_ptr->p_data;
+ iov[1].iov_len = ntohl(cm_ptr->dst.p_size);
+ len = writev(cm_ptr->socket, iov, 2);
+ } else {
+ len = writev(cm_ptr->socket, iov, 1);
}
- len = writev(cm_ptr->socket, iovec, (cm_ptr->dst.p_size ? 2:1));
- if (len != (ntohl(cm_ptr->dst.p_size) + sizeof(ib_qp_cm_t))) {
+ if (len != (ntohl(cm_ptr->dst.p_size) + sizeof(ib_qp_cm_t))) {
dapl_log(DAPL_DBG_TYPE_ERR,
" CONN_PENDING write: ERR %s, wcnt=%d -> %s\n",
strerror(errno), len,
dapl_dbg_log(DAPL_DBG_TYPE_CM,
" connected: sending SRC GID subnet %016llx id %016llx\n",
(unsigned long long)
- cpu_to_be64(cm_ptr->dst.gid.global.subnet_prefix),
+ htonll(cm_ptr->dst.gid.global.subnet_prefix),
(unsigned long long)
- cpu_to_be64(cm_ptr->dst.gid.global.interface_id));
+ htonll(cm_ptr->dst.gid.global.interface_id));
/* queue up to work thread to avoid blocking consumer */
cm_ptr->state = SCM_RTU_PENDING;
return DAT_INSUFFICIENT_RESOURCES;
/* create, connect, sockopt, and exchange QP information */
- if ((cm_ptr->socket = socket(AF_INET,SOCK_STREAM,0)) < 0 ) {
+ if ((cm_ptr->socket = socket(AF_INET,SOCK_STREAM,0)) == DAPL_INVALID_SOCKET) {
dapl_os_free( cm_ptr, sizeof( *cm_ptr ) );
return DAT_INSUFFICIENT_RESOURCES;
}
- /* non-blocking */
- ret = fcntl(cm_ptr->socket, F_GETFL);
- if (ret < 0 || fcntl(cm_ptr->socket,
- F_SETFL, ret | O_NONBLOCK) < 0) {
- dapl_log(DAPL_DBG_TYPE_ERR,
- " socket connect: fcntl on socket %d ERR %d %s\n",
- cm_ptr->socket, ret,
- strerror(errno));
- goto bail;
- }
+ ret = dapl_config_socket(cm_ptr->socket);
+ if (ret < 0) {
+ dapl_log(DAPL_DBG_TYPE_ERR,
+ " socket connect: config socket %d ERR %d %s\n",
+ cm_ptr->socket, ret, strerror(errno));
+ goto bail;
+ }
((struct sockaddr_in*)r_addr)->sin_port = htons(r_qual);
- ret = connect(cm_ptr->socket, r_addr, sizeof(*r_addr));
- if (ret && errno != EINPROGRESS) {
+ ret = dapl_connect_socket(cm_ptr->socket, (struct sockaddr *) r_addr,
+ sizeof(*r_addr));
+ if (ret && ret != EAGAIN) {
dapl_log(DAPL_DBG_TYPE_ERR,
" socket connect ERROR: %s -> %s r_qual %d\n",
strerror(errno),
{
DAPL_EP *ep_ptr = cm_ptr->ep;
int len;
- struct iovec iovec[2];
short rtu_data = htons(0x0E0F);
ib_cm_events_t event = IB_CME_DESTINATION_REJECT;
/* read DST information into cm_ptr, overwrite SRC info */
dapl_dbg_log(DAPL_DBG_TYPE_EP," connect_rtu: recv peer QP data\n");
- iovec[0].iov_base = &cm_ptr->dst;
- iovec[0].iov_len = sizeof(ib_qp_cm_t);
- len = readv(cm_ptr->socket, iovec, 1);
+ len = recv(cm_ptr->socket, (char *) &cm_ptr->dst, sizeof(ib_qp_cm_t), 0);
if (len != sizeof(ib_qp_cm_t) || ntohs(cm_ptr->dst.ver) != DSCM_VER) {
dapl_log(DAPL_DBG_TYPE_ERR,
" CONN_RTU read: ERR %s, rcnt=%d, ver=%d -> %s\n",
/* read private data into cm_handle if any present */
dapl_dbg_log(DAPL_DBG_TYPE_EP," socket connected, read private data\n");
if (cm_ptr->dst.p_size) {
- iovec[0].iov_base = cm_ptr->p_data;
- iovec[0].iov_len = cm_ptr->dst.p_size;
- len = readv(cm_ptr->socket, iovec, 1);
+ len = recv(cm_ptr->socket, cm_ptr->p_data, cm_ptr->dst.p_size, 0);
if (len != cm_ptr->dst.p_size) {
dapl_log(DAPL_DBG_TYPE_ERR,
" CONN_RTU read pdata: ERR %s, rcnt=%d -> %s\n",
dapl_dbg_log(DAPL_DBG_TYPE_EP," connect_rtu: send RTU\n");
/* complete handshake after final QP state change */
- if (write(cm_ptr->socket, &rtu_data, sizeof(rtu_data)) == -1) {
+ if (send(cm_ptr->socket, (char *) &rtu_data, sizeof(rtu_data), 0) == -1) {
dapl_log(DAPL_DBG_TYPE_ERR,
" CONN_RTU: write error = %s\n", strerror(errno));
goto bail;
cm_ptr->hca = ia_ptr->hca_ptr;
/* bind, listen, set sockopt, accept, exchange data */
- if ((cm_ptr->socket = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
+ if ((cm_ptr->socket = socket(AF_INET, SOCK_STREAM, 0)) == DAPL_INVALID_SOCKET) {
dapl_log(DAPL_DBG_TYPE_ERR,
" ERR: listen socket create: %s\n",
strerror(errno));
goto bail;
}
- setsockopt(cm_ptr->socket,SOL_SOCKET,SO_REUSEADDR,&opt,sizeof(opt));
+ setsockopt(cm_ptr->socket, SOL_SOCKET, SO_REUSEADDR,
+ (char *) &opt, sizeof(opt));
addr.sin_port = htons(serviceID);
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = INADDR_ANY;
(void) dapl_os_memzero(acm_ptr, sizeof(*acm_ptr));
- acm_ptr->socket = -1;
+ acm_ptr->socket = DAPL_INVALID_SOCKET;
acm_ptr->sp = cm_ptr->sp;
acm_ptr->hca = cm_ptr->hca;
acm_ptr->socket = accept(cm_ptr->socket,
(struct sockaddr*)&acm_ptr->dst.ia_address,
(socklen_t*)&len);
- if (acm_ptr->socket < 0) {
+ if (acm_ptr->socket == DAPL_INVALID_SOCKET) {
dapl_log(DAPL_DBG_TYPE_ERR,
" accept: ERR %s on FD %d l_cr %p\n",
strerror(errno),cm_ptr->socket,cm_ptr);
dapl_dbg_log(DAPL_DBG_TYPE_EP," socket accepted, read QP data\n");
/* read in DST QP info, IA address. check for private data */
- len = read(acm_ptr->socket, &acm_ptr->dst, sizeof(ib_qp_cm_t));
+ len = recv(acm_ptr->socket, (char *) &acm_ptr->dst, sizeof(ib_qp_cm_t), 0);
if (len != sizeof(ib_qp_cm_t) ||
ntohs(acm_ptr->dst.ver) != DSCM_VER) {
dapl_log(DAPL_DBG_TYPE_ERR,
/* read private data into cm_handle if any present */
if (acm_ptr->dst.p_size) {
- len = read( acm_ptr->socket,
- acm_ptr->p_data, acm_ptr->dst.p_size);
+ len = recv(acm_ptr->socket, acm_ptr->p_data, acm_ptr->dst.p_size, 0);
if (len != acm_ptr->dst.p_size) {
dapl_log(DAPL_DBG_TYPE_ERR,
" accept read pdata: ERR %s, rcnt=%d\n",
DAPL_IA *ia_ptr = ep_ptr->header.owner_ia;
dp_ib_cm_handle_t cm_ptr = cr_ptr->ib_cm_handle;
ib_qp_cm_t local;
- struct iovec iovec[2];
+ struct iovec iov[2];
int len;
if (p_size > IB_MAX_REP_PDATA_SIZE)
return DAT_LENGTH_ERROR;
/* must have a accepted socket */
- if (cm_ptr->socket < 0)
+ if (cm_ptr->socket == DAPL_INVALID_SOCKET)
return DAT_INTERNAL_ERROR;
dapl_dbg_log(DAPL_DBG_TYPE_EP,
local.ia_address = ia_ptr->hca_ptr->hca_address;
local.p_size = htonl(p_size);
- iovec[0].iov_base = &local;
- iovec[0].iov_len = sizeof(ib_qp_cm_t);
+ iov[0].iov_base = (void *) &local;
+ iov[0].iov_len = sizeof(ib_qp_cm_t);
if (p_size) {
- iovec[1].iov_base = p_data;
- iovec[1].iov_len = p_size;
+ iov[1].iov_base = p_data;
+ iov[1].iov_len = p_size;
+ len = writev(cm_ptr->socket, iov, 2);
+ } else {
+ len = writev(cm_ptr->socket, iov, 1);
}
- len = writev(cm_ptr->socket, iovec, (p_size ? 2:1));
- if (len != (p_size + sizeof(ib_qp_cm_t))) {
+
+ if (len != (p_size + sizeof(ib_qp_cm_t))) {
dapl_log(DAPL_DBG_TYPE_ERR,
" ACCEPT_USR: ERR %s, wcnt=%d -> %s\n",
strerror(errno), len,
&cm_ptr->dst.ia_address)->sin_addr));
goto bail;
}
+
dapl_dbg_log(DAPL_DBG_TYPE_CM,
" ACCEPT_USR: local port=0x%x lid=0x%x"
" qpn=0x%x psize=%d\n",
dapl_dbg_log(DAPL_DBG_TYPE_CM,
" ACCEPT_USR SRC GID subnet %016llx id %016llx\n",
(unsigned long long)
- cpu_to_be64(local.gid.global.subnet_prefix),
+ htonll(local.gid.global.subnet_prefix),
(unsigned long long)
- cpu_to_be64(local.gid.global.interface_id));
+ htonll(local.gid.global.interface_id));
/* save state and reference to EP, queue for RTU data */
cm_ptr->ep = ep_ptr;
short rtu_data = 0;
/* complete handshake after final QP state change */
- len = read(cm_ptr->socket, &rtu_data, sizeof(rtu_data));
+ len = recv(cm_ptr->socket, (char *) &rtu_data, sizeof(rtu_data), 0);
if (len != sizeof(rtu_data) || ntohs(rtu_data) != 0x0e0f) {
dapl_log(DAPL_DBG_TYPE_ERR,
" ACCEPT_RTU: ERR %s, rcnt=%d rdata=%x\n",
/* close accepted socket, free cm_srvc_handle and return */
if (cm_ptr != NULL) {
- if (cm_ptr->socket >= 0) {
- close(cm_ptr->socket );
- cm_ptr->socket = -1;
+ if (cm_ptr->socket != DAPL_INVALID_SOCKET) {
+ closesocket(cm_ptr->socket);
+ cm_ptr->socket = DAPL_INVALID_SOCKET;
}
/* cr_thread will free */
cm_ptr->state = SCM_DESTROY;
IN DAT_COUNT psize,
IN const DAT_PVOID pdata)
{
- struct iovec iovec[2];
+ struct iovec iov[2];
dapl_dbg_log (DAPL_DBG_TYPE_EP,
" reject(cm %p reason %x, pdata %p, psize %d)\n",
cm_ptr, reason, pdata, psize);
/* write reject data to indicate reject */
- if (cm_ptr->socket >= 0) {
+ if (cm_ptr->socket != DAPL_INVALID_SOCKET) {
cm_ptr->dst.rej = (uint16_t)reason;
cm_ptr->dst.rej = htons(cm_ptr->dst.rej);
- iovec[0].iov_base = &cm_ptr->dst;
- iovec[0].iov_len = sizeof(ib_qp_cm_t);
+
+ iov[0].iov_base = (void *) &cm_ptr->dst;
+ iov[0].iov_len = sizeof(ib_qp_cm_t);
if (psize) {
- iovec[1].iov_base = pdata;
- iovec[2].iov_len = psize;
- writev(cm_ptr->socket, &iovec[0], 2);
- } else
- writev(cm_ptr->socket, &iovec[0], 1);
-
- close(cm_ptr->socket);
- cm_ptr->socket = -1;
+ iov[1].iov_base = pdata;
+ iov[1].iov_len = psize;
+ writev(cm_ptr->socket, iov, 2);
+ } else {
+ writev(cm_ptr->socket, iov, 1);
+ }
+
+ closesocket(cm_ptr->socket);
+ cm_ptr->socket = DAPL_INVALID_SOCKET;
}
/* cr_thread will destroy CR */
}
/* outbound/inbound CR processing thread to avoid blocking applications */
-#define SCM_MAX_CONN 8192
void cr_thread(void *arg)
{
- struct dapl_hca *hca_ptr = arg;
- dp_ib_cm_handle_t cr, next_cr;
- int opt,ret,idx;
- socklen_t opt_len;
- char rbuf[2];
- struct pollfd ufds[SCM_MAX_CONN];
-
- dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cr_thread: ENTER hca %p\n",hca_ptr);
-
- dapl_os_lock( &hca_ptr->ib_trans.lock );
- hca_ptr->ib_trans.cr_state = IB_THREAD_RUN;
- while (hca_ptr->ib_trans.cr_state == IB_THREAD_RUN) {
- idx=0;
- ufds[idx].fd = g_scm_pipe[0]; /* wakeup and process work */
- ufds[idx].events = POLLIN;
- ufds[idx].revents = 0;
-
- if (!dapl_llist_is_empty(&hca_ptr->ib_trans.list))
- next_cr = dapl_llist_peek_head (&hca_ptr->ib_trans.list);
- else
- next_cr = NULL;
-
- while (next_cr) {
- cr = next_cr;
- if ((cr->socket == -1 && cr->state == SCM_DESTROY) ||
- hca_ptr->ib_trans.cr_state != IB_THREAD_RUN) {
-
- dapl_dbg_log(DAPL_DBG_TYPE_CM," cr_thread: Free %p\n", cr);
- next_cr = dapl_llist_next_entry(&hca_ptr->ib_trans.list,
- (DAPL_LLIST_ENTRY*)&cr->entry );
- dapl_llist_remove_entry(&hca_ptr->ib_trans.list,
- (DAPL_LLIST_ENTRY*)&cr->entry);
- dapl_os_free(cr, sizeof(*cr));
- continue;
- }
-
- if (idx==SCM_MAX_CONN-1) {
- dapl_dbg_log(DAPL_DBG_TYPE_ERR,
- "SCM ERR: cm_thread exceeded FD_SETSIZE %d\n",idx+1);
- continue;
- }
-
- /* Add to ufds for poll, check for immediate work */
- ufds[++idx].fd = cr->socket; /* add listen or cr */
- ufds[idx].revents = 0;
- if (cr->state == SCM_CONN_PENDING)
- ufds[idx].events = POLLOUT;
- else
- ufds[idx].events = POLLIN;
-
- /* check socket for event, accept in or connect out */
- dapl_dbg_log(DAPL_DBG_TYPE_CM," poll cr=%p, fd=%d,%d\n",
- cr, cr->socket, ufds[idx].fd);
- dapl_os_unlock(&hca_ptr->ib_trans.lock);
- ret = poll(&ufds[idx],1,0);
- dapl_dbg_log(DAPL_DBG_TYPE_CM,
- " poll wakeup ret=%d cr->st=%d"
- " ev=0x%x fd=%d\n",
- ret,cr->state,ufds[idx].revents,ufds[idx].fd);
-
- /* data on listen, qp exchange, and on disconnect request */
- if ((ret == 1) && ufds[idx].revents == POLLIN) {
- if (cr->socket > 0) {
- if (cr->state == SCM_LISTEN)
- dapli_socket_accept(cr);
- else if (cr->state == SCM_ACCEPTING)
- dapli_socket_accept_data(cr);
- else if (cr->state == SCM_ACCEPTED)
- dapli_socket_accept_rtu(cr);
- else if (cr->state == SCM_RTU_PENDING)
- dapli_socket_connect_rtu(cr);
- else if (cr->state == SCM_CONNECTED)
- dapli_socket_disconnect(cr);
+ struct dapl_hca *hca_ptr = arg;
+ dp_ib_cm_handle_t cr, next_cr;
+ int opt, ret;
+ socklen_t opt_len;
+ char rbuf[2];
+ struct dapl_fd_set *set;
+ enum DAPL_FD_EVENTS event;
+
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cr_thread: ENTER hca %p\n", hca_ptr);
+ set = dapl_alloc_fd_set();
+ if (!set)
+ goto out;
+
+ dapl_os_lock(&hca_ptr->ib_trans.lock);
+ hca_ptr->ib_trans.cr_state = IB_THREAD_RUN;
+
+ while (hca_ptr->ib_trans.cr_state == IB_THREAD_RUN) {
+ dapl_fd_zero(set);
+ dapl_fd_set(g_scm_pipe[0], set, DAPL_FD_READ);
+
+ if (!dapl_llist_is_empty(&hca_ptr->ib_trans.list))
+ next_cr = dapl_llist_peek_head(&hca_ptr->ib_trans.list);
+ else
+ next_cr = NULL;
+
+ while (next_cr) {
+ cr = next_cr;
+ if ((cr->socket == DAPL_INVALID_SOCKET && cr->state == SCM_DESTROY) ||
+ hca_ptr->ib_trans.cr_state != IB_THREAD_RUN) {
+ next_cr = dapl_llist_next_entry(&hca_ptr->ib_trans.list,
+ (DAPL_LLIST_ENTRY*)&cr->entry);
+ dapl_llist_remove_entry(&hca_ptr->ib_trans.list,
+ (DAPL_LLIST_ENTRY*)&cr->entry);
+ dapl_os_free(cr, sizeof(*cr));
+ continue;
+ }
+
+ event = (cr->state == SCM_CONN_PENDING) ?
+ DAPL_FD_WRITE : DAPL_FD_READ;
+ if (dapl_fd_set(cr->socket, set, event)) {
+ dapl_log(DAPL_DBG_TYPE_ERR,
+ " cr_thread: DESTROY CR st=%d fd %d"
+ " -> %s\n", cr->state, cr->socket,
+ inet_ntoa(((struct sockaddr_in*)
+ &cr->dst.ia_address)->sin_addr));
+ dapli_cm_destroy(cr);
+ continue;
+ }
+
+ dapl_dbg_log(DAPL_DBG_TYPE_CM, " poll cr=%p, fd=%d\n",
+ cr, cr->socket);
+ dapl_os_unlock(&hca_ptr->ib_trans.lock);
+
+ ret = dapl_poll(cr->socket, event);
+
+ dapl_dbg_log(DAPL_DBG_TYPE_CM,
+ " poll wakeup ret=%d cr->st=%d fd=%d\n",
+ ret, cr->state, cr->socket);
+
+ /* data on listen, qp exchange, and on disconnect request */
+ if (ret == DAPL_FD_READ) {
+ if (cr->socket != DAPL_INVALID_SOCKET) {
+ switch (cr->state) {
+ case SCM_LISTEN:
+ dapli_socket_accept(cr);
+ break;
+ case SCM_ACCEPTING:
+ dapli_socket_accept_data(cr);
+ break;
+ case SCM_ACCEPTED:
+ dapli_socket_accept_rtu(cr);
+ break;
+ case SCM_RTU_PENDING:
+ dapli_socket_connect_rtu(cr);
+ break;
+ case SCM_CONNECTED:
+ dapli_socket_disconnect(cr);
+ break;
+ default:
+ break;
+ }
+ }
+ /* connect socket is writable, check status */
+ } else if (ret == DAPL_FD_WRITE || ret == DAPL_FD_ERROR) {
+ if (cr->state == SCM_CONN_PENDING) {
+ opt = 0;
+ ret = getsockopt(cr->socket, SOL_SOCKET,
+ SO_ERROR, (char *) &opt, &opt_len);
+ if (!ret)
+ dapli_socket_connected(cr, opt);
+ else
+ dapli_socket_connected(cr, errno);
+ } else {
+ dapl_log(DAPL_DBG_TYPE_CM,
+ " CM poll ERR, wrong state(%d) -> %s SKIP\n", cr->state,
+ inet_ntoa(((struct sockaddr_in*)&cr->dst.ia_address)->sin_addr));
+ }
+ } else if (ret != 0) {
+ dapl_log(DAPL_DBG_TYPE_CM,
+ " CM poll warning %s, ret=%d st=%d -> %s\n",
+ strerror(errno), ret, cr->state,
+ inet_ntoa(((struct sockaddr_in*)
+ &cr->dst.ia_address)->sin_addr));
+
+ /* POLLUP, NVAL, or poll error, issue event if connected */
+ if (cr->state == SCM_CONNECTED)
+ dapli_socket_disconnect(cr);
+ }
+
+ dapl_os_lock(&hca_ptr->ib_trans.lock);
+ next_cr = dapl_llist_next_entry(&hca_ptr->ib_trans.list,
+ (DAPL_LLIST_ENTRY*)&cr->entry);
}
- /* connect socket is writable, check status */
- } else if ((ret == 1) &&
- (ufds[idx].revents & POLLOUT ||
- ufds[idx].revents & POLLERR)) {
- if (cr->state == SCM_CONN_PENDING) {
- opt = 0;
- ret = getsockopt(cr->socket, SOL_SOCKET,
- SO_ERROR, &opt, &opt_len);
- if (!ret)
- dapli_socket_connected(cr,opt);
- else
- dapli_socket_connected(cr,errno);
- } else {
- dapl_log(DAPL_DBG_TYPE_CM,
- " CM poll ERR, wrong state(%d) -> %s SKIP\n",
- cr->state,
- inet_ntoa(((struct sockaddr_in*)
- &cr->dst.ia_address)->sin_addr));
+
+ dapl_os_unlock(&hca_ptr->ib_trans.lock);
+ dapl_dbg_log(DAPL_DBG_TYPE_CM," cr_thread: sleep, fds=%d\n",
+ set->index+1);
+ dapl_select(set);
+ dapl_dbg_log(DAPL_DBG_TYPE_CM," cr_thread: wakeup\n");
+
+ /* if pipe used to wakeup, consume */
+ if (dapl_poll(g_scm_pipe[0], DAPL_FD_READ) == DAPL_FD_READ) {
+ if (read(g_scm_pipe[0], rbuf, 2) == -1)
+ dapl_log(DAPL_DBG_TYPE_CM,
+ " cr_thread: read pipe error = %s\n",
+ strerror(errno));
}
- } else if (ret != 0) {
- dapl_log(DAPL_DBG_TYPE_CM,
- " CM poll warning %s, ret=%d revnt=%x st=%d -> %s\n",
- strerror(errno), ret, ufds[idx].revents, cr->state,
- inet_ntoa(((struct sockaddr_in*)
- &cr->dst.ia_address)->sin_addr));
-
- /* POLLUP, NVAL, or poll error, issue event if connected */
- if (cr->state == SCM_CONNECTED)
- dapli_socket_disconnect(cr);
- }
- dapl_os_lock(&hca_ptr->ib_trans.lock);
- next_cr = dapl_llist_next_entry(&hca_ptr->ib_trans.list,
- (DAPL_LLIST_ENTRY*)&cr->entry);
+ dapl_os_lock(&hca_ptr->ib_trans.lock);
}
+
dapl_os_unlock(&hca_ptr->ib_trans.lock);
- dapl_dbg_log(DAPL_DBG_TYPE_CM," cr_thread: sleep, %d\n", idx+1);
- poll(ufds,idx+1,-1); /* infinite, all sockets and pipe */
- /* if pipe used to wakeup, consume */
- if (ufds[0].revents == POLLIN)
- if (read(g_scm_pipe[0], rbuf, 2) == -1)
- dapl_log(DAPL_DBG_TYPE_CM,
- " cr_thread: read pipe error = %s\n",
- strerror(errno));
- dapl_dbg_log(DAPL_DBG_TYPE_CM," cr_thread: wakeup\n");
- dapl_os_lock(&hca_ptr->ib_trans.lock);
- }
- dapl_os_unlock(&hca_ptr->ib_trans.lock);
- hca_ptr->ib_trans.cr_state = IB_THREAD_EXIT;
- dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cr_thread(hca %p) exit\n",hca_ptr);
+ free(set);
+out:
+ hca_ptr->ib_trans.cr_state = IB_THREAD_EXIT;
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cr_thread(hca %p) exit\n",hca_ptr);
}
-
-/*
- * Local variables:
- * c-indent-level: 4
- * c-basic-offset: 4
- * tab-width: 8
- * End:
- */
*
**************************************************************************/
+#include "openib_osd.h"
#include "dapl.h"
#include "dapl_adapter_util.h"
#include "dapl_lmr_util.h"
#include "dapl_evd_util.h"
#include "dapl_ring_buffer_util.h"
-#include <sys/poll.h>
-#include <signal.h>
-int dapli_cq_thread_init(struct dapl_hca *hca_ptr)
+#if defined(_WIN64) || defined(_WIN32)
+void dapli_cq_thread_destroy(struct dapl_hca *hca_ptr)
{
- DAT_RETURN dat_status;
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread_destroy(%p)\n", hca_ptr);
- dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread_init(%p)\n", hca_ptr);
+ if (hca_ptr->ib_trans.cq_state != IB_THREAD_RUN)
+ return;
- /* create thread to process inbound connect request */
- hca_ptr->ib_trans.cq_state = IB_THREAD_INIT;
- dat_status = dapl_os_thread_create(cq_thread, (void*)hca_ptr, &hca_ptr->ib_trans.cq_thread);
- if (dat_status != DAT_SUCCESS)
- {
- dapl_dbg_log(DAPL_DBG_TYPE_ERR,
- " cq_thread_init: failed to create thread\n");
- return 1;
- }
+ /* destroy cr_thread and lock */
+ hca_ptr->ib_trans.cq_state = IB_THREAD_CANCEL;
+ SetEvent(hca_ptr->ib_trans.ib_cq->event);
+ dapl_dbg_log(DAPL_DBG_TYPE_CM," cq_thread_destroy(%p) cancel\n",hca_ptr);
+ while (hca_ptr->ib_trans.cq_state != IB_THREAD_EXIT) {
+ dapl_os_sleep_usec(20000);
+ }
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread_destroy(%d) exit\n",dapl_os_getpid());
+}
+
+static void cq_thread(void *arg)
+{
+ struct dapl_hca *hca_ptr = arg;
+ struct dapl_evd *evd_ptr;
+ struct ibv_cq *ibv_cq = NULL;
+
+ hca_ptr->ib_trans.cq_state = IB_THREAD_RUN;
+
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread: ENTER hca %p\n",hca_ptr);
- /* wait for thread to start */
- while (hca_ptr->ib_trans.cq_state != IB_THREAD_RUN) {
- struct timespec sleep, remain;
- sleep.tv_sec = 0;
- sleep.tv_nsec = 20000000; /* 20 ms */
- dapl_dbg_log(DAPL_DBG_TYPE_UTIL,
- " cq_thread_init: waiting for cq_thread\n");
- nanosleep (&sleep, &remain);
- }
- dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread_init(%d) exit\n",getpid());
- return 0;
+ /* wait on DTO event, or signal to abort */
+ while (hca_ptr->ib_trans.cq_state == IB_THREAD_RUN) {
+ if (!ibv_get_cq_event(hca_ptr->ib_trans.ib_cq, &ibv_cq, (void*)&evd_ptr)) {
+
+ if (DAPL_BAD_HANDLE(evd_ptr, DAPL_MAGIC_EVD)) {
+ ibv_ack_cq_events(ibv_cq, 1);
+ return;
+ }
+
+ /* process DTO event via callback */
+ dapl_evd_dto_callback(hca_ptr->ib_hca_handle, evd_ptr->ib_cq_handle,
+ (void*)evd_ptr );
+
+ ibv_ack_cq_events(ibv_cq, 1);
+ }
+ }
+ hca_ptr->ib_trans.cq_state = IB_THREAD_EXIT;
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread: EXIT: hca %p \n", hca_ptr);
}
+#else // _WIN32 || _WIN64
+
void dapli_cq_thread_destroy(struct dapl_hca *hca_ptr)
{
- dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread_destroy(%p)\n", hca_ptr);
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread_destroy(%p)\n", hca_ptr);
if (hca_ptr->ib_trans.cq_state != IB_THREAD_RUN)
return;
- /* destroy cr_thread and lock */
- hca_ptr->ib_trans.cq_state = IB_THREAD_CANCEL;
- pthread_kill(hca_ptr->ib_trans.cq_thread, SIGUSR1);
- dapl_dbg_log(DAPL_DBG_TYPE_CM," cq_thread_destroy(%p) cancel\n",hca_ptr);
- while (hca_ptr->ib_trans.cq_state != IB_THREAD_EXIT) {
- struct timespec sleep, remain;
- sleep.tv_sec = 0;
- sleep.tv_nsec = 2000000; /* 2 ms */
- dapl_dbg_log(DAPL_DBG_TYPE_UTIL,
- " cq_thread_destroy: waiting for cq_thread\n");
- nanosleep (&sleep, &remain);
- }
- dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread_destroy(%d) exit\n",getpid());
+ /* destroy cr_thread and lock */
+ hca_ptr->ib_trans.cq_state = IB_THREAD_CANCEL;
+ pthread_kill(hca_ptr->ib_trans.cq_thread, SIGUSR1);
+ dapl_dbg_log(DAPL_DBG_TYPE_CM," cq_thread_destroy(%p) cancel\n",hca_ptr);
+ while (hca_ptr->ib_trans.cq_state != IB_THREAD_EXIT) {
+ dapl_os_sleep_usec(20000);
+ }
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread_destroy(%d) exit\n",dapl_os_getpid());
}
/* catch the signal */
static void ib_cq_handler(int signum)
{
- return;
+ return;
}
-void cq_thread( void *arg )
+static void cq_thread(void *arg)
{
- struct dapl_hca *hca_ptr = arg;
- struct dapl_evd *evd_ptr;
- struct ibv_cq *ibv_cq = NULL;
+ struct dapl_hca *hca_ptr = arg;
+ struct dapl_evd *evd_ptr;
+ struct ibv_cq *ibv_cq = NULL;
sigset_t sigset;
sigemptyset(&sigset);
- sigaddset(&sigset,SIGUSR1);
- pthread_sigmask(SIG_UNBLOCK, &sigset, NULL);
- signal(SIGUSR1, ib_cq_handler);
+ sigaddset(&sigset,SIGUSR1);
+ pthread_sigmask(SIG_UNBLOCK, &sigset, NULL);
+ signal(SIGUSR1, ib_cq_handler);
hca_ptr->ib_trans.cq_state = IB_THREAD_RUN;
-
+
dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread: ENTER hca %p\n",hca_ptr);
- /* wait on DTO event, or signal to abort */
- while (hca_ptr->ib_trans.cq_state == IB_THREAD_RUN) {
- struct pollfd cq_fd = {
- .fd = hca_ptr->ib_trans.ib_cq->fd,
- .events = POLLIN,
- .revents = 0
- };
+ /* wait on DTO event, or signal to abort */
+ while (hca_ptr->ib_trans.cq_state == IB_THREAD_RUN) {
+ struct pollfd cq_fd = {
+ .fd = hca_ptr->ib_trans.ib_cq->fd,
+ .events = POLLIN,
+ .revents = 0
+ };
if ((poll(&cq_fd, 1, -1) == 1) &&
- (!ibv_get_cq_event(hca_ptr->ib_trans.ib_cq,
- &ibv_cq, (void*)&evd_ptr))) {
+ (!ibv_get_cq_event(hca_ptr->ib_trans.ib_cq, &ibv_cq, (void*)&evd_ptr))) {
if (DAPL_BAD_HANDLE(evd_ptr, DAPL_MAGIC_EVD)) {
ibv_ack_cq_events(ibv_cq, 1);
}
/* process DTO event via callback */
- dapl_evd_dto_callback ( hca_ptr->ib_hca_handle,
- evd_ptr->ib_cq_handle,
- (void*)evd_ptr );
+ dapl_evd_dto_callback(hca_ptr->ib_hca_handle,
+ evd_ptr->ib_cq_handle, (void*)evd_ptr );
ibv_ack_cq_events(ibv_cq, 1);
}
- }
- hca_ptr->ib_trans.cq_state = IB_THREAD_EXIT;
- dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread: EXIT: hca %p \n", hca_ptr);
+ }
+ hca_ptr->ib_trans.cq_state = IB_THREAD_EXIT;
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread: EXIT: hca %p \n", hca_ptr);
+}
+
+#endif // _WIN32 || _WIN64
+
+
+int dapli_cq_thread_init(struct dapl_hca *hca_ptr)
+{
+ DAT_RETURN dat_status;
+
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread_init(%p)\n", hca_ptr);
+
+ /* create thread to process inbound connect request */
+ hca_ptr->ib_trans.cq_state = IB_THREAD_INIT;
+ dat_status = dapl_os_thread_create(cq_thread, (void*)hca_ptr, &hca_ptr->ib_trans.cq_thread);
+ if (dat_status != DAT_SUCCESS) {
+ dapl_dbg_log(DAPL_DBG_TYPE_ERR,
+ " cq_thread_init: failed to create thread\n");
+ return 1;
+ }
+
+ /* wait for thread to start */
+ while (hca_ptr->ib_trans.cq_state != IB_THREAD_RUN) {
+ dapl_os_sleep_usec(20000);
+ }
+ dapl_dbg_log(DAPL_DBG_TYPE_UTIL," cq_thread_init(%d) exit\n",dapl_os_getpid());
+ return 0;
}
IN DAPL_EVD *evd_ptr,
IN DAT_COUNT *cqlen )
{
+ struct ibv_comp_channel *channel = ia_ptr->hca_ptr->ib_trans.ib_cq;
+
dapl_dbg_log ( DAPL_DBG_TYPE_UTIL,
"dapls_ib_cq_alloc: evd %p cqlen=%d \n", evd_ptr, *cqlen );
- struct ibv_comp_channel *channel = ia_ptr->hca_ptr->ib_trans.ib_cq;
-
#ifdef CQ_WAIT_OBJECT
if (evd_ptr->cq_wait_obj_handle)
channel = evd_ptr->cq_wait_obj_handle;
projects / ~ardavis / dapl.git / commitdiff