feat: send notification failed then retry #925
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
HuaHuaY
requested review from
skyzh,
yezizp2012,
BugenZhao,
BowenXiao1999 and
zwang28
Codecov Report
@@ Coverage Diff @@
## main #925 +/- ##
=========================================
Coverage 71.80% 71.80%
Complexity 2766 2766
=========================================
Files 971 971
Lines 56790 56773 -17
Branches 1787 1787
=========================================
- Hits 40776 40764 -12
+ Misses 15124 15119 -5
Partials 890 890
Flags with carried forward coverage won't be shown. Click here to find out more.
📣 Codecov can now indicate which changes are the most critical in Pull Requests. Learn more |
BugenZhao
reviewed
Mar 15, 2022
Comment on lines
85
to
108
| loop { | ||
| // Heartbeat may delete worker. | ||
| // We assume that after a worker is disconnected, before it recalls subscribe, we | ||
| // will call notify. | ||
| while let Ok(x) = self.rx.try_recv() { | ||
| keys.insert(x); | ||
| } | ||
| if keys.contains(worker_key) { | ||
| break; | ||
| } | ||
| let result = sender | ||
| .send(Ok(SubscribeResponse { | ||
| status: None, | ||
| operation: operation as i32, | ||
| info: Some(info.clone()), | ||
| })) | ||
| .await; | ||
| if result.is_ok() { | ||
| break; | ||
| } | ||
| time::sleep(Duration::from_micros(NOTIFY_RETRY_INTERVAL)).await; | ||
| } | ||
| NotificationTarget::ComputeNode => Box::new(self.be_observers.iter()), | ||
| NotificationTarget::Frontend => Box::new(self.fe_observers.iter()), | ||
| } | ||
| self.remove_by_key(keys); |
There was a problem hiding this comment.
These snippets seem to be duplicated.
There was a problem hiding this comment.
Yeah, the difference of two methods is just use the HashMap of frontend or backend. I think there are only two functions, and in the same source file, there is no need to encapsulate another layer at present.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What's changed and what's your intention?
Mutex<NotificationManagerCore>instead ofDashMap, which help us to control concurrency issues. Side effect is we can't callnotifymethod in multiple threads at the same time.NotificationTarget.I assume that after a worker is disconnected, before we tell notification manager that its sender should be removed, the worker will not recall
subscribeRPC.Consider we store
Workerindelete_worker_nodemethod, I do not store information in notification manager when deleting the worker's sender.Checklist
Refer to a related PR or issue link (optional)
related #361