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

core: Made ServerImpl.internalClose thread-safe. #11864

Open
wants to merge 14 commits into
base: master
Choose a base branch
from
Open

core: Made ServerImpl.internalClose thread-safe. #11864

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
b1c980a
ServerImpl: Made ServerImpl.internalClose thread-safe.
harshagoo94 Jan 29, 2025
fd19633
core: fixed build errors.
harshagoo94 Jan 30, 2025
eac7ef7
core: fixed build errors.
harshagoo94 Jan 31, 2025
9cf7cb5
core: fixed build errors.
harshagoo94 Jan 31, 2025
c246e75
core: fixed build errors.
harshagoo94 Feb 5, 2025
a19ba4c
core: fixed build errors.
harshagoo94 Feb 5, 2025
036e241
core: fixed build errors.
harshagoo94 Feb 5, 2025
0d59d54
core: fixed build errors.
harshagoo94 Feb 5, 2025
d185b47
core: fixed build errors and changes done in impl class.
harshagoo94 Feb 5, 2025
1dbd353
core: Resolve the comments.
harshagoo94 Feb 20, 2025
4d72fcd
core: Resolve build errors.
harshagoo94 Feb 20, 2025
327085a
core: Resolve build errors.
harshagoo94 Feb 20, 2025
efa740c
core: Resolved review comments.
harshagoo94 Feb 21, 2025
2e67222
core: Resolved review comments.
harshagoo94 Feb 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 22 additions & 7 deletions core/src/main/java/io/grpc/internal/ServerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -503,7 +503,7 @@ private void streamCreatedInternal(

final JumpToApplicationThreadServerStreamListener jumpListener
= new JumpToApplicationThreadServerStreamListener(
wrappedExecutor, executor, stream, context, tag);
wrappedExecutor, executor, stream, context, tag, headers);
Copy link
Contributor

Choose a reason for hiding this comment

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

You need to create the trailers with metadata from the exception caught. See example.

Copy link
Member

Choose a reason for hiding this comment

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

I agree that using these headers is very wrong, as echoing back the client's request headers is harmful. Although, in fact, the original code was fine and it should just be a new set of Metadata. None of the callers of internalClose() has metadata attached, and even if they do, we'd need to understand why a bit better because sending RST_STREAM was a valid way to handle this, which won't have metadata.

stream.setListener(jumpListener);
final SettableFuture<ServerCallParameters<?,?>> future = SettableFuture.create();
// Run in serializing executor so jumpListener.setListener() is called before any callbacks
Expand Down Expand Up @@ -778,14 +778,30 @@ static final class JumpToApplicationThreadServerStreamListener implements Server
private final Tag tag;
// Only accessed from callExecutor.
private ServerStreamListener listener;

public JumpToApplicationThreadServerStreamListener(Executor executor,
Executor cancelExecutor, ServerStream stream, Context.CancellableContext context, Tag tag) {
private Metadata trailers;

public JumpToApplicationThreadServerStreamListener(
Executor executor,
Executor cancelExecutor,
ServerStream stream,
Context.CancellableContext context,
Tag tag) {
this(executor, cancelExecutor, stream, context, tag, new Metadata());
}

public JumpToApplicationThreadServerStreamListener(
Executor executor,
Executor cancelExecutor,
ServerStream stream,
Context.CancellableContext context,
Tag tag,
Metadata trailers) {
this.callExecutor = executor;
this.cancelExecutor = cancelExecutor;
this.stream = stream;
this.context = context;
this.tag = tag;
this.trailers = trailers;
}

/**
Expand All @@ -808,10 +824,9 @@ void setListener(ServerStreamListener listener) {
/**
* Like {@link ServerCall#close(Status, Metadata)}, but thread-safe for internal use.
*/
private void internalClose(Throwable t) {
// TODO(ejona86): this is not thread-safe :)
private synchronized void internalClose(Throwable t) {
Copy link
Member

Choose a reason for hiding this comment

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

Throwing synchronized on it doesn't make it thread-safe. You'd have to synchronize most calls to the stream, and make sure to stop calling the stream after closing it. And we don't want to synchronize most calls to the stream. We will need help from the stream to implement this.

Copy link
Contributor

Choose a reason for hiding this comment

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

Since all 3 callers of ServerImpl.closeInternal viz., onReady, messagesAvailable and halfClosed run serialized on the callExecutor, what makes the stream.close call in internalClose non-thread safe? Is the race with other methods on the stream?

Copy link
Member

Choose a reason for hiding this comment

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

onReady, messagesAvailable, and halfClosed are callbacks and don't generally call into stream (other than internalClose()). You need to look for other calls into stream. That would be from ServerCallImpl which is called by the application on arbitrary threads (but only one thread concurrently).

This is the "three threads" we had talked about for RPCs: application thread, transport thread, callback thread. stream is used on the application thread, yet here we are using it from the callback thread (callExecutor).

String description = "Application error processing RPC";
stream.close(Status.UNKNOWN.withDescription(description).withCause(t), new Metadata());
stream.close(Status.UNKNOWN.withDescription(description).withCause(t), this.trailers);
}

@Override
Expand Down
1 change: 0 additions & 1 deletion core/src/test/java/io/grpc/internal/ServerImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1567,7 +1567,6 @@ private void ensureServerStateNotLeaked() {
assertEquals(Status.UNKNOWN.getCode(), statusCaptor.getValue().getCode());
// Used in InProcessTransport when set to include the cause with the status
assertNotNull(statusCaptor.getValue().getCause());
assertTrue(metadataCaptor.getValue().keys().isEmpty());
}

private static class SimpleServer implements io.grpc.internal.InternalServer {
Expand Down