clients/rds/rdsib_cm.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */
/*
 * Copyright 2009 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */
/*
 * Copyright (c) 2005 SilverStorm Technologies, Inc. All rights reserved.
 *
 * This software is available to you under a choice of one of two
 * licenses.  You may choose to be licensed under the terms of the GNU
 * General Public License (GPL) Version 2, available from the file
 * COPYING in the main directory of this source tree, or 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 AND
 * 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.
 *
 */
/*
 * Sun elects to include this software in Sun product
 * under the OpenIB BSD license.
 *
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include <sys/ib/clients/rds/rdsib_cm.h>
#include <sys/ib/clients/rds/rdsib_ib.h>
#include <sys/ib/clients/rds/rdsib_buf.h>
#include <sys/ib/clients/rds/rdsib_ep.h>

/*
 * This file contains CM related work:
 *
 * Service registration/deregistration
 * Path lookup
 * CM connection callbacks
 * CM active and passive connection establishment
 * Connection failover
 */

#define	SRCIP	src_addr.un.ip4addr
#define	DSTIP	dst_addr.un.ip4addr

/*
 * Handle an incoming CM REQ
 */
/* ARGSUSED */
static ibt_cm_status_t
rds_handle_cm_req(rds_state_t *statep, ibt_cm_event_t *evp,
    ibt_cm_return_args_t *rargsp, void *rcmp, ibt_priv_data_len_t rcmp_len)
{
	ibt_cm_req_rcv_t	*reqp;
	ib_gid_t		lgid, rgid;
	rds_cm_private_data_t	cmp;
	rds_session_t		*sp;
	rds_ep_t		*ep;
	ibt_channel_hdl_t	chanhdl;
	ibt_ip_cm_info_t	ipcm_info;
	uint8_t			save_state, save_type;
	int			ret;

	RDS_DPRINTF2("rds_handle_cm_req", "Enter");

	reqp = &evp->cm_event.req;
	rgid = reqp->req_prim_addr.av_dgid; /* requester gid */
	lgid = reqp->req_prim_addr.av_sgid; /* receiver gid */

	RDS_DPRINTF2(LABEL, "REQ Received: From: %llx:%llx To: %llx:%llx",
	    rgid.gid_prefix, rgid.gid_guid, lgid.gid_prefix, lgid.gid_guid);

	/*
	 * CM private data brings IP information
	 * Private data received is a stream of bytes and may not be properly
	 * aligned. So, bcopy the data onto the stack before accessing it.
	 */
	bcopy((uint8_t *)evp->cm_priv_data, &cmp,
	    sizeof (rds_cm_private_data_t));

	/* extract the CM IP info */
	ret = ibt_get_ip_data(evp->cm_priv_data_len, evp->cm_priv_data,
	    &ipcm_info);
	if (ret != IBT_SUCCESS) {
		RDS_DPRINTF2("rds_handle_cm_req", "ibt_get_ip_data failed: %d",
		    ret);
		return (IBT_CM_REJECT);
	}

	RDS_DPRINTF2("rds_handle_cm_req",
	    "REQ Received: From IP: 0x%x To IP: 0x%x type: %d",
	    ipcm_info.SRCIP, ipcm_info.DSTIP, cmp.cmp_eptype);

	if (cmp.cmp_version != RDS_VERSION) {
		RDS_DPRINTF2(LABEL, "Version Mismatch: Local version: %d "
		    "Remote version: %d", RDS_VERSION, cmp.cmp_version);
		return (IBT_CM_REJECT);
	}

	/* RDS supports V4 addresses only */
	if ((ipcm_info.src_addr.family != AF_INET) ||
	    (ipcm_info.dst_addr.family != AF_INET)) {
		RDS_DPRINTF2(LABEL, "Unsupported Address Family: "
		    "src: %d dst: %d", ipcm_info.src_addr.family,
		    ipcm_info.dst_addr.family);
		return (IBT_CM_REJECT);
	}

	if (cmp.cmp_arch != RDS_THIS_ARCH) {
		RDS_DPRINTF2(LABEL, "ARCH does not match (%d != %d)",
		    cmp.cmp_arch, RDS_THIS_ARCH);
		return (IBT_CM_REJECT);
	}

	if ((cmp.cmp_eptype != RDS_EP_TYPE_CTRL) &&
	    (cmp.cmp_eptype != RDS_EP_TYPE_DATA)) {
		RDS_DPRINTF2(LABEL, "Unknown Channel type: %d", cmp.cmp_eptype);
		return (IBT_CM_REJECT);
	}

	/* user_buffer_size should be same on all nodes */
	if (cmp.cmp_user_buffer_size != UserBufferSize) {
		RDS_DPRINTF2(LABEL,
		    "UserBufferSize Mismatch, this node: %d remote node: %d",
		    UserBufferSize, cmp.cmp_user_buffer_size);
		return (IBT_CM_REJECT);
	}

	/*
	 * RDS needs more time to process a failover REQ so send an MRA.
	 * Otherwise, the remote may retry the REQ and fail the connection.
	 */
	if ((cmp.cmp_failover) && (cmp.cmp_eptype == RDS_EP_TYPE_DATA)) {
		RDS_DPRINTF2("rds_handle_cm_req", "Session Failover, send MRA");
		(void) ibt_cm_delay(IBT_CM_DELAY_REQ, evp->cm_session_id,
		    10000000 /* 10 sec */, NULL, 0);
	}

	/* Is there a session to the destination node? */
	rw_enter(&statep->rds_sessionlock, RW_READER);
	sp = rds_session_lkup(statep, ipcm_info.SRCIP, rgid.gid_guid);
	rw_exit(&statep->rds_sessionlock);

	if (sp == NULL) {
		/*
		 * currently there is no session to the destination
		 * remote ip in the private data is the local ip and vice
		 * versa
		 */
		sp = rds_session_create(statep, ipcm_info.DSTIP,
		    ipcm_info.SRCIP, reqp, RDS_SESSION_PASSIVE);
		if (sp == NULL) {
			/* Check the list anyway. */
			rw_enter(&statep->rds_sessionlock, RW_READER);
			sp = rds_session_lkup(statep, ipcm_info.SRCIP,
			    rgid.gid_guid);
			rw_exit(&statep->rds_sessionlock);
			if (sp == NULL) {
				/*
				 * The only way this can fail is due to lack
				 * of kernel resources
				 */
				return (IBT_CM_REJECT);
			}
		}
	}

	rw_enter(&sp->session_lock, RW_WRITER);

	/* catch peer-to-peer case as soon as possible */
	if ((sp->session_state == RDS_SESSION_STATE_CREATED) ||
	    (sp->session_state == RDS_SESSION_STATE_INIT)) {
		/* Check possible peer-to-peer case here */
		if (sp->session_type != RDS_SESSION_PASSIVE) {
			RDS_DPRINTF2("rds_handle_cm_req",
			    "SP(%p) Peer-peer connection handling", sp);
			if (lgid.gid_guid > rgid.gid_guid) {
				/* this node is active so reject this request */
				rw_exit(&sp->session_lock);
				return (IBT_CM_REJECT);
			} else {
				/* this node is passive, change the session */
				sp->session_type = RDS_SESSION_PASSIVE;
				sp->session_lgid = lgid;
				sp->session_rgid = rgid;
			}
		}
	}

	RDS_DPRINTF2(LABEL, "SP(%p) state: %d", sp, sp->session_state);
	save_state = sp->session_state;
	save_type = sp->session_type;

	switch (sp->session_state) {
	case RDS_SESSION_STATE_CONNECTED:
		RDS_DPRINTF2(LABEL, "STALE Session Detected SP(%p)", sp);
		sp->session_state = RDS_SESSION_STATE_ERROR;
		RDS_DPRINTF3("rds_handle_cm_req", "SP(%p) State "
		    "RDS_SESSION_STATE_ERROR", sp);

		/* FALLTHRU */
	case RDS_SESSION_STATE_ERROR:
	case RDS_SESSION_STATE_PASSIVE_CLOSING:
		/*
		 * Some other thread must be processing this session,
		 * this thread must wait until the other thread finishes.
		 */
		sp->session_type = RDS_SESSION_PASSIVE;
		rw_exit(&sp->session_lock);

		/* Handling this will take some time, so send an MRA */
		(void) ibt_cm_delay(IBT_CM_DELAY_REQ, evp->cm_session_id,
		    10000000 /* 10 sec */, NULL, 0);

		/*
		 * Any pending completions don't get flushed until the channel
		 * is closed. So, passing 0 here will not wait for pending
		 * completions in rds_session_close before closing the channel
		 */
		rds_session_close(sp, IBT_NOCALLBACKS, 0);

		rw_enter(&sp->session_lock, RW_WRITER);

		/*
		 * If the session was in ERROR, then either a failover thread
		 * or event_failure thread would be processing this session.
		 * This thread should wait for event_failure thread to
		 * complete. This need not wait for failover thread.
		 */
		if ((save_state != RDS_SESSION_STATE_CONNECTED) &&
		    (save_type == RDS_SESSION_PASSIVE)) {
				/*
				 * The other thread is event_failure thread,
				 * wait until it finishes.
				 */
				while (!((sp->session_state ==
				    RDS_SESSION_STATE_FAILED) ||
				    (sp->session_state ==
				    RDS_SESSION_STATE_FINI))) {
					rw_exit(&sp->session_lock);
					delay(drv_usectohz(1000000));
					rw_enter(&sp->session_lock, RW_WRITER);
				}
		}

		/* move the session to init state */
		if ((sp->session_state == RDS_SESSION_STATE_ERROR) ||
		    (sp->session_state == RDS_SESSION_STATE_PASSIVE_CLOSING)) {
			ret = rds_session_reinit(sp, lgid);
			sp->session_myip = ipcm_info.DSTIP;
			sp->session_lgid = lgid;
			sp->session_rgid = rgid;
			if (ret != 0) {
				rds_session_fini(sp);
				sp->session_state = RDS_SESSION_STATE_FAILED;
				RDS_DPRINTF3("rds_handle_cm_req",
				    "SP(%p) State RDS_SESSION_STATE_FAILED",
				    sp);
				rw_exit(&sp->session_lock);
				return (IBT_CM_REJECT);
			} else {
				sp->session_state = RDS_SESSION_STATE_INIT;
				RDS_DPRINTF3("rds_handle_cm_req",
				    "SP(%p) State RDS_SESSION_STATE_INIT", sp);
			}

			if (cmp.cmp_eptype == RDS_EP_TYPE_CTRL) {
				ep = &sp->session_ctrlep;
			} else {
				ep = &sp->session_dataep;
			}
			break;
		}

		/* FALLTHRU */
	case RDS_SESSION_STATE_CREATED:
	case RDS_SESSION_STATE_FAILED:
	case RDS_SESSION_STATE_FINI:
		/*
		 * Initialize both channels, we accept this connection
		 * only if both channels are initialized
		 */
		sp->session_type = RDS_SESSION_PASSIVE;
		sp->session_lgid = lgid;
		sp->session_rgid = rgid;
		sp->session_state = RDS_SESSION_STATE_CREATED;
		RDS_DPRINTF3("rds_handle_cm_req", "SP(%p) State "
		    "RDS_SESSION_STATE_CREATED", sp);
		ret = rds_session_init(sp);
		if (ret != 0) {
			/* Seems like there are not enough resources */
			sp->session_state = RDS_SESSION_STATE_FAILED;
			RDS_DPRINTF3("rds_handle_cm_req", "SP(%p) State "
			    "RDS_SESSION_STATE_FAILED", sp);
			rw_exit(&sp->session_lock);
			return (IBT_CM_REJECT);
		}
		sp->session_state = RDS_SESSION_STATE_INIT;
		RDS_DPRINTF3("rds_handle_cm_req", "SP(%p) State "
		    "RDS_SESSION_STATE_INIT", sp);

		/* FALLTHRU */
	case RDS_SESSION_STATE_INIT:
		/*
		 * When re-using an existing session, make sure the
		 * session is still through the same HCA. Otherwise, the
		 * memory registrations have to moved to the new HCA.
		 */
		if (cmp.cmp_eptype == RDS_EP_TYPE_DATA) {
			if (sp->session_lgid.gid_guid != lgid.gid_guid) {
				RDS_DPRINTF2("rds_handle_cm_req",
				    "Existing Session but different gid "
				    "existing: 0x%llx, new: 0x%llx, "
				    "sending an MRA",
				    sp->session_lgid.gid_guid, lgid.gid_guid);
				(void) ibt_cm_delay(IBT_CM_DELAY_REQ,
				    evp->cm_session_id, 10000000 /* 10 sec */,
				    NULL, 0);
				ret = rds_session_reinit(sp, lgid);
				if (ret != 0) {
					rds_session_fini(sp);
					sp->session_state =
					    RDS_SESSION_STATE_FAILED;
					sp->session_failover = 0;
					RDS_DPRINTF3("rds_failover_session",
					    "SP(%p) State "
					    "RDS_SESSION_STATE_FAILED", sp);
					rw_exit(&sp->session_lock);
					return (IBT_CM_REJECT);
				}
			}
			ep = &sp->session_dataep;
		} else {
			ep = &sp->session_ctrlep;
		}

		break;
	default:
		RDS_DPRINTF2(LABEL, "ERROR: SP(%p) is in an unexpected "
		    "state: %d", sp, sp->session_state);
		rw_exit(&sp->session_lock);
		return (IBT_CM_REJECT);
	}

	sp->session_failover = 0; /* reset any previous value */
	if (cmp.cmp_failover) {
		RDS_DPRINTF2("rds_handle_cm_req",
		    "SP(%p) Failover Session (BP %p)", sp, cmp.cmp_last_bufid);
		sp->session_failover = 1;
	}

	mutex_enter(&ep->ep_lock);
	if (ep->ep_state == RDS_EP_STATE_UNCONNECTED) {
		ep->ep_state = RDS_EP_STATE_PASSIVE_PENDING;
		sp->session_type = RDS_SESSION_PASSIVE;
		rw_exit(&sp->session_lock);
	} else if (ep->ep_state == RDS_EP_STATE_ACTIVE_PENDING) {
		rw_exit(&sp->session_lock);
		/*
		 * Peer to peer connection. There is an active
		 * connection pending on this ep. The one with
		 * greater port guid becomes active and the
		 * other becomes passive.
		 */
		RDS_DPRINTF2("rds_handle_cm_req",
		    "EP(%p) Peer-peer connection handling", ep);
		if (lgid.gid_guid > rgid.gid_guid) {
			/* this node is active so reject this request */
			mutex_exit(&ep->ep_lock);
			RDS_DPRINTF2(LABEL, "SP(%p) EP(%p): "
			    "Rejecting passive in favor of active", sp, ep);
			return (IBT_CM_REJECT);
		} else {
			/*
			 * This session is not the active end, change it
			 * to passive end.
			 */
			ep->ep_state = RDS_EP_STATE_PASSIVE_PENDING;

			rw_enter(&sp->session_lock, RW_WRITER);
			sp->session_type = RDS_SESSION_PASSIVE;
			sp->session_lgid = lgid;
			sp->session_rgid = rgid;
			rw_exit(&sp->session_lock);
		}
	} else {
		rw_exit(&sp->session_lock);
	}

	ep->ep_lbufid = cmp.cmp_last_bufid;
	ep->ep_ackwr.wr.rc.rcwr.rdma.rdma_raddr = (ib_vaddr_t)cmp.cmp_ack_addr;
	ep->ep_ackwr.wr.rc.rcwr.rdma.rdma_rkey = cmp.cmp_ack_rkey;
	cmp.cmp_last_bufid = ep->ep_rbufid;
	cmp.cmp_ack_addr = ep->ep_ack_addr;
	cmp.cmp_ack_rkey = ep->ep_ack_rkey;
	mutex_exit(&ep->ep_lock);

	/* continue with accepting the connection request for this channel */
	chanhdl = rds_ep_alloc_rc_channel(ep, reqp->req_prim_hca_port);
	if (chanhdl == NULL) {
		mutex_enter(&ep->ep_lock);
		ep->ep_state = RDS_EP_STATE_UNCONNECTED;
		mutex_exit(&ep->ep_lock);
		return (IBT_CM_REJECT);
	}

	/* pre-post recv buffers in the RQ */
	rds_post_recv_buf((void *)chanhdl);

	rargsp->cm_ret_len = sizeof (rds_cm_private_data_t);
	bcopy((uint8_t *)&cmp, rcmp, sizeof (rds_cm_private_data_t));
	rargsp->cm_ret.rep.cm_channel = chanhdl;
	rargsp->cm_ret.rep.cm_rdma_ra_out = 4;
	rargsp->cm_ret.rep.cm_rdma_ra_in = 4;
	rargsp->cm_ret.rep.cm_rnr_retry_cnt = MinRnrRetry;

	RDS_DPRINTF2("rds_handle_cm_req", "Return: SP(%p) EP(%p) Chan (%p)",
	    sp, ep, chanhdl);

	return (IBT_CM_ACCEPT);
}

