You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@kvrocks.apache.org by "LindaSummer (via GitHub)" <gi...@apache.org> on 2024/08/01 13:57:16 UTC
[PR] feat: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
LindaSummer opened a new pull request, #2461:
URL: https://github.com/apache/kvrocks/pull/2461
# Issue
#1389
# Proposed Changes
- refactor worker's connection structure from `std:::map` to `tbb::concurrent_hash_map`
- remove redundant mutex
# Comment
During my testing and researching on tbb, I find that the `tbb::concurrent_hash_map` is not thread safe on traversing and erasing.
Here is the related issue: https://github.com/oneapi-src/oneTBB/issues/183.
So, I have to keep the lock on iteration and erasing.
I have an idea that replace the erasing action to assigning a `nullptr` and cleanup the item during iteration.
Then we only need to lock the traversing.
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2294756243
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461) **Quality Gate passed**
Issues
 [13 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2461&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2461&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=2461&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [62.6% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2461&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2461&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461)
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "PragmaTwice (via GitHub)" <gi...@apache.org>.
PragmaTwice commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1709624742
##########
src/server/worker.cc:
##########
@@ -76,17 +78,29 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ auto collect_conns_fn = [](ConnMap &conns) {
+ return parallel_reduce(
Review Comment:
Is `parallel_reduce` necessary 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] feat: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2274798786
@mapleFU This PR generally looks good to me.
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1711633382
##########
src/server/worker.cc:
##########
@@ -409,31 +410,27 @@ void Worker::FreeConnection(redis::Connection *conn) {
}
void Worker::FreeConnectionByID(int fd, uint64_t id) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end() && iter->second->GetID() == id) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
if (rate_limit_group_ != nullptr) {
- bufferevent_remove_from_rate_limit_group(iter->second->GetBufferEvent());
+ bufferevent_remove_from_rate_limit_group(accessor->second->GetBufferEvent());
}
- delete iter->second;
- conns_.erase(iter);
+
+ delete accessor->second;
+ conns_.erase(accessor);
+
srv->DecrClientNum();
}
-
- iter = monitor_conns_.find(fd);
- if (iter != monitor_conns_.end() && iter->second->GetID() == id) {
- delete iter->second;
- monitor_conns_.erase(iter);
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, fd)) {
+ delete accessor->second;
Review Comment:
Hi @mapleFU ,
Erasing and deleting action has been reordered. 😊
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2286965422
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461) **Quality Gate passed**
Issues
 [15 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2461&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2461&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=2461&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [54.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2461&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2461&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461)
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2275934283
I don't have time for this today, I'll take a round tomorrow
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2277149643
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461) **Quality Gate failed**
Failed conditions
 [D Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1707235409
##########
src/server/worker.cc:
##########
@@ -333,18 +338,18 @@ Status Worker::AddConnection(redis::Connection *c) {
redis::Connection *Worker::removeConnection(int fd) {
redis::Connection *conn = nullptr;
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- conn = iter->second;
- conns_.erase(iter);
+ std::lock_guard<std::mutex> guard(conns_mu_);
Review Comment:
Hi @mapleFU ,
Sorry for delay reply.
I'm not so familiar with redis-benchmark.
I use the default redis benchmark command and generate two csv reports here.
It seems that the performance just has a slight change.
Here is my computer basic info.
```
OS: Rocky Linux 9.1 (Blue Onyx) x86_64
Kernel: 5.14.0-162.6.1.el9_1.0.1.x86_64
CPU: AMD Ryzen 5 1600 (12) @ 3.200GHz
Memory: 9970MiB / 31785MiB
```
This is the unstable branch's report with commit 6473d6c.
```csv
"test","rps","avg_latency_ms","min_latency_ms","p50_latency_ms","p95_latency_ms","p99_latency_ms","max_latency_ms"
"PING_INLINE","64808.82","0.404","0.072","0.391","0.511","0.671","3.647"
"PING_MBULK","65402.22","0.400","0.096","0.391","0.503","0.567","3.191"
"SET","64391.50","0.418","0.080","0.407","0.527","0.607","1.647"
"GET","64641.24","0.407","0.080","0.399","0.503","0.583","0.999"
"INCR","62500.00","0.559","0.056","0.503","0.951","1.215","2.127"
"LPUSH","56980.06","0.846","0.048","0.855","1.375","1.591","2.135"
"RPUSH","58685.45","0.801","0.048","0.799","1.199","1.391","2.511"
"LPOP","51572.98","0.939","0.048","0.967","1.479","1.783","3.247"
"RPOP","50838.84","0.960","0.048","0.927","1.607","1.967","3.743"
"SADD","62421.97","0.521","0.056","0.471","0.911","1.575","2.199"
"HSET","40064.10","1.229","0.056","0.999","2.655","3.023","3.999"
"SPOP","65146.58","0.403","0.088","0.399","0.511","0.567","0.903"
"ZADD","64061.50","0.627","0.040","0.575","1.111","1.279","1.775"
"ZPOPMIN","66225.16","0.398","0.080","0.391","0.511","0.599","1.031"
"LPUSH (needed to benchmark LRANGE)","54436.58","0.893","0.048","0.839","1.575","1.815","2.687"
"LRANGE_100 (first 100 elements)","40404.04","0.680","0.256","0.687","0.823","0.879","1.343"
"LRANGE_300 (first 300 elements)","19872.81","1.377","0.432","1.367","1.735","1.935","5.535"
"LRANGE_500 (first 500 elements)","13666.80","2.029","0.664","2.023","2.535","2.847","5.327"
"LRANGE_600 (first 600 elements)","12012.01","2.336","1.072","2.159","3.471","3.847","6.159"
"MSET (10 keys)","46772.68","1.030","0.048","1.007","2.047","2.519","18.703"
```
This is current tbb feature branch's report.
```csv
"test","rps","avg_latency_ms","min_latency_ms","p50_latency_ms","p95_latency_ms","p99_latency_ms","max_latency_ms"
"PING_INLINE","66269.05","0.395","0.072","0.391","0.479","0.551","3.055"
"PING_MBULK","65616.80","0.398","0.080","0.391","0.487","0.575","1.015"
"SET","61538.46","0.436","0.064","0.431","0.535","0.591","1.343"
"GET","65274.15","0.402","0.088","0.399","0.511","0.567","0.839"
"INCR","62695.92","0.572","0.056","0.503","1.039","1.295","2.103"
"LPUSH","58719.91","0.810","0.048","0.767","1.399","1.639","3.903"
"RPUSH","57703.40","0.829","0.048","0.879","1.415","1.655","3.031"
"LPOP","51177.07","0.952","0.048","0.951","1.615","1.927","3.175"
"RPOP","49776.01","0.981","0.048","0.999","1.839","2.223","3.391"
"SADD","63897.76","0.572","0.040","0.503","1.007","1.215","2.103"
"HSET","40866.37","1.205","0.056","1.223","1.999","2.287","3.295"
"SPOP","63171.20","0.416","0.088","0.407","0.535","0.639","1.175"
"ZADD","65274.15","0.685","0.048","0.679","1.103","1.263","1.975"
"ZPOPMIN","63451.78","0.414","0.064","0.415","0.511","0.575","0.927"
"LPUSH (needed to benchmark LRANGE)","54436.58","0.883","0.048","0.863","1.559","1.831","2.591"
"LRANGE_100 (first 100 elements)","38358.27","0.710","0.280","0.719","0.847","0.895","1.431"
"LRANGE_300 (first 300 elements)","17927.57","1.496","0.696","1.471","1.887","2.039","2.951"
"LRANGE_500 (first 500 elements)","12485.95","2.157","1.056","2.183","2.615","2.959","6.047"
"LRANGE_600 (first 600 elements)","11053.39","2.435","0.784","2.423","3.015","3.375","7.471"
"MSET (10 keys)","47505.94","1.030","0.056","1.087","1.567","1.783","2.719"
```
Here is the delta report of tbb minus unstable.
```csv
test,rps_diff,avg_latency_ms_diff,min_latency_ms_diff,p50_latency_ms_diff,p95_latency_ms_diff,p99_latency_ms_diff,max_latency_ms_diff
PING_INLINE,1460.23,-0.009,0.0,0.0,-0.032,-0.12,-0.592
PING_MBULK,214.58,-0.002,-0.016,0.0,-0.016,0.008,-2.176
SET,-2853.04,0.018,-0.016,0.024,0.008,-0.016,-0.304
GET,632.91,-0.005,0.008,0.0,0.008,-0.016,-0.16
INCR,195.92,0.013,0.0,0.0,0.088,0.08,-0.024
LPUSH,1739.85,-0.036,0.0,-0.088,0.024,0.048,1.768
RPUSH,-982.05,0.028,0.0,0.08,0.216,0.264,0.52
LPOP,-395.91,0.013,0.0,-0.016,0.136,0.144,-0.072
RPOP,-1062.83,0.021,0.0,0.072,0.232,0.256,-0.352
SADD,1475.79,0.051,-0.016,0.032,0.096,-0.36,-0.096
HSET,802.27,-0.024,0.0,0.224,-0.656,-0.736,-0.704
SPOP,-1975.38,0.013,0.0,0.008,0.024,0.072,0.272
ZADD,1212.65,0.058,0.008,0.104,-0.008,-0.016,0.2
ZPOPMIN,-2773.38,0.016,-0.016,0.024,0.0,-0.024,-0.104
LPUSH (needed to benchmark LRANGE),0.0,-0.01,0.0,0.024,-0.016,0.016,-0.096
LRANGE_100 (first 100 elements),-2045.77,0.03,0.024,0.032,0.024,0.016,0.088
LRANGE_300 (first 300 elements),-1945.24,0.119,0.264,0.104,0.152,0.104,-2.584
LRANGE_500 (first 500 elements),-1180.85,0.128,0.392,0.16,0.08,0.112,0.72
LRANGE_600 (first 600 elements),-958.62,0.099,-0.288,0.264,-0.456,-0.472,1.312
MSET (10 keys),733.26,0.0,0.008,0.08,-0.48,-0.736,-15.984
```
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1712996583
##########
src/server/worker.cc:
##########
@@ -38,6 +38,8 @@
#include <openssl/ssl.h>
#endif
+#include <oneapi/tbb/parallel_for.h>
+#include <oneapi/tbb/parallel_reduce.h>
Review Comment:
Hi @mapleFU ,
Thanks for your suggestions!
I find the formatter will automatically remove the blank line.
So, I make the tbb headers close to above glog headers.
Best Regards,
Edward
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1703265930
##########
src/server/worker.cc:
##########
@@ -76,17 +77,22 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ {
+ std::lock_guard<std::mutex> guard(conns_mu_);
+ for (const auto &iter : conns_) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, iter.first)) {
+ accessor->second->Close();
Review Comment:
Hi @mapleFU ,
Sorry for delay reply.
Thanks for your patience and suggestions!
I have updated code with `tbb::parallel_reduce` to collect the connections and close them out of tbb loop.
Best Regards,
Edward
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701977432
##########
src/server/worker.cc:
##########
@@ -333,18 +338,18 @@ Status Worker::AddConnection(redis::Connection *c) {
redis::Connection *Worker::removeConnection(int fd) {
redis::Connection *conn = nullptr;
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- conn = iter->second;
- conns_.erase(iter);
+ std::lock_guard<std::mutex> guard(conns_mu_);
Review Comment:
What brings up my attention is, if we really wants to do like this, can the critical section looks like:
```
if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
{
conn = accessor->second;
std::lock_guard<std::mutex> guard(conns_mu_);
conns_.erase(accessor);
}
```
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1710645447
##########
src/server/worker.cc:
##########
@@ -76,17 +78,29 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ auto collect_conns_fn = [](ConnMap &conns) {
+ return parallel_reduce(
Review Comment:
Hi @PragmaTwice ,
I have removed the `parallel_reduce` from dtor and all tests are passed in my repo's GitHub Actions.
I think in dtor the concurrency protection of internal state may be redundant since all references should be released except the dtor itself.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1707247473
##########
src/server/worker.cc:
##########
@@ -442,115 +461,146 @@ Status Worker::EnableWriteEvent(int fd) {
}
Status Worker::Reply(int fd, const std::string &reply) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- iter->second->SetLastInteraction();
- redis::Reply(iter->second->Output(), reply);
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
+ accessor->second->SetLastInteraction();
+ redis::Reply(accessor->second->Output(), reply);
return Status::OK();
}
return {Status::NotOK, "connection doesn't exist"};
}
void Worker::BecomeMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- conns_.erase(conn->GetFD());
- monitor_conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
+ conns_.erase(accessor);
+ accessor.release();
+
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, conn->GetFD())) {
+ accessor->second = conn;
+ } else {
+ monitor_conns_.insert(accessor, std::make_pair(conn->GetFD(), conn));
+ }
}
srv->IncrMonitorClientNum();
conn->EnableFlag(redis::Connection::kMonitor);
}
void Worker::QuitMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- monitor_conns_.erase(conn->GetFD());
- conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, conn->GetFD())) {
+ {
+ monitor_conns_.erase(accessor);
+ accessor.release();
+ }
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
+ accessor->second = conn;
+ } else {
+ conns_.insert(accessor, std::make_pair(conn->GetFD(), conn));
+ }
}
srv->DecrMonitorClientNum();
conn->DisableFlag(redis::Connection::kMonitor);
}
void Worker::FeedMonitorConns(redis::Connection *conn, const std::string &response) {
- std::unique_lock<std::mutex> lock(conns_mu_);
-
- for (const auto &iter : monitor_conns_) {
- if (conn == iter.second) continue; // skip the monitor command
-
- if (conn->GetNamespace() == iter.second->GetNamespace() || iter.second->GetNamespace() == kDefaultNamespace) {
- iter.second->Reply(response);
+ tbb::parallel_for(monitor_conns_.range(), [conn, response](const ConnMap::range_type &range) {
+ for (auto &it : range) {
+ const auto &value = it.second;
+ if (conn == value) continue; // skip the monitor command
+ if (conn->GetNamespace() == value->GetNamespace() || value->GetNamespace() == kDefaultNamespace) {
+ value->Reply(response);
+ }
}
- }
+ });
}
std::string Worker::GetClientsStr() {
- std::unique_lock<std::mutex> lock(conns_mu_);
-
- std::string output;
- for (const auto &iter : conns_) {
- redis::Connection *conn = iter.second;
- output.append(conn->ToString());
- }
-
- return output;
+ return tbb::parallel_reduce(
+ conns_.range(), std::string{},
+ [](const ConnMap::range_type &range, std::string &&result) {
+ for (auto &it : range) {
+ result.append(it.second->ToString());
+ }
+ return result;
+ },
+ [](const std::string &lhs, const std::string &rhs) {
+ std::string result = lhs;
+ result.append(rhs);
+ return result;
+ });
}
void Worker::KillClient(redis::Connection *self, uint64_t id, const std::string &addr, uint64_t type, bool skipme,
int64_t *killed) {
- std::lock_guard<std::mutex> guard(conns_mu_);
-
- for (const auto &iter : conns_) {
- redis::Connection *conn = iter.second;
- if (skipme && self == conn) continue;
-
- // no need to kill the client again if the kCloseAfterReply flag is set
- if (conn->IsFlagEnabled(redis::Connection::kCloseAfterReply)) {
- continue;
- }
-
- if ((type & conn->GetClientType()) ||
- (!addr.empty() && (conn->GetAddr() == addr || conn->GetAnnounceAddr() == addr)) ||
- (id != 0 && conn->GetID() == id)) {
- conn->EnableFlag(redis::Connection::kCloseAfterReply);
- // enable write event to notify worker wake up ASAP, and remove the connection
- if (!conn->IsFlagEnabled(redis::Connection::kSlave)) { // don't enable any event in slave connection
- auto bev = conn->GetBufferEvent();
- bufferevent_enable(bev, EV_WRITE);
+ for (const auto key : getConnFds()) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, key)) {
+ auto conn = accessor->second;
+ if (skipme && self == conn) continue;
+
+ // no need to kill the client again if the kCloseAfterReply flag is set
+ if (conn->IsFlagEnabled(redis::Connection::kCloseAfterReply)) {
+ continue;
+ }
+ if ((type & conn->GetClientType()) ||
+ (!addr.empty() && (conn->GetAddr() == addr || conn->GetAnnounceAddr() == addr)) ||
+ (id != 0 && conn->GetID() == id)) {
+ conn->EnableFlag(redis::Connection::kCloseAfterReply);
+ // enable write event to notify worker wake up ASAP, and remove the connection
+ if (!conn->IsFlagEnabled(redis::Connection::kSlave)) { // don't enable any event in slave connection
+ auto bev = conn->GetBufferEvent();
+ bufferevent_enable(bev, EV_WRITE);
+ }
+ (*killed)++;
}
- (*killed)++;
}
}
}
void Worker::KickoutIdleClients(int timeout) {
std::vector<std::pair<int, uint64_t>> to_be_killed_conns;
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- if (conns_.empty()) {
- return;
- }
+ auto fd_list = getConnFds();
+ if (fd_list.empty()) {
+ return;
+ }
- int iterations = std::min(static_cast<int>(conns_.size()), 50);
- auto iter = conns_.upper_bound(last_iter_conn_fd_);
- while (iterations--) {
- if (iter == conns_.end()) iter = conns_.begin();
- if (static_cast<int>(iter->second->GetIdleTime()) >= timeout) {
- to_be_killed_conns.emplace_back(iter->first, iter->second->GetID());
- }
- iter++;
+ std::set<int> fds(fd_list.cbegin(), fd_list.cend());
+
+ int iterations = std::min(static_cast<int>(conns_.size()), 50);
+ auto iter = fds.upper_bound(last_iter_conn_fd_);
+ while (iterations--) {
+ if (iter == fds.end()) {
+ iter = fds.begin();
+ }
+ if (ConnMap::const_accessor accessor;
+ conns_.find(accessor, *iter) && static_cast<int>(accessor->second->GetIdleTime()) >= timeout) {
+ to_be_killed_conns.emplace_back(accessor->first, accessor->second->GetID());
}
- iter--;
- last_iter_conn_fd_ = iter->first;
+ iter++;
}
+ iter--;
+ last_iter_conn_fd_ = *iter;
for (const auto &conn : to_be_killed_conns) {
FreeConnectionByID(conn.first, conn.second);
}
}
+std::vector<int> Worker::getConnFds() const {
+ return tbb::parallel_reduce(
Review Comment:
Hi @mapleFU ,
Thanks for your review suggestion.
I read the tbb's [official document](https://oneapi-src.github.io/oneTBB/main/tbb_userguide/Guiding_Task_Scheduler_Execution.html), and find the schedule uses all cores on default.
Maybe we should set a limitation for tbb schedule.
In fact, I'm not sure do we really need tbb hashmap to replace current mutex now. 🤔
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1712955758
##########
src/server/worker.cc:
##########
@@ -38,6 +38,8 @@
#include <openssl/ssl.h>
#endif
+#include <oneapi/tbb/parallel_for.h>
+#include <oneapi/tbb/parallel_reduce.h>
Review Comment:
nit: \n between 3rd and sys
##########
src/server/worker.cc:
##########
@@ -442,115 +439,146 @@ Status Worker::EnableWriteEvent(int fd) {
}
Status Worker::Reply(int fd, const std::string &reply) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- iter->second->SetLastInteraction();
- redis::Reply(iter->second->Output(), reply);
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
+ accessor->second->SetLastInteraction();
+ redis::Reply(accessor->second->Output(), reply);
return Status::OK();
}
return {Status::NotOK, "connection doesn't exist"};
}
void Worker::BecomeMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- conns_.erase(conn->GetFD());
- monitor_conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
+ conns_.erase(accessor);
+ accessor.release();
+
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, conn->GetFD())) {
+ accessor->second = conn;
+ } else {
+ monitor_conns_.insert(accessor, std::make_pair(conn->GetFD(), conn));
+ }
}
srv->IncrMonitorClientNum();
conn->EnableFlag(redis::Connection::kMonitor);
}
void Worker::QuitMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- monitor_conns_.erase(conn->GetFD());
- conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, conn->GetFD())) {
+ {
+ monitor_conns_.erase(accessor);
+ accessor.release();
+ }
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
+ accessor->second = conn;
+ } else {
+ conns_.insert(accessor, std::make_pair(conn->GetFD(), conn));
+ }
}
srv->DecrMonitorClientNum();
conn->DisableFlag(redis::Connection::kMonitor);
}
void Worker::FeedMonitorConns(redis::Connection *conn, const std::string &response) {
- std::unique_lock<std::mutex> lock(conns_mu_);
-
- for (const auto &iter : monitor_conns_) {
- if (conn == iter.second) continue; // skip the monitor command
-
- if (conn->GetNamespace() == iter.second->GetNamespace() || iter.second->GetNamespace() == kDefaultNamespace) {
- iter.second->Reply(response);
+ tbb::parallel_for(monitor_conns_.range(), [conn, response](const ConnMap::range_type &range) {
+ for (auto &it : range) {
+ const auto &value = it.second;
+ if (conn == value) continue; // skip the monitor command
+ if (conn->GetNamespace() == value->GetNamespace() || value->GetNamespace() == kDefaultNamespace) {
+ value->Reply(response);
+ }
}
- }
+ });
}
std::string Worker::GetClientsStr() {
- std::unique_lock<std::mutex> lock(conns_mu_);
-
- std::string output;
- for (const auto &iter : conns_) {
- redis::Connection *conn = iter.second;
- output.append(conn->ToString());
- }
-
- return output;
+ return tbb::parallel_reduce(
+ conns_.range(), std::string{},
+ [](const ConnMap::range_type &range, std::string &&result) {
Review Comment:
Emm why `&&result`? Using `std::string result` here?
##########
src/server/worker.cc:
##########
@@ -442,115 +439,146 @@ Status Worker::EnableWriteEvent(int fd) {
}
Status Worker::Reply(int fd, const std::string &reply) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- iter->second->SetLastInteraction();
- redis::Reply(iter->second->Output(), reply);
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
+ accessor->second->SetLastInteraction();
+ redis::Reply(accessor->second->Output(), reply);
return Status::OK();
}
return {Status::NotOK, "connection doesn't exist"};
}
void Worker::BecomeMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- conns_.erase(conn->GetFD());
- monitor_conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
Review Comment:
does this change the previous syntax? Previous if `conn->GetFD()` not erase anything, the insert would still happens?
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1712995356
##########
src/server/worker.cc:
##########
@@ -409,31 +410,27 @@ void Worker::FreeConnection(redis::Connection *conn) {
}
void Worker::FreeConnectionByID(int fd, uint64_t id) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end() && iter->second->GetID() == id) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
if (rate_limit_group_ != nullptr) {
- bufferevent_remove_from_rate_limit_group(iter->second->GetBufferEvent());
+ bufferevent_remove_from_rate_limit_group(accessor->second->GetBufferEvent());
}
- delete iter->second;
- conns_.erase(iter);
+
+ delete accessor->second;
+ conns_.erase(accessor);
+
srv->DecrClientNum();
}
-
- iter = monitor_conns_.find(fd);
- if (iter != monitor_conns_.end() && iter->second->GetID() == id) {
- delete iter->second;
- monitor_conns_.erase(iter);
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, fd)) {
+ delete accessor->second;
Review Comment:
Hi @mapleFU ,
Sorry for delay reply.
I read the code of tbb and find code below.
The data element is held by [`my_node`](
https://github.com/oneapi-src/oneTBB/blob/v2021.13.0/include/oneapi/tbb/concurrent_hash_map.h#L826).
The `my_node` will be [`released`](https://github.com/oneapi-src/oneTBB/blob/v2021.13.0/include/oneapi/tbb/concurrent_hash_map.h#L1418) in `erase()`.
I think accessing the element before `erase` may be a safer way.
I will move this line to above.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2324910119
Hi @mapleFU ,
Sorry for delay reply.
I have added constraints for `parallel_for` and `parallel_reduce`.
It should now only use one thread for execution.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1709665386
##########
src/server/worker.cc:
##########
@@ -76,17 +78,29 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ auto collect_conns_fn = [](ConnMap &conns) {
+ return parallel_reduce(
Review Comment:
Hi @PragmaTwice ,
I use a reduce action before iteration because the `parallel_for` will acquire a range rather than one item.
So, I want to move the time costly action `Close()` outside of the `parallel_for`.
But since this function is inside the dtor, if the ownership is managed correctly, it should be the only function uses internal state of the `Worker` object.
I think the `parallel_reduce` can be replaced by `parallel_for` or even removed.
I will refactor it and run tests before pushing to current PR.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1709665386
##########
src/server/worker.cc:
##########
@@ -76,17 +78,29 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ auto collect_conns_fn = [](ConnMap &conns) {
+ return parallel_reduce(
Review Comment:
Hi @PragmaTwice ,
I use a reduce action before iteration because the `parallel_for` will acquire a range rather than one item.
So, I want to move the time costly action `Close()` outside of the `parallel_for`.
But since this function is inside the dtor, if the ownership is managed correctly, it should be the only function uses internal state of the `Worker` object.
I think the `parallel_reduce` can be replaced by `parallel_for`.
I will refactor it and run tests before pushing to current PR.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2282800014
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461) **Quality Gate failed**
Failed conditions
 [D Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2286762036
General LGTM, maybe I'd testing it this week, if the operations is just raw read/erase it would works better
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701972560
##########
src/server/worker.h:
##########
@@ -88,9 +91,10 @@ class Worker : EventCallbackBase<Worker>, EvconnlistenerBase<Worker> {
UniqueEvent timer_;
std::thread::id tid_;
std::vector<evconnlistener *> listen_events_;
- std::mutex conns_mu_;
- std::map<int, redis::Connection *> conns_;
- std::map<int, redis::Connection *> monitor_conns_;
+ mutable std::mutex conns_mu_; // refer to https://github.com/oneapi-src/oneTBB/issues/183, traverse and erase should
Review Comment:
Hi @mapleFU ,
Thanks for your suggestions and help! ❤
I will add the tbb's code reference in comment.
> Would you mind add // guarded by for the protected element?
Of course, I will add comments for resources protected by mutex in next patches.
> would a extra lock and synchronization break the potential performance speedup and causing the performance worse than before?
Yes, during my development, I also concern about the performance of using both `tbb::concurrent_hash_map` and `std::mutex`. Maybe a simple read-write shared_mutex have a better performance and cleaner logic.
Best Regards,
Edward
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2264563873
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461) **Quality Gate failed**
Failed conditions
 [47.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2461&metric=new_coverage&view=list) (required ≥ 50%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461)
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1712997341
##########
src/server/worker.cc:
##########
@@ -442,115 +439,146 @@ Status Worker::EnableWriteEvent(int fd) {
}
Status Worker::Reply(int fd, const std::string &reply) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- iter->second->SetLastInteraction();
- redis::Reply(iter->second->Output(), reply);
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
+ accessor->second->SetLastInteraction();
+ redis::Reply(accessor->second->Output(), reply);
return Status::OK();
}
return {Status::NotOK, "connection doesn't exist"};
}
void Worker::BecomeMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- conns_.erase(conn->GetFD());
- monitor_conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
+ conns_.erase(accessor);
+ accessor.release();
+
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, conn->GetFD())) {
+ accessor->second = conn;
+ } else {
+ monitor_conns_.insert(accessor, std::make_pair(conn->GetFD(), conn));
+ }
}
srv->IncrMonitorClientNum();
conn->EnableFlag(redis::Connection::kMonitor);
}
void Worker::QuitMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- monitor_conns_.erase(conn->GetFD());
- conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, conn->GetFD())) {
+ {
+ monitor_conns_.erase(accessor);
+ accessor.release();
+ }
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
+ accessor->second = conn;
+ } else {
+ conns_.insert(accessor, std::make_pair(conn->GetFD(), conn));
+ }
}
srv->DecrMonitorClientNum();
conn->DisableFlag(redis::Connection::kMonitor);
}
void Worker::FeedMonitorConns(redis::Connection *conn, const std::string &response) {
- std::unique_lock<std::mutex> lock(conns_mu_);
-
- for (const auto &iter : monitor_conns_) {
- if (conn == iter.second) continue; // skip the monitor command
-
- if (conn->GetNamespace() == iter.second->GetNamespace() || iter.second->GetNamespace() == kDefaultNamespace) {
- iter.second->Reply(response);
+ tbb::parallel_for(monitor_conns_.range(), [conn, response](const ConnMap::range_type &range) {
+ for (auto &it : range) {
+ const auto &value = it.second;
+ if (conn == value) continue; // skip the monitor command
+ if (conn->GetNamespace() == value->GetNamespace() || value->GetNamespace() == kDefaultNamespace) {
+ value->Reply(response);
+ }
}
- }
+ });
}
std::string Worker::GetClientsStr() {
- std::unique_lock<std::mutex> lock(conns_mu_);
-
- std::string output;
- for (const auto &iter : conns_) {
- redis::Connection *conn = iter.second;
- output.append(conn->ToString());
- }
-
- return output;
+ return tbb::parallel_reduce(
+ conns_.range(), std::string{},
+ [](const ConnMap::range_type &range, std::string &&result) {
Review Comment:
Hi @mapleFU ,
Previously I thought the accumulation result should be a temporary value, so I used this signature.
But this right value reference is not necessary here.
If it's a returned temporary value, the RVO should work to help optimize.
I have updated it with a normal argument copy type.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1741016859
##########
src/server/worker.cc:
##########
@@ -442,115 +441,144 @@ Status Worker::EnableWriteEvent(int fd) {
}
Status Worker::Reply(int fd, const std::string &reply) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- iter->second->SetLastInteraction();
- redis::Reply(iter->second->Output(), reply);
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
+ accessor->second->SetLastInteraction();
+ redis::Reply(accessor->second->Output(), reply);
return Status::OK();
}
return {Status::NotOK, "connection doesn't exist"};
}
void Worker::BecomeMonitorConn(redis::Connection *conn) {
Review Comment:
Hi @mapleFU ,
Sorry for delay reply.
I'd like to use this way on all accessors to mitigate potential inconsistency in `monitor_conns_` and `conns_`.
```suggestion
void Worker::BecomeMonitorConn(redis::Connection *conn) {
ConnMap::accessor accessor;
ConnMap::accessor monitor_accessor;
bool find_conn = conns_.find(accessor, conn->GetFD());
bool find_monitor = monitor_conns_.find(monitor_accessor, conn->GetFD());
if (find_conn) {
conns_.erase(accessor);
}
if (find_monitor) {
monitor_accessor->second = conn;
} else {
monitor_conns_.insert(monitor_accessor, std::make_pair(conn->GetFD(), conn));
}
srv->IncrMonitorClientNum();
conn->EnableFlag(redis::Connection::kMonitor);
}
```
Best Regards,
Edward
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701943022
##########
src/server/worker.cc:
##########
@@ -76,17 +77,22 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ {
+ std::lock_guard<std::mutex> guard(conns_mu_);
+ for (const auto &iter : conns_) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, iter.first)) {
+ accessor->second->Close();
Review Comment:
I don't know whether close the connection in the critical section is really a good idea...
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1703267223
##########
src/server/worker.cc:
##########
@@ -564,6 +584,17 @@ void WorkerThread::Start() {
LOG(INFO) << "[worker] Thread #" << t_.get_id() << " started";
}
+std::map<int, redis::Connection *> Worker::GetConnections() const {
+ std::unique_lock<std::mutex> guard(conns_mu_);
+ std::map<int, redis::Connection *> result;
+ for (auto [fd, _] : conns_) {
Review Comment:
Hi @mapleFU ,
Yes, as you commented above, tbb doesn't guarantee the concurrency safe of STL style loop.
I have replaced it with `tbb::parallel_for` and `range()` method.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1704282111
##########
src/server/worker.cc:
##########
@@ -442,115 +461,146 @@ Status Worker::EnableWriteEvent(int fd) {
}
Status Worker::Reply(int fd, const std::string &reply) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- iter->second->SetLastInteraction();
- redis::Reply(iter->second->Output(), reply);
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
+ accessor->second->SetLastInteraction();
+ redis::Reply(accessor->second->Output(), reply);
return Status::OK();
}
return {Status::NotOK, "connection doesn't exist"};
}
void Worker::BecomeMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- conns_.erase(conn->GetFD());
- monitor_conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
+ conns_.erase(accessor);
+ accessor.release();
+
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, conn->GetFD())) {
+ accessor->second = conn;
+ } else {
+ monitor_conns_.insert(accessor, std::make_pair(conn->GetFD(), conn));
+ }
}
srv->IncrMonitorClientNum();
conn->EnableFlag(redis::Connection::kMonitor);
}
void Worker::QuitMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- monitor_conns_.erase(conn->GetFD());
- conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, conn->GetFD())) {
+ {
+ monitor_conns_.erase(accessor);
+ accessor.release();
+ }
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
+ accessor->second = conn;
+ } else {
+ conns_.insert(accessor, std::make_pair(conn->GetFD(), conn));
+ }
}
srv->DecrMonitorClientNum();
conn->DisableFlag(redis::Connection::kMonitor);
}
void Worker::FeedMonitorConns(redis::Connection *conn, const std::string &response) {
- std::unique_lock<std::mutex> lock(conns_mu_);
-
- for (const auto &iter : monitor_conns_) {
- if (conn == iter.second) continue; // skip the monitor command
-
- if (conn->GetNamespace() == iter.second->GetNamespace() || iter.second->GetNamespace() == kDefaultNamespace) {
- iter.second->Reply(response);
+ tbb::parallel_for(monitor_conns_.range(), [conn, response](const ConnMap::range_type &range) {
+ for (auto &it : range) {
+ const auto &value = it.second;
+ if (conn == value) continue; // skip the monitor command
+ if (conn->GetNamespace() == value->GetNamespace() || value->GetNamespace() == kDefaultNamespace) {
+ value->Reply(response);
+ }
}
- }
+ });
}
std::string Worker::GetClientsStr() {
- std::unique_lock<std::mutex> lock(conns_mu_);
-
- std::string output;
- for (const auto &iter : conns_) {
- redis::Connection *conn = iter.second;
- output.append(conn->ToString());
- }
-
- return output;
+ return tbb::parallel_reduce(
+ conns_.range(), std::string{},
+ [](const ConnMap::range_type &range, std::string &&result) {
+ for (auto &it : range) {
+ result.append(it.second->ToString());
+ }
+ return result;
+ },
+ [](const std::string &lhs, const std::string &rhs) {
+ std::string result = lhs;
+ result.append(rhs);
+ return result;
+ });
}
void Worker::KillClient(redis::Connection *self, uint64_t id, const std::string &addr, uint64_t type, bool skipme,
int64_t *killed) {
- std::lock_guard<std::mutex> guard(conns_mu_);
-
- for (const auto &iter : conns_) {
- redis::Connection *conn = iter.second;
- if (skipme && self == conn) continue;
-
- // no need to kill the client again if the kCloseAfterReply flag is set
- if (conn->IsFlagEnabled(redis::Connection::kCloseAfterReply)) {
- continue;
- }
-
- if ((type & conn->GetClientType()) ||
- (!addr.empty() && (conn->GetAddr() == addr || conn->GetAnnounceAddr() == addr)) ||
- (id != 0 && conn->GetID() == id)) {
- conn->EnableFlag(redis::Connection::kCloseAfterReply);
- // enable write event to notify worker wake up ASAP, and remove the connection
- if (!conn->IsFlagEnabled(redis::Connection::kSlave)) { // don't enable any event in slave connection
- auto bev = conn->GetBufferEvent();
- bufferevent_enable(bev, EV_WRITE);
+ for (const auto key : getConnFds()) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, key)) {
+ auto conn = accessor->second;
+ if (skipme && self == conn) continue;
+
+ // no need to kill the client again if the kCloseAfterReply flag is set
+ if (conn->IsFlagEnabled(redis::Connection::kCloseAfterReply)) {
+ continue;
+ }
+ if ((type & conn->GetClientType()) ||
+ (!addr.empty() && (conn->GetAddr() == addr || conn->GetAnnounceAddr() == addr)) ||
+ (id != 0 && conn->GetID() == id)) {
+ conn->EnableFlag(redis::Connection::kCloseAfterReply);
+ // enable write event to notify worker wake up ASAP, and remove the connection
+ if (!conn->IsFlagEnabled(redis::Connection::kSlave)) { // don't enable any event in slave connection
+ auto bev = conn->GetBufferEvent();
+ bufferevent_enable(bev, EV_WRITE);
+ }
+ (*killed)++;
}
- (*killed)++;
}
}
}
void Worker::KickoutIdleClients(int timeout) {
std::vector<std::pair<int, uint64_t>> to_be_killed_conns;
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- if (conns_.empty()) {
- return;
- }
+ auto fd_list = getConnFds();
+ if (fd_list.empty()) {
+ return;
+ }
- int iterations = std::min(static_cast<int>(conns_.size()), 50);
- auto iter = conns_.upper_bound(last_iter_conn_fd_);
- while (iterations--) {
- if (iter == conns_.end()) iter = conns_.begin();
- if (static_cast<int>(iter->second->GetIdleTime()) >= timeout) {
- to_be_killed_conns.emplace_back(iter->first, iter->second->GetID());
- }
- iter++;
+ std::set<int> fds(fd_list.cbegin(), fd_list.cend());
+
+ int iterations = std::min(static_cast<int>(conns_.size()), 50);
+ auto iter = fds.upper_bound(last_iter_conn_fd_);
+ while (iterations--) {
+ if (iter == fds.end()) {
+ iter = fds.begin();
+ }
+ if (ConnMap::const_accessor accessor;
+ conns_.find(accessor, *iter) && static_cast<int>(accessor->second->GetIdleTime()) >= timeout) {
+ to_be_killed_conns.emplace_back(accessor->first, accessor->second->GetID());
}
- iter--;
- last_iter_conn_fd_ = iter->first;
+ iter++;
}
+ iter--;
+ last_iter_conn_fd_ = *iter;
for (const auto &conn : to_be_killed_conns) {
FreeConnectionByID(conn.first, conn.second);
}
}
+std::vector<int> Worker::getConnFds() const {
+ return tbb::parallel_reduce(
Review Comment:
IMO parallel_for looks like some cpu-bound operations, and would it occupies more threads than expected 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] feat: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1710648034
##########
src/server/worker.cc:
##########
@@ -76,17 +78,29 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ auto collect_conns_fn = [](ConnMap &conns) {
+ return parallel_reduce(
Review Comment:
> I mean maybe a single-thread for is enough here? didn't have a deep look yet
Yeah parallel task also has a dop (degree of parallel), and I guess default of `parallel_for` would be high and be good at writing cpu-bound tasks like OpenMP
I think maybe lock is prefered 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] feat: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2290478183
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461) **Quality Gate passed**
Issues
 [27 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2461&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2461&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=2461&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [62.6% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2461&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2461&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461)
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701947296
##########
src/server/worker.h:
##########
@@ -88,9 +91,10 @@ class Worker : EventCallbackBase<Worker>, EvconnlistenerBase<Worker> {
UniqueEvent timer_;
std::thread::id tid_;
std::vector<evconnlistener *> listen_events_;
- std::mutex conns_mu_;
- std::map<int, redis::Connection *> conns_;
- std::map<int, redis::Connection *> monitor_conns_;
+ mutable std::mutex conns_mu_; // refer to https://github.com/oneapi-src/oneTBB/issues/183, traverse and erase should
Review Comment:
Besides, I'm still curious that what did the mutex protect after this patch. Since we use open-tbb now, the concurrent hash map would brings up much more cost ( on both memory usage and search time) than std::unordered_map, would a extra lock and synchronization break the potential performance speedup and causing the performance worse than before? Would you mind add `// guarded by` for the protected element?
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701965091
##########
src/server/worker.cc:
##########
@@ -564,6 +584,17 @@ void WorkerThread::Start() {
LOG(INFO) << "[worker] Thread #" << t_.get_id() << " started";
}
+std::map<int, redis::Connection *> Worker::GetConnections() const {
+ std::unique_lock<std::mutex> guard(conns_mu_);
+ std::map<int, redis::Connection *> result;
+ for (auto [fd, _] : conns_) {
Review Comment:
So it's because this traversing uses legacy style traverse, which bring up the problem. The iter would be invalidate when erase happens?
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701976227
##########
src/server/worker.cc:
##########
@@ -76,17 +77,22 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ {
+ std::lock_guard<std::mutex> guard(conns_mu_);
+ for (const auto &iter : conns_) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, iter.first)) {
+ accessor->second->Close();
Review Comment:
In dtor, we're able to just move out `conns_` with lock ( or even without lock ) seems other operations can merely concurrency with a 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] feat: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2268088864
> Would you mind try redis-benchmark to test some commands? I don't know how this would improve our performance, it would great to have some data
Hi @mapleFU ,
Thanks for your suggestion!
In fact I'm also not sure how much will tbb hashmap affect the performance. 😆
I will make a benchmark with redis benchmark later today.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1707235409
##########
src/server/worker.cc:
##########
@@ -333,18 +338,18 @@ Status Worker::AddConnection(redis::Connection *c) {
redis::Connection *Worker::removeConnection(int fd) {
redis::Connection *conn = nullptr;
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- conn = iter->second;
- conns_.erase(iter);
+ std::lock_guard<std::mutex> guard(conns_mu_);
Review Comment:
Hi @mapleFU ,
Sorry for delay reply.
I'm not so familiar with redis-benchmark.
I use the default redis benchmark command and generate two csv reports here.
It seems that the performance just has a slight change.
Here is my computer basic info.
```
OS: Rocky Linux 9.1 (Blue Onyx) x86_64
Kernel: 5.14.0-162.6.1.el9_1.0.1.x86_64
CPU: AMD Ryzen 5 1600 (12) @ 3.200GHz
Memory: 9970MiB / 31785MiB
```
This is the unstable branch's report with commit 6473d6c.
```csv
"test","rps","avg_latency_ms","min_latency_ms","p50_latency_ms","p95_latency_ms","p99_latency_ms","max_latency_ms"
"PING_INLINE","64808.82","0.404","0.072","0.391","0.511","0.671","3.647"
"PING_MBULK","65402.22","0.400","0.096","0.391","0.503","0.567","3.191"
"SET","64391.50","0.418","0.080","0.407","0.527","0.607","1.647"
"GET","64641.24","0.407","0.080","0.399","0.503","0.583","0.999"
"INCR","62500.00","0.559","0.056","0.503","0.951","1.215","2.127"
"LPUSH","56980.06","0.846","0.048","0.855","1.375","1.591","2.135"
"RPUSH","58685.45","0.801","0.048","0.799","1.199","1.391","2.511"
"LPOP","51572.98","0.939","0.048","0.967","1.479","1.783","3.247"
"RPOP","50838.84","0.960","0.048","0.927","1.607","1.967","3.743"
"SADD","62421.97","0.521","0.056","0.471","0.911","1.575","2.199"
"HSET","40064.10","1.229","0.056","0.999","2.655","3.023","3.999"
"SPOP","65146.58","0.403","0.088","0.399","0.511","0.567","0.903"
"ZADD","64061.50","0.627","0.040","0.575","1.111","1.279","1.775"
"ZPOPMIN","66225.16","0.398","0.080","0.391","0.511","0.599","1.031"
"LPUSH (needed to benchmark LRANGE)","54436.58","0.893","0.048","0.839","1.575","1.815","2.687"
"LRANGE_100 (first 100 elements)","40404.04","0.680","0.256","0.687","0.823","0.879","1.343"
"LRANGE_300 (first 300 elements)","19872.81","1.377","0.432","1.367","1.735","1.935","5.535"
"LRANGE_500 (first 500 elements)","13666.80","2.029","0.664","2.023","2.535","2.847","5.327"
"LRANGE_600 (first 600 elements)","12012.01","2.336","1.072","2.159","3.471","3.847","6.159"
"MSET (10 keys)","46772.68","1.030","0.048","1.007","2.047","2.519","18.703"
```
This is current tbb feature branch's report.
```csv
"test","rps","avg_latency_ms","min_latency_ms","p50_latency_ms","p95_latency_ms","p99_latency_ms","max_latency_ms"
"PING_INLINE","66269.05","0.395","0.072","0.391","0.479","0.551","3.055"
"PING_MBULK","65616.80","0.398","0.080","0.391","0.487","0.575","1.015"
"SET","61538.46","0.436","0.064","0.431","0.535","0.591","1.343"
"GET","65274.15","0.402","0.088","0.399","0.511","0.567","0.839"
"INCR","62695.92","0.572","0.056","0.503","1.039","1.295","2.103"
"LPUSH","58719.91","0.810","0.048","0.767","1.399","1.639","3.903"
"RPUSH","57703.40","0.829","0.048","0.879","1.415","1.655","3.031"
"LPOP","51177.07","0.952","0.048","0.951","1.615","1.927","3.175"
"RPOP","49776.01","0.981","0.048","0.999","1.839","2.223","3.391"
"SADD","63897.76","0.572","0.040","0.503","1.007","1.215","2.103"
"HSET","40866.37","1.205","0.056","1.223","1.999","2.287","3.295"
"SPOP","63171.20","0.416","0.088","0.407","0.535","0.639","1.175"
"ZADD","65274.15","0.685","0.048","0.679","1.103","1.263","1.975"
"ZPOPMIN","63451.78","0.414","0.064","0.415","0.511","0.575","0.927"
"LPUSH (needed to benchmark LRANGE)","54436.58","0.883","0.048","0.863","1.559","1.831","2.591"
"LRANGE_100 (first 100 elements)","38358.27","0.710","0.280","0.719","0.847","0.895","1.431"
"LRANGE_300 (first 300 elements)","17927.57","1.496","0.696","1.471","1.887","2.039","2.951"
"LRANGE_500 (first 500 elements)","12485.95","2.157","1.056","2.183","2.615","2.959","6.047"
"LRANGE_600 (first 600 elements)","11053.39","2.435","0.784","2.423","3.015","3.375","7.471"
"MSET (10 keys)","47505.94","1.030","0.056","1.087","1.567","1.783","2.719"
```
Here is the delta report of tbb minus unstable.
```csv
test,rps_diff,avg_latency_ms_diff,min_latency_ms_diff,p50_latency_ms_diff,p95_latency_ms_diff,p99_latency_ms_diff,max_latency_ms_diff
PING_INLINE,1460.23,-0.009,0.0,0.0,-0.032,-0.12,-0.592
PING_MBULK,214.58,-0.002,-0.016,0.0,-0.016,0.008,-2.176
SET,-2853.04,0.018,-0.016,0.024,0.008,-0.016,-0.304
GET,632.91,-0.005,0.008,0.0,0.008,-0.016,-0.16
INCR,195.92,0.013,0.0,0.0,0.088,0.08,-0.024
LPUSH,1739.85,-0.036,0.0,-0.088,0.024,0.048,1.768
RPUSH,-982.05,0.028,0.0,0.08,0.216,0.264,0.52
LPOP,-395.91,0.013,0.0,-0.016,0.136,0.144,-0.072
RPOP,-1062.83,0.021,0.0,0.072,0.232,0.256,-0.352
SADD,1475.79,0.051,-0.016,0.032,0.096,-0.36,-0.096
HSET,802.27,-0.024,0.0,0.224,-0.656,-0.736,-0.704
SPOP,-1975.38,0.013,0.0,0.008,0.024,0.072,0.272
ZADD,1212.65,0.058,0.008,0.104,-0.008,-0.016,0.2
ZPOPMIN,-2773.38,0.016,-0.016,0.024,0.0,-0.024,-0.104
LPUSH (needed to benchmark LRANGE),0.0,-0.01,0.0,0.024,-0.016,0.016,-0.096
LRANGE_100 (first 100 elements),-2045.77,0.03,0.024,0.032,0.024,0.016,0.088
LRANGE_300 (first 300 elements),-1945.24,0.119,0.264,0.104,0.152,0.104,-2.584
LRANGE_500 (first 500 elements),-1180.85,0.128,0.392,0.16,0.08,0.112,0.72
LRANGE_600 (first 600 elements),-958.62,0.099,-0.288,0.264,-0.456,-0.472,1.312
MSET (10 keys),733.26,0.0,0.008,0.08,-0.48,-0.736,-15.984
```
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2273719129
> Would you mind try redis-benchmark to test some commands? I don't know how this would improve our performance, it would great to have some data
Hi @mapleFU ,
Sorry for delay reply.
I'm not so familiar with redis-benchmark.
I use the default redis benchmark command and generate two csv reports here.
It seems that the performance just has a slight change.
Here is my computer basic info.
```
OS: Rocky Linux 9.1 (Blue Onyx) x86_64
Kernel: 5.14.0-162.6.1.el9_1.0.1.x86_64
CPU: AMD Ryzen 5 1600 (12) @ 3.200GHz
Memory: 9970MiB / 31785MiB
```
This is the unstable branch's report with commit 6473d6c.
```csv
"test","rps","avg_latency_ms","min_latency_ms","p50_latency_ms","p95_latency_ms","p99_latency_ms","max_latency_ms"
"PING_INLINE","64808.82","0.404","0.072","0.391","0.511","0.671","3.647"
"PING_MBULK","65402.22","0.400","0.096","0.391","0.503","0.567","3.191"
"SET","64391.50","0.418","0.080","0.407","0.527","0.607","1.647"
"GET","64641.24","0.407","0.080","0.399","0.503","0.583","0.999"
"INCR","62500.00","0.559","0.056","0.503","0.951","1.215","2.127"
"LPUSH","56980.06","0.846","0.048","0.855","1.375","1.591","2.135"
"RPUSH","58685.45","0.801","0.048","0.799","1.199","1.391","2.511"
"LPOP","51572.98","0.939","0.048","0.967","1.479","1.783","3.247"
"RPOP","50838.84","0.960","0.048","0.927","1.607","1.967","3.743"
"SADD","62421.97","0.521","0.056","0.471","0.911","1.575","2.199"
"HSET","40064.10","1.229","0.056","0.999","2.655","3.023","3.999"
"SPOP","65146.58","0.403","0.088","0.399","0.511","0.567","0.903"
"ZADD","64061.50","0.627","0.040","0.575","1.111","1.279","1.775"
"ZPOPMIN","66225.16","0.398","0.080","0.391","0.511","0.599","1.031"
"LPUSH (needed to benchmark LRANGE)","54436.58","0.893","0.048","0.839","1.575","1.815","2.687"
"LRANGE_100 (first 100 elements)","40404.04","0.680","0.256","0.687","0.823","0.879","1.343"
"LRANGE_300 (first 300 elements)","19872.81","1.377","0.432","1.367","1.735","1.935","5.535"
"LRANGE_500 (first 500 elements)","13666.80","2.029","0.664","2.023","2.535","2.847","5.327"
"LRANGE_600 (first 600 elements)","12012.01","2.336","1.072","2.159","3.471","3.847","6.159"
"MSET (10 keys)","46772.68","1.030","0.048","1.007","2.047","2.519","18.703"
```
This is current tbb feature branch's report.
```csv
"test","rps","avg_latency_ms","min_latency_ms","p50_latency_ms","p95_latency_ms","p99_latency_ms","max_latency_ms"
"PING_INLINE","66269.05","0.395","0.072","0.391","0.479","0.551","3.055"
"PING_MBULK","65616.80","0.398","0.080","0.391","0.487","0.575","1.015"
"SET","61538.46","0.436","0.064","0.431","0.535","0.591","1.343"
"GET","65274.15","0.402","0.088","0.399","0.511","0.567","0.839"
"INCR","62695.92","0.572","0.056","0.503","1.039","1.295","2.103"
"LPUSH","58719.91","0.810","0.048","0.767","1.399","1.639","3.903"
"RPUSH","57703.40","0.829","0.048","0.879","1.415","1.655","3.031"
"LPOP","51177.07","0.952","0.048","0.951","1.615","1.927","3.175"
"RPOP","49776.01","0.981","0.048","0.999","1.839","2.223","3.391"
"SADD","63897.76","0.572","0.040","0.503","1.007","1.215","2.103"
"HSET","40866.37","1.205","0.056","1.223","1.999","2.287","3.295"
"SPOP","63171.20","0.416","0.088","0.407","0.535","0.639","1.175"
"ZADD","65274.15","0.685","0.048","0.679","1.103","1.263","1.975"
"ZPOPMIN","63451.78","0.414","0.064","0.415","0.511","0.575","0.927"
"LPUSH (needed to benchmark LRANGE)","54436.58","0.883","0.048","0.863","1.559","1.831","2.591"
"LRANGE_100 (first 100 elements)","38358.27","0.710","0.280","0.719","0.847","0.895","1.431"
"LRANGE_300 (first 300 elements)","17927.57","1.496","0.696","1.471","1.887","2.039","2.951"
"LRANGE_500 (first 500 elements)","12485.95","2.157","1.056","2.183","2.615","2.959","6.047"
"LRANGE_600 (first 600 elements)","11053.39","2.435","0.784","2.423","3.015","3.375","7.471"
"MSET (10 keys)","47505.94","1.030","0.056","1.087","1.567","1.783","2.719"
```
Here is the delta report of tbb minus unstable.
```csv
test,rps_diff,avg_latency_ms_diff,min_latency_ms_diff,p50_latency_ms_diff,p95_latency_ms_diff,p99_latency_ms_diff,max_latency_ms_diff
PING_INLINE,1460.23,-0.009,0.0,0.0,-0.032,-0.12,-0.592
PING_MBULK,214.58,-0.002,-0.016,0.0,-0.016,0.008,-2.176
SET,-2853.04,0.018,-0.016,0.024,0.008,-0.016,-0.304
GET,632.91,-0.005,0.008,0.0,0.008,-0.016,-0.16
INCR,195.92,0.013,0.0,0.0,0.088,0.08,-0.024
LPUSH,1739.85,-0.036,0.0,-0.088,0.024,0.048,1.768
RPUSH,-982.05,0.028,0.0,0.08,0.216,0.264,0.52
LPOP,-395.91,0.013,0.0,-0.016,0.136,0.144,-0.072
RPOP,-1062.83,0.021,0.0,0.072,0.232,0.256,-0.352
SADD,1475.79,0.051,-0.016,0.032,0.096,-0.36,-0.096
HSET,802.27,-0.024,0.0,0.224,-0.656,-0.736,-0.704
SPOP,-1975.38,0.013,0.0,0.008,0.024,0.072,0.272
ZADD,1212.65,0.058,0.008,0.104,-0.008,-0.016,0.2
ZPOPMIN,-2773.38,0.016,-0.016,0.024,0.0,-0.024,-0.104
LPUSH (needed to benchmark LRANGE),0.0,-0.01,0.0,0.024,-0.016,0.016,-0.096
LRANGE_100 (first 100 elements),-2045.77,0.03,0.024,0.032,0.024,0.016,0.088
LRANGE_300 (first 300 elements),-1945.24,0.119,0.264,0.104,0.152,0.104,-2.584
LRANGE_500 (first 500 elements),-1180.85,0.128,0.392,0.16,0.08,0.112,0.72
LRANGE_600 (first 600 elements),-958.62,0.099,-0.288,0.264,-0.456,-0.472,1.312
MSET (10 keys),733.26,0.0,0.008,0.08,-0.48,-0.736,-15.984
```
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2274627265
> Thanks for your benchmarking! Maybe I'd like to run this and trying to perf the mutex contention when I have time.
>
> I've check out the benchmark result but the benchmark result turns to be unstable 🤔 The graph below shows that in different percentage result would be different...So maybe we should analysis the performance bottleneck for connections handling and check whether it could bring up performance improvement
>
> 
Hi @mapleFU ,
Thanks very much for your help! 😊
I will learn more about redis-benchamrk and try to use a larger test sample for report.
Best Regards,
Edward
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701612282
##########
src/server/worker.h:
##########
@@ -88,9 +91,10 @@ class Worker : EventCallbackBase<Worker>, EvconnlistenerBase<Worker> {
UniqueEvent timer_;
std::thread::id tid_;
std::vector<evconnlistener *> listen_events_;
- std::mutex conns_mu_;
- std::map<int, redis::Connection *> conns_;
- std::map<int, redis::Connection *> monitor_conns_;
+ mutable std::mutex conns_mu_; // refer to https://github.com/oneapi-src/oneTBB/issues/183, traverse and erase should
Review Comment:
Hi @mapleFU ,
Thanks very much for your review suggestions!
In fact I tried to get some official document after the TSAN issue emerged.
But It seems the tbb official document has no description on this behavior.
I finally found the problem in a [stackoverflow post](https://stackoverflow.com/questions/77673606/when-find-of-tbbconcurrent-hash-map-is-used-in-parallel-with-iteration-the).
This detail page was in intel's documentation site but now disappeared.
With the web archive, I got these two pages archive.
[Traversing concurrent_hash_map concurrently](https://web.archive.org/web/20200818130444/https://software.intel.com/content/www/us/en/develop/blogs/traversing-concurrent_hash_map-concurrently.html).
[concurrent_hash_map iterators](https://web.archive.org/web/20240429165704/https://spec.oneapi.io/versions/latest/elements/oneTBB/source/containers/concurrent_hash_map_cls/iterators.html)
The first archive has below sentences.
> Although it is not documented, you can traverse tbb::concurrent_hash_map concurrently if you are fine with the following restrictions and possible workarounds for them:
>
> You cannot traverse concurrently with an erase() operation. Use a mutex around erase and traversal code if you still need both running in different threads.
> - Some items can be missed due to concurrent rehashing. Try to reserve enough buckets before traversal using rehash() to prevent this or repeat the pass.
> - Some items can be visited twice or more times. Try to reserve enough buckets before traversal using rehash() to prevent this or put everything to a local serial map first to remove duplicates.
> - Access through iterators is not protected. Know what are you doing, e.g. use additional synchronization primitives like tbb::atomic to protect one member or acquire accessor to protect the rest.
In the second archive captured on 2024/04/29, there are some sentences on iteration.
> All member functions in this section can only be performed serially. The behavior is undefined in case of concurrent execution of these member functions with other (either concurrently safe) methods.
I don't know the exact reason of removing this page on current tbb documentation. 🤣
The latest official tbb release version and our dependency are same `oneapi-src/oneTBB v2021.13.0`.
It should not have any change on stable releases since the second archive published.
At least in my [GitHub actions](https://github.com/LindaSummer/kvrocks/actions/runs/10130746250/job/28017077204) this issue still exists.
Maybe we should create a new issue on tbb community for more official answers? 😄
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2268081444
Would you mind try redis-benchmark to test some commands? I don't know how this would improve our performance, it would great to have some data
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1703515575
##########
src/server/worker.cc:
##########
@@ -76,17 +78,40 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ auto conns = tbb::parallel_reduce(
+ conns_.range(), std::vector<redis::Connection *>{},
+ [](const ConnMap::range_type &range, std::vector<redis::Connection *> &&result) {
+ for (auto &it : range) {
+ result.push_back(it.second);
+ }
+ return result;
+ },
+ [](const std::vector<redis::Connection *> &lhs, const std::vector<redis::Connection *> &rhs) {
+ std::vector<redis::Connection *> result = lhs;
+ result.insert(result.end(), rhs.begin(), rhs.end());
+ return result;
+ });
+
+ auto monitor_conns = tbb::parallel_reduce(
+ monitor_conns_.range(), std::vector<redis::Connection *>{},
+ [](const ConnMap::range_type &range, std::vector<redis::Connection *> &&result) {
+ for (auto &it : range) {
+ result.push_back(it.second);
+ }
+ return result;
+ },
+ [](const std::vector<redis::Connection *> &lhs, const std::vector<redis::Connection *> &rhs) {
+ std::vector<redis::Connection *> result = lhs;
+ result.insert(result.end(), rhs.begin(), rhs.end());
+ return result;
+ });
Review Comment:
```suggestion
auto collect_conns_fn = [](ConnMap &conns) {
return parallel_reduce(
conns.range(), std::vector<redis::Connection *>{},
[](const ConnMap::range_type &range, std::vector<redis::Connection *> &&result) {
for (auto &it : range) {
result.push_back(it.second);
}
return result;
},
[](const std::vector<redis::Connection *> &lhs, const std::vector<redis::Connection *> &rhs) {
std::vector<redis::Connection *> result = lhs;
result.insert(result.end(), rhs.begin(), rhs.end());
return result;
});
};
auto conns = collect_conns_fn(conns_);
auto monitor_conns = collect_conns_fn(monitor_conns_);
```
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1703265043
##########
src/server/worker.h:
##########
@@ -88,9 +91,10 @@ class Worker : EventCallbackBase<Worker>, EvconnlistenerBase<Worker> {
UniqueEvent timer_;
std::thread::id tid_;
std::vector<evconnlistener *> listen_events_;
- std::mutex conns_mu_;
- std::map<int, redis::Connection *> conns_;
- std::map<int, redis::Connection *> monitor_conns_;
+ mutable std::mutex conns_mu_; // refer to https://github.com/oneapi-src/oneTBB/issues/183, traverse and erase should
Review Comment:
Hi @mapleFU ,
Thanks for your suggestions!
I have been inspired by your reference of [tbb source file](https://github.com/oneapi-src/oneTBB/blob/v2021.13.0/include/oneapi/tbb/concurrent_hash_map.h#L1033-L1051).
I switched from the cpp style for loop to `tbb::parallel_for` and `tbb_parallel_reduce`.
The `range()` method is commented as `Parallel algorithm support`.
It works now and the mutex has been removed. 😊
I read the source code of `tbb::parallel_recude` and `tbb::concurrent_hashmap::range()` and it seems that the partitioner will help [split the range](https://github.com/oneapi-src/oneTBB/blob/v2021.13.0/include/oneapi/tbb/partitioner.h#L273-L291) before execution.
Best Regards,
Edward
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1703268623
##########
src/server/worker.cc:
##########
@@ -333,18 +338,18 @@ Status Worker::AddConnection(redis::Connection *c) {
redis::Connection *Worker::removeConnection(int fd) {
redis::Connection *conn = nullptr;
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- conn = iter->second;
- conns_.erase(iter);
+ std::lock_guard<std::mutex> guard(conns_mu_);
Review Comment:
Hi @mapleFU ,
Thanks for your suggestions.
I agree on your view, if we need to protect the erase action, we should only lock the erase itself.
Lock the outside scope is an insufficient way.
I have removed the mutex now.
So, we have no more lock here and we all rely on tbb hashmap itself.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1711521700
##########
src/server/worker.cc:
##########
@@ -47,6 +47,8 @@
#include <list>
#include <utility>
+#include "oneapi/tbb/parallel_for.h"
Review Comment:
using `<>` for 3rdb libs?
##########
src/server/worker.cc:
##########
@@ -409,31 +410,27 @@ void Worker::FreeConnection(redis::Connection *conn) {
}
void Worker::FreeConnectionByID(int fd, uint64_t id) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end() && iter->second->GetID() == id) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
if (rate_limit_group_ != nullptr) {
- bufferevent_remove_from_rate_limit_group(iter->second->GetBufferEvent());
+ bufferevent_remove_from_rate_limit_group(accessor->second->GetBufferEvent());
}
- delete iter->second;
- conns_.erase(iter);
+
+ delete accessor->second;
+ conns_.erase(accessor);
Review Comment:
can we erase it before delete it? previously this is guarded by global lock, but now I think it would become dangerous without fencing
##########
src/server/worker.cc:
##########
@@ -409,31 +410,27 @@ void Worker::FreeConnection(redis::Connection *conn) {
}
void Worker::FreeConnectionByID(int fd, uint64_t id) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end() && iter->second->GetID() == id) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
if (rate_limit_group_ != nullptr) {
- bufferevent_remove_from_rate_limit_group(iter->second->GetBufferEvent());
+ bufferevent_remove_from_rate_limit_group(accessor->second->GetBufferEvent());
}
- delete iter->second;
- conns_.erase(iter);
+
+ delete accessor->second;
+ conns_.erase(accessor);
+
srv->DecrClientNum();
}
-
- iter = monitor_conns_.find(fd);
- if (iter != monitor_conns_.end() && iter->second->GetID() == id) {
- delete iter->second;
- monitor_conns_.erase(iter);
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, fd)) {
+ delete accessor->second;
Review Comment:
ditto
##########
src/server/worker.cc:
##########
@@ -442,115 +461,146 @@ Status Worker::EnableWriteEvent(int fd) {
}
Status Worker::Reply(int fd, const std::string &reply) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- iter->second->SetLastInteraction();
- redis::Reply(iter->second->Output(), reply);
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
+ accessor->second->SetLastInteraction();
+ redis::Reply(accessor->second->Output(), reply);
return Status::OK();
}
return {Status::NotOK, "connection doesn't exist"};
}
void Worker::BecomeMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- conns_.erase(conn->GetFD());
- monitor_conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
+ conns_.erase(accessor);
+ accessor.release();
+
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, conn->GetFD())) {
+ accessor->second = conn;
+ } else {
+ monitor_conns_.insert(accessor, std::make_pair(conn->GetFD(), conn));
+ }
}
srv->IncrMonitorClientNum();
conn->EnableFlag(redis::Connection::kMonitor);
}
void Worker::QuitMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- monitor_conns_.erase(conn->GetFD());
- conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, conn->GetFD())) {
+ {
+ monitor_conns_.erase(accessor);
+ accessor.release();
+ }
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
+ accessor->second = conn;
+ } else {
+ conns_.insert(accessor, std::make_pair(conn->GetFD(), conn));
+ }
}
srv->DecrMonitorClientNum();
conn->DisableFlag(redis::Connection::kMonitor);
}
void Worker::FeedMonitorConns(redis::Connection *conn, const std::string &response) {
- std::unique_lock<std::mutex> lock(conns_mu_);
-
- for (const auto &iter : monitor_conns_) {
- if (conn == iter.second) continue; // skip the monitor command
-
- if (conn->GetNamespace() == iter.second->GetNamespace() || iter.second->GetNamespace() == kDefaultNamespace) {
- iter.second->Reply(response);
+ tbb::parallel_for(monitor_conns_.range(), [conn, response](const ConnMap::range_type &range) {
+ for (auto &it : range) {
+ const auto &value = it.second;
+ if (conn == value) continue; // skip the monitor command
+ if (conn->GetNamespace() == value->GetNamespace() || value->GetNamespace() == kDefaultNamespace) {
+ value->Reply(response);
+ }
}
- }
+ });
}
std::string Worker::GetClientsStr() {
- std::unique_lock<std::mutex> lock(conns_mu_);
-
- std::string output;
- for (const auto &iter : conns_) {
- redis::Connection *conn = iter.second;
- output.append(conn->ToString());
- }
-
- return output;
+ return tbb::parallel_reduce(
+ conns_.range(), std::string{},
+ [](const ConnMap::range_type &range, std::string &&result) {
+ for (auto &it : range) {
+ result.append(it.second->ToString());
+ }
+ return result;
+ },
+ [](const std::string &lhs, const std::string &rhs) {
+ std::string result = lhs;
+ result.append(rhs);
+ return result;
+ });
}
void Worker::KillClient(redis::Connection *self, uint64_t id, const std::string &addr, uint64_t type, bool skipme,
int64_t *killed) {
- std::lock_guard<std::mutex> guard(conns_mu_);
-
- for (const auto &iter : conns_) {
- redis::Connection *conn = iter.second;
- if (skipme && self == conn) continue;
-
- // no need to kill the client again if the kCloseAfterReply flag is set
- if (conn->IsFlagEnabled(redis::Connection::kCloseAfterReply)) {
- continue;
- }
-
- if ((type & conn->GetClientType()) ||
- (!addr.empty() && (conn->GetAddr() == addr || conn->GetAnnounceAddr() == addr)) ||
- (id != 0 && conn->GetID() == id)) {
- conn->EnableFlag(redis::Connection::kCloseAfterReply);
- // enable write event to notify worker wake up ASAP, and remove the connection
- if (!conn->IsFlagEnabled(redis::Connection::kSlave)) { // don't enable any event in slave connection
- auto bev = conn->GetBufferEvent();
- bufferevent_enable(bev, EV_WRITE);
+ for (const auto key : getConnFds()) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, key)) {
+ auto conn = accessor->second;
+ if (skipme && self == conn) continue;
+
+ // no need to kill the client again if the kCloseAfterReply flag is set
+ if (conn->IsFlagEnabled(redis::Connection::kCloseAfterReply)) {
+ continue;
+ }
+ if ((type & conn->GetClientType()) ||
+ (!addr.empty() && (conn->GetAddr() == addr || conn->GetAnnounceAddr() == addr)) ||
+ (id != 0 && conn->GetID() == id)) {
+ conn->EnableFlag(redis::Connection::kCloseAfterReply);
+ // enable write event to notify worker wake up ASAP, and remove the connection
+ if (!conn->IsFlagEnabled(redis::Connection::kSlave)) { // don't enable any event in slave connection
+ auto bev = conn->GetBufferEvent();
+ bufferevent_enable(bev, EV_WRITE);
+ }
+ (*killed)++;
}
- (*killed)++;
}
}
}
void Worker::KickoutIdleClients(int timeout) {
std::vector<std::pair<int, uint64_t>> to_be_killed_conns;
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- if (conns_.empty()) {
- return;
- }
+ auto fd_list = getConnFds();
+ if (fd_list.empty()) {
+ return;
+ }
- int iterations = std::min(static_cast<int>(conns_.size()), 50);
- auto iter = conns_.upper_bound(last_iter_conn_fd_);
- while (iterations--) {
- if (iter == conns_.end()) iter = conns_.begin();
- if (static_cast<int>(iter->second->GetIdleTime()) >= timeout) {
- to_be_killed_conns.emplace_back(iter->first, iter->second->GetID());
- }
- iter++;
+ std::set<int> fds(fd_list.cbegin(), fd_list.cend());
+
+ int iterations = std::min(static_cast<int>(conns_.size()), 50);
+ auto iter = fds.upper_bound(last_iter_conn_fd_);
+ while (iterations--) {
+ if (iter == fds.end()) {
+ iter = fds.begin();
+ }
+ if (ConnMap::const_accessor accessor;
+ conns_.find(accessor, *iter) && static_cast<int>(accessor->second->GetIdleTime()) >= timeout) {
+ to_be_killed_conns.emplace_back(accessor->first, accessor->second->GetID());
}
- iter--;
- last_iter_conn_fd_ = iter->first;
+ iter++;
}
+ iter--;
+ last_iter_conn_fd_ = *iter;
for (const auto &conn : to_be_killed_conns) {
FreeConnectionByID(conn.first, conn.second);
}
}
+std::vector<int> Worker::getConnFds() const {
+ return tbb::parallel_reduce(
Review Comment:
> In fact, I'm not sure do we really need tbb hashmap to replace current mutex now.
I think it's worth doing if we don't traverse all `conns_` frequently.
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1711631422
##########
src/server/worker.cc:
##########
@@ -409,31 +410,27 @@ void Worker::FreeConnection(redis::Connection *conn) {
}
void Worker::FreeConnectionByID(int fd, uint64_t id) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end() && iter->second->GetID() == id) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
if (rate_limit_group_ != nullptr) {
- bufferevent_remove_from_rate_limit_group(iter->second->GetBufferEvent());
+ bufferevent_remove_from_rate_limit_group(accessor->second->GetBufferEvent());
}
- delete iter->second;
- conns_.erase(iter);
+
+ delete accessor->second;
+ conns_.erase(accessor);
Review Comment:
Hi @mapleFU ,
Thanks for your suggestion! 😊
I use an `accessor` rather than a `const_accessor` here.
I think other threads should have no access before `accessor` released.
And erasing before modifying the object is a much safer operation.
I have reordered the deleting and erasing action.
Bes Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1711681147
##########
src/server/worker.cc:
##########
@@ -409,31 +410,27 @@ void Worker::FreeConnection(redis::Connection *conn) {
}
void Worker::FreeConnectionByID(int fd, uint64_t id) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end() && iter->second->GetID() == id) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
if (rate_limit_group_ != nullptr) {
- bufferevent_remove_from_rate_limit_group(iter->second->GetBufferEvent());
+ bufferevent_remove_from_rate_limit_group(accessor->second->GetBufferEvent());
}
- delete iter->second;
- conns_.erase(iter);
+
+ delete accessor->second;
+ conns_.erase(accessor);
+
srv->DecrClientNum();
}
-
- iter = monitor_conns_.find(fd);
- if (iter != monitor_conns_.end() && iter->second->GetID() == id) {
- delete iter->second;
- monitor_conns_.erase(iter);
+ if (ConnMap::accessor accessor; monitor_conns_.find(accessor, fd)) {
+ delete accessor->second;
Review Comment:
My bad, so delete accessor before erase is ok, and you can add comment for that ( I don't know whether visit accessor after erase is ok)
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701931221
##########
src/server/worker.h:
##########
@@ -88,9 +91,10 @@ class Worker : EventCallbackBase<Worker>, EvconnlistenerBase<Worker> {
UniqueEvent timer_;
std::thread::id tid_;
std::vector<evconnlistener *> listen_events_;
- std::mutex conns_mu_;
- std::map<int, redis::Connection *> conns_;
- std::map<int, redis::Connection *> monitor_conns_;
+ mutable std::mutex conns_mu_; // refer to https://github.com/oneapi-src/oneTBB/issues/183, traverse and erase should
Review Comment:
I've take a look at tbb's code, and seems that the iterator it's just a wrapper to making the hash_map looks like a hash map in STL 😅
https://github.com/oneapi-src/oneTBB/blob/master/include/oneapi/tbb/concurrent_hash_map.h#L1043-L1046
What about 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] feat: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701987171
##########
src/server/worker.h:
##########
@@ -88,9 +91,10 @@ class Worker : EventCallbackBase<Worker>, EvconnlistenerBase<Worker> {
UniqueEvent timer_;
std::thread::id tid_;
std::vector<evconnlistener *> listen_events_;
- std::mutex conns_mu_;
- std::map<int, redis::Connection *> conns_;
- std::map<int, redis::Connection *> monitor_conns_;
+ mutable std::mutex conns_mu_; // refer to https://github.com/oneapi-src/oneTBB/issues/183, traverse and erase should
Review Comment:
> Maybe a simple read-write shared_mutex have a better performance and cleaner logic.
read-write lock is a tricky thing, usally it would bring up higher cost in checking the locks, and it's suitable for the cases that are merely updated. If we're sure the read is much more than write operations we can use that, otherwise just simple lock would be ok
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2325523932
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461) **Quality Gate passed**
Issues
 [12 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2461&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2461&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=2461&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [62.8% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2461&metric=new_coverage&view=list)
 [1.4% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2461&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461)
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1712996914
##########
src/server/worker.cc:
##########
@@ -442,115 +439,146 @@ Status Worker::EnableWriteEvent(int fd) {
}
Status Worker::Reply(int fd, const std::string &reply) {
- std::unique_lock<std::mutex> lock(conns_mu_);
- auto iter = conns_.find(fd);
- if (iter != conns_.end()) {
- iter->second->SetLastInteraction();
- redis::Reply(iter->second->Output(), reply);
+ if (ConnMap::accessor accessor; conns_.find(accessor, fd)) {
+ accessor->second->SetLastInteraction();
+ redis::Reply(accessor->second->Output(), reply);
return Status::OK();
}
return {Status::NotOK, "connection doesn't exist"};
}
void Worker::BecomeMonitorConn(redis::Connection *conn) {
- {
- std::lock_guard<std::mutex> guard(conns_mu_);
- conns_.erase(conn->GetFD());
- monitor_conns_[conn->GetFD()] = conn;
+ if (ConnMap::accessor accessor; conns_.find(accessor, conn->GetFD())) {
Review Comment:
Hi @mapleFU ,
Thanks very much for your advice! 😊
Yes, this will change the behavior of `monitor_conns_`.
I have moved the `monitor_conns_` inserting out of the scope.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1707299725
##########
src/server/worker.cc:
##########
@@ -76,17 +78,40 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ auto conns = tbb::parallel_reduce(
+ conns_.range(), std::vector<redis::Connection *>{},
+ [](const ConnMap::range_type &range, std::vector<redis::Connection *> &&result) {
+ for (auto &it : range) {
+ result.push_back(it.second);
+ }
+ return result;
+ },
+ [](const std::vector<redis::Connection *> &lhs, const std::vector<redis::Connection *> &rhs) {
+ std::vector<redis::Connection *> result = lhs;
+ result.insert(result.end(), rhs.begin(), rhs.end());
+ return result;
+ });
+
+ auto monitor_conns = tbb::parallel_reduce(
+ monitor_conns_.range(), std::vector<redis::Connection *>{},
+ [](const ConnMap::range_type &range, std::vector<redis::Connection *> &&result) {
+ for (auto &it : range) {
+ result.push_back(it.second);
+ }
+ return result;
+ },
+ [](const std::vector<redis::Connection *> &lhs, const std::vector<redis::Connection *> &rhs) {
+ std::vector<redis::Connection *> result = lhs;
+ result.insert(result.end(), rhs.begin(), rhs.end());
+ return result;
+ });
Review Comment:
Hi @git-hulk ,
Thanks very much for your suggestions! 😊
I have refactored the duplicated code.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2273780018
Thanks for your benchmarking! Maybe I'd like to run this and trying to perf the mutex contention when I have time.
I've check out the benchmark result but the benchmark result turns to be unstable 🤔 The graph below shows that in different percentage result would be different...So maybe we should analysis the performance bottleneck for connections handling and check whether it could bring up performance improvement
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2297818560
Hi @mapleFU ,
Thanks very much for your review suggestions! 😄
> it would brings up inconsistency between `monitor_conns_` and `conns_`
I agree on this concern. Removing the lock will lead to two connection collection's inconsistency.
It can be mitigated if we fetch the `monitor_conns_` accessor inside the related accessor scope of `conns_`.
But frankly I think this way increased the complexity of current logic.
If the connections quantity is not so much, the performance may not have improvement.
> tbb::parallel_for would goes here: https://github.com/oneapi-src/oneTBB/blob/2d516c871b79d81032f0ca44d76e5072bc62d479/src/tbb/task_dispatcher.h#L243 which requires task_scheduler and an independent pool.
Thanks for your investigation!
I think that I should follow current's way and construct a single thread execution pool for the `parallel_for` and `parallel_reduce`.
By the way, after reading our codebase, I think that the tbb container may more suitable for containers which just need to protect themselves rather than more related states.
Maybe states like [`block_keys_`](https://github.com/apache/kvrocks/blob/f38557b1d87eca93c8e360f4f450bfdf4103ed61/src/server/server.h#L380) and [`blocked_stream_consumers_`](https://github.com/apache/kvrocks/blob/f38557b1d87eca93c8e360f4f450bfdf4103ed61/src/server/server.h#L386) are more suitable for tbb containers.
Best Regards,
Edward
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701977622
##########
src/server/worker.cc:
##########
@@ -76,17 +77,22 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ {
+ std::lock_guard<std::mutex> guard(conns_mu_);
+ for (const auto &iter : conns_) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, iter.first)) {
+ accessor->second->Close();
Review Comment:
Hi @mapleFU ,
Sorry for putting time-costly tasks into a critical section.
I will make this action only manipulate with the container and leave `Close()` action outside the critical section.
Best Regards,
Edward
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#issuecomment-2267655996
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461) **Quality Gate failed**
Failed conditions
 [42.5% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2461&metric=new_coverage&view=list) (required ≥ 50%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2461)
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1702006565
##########
src/server/worker.cc:
##########
@@ -76,17 +77,22 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ {
+ std::lock_guard<std::mutex> guard(conns_mu_);
+ for (const auto &iter : conns_) {
+ if (ConnMap::accessor accessor; conns_.find(accessor, iter.first)) {
+ accessor->second->Close();
Review Comment:
It's ok, we can use the method here: https://github.com/apache/kvrocks/pull/2461#discussion_r1701976227 or using a vector to collect the conns and close them after lock. Generally, IO related options should be better outside critical section lock for performance
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701375969
##########
src/server/worker.h:
##########
@@ -88,9 +91,10 @@ class Worker : EventCallbackBase<Worker>, EvconnlistenerBase<Worker> {
UniqueEvent timer_;
std::thread::id tid_;
std::vector<evconnlistener *> listen_events_;
- std::mutex conns_mu_;
- std::map<int, redis::Connection *> conns_;
- std::map<int, redis::Connection *> monitor_conns_;
+ mutable std::mutex conns_mu_; // refer to https://github.com/oneapi-src/oneTBB/issues/183, traverse and erase should
Review Comment:
Can we reference to the document of TBB? 🤔 An issue might be closed or fixed later so a official doc which point out the concurrent behavior here might be beeter
--
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: replace worker connection's mutex partially with `tbb::concurrent_hash_map` [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1701931221
##########
src/server/worker.h:
##########
@@ -88,9 +91,10 @@ class Worker : EventCallbackBase<Worker>, EvconnlistenerBase<Worker> {
UniqueEvent timer_;
std::thread::id tid_;
std::vector<evconnlistener *> listen_events_;
- std::mutex conns_mu_;
- std::map<int, redis::Connection *> conns_;
- std::map<int, redis::Connection *> monitor_conns_;
+ mutable std::mutex conns_mu_; // refer to https://github.com/oneapi-src/oneTBB/issues/183, traverse and erase should
Review Comment:
https://github.com/oneapi-src/oneTBB/blob/master/include/oneapi/tbb/concurrent_hash_map.h#L1043-L1046
What about 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] feat: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "PragmaTwice (via GitHub)" <gi...@apache.org>.
PragmaTwice commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1710636021
##########
src/server/worker.cc:
##########
@@ -76,17 +78,29 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
}
Worker::~Worker() {
- std::vector<redis::Connection *> conns;
- conns.reserve(conns_.size() + monitor_conns_.size());
-
- for (const auto &iter : conns_) {
- conns.emplace_back(iter.second);
- }
- for (const auto &iter : monitor_conns_) {
- conns.emplace_back(iter.second);
+ auto collect_conns_fn = [](ConnMap &conns) {
+ return parallel_reduce(
Review Comment:
I mean maybe a single-thread for is enough here? didn't have a deep look yet cc @mapleFU
--
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: replace worker connection's mutex with `tbb::concurrent_hash_map` [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2461:
URL: https://github.com/apache/kvrocks/pull/2461#discussion_r1711621495
##########
src/server/worker.cc:
##########
@@ -47,6 +47,8 @@
#include <list>
#include <utility>
+#include "oneapi/tbb/parallel_for.h"
Review Comment:
Hi @mapleFU ,
Thanks very much for your suggestions! 😊
I have made all tbb headers including with `<>`.
Best Regards,
Edward
--
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