Issue 340: Fix event writer implementation and add flush #341

dellThejas · 2021-11-17T11:37:06Z

Signed-off-by: dellThejas [email protected]

Change log description
Added flush for event writer
Updated implementation of Event Writer

Purpose of the change
#340

What the code does
Flush implementation added to event writer similar to Byte writer
Also, event writer will not be return the receiver anymore. This implementation helps setup the flush, and makes the write operation easier to implement for users.

How to verify it
ran existing tests

src/event/writer.rs

codecov-commenter · 2021-11-29T21:08:37Z

Codecov Report

Merging #341 (d64601f) into master (45aa5dc) will increase coverage by 7.42%.
The diff coverage is 86.36%.

@@            Coverage Diff             @@
##           master     #341      +/-   ##
==========================================
+ Coverage   67.43%   74.86%   +7.42%     
==========================================
  Files          26       46      +20     
  Lines        6243    11671    +5428     
==========================================
+ Hits         4210     8737    +4527     
- Misses       2033     2934     +901

Impacted Files	Coverage Δ
src/event/writer.rs	`100.00% <ø> (ø)`
src/segment/writer.rs	`83.84% <50.00%> (ø)`
src/segment/event.rs	`52.94% <94.11%> (ø)`
src/segment/reactor.rs	`69.49% <100.00%> (ø)`
controller-client/src/model_helper.rs	`55.81% <0.00%> (-9.06%)`	⬇️
wire_protocol/src/wire_commands.rs	`90.46% <0.00%> (-7.63%)`	⬇️
wire_protocol/src/commands.rs	`79.67% <0.00%> (-7.42%)`	⬇️
controller-client/src/lib.rs	`55.50% <0.00%> (-5.46%)`	⬇️
connection_pool/src/connection_pool.rs	`83.33% <0.00%> (-1.76%)`	⬇️
wire_protocol/src/client_connection.rs	`91.39% <0.00%> (-1.46%)`	⬇️
... and 32 more

shrids

The changes are looking good. I have a couple of comments/questions.

shrids · 2021-12-01T06:02:49Z

