mpxyd: need CM to QP linking with CM references

Complete coding support for ref_cnt on CM to allow for
proper destruction of CM resourses. Ref count for CM alloc,
QP linking, and queue list. List dequeue will trigger CM
free, move to destroy state, and dealloc if ref_cnt is zero.

Signed-off-by: Arlin Davis <arlin.r.davis@intel.com>

diff --git a/dapl/svc/mcm.c b/dapl/svc/mcm.c
index 3e389157581badc34f144e6a7c0ae5c2894b7197..2034ad0d7396afa1dd3c403b1c78f55fee4c6bdd 100644 (file)
--- a/dapl/svc/mcm.c
+++ b/dapl/svc/mcm.c
@@ -443,15 +443,29 @@ void mcm_flush_qp(struct mcm_qp *m_qp)
 /* MCM Endpoint CM objects */
 void m_cm_free(mcm_cm_t *cm)
 {
+       mlog(2, "CM %p free: qp %p ref_cnt %d state %s sid %x\n",
+               cm, cm->m_qp, cm->ref_cnt,
+               mcm_state_str(cm->state), cm->sid);
+
+       mpxy_lock(&cm->lock);
+       if (cm->state != MCM_DESTROY) {
+               cm->ref_cnt--; /* alloc ref */
+               cm->state = MCM_DESTROY;
+       }
+
        /* client, release local conn id port */
        if (!cm->l_ep && cm->sid) {
                mcm_free_port(cm->md->ports, cm->sid);
                cm->sid = 0;
        }
-
-       mpxy_lock_destroy(&cm->lock);
-       cm->smd->ref_cnt--;
-       free(cm);
+       /* last reference, destroy */
+       if (!cm->ref_cnt) {
+               mpxy_unlock(&cm->lock);
+               mpxy_lock_destroy(&cm->lock);
+               cm->smd->ref_cnt--;
+               free(cm);
+       } else
+               mpxy_unlock(&cm->lock);
 }
 
 mcm_cm_t *m_cm_create(mcm_scif_dev_t *smd, mcm_qp_t *m_qp, dat_mcm_addr_t *r_addr)
