Merge pull request #6028 from evgeny-leksikov/uct_rc_ep_list_lock

UCT/RC: Protect rc_iface->ep_list and rc_iface->eps with a spinlock

src/uct/ib/rc/base/rc_ep.c

@@ -158,7 +158,11 @@ UCS_CLASS_INIT_FUNC(uct_rc_ep_t, uct_rc_iface_t *iface, uint32_t qp_num,
 
     ucs_arbiter_group_init(&self->arb_group);
 
+    ucs_spin_lock(&iface->eps_lock);
     ucs_list_add_head(&iface->ep_list, &self->list);
+    ucs_spin_unlock(&iface->eps_lock);
+
+    ucs_debug("created rc ep %p", self);
     return UCS_OK;
 
 err_txqp_cleanup:

src/uct/ib/rc/base/rc_iface.c

@@ -250,6 +250,7 @@ void uct_rc_iface_add_qp(uct_rc_iface_t *iface, uct_rc_ep_t *ep,
 {
     uct_rc_ep_t ***ptr, **memb;
 
+    ucs_spin_lock(&iface->eps_lock);
     ptr = &iface->eps[qp_num >> UCT_RC_QP_TABLE_ORDER];
     if (*ptr == NULL) {
         *ptr = ucs_calloc(UCS_BIT(UCT_RC_QP_TABLE_MEMB_ORDER), sizeof(**ptr),
@@ -259,16 +260,19 @@ void uct_rc_iface_add_qp(uct_rc_iface_t *iface, uct_rc_ep_t *ep,
     memb = &(*ptr)[qp_num &  UCS_MASK(UCT_RC_QP_TABLE_MEMB_ORDER)];
     ucs_assert(*memb == NULL);
     *memb = ep;
+    ucs_spin_unlock(&iface->eps_lock);
 }
 
 void uct_rc_iface_remove_qp(uct_rc_iface_t *iface, unsigned qp_num)
 {
     uct_rc_ep_t **memb;
 
+    ucs_spin_lock(&iface->eps_lock);
     memb = &iface->eps[qp_num >> UCT_RC_QP_TABLE_ORDER]
                       [qp_num &  UCS_MASK(UCT_RC_QP_TABLE_MEMB_ORDER)];
     ucs_assert(*memb != NULL);
     *memb = NULL;
+    ucs_spin_unlock(&iface->eps_lock);
 }
 
 ucs_status_t uct_rc_iface_flush(uct_iface_h tl_iface, unsigned flags,
@@ -289,14 +293,17 @@ ucs_status_t uct_rc_iface_flush(uct_iface_h tl_iface, unsigned flags,
     }
 
     count = 0;
+    ucs_spin_lock(&iface->eps_lock);
     ucs_list_for_each(ep, &iface->ep_list, list) {
         status = uct_ep_flush(&ep->super.super, 0, NULL);
         if ((status == UCS_ERR_NO_RESOURCE) || (status == UCS_INPROGRESS)) {
             ++count;
         } else if (status != UCS_OK) {
+            ucs_spin_unlock(&iface->eps_lock);
             return status;
         }
     }
+    ucs_spin_unlock(&iface->eps_lock);
 
     if (count != 0) {
         UCT_TL_IFACE_STAT_FLUSH_WAIT(&iface->super.super);
@@ -569,6 +576,12 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_rc_iface_ops_t *ops, uct_md_h md,
     self->tx.reads_completed = 0;
 
     uct_ib_fence_info_init(&self->tx.fi);
+
+    status = ucs_spinlock_init(&self->eps_lock, 0);
+    if (status != UCS_OK) {
+        goto err;
+    }
+
     memset(self->eps, 0, sizeof(self->eps));
     ucs_arbiter_init(&self->tx.arbiter);
     ucs_list_head_init(&self->ep_list);
@@ -586,7 +599,7 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_rc_iface_ops_t *ops, uct_md_h md,
     status = uct_ib_iface_recv_mpool_init(&self->super, &config->super,
                                           "rc_recv_desc", &self->rx.mp);
     if (status != UCS_OK) {
-        goto err;
+        goto err_destroy_eps_lock;
     }
 
     /* Create TX buffers mempool */
@@ -668,6 +681,8 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_rc_iface_ops_t *ops, uct_md_h md,
     ucs_mpool_cleanup(&self->tx.mp, 1);
 err_destroy_rx_mp:
     ucs_mpool_cleanup(&self->rx.mp, 1);
+err_destroy_eps_lock:
+    ucs_spinlock_destroy(&self->eps_lock);
 err:
     return status;
 }
@@ -702,6 +717,7 @@ static UCS_CLASS_CLEANUP_FUNC(uct_rc_iface_t)
 
     UCS_STATS_NODE_FREE(self->stats);
 
+    ucs_spinlock_destroy(&self->eps_lock);
     ops->cleanup_rx(self);
     uct_rc_iface_tx_ops_cleanup(self);
     ucs_mpool_cleanup(&self->tx.mp, 1);

src/uct/ib/rc/base/rc_iface.h

@@ -271,6 +271,7 @@ struct uct_rc_iface {
 
     UCS_STATS_NODE_DECLARE(stats)
 
+    ucs_spinlock_t           eps_lock; /* common lock for eps and ep_list */
     uct_rc_ep_t              **eps[UCT_RC_QP_TABLE_SIZE];
     ucs_list_link_t          ep_list;
     ucs_list_link_t          ep_gc_list;