src/event/writer.rs

        if let Some(pending_event) = PendingEvent::with_header(routing_info, event, None, tx) {
            let append_event = Incoming::AppendEvent(pending_event);
-            self.writer_event_internal(append_event, size, rx).await
+            let write_event = self.writer_event_internal(append_event, size, rx).await;
+            if let Some(pending_event_flush) =


Should we handle the else part here?
similar comment for the method above...

Added a panic statement

src/event/writer.rs

shrids · 2021-12-01T10:03:12Z

src/event/writer.rs

+                let flush_rec = self.writer_event_internal(append_event_flush, 0, rx_flush).await;
+                self.event_handles.push_back(flush_rec);
+            } else {
+                panic!("Failed to track event in flush.");


@tkaitchuck is there a better way to solve this ?

.expect()

tkaitchuck · 2021-12-03T01:26:27Z

src/event/writer.rs

        if let Some(pending_event) = PendingEvent::with_header(routing_info, event, None, tx) {
            let append_event = Incoming::AppendEvent(pending_event);
-            self.writer_event_internal(append_event, size, rx).await
+            let write_event = self.writer_event_internal(append_event, size, rx).await;


Instead of calling writer_event_internal here, it might be better to modify it to have a single call which takes two rxes. (IE embedding both txes into the same PendingEvent.)

This would avoid a bug which is currently in this code which is that it is passing None as the routing key for both the rx_flush and the rx which implies that it is legal for these to be sent to different segments.

I added an additional optional Sender within PendingEvent, and write_event_internal now is only called once to track flush activity as well as write. This should eliminate our concerns dealing with different routing keys, and different segments.

tkaitchuck · 2021-12-03T01:27:49Z

src/event/writer.rs

+                let flush_rec = self.writer_event_internal(append_event_flush, 0, rx_flush).await;
+                self.event_handles.push_back(flush_rec);
+            } else {
+                panic!("Failed to track event in flush.");


.expect()

tkaitchuck · 2021-12-03T01:29:04Z

src/event/writer.rs

        let routing_info = RoutingInfo::RoutingKey(Some(routing_key));
+        let routing_info_flush = RoutingInfo::RoutingKey(None);


This will route the flush event differently from the event it is supposed to be following. This could result in flush returning even when the data has not yet been written. Please add a test that covers this case.

Once we embedd the tx_flush within PendingEvent, we wont need to have a separate Routing Key.

src/event/writer.rs

Signed-off-by: dellThejas <[email protected]>

tkaitchuck · 2022-01-04T00:21:36Z

src/event/writer.rs

+    /// ```
+    pub async fn flush(&mut self) -> Result<(), Error> {
+        while self.event_handles.front().is_some() {
+            let mut receiver = self.event_handles.swap_remove_front(0).expect("get first handle");


I think pop_front() would be preferable to swap_remove_front(0) here.
This could also be combined with the conditional as in:

while let Some(receiver) = self.event_handles.front()

To be more efficient, clear, and concise.

tkaitchuck · 2022-01-04T00:25:32Z

src/event/writer.rs

+    /// It will wait until all pending appends have acknowledgment.
+    ///
+    /// # Examples
+    /// ```ignore


Remove the ignore. This example is broken.

tkaitchuck · 2022-01-04T00:25:48Z

src/event/writer.rs

+    /// ```ignore
+    /// let mut byte_writer = client_factory.create_byte_writer(segment).await;
+    /// let payload = vec![0; 8];
+    /// let size = event_writer.write_event(&payload).await;


A size is not returned here.

tkaitchuck · 2022-01-04T00:29:37Z

src/event/writer.rs

+                _ => {
+                    try_recv.map_err(|e| Error::InternalFailure {
+                        msg: format!("oneshot error {:?}", e),
+                    })??;


is there an await missing here?
It looks like this method will just return without actually waiting on anything.
Also instead of _ it's probably worth explicitly having an Ok case as this branch is currently handling both all unlisted errors and the success case.

Issue here was that I wasnt sending anything to tx_flush sender. Now, I am so we wont have this issue.

tkaitchuck · 2022-01-04T00:30:21Z

src/event/writer.rs

@@ -161,9 +172,39 @@ impl EventWriter {
                .expect("send error");
            rx_error
        } else {
+            self.event_handles.push_back(rx_flush);


We need to support the case where the caller never calls flush without leaking memory. Currently the flush call removes items from event_handles but if it is not called this code will OOM.

Do we need to periodically call flush to clear this? Or upon every write, call flush on all previous writes? Or clean up event_handles while doing every write? (It would need to go through every write call to check if its processed or not)

Yes. On a write call, clear out any completed ones stopping at the first uncompleted one.

shrids · 2022-01-04T10:36:54Z

src/segment/event.rs

@@ -50,6 +50,7 @@ pub(crate) struct PendingEvent {
    pub(crate) data: Vec<u8>,
    pub(crate) conditional_offset: Option<i64>,
    pub(crate) oneshot_sender: oneshot::Sender<Result<(), Error>>,
+    pub(crate) flush_oneshot_sender: Option<oneshot::Sender<Result<(), Error>>>,


flush_oneshot_sender is never used anywhere; we would need to invoke a send() with the write status on it.

Signed-off-by: dellThejas <[email protected]>

shrids · 2022-01-06T10:45:59Z

src/event/writer.rs

+    /// It will wait until all pending appends have acknowledgment.
+    /// ```
+    pub async fn flush(&mut self) -> Result<(), Error> {
+        self.clear_initial_complete_events();


is this required here ?

shrids · 2022-01-06T11:01:15Z

src/event/writer.rs

+                    self.event_handles.push_front(receiver);
+                    break;
+                }
+                Err(TryRecvError::Closed) => {}


I think we should return an error here and fail fast.

tkaitchuck · 2022-01-07T00:06:36Z

src/event/writer.rs

+                    break;
+                }
+                Err(TryRecvError::Closed) => {}
+                _ => {}


Please make this:

Ok(_) => {}

That way if a new error value is ever added to try_receive we'll get a compile error rather than being unaware of it and simply skipping over the error at runtime.

tkaitchuck · 2022-01-07T00:52:58Z

src/segment/writer.rs

@@ -377,6 +377,14 @@ impl SegmentWriter {
                    acked.event_id
                );
            }
+            if let Some(flush_sender) = acked.event.flush_oneshot_sender {
+                if flush_sender.send(Result::Ok(())).is_err() {
+                    trace!(


This should probably be info.

Signed-off-by: dellThejas <[email protected]>

shrids

LGTM.

dellThejas requested a review from tkaitchuck November 17, 2021 11:37

dellThejas force-pushed the large-event-flush branch from 8dd54dc to 1456b1e Compare November 22, 2021 21:23

shrids reviewed Nov 26, 2021

View reviewed changes

src/event/writer.rs Outdated Show resolved Hide resolved

dellThejas requested a review from shrids November 29, 2021 23:47

shrids changed the title ~~fix event writer implementation and add flush~~ Issue 340: Fix event writer implementation and add flush Dec 1, 2021

shrids reviewed Dec 1, 2021

View reviewed changes

dellThejas requested a review from shrids December 1, 2021 08:47

shrids approved these changes Dec 1, 2021

View reviewed changes

tkaitchuck requested changes Dec 3, 2021

View reviewed changes

dellThejas requested a review from tkaitchuck December 15, 2021 12:25

tkaitchuck reviewed Dec 16, 2021

View reviewed changes

src/event/writer.rs Outdated Show resolved Hide resolved

src/event/writer.rs Show resolved Hide resolved

src/event/writer.rs Outdated Show resolved Hide resolved

src/event/writer.rs Outdated Show resolved Hide resolved

dellThejas requested a review from shrids December 16, 2021 05:02

dellThejas added 9 commits January 3, 2022 15:54

add flush for event writer

14b64ae

Signed-off-by: dellThejas <[email protected]>

make sure flush events dont have header

7d4e4dc

Signed-off-by: dellThejas <[email protected]>

updated order of write and flush

500a88c

Signed-off-by: dellThejas <[email protected]>

add panic when failed to track event in flush

ad3c48e

Signed-off-by: dellThejas <[email protected]>

embedd flush sender within Pending event

0d18eb9

Signed-off-by: dellThejas <[email protected]>

update format

dfd70b0

Signed-off-by: dellThejas <[email protected]>

ignore receiver error upon calling flush

c2219ef

Signed-off-by: dellThejas <[email protected]>

cleanup

d228d36

Signed-off-by: dellThejas <[email protected]>

removed flush_cleared and updated flush functionality

61c7df3

Signed-off-by: dellThejas <[email protected]>

dellThejas force-pushed the large-event-flush branch from 4ca2e97 to 61c7df3 Compare January 4, 2022 00:03

tkaitchuck requested changes Jan 4, 2022

View reviewed changes

shrids requested changes Jan 4, 2022

View reviewed changes

send empty event to flush call in same segment

92a712a

Signed-off-by: dellThejas <[email protected]>

dellThejas requested a review from tkaitchuck January 5, 2022 16:07

clear completed events in beginning of queue

d64601f

Signed-off-by: dellThejas <[email protected]>

dellThejas requested a review from shrids January 6, 2022 09:06

shrids reviewed Jan 6, 2022

View reviewed changes

tkaitchuck approved these changes Jan 7, 2022

View reviewed changes

throw error fast if trying to flush closed channels

117724b

Signed-off-by: dellThejas <[email protected]>

dellThejas requested a review from shrids January 7, 2022 11:02

shrids approved these changes Jan 7, 2022

View reviewed changes

tkaitchuck approved these changes Jan 7, 2022

View reviewed changes

tkaitchuck merged commit 6f67ec1 into pravega:master Jan 7, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Issue 340: Fix event writer implementation and add flush #341

Issue 340: Fix event writer implementation and add flush #341

dellThejas commented Nov 17, 2021 •

edited

Loading

codecov-commenter commented Nov 29, 2021 •

edited

Loading

shrids left a comment

shrids Dec 1, 2021 •

edited

Loading

dellThejas Dec 1, 2021

shrids Dec 1, 2021

tkaitchuck Dec 3, 2021

tkaitchuck Dec 3, 2021

dellThejas Dec 15, 2021

tkaitchuck Dec 3, 2021

tkaitchuck Dec 3, 2021

dellThejas Dec 15, 2021

tkaitchuck Jan 4, 2022

tkaitchuck Jan 4, 2022

tkaitchuck Jan 4, 2022

tkaitchuck Jan 4, 2022

dellThejas Jan 5, 2022

tkaitchuck Jan 4, 2022

dellThejas Jan 4, 2022

tkaitchuck Jan 5, 2022

shrids Jan 4, 2022

shrids Jan 6, 2022

shrids Jan 6, 2022

tkaitchuck Jan 7, 2022

tkaitchuck Jan 7, 2022

shrids left a comment

		let routing_info = RoutingInfo::RoutingKey(Some(routing_key));
		let routing_info_flush = RoutingInfo::RoutingKey(None);

Issue 340: Fix event writer implementation and add flush #341

Issue 340: Fix event writer implementation and add flush #341

Conversation

dellThejas commented Nov 17, 2021 • edited Loading

codecov-commenter commented Nov 29, 2021 • edited Loading

Codecov Report

shrids left a comment

Choose a reason for hiding this comment

shrids Dec 1, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shrids left a comment

Choose a reason for hiding this comment

dellThejas commented Nov 17, 2021 •

edited

Loading

codecov-commenter commented Nov 29, 2021 •

edited

Loading

shrids Dec 1, 2021 •

edited

Loading