diff --git a/src/tendisplus/network/blocking_tcp_client.cpp b/src/tendisplus/network/blocking_tcp_client.cpp index ad772046..079abf1f 100644 --- a/src/tendisplus/network/blocking_tcp_client.cpp +++ b/src/tendisplus/network/blocking_tcp_client.cpp @@ -70,6 +70,46 @@ void BlockingTcpClient::closeSocket() { _socket.close(ignoredEc); } +void BlockingTcpClient::beginWait() { + std::lock_guard lk(_mutex); + _notified = false; +} + +void BlockingTcpClient::completeAsync(const asio::error_code& oec) { + std::lock_guard lk(_mutex); + _ec = oec; + _notified = true; + _cv.notify_one(); +} + +Status BlockingTcpClient::waitNotified( + std::chrono::steady_clock::duration timeout, const std::string& timeoutMsg) { + std::unique_lock lk(_mutex); + if (_notified || _cv.wait_for(lk, timeout, [this] { return _notified; })) { + if (_ec) { + auto msg = _ec.message(); + lk.unlock(); + closeSocket(); + return {ErrorCodes::ERR_NETWORK, msg}; + } + return {ErrorCodes::ERR_OK, ""}; + } + lk.unlock(); + closeSocket(); + return {ErrorCodes::ERR_TIMEOUT, timeoutMsg}; +} + +Status BlockingTcpClient::configureConnectedSocket() { + std::error_code ec; + _socket.non_blocking(true, ec); + if (ec) { + return {ErrorCodes::ERR_NETWORK, ec.message()}; + } + _socket.set_option(asio::ip::tcp::no_delay(true)); + _socket.set_option(asio::socket_base::keep_alive(true)); + return {ErrorCodes::ERR_OK, ""}; +} + Status BlockingTcpClient::connect(const std::string& host, uint16_t port, std::chrono::milliseconds timeout, @@ -80,6 +120,7 @@ Status BlockingTcpClient::connect(const std::string& host, return {ErrorCodes::ERR_NETWORK, "already inited sock"}; } _inited = true; + _timeout = timeout; } std::stringstream ss; ss << port; @@ -92,45 +133,23 @@ Status BlockingTcpClient::connect(const std::string& host, return {ErrorCodes::ERR_NETWORK, "resolve domain name fail"}; } - _notified = false; + beginWait(); auto self(shared_from_this()); asio::async_connect( _socket, endpoints, - [this, self](const std::error_code& oec, asio::ip::tcp::endpoint) { - std::unique_lock lk(_mutex); - _ec = oec; - _notified = true; - _cv.notify_one(); + [self](const std::error_code& oec, asio::ip::tcp::endpoint) { + self->completeAsync(oec); }); - _timeout = timeout; - if (isBlockingConnect) { - std::unique_lock lk(_mutex); - /* * - * NOTE(vinchen): It's possible that _cv.notify_one() was - called before _cv.wait_for(). So it check _notified - first. - */ - if (_notified || _cv.wait_for(lk, timeout, [this] { return _notified; })) { - if (_ec) { - closeSocket(); - return {ErrorCodes::ERR_NETWORK, _ec.message()}; - } - std::error_code ec; - _socket.non_blocking(true, ec); - INVARIANT(ec.value() == 0); - _socket.set_option(asio::ip::tcp::no_delay(true)); - _socket.set_option(asio::socket_base::keep_alive(true)); - return {ErrorCodes::ERR_OK, ""}; - } else { - closeSocket(); - return {ErrorCodes::ERR_TIMEOUT, "conn timeout"}; + auto s = waitNotified(timeout, "conn timeout"); + if (!s.ok()) { + return s; } - } else { - return {ErrorCodes::ERR_OK, ""}; + return configureConnectedSocket(); } + return {ErrorCodes::ERR_OK, ""}; } Status BlockingTcpClient::tryWaitConnect() { @@ -138,87 +157,61 @@ Status BlockingTcpClient::tryWaitConnect() { if (_cv.wait_for( lk, std::chrono::milliseconds(0), [this] { return _notified; })) { if (_ec) { + auto msg = _ec.message(); + lk.unlock(); closeSocket(); - return {ErrorCodes::ERR_NETWORK, _ec.message()}; - } - std::error_code ec; - _socket.non_blocking(true, ec); - INVARIANT_D(ec.value() == 0); - _socket.set_option(asio::ip::tcp::no_delay(true)); - _socket.set_option(asio::socket_base::keep_alive(true)); - return {ErrorCodes::ERR_OK, ""}; - } else { - if (msSinceEpoch() - _ctime > static_cast(_timeout.count())) { - return {ErrorCodes::ERR_TIMEOUT, "conn timeout"}; + return {ErrorCodes::ERR_NETWORK, msg}; } - - return {ErrorCodes::ERR_CONNECT_TRY, "conn try again"}; + lk.unlock(); + return configureConnectedSocket(); + } + if (msSinceEpoch() - _ctime > static_cast(_timeout.count())) { + return {ErrorCodes::ERR_TIMEOUT, "conn timeout"}; } + return {ErrorCodes::ERR_CONNECT_TRY, "conn try again"}; } Expected BlockingTcpClient::readLine( std::chrono::seconds timeout) { - _notified = false; + beginWait(); auto self(shared_from_this()); asio::async_read_until( - _socket, - _inputBuf, - "\n", - [this, self](const asio::error_code& oec, size_t size) { - std::unique_lock lk(_mutex); - _ec = oec; - _notified = true; - _cv.notify_one(); + _socket, _inputBuf, "\n", [self](const asio::error_code& oec, size_t) { + self->completeAsync(oec); }); - std::unique_lock lk(_mutex); - if (_cv.wait_for(lk, timeout, [this] { return _notified; })) { - if (_ec) { - closeSocket(); - return {ErrorCodes::ERR_NETWORK, _ec.message()}; - } + auto s = waitNotified(timeout, "readLine timeout"); + if (!s.ok()) { + return s; + } - std::string line; - std::istream is(&_inputBuf); - std::getline(is, line); - if (line[line.size() - 1] != '\r') { - closeSocket(); - return {ErrorCodes::ERR_NETWORK, "line not ended with \\r\\n"}; - } - line.erase(line.size() - 1); - return line; - } else { + std::string line; + std::istream is(&_inputBuf); + std::getline(is, line); + if (line[line.size() - 1] != '\r') { closeSocket(); - return {ErrorCodes::ERR_TIMEOUT, "readLine timeout"}; + return {ErrorCodes::ERR_NETWORK, "line not ended with \\r\\n"}; } + line.erase(line.size() - 1); + return line; } Expected BlockingTcpClient::realRead( size_t remain, std::chrono::seconds timeout) { if (remain > 0) { - _notified = false; + beginWait(); auto self(shared_from_this()); asio::async_read(_socket, _inputBuf, asio::transfer_exactly(remain), - [this, self](const asio::error_code& oec, size_t) { - std::unique_lock lk(_mutex); - _ec = oec; - _notified = true; - _cv.notify_one(); + [self](const asio::error_code& oec, size_t) { + self->completeAsync(oec); }); - // Block until the asynchronous operation has completed. - std::unique_lock lk(_mutex); - if (!_cv.wait_for(lk, timeout, [this] { return _notified; })) { - closeSocket(); - return {ErrorCodes::ERR_TIMEOUT, - "read timeout,remain:" + std::to_string(remain)}; - } else if (_ec) { - closeSocket(); - return {ErrorCodes::ERR_NETWORK, _ec.message()}; - } else { - // everything is ok, stepout this scope and process buffer + auto s = + waitNotified(timeout, "read timeout,remain:" + std::to_string(remain)); + if (!s.ok()) { + return s; } } return {ErrorCodes::ERR_OK, ""}; @@ -289,29 +282,14 @@ Status BlockingTcpClient::writeData(const std::string& data) { Status BlockingTcpClient::writeOneBatch(const char* data, uint32_t size, std::chrono::seconds timeout) { - _notified = false; + beginWait(); auto self(shared_from_this()); - asio::async_write(_socket, - asio::buffer(data, size), - [this, self](const asio::error_code& oec, size_t) { - std::unique_lock lk(_mutex); - _ec = oec; - _notified = true; - _cv.notify_one(); - }); + asio::async_write( + _socket, + asio::buffer(data, size), + [self](const asio::error_code& oec, size_t) { self->completeAsync(oec); }); - std::unique_lock lk(_mutex); - if (_cv.wait_for(lk, timeout, [this] { return _notified; })) { - if (_ec) { - closeSocket(); - return {ErrorCodes::ERR_NETWORK, _ec.message()}; - } else { - return {ErrorCodes::ERR_OK, ""}; - } - } else { - closeSocket(); - return {ErrorCodes::ERR_TIMEOUT, "writeData timeout"}; - } + return waitNotified(timeout, "writeData timeout"); } Status BlockingTcpClient::writeLine(const std::string& line) { diff --git a/src/tendisplus/network/blocking_tcp_client.h b/src/tendisplus/network/blocking_tcp_client.h index c0a1ab0d..1220913d 100644 --- a/src/tendisplus/network/blocking_tcp_client.h +++ b/src/tendisplus/network/blocking_tcp_client.h @@ -137,6 +137,11 @@ class BlockingTcpClient private: Expected realRead(size_t bufSize, std::chrono::seconds timeout); + void beginWait(); + void completeAsync(const asio::error_code& oec); + Status waitNotified(std::chrono::steady_clock::duration timeout, + const std::string& timeoutMsg); + Status configureConnectedSocket(); std::mutex _mutex; std::condition_variable _cv; bool _inited; diff --git a/src/tendisplus/server/server_entry.cpp b/src/tendisplus/server/server_entry.cpp index f3b02ca6..27caf4d3 100644 --- a/src/tendisplus/server/server_entry.cpp +++ b/src/tendisplus/server/server_entry.cpp @@ -218,14 +218,14 @@ void SlowlogStat::closeSlowlogFile() { } void SlowlogStat::slowlogFlush() { - if (!_waitingFlush) { + std::lock_guard lk(_fileMutex); + if (!_waitingFlush.load(std::memory_order_relaxed)) { return; } - std::lock_guard lk(_fileMutex); if (_slowLog.is_open()) { _slowLog.flush(); } - _waitingFlush = false; + _waitingFlush.store(false, std::memory_order_relaxed); } void SlowlogStat::slowlogDataPushEntryIfNeeded( @@ -241,7 +241,8 @@ void SlowlogStat::slowlogDataPushEntryIfNeeded( if (cfgs->slowlogFileEnabled) { if (cfgs->slowlogFileSplitEnabled && - (_filesize >> 20) >= cfgs->slowlogFileMaxSizeMb) { + (_filesize.load(std::memory_order_relaxed) >> 20) >= + cfgs->slowlogFileMaxSizeMb) { std::lock_guard lk(_fileMutex); if (_slowLog.is_open()) _slowLog.flush(); @@ -885,6 +886,11 @@ Status ServerEntry::startup(const std::shared_ptr& cfg) { return s; } + // Must init before _network->run() and before the cron thread starts: + // both record latency / flush slowlog as soon as they are up. + _latencyMonitorSet.init(); + _slowlogStat.initSlowlogFile(cfg); + // listener should be the lastone to run. s = _network->run(); if (!s.ok()) { @@ -909,11 +915,6 @@ Status ServerEntry::startup(const std::shared_ptr& cfg) { bgCompactCron(); }); - _latencyMonitorSet.init(); - - // init slowlog - _slowlogStat.initSlowlogFile(cfg); - _lastJeprofDumpMemoryGB = 0; jemallocBgThreadConf(); diff --git a/src/tendisplus/server/server_entry.h b/src/tendisplus/server/server_entry.h index cc9498a6..e3fea7e7 100644 --- a/src/tendisplus/server/server_entry.h +++ b/src/tendisplus/server/server_entry.h @@ -5,6 +5,7 @@ #ifndef SRC_TENDISPLUS_SERVER_SERVER_ENTRY_H_ #define SRC_TENDISPLUS_SERVER_SERVER_ENTRY_H_ +#include #include #include #include @@ -149,8 +150,8 @@ class SlowlogStat { std::atomic _slowlogId; mutable std::mutex _dataMutex; mutable std::mutex _fileMutex; - bool _waitingFlush; - uint64_t _filesize; + std::atomic _waitingFlush; + std::atomic _filesize; }; #define THREAD_SLEEP(n_secs) \