/*
 * Handle an incoming CM REP
 * Pre-post recv buffers for the QP
 */
/* ARGSUSED */
static ibt_cm_status_t
rds_handle_cm_rep(ibt_cm_event_t *evp, ibt_cm_return_args_t *rargsp,
    void *rcmp, ibt_priv_data_len_t rcmp_len)
{
	rds_ep_t	*ep;
	rds_cm_private_data_t	cmp;

	RDS_DPRINTF2("rds_handle_cm_rep", "Enter");

	/* pre-post recv buffers in the RQ */
	rds_post_recv_buf((void *)evp->cm_channel);

	ep = (rds_ep_t *)ibt_get_chan_private(evp->cm_channel);
	bcopy((uint8_t *)evp->cm_priv_data, &cmp,
	    sizeof (rds_cm_private_data_t));
	ep->ep_lbufid = cmp.cmp_last_bufid;
	ep->ep_ackwr.wr.rc.rcwr.rdma.rdma_raddr = (ib_vaddr_t)cmp.cmp_ack_addr;
	ep->ep_ackwr.wr.rc.rcwr.rdma.rdma_rkey = cmp.cmp_ack_rkey;

	rargsp->cm_ret_len = 0;

	RDS_DPRINTF2("rds_handle_cm_rep", "Return: lbufid: %p", ep->ep_lbufid);

	return (IBT_CM_ACCEPT);
}

