cma: add support for IB collective providers

Add collective member address and threading information
on a per transport basis. Call create/free service with
HCA open/close.

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

diff --git a/dapl/openib_cma/dapl_ib_util.h b/dapl/openib_cma/dapl_ib_util.h
index 471bd7ffe51cfac5208f0620674377148d304db2..454f7e159cded120b697c9cfebfcaab9da508768 100755 (executable)
--- a/dapl/openib_cma/dapl_ib_util.h
+++ b/dapl/openib_cma/dapl_ib_util.h
@@ -124,6 +124,23 @@ typedef struct _ib_hca_transport
        uint8_t                 sl;
        uint16_t                pkey;
        int                     pkey_idx;
+#ifdef DAT_IB_COLLECTIVES
+       /* Collective member device and address information */
+       ib_thread_state_t       coll_thread_state;
+       DAPL_OS_THREAD          coll_thread;
+       DAPL_OS_LOCK            coll_lock;
+       DAPL_OS_WAIT_OBJECT     coll_event;
+       struct dapl_llist_entry *grp_list;
+       user_progress_func_t    *user_func;
+       int                     l_sock;
+       struct sockaddr_in      m_addr;
+       void                    *m_ctx;
+       void                    *m_info;
+       void                    *f_info;
+       int                     m_size;
+       int                     f_size;
+       int                     t_id;
+#endif
 } ib_hca_transport_t;
 
 /* prototypes */

diff --git a/dapl/openib_cma/device.c b/dapl/openib_cma/device.c
index 454c394f50151e1655a0aa387978bf3f5790a494..d1a3ab635a81b32d246f3d2b2622510e8642cbf2 100644 (file)
--- a/dapl/openib_cma/device.c
+++ b/dapl/openib_cma/device.c
@@ -47,6 +47,10 @@ static const char rcsid[] = "$Id:  $";
 
 #include <stdlib.h>
 
+#ifdef DAT_IB_COLLECTIVES
+#include <collectives/ib_collectives.h>
+#endif
+
 struct rdma_event_channel *g_cm_events = NULL;
 ib_thread_state_t g_ib_thread_state = 0;
 DAPL_OS_THREAD g_ib_thread;
@@ -376,6 +380,11 @@ DAT_RETURN dapls_ib_open_hca(IN IB_HCA_NAME hca_name, IN DAPL_HCA * hca_ptr)
                     &hca_ptr->hca_address)->sin_addr.s_addr >> 24 & 0xff, 
                     hca_ptr->ib_trans.max_inline_send);
 
+#ifdef DAT_IB_COLLECTIVES
+       if (dapli_create_collective_service(hca_ptr))
+               return DAT_INTERNAL_ERROR;
+#endif
+
        return DAT_SUCCESS;
 }
 
@@ -400,6 +409,10 @@ DAT_RETURN dapls_ib_close_hca(IN DAPL_HCA * hca_ptr)
        dapl_dbg_log(DAPL_DBG_TYPE_UTIL, " close_hca: %p->%p\n",
                     hca_ptr, hca_ptr->ib_hca_handle);
 
+#ifdef DAT_IB_COLLECTIVES
+       dapli_free_collective_service(hca_ptr);
+#endif
+
        dapl_os_lock(&g_hca_lock);
        if (g_ib_thread_state != IB_THREAD_RUN) {
                dapl_os_unlock(&g_hca_lock);