jihuayu commented on code in PR #3560:
URL: https://github.com/apache/kvrocks/pull/3560#discussion_r3583906452
##########
src/config/config.cc:
##########
@@ -444,6 +446,47 @@ void Config::initFieldValidator() {
// The callback function would be invoked after the field was set,
// it may change related fields or re-format the field. for example,
// when the 'dir' was set, the db-dir or backup-dir should be reset as well.
+// Parses a client-output-buffer-limit spec like "normal 0 0 0 pubsub 32m 8m
60"
+// and applies it to the limits of the specified client kinds. The full spec is
+// parsed before applying anything, so a malformed quadruple cannot leave the
+// limits partially updated.
+Status Config::parseClientOutputBufferLimits(const std::string &v) {
+ std::vector<std::string> args = util::Split(v, " \t");
+ if (args.empty() || args.size() % 4 != 0) {
+ return {Status::NotOK, "should be in the format of <class> <hard limit>
<soft limit> <soft seconds> ..."};
+ }
+
+ struct ParsedLimit {
+ ClientKind kind;
+ uint64_t hard_limit_bytes, soft_limit_bytes, soft_limit_seconds;
+ };
+ std::vector<ParsedLimit> parsed;
+ for (size_t i = 0; i < args.size(); i += 4) {
+ ClientKind kind = ClientKind::kNormal;
+ if (util::EqualICase(args[i], "normal")) {
+ kind = ClientKind::kNormal;
+ } else if (util::EqualICase(args[i], "slave") || util::EqualICase(args[i],
"replica")) {
+ kind = ClientKind::kSlave;
+ } else if (util::EqualICase(args[i], "pubsub")) {
+ kind = ClientKind::kPubsub;
+ } else {
+ return {Status::NotOK, fmt::format("unknown client kind '{}'", args[i])};
+ }
+ auto hard = GET_OR_RET(ParseSizeAndUnit(args[i + 1]).Prefixed("invalid
hard limit"));
Review Comment:
It looks a little different from Redis:
https://github.com/redis/redis/blob/e5e1eaa97ad3020305b84442197011abc8ce40bc/redis.conf#L8-L18
##########
src/server/redis_connection.cc:
##########
@@ -93,11 +93,60 @@ std::string Connection::ToString() {
evbuffer_get_length(Input()), evbuffer_get_length(Output()), last_cmd_,
set_info_.lib_name, set_info_.lib_ver);
}
-void Connection::Close() {
+void Connection::Close(bool is_async) {
+ if (is_async) {
+ // Only the first caller should schedule the close since concurrent reply
+ // paths (e.g. publishers on other workers) may race here.
+ if (flags_.fetch_or(kCloseAsync) & kCloseAsync) return;
+
+ // The write callback of a stuck client may never be invoked since its
output
+ // buffer cannot drain, so trigger the callback manually instead of waiting
+ // for it. Ignoring watermarks is required because the write callback is
only
+ // triggered when the output buffer size is not larger than the low
watermark.
+ // The callback is deferred(BEV_OPT_DEFER_CALLBACKS) and runs in the owner
+ // worker's event loop, where it's safe to free the connection.
+ bufferevent_trigger(bev_, EV_WRITE, BEV_TRIG_IGNORE_WATERMARKS |
BEV_TRIG_DEFER_CALLBACKS);
+ return;
+ }
+
if (close_cb) close_cb(GetFD());
owner_->FreeConnection(this);
}
+bool Connection::IsExceedOutputBufferLimit() {
+ // Connections that are already scheduled to close don't need to be checked
+ // again. The replication stream is written to the socket directly instead of
+ // going through the connection output buffer, so the slave kind is not
+ // applicable here: slow replicas are handled by max-replication-lag and
+ // replication-send-timeout-ms.
+ if (IsFlagEnabled(kCloseAsync) || IsFlagEnabled(kCloseAfterReply) ||
IsFlagEnabled(kSlave)) return false;
+
+ auto kind = GetClientType() == kTypePubsub ? ClientKind::kPubsub :
ClientKind::kNormal;
+ const auto &limit = srv_->GetConfig()->GetClientOutputBufferLimit(kind);
+ uint64_t hard_limit_bytes =
limit.hard_limit_bytes.load(std::memory_order_relaxed);
+ uint64_t soft_limit_bytes =
limit.soft_limit_bytes.load(std::memory_order_relaxed);
+ if (hard_limit_bytes == 0 && soft_limit_bytes == 0) return false;
+
+ uint64_t used_bytes = evbuffer_get_length(Output());
+ if (hard_limit_bytes != 0 && used_bytes >= hard_limit_bytes) return true;
+
+ if (soft_limit_bytes != 0) {
+ if (used_bytes >= soft_limit_bytes) {
+ auto soft_limit_seconds =
static_cast<int64_t>(limit.soft_limit_seconds.load(std::memory_order_relaxed));
Review Comment:
Can we use the same type for limit.soft_limit_seconds and soft_limit_seconds
to avoid overflow?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