/*
 * Handle CONN EST
 */
static ibt_cm_status_t
rds_handle_cm_conn_est(ibt_cm_event_t *evp)
{
	rds_session_t	*sp;
	rds_ep_t	*ep;

	ep = (rds_ep_t *)ibt_get_chan_private(evp->cm_channel);

	RDS_DPRINTF2("rds_handle_cm_conn_est", "EP(%p) State: %d", ep,
	    ep->ep_state);

	mutex_enter(&ep->ep_lock);
	ASSERT((ep->ep_state == RDS_EP_STATE_ACTIVE_PENDING) ||
	    (ep->ep_state == RDS_EP_STATE_PASSIVE_PENDING));
	ep->ep_state = RDS_EP_STATE_CONNECTED;
	ep->ep_chanhdl = evp->cm_channel;
	sp = ep->ep_sp;
	mutex_exit(&ep->ep_lock);

	(void) rds_session_active(sp);

	RDS_DPRINTF2("rds_handle_cm_conn_est", "Return");
	return (IBT_CM_ACCEPT);
}

/*
 * Handle CONN CLOSED
 */
static ibt_cm_status_t
rds_handle_cm_conn_closed(ibt_cm_event_t *evp)
{
	rds_ep_t	*ep;
	rds_session_t	*sp;

	/* Catch DREQs but ignore DREPs */
	if (evp->cm_event.closed != IBT_CM_CLOSED_DREQ_RCVD) {
		RDS_DPRINTF2("rds_handle_cm_conn_closed",
		    "Ignoring Event: %d received", evp->cm_event.closed);
		return (IBT_CM_ACCEPT);
	}

	ep = (rds_ep_t *)ibt_get_chan_private(evp->cm_channel);
	sp = ep->ep_sp;
	RDS_DPRINTF2("rds_handle_cm_conn_closed", "EP(%p) Chan(%p) Enter",
	    ep, evp->cm_channel);

	mutex_enter(&ep->ep_lock);
	if (ep->ep_state != RDS_EP_STATE_CONNECTED) {
		/* Ignore this DREQ */
		RDS_DPRINTF2("rds_handle_cm_conn_closed",
		    "EP(%p) not connected, state: %d", ep, ep->ep_state);
		mutex_exit(&ep->ep_lock);
		return (IBT_CM_ACCEPT);
	}
	ep->ep_state = RDS_EP_STATE_CLOSING;
	mutex_exit(&ep->ep_lock);

	rw_enter(&sp->session_lock, RW_WRITER);
	RDS_DPRINTF2("rds_handle_cm_conn_closed", "SP(%p) - state: %d", sp,
	    sp->session_state);

	switch (sp->session_state) {
	case RDS_SESSION_STATE_CONNECTED:
	case RDS_SESSION_STATE_HCA_CLOSING:
		sp->session_state = RDS_SESSION_STATE_PASSIVE_CLOSING;
		RDS_DPRINTF3("rds_handle_cm_conn_closed", "SP(%p) State "
		    "RDS_SESSION_STATE_PASSIVE_CLOSING", sp);
		break;

	case RDS_SESSION_STATE_PASSIVE_CLOSING:
		sp->session_state = RDS_SESSION_STATE_CLOSED;
		RDS_DPRINTF3("rds_handle_cm_conn_closed", "SP(%p) State "
		    "RDS_SESSION_STATE_CLOSED", sp);
		rds_passive_session_fini(sp);
		sp->session_state = RDS_SESSION_STATE_FINI;
		RDS_DPRINTF3("rds_handle_cm_conn_closed",
		    "SP(%p) State RDS_SESSION_STATE_FINI", sp);
		break;

	case RDS_SESSION_STATE_ACTIVE_CLOSING:
	case RDS_SESSION_STATE_ERROR:
	case RDS_SESSION_STATE_CLOSED:
		break;

	case RDS_SESSION_STATE_INIT:
		sp->session_state = RDS_SESSION_STATE_ERROR;
		RDS_DPRINTF3("rds_handle_cm_conn_closed", "SP(%p) State "
		    "RDS_SESSION_STATE_ERROR", sp);
		rds_passive_session_fini(sp);
		sp->session_state = RDS_SESSION_STATE_FAILED;
		RDS_DPRINTF3("rds_handle_cm_conn_closed",
		    "SP(%p) State RDS_SESSION_STATE_FAILED", sp);
		break;

	default:
		RDS_DPRINTF2("rds_handle_cm_conn_closed",
		    "SP(%p) - Unexpected state: %d", sp, sp->session_state);
		rds_passive_session_fini(sp);
		sp->session_state = RDS_SESSION_STATE_FAILED;
		RDS_DPRINTF3("rds_handle_cm_conn_closed", "SP(%p) State "
		    "RDS_SESSION_STATE_FAILED", sp);
	}
	rw_exit(&sp->session_lock);

	mutex_enter(&ep->ep_lock);
	ep->ep_state = RDS_EP_STATE_CLOSED;
	mutex_exit(&ep->ep_lock);

	RDS_DPRINTF2("rds_handle_cm_conn_closed", "SP(%p) Return", sp);
	return (IBT_CM_ACCEPT);
}

