Replace event channel with broadcast channel #5478

link2xt · 2024-04-19T02:41:56Z

Purpose of this PR is to make emitter events immediately available to TestContext "event tracker". Before this PR "event tracker" relied on this task copying events from EventEmitter to "event tracker" and LogSink:

deltachat-core-rust/src/test_utils.rs

Lines 367 to 376 in 65822e5

    
           task::spawn(async move { 
        
               while let Some(event) = events.recv().await { 
        
                   for sender in senders.read().await.iter() { 
        
                       // Don't block because someone wanted to use a oneshot receiver, use 
        
                       // an unbounded channel if you want all events. 
        
                       sender.try_send(event.clone()).ok(); 
        
                   } 
        
                   log_sender.try_send(LogEvent::Event(event.clone())).ok(); 
        
               } 
        
           });

The problem with current approach is that events are not available to event tracker immediately after emitting them because this task may not be waken up immediately after emitting the event. This makes clear_events unreliable as it only consumes events that have already been moved from event emitter to event tracker.

This PR replaces event channel with broadcast channel, so it is possible to have primary EventEmitter and "event tracker" receiving events immediately. LogSink now also gets its own broadcast receiver for each account and runs a task moving events from receivers to the log channel. This may result in reordering of events in the logs, but it is not as critical as "event tracker" not receiving events immediately as it does not break tests.

Related comment:
#5471 (comment)

There is another PR solving the problem of clearing events in event tracker in Python by using special "checkpoint" events: #5477
In Python current approach from this PR is not possible because events are polled via get_next_event call which is not guaranteed to return result immediately as event is available.

link2xt · 2024-04-20T10:31:27Z

src/events.rs

