Skip to content

Commit

Permalink
f Run async event processing in loop until we're done
Browse files Browse the repository at this point in the history
  • Loading branch information
tnull committed Apr 19, 2023
1 parent ab7fa89 commit c9cfd20
Showing 1 changed file with 27 additions and 22 deletions.
49 changes: 27 additions & 22 deletions lightning/src/ln/channelmanager.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5779,35 +5779,40 @@ where
pub async fn process_pending_events_async<Future: core::future::Future, H: Fn(Event) -> Future>(
&self, handler: H
) {
if self.pending_events_processor.compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed).is_err() {
return;
}
let mut processed_all_events = false;
while !processed_all_events {
if self.pending_events_processor.compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed).is_err() {
return;
}

let mut result = NotifyOption::SkipPersist;
let mut result = NotifyOption::SkipPersist;

// TODO: This behavior should be documented. It's unintuitive that we query
// ChannelMonitors when clearing other events.
if self.process_pending_monitor_events() {
result = NotifyOption::DoPersist;
}
// TODO: This behavior should be documented. It's unintuitive that we query
// ChannelMonitors when clearing other events.
if self.process_pending_monitor_events() {
result = NotifyOption::DoPersist;
}

let pending_events = self.pending_events.lock().unwrap().clone();
let num_events = pending_events.len();
if !pending_events.is_empty() {
result = NotifyOption::DoPersist;
}
let pending_events = self.pending_events.lock().unwrap().clone();
let num_events = pending_events.len();
if !pending_events.is_empty() {
result = NotifyOption::DoPersist;
}

for event in pending_events {
handler(event).await;
}
for event in pending_events {
handler(event).await;
}

self.pending_events.lock().unwrap().drain(..num_events);
let mut pending_events = self.pending_events.lock().unwrap().clone();
pending_events.drain(..num_events);
processed_all_events = pending_events.is_empty();

if result == NotifyOption::DoPersist {
self.persistence_notifier.notify();
}
if result == NotifyOption::DoPersist {
self.persistence_notifier.notify();
}

self.pending_events_processor.store(false, Ordering::Release);
self.pending_events_processor.store(false, Ordering::Release);
}
}
}

Expand Down

0 comments on commit c9cfd20

Please sign in to comment.