/*
 * Handle EVENT FAILURE
 */
static ibt_cm_status_t
rds_handle_cm_event_failure(ibt_cm_event_t *evp)
{
	rds_ep_t	*ep;
	rds_session_t	*sp;
	int		ret;

	RDS_DPRINTF2("rds_handle_cm_event_failure", "Enter: Chan hdl: 0x%p "
	    "Code: %d msg: %d reason: %d", evp->cm_channel,
	    evp->cm_event.failed.cf_code, evp->cm_event.failed.cf_msg,
	    evp->cm_event.failed.cf_reason);

	if (evp->cm_event.failed.cf_reason == IBT_CM_INVALID_SID) {
		RDS_DPRINTF2(LABEL,
		    "Received REJ with reason IBT_CM_INVALID_SID: "
		    "RDS may not be loaded on the remote system");
	}

	if (evp->cm_channel == NULL) {
		return (IBT_CM_ACCEPT);
	}

	if ((evp->cm_event.failed.cf_code != IBT_CM_FAILURE_STALE) &&
	    (evp->cm_event.failed.cf_msg == IBT_CM_FAILURE_REQ)) {
		/*
		 * This end is active, just ignore, ibt_open_rc_channel()
		 * caller will take care of cleanup.
		 */
		RDS_DPRINTF2("rds_handle_cm_event_failure",
		    "Ignoring this event: Chan hdl: 0x%p", evp->cm_channel);
		return (IBT_CM_ACCEPT);
	}

	ep = (rds_ep_t *)ibt_get_chan_private(evp->cm_channel);
	sp = ep->ep_sp;

	rw_enter(&sp->session_lock, RW_WRITER);
	if (sp->session_type == RDS_SESSION_PASSIVE) {
		RDS_DPRINTF2("rds_handle_cm_event_failure",
		    "SP(%p) - state: %d", sp, sp->session_state);
		if ((sp->session_state == RDS_SESSION_STATE_INIT) ||
		    (sp->session_state == RDS_SESSION_STATE_CONNECTED)) {
			sp->session_state = RDS_SESSION_STATE_ERROR;
			RDS_DPRINTF3("rds_handle_cm_event_failure",
			    "SP(%p) State RDS_SESSION_STATE_ERROR", sp);

			/*
			 * Store the cm_channel for freeing later
			 * Active side frees it on ibt_open_rc_channel
			 * failure
			 */
			if (ep->ep_chanhdl == NULL) {
				ep->ep_chanhdl = evp->cm_channel;
			}
			rw_exit(&sp->session_lock);

			/*
			 * rds_passive_session_fini should not be called
			 * directly in the CM handler. It will cause a deadlock.
			 */
			ret = ddi_taskq_dispatch(rds_taskq,
			    rds_cleanup_passive_session, (void *)sp,
			    DDI_NOSLEEP);
			if (ret != DDI_SUCCESS) {
				RDS_DPRINTF2("rds_handle_cm_event_failure",
				    "SP(%p) TaskQ dispatch FAILED:%d", sp, ret);
			}
			return (IBT_CM_ACCEPT);
		}
	}
	rw_exit(&sp->session_lock);

	RDS_DPRINTF2("rds_handle_cm_event_failure", "SP(%p) Return", sp);
	return (IBT_CM_ACCEPT);
}

