[rocketmq-client-cpp] branch master updated: Optimize transport layer (#165)

[di...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dinglei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-client-cpp.git


The following commit(s) were added to refs/heads/master by this push:
     new ce126b7  Optimize transport layer (#165)
ce126b7 is described below

commit ce126b723672d1130ed3660d02d61057e5af5f5c
Author: James Yin <yw...@hotmail.com>
AuthorDate: Mon Nov 11 09:51:24 2019 +0800

    Optimize transport layer (#165)
    
    * always addTimerCallback for avoid memory leak when pass NULL cbw and not response.
---
 src/transport/TcpRemotingClient.cpp | 13 ++++++-------
 src/transport/TcpTransport.cpp      |  7 ++++---
 2 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/src/transport/TcpRemotingClient.cpp b/src/transport/TcpRemotingClient.cpp
index cd7fbfa..0207bbe 100755
--- a/src/transport/TcpRemotingClient.cpp
+++ b/src/transport/TcpRemotingClient.cpp
@@ -242,13 +242,12 @@ bool TcpRemotingClient::invokeAsync(const string& addr,
     responseFuture->setRequestCommand(request);
     addResponseFuture(opaque, responseFuture);
 
-    if (callback) {
-      boost::asio::deadline_timer* t =
-          new boost::asio::deadline_timer(m_timerService, boost::posix_time::milliseconds(timeoutMillis));
-      addTimerCallback(t, opaque);
-      t->async_wait(
-          boost::bind(&TcpRemotingClient::handleAsyncRequestTimeout, this, boost::asio::placeholders::error, opaque));
-    }
+    // timeout monitor
+    boost::asio::deadline_timer* t =
+        new boost::asio::deadline_timer(m_timerService, boost::posix_time::milliseconds(timeoutMillis));
+    addTimerCallback(t, opaque);
+    t->async_wait(
+        boost::bind(&TcpRemotingClient::handleAsyncRequestTimeout, this, boost::asio::placeholders::error, opaque));
 
     // even if send failed, asyncTimerThread will trigger next pull request or report send msg failed
     if (SendCommand(pTcp, request)) {
diff --git a/src/transport/TcpTransport.cpp b/src/transport/TcpTransport.cpp
index 884304c..641c516 100644
--- a/src/transport/TcpTransport.cpp
+++ b/src/transport/TcpTransport.cpp
@@ -67,9 +67,10 @@ TcpConnectStatus TcpTransport::getTcpConnectStatus() {
 }
 
 TcpConnectStatus TcpTransport::waitTcpConnectEvent(int timeoutMillis) {
-  std::unique_lock<std::mutex> eventLock(m_connectEventLock);
   if (m_tcpConnectStatus == TCP_CONNECT_STATUS_WAIT) {
-    if (m_connectEvent.wait_for(eventLock, std::chrono::milliseconds(timeoutMillis)) == std::cv_status::timeout) {
+    std::unique_lock<std::mutex> eventLock(m_connectEventLock);
+    if (!m_connectEvent.wait_for(eventLock, std::chrono::milliseconds(timeoutMillis),
+                                 [&] { return m_tcpConnectStatus != TCP_CONNECT_STATUS_WAIT; })) {
       LOG_INFO("connect timeout");
     }
   }
@@ -80,7 +81,7 @@ TcpConnectStatus TcpTransport::waitTcpConnectEvent(int timeoutMillis) {
 void TcpTransport::setTcpConnectEvent(TcpConnectStatus connectStatus) {
   TcpConnectStatus baseStatus = m_tcpConnectStatus.exchange(connectStatus, std::memory_order_relaxed);
   if (baseStatus == TCP_CONNECT_STATUS_WAIT) {
-    std::unique_lock<std::mutex> eventLock(m_connectEventLock);
+    // awake waiting thread
     m_connectEvent.notify_all();
   }
 }