You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kvrocks.apache.org by tw...@apache.org on 2023/03/09 06:24:31 UTC
[incubator-kvrocks] branch unstable updated: Add CreateThread to avoid manual try-catch (#1302)
This is an automated email from the ASF dual-hosted git repository.
twice pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/incubator-kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new d48edac7 Add CreateThread to avoid manual try-catch (#1302)
d48edac7 is described below
commit d48edac7100bdc94e247b0aa8279335d39c80032
Author: Twice <tw...@gmail.com>
AuthorDate: Thu Mar 9 14:24:25 2023 +0800
Add CreateThread to avoid manual try-catch (#1302)
---
src/cluster/replication.cc | 63 +++++++++++++++++++++--------------------
src/cluster/slot_migrate.cc | 17 +++++------
src/commands/cmd_replication.cc | 20 +++++++------
src/common/task_runner.cc | 6 ++--
src/common/thread_util.cc | 16 +++++++++++
src/common/thread_util.h | 21 ++++++++++++++
src/server/server.cc | 18 ++++++------
src/server/worker.cc | 18 ++++++------
8 files changed, 111 insertions(+), 68 deletions(-)
diff --git a/src/cluster/replication.cc b/src/cluster/replication.cc
index 51a86571..073dbe06 100644
--- a/src/cluster/replication.cc
+++ b/src/cluster/replication.cc
@@ -46,28 +46,29 @@
#include "time_util.h"
Status FeedSlaveThread::Start() {
- try {
- t_ = std::thread([this]() {
- Util::ThreadSetName("feed-replica");
- sigset_t mask, omask;
- sigemptyset(&mask);
- sigemptyset(&omask);
- sigaddset(&mask, SIGCHLD);
- sigaddset(&mask, SIGHUP);
- sigaddset(&mask, SIGPIPE);
- pthread_sigmask(SIG_BLOCK, &mask, &omask);
- auto s = Util::SockSend(conn_->GetFD(), "+OK\r\n");
- if (!s.IsOK()) {
- LOG(ERROR) << "failed to send OK response to the replica: " << s.Msg();
- return;
- }
- this->loop();
- });
- } catch (const std::system_error &e) {
+ auto s = Util::CreateThread("feed-replica", [this] {
+ sigset_t mask, omask;
+ sigemptyset(&mask);
+ sigemptyset(&omask);
+ sigaddset(&mask, SIGCHLD);
+ sigaddset(&mask, SIGHUP);
+ sigaddset(&mask, SIGPIPE);
+ pthread_sigmask(SIG_BLOCK, &mask, &omask);
+ auto s = Util::SockSend(conn_->GetFD(), "+OK\r\n");
+ if (!s.IsOK()) {
+ LOG(ERROR) << "failed to send OK response to the replica: " << s.Msg();
+ return;
+ }
+ this->loop();
+ });
+
+ if (s) {
+ t_ = std::move(*s);
+ } else {
conn_ = nullptr; // prevent connection was freed when failed to start the thread
- return {Status::NotOK, e.what()};
}
- return Status::OK();
+
+ return s;
}
void FeedSlaveThread::Stop() {
@@ -76,7 +77,9 @@ void FeedSlaveThread::Stop() {
}
void FeedSlaveThread::Join() {
- if (t_.joinable()) t_.join();
+ if (auto s = Util::ThreadJoin(t_); !s) {
+ LOG(WARNING) << "Slave thread operation failed: " << s.Msg();
+ }
}
void FeedSlaveThread::checkLivenessIfNeed() {
@@ -323,15 +326,11 @@ Status ReplicationThread::Start(std::function<void()> &&pre_fullsync_cb, std::fu
// cleanup the old backups, so we can start replication in a clean state
storage_->PurgeOldBackups(0, 0);
- try {
- t_ = std::thread([this]() {
- Util::ThreadSetName("master-repl");
- this->run();
- assert(stop_flag_);
- });
- } catch (const std::system_error &e) {
- return Status(Status::NotOK, e.what());
- }
+ t_ = GET_OR_RET(Util::CreateThread("master-repl", [this] {
+ this->run();
+ assert(stop_flag_);
+ }));
+
return Status::OK();
}
@@ -340,7 +339,9 @@ void ReplicationThread::Stop() {
stop_flag_ = true; // Stopping procedure is asynchronous,
// handled by timer
- if (t_.joinable()) t_.join();
+ if (auto s = Util::ThreadJoin(t_); !s) {
+ LOG(WARNING) << "Replication thread operation failed: " << s.Msg();
+ }
LOG(INFO) << "[replication] Stopped";
}
diff --git a/src/cluster/slot_migrate.cc b/src/cluster/slot_migrate.cc
index 24e1f806..b62562ac 100644
--- a/src/cluster/slot_migrate.cc
+++ b/src/cluster/slot_migrate.cc
@@ -118,20 +118,17 @@ SlotMigrate::~SlotMigrate() {
stop_migrate_ = true;
thread_state_ = ThreadState::Terminated;
job_cv_.notify_all();
- if (t_.joinable()) t_.join();
+ if (auto s = Util::ThreadJoin(t_); !s) {
+ LOG(WARNING) << "Slot migrating thread operation failed: " << s.Msg();
+ }
}
}
Status SlotMigrate::CreateMigrateHandleThread() {
- try {
- t_ = std::thread([this]() {
- Util::ThreadSetName("slot-migrate");
- thread_state_ = ThreadState::Running;
- this->Loop();
- });
- } catch (const std::exception &e) {
- return {Status::NotOK, std::string(e.what())};
- }
+ t_ = GET_OR_RET(Util::CreateThread("slot-migrate", [this] {
+ thread_state_ = ThreadState::Running;
+ this->Loop();
+ }));
return Status::OK();
}
diff --git a/src/commands/cmd_replication.cc b/src/commands/cmd_replication.cc
index 5660c81c..369b4307 100644
--- a/src/commands/cmd_replication.cc
+++ b/src/commands/cmd_replication.cc
@@ -217,8 +217,7 @@ class CommandFetchMeta : public Commander {
svr->stats_.IncrFullSyncCounter();
// Feed-replica-meta thread
- std::thread t = std::thread([svr, repl_fd, ip, bev = conn->GetBufferEvent()]() {
- Util::ThreadSetName("feed-repl-info");
+ auto t = GET_OR_RET(Util::CreateThread("feed-repl-info", [svr, repl_fd, ip, bev = conn->GetBufferEvent()] {
svr->IncrFetchFileThread();
auto exit = MakeScopeExit([svr, bev] {
bufferevent_free(bev);
@@ -243,8 +242,11 @@ class CommandFetchMeta : public Commander {
}
auto now = static_cast<time_t>(Util::GetTimeStamp());
svr->storage_->SetCheckpointAccessTime(now);
- });
- t.detach();
+ }));
+
+ if (auto s = Util::ThreadDetach(t); !s) {
+ return s;
+ }
return Status::OK();
}
@@ -271,8 +273,7 @@ class CommandFetchFile : public Commander {
conn->NeedNotFreeBufferEvent(); // Feed-replica-file thread will close the replica bufferevent
conn->EnableFlag(Redis::Connection::kCloseAsync);
- std::thread t = std::thread([svr, repl_fd, ip, files, bev = conn->GetBufferEvent()]() {
- Util::ThreadSetName("feed-repl-file");
+ auto t = GET_OR_RET(Util::CreateThread("feed-repl-file", [svr, repl_fd, ip, files, bev = conn->GetBufferEvent()]() {
auto exit = MakeScopeExit([bev] { bufferevent_free(bev); });
svr->IncrFetchFileThread();
@@ -311,8 +312,11 @@ class CommandFetchFile : public Commander {
auto now = static_cast<time_t>(Util::GetTimeStamp());
svr->storage_->SetCheckpointAccessTime(now);
svr->DecrFetchFileThread();
- });
- t.detach();
+ }));
+
+ if (auto s = Util::ThreadDetach(t); !s) {
+ return s;
+ }
return Status::OK();
}
diff --git a/src/common/task_runner.cc b/src/common/task_runner.cc
index 9efe930e..63852d46 100644
--- a/src/common/task_runner.cc
+++ b/src/common/task_runner.cc
@@ -42,7 +42,7 @@ Status TaskRunner::Publish(const Task &task) {
void TaskRunner::Start() {
stop_ = false;
for (int i = 0; i < n_thread_; i++) {
- threads_.emplace_back([this]() {
+ threads_.emplace_back([this] {
Util::ThreadSetName("task-runner");
this->run();
});
@@ -57,7 +57,9 @@ void TaskRunner::Stop() {
void TaskRunner::Join() {
for (auto &thread : threads_) {
- if (thread.joinable()) thread.join();
+ if (auto s = Util::ThreadJoin(thread); !s) {
+ LOG(WARNING) << "Task thread operation failed: " << s.Msg();
+ }
}
}
diff --git a/src/common/thread_util.cc b/src/common/thread_util.cc
index b93d3fec..01afaa7d 100644
--- a/src/common/thread_util.cc
+++ b/src/common/thread_util.cc
@@ -20,6 +20,7 @@
#include "thread_util.h"
+#include <fmt/std.h>
#include <pthread.h>
namespace Util {
@@ -32,4 +33,19 @@ void ThreadSetName(const char *name) {
#endif
}
+template <void (std::thread::*F)(), typename... Args>
+Status ThreadOperationImpl(std::thread &t, const char *op, Args &&...args) {
+ try {
+ (t.*F)(std::forward<Args>(args)...);
+ } catch (const std::system_error &e) {
+ return {Status::NotOK, fmt::format("thread #{} cannot be `{}`ed: {}", t.get_id(), op, e.what())};
+ }
+
+ return Status::OK();
+}
+
+Status ThreadJoin(std::thread &t) { return ThreadOperationImpl<&std::thread::join>(t, "join"); }
+
+Status ThreadDetach(std::thread &t) { return ThreadOperationImpl<&std::thread::detach>(t, "detach"); }
+
} // namespace Util
diff --git a/src/common/thread_util.h b/src/common/thread_util.h
index f4ffc789..937ed963 100644
--- a/src/common/thread_util.h
+++ b/src/common/thread_util.h
@@ -20,8 +20,29 @@
#pragma once
+#include <system_error>
+#include <thread>
+
+#include "fmt/core.h"
+#include "status.h"
+
namespace Util {
void ThreadSetName(const char *name);
+template <typename F>
+StatusOr<std::thread> CreateThread(const char *name, F f) {
+ try {
+ return std::thread([name, f = std::move(f)] {
+ ThreadSetName(name);
+ f();
+ });
+ } catch (const std::system_error &e) {
+ return {Status::NotOK, fmt::format("thread '{}' cannot be started: {}", name, e.what())};
+ }
+}
+
+Status ThreadJoin(std::thread &t);
+Status ThreadDetach(std::thread &t);
+
} // namespace Util
diff --git a/src/server/server.cc b/src/server/server.cc
index e02005a3..8e4da917 100644
--- a/src/server/server.cc
+++ b/src/server/server.cc
@@ -165,15 +165,11 @@ Status Server::Start() {
task_runner_.Start();
// setup server cron thread
- cron_thread_ = std::thread([this]() {
- Util::ThreadSetName("server-cron");
- this->cron();
- });
+ cron_thread_ = GET_OR_RET(Util::CreateThread("server-cron", [this] { this->cron(); }));
- compaction_checker_thread_ = std::thread([this]() {
+ compaction_checker_thread_ = GET_OR_RET(Util::CreateThread("compact-check", [this] {
uint64_t counter = 0;
time_t last_compact_date = 0;
- Util::ThreadSetName("compact-check");
CompactionChecker compaction_checker(this->storage_);
while (!stop_) {
@@ -204,7 +200,7 @@ Status Server::Start() {
}
}
}
- });
+ }));
memory_startup_use_.store(Stats::GetMemoryRSS(), std::memory_order_relaxed);
LOG(INFO) << "[server] Ready to accept connections";
@@ -233,8 +229,12 @@ void Server::Join() {
}
task_runner_.Join();
- if (cron_thread_.joinable()) cron_thread_.join();
- if (compaction_checker_thread_.joinable()) compaction_checker_thread_.join();
+ if (auto s = Util::ThreadJoin(cron_thread_); !s) {
+ LOG(WARNING) << "Cron thread operation failed: " << s.Msg();
+ }
+ if (auto s = Util::ThreadJoin(compaction_checker_thread_); !s) {
+ LOG(WARNING) << "Compaction checker thread operation failed: " << s.Msg();
+ }
}
Status Server::AddMaster(const std::string &host, uint32_t port, bool force_reconnect) {
diff --git a/src/server/worker.cc b/src/server/worker.cc
index e4f70eb7..22ccaaea 100644
--- a/src/server/worker.cc
+++ b/src/server/worker.cc
@@ -518,20 +518,22 @@ void Worker::KickoutIdleClients(int timeout) {
}
void WorkerThread::Start() {
- try {
- t_ = std::thread([this]() {
- Util::ThreadSetName("worker");
- this->worker_->Run(std::this_thread::get_id());
- });
- } catch (const std::system_error &e) {
- LOG(ERROR) << "[worker] Failed to start worker thread, err: " << e.what();
+ auto s = Util::CreateThread("worker", [this] { this->worker_->Run(std::this_thread::get_id()); });
+
+ if (s) {
+ t_ = std::move(*s);
+ } else {
+ LOG(ERROR) << "[worker] Failed to start worker thread, err: " << s.Msg();
return;
}
+
LOG(INFO) << "[worker] Thread #" << t_.get_id() << " started";
}
void WorkerThread::Stop() { worker_->Stop(); }
void WorkerThread::Join() {
- if (t_.joinable()) t_.join();
+ if (auto s = Util::ThreadJoin(t_); !s) {
+ LOG(WARNING) << "[worker] " << s.Msg();
+ }
}