/*
 * CM Handler
 *
 * Called by IBCM
 * The cm_private type differs for active and passive events.
 */
ibt_cm_status_t
rds_cm_handler(void *cm_private, ibt_cm_event_t *eventp,
    ibt_cm_return_args_t *ret_args, void *ret_priv_data,
    ibt_priv_data_len_t ret_len_max)
{
	ibt_cm_status_t		ret = IBT_CM_ACCEPT;

	RDS_DPRINTF2("rds_cm_handler", "Enter: event: %d", eventp->cm_type);

	switch (eventp->cm_type) {
	case IBT_CM_EVENT_REQ_RCV:
		ret = rds_handle_cm_req((rds_state_t *)cm_private, eventp,
		    ret_args, ret_priv_data, ret_len_max);
		break;
	case IBT_CM_EVENT_REP_RCV:
		ret = rds_handle_cm_rep(eventp, ret_args, ret_priv_data,
		    ret_len_max);
		break;
	case IBT_CM_EVENT_MRA_RCV:
		/* Not supported */
		break;
	case IBT_CM_EVENT_CONN_EST:
		ret = rds_handle_cm_conn_est(eventp);
		break;
	case IBT_CM_EVENT_CONN_CLOSED:
		ret = rds_handle_cm_conn_closed(eventp);
		break;
	case IBT_CM_EVENT_FAILURE:
		ret = rds_handle_cm_event_failure(eventp);
		break;
	case IBT_CM_EVENT_LAP_RCV:
		/* Not supported */
		RDS_DPRINTF2(LABEL, "LAP message received");
		break;
	case IBT_CM_EVENT_APR_RCV:
		/* Not supported */
		RDS_DPRINTF2(LABEL, "APR message received");
		break;
	default:
		break;
	}

	RDS_DPRINTF2("rds_cm_handler", "Return");

	return (ret);
}