@@ -11,8 +10,11 @@ pub use self::payload::EventType;
 /// Event channel.
 #[derive(Debug, Clone)]
 pub struct Events {
-    receiver: Receiver<Event>,
-    sender: Sender<Event>,
+    /// Unused receiver to prevent the channel from closing.


This is ugly, but removing it actually makes a lot of tests fail.

I have deactivated the receiver. This is a documented use of InactiveReceiver, so I guess this is how this is supposed to be done:
https://docs.rs/async-broadcast/0.7.0/async_broadcast/struct.InactiveReceiver.html

link2xt · 2024-04-20T10:45:51Z

src/events.rs

-#[pin_project]
-pub struct EventEmitter(#[pin] Receiver<Event>);
+#[derive(Debug)]
+pub struct EventEmitter(Mutex<async_broadcast::Receiver<Event>>);


This Mutex is needed because recv of broadcast channel takes &mut self instead of &self like with a normal channel. This is the same for async-broadcast crate I used and tokio broadcast channels. I think the idea is that there is no need for synchronization if receiver is only used by a single thread as in broadcast channels each receiver reads from its own chunk of memory, but in our case we expose event emitter to FFI and otherwise allow using it from multiple threads, so we want thread-safety and have to add our own synchronization.

iequidoo · 2024-04-21T01:08:04Z

src/events.rs

+    ///
+    /// Returns `None` if no events are available for reception.
+    pub async fn try_recv(&self) -> Option<Event> {
+        let mut lock = self.0.lock().await;


So, EventEmitter::recv() can be called in parallel by multiple tasks, try_recv() as well, but together they can't be called in parallel as try_recv() would wait for recv(). How critical it is? Afaiu not at all as each task should use its own emitter?

You mean if we call recv() and it starts waiting, attempt to call try_recv() will lock until recv() returns? The other way round we cannot easily deadlock as try_recv() returns immediately.

This is actually a bug, at least needs a comment above recv that it may lock out try_recv if there is no easy way to solve it.

https://docs.rs/async-broadcast/0.7.0/async_broadcast/index.html describes broadcast channel as "MPMC". But it accepts &mut self in both https://docs.rs/async-broadcast/0.7.0/async_broadcast/struct.Receiver.html#method.recv and https://docs.rs/async-broadcast/0.7.0/async_broadcast/struct.Receiver.html#method.try_recv

So it is not MPMC in the same sense as ordinary channel is, multiple "consumers" cannot read from the same receiver at the same time. I filed upstream issue: smol-rs/async-broadcast#57

So it is not MPMC in the same sense as ordinary channel is, multiple "consumers" cannot read from the same receiver at the same time. Maybe this should be filed as a bug to async-broadcast?

I think they meant that multiple consumers should use multiple receivers. But then yes, it's not MPMC because those receivers don't "steal" messages from each other, but receive message copies.

This is actually a bug, at least needs a comment above recv that it may lock out try_recv if there is no easy way to solve it.

Seems that adding some flag like is_empty: RwLock<bool> (set by recv()) solves this, but then try_recv() can miss events if the flag isn't yet reset, but new events have already arrived.

EDIT: But probably it's ok to miss events if there are other consumers working in parallel.

Even bool isn't needed, just RwLock<()>. recv() should do try_recv() first which takes a read lock (i.e. calls try_read()) and if try_recv() didn't succeed, take a write lock

I have made try_recv() use try_lock on the mutex. So try_recv may return an error if there are concurrent calls to recv or try_recv, but we don't use it from concurrent threads so it is fine. It is now documented that it may return an error and good thing is that try_recv is non-async again.

src/test_utils.rs

iequidoo · 2024-04-22T05:11:46Z

src/events.rs

+    /// Tries to receive an event without blocking.
+    ///
+    /// Returns error if no events are available for reception
+    /// or if receiver is blocked by a concurrent call to [`recv`].


Or a concurrent call to try_recv() i guess. If it's done from another os thread

This makes `EventTracker` receive events immediately instead of being moved from event emitter to event tracker by a task spawned from `TestContext::new_internal`. This makes `EventTracker.clear_events` reliable as it is guaranteed to remove all events emitted by the time it is called rather than only events that have been moved already.

link2xt force-pushed the link2xt/broadcast-event-channel branch 11 times, most recently from 6d0725b to c86528c Compare April 20, 2024 07:40

link2xt changed the title ~~WIP: Replace event channel with broadcast channel~~ Replace event channel with broadcast channel Apr 20, 2024

link2xt force-pushed the link2xt/broadcast-event-channel branch 4 times, most recently from 2277ddc to 3853696 Compare April 20, 2024 10:07

link2xt marked this pull request as ready for review April 20, 2024 10:08

link2xt force-pushed the link2xt/broadcast-event-channel branch 3 times, most recently from 6836aab to 4928dc2 Compare April 20, 2024 10:25

link2xt requested review from iequidoo and Simon-Laux April 20, 2024 10:25

link2xt commented Apr 20, 2024

View reviewed changes

link2xt force-pushed the link2xt/broadcast-event-channel branch from 4928dc2 to aca11ab Compare April 20, 2024 10:36

link2xt mentioned this pull request Apr 20, 2024

test: fix flaky chatlist_events test test_update_after_ephemeral_messages #5471

Merged

link2xt commented Apr 20, 2024

View reviewed changes

iequidoo reviewed Apr 21, 2024

View reviewed changes

link2xt force-pushed the link2xt/broadcast-event-channel branch from 163959d to 2b6c35b Compare April 21, 2024 02:30

api!: remove Stream implementation for EventEmitter

5f82ba5

link2xt force-pushed the link2xt/broadcast-event-channel branch from 2b6c35b to d7a16c4 Compare April 21, 2024 02:33

link2xt force-pushed the link2xt/broadcast-event-channel branch from 1357019 to 2c0592a Compare April 21, 2024 20:21

link2xt mentioned this pull request Apr 21, 2024

Receiver is not multi-consumer smol-rs/async-broadcast#57

Closed

iequidoo approved these changes Apr 22, 2024

View reviewed changes

link2xt force-pushed the link2xt/broadcast-event-channel branch from 2c0592a to 5241ed6 Compare April 22, 2024 06:24

link2xt merged commit 34f4ec0 into main Apr 22, 2024
38 checks passed

link2xt deleted the link2xt/broadcast-event-channel branch April 22, 2024 07:44

link2xt mentioned this pull request May 20, 2024

fix: ignore event channel overflows #5605

Merged

link2xt mentioned this pull request May 30, 2024

test: fix logging of TestContext created using TestContext::new_alice() #5641

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Replace event channel with broadcast channel #5478

Replace event channel with broadcast channel #5478

link2xt commented Apr 19, 2024 •

edited

Loading

link2xt Apr 20, 2024

link2xt Apr 20, 2024

link2xt Apr 20, 2024 •

edited

Loading

iequidoo Apr 21, 2024

link2xt Apr 21, 2024

link2xt Apr 21, 2024 •

edited

Loading

iequidoo Apr 21, 2024 •

edited

Loading

iequidoo Apr 21, 2024 •

edited

Loading

link2xt Apr 21, 2024

iequidoo Apr 22, 2024

	task::spawn(async move {
	while let Some(event) = events.recv().await {
	for sender in senders.read().await.iter() {
	// Don't block because someone wanted to use a oneshot receiver, use
	// an unbounded channel if you want all events.
	sender.try_send(event.clone()).ok();
	}
	log_sender.try_send(LogEvent::Event(event.clone())).ok();
	}
	});

Replace event channel with broadcast channel #5478

Replace event channel with broadcast channel #5478

Conversation

link2xt commented Apr 19, 2024 • edited Loading

link2xt Apr 20, 2024

Choose a reason for hiding this comment

link2xt Apr 20, 2024

Choose a reason for hiding this comment

link2xt Apr 20, 2024 • edited Loading

Choose a reason for hiding this comment

iequidoo Apr 21, 2024

Choose a reason for hiding this comment

link2xt Apr 21, 2024

Choose a reason for hiding this comment

link2xt Apr 21, 2024 • edited Loading

Choose a reason for hiding this comment

iequidoo Apr 21, 2024 • edited Loading

Choose a reason for hiding this comment

iequidoo Apr 21, 2024 • edited Loading

Choose a reason for hiding this comment

link2xt Apr 21, 2024

Choose a reason for hiding this comment

iequidoo Apr 22, 2024

Choose a reason for hiding this comment

link2xt commented Apr 19, 2024 •

edited

Loading

link2xt Apr 20, 2024 •

edited

Loading

link2xt Apr 21, 2024 •

edited

Loading

iequidoo Apr 21, 2024 •

edited

Loading

iequidoo Apr 21, 2024 •

edited

Loading