@@ -667,7 +667,6 @@ where
667
667
D :: Target : ' static + ChangeDestinationSource ,
668
668
K :: Target : ' static + KVStore ,
669
669
{
670
- let mut should_break = false ;
671
670
let async_event_handler = |event| {
672
671
let network_graph = gossip_sync. network_graph ( ) ;
673
672
let event_handler = & event_handler;
@@ -744,24 +743,14 @@ where
744
743
// generally, and as a fallback place such blocking only immediately before
745
744
// persistence.
746
745
peer_manager. as_ref ( ) . process_events ( ) ;
747
- if ( |fut : & mut SleepFuture | {
748
- let mut waker = dummy_waker ( ) ;
749
- let mut ctx = task:: Context :: from_waker ( & mut waker) ;
750
- match core:: pin:: Pin :: new ( fut) . poll ( & mut ctx) {
751
- task:: Poll :: Ready ( exit) => {
752
- should_break = exit;
753
- true
754
- } ,
755
- task:: Poll :: Pending => false ,
756
- }
757
- } ) ( & mut last_forwards_processing_call)
758
- {
759
- channel_manager. get_cm ( ) . process_pending_htlc_forwards ( ) ;
760
- cur_batch_delay = batch_delay. next ( ) ;
761
- last_forwards_processing_call = sleeper ( cur_batch_delay) ;
762
- }
763
- if should_break {
764
- break ;
746
+ match check_sleeper ( & mut last_forwards_processing_call) {
747
+ Some ( false ) => {
748
+ channel_manager. get_cm ( ) . process_pending_htlc_forwards ( ) ;
749
+ cur_batch_delay = batch_delay. next ( ) ;
750
+ last_forwards_processing_call = sleeper ( cur_batch_delay) ;
751
+ } ,
752
+ Some ( true ) => break ,
753
+ None => { } ,
765
754
}
766
755
767
756
// We wait up to 100ms, but track how long it takes to detect being put to sleep,
@@ -799,27 +788,21 @@ where
799
788
match fut. await {
800
789
SelectorOutput :: A | SelectorOutput :: B | SelectorOutput :: C | SelectorOutput :: D => { } ,
801
790
SelectorOutput :: E ( exit) => {
802
- should_break = exit;
791
+ if exit {
792
+ break ;
793
+ }
803
794
} ,
804
795
}
796
+
805
797
let await_slow = if mobile_interruptable_platform {
806
- ( |fut : & mut SleepFuture | {
807
- let mut waker = dummy_waker ( ) ;
808
- let mut ctx = task:: Context :: from_waker ( & mut waker) ;
809
- match core:: pin:: Pin :: new ( fut) . poll ( & mut ctx) {
810
- task:: Poll :: Ready ( exit) => {
811
- should_break = exit;
812
- true
813
- } ,
814
- task:: Poll :: Pending => false ,
815
- }
816
- } ) ( & mut await_start. unwrap ( ) )
798
+ match check_sleeper ( & mut await_start. unwrap ( ) ) {
799
+ Some ( true ) => break ,
800
+ Some ( false ) => true ,
801
+ None => false ,
802
+ }
817
803
} else {
818
804
false
819
805
} ;
820
- if should_break {
821
- break ;
822
- }
823
806
if channel_manager. get_cm ( ) . get_and_clear_needs_persistence ( ) {
824
807
log_trace ! ( logger, "Persisting ChannelManager..." ) ;
825
808
kv_store
@@ -832,39 +815,25 @@ where
832
815
. await ?;
833
816
log_trace ! ( logger, "Done persisting ChannelManager." ) ;
834
817
}
835
- if ( |fut : & mut SleepFuture | {
836
- let mut waker = dummy_waker ( ) ;
837
- let mut ctx = task:: Context :: from_waker ( & mut waker) ;
838
- match core:: pin:: Pin :: new ( fut) . poll ( & mut ctx) {
839
- task:: Poll :: Ready ( exit) => {
840
- should_break = exit;
841
- true
842
- } ,
843
- task:: Poll :: Pending => false ,
844
- }
845
- } ) ( & mut last_freshness_call)
846
- {
847
- log_trace ! ( logger, "Calling ChannelManager's timer_tick_occurred" ) ;
848
- channel_manager. get_cm ( ) . timer_tick_occurred ( ) ;
849
- last_freshness_call = sleeper ( FRESHNESS_TIMER ) ;
850
- }
851
- if ( |fut : & mut SleepFuture | {
852
- let mut waker = dummy_waker ( ) ;
853
- let mut ctx = task:: Context :: from_waker ( & mut waker) ;
854
- match core:: pin:: Pin :: new ( fut) . poll ( & mut ctx) {
855
- task:: Poll :: Ready ( exit) => {
856
- should_break = exit;
857
- true
858
- } ,
859
- task:: Poll :: Pending => false ,
860
- }
861
- } ) ( & mut last_onion_message_handler_call)
862
- {
863
- if let Some ( om) = & onion_messenger {
864
- log_trace ! ( logger, "Calling OnionMessageHandler's timer_tick_occurred" ) ;
865
- om. get_om ( ) . timer_tick_occurred ( ) ;
866
- }
867
- last_onion_message_handler_call = sleeper ( ONION_MESSAGE_HANDLER_TIMER ) ;
818
+ match check_sleeper ( & mut last_freshness_call) {
819
+ Some ( false ) => {
820
+ log_trace ! ( logger, "Calling ChannelManager's timer_tick_occurred" ) ;
821
+ channel_manager. get_cm ( ) . timer_tick_occurred ( ) ;
822
+ last_freshness_call = sleeper ( FRESHNESS_TIMER ) ;
823
+ } ,
824
+ Some ( true ) => break ,
825
+ None => { } ,
826
+ }
827
+ match check_sleeper ( & mut last_onion_message_handler_call) {
828
+ Some ( false ) => {
829
+ if let Some ( om) = & onion_messenger {
830
+ log_trace ! ( logger, "Calling OnionMessageHandler's timer_tick_occurred" ) ;
831
+ om. get_om ( ) . timer_tick_occurred ( ) ;
832
+ }
833
+ last_onion_message_handler_call = sleeper ( ONION_MESSAGE_HANDLER_TIMER ) ;
834
+ } ,
835
+ Some ( true ) => break ,
836
+ None => { } ,
868
837
}
869
838
if await_slow {
870
839
// On various platforms, we may be starved of CPU cycles for several reasons.
@@ -882,39 +851,31 @@ where
882
851
log_trace ! ( logger, "100ms sleep took more than a second, disconnecting peers." ) ;
883
852
peer_manager. as_ref ( ) . disconnect_all_peers ( ) ;
884
853
last_ping_call = sleeper ( PING_TIMER ) ;
885
- } else if ( |fut : & mut SleepFuture | {
886
- let mut waker = dummy_waker ( ) ;
887
- let mut ctx = task:: Context :: from_waker ( & mut waker) ;
888
- match core:: pin:: Pin :: new ( fut) . poll ( & mut ctx) {
889
- task:: Poll :: Ready ( exit) => {
890
- should_break = exit;
891
- true
854
+ } else {
855
+ match check_sleeper ( & mut last_ping_call) {
856
+ Some ( false ) => {
857
+ log_trace ! ( logger, "Calling PeerManager's timer_tick_occurred" ) ;
858
+ peer_manager. as_ref ( ) . timer_tick_occurred ( ) ;
859
+ last_ping_call = sleeper ( PING_TIMER ) ;
892
860
} ,
893
- task:: Poll :: Pending => false ,
861
+ Some ( true ) => break ,
862
+ _ => { } ,
894
863
}
895
- } ) ( & mut last_ping_call)
896
- {
897
- log_trace ! ( logger, "Calling PeerManager's timer_tick_occurred" ) ;
898
- peer_manager. as_ref ( ) . timer_tick_occurred ( ) ;
899
- last_ping_call = sleeper ( PING_TIMER ) ;
900
864
}
901
865
902
866
// Note that we want to run a graph prune once not long after startup before
903
867
// falling back to our usual hourly prunes. This avoids short-lived clients never
904
868
// pruning their network graph. We run once 60 seconds after startup before
905
869
// continuing our normal cadence. For RGS, since 60 seconds is likely too long,
906
870
// we prune after an initial sync completes.
907
- let prune_timer_elapsed = ( |fut : & mut SleepFuture | {
908
- let mut waker = dummy_waker ( ) ;
909
- let mut ctx = task:: Context :: from_waker ( & mut waker) ;
910
- match core:: pin:: Pin :: new ( fut) . poll ( & mut ctx) {
911
- task:: Poll :: Ready ( exit) => {
912
- should_break = exit;
913
- true
914
- } ,
915
- task:: Poll :: Pending => false ,
871
+ let prune_timer_elapsed = {
872
+ match check_sleeper ( & mut last_prune_call) {
873
+ Some ( false ) => true ,
874
+ Some ( true ) => break ,
875
+ None => false ,
916
876
}
917
- } ) ( & mut last_prune_call) ;
877
+ } ;
878
+
918
879
let should_prune = match gossip_sync {
919
880
GossipSync :: Rapid ( _) => !have_pruned || prune_timer_elapsed,
920
881
_ => prune_timer_elapsed,
@@ -957,76 +918,55 @@ where
957
918
}
958
919
have_decayed_scorer = true ;
959
920
}
960
- if ( |fut : & mut SleepFuture | {
961
- let mut waker = dummy_waker ( ) ;
962
- let mut ctx = task:: Context :: from_waker ( & mut waker) ;
963
- match core:: pin:: Pin :: new ( fut) . poll ( & mut ctx) {
964
- task:: Poll :: Ready ( exit) => {
965
- should_break = exit;
966
- true
967
- } ,
968
- task:: Poll :: Pending => false ,
969
- }
970
- } ) ( & mut last_scorer_persist_call)
971
- {
972
- if let Some ( ref scorer) = scorer {
973
- if let Some ( duration_since_epoch) = fetch_time ( ) {
974
- log_trace ! ( logger, "Calling time_passed and persisting scorer" ) ;
975
- scorer. write_lock ( ) . time_passed ( duration_since_epoch) ;
976
- } else {
977
- log_trace ! ( logger, "Persisting scorer" ) ;
921
+ match check_sleeper ( & mut last_scorer_persist_call) {
922
+ Some ( false ) => {
923
+ if let Some ( ref scorer) = scorer {
924
+ if let Some ( duration_since_epoch) = fetch_time ( ) {
925
+ log_trace ! ( logger, "Calling time_passed and persisting scorer" ) ;
926
+ scorer. write_lock ( ) . time_passed ( duration_since_epoch) ;
927
+ } else {
928
+ log_trace ! ( logger, "Persisting scorer" ) ;
929
+ }
930
+ if let Err ( e) = kv_store
931
+ . write (
932
+ SCORER_PERSISTENCE_PRIMARY_NAMESPACE ,
933
+ SCORER_PERSISTENCE_SECONDARY_NAMESPACE ,
934
+ SCORER_PERSISTENCE_KEY ,
935
+ & scorer. encode ( ) ,
936
+ )
937
+ . await
938
+ {
939
+ log_error ! (
940
+ logger,
941
+ "Error: Failed to persist scorer, check your disk and permissions {}" ,
942
+ e
943
+ ) ;
944
+ }
978
945
}
979
- if let Err ( e) = kv_store
980
- . write (
981
- SCORER_PERSISTENCE_PRIMARY_NAMESPACE ,
982
- SCORER_PERSISTENCE_SECONDARY_NAMESPACE ,
983
- SCORER_PERSISTENCE_KEY ,
984
- & scorer. encode ( ) ,
985
- )
986
- . await
987
- {
988
- log_error ! (
989
- logger,
990
- "Error: Failed to persist scorer, check your disk and permissions {}" ,
991
- e
992
- ) ;
946
+ last_scorer_persist_call = sleeper ( SCORER_PERSIST_TIMER ) ;
947
+ } ,
948
+ Some ( true ) => break ,
949
+ None => { } ,
950
+ }
951
+ match check_sleeper ( & mut last_rebroadcast_call) {
952
+ Some ( false ) => {
953
+ log_trace ! ( logger, "Rebroadcasting monitor's pending claims" ) ;
954
+ chain_monitor. rebroadcast_pending_claims ( ) ;
955
+ last_rebroadcast_call = sleeper ( REBROADCAST_TIMER ) ;
956
+ } ,
957
+ Some ( true ) => break ,
958
+ None => { } ,
959
+ }
960
+ match check_sleeper ( & mut last_sweeper_call) {
961
+ Some ( false ) => {
962
+ log_trace ! ( logger, "Regenerating sweeper spends if necessary" ) ;
963
+ if let Some ( ref sweeper) = sweeper {
964
+ let _ = sweeper. regenerate_and_broadcast_spend_if_necessary ( ) . await ;
993
965
}
994
- }
995
- last_scorer_persist_call = sleeper ( SCORER_PERSIST_TIMER ) ;
996
- }
997
- if ( |fut : & mut SleepFuture | {
998
- let mut waker = dummy_waker ( ) ;
999
- let mut ctx = task:: Context :: from_waker ( & mut waker) ;
1000
- match core:: pin:: Pin :: new ( fut) . poll ( & mut ctx) {
1001
- task:: Poll :: Ready ( exit) => {
1002
- should_break = exit;
1003
- true
1004
- } ,
1005
- task:: Poll :: Pending => false ,
1006
- }
1007
- } ) ( & mut last_rebroadcast_call)
1008
- {
1009
- log_trace ! ( logger, "Rebroadcasting monitor's pending claims" ) ;
1010
- chain_monitor. rebroadcast_pending_claims ( ) ;
1011
- last_rebroadcast_call = sleeper ( REBROADCAST_TIMER ) ;
1012
- }
1013
- if ( |fut : & mut SleepFuture | {
1014
- let mut waker = dummy_waker ( ) ;
1015
- let mut ctx = task:: Context :: from_waker ( & mut waker) ;
1016
- match core:: pin:: Pin :: new ( fut) . poll ( & mut ctx) {
1017
- task:: Poll :: Ready ( exit) => {
1018
- should_break = exit;
1019
- true
1020
- } ,
1021
- task:: Poll :: Pending => false ,
1022
- }
1023
- } ) ( & mut last_sweeper_call)
1024
- {
1025
- log_trace ! ( logger, "Regenerating sweeper spends if necessary" ) ;
1026
- if let Some ( ref sweeper) = sweeper {
1027
- let _ = sweeper. regenerate_and_broadcast_spend_if_necessary ( ) . await ;
1028
- }
1029
- last_sweeper_call = sleeper ( SWEEPER_TIMER ) ;
966
+ last_sweeper_call = sleeper ( SWEEPER_TIMER ) ;
967
+ } ,
968
+ Some ( true ) => break ,
969
+ None => { } ,
1030
970
}
1031
971
}
1032
972
log_trace ! ( logger, "Terminating background processor." ) ;
@@ -1065,6 +1005,17 @@ where
1065
1005
Ok ( ( ) )
1066
1006
}
1067
1007
1008
+ fn check_sleeper < SleepFuture : core:: future:: Future < Output = bool > + core:: marker:: Unpin > (
1009
+ fut : & mut SleepFuture ,
1010
+ ) -> Option < bool > {
1011
+ let mut waker = dummy_waker ( ) ;
1012
+ let mut ctx = task:: Context :: from_waker ( & mut waker) ;
1013
+ match core:: pin:: Pin :: new ( fut) . poll ( & mut ctx) {
1014
+ task:: Poll :: Ready ( exit) => Some ( exit) ,
1015
+ task:: Poll :: Pending => None ,
1016
+ }
1017
+ }
1018
+
1068
1019
/// Async events processor that is based on [`process_events_async`] but allows for [`KVStoreSync`] to be used for
1069
1020
/// synchronous background persistence.
1070
1021
pub async fn process_events_async_with_kv_store_sync <
0 commit comments