/* This is based on OFED Linux RDS */
#define	RDS_PORT_NUM	6556

/*
 * Register the wellknown service with service id: RDS_SERVICE_ID
 * Incoming connection requests should arrive on this service id.
 */
ibt_srv_hdl_t
rds_register_service(ibt_clnt_hdl_t rds_ibhdl)
{
	ibt_srv_hdl_t	srvhdl;
	ibt_srv_desc_t	srvdesc;
	int		ret;

	RDS_DPRINTF2("rds_register_service", "Enter: 0x%p", rds_ibhdl);

	bzero(&srvdesc, sizeof (ibt_srv_desc_t));
	srvdesc.sd_handler = rds_cm_handler;
	srvdesc.sd_flags = IBT_SRV_NO_FLAGS;

	/*
	 * This is the new service id as per:
	 * Annex A11: RDMA IP CM Service
	 */
	rdsib_statep->rds_service_id = ibt_get_ip_sid(IPPROTO_TCP,
	    RDS_PORT_NUM);
	ret = ibt_register_service(rds_ibhdl, &srvdesc,
	    rdsib_statep->rds_service_id, 1, &srvhdl, NULL);
	if (ret != IBT_SUCCESS) {
		RDS_DPRINTF2(LABEL,
		    "RDS Service (0x%llx) Registration Failed: %d",
		    rdsib_statep->rds_service_id, ret);
		return (NULL);
	}

	RDS_DPRINTF2("rds_register_service", "Return: 0x%p", srvhdl);
	return (srvhdl);
}

