@@ -51,28 +51,20 @@ crate::di_service!(SchedulerImpl, []);
5151
5252struct LastNodeActionTimes {
5353 in_memory_compaction : SystemTime ,
54- deleted_chunks_release : SystemTime ,
5554}
5655
5756impl LastNodeActionTimes {
5857 pub fn new ( ) -> Self {
5958 Self {
6059 in_memory_compaction : SystemTime :: now ( ) ,
61- deleted_chunks_release : SystemTime :: now ( ) ,
6260 }
6361 }
6462 pub fn in_memory_compaction ( & self ) -> SystemTime {
6563 self . in_memory_compaction . clone ( )
6664 }
67- pub fn deleted_chunks_release ( & self ) -> SystemTime {
68- self . deleted_chunks_release . clone ( )
69- }
7065 pub fn set_in_memory_compaction ( & mut self , time : SystemTime ) {
7166 self . in_memory_compaction = time;
7267 }
73- pub fn set_deleted_chunks_release ( & mut self , time : SystemTime ) {
74- self . deleted_chunks_release = time;
75- }
7668}
7769
7870impl SchedulerImpl {
@@ -758,7 +750,16 @@ impl SchedulerImpl {
758750 tokio:: fs:: remove_file ( file) . await ?;
759751 }
760752 if let MetaStoreEvent :: DeleteChunk ( chunk) = & event {
761- if !chunk. get_row ( ) . in_memory ( ) && chunk. get_row ( ) . uploaded ( ) {
753+ if chunk. get_row ( ) . in_memory ( ) {
754+ let partition = self
755+ . meta_store
756+ . get_partition ( chunk. get_row ( ) . get_partition_id ( ) )
757+ . await ?;
758+ let node_name = self . cluster . node_name_by_partition ( & partition) ;
759+ self . cluster
760+ . free_memory_chunk ( & node_name, chunk. get_id ( ) )
761+ . await ?;
762+ } else if chunk. get_row ( ) . uploaded ( ) {
762763 let file_name =
763764 ChunkStore :: chunk_remote_path ( chunk. get_id ( ) , chunk. get_row ( ) . suffix ( ) ) ;
764765 let deadline = Instant :: now ( )
@@ -898,25 +899,6 @@ impl SchedulerImpl {
898899 self . process_active_chunks ( active_chunks) . await ?;
899900 self . process_inactive_chunks ( inactive_chunks) . await ?;
900901 }
901- let mut node_last_actions = self . node_last_actions . lock ( ) . await ;
902- for ( node, last_action) in node_last_actions. iter_mut ( ) {
903- if last_action
904- . deleted_chunks_release ( )
905- . elapsed ( )
906- . ok ( )
907- . map_or ( false , |d| d >= Duration :: from_secs ( 2 ) )
908- {
909- if let Err ( e) = self . cluster . free_deleted_memory_chunks ( & node) . await {
910- log:: error!(
911- "Error while trying release in memory chunks in node {}: {}" ,
912- node,
913- e
914- ) ;
915- } else {
916- last_action. set_deleted_chunks_release ( SystemTime :: now ( ) ) ;
917- }
918- }
919- }
920902
921903 Ok ( ( ) )
922904 }
0 commit comments