@@ -21,7 +21,7 @@ use std::{collections::HashMap, sync::Arc, time::SystemTime};
21
21
use thiserror:: Error ;
22
22
use tokio:: sync:: mpsc:: { channel, Receiver , Sender } ;
23
23
use tokio:: sync:: Mutex ;
24
- use tokio:: { select, time, time:: Duration , time:: Instant } ;
24
+ use tokio:: { select, time, time:: Duration , time:: Instant , time :: Sleep } ;
25
25
use tokio_util:: task:: TaskTracker ;
26
26
use triggered:: { Listener , Trigger } ;
27
27
@@ -1345,8 +1345,7 @@ async fn run_results_logger(
1345
1345
///
1346
1346
/// Note: this producer does not accept a shutdown trigger because it only expects to be dispatched once. In the single
1347
1347
/// producer case exit will drop the only sending channel and the receiving channel provided to the consumer will error
1348
- /// out. In the multiple-producer case, a single producer shutting down does not drop *all* sending channels so the
1349
- /// consumer will not exit and a trigger is required.
1348
+ /// out.
1350
1349
async fn produce_simulation_results (
1351
1350
nodes : HashMap < PublicKey , Arc < Mutex < dyn LightningNode > > > ,
1352
1351
mut output_receiver : Receiver < SimulationOutput > ,
@@ -1409,27 +1408,28 @@ async fn track_payment_result(
1409
1408
log:: debug!( "Tracking payment outcome for: {}." , hex:: encode( hash. 0 ) ) ;
1410
1409
1411
1410
// Trigger and listener to stop the implementation specific track payment functions (node.track_payment())
1412
- let ( stop , listen ) = triggered:: trigger ( ) ;
1411
+ let ( track_payment_trigger , track_payment_listener ) = triggered:: trigger ( ) ;
1413
1412
1414
1413
// Timer for waiting after getting the shutdown signal in order for current tracking to complete
1415
- let mut timer: Option < tokio:: time:: Sleep > = None ;
1414
+ let mut timer: Option < Sleep > = None ;
1415
+ let mut timer_started = false ;
1416
1416
1417
1417
loop {
1418
1418
tokio:: select! {
1419
- biased;
1420
1419
// The shutdown listener is triggered and we have not started a timer yet
1421
- _ = async { } , if listener. clone( ) . is_triggered( ) && timer . is_none ( ) => {
1420
+ _ = async { } , if listener. clone( ) . is_triggered( ) && !timer_started => {
1422
1421
log:: debug!( "Shutdown received by track_payment_result, starting timer..." ) ;
1423
1422
timer = Some ( time:: sleep_until( Instant :: now( ) + Duration :: from_secs( 3 ) ) ) ;
1423
+ timer_started = true ;
1424
1424
} ,
1425
1425
// The timer has been started and it expires
1426
1426
Some ( _) = conditional_sleeper( timer) => {
1427
1427
log:: error!( "Track payment failed for {}. The shutdown timer expired." , hex:: encode( hash. 0 ) ) ;
1428
- stop . trigger( ) ;
1428
+ track_payment_trigger . trigger( ) ;
1429
1429
timer = None ;
1430
1430
}
1431
1431
// The payment tracking completes
1432
- res = node. track_payment( & hash, listen . clone( ) ) => {
1432
+ res = node. track_payment( & hash, track_payment_listener . clone( ) ) => {
1433
1433
match res {
1434
1434
Ok ( res) => {
1435
1435
log:: info!(
@@ -1469,7 +1469,7 @@ async fn track_payment_result(
1469
1469
Ok ( ( ) )
1470
1470
}
1471
1471
1472
- async fn conditional_sleeper ( t : Option < tokio :: time :: Sleep > ) -> Option < ( ) > {
1472
+ async fn conditional_sleeper ( t : Option < Sleep > ) -> Option < ( ) > {
1473
1473
match t {
1474
1474
Some ( timer) => {
1475
1475
timer. await ;
0 commit comments