/* Bind the RDS service on all ports */
int
rds_bind_service(rds_state_t *statep)
{
	rds_hca_t	*hcap;
	ib_gid_t	gid;
	uint_t		jx, nbinds = 0, nports = 0;
	int		ret;

	RDS_DPRINTF2("rds_bind_service", "Enter: 0x%p", statep);

	rw_enter(&statep->rds_hca_lock, RW_READER);

	hcap = statep->rds_hcalistp;
	while (hcap != NULL) {

		/* skip the HCAs that are not fully online */
		if ((hcap->hca_state != RDS_HCA_STATE_OPEN) &&
		    (hcap->hca_state != RDS_HCA_STATE_MEM_REGISTERED)) {
			RDS_DPRINTF2("rds_bind_service",
			    "Skipping HCA: 0x%llx, state: %d",
			    hcap->hca_guid, hcap->hca_state);
			hcap = hcap->hca_nextp;
			continue;
		}

		/* currently, we have space for only 4 bindhdls */
		ASSERT(hcap->hca_nports < 4);
		for (jx = 0; jx < hcap->hca_nports; jx++) {
			nports++;
			if (hcap->hca_pinfop[jx].p_linkstate !=
			    IBT_PORT_ACTIVE) {
				/*
				 * service bind will be called in the async
				 * handler when the port comes up. Clear any
				 * stale bind handle.
				 */
				hcap->hca_bindhdl[jx] = NULL;
				continue;
			}

			gid = hcap->hca_pinfop[jx].p_sgid_tbl[0];
			RDS_DPRINTF5(LABEL, "HCA: 0x%llx Port: %d "
			    "gid: %llx:%llx", hcap->hca_guid,
			    hcap->hca_pinfop[jx].p_port_num, gid.gid_prefix,
			    gid.gid_guid);

			/* pass statep as cm_private */
			ret = ibt_bind_service(statep->rds_srvhdl, gid,
			    NULL, statep, &hcap->hca_bindhdl[jx]);
			if (ret != IBT_SUCCESS) {
				RDS_DPRINTF2(LABEL, "Bind service for "
				    "HCA: 0x%llx Port: %d gid %llx:%llx "
				    "failed: %d", hcap->hca_guid,
				    hcap->hca_pinfop[jx].p_port_num,
				    gid.gid_prefix, gid.gid_guid, ret);
				continue;
			}

			nbinds++;
		}
		hcap = hcap->hca_nextp;
	}

	rw_exit(&statep->rds_hca_lock);

	RDS_DPRINTF2(LABEL, "RDS Service available on %d/%d ports",
	    nbinds, nports);

#if 0
	if (nbinds == 0) {
		return (-1);
	}
#endif

	RDS_DPRINTF2("rds_bind_service", "Return");

	return (0);
}