@@ -471,6 +485,7 @@ mcm_cm_t *m_cm_create(mcm_scif_dev_t *smd, mcm_qp_t *m_qp, dat_mcm_addr_t *r_add
        if (mpxy_lock_init(&cm->lock, NULL))
                goto bail;
 
+       cm->ref_cnt++; /* alloc ref */
        cm->state = MCM_INIT;
        cm->smd = smd;
        cm->md = smd->md;
@@ -486,6 +501,7 @@ mcm_cm_t *m_cm_create(mcm_scif_dev_t *smd, mcm_qp_t *m_qp, dat_mcm_addr_t *r_add
                        mpxy_lock_destroy(&cm->lock);
                        goto bail;
                }
+               cm->ref_cnt++; /* Active: QP ref */
                cm->m_qp = m_qp;
                m_qp->cm = cm;
 
@@ -546,6 +562,10 @@ bail:
 /* queue up connection object on CM list */
 void mcm_qconn(mcm_scif_dev_t *smd, mcm_cm_t *cm)
 {
+       mpxy_lock(&cm->lock);
+       cm->ref_cnt++; /* clist ref */
+       mpxy_unlock(&cm->lock);
+
        /* add to CONN work queue, list, for mcm fabric CM */
        mpxy_lock(&smd->clock);
        insert_tail(&cm->entry, &smd->clist, (void *)cm);
@@ -559,10 +579,19 @@ void mcm_dqconn(mcm_scif_dev_t *smd, mcm_cm_t *cm)
        remove_entry(&cm->entry);
        mpxy_unlock(&smd->clock);
 
+       mpxy_lock(&cm->lock);
+       cm->ref_cnt--; /* clist ref */
+       mpxy_unlock(&cm->lock);
+
+       m_cm_free(cm);
 }
 /* queue listen object on listen list */
 void mcm_qlisten(mcm_scif_dev_t *smd, mcm_cm_t *cm)
 {
+       mpxy_lock(&cm->lock);
+       cm->ref_cnt++; /* llist ref */
+       mpxy_unlock(&cm->lock);
+
        /* add to LISTEN work queue, list, for mcm fabric CM */
        mpxy_lock(&smd->llock);
        insert_tail(&cm->entry, &smd->llist, (void *)cm);
@@ -575,8 +604,13 @@ void mcm_dqlisten(mcm_scif_dev_t *smd, mcm_cm_t *cm)
        remove_entry(&cm->entry);
        mcm_free_port(smd->md->ports, cm->sid);
        cm->sid = 0;
-       m_cm_free(cm);
        mpxy_unlock(&smd->llock);
+
+       mpxy_lock(&cm->lock);
+       cm->ref_cnt--; /* llist ref */
+       mpxy_unlock(&cm->lock);
+
+       m_cm_free(cm);
 }
 
 /*
@@ -1008,6 +1042,7 @@ static void mcm_process_recv(mcm_ib_dev_t *md, dat_mcm_msg_t *msg, mcm_cm_t *cm,
                break;
        case MCM_DISCONNECTED:
        case MCM_FREE:
+       case MCM_DESTROY:
                /* DREQ dropped, resend */
                if (ntohs(msg->op) == MCM_DREQ) {
                        MCNTR(md, MCM_CM_DREQ_DUP);
@@ -1043,7 +1078,7 @@ static void mcm_process_recv(mcm_ib_dev_t *md, dat_mcm_msg_t *msg, mcm_cm_t *cm,
                        break;
                }
        default:
-               mlog(8, " mcm_recv: Warning, UNKNOWN state"
+               mlog(2, " mcm_recv: Warning, UNKNOWN state"
                        " <- op %s, %s spsp %x sqpn %x slid %x\n",
                        mcm_op_str(ntohs(msg->op)), mcm_state_str(cm->state),
                        ntohs(msg->sport), ntohl(msg->sqpn), ntohs(msg->saddr1.lid));
@@ -1208,7 +1243,7 @@ mcm_cm_t *mcm_get_cm(mcm_ib_dev_t *md, dat_mcm_msg_t *msg)
        mpxy_unlock(&md->slock);
 
        if (!cm && !dup) {
-               mlog(1,  " %s - op %s [lid, port, cqp, iqp, pid]:"
+               mlog(2,  " %s - op %s [lid, port, cqp, iqp, pid]:"
                         " %x %x %x %x %x <- %x %x %x %x lpid %x rpid %x\n",
                         ntohs(msg->op) == MCM_REQ ? "NO LISTENER":"NO MATCH",
                         mcm_op_str(ntohs(msg->op)),

diff --git a/dapl/svc/mix.c b/dapl/svc/mix.c
index c7e1dadf4f984ed33ab31fa28bd8cfc1a988f201..8774e5d34a1dbca9d9cbe038ee41896336e2398f 100644 (file)
--- a/dapl/svc/mix.c
+++ b/dapl/svc/mix.c
@@ -498,11 +498,18 @@ void m_qp_free(struct mcm_qp *m_qp)
        struct ibv_qp *qp1 = m_qp->ib_qp1;
        struct ibv_qp *qp2 = m_qp->ib_qp2;
 
+
        m_qp->ib_qp1 = NULL;
        m_qp->ib_qp2 = NULL;
-       if (m_qp->cm) { /* unlink CM */
+       if (m_qp->cm) { /* unlink CM, serialized */
+               struct mcm_cm *cm = m_qp->cm;
+
+               mpxy_lock(&cm->lock);
+               m_qp->cm->ref_cnt--; /* QP ref */
                m_qp->cm->m_qp = NULL;
                m_qp->cm = NULL;
+               mpxy_unlock(&cm->lock);
+               mcm_dqconn(m_qp->smd, cm);
        }
        if (qp1) {
                mcm_modify_qp(qp1, IBV_QPS_ERR, 0, 0, NULL);
@@ -1484,8 +1491,8 @@ static int mix_cm_rep_out(mcm_scif_dev_t *smd, dat_mix_cm_t *pmsg, scif_epd_t sc
        struct mcm_cm *m_cm;
        struct ibv_qp *qp, *qp2 = NULL;
        union ibv_gid *dgid = NULL, *dgid2 = NULL;
-       uint32_t dqpn, dqpn2 = 0;
-       uint16_t dlid, dlid2 = 0;
+       uint32_t dqpn = 0, dqpn2 = 0;
+       uint16_t dlid = 0, dlid2 = 0;
 
        /* hdr already read, get operation data */
        len = sizeof(dat_mix_cm_t) - sizeof(dat_mix_hdr_t);
@@ -1503,7 +1510,7 @@ static int mix_cm_rep_out(mcm_scif_dev_t *smd, dat_mix_cm_t *pmsg, scif_epd_t sc
                mlog(0, " ERR: mix_get_cm, id %d, not found\n", pmsg->cm_id);
                return -1;
        }
-
+       mpxy_lock(&m_cm->lock);
        /* update CM message from MIX client, save clients id,ctx */
        m_cm->cm_id = 0; /* no client id for now, just ctx */
        m_cm->cm_ctx = pmsg->sp_ctx;
@@ -1515,8 +1522,10 @@ static int mix_cm_rep_out(mcm_scif_dev_t *smd, dat_mix_cm_t *pmsg, scif_epd_t sc
        m_cm->m_qp = mix_get_qp(smd, pmsg->qp_id, 0);
        if (!m_cm->m_qp) {
                mlog(0, " ERR: mix_get_qp, id %d, not found\n", pmsg->qp_id);
+               mpxy_unlock(&m_cm->lock);
                return -1;
        }
+       m_cm->ref_cnt++; /* Passive: QP ref */
        m_cm->m_qp->cm = m_cm;
        mcm_save_wrc(m_cm); /* save remote proxy-in WRC QP info */
 
@@ -1529,10 +1538,10 @@ static int mix_cm_rep_out(mcm_scif_dev_t *smd, dat_mix_cm_t *pmsg, scif_epd_t sc
                mlog(2, " MXS -> MSS remote \n");
 
                if (m_qp_create_pi(smd, m_cm->m_qp))
-                       return -1;
+                       goto err;
 
                if (m_pi_prep_rcv_q(m_cm->m_qp))
-                       return -1;
+                       goto err;
 
                /* KR to KL or XEON, QP1<-QP2 and QP2->QP1 */
                /* update the src information in CM msg */
@@ -1571,7 +1580,7 @@ static int mix_cm_rep_out(mcm_scif_dev_t *smd, dat_mix_cm_t *pmsg, scif_epd_t sc
                mlog(2, " MXS -> MXS remote \n");
 
                if (m_pi_prep_rcv_q(m_cm->m_qp))
-                       return -1;
+                       goto err;
 
                /* update the QPt src information in CM msg */
                m_cm->msg.saddr1.ep_map = MIC_XSOCK_DEV;
@@ -1627,7 +1636,7 @@ static int mix_cm_rep_out(mcm_scif_dev_t *smd, dat_mix_cm_t *pmsg, scif_epd_t sc
                        goto err;
 
                if (m_pi_prep_rcv_q(m_cm->m_qp))
-                       return -1;
+                       goto err;
        }
        mcm_init_wrc(m_cm); /* send back proxy-in WR/WC raddr,rkey info */
        mcm_pr_addrs(2, &m_cm->msg, m_cm->state, 0);
@@ -1652,6 +1661,7 @@ static int mix_cm_rep_out(mcm_scif_dev_t *smd, dat_mix_cm_t *pmsg, scif_epd_t sc
 
        /* send RTU on wire, monitor for retries */
        m_cm->state = MCM_RTU_PENDING;
+       mpxy_unlock(&m_cm->lock);
        mcm_cm_rep_out(m_cm);
        return 0;
 err:
@@ -1666,6 +1676,7 @@ err:
              ntohl(dqpn), ntohs(dlid));
 
        mcm_pr_addrs(0, &m_cm->msg, m_cm->state, 0);
+       mpxy_unlock(&m_cm->lock);
        return -1;
 }
 

diff --git a/dapl/svc/mpxyd.c b/dapl/svc/mpxyd.c
index abbfac0e150e72cdca3342ea4224872ed772cade..2f7f6ced7b19ef34e0cf884138c79ad7699d5a44 100644 (file)
--- a/dapl/svc/mpxyd.c
+++ b/dapl/svc/mpxyd.c
@@ -310,9 +310,7 @@ void mpxy_destroy_smd(mcm_scif_dev_t *smd)
        m_cm = get_head_entry(&smd->llist);
        while (m_cm) {
                next_cm = get_next_entry(&m_cm->entry, &smd->llist);
-               if (m_cm->sid)
-                       mcm_free_port(smd->md->ports, m_cm->sid);
-               m_cm_free(m_cm);
+               mcm_dqlisten(smd, m_cm); /* dequeue and free */
                m_cm = next_cm;
        }
        init_list(&smd->llist);
@@ -324,7 +322,7 @@ void mpxy_destroy_smd(mcm_scif_dev_t *smd)
        m_cm = get_head_entry(&smd->clist);
        while (m_cm) {
                next_cm = get_next_entry(&m_cm->entry, &smd->clist);
-               m_cm_free(m_cm);
+               mcm_dqconn(smd, m_cm); /* dequeue and free */
                m_cm = next_cm;
        }
        init_list(&smd->clist);

diff --git a/dapl/svc/mpxyd.h b/dapl/svc/mpxyd.h
index a81be03b7ee03347ef6b1485b3fe92e94580e099..832ab9d8d38faf4f1d78aa61652148c73cddd630 100644 (file)
--- a/dapl/svc/mpxyd.h
+++ b/dapl/svc/mpxyd.h
@@ -418,7 +418,7 @@ typedef struct mcm_mr {
 /*  DAPL MCM Connection/Listen object */
 typedef struct mcm_cm {
        LLIST_ENTRY             entry;
-       mpxy_lock_t     lock;
+       mpxy_lock_t             lock;
        struct mcm_ib_dev       *md;    /* mcm_ib_dev parent reference */
        struct mcm_scif_dev     *smd;   /* mcm_scif_dev parent reference */
        struct mcm_cm           *l_ep;  /* listen reference, passive */
@@ -671,7 +671,6 @@ void mcm_dump_cm_lists(mcm_scif_dev_t *smd);
 /* mix.c, MIC message exchange (MIX) services */
 void m_cq_free(struct mcm_cq *m_cq);
 void m_qp_free(struct mcm_qp *m_qp);
-void m_cm_free(struct mcm_cm *m_cm);
 void m_mr_free(struct mcm_mr *m_mr);
 int mix_scif_recv(mcm_scif_dev_t *smd, scif_epd_t scif_ep);
 int mix_cm_disc_in(mcm_cm_t *m_cm);