UCP/WIREUP/GTEST: Fix dead code in CM disconnect

src/ucp/wireup/wireup_cm.c

@@ -638,10 +638,9 @@ static void ucp_ep_cm_disconnect_flushed_cb(ucp_request_t *req)
     UCS_ASYNC_UNBLOCK(async);
 }
 
-static unsigned ucp_ep_cm_remote_disconnect_progress(void *arg)
+static unsigned ucp_ep_cm_remote_disconnect_progress(ucp_ep_h ucp_ep)
 {
     ucs_status_t status = UCS_ERR_CONNECTION_RESET;
-    ucp_ep_h ucp_ep     = arg;
     void *req;
 
     ucs_trace("ep %p: flags 0x%x cm_remote_disconnect_progress", ucp_ep,
@@ -663,15 +662,18 @@ static unsigned ucp_ep_cm_remote_disconnect_progress(void *arg)
          * @ref ucp_ep_close_flushed_callback */
         ucs_debug("ep %p: ep closed but request is not set, waiting for"
                   " the flush callback", ucp_ep);
-        goto err;
+        return 1;
     }
 
     if (!(ucp_ep->flags & UCP_EP_FLAG_REMOTE_CONNECTED)) {
         /* CM disconnect happens during WIREUP MSGs exchange phase, when EP
          * is locally connected to the peer */
-        goto err;
+        goto set_ep_failed;
     }
 
+    /* if the EP is local and remote connected, need to flush it from main
+     * thread first */
+
     /*
      * TODO: set the ucp_ep to error state to prevent user from sending more
      *       ops.
@@ -689,12 +691,13 @@ static unsigned ucp_ep_cm_remote_disconnect_progress(void *arg)
         status = UCS_PTR_STATUS(req);
         ucs_error("ucp_ep_flush_internal completed with error: %s",
                   ucs_status_string(status));
-        goto err;
+        goto set_ep_failed;
     }
 
+    ucp_ep_invoke_err_cb(ucp_ep, UCS_ERR_CONNECTION_RESET);
     return 1;
 
-err:
+set_ep_failed:
     ucp_worker_set_ep_failed(ucp_ep->worker, ucp_ep,
                              ucp_ep_get_cm_uct_ep(ucp_ep),
                              ucp_ep_get_cm_lane(ucp_ep), status);
@@ -714,8 +717,6 @@ static unsigned ucp_ep_cm_disconnect_progress(void *arg)
               uct_cm_ep, ucp_ep->flags);
     ucs_assert(ucp_ep_get_cm_uct_ep(ucp_ep) == uct_cm_ep);
 
-    ucp_ep->flags &= ~UCP_EP_FLAG_REMOTE_CONNECTED;
-
     if (ucp_ep->flags & UCP_EP_FLAG_FAILED) {
         /* - ignore close event on failed ep, since all lanes are destroyed in
              generic err flow
@@ -738,6 +739,8 @@ static unsigned ucp_ep_cm_disconnect_progress(void *arg)
                  ucp_ep, ucp_ep->flags);
     }
 
+    ucp_ep->flags &= ~UCP_EP_FLAG_REMOTE_CONNECTED;
+
     UCS_ASYNC_UNBLOCK(async);
     return 1;
 }

test/gtest/ucp/test_ucp_sockaddr.cc

@@ -61,6 +61,8 @@ class test_ucp_sockaddr : public ucp_test {
     ucs::sock_addr_storage m_test_addr;
 
     void init() {
+        m_err_count = 0;
+
         if (get_variant_value() & TEST_MODIFIER_CM) {
             modify_config("SOCKADDR_CM_ENABLE", "yes");
         }
@@ -398,14 +400,14 @@ class test_ucp_sockaddr : public ucp_test {
     virtual ucp_ep_params_t get_ep_params()
     {
         ucp_ep_params_t ep_params = ucp_test::get_ep_params();
-        ep_params.field_mask      |= UCP_EP_PARAM_FIELD_ERR_HANDLING_MODE |
-                                     UCP_EP_PARAM_FIELD_ERR_HANDLER;
+        ep_params.field_mask     |= UCP_EP_PARAM_FIELD_ERR_HANDLING_MODE |
+                                    UCP_EP_PARAM_FIELD_ERR_HANDLER;
         /* The error handling requirement is needed since we need to take
          * care of a case where the client gets an error. In case ucp needs to
          * handle a large worker address but neither ud nor ud_x are present */
-        ep_params.err_mode         = UCP_ERR_HANDLING_MODE_PEER;
-        ep_params.err_handler.cb   = err_handler_cb;
-        ep_params.err_handler.arg  = NULL;
+        ep_params.err_mode        = UCP_ERR_HANDLING_MODE_PEER;
+        ep_params.err_handler.cb  = err_handler_cb;
+        ep_params.err_handler.arg = NULL;
         return ep_params;
     }
 
@@ -512,6 +514,8 @@ class test_ucp_sockaddr : public ucp_test {
     static void err_handler_cb(void *arg, ucp_ep_h ep, ucs_status_t status) {
         ucp_test::err_handler_cb(arg, ep, status);
 
+        ++m_err_count;
+
         /* The current expected errors are only from the err_handle test
          * and from transports where the worker address is too long but ud/ud_x
          * are not present, or ud/ud_x are present but their addresses are too
@@ -566,8 +570,14 @@ class test_ucp_sockaddr : public ucp_test {
                     ((lane1 != UCP_NULL_LANE) && (lane2 != UCP_NULL_LANE) &&
                      ucp_ep_config_lane_is_peer_equal(key1, lane1, key2, lane2)));
     }
+
+protected:
+    static unsigned m_err_count;
 };
 
+unsigned test_ucp_sockaddr::m_err_count = 0;
+
+
 UCS_TEST_SKIP_COND_P(test_ucp_sockaddr, listen, no_close_protocol()) {
     listen_and_communicate(false, 0);
 }
@@ -606,6 +616,16 @@ UCS_TEST_P(test_ucp_sockaddr, onesided_disconnect_bidi) {
     one_sided_disconnect(sender(), UCP_EP_CLOSE_MODE_FLUSH);
 }
 
+UCS_TEST_SKIP_COND_P(test_ucp_sockaddr, onesided_disconnect_bidi_wait_err_cb,
+                     no_close_protocol()) {
+    listen_and_communicate(false, SEND_DIRECTION_BIDI);
+
+    one_sided_disconnect(sender(), UCP_EP_CLOSE_MODE_FLUSH);
+
+    wait_for_flag(&m_err_count);
+    EXPECT_EQ(1, m_err_count);
+}
+
 UCS_TEST_SKIP_COND_P(test_ucp_sockaddr, concurrent_disconnect,
                      no_close_protocol()) {
     listen_and_communicate(false, 0);