/* Open an RC connection */
int
rds_open_rc_channel(rds_ep_t *ep, ibt_path_info_t *pinfo,
    ibt_execution_mode_t mode, ibt_channel_hdl_t *chanhdl)
{
	rds_session_t		*sp;
	ibt_chan_open_args_t	ocargs;
	ibt_rc_returns_t	ocrets;
	rds_cm_private_data_t	cmp;
	uint8_t			hca_port;
	ibt_channel_hdl_t	hdl;
	ibt_status_t		ret = 0;
	ibt_ip_cm_info_t	ipcm_info;

	RDS_DPRINTF2("rds_open_rc_channel", "Enter: EP(%p) mode: %d", ep, mode);

	sp = ep->ep_sp;

	bzero(&ipcm_info, sizeof (ibt_ip_cm_info_t));
	ipcm_info.src_addr.family = AF_INET;
	ipcm_info.SRCIP = sp->session_myip;
	ipcm_info.dst_addr.family = AF_INET;
	ipcm_info.DSTIP = sp->session_remip;
	ipcm_info.src_port = RDS_PORT_NUM;
	ret = ibt_format_ip_private_data(&ipcm_info,
	    sizeof (rds_cm_private_data_t), &cmp);
	if (ret != IBT_SUCCESS) {
		RDS_DPRINTF2(LABEL, "SP(%p) EP(%p) ibt_format_ip_private_data "
		    "failed: %d", sp, ep, ret);
		return (-1);
	}

	hca_port = pinfo->pi_prim_cep_path.cep_hca_port_num;

	hdl = rds_ep_alloc_rc_channel(ep, hca_port);
	if (hdl == NULL) {
		return (-1);
	}

	cmp.cmp_version = RDS_VERSION;
	cmp.cmp_arch = RDS_THIS_ARCH;
	cmp.cmp_eptype = ep->ep_type;
	cmp.cmp_failover = sp->session_failover;
	cmp.cmp_last_bufid = ep->ep_rbufid;
	cmp.cmp_user_buffer_size = UserBufferSize;
	cmp.cmp_ack_addr = ep->ep_ack_addr;
	cmp.cmp_ack_rkey = ep->ep_ack_rkey;

	bzero(&ocargs, sizeof (ibt_chan_open_args_t));
	bzero(&ocrets, sizeof (ibt_rc_returns_t));
	ocargs.oc_path = pinfo;
	ocargs.oc_cm_handler = rds_cm_handler;
	ocargs.oc_cm_clnt_private = NULL;
	ocargs.oc_rdma_ra_out = 4;
	ocargs.oc_rdma_ra_in = 4;
	ocargs.oc_priv_data_len = sizeof (rds_cm_private_data_t);
	ocargs.oc_priv_data = &cmp;
	ocargs.oc_path_retry_cnt = IBPathRetryCount;
	ocargs.oc_path_rnr_retry_cnt = MinRnrRetry;
	ret = ibt_open_rc_channel(hdl, IBT_OCHAN_NO_FLAGS,
	    mode, &ocargs, &ocrets);
	if (ret != IBT_SUCCESS) {
		RDS_DPRINTF2(LABEL, "SP(%p) EP(%p) ibt_open_rc_channel "
		    "failed: %d", sp, ep, ret);
		(void) ibt_flush_channel(hdl);
		(void) ibt_free_channel(hdl);

		mutex_enter(&ep->ep_lock);
		/* don't cleanup if this failure is due to peer-peer race */
		if (ep->ep_state == RDS_EP_STATE_ACTIVE_PENDING) {
			/* cleanup stuff allocated in rds_ep_alloc_rc_channel */
			ep->ep_state = RDS_EP_STATE_ERROR;
			rds_ep_free_rc_channel(ep);
		}
		mutex_exit(&ep->ep_lock);

		return (-1);
	}

	*chanhdl = hdl;

	RDS_DPRINTF2("rds_open_rc_channel", "Return: EP(%p) Chan: %p", ep,
	    *chanhdl);

	return (0);
}

int
rds_close_rc_channel(ibt_channel_hdl_t chanhdl, ibt_execution_mode_t mode)
{
	int	ret;

	RDS_DPRINTF2("rds_close_rc_channel", "Enter: Chan(%p) Mode(%d)",
	    chanhdl, mode);

	ret = ibt_close_rc_channel(chanhdl, mode, NULL, 0, NULL, NULL, 0);

	RDS_DPRINTF2("rds_close_rc_channel", "Return Chan(%p)", chanhdl);

	return (ret);
}