-
Notifications
You must be signed in to change notification settings - Fork 613
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor(meta): unify hummock manager worker (#6675)
refactor(meta): add hummock manager worker Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
- Loading branch information
1 parent
bb051f7
commit 3ec1ea0
Showing
8 changed files
with
195 additions
and
190 deletions.
There are no files selected for viewing
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
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,154 @@ | ||
| // Copyright 2022 Singularity Data | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| use std::time::Duration; | ||
|
|
||
| use sync_point::sync_point; | ||
| use tokio::task::JoinHandle; | ||
| use tokio_retry::strategy::{jitter, ExponentialBackoff}; | ||
|
|
||
| use crate::hummock::utils::RetryableError; | ||
| use crate::hummock::{HummockManager, HummockManagerRef}; | ||
| use crate::manager::LocalNotification; | ||
| use crate::storage::MetaStore; | ||
|
|
||
| pub type HummockManagerEventSender = tokio::sync::mpsc::UnboundedSender<HummockManagerEvent>; | ||
| pub type HummockManagerEventReceiver = tokio::sync::mpsc::UnboundedReceiver<HummockManagerEvent>; | ||
|
|
||
| pub enum HummockManagerEvent { | ||
| Shutdown, | ||
| } | ||
|
|
||
| impl<S> HummockManager<S> | ||
| where | ||
| S: MetaStore, | ||
| { | ||
| pub(crate) async fn start_worker( | ||
| self: &HummockManagerRef<S>, | ||
| mut receiver: HummockManagerEventReceiver, | ||
| ) -> JoinHandle<()> { | ||
| let (local_notification_tx, mut local_notification_rx) = | ||
| tokio::sync::mpsc::unbounded_channel(); | ||
| self.env | ||
| .notification_manager() | ||
| .insert_local_sender(local_notification_tx) | ||
| .await; | ||
| let hummock_manager = self.clone(); | ||
| tokio::spawn(async move { | ||
| loop { | ||
| tokio::select! { | ||
| notification = local_notification_rx.recv() => { | ||
| match notification { | ||
| Some(notification) => { | ||
| hummock_manager | ||
| .handle_local_notification(notification) | ||
| .await; | ||
| } | ||
| None => { | ||
| return; | ||
| } | ||
| } | ||
| } | ||
| hummock_manager_event = receiver.recv() => { | ||
| match hummock_manager_event { | ||
| Some(hummock_manager_event) => { | ||
| if !hummock_manager | ||
| .handle_hummock_manager_event(hummock_manager_event) | ||
| .await { | ||
| return; | ||
| } | ||
| } | ||
| None => { | ||
| return; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| }) | ||
| } | ||
|
|
||
| /// Returns false indicates to shutdown worker | ||
| #[expect(clippy::unused_async)] | ||
| async fn handle_hummock_manager_event(&self, event: HummockManagerEvent) -> bool { | ||
| match event { | ||
| HummockManagerEvent::Shutdown => { | ||
| tracing::info!("Hummock manager worker is stopped"); | ||
| false | ||
| } | ||
| } | ||
| } | ||
|
|
||
| async fn handle_local_notification(&self, notification: LocalNotification) { | ||
| let retry_strategy = ExponentialBackoff::from_millis(10) | ||
| .max_delay(Duration::from_secs(60)) | ||
| .map(jitter); | ||
| match notification { | ||
| LocalNotification::WorkerNodeIsDeleted(worker_node) => { | ||
| self.compactor_manager.remove_compactor(worker_node.id); | ||
| tokio_retry::RetryIf::spawn( | ||
| retry_strategy.clone(), | ||
| || async { | ||
| if let Err(err) = self.release_contexts(vec![worker_node.id]).await { | ||
| tracing::warn!( | ||
| "Failed to release hummock context {}. {}. Will retry.", | ||
| worker_node.id, | ||
| err | ||
| ); | ||
| return Err(err); | ||
| } | ||
| Ok(()) | ||
| }, | ||
| RetryableError::default(), | ||
| ) | ||
| .await | ||
| .expect("retry until success"); | ||
| tracing::info!("Released hummock context {}", worker_node.id); | ||
| sync_point!("AFTER_RELEASE_HUMMOCK_CONTEXTS_ASYNC"); | ||
| } | ||
| // TODO move `CompactionTaskNeedCancel` to `handle_hummock_manager_event` | ||
| // TODO extract retry boilerplate code | ||
| LocalNotification::CompactionTaskNeedCancel(compact_task) => { | ||
| let task_id = compact_task.task_id; | ||
| tokio_retry::RetryIf::spawn( | ||
| retry_strategy.clone(), | ||
| || async { | ||
| let mut compact_task_mut = compact_task.clone(); | ||
| if let Err(err) = self.cancel_compact_task_impl(&mut compact_task_mut).await | ||
| { | ||
| tracing::warn!( | ||
| "Failed to cancel compaction task {}. {}. Will retry.", | ||
| compact_task.task_id, | ||
| err | ||
| ); | ||
| return Err(err); | ||
| } | ||
| Ok(()) | ||
| }, | ||
| RetryableError::default(), | ||
| ) | ||
| .await | ||
| .expect("retry until success"); | ||
| tracing::info!("Cancelled compaction task {}", task_id); | ||
| sync_point!("AFTER_CANCEL_COMPACTION_TASK_ASYNC"); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| pub fn try_send_event(&self, event: HummockManagerEvent) { | ||
| if let Err(e) = self.event_sender.send(event) { | ||
| tracing::warn!("failed to send event to hummock manager {}", e); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.