You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@kvrocks.apache.org by "AntiTopQuark (via GitHub)" <gi...@apache.org> on 2024/08/20 05:49:40 UTC
[PR] [WIP]support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
AntiTopQuark opened a new pull request, #2498:
URL: https://github.com/apache/kvrocks/pull/2498
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "torwig (via GitHub)" <gi...@apache.org>.
torwig commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1731529953
##########
src/server/redis_connection.cc:
##########
@@ -97,6 +103,9 @@ void Connection::OnRead(struct bufferevent *bev) {
if (IsFlagEnabled(kCloseAsync)) {
Close();
}
+ if (IsReachOBufLimit()) {
Review Comment:
We can expand the letter `O` to `Output` and have something like `IsOutputBufferLimitReached`.
##########
src/server/redis_connection.cc:
##########
@@ -559,4 +652,86 @@ void Connection::ResetMultiExec() {
DisableFlag(Connection::kMultiExec);
}
+bool Connection::CheckClientReachOBufLimits(size_t reply_bytes) {
+ assert(reply_bytes < SIZE_MAX - (1024 * 64));
+ if (reply_bytes == 0 && GetClientType() != kTypeSlave) {
+ return false;
+ }
+ int client_type = 0;
+ if (GetClientType() == kTypeSlave) {
+ client_type = 1;
+ } else if (GetClientType() == kTypePubsub) {
+ client_type = 2;
+ } else {
+ client_type = 0;
+ }
+
+ bool hard = false;
+ bool soft = false;
+ auto hard_limit_bytes = GetServer()->GetConfig()->GetClientOutputBufferLimits()[client_type].hard_limit_bytes;
+ auto soft_limit_bytes = GetServer()->GetConfig()->GetClientOutputBufferLimits()[client_type].soft_limit_bytes;
+ if (hard_limit_bytes && reply_bytes >= hard_limit_bytes) {
+ hard = true;
+ }
+ if (soft_limit_bytes && reply_bytes >= soft_limit_bytes) {
+ soft = true;
+ }
+
+ if (soft) {
+ if (GetObufSoftLimitReachedTime() == 0) {
+ SetObufSoftLimitReachedTime(util::GetTimeStamp());
+ soft = false;
+ } else {
+ uint64_t elapsed = util::GetTimeStamp() - GetObufSoftLimitReachedTime();
+ if (elapsed <= GetServer()->GetConfig()->GetClientOutputBufferLimits()[0].soft_limit_seconds) {
+ soft = false;
+ }
+ }
+ } else {
+ SetObufSoftLimitReachedTime(0);
+ }
+ return hard || soft;
+}
+
+std::string Connection::CheckClientReachOBufLimits(const std::string &msg) {
+ auto memSize = msg.size() + GetOutputBuffer().capacity() + evbuffer_get_length(Output());
Review Comment:
`memSize` -> `mem_size`
##########
src/server/redis_connection.cc:
##########
@@ -559,4 +652,86 @@ void Connection::ResetMultiExec() {
DisableFlag(Connection::kMultiExec);
}
+bool Connection::CheckClientReachOBufLimits(size_t reply_bytes) {
Review Comment:
The same here: `O` -> `Output`.
##########
src/server/redis_connection.cc:
##########
@@ -559,4 +652,86 @@ void Connection::ResetMultiExec() {
DisableFlag(Connection::kMultiExec);
}
+bool Connection::CheckClientReachOBufLimits(size_t reply_bytes) {
+ assert(reply_bytes < SIZE_MAX - (1024 * 64));
+ if (reply_bytes == 0 && GetClientType() != kTypeSlave) {
+ return false;
+ }
+ int client_type = 0;
Review Comment:
Maybe we can extract this logic to a separate function, introducing descriptive constants for 0, 1, 2.
##########
src/server/server.cc:
##########
@@ -839,6 +839,11 @@ void Server::cron() {
cleanupExitedWorkerThreads(false);
}
+ // check if we need to evict connections every 10s
+ if (counter != 0 && counter % 100 == 0 && config_->max_memory_clients > 0) {
Review Comment:
Can we make the condition `config_->max_memory_clients > 0` the first one to benefit from fast evaluation if this option is disabled in the config?
##########
src/server/redis_connection.h:
##########
@@ -180,6 +172,15 @@ class Connection : public EvbufCallbackBase<Connection> {
std::set<std::string> watched_keys;
std::atomic<bool> watched_keys_modified = false;
+ bool CheckClientReachOBufLimits(size_t reply_bytes);
Review Comment:
In all functions of this "family" we can expand a single `O` letter to `Output`
##########
src/config/config_type.h:
##########
@@ -338,3 +338,119 @@ class IntWithUnitField : public ConfigField {
T max_;
IntUnit current_unit_ = IntUnit::None;
};
+
+struct ClientOutputBufferLimitConfig {
+ uint64_t hard_limit_bytes = 0;
+ uint64_t soft_limit_bytes = 0;
+ uint64_t soft_limit_seconds = 0;
+};
+
+constexpr int CLIENT_TYPE_OBUF_COUNT = 3;
+
+class ClientOutputBufferLimitField : public ConfigField {
+ public:
+ ClientOutputBufferLimitField(std::vector<ClientOutputBufferLimitConfig> *limits, std::string default_val)
+ : receiver_(limits), default_val_(std::move(default_val)) {
+ if (receiver_->empty()) {
+ receiver_->resize(CLIENT_TYPE_OBUF_COUNT);
+ }
+ CHECK(Set(default_val_));
+ config_type = ConfigType::SingleConfig;
+ }
+ std::string Default() const override { return default_val_; }
+
+ std::string ToString() const override {
+ std::ostringstream ss;
+ for (size_t i = 0; i < receiver_->size(); ++i) {
+ if (i != 0) ss << " ";
+ ss << clientTypeToString(i) << " " << receiver_->at(i).hard_limit_bytes << " "
+ << receiver_->at(i).soft_limit_bytes << " " << receiver_->at(i).soft_limit_seconds;
+ }
+ return ss.str();
+ }
+
+ Status Set(const std::string &v) override {
+ std::vector<std::string> tokens = splitString(v, ' ');
+ if (tokens.size() % 4 != 0) {
+ return {Status::NotOK, "Wrong number of arguments in buffer limit configuration."};
+ }
+
+ for (size_t i = 0; i < tokens.size(); i += 4) {
+ int class_type = getClientTypeByName(tokens[i]);
+ if (class_type == -1) {
+ return {Status::NotOK, "Invalid client class specified in buffer limit configuration."};
+ }
+
+ uint64_t hard_limit = 0, soft_limit = 0;
+ auto s = capacityToInt(tokens[i + 1], hard_limit);
+ if (!s.IsOK()) return s;
+ s = capacityToInt(tokens[i + 2], soft_limit);
+ if (!s.IsOK()) return s;
+ auto [soft_seconds, rest] = GET_OR_RET(TryParseInt(tokens[i + 3].c_str(), 10));
+ if (*rest != 0) {
+ return {Status::NotOK, "Error in soft_seconds setting in buffer limit configuration."};
+ }
+
+ ClientOutputBufferLimitConfig config = {hard_limit, soft_limit, static_cast<uint64_t>(soft_seconds)};
+ (*receiver_)[class_type] = config;
+ }
+ return Status::OK();
+ }
+
+ private:
+ static std::string clientTypeToString(size_t type) {
+ switch (type) {
+ case 0:
+ return "normal";
+ case 1:
+ return "replica";
+ case 2:
+ return "pubsub";
+ default:
+ return "unknown";
+ }
+ }
+
+ static std::vector<std::string> splitString(const std::string &str, char delim) {
+ std::vector<std::string> tokens;
+ std::string token;
+ std::istringstream token_stream(str);
+ while (std::getline(token_stream, token, delim)) {
+ tokens.push_back(token);
+ }
+ return tokens;
+ }
+
+ static Status capacityToInt(const std::string &str, uint64_t &capacity) {
Review Comment:
We have something similar here src/common/parse_util.cc -> ParseSizeAndUnit(). Can it serve your purpose here?
##########
src/server/redis_connection.h:
##########
@@ -180,6 +172,15 @@ class Connection : public EvbufCallbackBase<Connection> {
std::set<std::string> watched_keys;
std::atomic<bool> watched_keys_modified = false;
+ bool CheckClientReachOBufLimits(size_t reply_bytes);
+ std::string CheckClientReachOBufLimits(const std::string &msg);
+ void SetObufSoftLimitReachedTime(int64_t time) { obuf_soft_limit_reached_time_ = time; }
+ int64_t GetObufSoftLimitReachedTime() const { return obuf_soft_limit_reached_time_; }
+ inline std::string &GetOutputBuffer() { return output_buffer_; }
+ size_t GetConnectionMemoryUsed() const;
+ void SetReachOBufLimit(bool reach) { is_reach_obuf_limit_ = reach; }
+ bool IsReachOBufLimit() const { return is_reach_obuf_limit_; }
Review Comment:
`IsReachOBufLimit` -> `IsOutputBufLimitReached`.
##########
src/server/redis_connection.cc:
##########
@@ -559,4 +652,86 @@ void Connection::ResetMultiExec() {
DisableFlag(Connection::kMultiExec);
}
+bool Connection::CheckClientReachOBufLimits(size_t reply_bytes) {
+ assert(reply_bytes < SIZE_MAX - (1024 * 64));
+ if (reply_bytes == 0 && GetClientType() != kTypeSlave) {
+ return false;
+ }
+ int client_type = 0;
+ if (GetClientType() == kTypeSlave) {
+ client_type = 1;
+ } else if (GetClientType() == kTypePubsub) {
+ client_type = 2;
+ } else {
+ client_type = 0;
+ }
+
+ bool hard = false;
+ bool soft = false;
+ auto hard_limit_bytes = GetServer()->GetConfig()->GetClientOutputBufferLimits()[client_type].hard_limit_bytes;
+ auto soft_limit_bytes = GetServer()->GetConfig()->GetClientOutputBufferLimits()[client_type].soft_limit_bytes;
+ if (hard_limit_bytes && reply_bytes >= hard_limit_bytes) {
+ hard = true;
+ }
+ if (soft_limit_bytes && reply_bytes >= soft_limit_bytes) {
+ soft = true;
+ }
+
+ if (soft) {
+ if (GetObufSoftLimitReachedTime() == 0) {
+ SetObufSoftLimitReachedTime(util::GetTimeStamp());
+ soft = false;
+ } else {
+ uint64_t elapsed = util::GetTimeStamp() - GetObufSoftLimitReachedTime();
+ if (elapsed <= GetServer()->GetConfig()->GetClientOutputBufferLimits()[0].soft_limit_seconds) {
+ soft = false;
+ }
+ }
+ } else {
+ SetObufSoftLimitReachedTime(0);
+ }
+ return hard || soft;
+}
+
+std::string Connection::CheckClientReachOBufLimits(const std::string &msg) {
+ auto memSize = msg.size() + GetOutputBuffer().capacity() + evbuffer_get_length(Output());
+ if (CheckClientReachOBufLimits(memSize)) {
+ SetReachOBufLimit(true);
+ return "";
+ }
+ return msg;
+}
+
+size_t Connection::GetConnectionMemoryUsed() const {
+ size_t total_memory = sizeof(*this); // 包含所有成员变量的静态内存大小
Review Comment:
Could you please translate the comment into English?
##########
src/server/worker.cc:
##########
@@ -102,8 +102,13 @@ Worker::~Worker() {
void Worker::TimerCB(int, int16_t events) {
auto config = srv->GetConfig();
- if (config->timeout == 0) return;
- KickoutIdleClients(config->timeout);
+ if (config->timeout != 0) {
+ KickoutIdleClients(config->timeout);
+ }
+
+ if (config->IsClientOutputBufferLimitsEnabled()) {
+ KickoutReachOBufLimitsClients();
Review Comment:
The same: `O` -> `Output`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "torwig (via GitHub)" <gi...@apache.org>.
torwig commented on PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#issuecomment-2310703089
@AntiTopQuark Running `./x.py format` should make the linter happy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730777654
##########
src/server/worker.cc:
##########
@@ -446,6 +452,13 @@ Status Worker::Reply(int fd, const std::string &reply) {
auto iter = conns_.find(fd);
if (iter != conns_.end()) {
iter->second->SetLastInteraction();
+ if (iter->second->CheckClientReachOBufLimits(reply.size() + evbuffer_get_length(iter->second->Output()))) {
Review Comment:
There are not many such dangerous commands. I don't think the cost per check will be too high (or we can check once every N times).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "AntiTopQuark (via GitHub)" <gi...@apache.org>.
AntiTopQuark commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730751617
##########
src/server/worker.cc:
##########
@@ -446,6 +452,13 @@ Status Worker::Reply(int fd, const std::string &reply) {
auto iter = conns_.find(fd);
if (iter != conns_.end()) {
iter->second->SetLastInteraction();
+ if (iter->second->CheckClientReachOBufLimits(reply.size() + evbuffer_get_length(iter->second->Output()))) {
Review Comment:
There are many commands that return results by using the `operator+= ` operation on the output string. Are you suggesting that every time `operator+=` is used, we should monitor and check the memory usage? Wouldn't this cause significant performance issues? Is it acceptable to make extensive changes to the code related to many commands?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1729173952
##########
src/server/worker.cc:
##########
@@ -446,6 +452,13 @@ Status Worker::Reply(int fd, const std::string &reply) {
auto iter = conns_.find(fd);
if (iter != conns_.end()) {
iter->second->SetLastInteraction();
+ if (iter->second->CheckClientReachOBufLimits(reply.size())) {
+ srv->stats.IncrReachOutbufLimitDisconnections();
+ // unlock before calling FreeConnectionByID, otherwise it will cause nested deadlock
+ lock.unlock();
+ FreeConnectionByID(iter->first, iter->second->GetID());
+ return {Status::NotOK, "connection reach output buffer limits"};
+ }
redis::Reply(iter->second->Output(), reply);
Review Comment:
The reply will be added to libevent's evbuffer. Will the data accumulate in the evbuffer?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "torwig (via GitHub)" <gi...@apache.org>.
torwig commented on PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#issuecomment-2310701261
@AntiTopQuark Ideally extract each feature in a separate PR as they are independent as far as I understand.
in a separate PR. It's not **strongly** required but it would make your PRs more granular and [maybe] smaller in size :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730621616
##########
src/server/worker.cc:
##########
@@ -446,6 +452,13 @@ Status Worker::Reply(int fd, const std::string &reply) {
auto iter = conns_.find(fd);
if (iter != conns_.end()) {
iter->second->SetLastInteraction();
+ if (iter->second->CheckClientReachOBufLimits(reply.size() + evbuffer_get_length(iter->second->Output()))) {
Review Comment:
This reply may become large during generation, such as the `KEYS` command.
##########
src/server/server.cc:
##########
@@ -2115,3 +2122,40 @@ AuthResult Server::AuthenticateUser(const std::string &user_password, std::strin
*ns = kDefaultNamespace;
return AuthResult::IS_ADMIN;
}
+
+void Server::evictionClients() {
+ size_t mem = 0;
+ for (const auto &t : worker_threads_) {
+ mem += t->GetWorker()->GetConnectionsMemoryUsed();
+ }
+ if (mem < config_->max_memory_clients) {
+ return;
+ }
+
+ std::vector<redis::Connection *> conns;
+ for (const auto &t : worker_threads_) {
+ auto worker_conns = t->GetWorker()->GetConnections();
+ for (const auto &iter : worker_conns) {
+ conns.push_back(iter.second);
Review Comment:
The lock will not work here, and these connections may still be released when you access them below.
One of my own practices is to `maxmemory-clients/worker`, estimate the `maxmemory` of each worker, and then record (update in real time) the memory of each connection in each worker, which can be more timely and without locking.
##########
src/config/config_type.h:
##########
@@ -338,3 +338,119 @@ class IntWithUnitField : public ConfigField {
T max_;
IntUnit current_unit_ = IntUnit::None;
};
+
+struct ClientOutputBufferLimitConfig {
+ uint64_t hard_limit_bytes = 0;
+ uint64_t soft_limit_bytes = 0;
+ uint64_t soft_limit_seconds = 0;
+};
+
+constexpr int CLIENT_TYPE_OBUF_COUNT = 3;
+
+class ClientOutputBufferLimitFiled : public ConfigField {
Review Comment:
field
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#issuecomment-2312003179
> Maybe we should separate it into following PRs to make it easier to merge and review
I agree with it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#issuecomment-2311996885
To be honest, I'm worried about the performance and maintenance burden of this implementation. Maybe we should separate it into following PRs to make it easier to merge and review:
- Allow to control the maximum size of the write batch
- Allow to control the maximum size of connection input/output buffer(just check the size of libevent buffer)
- Allow to control the maximum size of all clients(need a clear implementation proposal first)
What do you think? @torwig @caipengbo @mapleFU @PragmaTwice
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730762127
##########
src/server/server.cc:
##########
@@ -2115,3 +2122,40 @@ AuthResult Server::AuthenticateUser(const std::string &user_password, std::strin
*ns = kDefaultNamespace;
return AuthResult::IS_ADMIN;
}
+
+void Server::evictionClients() {
+ size_t mem = 0;
+ for (const auto &t : worker_threads_) {
+ mem += t->GetWorker()->GetConnectionsMemoryUsed();
+ }
+ if (mem < config_->max_memory_clients) {
+ return;
+ }
+
+ std::vector<redis::Connection *> conns;
+ for (const auto &t : worker_threads_) {
+ auto worker_conns = t->GetWorker()->GetConnections();
+ for (const auto &iter : worker_conns) {
+ conns.push_back(iter.second);
Review Comment:
Yes, we usually assume that the worker is balanced. In fact, I do this, it is just a trade-off
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730364698
##########
src/server/worker.cc:
##########
@@ -446,6 +452,13 @@ Status Worker::Reply(int fd, const std::string &reply) {
auto iter = conns_.find(fd);
if (iter != conns_.end()) {
iter->second->SetLastInteraction();
+ if (iter->second->CheckClientReachOBufLimits(reply.size())) {
+ srv->stats.IncrReachOutbufLimitDisconnections();
+ // unlock before calling FreeConnectionByID, otherwise it will cause nested deadlock
+ lock.unlock();
+ FreeConnectionByID(iter->first, iter->second->GetID());
+ return {Status::NotOK, "connection reach output buffer limits"};
+ }
redis::Reply(iter->second->Output(), reply);
Review Comment:
Yes, I think the evbuffer is the client's final output buffer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "AntiTopQuark (via GitHub)" <gi...@apache.org>.
AntiTopQuark closed pull request #2498: feat: support limit WriteBatch size,client outbuf limit and maxmemory clients
URL: https://github.com/apache/kvrocks/pull/2498
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "AntiTopQuark (via GitHub)" <gi...@apache.org>.
AntiTopQuark commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730391892
##########
src/server/worker.cc:
##########
@@ -446,6 +452,13 @@ Status Worker::Reply(int fd, const std::string &reply) {
auto iter = conns_.find(fd);
if (iter != conns_.end()) {
iter->second->SetLastInteraction();
+ if (iter->second->CheckClientReachOBufLimits(reply.size())) {
+ srv->stats.IncrReachOutbufLimitDisconnections();
+ // unlock before calling FreeConnectionByID, otherwise it will cause nested deadlock
+ lock.unlock();
+ FreeConnectionByID(iter->first, iter->second->GetID());
+ return {Status::NotOK, "connection reach output buffer limits"};
+ }
redis::Reply(iter->second->Output(), reply);
Review Comment:
done, thank you
##########
src/server/server.cc:
##########
@@ -2115,3 +2122,40 @@ AuthResult Server::AuthenticateUser(const std::string &user_password, std::strin
*ns = kDefaultNamespace;
return AuthResult::IS_ADMIN;
}
+
+void Server::evictionClients() {
+ size_t mem = 0;
+ for (const auto &t : worker_threads_) {
+ mem += t->GetWorker()->GetConnectionsMemoryUsed();
+ }
+ if (mem < config_->max_memory_clients) {
+ return;
+ }
+
+ std::vector<redis::Connection *> conns;
+ for (const auto &t : worker_threads_) {
+ auto worker_conns = t->GetWorker()->GetConnections();
+ for (const auto &iter : worker_conns) {
+ conns.push_back(iter.second);
Review Comment:
done, thank you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "AntiTopQuark (via GitHub)" <gi...@apache.org>.
AntiTopQuark commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730203259
##########
src/server/worker.cc:
##########
@@ -446,6 +452,13 @@ Status Worker::Reply(int fd, const std::string &reply) {
auto iter = conns_.find(fd);
if (iter != conns_.end()) {
iter->second->SetLastInteraction();
+ if (iter->second->CheckClientReachOBufLimits(reply.size())) {
+ srv->stats.IncrReachOutbufLimitDisconnections();
+ // unlock before calling FreeConnectionByID, otherwise it will cause nested deadlock
+ lock.unlock();
+ FreeConnectionByID(iter->first, iter->second->GetID());
+ return {Status::NotOK, "connection reach output buffer limits"};
+ }
redis::Reply(iter->second->Output(), reply);
Review Comment:
Do you mean that publishing a large number of small messages caused the module consuming the `evbuffer` to not process them immediately, leading to memory bloat?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "AntiTopQuark (via GitHub)" <gi...@apache.org>.
AntiTopQuark commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1768895709
##########
src/server/worker.cc:
##########
@@ -446,6 +452,13 @@ Status Worker::Reply(int fd, const std::string &reply) {
auto iter = conns_.find(fd);
if (iter != conns_.end()) {
iter->second->SetLastInteraction();
+ if (iter->second->CheckClientReachOBufLimits(reply.size() + evbuffer_get_length(iter->second->Output()))) {
Review Comment:
Hi, I noticed that the operator += and append are used quite frequently. I need to convert all `redis::xxx` to `GET_OR_RET(conn->xxx())` to check whether the output buffer size meets the expectations each time. Even if I change it to check the client output buffer every 10 times, I suspect this might have a significant impact on performance.
My idea is to create multiple member functions for conn to replace the original redis::xxxx() functions, so that the memory size check can be performed each time an output is made. If the check fails, an error code will be thrown.
```c++
StatusOr<std::string> Connection::SimpleString(const std::string &msg) {
auto res = redis::SimpleString(msg);
if (CheckClientReachOutputBufferLimits(res)) {
return {Status::ReachClientOutputBufferLimit, "client reached output buffer limits"};
}
return res;
}
- *output += redis::SimpleString("Capacity");
- *output += redis::SimpleString("Size");
+ *output += GET_OR_RET((conn->SimpleString("Capacity"));
+ *output += GET_OR_RET((conn->SimpleString("Size"));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1729180650
##########
src/server/server.cc:
##########
@@ -2115,3 +2122,40 @@ AuthResult Server::AuthenticateUser(const std::string &user_password, std::strin
*ns = kDefaultNamespace;
return AuthResult::IS_ADMIN;
}
+
+void Server::evictionClients() {
+ size_t mem = 0;
+ for (const auto &t : worker_threads_) {
+ mem += t->GetWorker()->GetConnectionsMemoryUsed();
+ }
+ if (mem < config_->max_memory_clients) {
+ return;
+ }
+
+ std::vector<redis::Connection *> conns;
+ for (const auto &t : worker_threads_) {
+ auto worker_conns = t->GetWorker()->GetConnections();
+ for (const auto &iter : worker_conns) {
+ conns.push_back(iter.second);
Review Comment:
Are these thread-safe? Is it possible that the connections has been released?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "AntiTopQuark (via GitHub)" <gi...@apache.org>.
AntiTopQuark commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730203435
##########
src/server/server.cc:
##########
@@ -2115,3 +2122,40 @@ AuthResult Server::AuthenticateUser(const std::string &user_password, std::strin
*ns = kDefaultNamespace;
return AuthResult::IS_ADMIN;
}
+
+void Server::evictionClients() {
+ size_t mem = 0;
+ for (const auto &t : worker_threads_) {
+ mem += t->GetWorker()->GetConnectionsMemoryUsed();
+ }
+ if (mem < config_->max_memory_clients) {
+ return;
+ }
+
+ std::vector<redis::Connection *> conns;
+ for (const auto &t : worker_threads_) {
+ auto worker_conns = t->GetWorker()->GetConnections();
+ for (const auto &iter : worker_conns) {
+ conns.push_back(iter.second);
Review Comment:
I referred to the implementation of the `void Server::decreaseWorkerThreads(size_t delta)` function, which is also called without locking.
https://github.com/AntiTopQuark/kvrocks/blob/43f14db05f73d02a6fefc95347fd5c79781f85a5/src/server/server.cc#L1895
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730366878
##########
src/server/server.cc:
##########
@@ -2115,3 +2122,40 @@ AuthResult Server::AuthenticateUser(const std::string &user_password, std::strin
*ns = kDefaultNamespace;
return AuthResult::IS_ADMIN;
}
+
+void Server::evictionClients() {
+ size_t mem = 0;
+ for (const auto &t : worker_threads_) {
+ mem += t->GetWorker()->GetConnectionsMemoryUsed();
+ }
+ if (mem < config_->max_memory_clients) {
+ return;
+ }
+
+ std::vector<redis::Connection *> conns;
+ for (const auto &t : worker_threads_) {
+ auto worker_conns = t->GetWorker()->GetConnections();
+ for (const auto &iter : worker_conns) {
+ conns.push_back(iter.second);
Review Comment:
@git-hulk I think there's something wrong here. Am I missing something?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730777654
##########
src/server/worker.cc:
##########
@@ -446,6 +452,13 @@ Status Worker::Reply(int fd, const std::string &reply) {
auto iter = conns_.find(fd);
if (iter != conns_.end()) {
iter->second->SetLastInteraction();
+ if (iter->second->CheckClientReachOBufLimits(reply.size() + evbuffer_get_length(iter->second->Output()))) {
Review Comment:
There are not many such dangerous commands. I don't think the overhead per check will be too high (or we can check once every N times).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "AntiTopQuark (via GitHub)" <gi...@apache.org>.
AntiTopQuark commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730750660
##########
src/server/server.cc:
##########
@@ -2115,3 +2122,40 @@ AuthResult Server::AuthenticateUser(const std::string &user_password, std::strin
*ns = kDefaultNamespace;
return AuthResult::IS_ADMIN;
}
+
+void Server::evictionClients() {
+ size_t mem = 0;
+ for (const auto &t : worker_threads_) {
+ mem += t->GetWorker()->GetConnectionsMemoryUsed();
+ }
+ if (mem < config_->max_memory_clients) {
+ return;
+ }
+
+ std::vector<redis::Connection *> conns;
+ for (const auto &t : worker_threads_) {
+ auto worker_conns = t->GetWorker()->GetConnections();
+ for (const auto &iter : worker_conns) {
+ conns.push_back(iter.second);
Review Comment:
I referred to the implementation of the KickoutIdleClients() function. In FreeConnectionByID(), if the connection (conn) has already been released, no further action will be taken.
Additionally, may I ask if I can assume that, in most cases, the memory usage across different workers in the system is almost uniform, and that there is rarely a situation where memory usage is uneven?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "AntiTopQuark (via GitHub)" <gi...@apache.org>.
AntiTopQuark commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730750147
##########
src/config/config_type.h:
##########
@@ -338,3 +338,119 @@ class IntWithUnitField : public ConfigField {
T max_;
IntUnit current_unit_ = IntUnit::None;
};
+
+struct ClientOutputBufferLimitConfig {
+ uint64_t hard_limit_bytes = 0;
+ uint64_t soft_limit_bytes = 0;
+ uint64_t soft_limit_seconds = 0;
+};
+
+constexpr int CLIENT_TYPE_OBUF_COUNT = 3;
+
+class ClientOutputBufferLimitFiled : public ConfigField {
Review Comment:
done, thank you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730366878
##########
src/server/server.cc:
##########
@@ -2115,3 +2122,40 @@ AuthResult Server::AuthenticateUser(const std::string &user_password, std::strin
*ns = kDefaultNamespace;
return AuthResult::IS_ADMIN;
}
+
+void Server::evictionClients() {
+ size_t mem = 0;
+ for (const auto &t : worker_threads_) {
+ mem += t->GetWorker()->GetConnectionsMemoryUsed();
+ }
+ if (mem < config_->max_memory_clients) {
+ return;
+ }
+
+ std::vector<redis::Connection *> conns;
+ for (const auto &t : worker_threads_) {
+ auto worker_conns = t->GetWorker()->GetConnections();
+ for (const auto &iter : worker_conns) {
+ conns.push_back(iter.second);
Review Comment:
@git-hulk I think there's something wrong here. Am I missing something?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support limit WriteBatch size,client outbuf limit and maxmemory clients [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2498:
URL: https://github.com/apache/kvrocks/pull/2498#discussion_r1730368555
##########
src/server/server.cc:
##########
@@ -2115,3 +2122,40 @@ AuthResult Server::AuthenticateUser(const std::string &user_password, std::strin
*ns = kDefaultNamespace;
return AuthResult::IS_ADMIN;
}
+
+void Server::evictionClients() {
+ size_t mem = 0;
+ for (const auto &t : worker_threads_) {
+ mem += t->GetWorker()->GetConnectionsMemoryUsed();
+ }
+ if (mem < config_->max_memory_clients) {
+ return;
+ }
+
+ std::vector<redis::Connection *> conns;
+ for (const auto &t : worker_threads_) {
+ auto worker_conns = t->GetWorker()->GetConnections();
+ for (const auto &iter : worker_conns) {
+ conns.push_back(iter.second);
Review Comment:
In `MigrateConnection`, the connection cannot be released.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org