You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@kvrocks.apache.org by "git-hulk (via GitHub)" <gi...@apache.org> on 2023/10/24 13:13:55 UTC
[PR] Allow to change the worker thread number in-flight [kvrocks]
git-hulk opened a new pull request, #1855:
URL: https://github.com/apache/kvrocks/pull/1855
TODO:
- [ ] Add test cases
- [ ] Update the owner when the connection is blocking on keys or stream
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375396597
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
+ auto new_worker_threads = static_cast<size_t>(config_->workers);
+ if (new_worker_threads == worker_threads_.size()) {
+ return;
+ }
+ size_t delta = 0;
+ if (new_worker_threads > worker_threads_.size()) {
+ delta = new_worker_threads - worker_threads_.size();
+ increaseWorkerThreads(delta);
+ LOG(INFO) << "[server] Increase worker threads to " << new_worker_threads;
+ return;
+ }
+
+ delta = worker_threads_.size() - new_worker_threads;
+ LOG(INFO) << "[server] Decrease worker threads to " << new_worker_threads;
+ decreaseWorkerThreads(delta);
+}
+
+void Server::increaseWorkerThreads(size_t delta) {
+ std::vector<std::unique_ptr<WorkerThread>> new_threads;
+ for (size_t i = 0; i < delta; i++) {
+ auto worker = std::make_unique<Worker>(this, config_);
+ auto worker_thread = std::make_unique<WorkerThread>(std::move(worker));
+ worker_thread->Start();
+ worker_threads_.emplace_back(std::move(worker_thread));
+ }
+}
+
+void Server::decreaseWorkerThreads(size_t delta) {
+ auto current_worker_threads = worker_threads_.size();
+ auto remain_worker_threads = current_worker_threads - delta;
+ for (size_t i = remain_worker_threads; i < current_worker_threads; i++) {
+ // Unix socket will be listening on the first worker,
+ // so it MUST remove workers from the end of the vector.
+ // Otherwise, the unix socket will be closed.
+ auto worker_thread = std::move(worker_threads_.back());
+ worker_threads_.pop_back();
+ // Migrate connections to other workers before stopping the worker,
+ // we use round-robin to choose the target worker here.
+ auto connections = worker_thread->GetWorker()->GetConnections();
+ for (const auto &iter : connections) {
+ auto target_worker = worker_threads_[iter.first % remain_worker_threads]->GetWorker();
+ worker_thread->GetWorker()->MigrateConnection(target_worker, iter.second);
+ }
+ worker_thread->Stop();
+ // Don't join the worker thread here, because it may join itself.
+ recycle_worker_threads_.push(std::move(worker_thread));
+ }
+}
+
+void Server::cleanupExitedWorkerThreads() {
+ std::unique_ptr<WorkerThread> worker_thread = nullptr;
Review Comment:
Yes
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375395791
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
+ auto new_worker_threads = static_cast<size_t>(config_->workers);
+ if (new_worker_threads == worker_threads_.size()) {
+ return;
+ }
+ size_t delta = 0;
+ if (new_worker_threads > worker_threads_.size()) {
+ delta = new_worker_threads - worker_threads_.size();
+ increaseWorkerThreads(delta);
+ LOG(INFO) << "[server] Increase worker threads to " << new_worker_threads;
+ return;
+ }
+
+ delta = worker_threads_.size() - new_worker_threads;
+ LOG(INFO) << "[server] Decrease worker threads to " << new_worker_threads;
+ decreaseWorkerThreads(delta);
+}
+
+void Server::increaseWorkerThreads(size_t delta) {
+ std::vector<std::unique_ptr<WorkerThread>> new_threads;
+ for (size_t i = 0; i < delta; i++) {
+ auto worker = std::make_unique<Worker>(this, config_);
+ auto worker_thread = std::make_unique<WorkerThread>(std::move(worker));
+ worker_thread->Start();
+ worker_threads_.emplace_back(std::move(worker_thread));
+ }
+}
+
+void Server::decreaseWorkerThreads(size_t delta) {
+ auto current_worker_threads = worker_threads_.size();
+ auto remain_worker_threads = current_worker_threads - delta;
Review Comment:
Yeah
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#issuecomment-1784049324
I'm a bit confused about the ctor ordering here:
1. `Server::~Server` calls `cleanupExitedWorkerThreads()`
2. cron calls `cleanupExitedWorkerThreads()`
However, when `Join`, would the exited worker thread be cleaned?
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "PragmaTwice (via GitHub)" <gi...@apache.org>.
PragmaTwice commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1374464238
##########
src/server/redis_connection.cc:
##########
@@ -74,6 +75,17 @@ void Connection::Close() {
void Connection::Detach() { owner_->DetachConnection(this); }
+bool Connection::IsBlockingMode() const {
+ if (current_cmd == nullptr) return false;
+ if (dynamic_cast<BlockingCommander *>(current_cmd.get()) != nullptr) return true;
+
+ auto args = current_cmd->GetArgs();
+ if (args.empty()) return false;
+
+ std::string cmd_name = util::ToLower(args.front());
+ return cmd_name == "xread";
Review Comment:
Seems it can cause data race since `args` are used in multiple threads. And the check `cmd_name == "xread"` is also a disaster of maintainability (we can easily forget while facing new blocking commands).
So I think a better solution is add an new attribute like `blocking` to these commands and check it here. Then we actually do not need `args` and even `dynamic_cast`
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#issuecomment-1784083494
Got it, so finally resource is handled by dtor.
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#issuecomment-1778839866
> nice, long connect will not disconnect when change workers, I use redis-benchmark continuous write
Thank you! I'm working on test cases.
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "PragmaTwice (via GitHub)" <gi...@apache.org>.
PragmaTwice commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1374464238
##########
src/server/redis_connection.cc:
##########
@@ -74,6 +75,17 @@ void Connection::Close() {
void Connection::Detach() { owner_->DetachConnection(this); }
+bool Connection::IsBlockingMode() const {
+ if (current_cmd == nullptr) return false;
+ if (dynamic_cast<BlockingCommander *>(current_cmd.get()) != nullptr) return true;
+
+ auto args = current_cmd->GetArgs();
+ if (args.empty()) return false;
+
+ std::string cmd_name = util::ToLower(args.front());
+ return cmd_name == "xread";
Review Comment:
Seems it can cause data race since `args` are used in multiple threads. And the check `cmd_name == "xread"` is really a disaster of maintainability (we can easily forget while facing new blocking commands).
So I think a better solution is add an new attribute like `blocking` to these commands and check it here. Then we actually do not need `args` and even `dynamic_cast`
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1374529815
##########
src/server/redis_connection.cc:
##########
@@ -74,6 +75,17 @@ void Connection::Close() {
void Connection::Detach() { owner_->DetachConnection(this); }
+bool Connection::IsBlockingMode() const {
+ if (current_cmd == nullptr) return false;
+ if (dynamic_cast<BlockingCommander *>(current_cmd.get()) != nullptr) return true;
+
+ auto args = current_cmd->GetArgs();
+ if (args.empty()) return false;
+
+ std::string cmd_name = util::ToLower(args.front());
+ return cmd_name == "xread";
Review Comment:
Gotcha, 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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#issuecomment-1780375620
Should we prevent from setting the thread-cnt to a dangerous value( .e.g `0`)?
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "PragmaTwice (via GitHub)" <gi...@apache.org>.
PragmaTwice commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1374464238
##########
src/server/redis_connection.cc:
##########
@@ -74,6 +75,17 @@ void Connection::Close() {
void Connection::Detach() { owner_->DetachConnection(this); }
+bool Connection::IsBlockingMode() const {
+ if (current_cmd == nullptr) return false;
+ if (dynamic_cast<BlockingCommander *>(current_cmd.get()) != nullptr) return true;
+
+ auto args = current_cmd->GetArgs();
+ if (args.empty()) return false;
+
+ std::string cmd_name = util::ToLower(args.front());
+ return cmd_name == "xread";
Review Comment:
Seems it can cause data race since args are used in multiple threads. And the check `cmd_name == "xread"` is really a disaster of maintainability (we can easily forget while facing new blocking commands).
So I think a better solution is add an new attribute like `blocking` to these commands and check it here.
##########
src/server/redis_connection.cc:
##########
@@ -74,6 +75,17 @@ void Connection::Close() {
void Connection::Detach() { owner_->DetachConnection(this); }
+bool Connection::IsBlockingMode() const {
+ if (current_cmd == nullptr) return false;
+ if (dynamic_cast<BlockingCommander *>(current_cmd.get()) != nullptr) return true;
+
+ auto args = current_cmd->GetArgs();
+ if (args.empty()) return false;
+
+ std::string cmd_name = util::ToLower(args.front());
+ return cmd_name == "xread";
Review Comment:
Seems it can cause data race since `args` are used in multiple threads. And the check `cmd_name == "xread"` is really a disaster of maintainability (we can easily forget while facing new blocking commands).
So I think a better solution is add an new attribute like `blocking` to these commands and check it here.
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375414255
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
+ auto new_worker_threads = static_cast<size_t>(config_->workers);
+ if (new_worker_threads == worker_threads_.size()) {
+ return;
+ }
+ size_t delta = 0;
+ if (new_worker_threads > worker_threads_.size()) {
+ delta = new_worker_threads - worker_threads_.size();
+ increaseWorkerThreads(delta);
+ LOG(INFO) << "[server] Increase worker threads to " << new_worker_threads;
+ return;
+ }
+
+ delta = worker_threads_.size() - new_worker_threads;
+ LOG(INFO) << "[server] Decrease worker threads to " << new_worker_threads;
+ decreaseWorkerThreads(delta);
+}
+
+void Server::increaseWorkerThreads(size_t delta) {
+ std::vector<std::unique_ptr<WorkerThread>> new_threads;
+ for (size_t i = 0; i < delta; i++) {
+ auto worker = std::make_unique<Worker>(this, config_);
+ auto worker_thread = std::make_unique<WorkerThread>(std::move(worker));
+ worker_thread->Start();
+ worker_threads_.emplace_back(std::move(worker_thread));
+ }
+}
+
+void Server::decreaseWorkerThreads(size_t delta) {
+ auto current_worker_threads = worker_threads_.size();
+ auto remain_worker_threads = current_worker_threads - delta;
+ for (size_t i = remain_worker_threads; i < current_worker_threads; i++) {
+ // Unix socket will be listening on the first worker,
+ // so it MUST remove workers from the end of the vector.
+ // Otherwise, the unix socket will be closed.
+ auto worker_thread = std::move(worker_threads_.back());
+ worker_threads_.pop_back();
+ // Migrate connections to other workers before stopping the worker,
+ // we use round-robin to choose the target worker here.
+ auto connections = worker_thread->GetWorker()->GetConnections();
+ for (const auto &iter : connections) {
+ auto target_worker = worker_threads_[iter.first % remain_worker_threads]->GetWorker();
+ worker_thread->GetWorker()->MigrateConnection(target_worker, iter.second);
+ }
+ worker_thread->Stop();
+ // Don't join the worker thread here, because it may join itself.
+ recycle_worker_threads_.push(std::move(worker_thread));
+ }
+}
+
+void Server::cleanupExitedWorkerThreads() {
+ std::unique_ptr<WorkerThread> worker_thread = nullptr;
Review Comment:
It has been done before destroying the server.
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375341120
##########
tests/gocase/unit/config/config_test.go:
##########
@@ -141,3 +144,92 @@ func TestStartWithoutConfigurationFile(t *testing.T) {
require.NoError(t, rdb.Do(ctx, "SET", "foo", "bar").Err())
require.Equal(t, "bar", rdb.Do(ctx, "GET", "foo").Val())
}
+
+func TestDynamicChangeWorkerThread(t *testing.T) {
+ configs := map[string]string{}
+ srv := util.StartServer(t, configs)
+ defer srv.Close()
+
+ ctx := context.Background()
+ rdb := srv.NewClientWithOption(&redis.Options{
+ MaxIdleConns: 20,
+ MaxRetries: -1, // Disable retry to check connections are alive after config change
+ })
+ defer func() { require.NoError(t, rdb.Close()) }()
+
+ t.Run("Test dynamic change worker thread", func(t *testing.T) {
+ runCommands := func(workers int) {
+ var wg sync.WaitGroup
+ require.NoError(t, rdb.Do(ctx, "CONFIG", "SET", "workers", strconv.Itoa(workers)).Err())
+ for i := 0; i < 10; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for j := 0; j < 10; j++ {
+ require.NoError(t, rdb.Set(ctx, "foo", "bar", 0).Err())
+ }
+ }()
+ }
+ wg.Wait()
+ }
+ // Reduce worker threads to 4
+ runCommands(4)
+
+ // Reduce worker threads to 1
+ runCommands(1)
+
+ // Reduce worker threads to 12
Review Comment:
Good catch, 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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk merged PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375431766
##########
src/server/server.cc:
##########
@@ -117,6 +117,9 @@ Server::~Server() {
}
lua::DestroyState(lua_);
Review Comment:
> Since conn_ctxs_ is removed, should we remove or change the comment in L114?
Yes, deleted.
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375400728
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
+ auto new_worker_threads = static_cast<size_t>(config_->workers);
+ if (new_worker_threads == worker_threads_.size()) {
+ return;
+ }
+ size_t delta = 0;
+ if (new_worker_threads > worker_threads_.size()) {
+ delta = new_worker_threads - worker_threads_.size();
+ increaseWorkerThreads(delta);
+ LOG(INFO) << "[server] Increase worker threads to " << new_worker_threads;
+ return;
+ }
+
+ delta = worker_threads_.size() - new_worker_threads;
+ LOG(INFO) << "[server] Decrease worker threads to " << new_worker_threads;
+ decreaseWorkerThreads(delta);
+}
+
+void Server::increaseWorkerThreads(size_t delta) {
+ std::vector<std::unique_ptr<WorkerThread>> new_threads;
+ for (size_t i = 0; i < delta; i++) {
+ auto worker = std::make_unique<Worker>(this, config_);
+ auto worker_thread = std::make_unique<WorkerThread>(std::move(worker));
+ worker_thread->Start();
+ worker_threads_.emplace_back(std::move(worker_thread));
+ }
+}
+
+void Server::decreaseWorkerThreads(size_t delta) {
+ auto current_worker_threads = worker_threads_.size();
+ auto remain_worker_threads = current_worker_threads - delta;
+ for (size_t i = remain_worker_threads; i < current_worker_threads; i++) {
+ // Unix socket will be listening on the first worker,
+ // so it MUST remove workers from the end of the vector.
+ // Otherwise, the unix socket will be closed.
+ auto worker_thread = std::move(worker_threads_.back());
+ worker_threads_.pop_back();
+ // Migrate connections to other workers before stopping the worker,
+ // we use round-robin to choose the target worker here.
+ auto connections = worker_thread->GetWorker()->GetConnections();
+ for (const auto &iter : connections) {
+ auto target_worker = worker_threads_[iter.first % remain_worker_threads]->GetWorker();
+ worker_thread->GetWorker()->MigrateConnection(target_worker, iter.second);
+ }
+ worker_thread->Stop();
+ // Don't join the worker thread here, because it may join itself.
+ recycle_worker_threads_.push(std::move(worker_thread));
+ }
+}
+
+void Server::cleanupExitedWorkerThreads() {
+ std::unique_ptr<WorkerThread> worker_thread = nullptr;
Review Comment:
Also, should `Server::Join` also try to check `recycle_worker_threads_`?
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375394910
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
Review Comment:
It's unnecessary since Config::Set is an exclusive command, so it won't have the data race condition.
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "torwig (via GitHub)" <gi...@apache.org>.
torwig commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375265538
##########
src/server/worker.cc:
##########
@@ -344,6 +344,31 @@ redis::Connection *Worker::removeConnection(int fd) {
return conn;
}
+// MigrateConnection moves the connection to another worker
+// when reducing the number of workers.
+//
+// To make it simple, we would close the connection if it's
+// blocked on a key or stream.
+void Worker::MigrateConnection(Worker *target, redis::Connection *conn) {
+ if (!target || !conn) return;
+ if (conn->current_cmd != nullptr && conn->current_cmd->IsBlocking()) {
+ // don't need to close the connection since destroy worker thread will close it
+ return;
+ }
+
Review Comment:
Could any blocking command be received after the previous check and before the actual connection migrating?
##########
tests/gocase/unit/config/config_test.go:
##########
@@ -141,3 +144,92 @@ func TestStartWithoutConfigurationFile(t *testing.T) {
require.NoError(t, rdb.Do(ctx, "SET", "foo", "bar").Err())
require.Equal(t, "bar", rdb.Do(ctx, "GET", "foo").Val())
}
+
+func TestDynamicChangeWorkerThread(t *testing.T) {
+ configs := map[string]string{}
+ srv := util.StartServer(t, configs)
+ defer srv.Close()
+
+ ctx := context.Background()
+ rdb := srv.NewClientWithOption(&redis.Options{
+ MaxIdleConns: 20,
+ MaxRetries: -1, // Disable retry to check connections are alive after config change
+ })
+ defer func() { require.NoError(t, rdb.Close()) }()
+
+ t.Run("Test dynamic change worker thread", func(t *testing.T) {
+ runCommands := func(workers int) {
+ var wg sync.WaitGroup
+ require.NoError(t, rdb.Do(ctx, "CONFIG", "SET", "workers", strconv.Itoa(workers)).Err())
+ for i := 0; i < 10; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for j := 0; j < 10; j++ {
+ require.NoError(t, rdb.Set(ctx, "foo", "bar", 0).Err())
+ }
+ }()
+ }
+ wg.Wait()
+ }
+ // Reduce worker threads to 4
+ runCommands(4)
+
+ // Reduce worker threads to 1
+ runCommands(1)
+
+ // Reduce worker threads to 12
Review Comment:
Just nitpicking: in this case, we **increasing** worker threads to 12 :)
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375395917
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
+ auto new_worker_threads = static_cast<size_t>(config_->workers);
+ if (new_worker_threads == worker_threads_.size()) {
+ return;
+ }
+ size_t delta = 0;
+ if (new_worker_threads > worker_threads_.size()) {
+ delta = new_worker_threads - worker_threads_.size();
+ increaseWorkerThreads(delta);
+ LOG(INFO) << "[server] Increase worker threads to " << new_worker_threads;
+ return;
+ }
+
+ delta = worker_threads_.size() - new_worker_threads;
+ LOG(INFO) << "[server] Decrease worker threads to " << new_worker_threads;
+ decreaseWorkerThreads(delta);
+}
+
+void Server::increaseWorkerThreads(size_t delta) {
+ std::vector<std::unique_ptr<WorkerThread>> new_threads;
+ for (size_t i = 0; i < delta; i++) {
+ auto worker = std::make_unique<Worker>(this, config_);
+ auto worker_thread = std::make_unique<WorkerThread>(std::move(worker));
+ worker_thread->Start();
+ worker_threads_.emplace_back(std::move(worker_thread));
+ }
+}
+
+void Server::decreaseWorkerThreads(size_t delta) {
+ auto current_worker_threads = worker_threads_.size();
+ auto remain_worker_threads = current_worker_threads - delta;
+ for (size_t i = remain_worker_threads; i < current_worker_threads; i++) {
+ // Unix socket will be listening on the first worker,
+ // so it MUST remove workers from the end of the vector.
+ // Otherwise, the unix socket will be closed.
+ auto worker_thread = std::move(worker_threads_.back());
+ worker_threads_.pop_back();
+ // Migrate connections to other workers before stopping the worker,
+ // we use round-robin to choose the target worker here.
+ auto connections = worker_thread->GetWorker()->GetConnections();
+ for (const auto &iter : connections) {
+ auto target_worker = worker_threads_[iter.first % remain_worker_threads]->GetWorker();
+ worker_thread->GetWorker()->MigrateConnection(target_worker, iter.second);
+ }
+ worker_thread->Stop();
+ // Don't join the worker thread here, because it may join itself.
+ recycle_worker_threads_.push(std::move(worker_thread));
+ }
+}
+
+void Server::cleanupExitedWorkerThreads() {
+ std::unique_ptr<WorkerThread> worker_thread = nullptr;
Review Comment:
(That's why we only need DCHECK?)
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375423088
##########
src/server/server.cc:
##########
@@ -117,6 +117,9 @@ Server::~Server() {
}
lua::DestroyState(lua_);
Review Comment:
Also. should we adjust the ordering of `lua::DestroyState` and cleanup?( not so familiar with lua here, so don't know would we meet the problem)
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1374636419
##########
src/server/server.cc:
##########
@@ -417,8 +419,6 @@ void Server::SubscribeChannel(const std::string &channel, redis::Connection *con
std::lock_guard<std::mutex> guard(pubsub_channels_mu_);
auto conn_ctx = ConnContext(conn->Owner(), conn->GetFD());
- conn_ctxs_[conn_ctx] = true;
Review Comment:
No where is using conn_ctxs_, so delete 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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#issuecomment-1780829110
> Should we prevent from setting the thread-cnt to a dangerous value( .e.g `0`)?
@mapleFU Yes, IntField will check the value range before setting.
`{"workers", true, new IntField(&workers, 8, 1, 256)},`
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375429814
##########
src/server/server.cc:
##########
@@ -117,6 +117,9 @@ Server::~Server() {
}
lua::DestroyState(lua_);
Review Comment:
It should be fine to leave `lua::DestroyState` as it is because workers are stopped.
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375394683
##########
src/server/server.cc:
##########
@@ -245,6 +244,9 @@ void Server::Join() {
if (auto s = task_runner_.Join(); !s) {
LOG(WARNING) << s.Msg();
}
+ for (const auto &worker : worker_threads_) {
Review Comment:
It doesn't wait for other threads but can't be the first one, because worker_threads_ may be modified after starting, so I move it to the last.
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#issuecomment-1784081381
> I'm a bit confused about the ctor ordering here:
>
> 1. `Server::~Server` calls `cleanupExitedWorkerThreads()`
> 2. cron calls `cleanupExitedWorkerThreads()`
>
> However, when `Join`, would the exited worker thread be cleaned?
The `cron` thread periodic call `cleanupExitedWorkerThreads()` but it can't promise it will call `cleanupExitedWorkerThreads()` before exiting the cron thread. So for the server destructor, it MUST check again to prevent exceptions.
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375422895
##########
src/server/server.cc:
##########
@@ -117,6 +117,9 @@ Server::~Server() {
}
lua::DestroyState(lua_);
Review Comment:
Since `conn_ctxs_` is freed, should we remove or change the comment in L114?
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1374632624
##########
src/config/config.cc:
##########
@@ -359,296 +359,304 @@ void Config::initFieldCallback() {
};
#endif
- std::map<std::string, CallbackFn> callbacks = {
- {"dir",
- [this](Server *srv, const std::string &k, const std::string &v) -> Status {
- db_dir = dir + "/db";
- {
- std::lock_guard<std::mutex> lg(this->backup_mu);
- if (backup_dir.empty()) {
- backup_dir = dir + "/backup";
- }
- }
- if (log_dir.empty()) log_dir = dir;
- checkpoint_dir = dir + "/checkpoint";
- sync_checkpoint_dir = dir + "/sync_checkpoint";
- backup_sync_dir = dir + "/backup_for_sync";
- return Status::OK();
- }},
- {"backup-dir",
- [this](Server *srv, const std::string &k, const std::string &v) -> Status {
- std::string previous_backup;
- {
- // Note: currently, backup_mu_ may block by backing up or purging,
- // the command may wait for seconds.
- std::lock_guard<std::mutex> lg(this->backup_mu);
- previous_backup = std::move(backup_dir);
- backup_dir = v;
- }
- if (!previous_backup.empty() && srv != nullptr && !srv->IsLoading()) {
- // LOG(INFO) should be called after log is initialized and server is loaded.
- LOG(INFO) << "change backup dir from " << previous_backup << " to " << v;
- }
- return Status::OK();
- }},
- {"cluster-enabled",
- [this](Server *srv, const std::string &k, const std::string &v) -> Status {
- if (cluster_enabled) slot_id_encoded = true;
- return Status::OK();
- }},
- {"bind",
- [this](Server *srv, const std::string &k, const std::string &v) -> Status {
- std::vector<std::string> args = util::Split(v, " \t");
- binds = std::move(args);
- return Status::OK();
- }},
- {"maxclients",
- [](Server *srv, const std::string &k, const std::string &v) -> Status {
- if (!srv) return Status::OK();
- srv->AdjustOpenFilesLimit();
- return Status::OK();
- }},
- {"slaveof",
- [this](Server *srv, const std::string &k, const std::string &v) -> Status {
- if (v.empty()) {
- return Status::OK();
- }
- std::vector<std::string> args = util::Split(v, " \t");
- if (args.size() != 2) return {Status::NotOK, "wrong number of arguments"};
- if (args[0] != "no" && args[1] != "one") {
- master_host = args[0];
- auto parse_result = ParseInt<int>(args[1], NumericRange<int>{1, PORT_LIMIT - 1}, 10);
- if (!parse_result) {
- return {Status::NotOK, "should be between 0 and 65535"};
- }
- master_port = *parse_result;
- }
- return Status::OK();
- }},
- {"profiling-sample-commands",
- [this](Server *srv, const std::string &k, const std::string &v) -> Status {
- std::vector<std::string> cmds = util::Split(v, ",");
- profiling_sample_all_commands = false;
- profiling_sample_commands.clear();
- for (auto const &cmd : cmds) {
- if (cmd == "*") {
- profiling_sample_all_commands = true;
+ std::map<std::string, CallbackFn> callbacks =
+ {
+ {"workers",
+ [](Server *srv, const std::string &k, const std::string &v) -> Status {
+ if (!srv) return Status::OK();
+ srv->AdjustWorkerThreads();
+ return Status::OK();
+ }},
Review Comment:
Only those lines are newly added, rest is formatted by clang-format
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375422895
##########
src/server/server.cc:
##########
@@ -117,6 +117,9 @@ Server::~Server() {
}
lua::DestroyState(lua_);
Review Comment:
Since `conn_ctxs_` is removed, should we remove or change the comment in L114?
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375392220
##########
src/server/server.cc:
##########
@@ -245,6 +244,9 @@ void Server::Join() {
if (auto s = task_runner_.Join(); !s) {
LOG(WARNING) << s.Msg();
}
+ for (const auto &worker : worker_threads_) {
Review Comment:
Just curious, why worker would shutdown after the other threads shutdown?
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
Review Comment:
Do we need some kind of locking ( Or CAS ) here to avoid concurrent `AdjustWorkerThreads`?
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
+ auto new_worker_threads = static_cast<size_t>(config_->workers);
+ if (new_worker_threads == worker_threads_.size()) {
+ return;
+ }
+ size_t delta = 0;
+ if (new_worker_threads > worker_threads_.size()) {
+ delta = new_worker_threads - worker_threads_.size();
+ increaseWorkerThreads(delta);
+ LOG(INFO) << "[server] Increase worker threads to " << new_worker_threads;
+ return;
+ }
+
+ delta = worker_threads_.size() - new_worker_threads;
+ LOG(INFO) << "[server] Decrease worker threads to " << new_worker_threads;
+ decreaseWorkerThreads(delta);
+}
+
+void Server::increaseWorkerThreads(size_t delta) {
+ std::vector<std::unique_ptr<WorkerThread>> new_threads;
+ for (size_t i = 0; i < delta; i++) {
+ auto worker = std::make_unique<Worker>(this, config_);
+ auto worker_thread = std::make_unique<WorkerThread>(std::move(worker));
+ worker_thread->Start();
+ worker_threads_.emplace_back(std::move(worker_thread));
+ }
+}
+
+void Server::decreaseWorkerThreads(size_t delta) {
+ auto current_worker_threads = worker_threads_.size();
+ auto remain_worker_threads = current_worker_threads - delta;
Review Comment:
Should we better adding a `DCHECK(current_worker_threads > delta);`
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
+ auto new_worker_threads = static_cast<size_t>(config_->workers);
+ if (new_worker_threads == worker_threads_.size()) {
+ return;
+ }
+ size_t delta = 0;
+ if (new_worker_threads > worker_threads_.size()) {
+ delta = new_worker_threads - worker_threads_.size();
+ increaseWorkerThreads(delta);
+ LOG(INFO) << "[server] Increase worker threads to " << new_worker_threads;
+ return;
+ }
+
+ delta = worker_threads_.size() - new_worker_threads;
+ LOG(INFO) << "[server] Decrease worker threads to " << new_worker_threads;
+ decreaseWorkerThreads(delta);
+}
+
+void Server::increaseWorkerThreads(size_t delta) {
+ std::vector<std::unique_ptr<WorkerThread>> new_threads;
+ for (size_t i = 0; i < delta; i++) {
+ auto worker = std::make_unique<Worker>(this, config_);
+ auto worker_thread = std::make_unique<WorkerThread>(std::move(worker));
+ worker_thread->Start();
+ worker_threads_.emplace_back(std::move(worker_thread));
+ }
+}
+
+void Server::decreaseWorkerThreads(size_t delta) {
+ auto current_worker_threads = worker_threads_.size();
+ auto remain_worker_threads = current_worker_threads - delta;
+ for (size_t i = remain_worker_threads; i < current_worker_threads; i++) {
+ // Unix socket will be listening on the first worker,
+ // so it MUST remove workers from the end of the vector.
+ // Otherwise, the unix socket will be closed.
+ auto worker_thread = std::move(worker_threads_.back());
+ worker_threads_.pop_back();
+ // Migrate connections to other workers before stopping the worker,
+ // we use round-robin to choose the target worker here.
+ auto connections = worker_thread->GetWorker()->GetConnections();
+ for (const auto &iter : connections) {
+ auto target_worker = worker_threads_[iter.first % remain_worker_threads]->GetWorker();
+ worker_thread->GetWorker()->MigrateConnection(target_worker, iter.second);
+ }
+ worker_thread->Stop();
+ // Don't join the worker thread here, because it may join itself.
+ recycle_worker_threads_.push(std::move(worker_thread));
+ }
+}
+
+void Server::cleanupExitedWorkerThreads() {
+ std::unique_ptr<WorkerThread> worker_thread = nullptr;
Review Comment:
Should we dcheck thread-id not eq?
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375395364
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
+ auto new_worker_threads = static_cast<size_t>(config_->workers);
+ if (new_worker_threads == worker_threads_.size()) {
+ return;
+ }
+ size_t delta = 0;
+ if (new_worker_threads > worker_threads_.size()) {
+ delta = new_worker_threads - worker_threads_.size();
+ increaseWorkerThreads(delta);
+ LOG(INFO) << "[server] Increase worker threads to " << new_worker_threads;
+ return;
+ }
+
+ delta = worker_threads_.size() - new_worker_threads;
+ LOG(INFO) << "[server] Decrease worker threads to " << new_worker_threads;
+ decreaseWorkerThreads(delta);
+}
+
+void Server::increaseWorkerThreads(size_t delta) {
+ std::vector<std::unique_ptr<WorkerThread>> new_threads;
+ for (size_t i = 0; i < delta; i++) {
+ auto worker = std::make_unique<Worker>(this, config_);
+ auto worker_thread = std::make_unique<WorkerThread>(std::move(worker));
+ worker_thread->Start();
+ worker_threads_.emplace_back(std::move(worker_thread));
+ }
+}
+
+void Server::decreaseWorkerThreads(size_t delta) {
+ auto current_worker_threads = worker_threads_.size();
+ auto remain_worker_threads = current_worker_threads - delta;
+ for (size_t i = remain_worker_threads; i < current_worker_threads; i++) {
+ // Unix socket will be listening on the first worker,
+ // so it MUST remove workers from the end of the vector.
+ // Otherwise, the unix socket will be closed.
+ auto worker_thread = std::move(worker_threads_.back());
+ worker_threads_.pop_back();
+ // Migrate connections to other workers before stopping the worker,
+ // we use round-robin to choose the target worker here.
+ auto connections = worker_thread->GetWorker()->GetConnections();
+ for (const auto &iter : connections) {
+ auto target_worker = worker_threads_[iter.first % remain_worker_threads]->GetWorker();
+ worker_thread->GetWorker()->MigrateConnection(target_worker, iter.second);
+ }
+ worker_thread->Stop();
+ // Don't join the worker thread here, because it may join itself.
+ recycle_worker_threads_.push(std::move(worker_thread));
+ }
+}
+
+void Server::cleanupExitedWorkerThreads() {
+ std::unique_ptr<WorkerThread> worker_thread = nullptr;
Review Comment:
I guess it's fine since those threads are from decreaseWorkerThreads
##########
src/server/server.cc:
##########
@@ -1680,6 +1677,64 @@ void Server::AdjustOpenFilesLimit() {
}
}
+void Server::AdjustWorkerThreads() {
+ auto new_worker_threads = static_cast<size_t>(config_->workers);
+ if (new_worker_threads == worker_threads_.size()) {
+ return;
+ }
+ size_t delta = 0;
+ if (new_worker_threads > worker_threads_.size()) {
+ delta = new_worker_threads - worker_threads_.size();
+ increaseWorkerThreads(delta);
+ LOG(INFO) << "[server] Increase worker threads to " << new_worker_threads;
+ return;
+ }
+
+ delta = worker_threads_.size() - new_worker_threads;
+ LOG(INFO) << "[server] Decrease worker threads to " << new_worker_threads;
+ decreaseWorkerThreads(delta);
+}
+
+void Server::increaseWorkerThreads(size_t delta) {
+ std::vector<std::unique_ptr<WorkerThread>> new_threads;
+ for (size_t i = 0; i < delta; i++) {
+ auto worker = std::make_unique<Worker>(this, config_);
+ auto worker_thread = std::make_unique<WorkerThread>(std::move(worker));
+ worker_thread->Start();
+ worker_threads_.emplace_back(std::move(worker_thread));
+ }
+}
+
+void Server::decreaseWorkerThreads(size_t delta) {
+ auto current_worker_threads = worker_threads_.size();
+ auto remain_worker_threads = current_worker_threads - delta;
Review Comment:
@mapleFU Do you mean this?
```suggestion
DCHECK(current_worker_threads > delta);
auto remain_worker_threads = current_worker_threads - delta;
```
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#discussion_r1375346412
##########
src/server/worker.cc:
##########
@@ -344,6 +344,31 @@ redis::Connection *Worker::removeConnection(int fd) {
return conn;
}
+// MigrateConnection moves the connection to another worker
+// when reducing the number of workers.
+//
+// To make it simple, we would close the connection if it's
+// blocked on a key or stream.
+void Worker::MigrateConnection(Worker *target, redis::Connection *conn) {
+ if (!target || !conn) return;
+ if (conn->current_cmd != nullptr && conn->current_cmd->IsBlocking()) {
+ // don't need to close the connection since destroy worker thread will close it
+ return;
+ }
+
Review Comment:
It's not since Config::Set is an exclusive command, so it won't receive any other commands before this process was finished.
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#issuecomment-1784327393
Thanks all, merging..
--
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] Allow to change the worker thread number in-flight [kvrocks]
Posted by "guojidan (via GitHub)" <gi...@apache.org>.
guojidan commented on PR #1855:
URL: https://github.com/apache/kvrocks/pull/1855#issuecomment-1778808832
nice, long connect will not disconnect when change workers, I use redis-benchmark continuous write
--
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