Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

chore: Incorporate Proto Definition Changes For Reduce #52

Merged
merged 40 commits into from
Jun 29, 2024
Merged

chore: Incorporate Proto Definition Changes For Reduce #52

merged 40 commits into from
Jun 29, 2024

Conversation

yhl25
Copy link
Contributor

@yhl25 yhl25 commented May 29, 2024 •
edited
Loading

fixes #46

@yhl25 yhl25 marked this pull request as ready for review May 30, 2024 05:07
@yhl25 yhl25 requested a review from vigith May 30, 2024 05:07
@yhl25
proto
c240e0c 
Signed-off-by: Yashash H L <[email protected]>
@yhl25 yhl25 requested a review from BulkBeing May 30, 2024 05:07
yhl25 and others added 10 commits May 30, 2024 14:31
@yhl25
resolve conflicts
3c84c40 
Signed-off-by: Yashash H L <[email protected]>
@vigith
chore: give task ownership
991b541 
Signed-off-by: Vigith Maurice <[email protected]>
@yhl25
return error
888a12f 
Signed-off-by: Yashash H L <[email protected]>
@yhl25
Merge branch 'fix-reduce' of github.com:numaproj/numaflow-rs into fix…
9989e5b 
…-reduce
@yhl25
remove from map
aa3da61 
Signed-off-by: Yashash H L <[email protected]>
@yhl25
use drain
eeac6bd 
Signed-off-by: Yashash H L <[email protected]>
@vigith
fix: some minor opinions
3e2ca4c 
Signed-off-by: Vigith Maurice <[email protected]>
@yhl25
error handling
0313680 
Signed-off-by: Yashash H L <[email protected]>
@yhl25
fixes
786a72e 
Signed-off-by: Yashash H L <[email protected]>
@yhl25
fixes
5e9763d 
Signed-off-by: Yashash H L <[email protected]>
@yhl25 yhl25 marked this pull request as draft June 3, 2024 15:06
@vigith
chore: review and FIXME
75534f1 
Signed-off-by: Vigith Maurice <[email protected]>
@yhl25
refactor
9a9e622 
Signed-off-by: Yashash H L <[email protected]>
@yhl25
cancel tasks across streams
506a38b 
Signed-off-by: Yashash H L <[email protected]>
@yhl25
update example
9e63aac 
Signed-off-by: Yashash H L <[email protected]>
vigith
vigith reviewed Jun 25, 2024
View reviewed changes
src/error.rs Outdated
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thiserror will make life much easier here

src/map.rs
let map_svc = MapService { handler };

// Create a channel to send shutdown signal to the server to do graceful shutdown in case of non retryable errors.
let (internal_shutdown_tx, internal_shutdown_rx) = mpsc::channel(1);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I forget, why not oneshot?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to store that as a variable inside the ReduceService struct for it to be used inside the reduce_fn function. Since oneshot is not cloneable, I had to opt for the mpsc method.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please add a comment there?

src/shared.rs
use tracing::info;

// #[tracing::instrument(skip(path), fields(path = ?path.as_ref()))]
#[tracing::instrument(fields(path = ?path.as_ref()))]
#[tracing::instrument(fields(path = ? path.as_ref()))]
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
#[tracing::instrument(fields(path = ? path.as_ref()))]
#[tracing::instrument(fields(path = ?path.as_ref()))]

src/shared.rs Outdated
Comment on lines 78 to 86
let custom1 = async {
internal_rx.recv().await;
};

let custom2 = async {
if let Some(rx) = user_rx {
rx.await.ok();
}
};
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what is this?

src/shared.rs Outdated
Comment on lines 163 to 166
assert!(contents.contains("\"language\":\"rust\""));
assert!(contents.contains("\"version\":\"0.0.1\""));
assert!(contents.contains("\"metadata\":{}"));
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

use r#""#?

yhl25 and others added 8 commits June 25, 2024 20:40
@yhl25
review comments
66a0519 
Signed-off-by: Yashash H L <[email protected]>
@vigith
doc: on errors
37282a8 
Signed-off-by: Vigith Maurice <[email protected]>
@vigith
fix: comment
d016dd2 
Signed-off-by: Vigith Maurice <[email protected]>
@yhl25
update test
e88304d 
Signed-off-by: Yashash H L <[email protected]>
@yhl25
avoid creating socket file manually
bf59771 
Signed-off-by: Yashash H L <[email protected]>
@yhl25
remove socket file while shutting down
98a3b7f 
Signed-off-by: Yashash H L <[email protected]>
@yhl25
remove socket file
bd0a7b2 
Signed-off-by: Yashash H L <[email protected]>
@yhl25
fix fmt
bdaace7 
Signed-off-by: Yashash H L <[email protected]>
@yhl25 yhl25 marked this pull request as ready for review June 27, 2024 06:52
@yhl25 yhl25 requested a review from vigith June 27, 2024 06:53
vigith
vigith reviewed Jun 27, 2024
View reviewed changes
src/sourcetransform.rs Outdated
Comment on lines 340 to 344
.await?;

// cleanup the socket file after the server is shutdown
// UnixListener doesn't implement Drop trait, so we have to manually remove the socket file
let _ = fs::remove_file(&self.sock_addr);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we will not remove the file if we short-circuit using ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for catching this, totally missed that case.

@yhl25
Copy link
Contributor Author

yhl25 commented Jun 29, 2024

I will create a follow-up PR to review error handling in services, excluding 'reduce'. Currently, if the user code has a panic, it is not being propagated back in 'map'.

@yhl25 yhl25 requested a review from vigith June 29, 2024 13:30
@yhl25 yhl25 merged commit 7dff517 into main Jun 29, 2024
2 checks passed
@yhl25 yhl25 deleted the fix-reduce branch June 29, 2024 14:38
@vigith vigith mentioned this pull request Jun 29, 2024
Closed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@vigith vigith vigith approved these changes

@BulkBeing BulkBeing Awaiting requested review from BulkBeing

Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.

Incorporate Proto Definition Changes For Reduce
2 participants
@yhl25 @vigith