remove boost mutex and condition_variable in TcpTransport and ReponseFunture.

Author: ifplusor

src/transport/ResponseFuture.cpp

@@ -15,6 +15,9 @@
  * limitations under the License.
  */
 #include "ResponseFuture.h"
+
+#include <chrono>
+
 #include "Logging.h"
 #include "TcpRemotingClient.h"
 
@@ -52,12 +55,12 @@ void ResponseFuture::releaseThreadCondition() {
 }
 
 RemotingCommand *ResponseFuture::waitResponse(int timeoutMillis) {
-  boost::unique_lock<boost::mutex> eventLock(m_defaultEventLock);
+  std::unique_lock<std::mutex> eventLock(m_defaultEventLock);
   if (!m_haveResponse) {
     if (timeoutMillis <= 0) {
       timeoutMillis = m_timeout;
     }
-    if (!m_defaultEvent.timed_wait(eventLock, boost::posix_time::milliseconds(timeoutMillis))) {
+    if (m_defaultEvent.wait_for(eventLock, std::chrono::milliseconds(timeoutMillis)) == std::cv_status::timeout) {
       LOG_WARN("waitResponse of code:%d with opaque:%d timeout", m_requestCode, m_opaque);
       m_haveResponse = true;
     }
@@ -66,7 +69,7 @@ RemotingCommand *ResponseFuture::waitResponse(int timeoutMillis) {
 }
 
 bool ResponseFuture::setResponse(RemotingCommand *pResponseCommand) {
-  boost::unique_lock<boost::mutex> eventLock(m_defaultEventLock);
+  std::unique_lock<std::mutex> eventLock(m_defaultEventLock);
 
   if (m_haveResponse) {
     return false;

src/transport/ResponseFuture.h

@@ -17,8 +17,8 @@
 #ifndef __RESPONSEFUTURE_H__
 #define __RESPONSEFUTURE_H__
 
-#include <boost/atomic.hpp>
-#include <boost/thread/condition_variable.hpp>
+#include <atomic>
+#include <condition_variable>
 
 #include "AsyncCallbackWrap.h"
 #include "RemotingCommand.h"
@@ -77,11 +77,11 @@ class ResponseFuture {
   AsyncCallbackWrap *m_pCallbackWrap;
 
   AsyncCallbackStatus m_asyncCallbackStatus;
-  boost::mutex m_asyncCallbackLock;
+  std::mutex m_asyncCallbackLock;
 
   bool m_haveResponse;
-  boost::mutex m_defaultEventLock;
-  boost::condition_variable_any m_defaultEvent;
+  std::mutex m_defaultEventLock;
+  std::condition_variable m_defaultEvent;
 
   int64 m_beginTimestamp;
   bool m_sendRequestOK;

src/transport/TcpTransport.cpp

@@ -16,6 +16,8 @@
  */
 #include "TcpTransport.h"
 
+#include <chrono>
+
 #ifndef WIN32
 #include <arpa/inet.h>  // for sockaddr_in and inet_ntoa...
 #include <netinet/tcp.h>
@@ -65,9 +67,9 @@ TcpConnectStatus TcpTransport::getTcpConnectStatus() {
 }
 
 TcpConnectStatus TcpTransport::waitTcpConnectEvent(int timeoutMillis) {
-  boost::unique_lock<boost::mutex> eventLock(m_connectEventLock);
+  std::unique_lock<std::mutex> eventLock(m_connectEventLock);
   if (m_tcpConnectStatus == TCP_CONNECT_STATUS_WAIT) {
-    if (!m_connectEvent.timed_wait(eventLock, boost::posix_time::milliseconds(timeoutMillis))) {
+    if (m_connectEvent.wait_for(eventLock, std::chrono::milliseconds(timeoutMillis)) == std::cv_status::timeout) {
       LOG_INFO("connect timeout");
     }
   }
@@ -76,16 +78,16 @@ TcpConnectStatus TcpTransport::waitTcpConnectEvent(int timeoutMillis) {
 
 // internal method
 void TcpTransport::setTcpConnectEvent(TcpConnectStatus connectStatus) {
-  TcpConnectStatus baseStatus = m_tcpConnectStatus.exchange(connectStatus, boost::memory_order_relaxed);
+  TcpConnectStatus baseStatus = m_tcpConnectStatus.exchange(connectStatus, std::memory_order_relaxed);
   if (baseStatus == TCP_CONNECT_STATUS_WAIT) {
-    boost::unique_lock<boost::mutex> eventLock(m_connectEventLock);
+    std::unique_lock<std::mutex> eventLock(m_connectEventLock);
     m_connectEvent.notify_all();
   }
 }
 
 void TcpTransport::disconnect(const string &addr) {
   // disconnect is idempotent.
-  boost::lock_guard<boost::mutex> lock(m_eventLock);
+  std::lock_guard<std::mutex> lock(m_eventLock);
   if (getTcpConnectStatus() != TCP_CONNECT_STATUS_INIT) {
     LOG_INFO("disconnect:%s start. event:%p", addr.c_str(), m_event.get());
     freeBufferEvent();
@@ -104,7 +106,7 @@ TcpConnectStatus TcpTransport::connect(const string &strServerURL, int timeoutMi
   // TODO: call DNS for hostname
 
   {
-    boost::lock_guard<boost::mutex> lock(m_eventLock);
+    std::lock_guard<std::mutex> lock(m_eventLock);
 
     struct sockaddr_in sin;
     memset(&sin, 0, sizeof(sin));
@@ -135,7 +137,7 @@ TcpConnectStatus TcpTransport::connect(const string &strServerURL, int timeoutMi
   if (connectStatus != TCP_CONNECT_STATUS_SUCCESS) {
     LOG_WARN("can not connect to server:%s", strServerURL.c_str());
 
-    boost::lock_guard<boost::mutex> lock(m_eventLock);
+    std::lock_guard<std::mutex> lock(m_eventLock);
     freeBufferEvent();
     setTcpConnectStatus(TCP_CONNECT_STATUS_FAILED);
     return TCP_CONNECT_STATUS_FAILED;
@@ -228,7 +230,7 @@ void TcpTransport::messageReceived(const MemoryBlock &mem, const std::string &ad
 }
 
 bool TcpTransport::sendMessage(const char *pData, size_t len) {
-  boost::lock_guard<boost::mutex> lock(m_eventLock);
+  std::lock_guard<std::mutex> lock(m_eventLock);
   if (getTcpConnectStatus() != TCP_CONNECT_STATUS_SUCCESS) {
     return false;
   }
@@ -241,7 +243,7 @@ bool TcpTransport::sendMessage(const char *pData, size_t len) {
 }
 
 const string TcpTransport::getPeerAddrAndPort() {
-  boost::lock_guard<boost::mutex> lock(m_eventLock);
+  std::lock_guard<std::mutex> lock(m_eventLock);
   return m_event ? m_event->getPeerAddrPort() : "";
 }
 

src/transport/TcpTransport.h

@@ -17,10 +17,9 @@
 #ifndef __TCPTRANSPORT_H__
 #define __TCPTRANSPORT_H__
 
-#include <boost/atomic.hpp>
-#include <boost/thread/condition_variable.hpp>
-#include <boost/thread/mutex.hpp>
-#include <boost/thread/thread.hpp>
+#include <atomic>
+#include <mutex>
+#include <condition_variable>
 
 #include "dataBlock.h"
 #include "EventLoop.h"
@@ -76,11 +75,11 @@ class TcpTransport : public std::enable_shared_from_this<TcpTransport> {
   uint64_t m_startTime;
 
   std::shared_ptr<BufferEvent> m_event;  // NOTE: use m_event in callback is unsafe.
-  boost::mutex m_eventLock;
-  boost::atomic<TcpConnectStatus> m_tcpConnectStatus;
+  std::mutex m_eventLock;
+  std::atomic<TcpConnectStatus> m_tcpConnectStatus;
 
-  boost::mutex m_connectEventLock;
-  boost::condition_variable_any m_connectEvent;
+  std::mutex m_connectEventLock;
+  std::condition_variable m_connectEvent;
 
   //<! read data callback
   TcpTransportReadCallback m_readCallback;