Aetherance commented on code in PR #3541: URL: https://github.com/apache/kvrocks/pull/3541#discussion_r3673781395
########## src/common/keyspace_events.cc: ########## @@ -0,0 +1,96 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +#include "keyspace_events.h" + +#include <cstring> +#include <utility> + +#include "config/config.h" +#include "fmt/format.h" + +bool ShouldNotifyKeyspaceEvent(int notify_flags, int type_flag) { Review Comment: A notification is generated here only when both conditions are met: the event type is enabled, and at least one of K or E is enabled. ########## src/server/redis_connection.cc: ########## @@ -466,7 +467,17 @@ Status Connection::ExecuteCommand(engine::Context &ctx, const std::string &cmd_n auto start = std::chrono::high_resolution_clock::now(); bool is_profiling = IsProfilingEnabled(cmd_name); + + keyspace_event_notify_flags_ = srv_->GetConfig()->notify_keyspace_events; Review Comment: The keyspace_event_collector is created only when keyspace notifications are enabled, and notifications are emitted only when the collector exists. Therefore, when keyspace notifications are disabled—or when a command’s event type is not enabled—the command skips all notification-related work except for a few lightweight condition checks. As a result, the performance impact in these cases should be negligible. ########## src/commands/cmd_key.cc: ########## @@ -372,9 +374,17 @@ class CommandDel : public Commander { uint64_t cnt = 0; redis::Database redis(srv->storage, conn->GetNamespace()); - auto s = redis.MDel(ctx, keys, &cnt); + const bool notify_del = GetAttributes()->name == "del" && conn->IsKeyspaceEventEnabled(kNotifyGeneric); Review Comment: UNLINK is intentionally excluded here. Whether to support it will be decided in a follow-up PR or discussion. ########## src/types/redis_string.cc: ########## @@ -239,7 +239,9 @@ rocksdb::Status String::Set(engine::Context &ctx, const std::string &user_key, c } rocksdb::Status String::Set(engine::Context &ctx, const std::string &user_key, const std::string &value, - const StringSetArgs &args, std::optional<std::string> &ret) { + const StringSetArgs &args, std::optional<std::string> &ret, bool *applied) { Review Comment: Here, `Status::OK` does not necessarily mean that a conditional `SET` was applied, so an additional `applied` flag is needed. So does it in `DEL` ########## src/server/redis_connection.cc: ########## @@ -709,10 +734,40 @@ void Connection::ExecuteCommands(std::deque<CommandTokens> *to_process_cmds) { } } +void Connection::queueOrPublishKeyspaceEvents(std::vector<KeyspaceEvent> &&events) { + if (events.empty()) return; + + if (in_exec_) { + // Queue transaction events until commit. + for (auto &event : events) { + pending_keyspace_events_.emplace_back(std::move(event)); + } + return; + } + + for (const auto &event : events) { + srv_->NotifyKeyspaceEvent(event.channel_flags, event.event, event.ns, event.key); + } +} + +void Connection::FlushKeyspaceEvents() { + for (const auto &e : pending_keyspace_events_) { + srv_->NotifyKeyspaceEvent(e.channel_flags, e.event, e.ns, e.key); + } + pending_keyspace_events_.clear(); +} + void Connection::ResetMultiExec() { in_exec_ = false; multi_error_ = false; multi_cmds_.clear(); + // Drop events from failed or aborted transactions. + pending_keyspace_events_.clear(); + // Retain capacity for typical transactions, but request releasing unusually large buffers. + constexpr std::size_t kMaxRetainedKeyspaceEvents = 1024; + if (pending_keyspace_events_.capacity() > kMaxRetainedKeyspaceEvents) { + pending_keyspace_events_.shrink_to_fit(); Review Comment: clear() does not release the vector’s capacity, so a large transaction may cause a long-lived connection to retain excessive memory and potentially lead to OOM. To mitigate this, I call shrink_to_fit() when the capacity exceeds 1,024. For small transactions, we can simply call `clear()` and reuse the allocated memory. ########## src/common/keyspace_events.h: ########## @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +#pragma once + +#include <string> +#include <string_view> +#include <vector> + +#include "status.h" + +// Flags for notify-keyspace-events, separate from RedisType. Review Comment: To reduce the review burden and keep the scope of the changes manageable, this PR only adds support for the SET and DEL commands. Currently, only the $ and g event classes are supported. Support for other commands will be added in follow-up PRs. ########## src/server/redis_connection.cc: ########## @@ -466,7 +467,17 @@ Status Connection::ExecuteCommand(engine::Context &ctx, const std::string &cmd_n auto start = std::chrono::high_resolution_clock::now(); bool is_profiling = IsProfilingEnabled(cmd_name); + + keyspace_event_notify_flags_ = srv_->GetConfig()->notify_keyspace_events; Review Comment: By the way, support for additional commands can be added by simply calling conn->AddKeyspaceEvent. Some minor changes may also be needed to accurately capture the full semantics of the event, but no major changes to the notification mechanism should be necessary. ########## src/server/server.cc: ########## @@ -478,6 +479,17 @@ int Server::PublishMessage(const std::string &channel, const std::string &msg) { return cnt; } +void Server::NotifyKeyspaceEvent(int flags, const std::string &event, const std::string &ns, const std::string &key) { + const std::string db = MapNamespaceToKeyspaceDB(ns, GetConfig()->redis_databases); Review Comment: Kvrocks has both ns and db. `MapNamespaceToKeyspaceDB` provides a unified representation of the two when naming Pub/Sub channels. -- 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]
