This repository was archived by the owner on Nov 15, 2023. It is now read-only.
Fix import queue thread pool shutdown #4929
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
bkchr
suggested changes
Feb 14, 2020
| // Flush the queue and close the receiver to terminate the future. | ||
| let _ = self.sender.unbounded_send(ToWorkerMsg::Shutdown); | ||
| let (_, closed) = buffered_link::buffered_link(); | ||
| drop(std::mem::replace(&mut self.result_port, closed)); |
| fn drop(&mut self) { | ||
| drop(self.pool.take()); | ||
| // Flush the queue and close the receiver to terminate the future. | ||
| let _ = self.sender.unbounded_send(ToWorkerMsg::Shutdown); |
There was a problem hiding this comment.
I think we should just close it: https://docs.rs/futures/0.3.4/futures/channel/mpsc/struct.UnboundedSender.html#method.close_channel
There was a problem hiding this comment.
Suggested change
| let _ = self.sender.unbounded_send(ToWorkerMsg::Shutdown); | |
| self.sender.close_channel(); |
| let mut pool = futures::executor::ThreadPool::builder() | ||
| .name_prefix("import-queue-worker-") | ||
| .pool_size(1) | ||
| .pool_size(2) |
There was a problem hiding this comment.
Isn't there just one worker anyway?
There was a problem hiding this comment.
Was testing if thread tracking code is working properly. Eventually there should be more workers, but this is out of scope of this PR.
|
Also fixed an issue with slog global guard being disposed too early. |
bkchr
reviewed
Feb 14, 2020
| @@ -144,6 +178,7 @@ enum ToWorkerMsg<B: BlockT> { | |||
| ImportBlocks(BlockOrigin, Vec<IncomingBlock<B>>), | |||
| ImportJustification(Origin, B::Hash, NumberFor<B>, Justification), | |||
| ImportFinalityProof(Origin, B::Hash, NumberFor<B>, Vec<u8>), | |||
| Shutdown, | |||
| @@ -239,6 +274,7 @@ impl<B: BlockT, Transaction: Send> BlockImportWorker<B, Transaction> { | |||
| ToWorkerMsg::ImportJustification(who, hash, number, justification) => { | |||
| worker.import_justification(who, hash, number, justification); | |||
| } | |||
| ToWorkerMsg::Shutdown => return Poll::Ready(()), | |||
There was a problem hiding this comment.
Suggested change
| ToWorkerMsg::Shutdown => return Poll::Ready(()), |
| fn drop(&mut self) { | ||
| drop(self.pool.take()); | ||
| // Flush the queue and close the receiver to terminate the future. | ||
| let _ = self.sender.unbounded_send(ToWorkerMsg::Shutdown); |
There was a problem hiding this comment.
Suggested change
| let _ = self.sender.unbounded_send(ToWorkerMsg::Shutdown); | |
| self.sender.close_channel(); |
bkchr
approved these changes
Feb 14, 2020
ordian
reviewed
Feb 15, 2020
| @@ -40,9 +41,28 @@ pub struct BasicQueue<B: BlockT, Transaction> { | |||
| manual_poll: Option<Pin<Box<dyn Future<Output = ()> + Send>>>, | |||
| /// A thread pool where the background worker is being run. | |||
| pool: Option<futures::executor::ThreadPool>, | |||
| pool_guard: Arc<(Mutex<usize>, Condvar)>, | |||
There was a problem hiding this comment.
Any reason not to use a https://docs.rs/crossbeam/0.7.3/crossbeam/sync/struct.WaitGroup.html?
There was a problem hiding this comment.
- There's no control over the lifetimes of objects passed to
ThreadPoolBuilder. It uses some reference counting internally and relying on implementation details seems wrong. - Don't want to introduce another dependency over a trivial thing.
This was referenced Feb 17, 2020
General-Beck
pushed a commit
to General-Beck/substrate
that referenced
this pull request
Feb 18, 2020
* Fix import queue thread pool shutdown * Make sure runtime is disposed before telemetry * Close channel istead of sending a message * Fixed test
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
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.
Due to the way rs-futures
ThreadPoolis implemented, Importing threads may outlive the queue instance and even the main thread. This leads to a race on closing rocksdb database vs cleaning the C++ runtime in the main thread, causing segfaults and preventing the database from being closed properly.This PR adds explicit synchronization between importing threads and the queue shutdown.
Fixes #4913