You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@kvrocks.apache.org by "hll1213181368 (via GitHub)" <gi...@apache.org> on 2025/10/30 08:29:28 UTC
[PR] chore(cluster): parallel to sendSnapshotByRawKV [kvrocks]
hll1213181368 opened a new pull request, #3240:
URL: https://github.com/apache/kvrocks/pull/3240
Use multithreads to improve slotRange migration speed of sendSnapshotByRawKV.
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3485207308
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=3240) **Quality Gate passed**
Issues
 [0 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=3240&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=3240&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=3240&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=3240&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=3240&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=3240)
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#discussion_r2490342741
##########
src/cluster/slot_migrate.cc:
##########
@@ -1260,8 +1263,51 @@ Status SlotMigrator::sendSnapshotByRawKV() {
auto slot_range = slot_range_.load();
info("[migrate] Migrating snapshot of slot(s) {} by raw key value", slot_range.String());
- auto prefix = ComposeSlotKeyPrefix(namespace_, slot_range.start);
- auto upper_bound = ComposeSlotKeyUpperBound(namespace_, slot_range.end);
+ int total_slots = slot_range.end - slot_range.start + 1;
+ unsigned int max_parallelism = std::thread::hardware_concurrency();
+ unsigned int available_slots = static_cast<unsigned int>(total_slots);
+ int parallelism = static_cast<int>(std::min(max_parallelism, available_slots));
+ int base = total_slots / parallelism;
+ int rem = total_slots % parallelism;
+
+ std::vector<std::future<Status>> results;
+ int cur_start = slot_range.start;
+ for (int i = 0; i < parallelism; i++) {
+ int count = base + (i < rem ? 1 : 0);
+ int cur_end = cur_start + count - 1;
+ if (count <= 0) break;
+
+ results.emplace_back(std::async(std::launch::async, [=]() -> Status {
+ int fd = createConnectToDstNode();
+ if (fd < 0) {
+ return Status(Status::NotOK, fmt::format("thread[{}] createConnectToDstNode failed", i));
+ }
+ auto s = migrateSlotRange(cur_start, cur_end, parallelism, fd);
+ close(fd);
+ return s;
+ }));
+
+ cur_start = cur_end + 1;
+ }
+
+ // Wait til finish
+ for (auto &result : results) {
+ auto s = result.get();
Review Comment:
@hll1213181368 Seems this comment was resolved? Did I miss anything?
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3585923573
> if one of the threads fails, Others thread still migrate. But the total result is NotOK. This slotRange migrate failed. we need to analysis the problem or retry to migrate this slotRange.
>
> @git-hulk Do you think so? Or do you have any good idea?
I'm good with that allows the partial failure while doing the parallel migration. But the current implementation doesn't respect this behavior. We should collect and wait for all migration threads to finish, then check whether each is successful, partially successful, or a failure.
--
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] chore(cluster): parallel to sendSnapshotByRawKV [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3475822321
@hll1213181368 Do you test how much it will be improved if you increase the number of threads?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "hll1213181368 (via GitHub)" <gi...@apache.org>.
hll1213181368 commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3544593256
> @hll1213181368 I didn't see that. Please check if your reply comments are in pending status.
@PragmaTwice how to set threads for this migrate task. I don't think more better idea.
<img width="793" height="636" alt="image" src="https://github.com/user-attachments/assets/1da242a1-b3b7-4498-9c0d-cd69bc333861" />
@git-hulk @PragmaTwice Later I will commit another pr. parallel to improve AsyncScanDBSize task. Similar to this deal principle.
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "hll1213181368 (via GitHub)" <gi...@apache.org>.
hll1213181368 commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3551399481
@git-hulk Above I have reply comments. Better to solve one threads fail is to try migrate this failed slotRange again.
--
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] chore(cluster): parallel to sendSnapshotByRawKV [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#discussion_r2484151430
##########
src/cluster/slot_migrate.cc:
##########
@@ -1272,12 +1318,11 @@ Status SlotMigrator::sendSnapshotByRawKV() {
auto no_txn_ctx = engine::Context::NoTransactionContext(storage_);
engine::DBIterator iter(no_txn_ctx, read_options);
- BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+ BatchSender batch_sender(fd, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_ / parallelism);
Review Comment:
`migrate_batch_bytes_per_sec_ / parallelism`, this might affect the migration performance when the keys exist in a few slots. The better way is to share a global rate limiter between threads.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "PragmaTwice (via GitHub)" <gi...@apache.org>.
PragmaTwice commented on code in PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#discussion_r2484237312
##########
src/cluster/slot_migrate.cc:
##########
@@ -1325,16 +1370,44 @@ Status SlotMigrator::sendSnapshotByRawKV() {
GET_OR_RET(sendMigrationBatch(&batch_sender));
- auto elapsed = util::GetTimeStampMS() - start_ts;
- info(
- "[migrate] Succeed to migrate snapshot range, slot(s): {}, elapsed: {} ms, sent: {} bytes, rate: {:.2f} kb/s, "
- "batches: {}, entries: {}",
- slot_range.String(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
- batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
-
return Status::OK();
}
+int SlotMigrator::createConnectToDstNode() {
+ struct sockaddr_in addr {};
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(dst_port_);
+ if (inet_pton(AF_INET, dst_ip_.c_str(), &addr.sin_addr) != 1) {
+ error("Invalid dst IP: {}", dst_ip_);
+ return -1;
+ }
+
+ int fd = socket(AF_INET, SOCK_STREAM, 0);
+ if (fd < 0) {
+ error("socket() failed: {}", strerror(errno));
+ return -1;
+ }
+ int one = 1;
+ setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
+
+ if (::connect(fd, (struct sockaddr *)&addr, sizeof(addr)) != 0) {
+ error("connect to {}:{} failed: {}", dst_ip_, dst_port_, strerror(errno));
+ close(fd);
+ return -1;
+ }
+
+ // add fd authOnDstNode
Review Comment:
if you want to add a comment, make sure that it is readable by other people.
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "hll1213181368 (via GitHub)" <gi...@apache.org>.
hll1213181368 closed pull request #3240: perf(cluster): use multithreads to optimize sendSnapshotByRawKV
URL: https://github.com/apache/kvrocks/pull/3240
--
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] chore(cluster): parallel to sendSnapshotByRawKV [kvrocks]
Posted by "PragmaTwice (via GitHub)" <gi...@apache.org>.
PragmaTwice commented on code in PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#discussion_r2484237312
##########
src/cluster/slot_migrate.cc:
##########
@@ -1325,16 +1370,44 @@ Status SlotMigrator::sendSnapshotByRawKV() {
GET_OR_RET(sendMigrationBatch(&batch_sender));
- auto elapsed = util::GetTimeStampMS() - start_ts;
- info(
- "[migrate] Succeed to migrate snapshot range, slot(s): {}, elapsed: {} ms, sent: {} bytes, rate: {:.2f} kb/s, "
- "batches: {}, entries: {}",
- slot_range.String(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
- batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
-
return Status::OK();
}
+int SlotMigrator::createConnectToDstNode() {
+ struct sockaddr_in addr {};
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(dst_port_);
+ if (inet_pton(AF_INET, dst_ip_.c_str(), &addr.sin_addr) != 1) {
+ error("Invalid dst IP: {}", dst_ip_);
+ return -1;
+ }
+
+ int fd = socket(AF_INET, SOCK_STREAM, 0);
+ if (fd < 0) {
+ error("socket() failed: {}", strerror(errno));
+ return -1;
+ }
+ int one = 1;
+ setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
+
+ if (::connect(fd, (struct sockaddr *)&addr, sizeof(addr)) != 0) {
+ error("connect to {}:{} failed: {}", dst_ip_, dst_port_, strerror(errno));
+ close(fd);
+ return -1;
+ }
+
+ // add fd authOnDstNode
Review Comment:
if you want to add a comment, make sure that it is readable to other people.
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "hll1213181368 (via GitHub)" <gi...@apache.org>.
hll1213181368 commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3544575153
> @hll1213181368 I didn't see that. Please check if your reply comments are in pending status.
<img width="784" height="172" alt="image" src="https://github.com/user-attachments/assets/912bae3a-293e-4388-8a13-33dac0d3b8ac" />
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3491705593
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=3240) **Quality Gate passed**
Issues
 [0 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=3240&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=3240&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=3240&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=3240&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=3240&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=3240)
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "hll1213181368 (via GitHub)" <gi...@apache.org>.
hll1213181368 commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3541582274
> @hll1213181368 There are still two comments to be resolved, and you won't be able to mark them as resolved without any answer or actions.
>
> For [#3240 (comment)](https://github.com/apache/kvrocks/pull/3240#discussion_r2484096543), we need to handle the partial failure case.
>
> For [#3240 (comment)](https://github.com/apache/kvrocks/pull/3240#discussion_r2484237567), you should support a configuration, or explain what do you think.
@git-hulk @PragmaTwice I have response this two comment. Do you have 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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3541713041
@hll1213181368 I didn't see that. Please check if your reply comments are in pending status?
--
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] chore(cluster): parallel to sendSnapshotByRawKV [kvrocks]
Posted by "hll1213181368 (via GitHub)" <gi...@apache.org>.
hll1213181368 commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3475952552
> @hll1213181368 Do you test how much it will be improved if you increase the number of threads?
The maximum length of the migration interval is designed to be 10 slots, which can improve the migration speed by about 5 times.
--
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] chore(cluster): parallel to sendSnapshotByRawKV [kvrocks]
Posted by "PragmaTwice (via GitHub)" <gi...@apache.org>.
PragmaTwice commented on code in PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#discussion_r2484237567
##########
src/cluster/slot_migrate.cc:
##########
@@ -1260,8 +1263,51 @@ Status SlotMigrator::sendSnapshotByRawKV() {
auto slot_range = slot_range_.load();
info("[migrate] Migrating snapshot of slot(s) {} by raw key value", slot_range.String());
- auto prefix = ComposeSlotKeyPrefix(namespace_, slot_range.start);
- auto upper_bound = ComposeSlotKeyUpperBound(namespace_, slot_range.end);
+ int total_slots = slot_range.end - slot_range.start + 1;
+ unsigned int max_parallelism = std::thread::hardware_concurrency();
Review Comment:
better to have an option instead of just querying nproc.
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3548574819
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=3240) **Quality Gate passed**
Issues
 [0 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=3240&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=3240&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=3240&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=3240&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=3240&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=3240)
--
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] chore(cluster): parallel to sendSnapshotByRawKV [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#discussion_r2484094630
##########
src/cluster/slot_migrate.cc:
##########
@@ -1260,8 +1263,51 @@ Status SlotMigrator::sendSnapshotByRawKV() {
auto slot_range = slot_range_.load();
info("[migrate] Migrating snapshot of slot(s) {} by raw key value", slot_range.String());
- auto prefix = ComposeSlotKeyPrefix(namespace_, slot_range.start);
- auto upper_bound = ComposeSlotKeyUpperBound(namespace_, slot_range.end);
+ int total_slots = slot_range.end - slot_range.start + 1;
+ unsigned int max_parallelism = std::thread::hardware_concurrency();
+ unsigned int available_slots = static_cast<unsigned int>(total_slots);
+ int parallelism = static_cast<int>(std::min(max_parallelism, available_slots));
+ int base = total_slots / parallelism;
+ int rem = total_slots % parallelism;
+
+ std::vector<std::future<Status>> results;
+ int cur_start = slot_range.start;
+ for (int i = 0; i < parallelism; i++) {
+ int count = base + (i < rem ? 1 : 0);
+ int cur_end = cur_start + count - 1;
+ if (count <= 0) break;
Review Comment:
When the count would be 0?
##########
src/cluster/slot_migrate.cc:
##########
@@ -1260,8 +1263,51 @@ Status SlotMigrator::sendSnapshotByRawKV() {
auto slot_range = slot_range_.load();
info("[migrate] Migrating snapshot of slot(s) {} by raw key value", slot_range.String());
- auto prefix = ComposeSlotKeyPrefix(namespace_, slot_range.start);
- auto upper_bound = ComposeSlotKeyUpperBound(namespace_, slot_range.end);
+ int total_slots = slot_range.end - slot_range.start + 1;
+ unsigned int max_parallelism = std::thread::hardware_concurrency();
+ unsigned int available_slots = static_cast<unsigned int>(total_slots);
+ int parallelism = static_cast<int>(std::min(max_parallelism, available_slots));
+ int base = total_slots / parallelism;
+ int rem = total_slots % parallelism;
Review Comment:
Those two should be `slots_per_thread` and `remain_slots?`
##########
src/cluster/slot_migrate.cc:
##########
@@ -1260,8 +1263,51 @@ Status SlotMigrator::sendSnapshotByRawKV() {
auto slot_range = slot_range_.load();
info("[migrate] Migrating snapshot of slot(s) {} by raw key value", slot_range.String());
- auto prefix = ComposeSlotKeyPrefix(namespace_, slot_range.start);
- auto upper_bound = ComposeSlotKeyUpperBound(namespace_, slot_range.end);
+ int total_slots = slot_range.end - slot_range.start + 1;
+ unsigned int max_parallelism = std::thread::hardware_concurrency();
+ unsigned int available_slots = static_cast<unsigned int>(total_slots);
+ int parallelism = static_cast<int>(std::min(max_parallelism, available_slots));
+ int base = total_slots / parallelism;
+ int rem = total_slots % parallelism;
+
+ std::vector<std::future<Status>> results;
+ int cur_start = slot_range.start;
+ for (int i = 0; i < parallelism; i++) {
+ int count = base + (i < rem ? 1 : 0);
+ int cur_end = cur_start + count - 1;
+ if (count <= 0) break;
+
+ results.emplace_back(std::async(std::launch::async, [=]() -> Status {
+ int fd = createConnectToDstNode();
+ if (fd < 0) {
+ return Status(Status::NotOK, fmt::format("thread[{}] createConnectToDstNode failed", i));
Review Comment:
```suggestion
return Status(Status::NotOK, fmt::format("failed to connect the destination node in thread[{}]", i));
```
##########
src/cluster/slot_migrate.cc:
##########
@@ -1325,16 +1370,44 @@ Status SlotMigrator::sendSnapshotByRawKV() {
GET_OR_RET(sendMigrationBatch(&batch_sender));
- auto elapsed = util::GetTimeStampMS() - start_ts;
- info(
- "[migrate] Succeed to migrate snapshot range, slot(s): {}, elapsed: {} ms, sent: {} bytes, rate: {:.2f} kb/s, "
- "batches: {}, entries: {}",
- slot_range.String(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
- batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
-
return Status::OK();
}
+int SlotMigrator::createConnectToDstNode() {
Review Comment:
You can just `SockConnect` instead of creating another one.
##########
src/cluster/slot_migrate.cc:
##########
@@ -1260,8 +1263,51 @@ Status SlotMigrator::sendSnapshotByRawKV() {
auto slot_range = slot_range_.load();
info("[migrate] Migrating snapshot of slot(s) {} by raw key value", slot_range.String());
- auto prefix = ComposeSlotKeyPrefix(namespace_, slot_range.start);
- auto upper_bound = ComposeSlotKeyUpperBound(namespace_, slot_range.end);
+ int total_slots = slot_range.end - slot_range.start + 1;
+ unsigned int max_parallelism = std::thread::hardware_concurrency();
+ unsigned int available_slots = static_cast<unsigned int>(total_slots);
Review Comment:
I think this line is unnecessary; you can just cast max_parallelism to `int`.
##########
src/cluster/slot_migrate.cc:
##########
@@ -1260,8 +1263,51 @@ Status SlotMigrator::sendSnapshotByRawKV() {
auto slot_range = slot_range_.load();
info("[migrate] Migrating snapshot of slot(s) {} by raw key value", slot_range.String());
- auto prefix = ComposeSlotKeyPrefix(namespace_, slot_range.start);
- auto upper_bound = ComposeSlotKeyUpperBound(namespace_, slot_range.end);
+ int total_slots = slot_range.end - slot_range.start + 1;
+ unsigned int max_parallelism = std::thread::hardware_concurrency();
+ unsigned int available_slots = static_cast<unsigned int>(total_slots);
+ int parallelism = static_cast<int>(std::min(max_parallelism, available_slots));
+ int base = total_slots / parallelism;
+ int rem = total_slots % parallelism;
+
+ std::vector<std::future<Status>> results;
+ int cur_start = slot_range.start;
+ for (int i = 0; i < parallelism; i++) {
+ int count = base + (i < rem ? 1 : 0);
+ int cur_end = cur_start + count - 1;
+ if (count <= 0) break;
+
+ results.emplace_back(std::async(std::launch::async, [=]() -> Status {
+ int fd = createConnectToDstNode();
+ if (fd < 0) {
+ return Status(Status::NotOK, fmt::format("thread[{}] createConnectToDstNode failed", i));
+ }
+ auto s = migrateSlotRange(cur_start, cur_end, parallelism, fd);
+ close(fd);
+ return s;
+ }));
+
+ cur_start = cur_end + 1;
+ }
+
+ // Wait til finish
+ for (auto &result : results) {
+ auto s = result.get();
Review Comment:
What if one of the threads fails? Others are still migrating?
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3491219443
@hll1213181368 There are still two comments to be resolved, and you won't be able to mark them as resolved without any answer or actions.
For https://github.com/apache/kvrocks/pull/3240#discussion_r2484096543, we need to handle the partial failure case.
For https://github.com/apache/kvrocks/pull/3240#discussion_r2484237567, you should support a configuration, or explain what do you think.
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3545169354
> > @hll1213181368 I didn't see that. Please check if your reply comments are in pending status.
>
> @PragmaTwice how to set threads for this migrate task. I don't think more better idea. <img alt="image" width="793" height="636" src="https://private-user-images.githubusercontent.com/17724484/515460731-1da242a1-b3b7-4498-9c0d-cd69bc333861.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NjM0NDQ0MDYsIm5iZiI6MTc2MzQ0NDEwNiwicGF0aCI6Ii8xNzcyNDQ4NC81MTU0NjA3MzEtMWRhMjQyYTEtYjNiNy00NDk4LTljMGQtY2Q2OWJjMzMzODYxLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNTExMTglMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjUxMTE4VDA1MzUwNlomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTcxYmEwYWFmNWQ1NjBmMmU0ZDQyZTY2MTIwZmE1NmYxNjJjYTE4YTI2Nzc4ZGU4ZTYyOTg2NzIwZjdjOGEwMzgmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0In0.uM0r4KHw6cI2zu0jtK6xNl13h8ixCLzn8J0a5GFjueA">
>
> @git-hulk @PragmaTwice Later I will commit another pr. parallel to improve AsyncScanDBSize task. Similar to this deal principle.
Those comments are pending and not visible to others. Please release them via the comment button.
--
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] perf(cluster): use multithreads to optimize sendSnapshotByRawKV [kvrocks]
Posted by "hll1213181368 (via GitHub)" <gi...@apache.org>.
hll1213181368 commented on PR #3240:
URL: https://github.com/apache/kvrocks/pull/3240#issuecomment-3596025565
> > if one of the threads fails, Others thread still migrate. But the total result is NotOK. This slotRange migrate failed. we need to analysis the problem or retry to migrate this slotRange.
> > @git-hulk Do you think so? Or do you have any good idea?
>
> I'm good with that allows the partial failure while doing the parallel migration. But the current implementation doesn't respect this behavior. We should collect and wait for all migration threads to finish, then check whether each is successful, partially successful, or a failure.
@git-hulk The kvrocks-controller sends a range slots migration request, and the kvrocks node returns either success or failure result for the entire range. For a failed range, the kvrocks-controller is called again to retry the entire range. Successful slots within the range will be returned as successfully migrated, while failed slots will be migrated again. I think it's unnecessary for the kvrocks node to return which slots in the range succeeded and which failed.
--
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