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

okhttp: forceful close after MAX_CONNECTION_AGE_GRACE_TIME #9968

Merged
merged 5 commits into from
Mar 27, 2023
Merged

okhttp: forceful close after MAX_CONNECTION_AGE_GRACE_TIME #9968

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
08df244
Merge branch 'master' of https://github.com/grpc/grpc-java into okhtt…
YifeiZhuang Mar 20, 2023
f6af33f
okhttp: forceful close after MAX_CONNECTION_AGE_GRACE_TIME
YifeiZhuang Mar 21, 2023
e3c5690
graceful shutdown
YifeiZhuang Mar 21, 2023
648c59e
Merge branch 'master' of https://github.com/grpc/grpc-java into okhtt…
YifeiZhuang Mar 23, 2023
6486b86
server.shutdown should not trigger forcefulclosure
YifeiZhuang Mar 27, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
19 changes: 14 additions & 5 deletions okhttp/src/main/java/io/grpc/okhttp/OkHttpServerTransport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
import okio.Buffer;
import okio.BufferedSource;
Expand All @@ -73,6 +74,9 @@ final class OkHttpServerTransport implements ServerTransport,
ExceptionHandlingFrameWriter.TransportExceptionHandler, OutboundFlowController.Transport {
private static final Logger log = Logger.getLogger(OkHttpServerTransport.class.getName());
private static final int GRACEFUL_SHUTDOWN_PING = 0x1111;

private static final long GRACEFUL_SHUTDOWN_PING_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(1);

private static final int KEEPALIVE_PING = 0xDEAD;
private static final ByteString HTTP_METHOD = ByteString.encodeUtf8(":method");
private static final ByteString CONNECT_METHOD = ByteString.encodeUtf8("CONNECT");
Expand Down Expand Up @@ -250,10 +254,10 @@ public void data(boolean outFinished, int streamId, Buffer source, int byteCount

@Override
public void shutdown() {
shutdown(TimeUnit.SECONDS.toNanos(1L));
shutdown(null);
}

private void shutdown(Long graceTimeInNanos) {
private void shutdown(@Nullable Long graceTimeInNanos) {
synchronized (lock) {
if (gracefulShutdown || abruptShutdown) {
return;
Expand All @@ -267,15 +271,16 @@ private void shutdown(Long graceTimeInNanos) {
// we also set a timer to limit the upper bound in case the PING is excessively stalled or
// the client is malicious.
secondGoawayTimer = scheduledExecutorService.schedule(
this::triggerGracefulSecondGoaway, graceTimeInNanos, TimeUnit.NANOSECONDS);
() -> triggerGracefulSecondGoaway(graceTimeInNanos),
GRACEFUL_SHUTDOWN_PING_TIMEOUT_NANOS, TimeUnit.NANOSECONDS);
frameWriter.goAway(Integer.MAX_VALUE, ErrorCode.NO_ERROR, new byte[0]);
frameWriter.ping(false, 0, GRACEFUL_SHUTDOWN_PING);
frameWriter.flush();
}
}
}

private void triggerGracefulSecondGoaway() {
private void triggerGracefulSecondGoaway(@Nullable Long gracePeriodNanos) {
synchronized (lock) {
if (secondGoawayTimer == null) {
return;
Expand All @@ -289,6 +294,10 @@ private void triggerGracefulSecondGoaway() {
} else {
frameWriter.flush();
}
if (gracePeriodNanos != null) {
forcefulCloseTimer = scheduledExecutorService.schedule(
this::triggerForcefulClose, gracePeriodNanos, TimeUnit.NANOSECONDS);
}
}
}

Expand Down Expand Up @@ -926,7 +935,7 @@ public void ping(boolean ack, int payload1, int payload2) {
return;
}
if (GRACEFUL_SHUTDOWN_PING == payload) {
triggerGracefulSecondGoaway();
triggerGracefulSecondGoaway(null);
Copy link
Member

Choose a reason for hiding this comment

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

This is actually the primary flow. secondGoawayTimer is just in case the ping takes too long, we still want to continue the shutdown process. So using null here means that shutdown(void) is correct, but the max connection age won't have a grace time.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Ah right that's true.
So after the first goaway and ping, the secondGoawayTimer is set with 1s ping timeout.
A normal successful ping ack within 1s cancels the timer and trigger second goAway and graceful shutdown. No ping after 1s also triggers the same path.

The ping method does not take any more parameter, as it implements frame handler interface. So I made the grace time a field. And it is synchronized on the lock to allow read and write to be thread safe.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

@ejona86 wdyt?

return;
}
log.log(Level.INFO, "Received unexpected ping ack: " + payload);
Expand Down
18 changes: 12 additions & 6 deletions okhttp/src/test/java/io/grpc/okhttp/OkHttpServerTransportTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import okio.Buffer;
import okio.BufferedSource;
import okio.ByteString;
Expand Down Expand Up @@ -170,7 +171,7 @@ public void maxConnectionAge() throws Exception {
pingPong();
fakeClock.forwardNanos(TimeUnit.SECONDS.toNanos(6)); // > 1.1 * 5
fakeClock.forwardNanos(TimeUnit.SECONDS.toNanos(1));
verifyGracefulShutdown(1);
verifyGracefulShutdown(1, TimeUnit.SECONDS.toNanos(1));
}

@Test
Expand Down Expand Up @@ -204,7 +205,7 @@ public void maxConnectionIdleTimer() throws Exception {

fakeClock.forwardNanos(MAX_CONNECTION_IDLE);
fakeClock.forwardNanos(MAX_CONNECTION_IDLE);
verifyGracefulShutdown(1);
verifyGracefulShutdown(1, null);
}

@Test
Expand All @@ -228,7 +229,7 @@ public void maxConnectionIdleTimer_respondWithError() throws Exception {
pingPong();
fakeClock.forwardNanos(MAX_CONNECTION_IDLE);
fakeClock.forwardNanos(MAX_CONNECTION_IDLE);
verifyGracefulShutdown(1);
verifyGracefulShutdown(1, null);
}

@Test
Expand Down Expand Up @@ -402,7 +403,7 @@ public void activeRpc_delaysShutdownTermination() throws Exception {
pingPong();

serverTransport.shutdown();
verifyGracefulShutdown(1);
verifyGracefulShutdown(1, null);
verify(transportListener, never()).transportTerminated();

MockStreamListener streamListener = mockTransportListener.newStreams.pop();
Expand Down Expand Up @@ -1219,7 +1220,7 @@ private Metadata metadata(String... keysAndValues) {
return metadata;
}

private void verifyGracefulShutdown(int lastStreamId)
private void verifyGracefulShutdown(int lastStreamId, @Nullable Long gracePeriodNanos)
throws IOException {
assertThat(clientFrameReader.nextFrame(clientFramesRead)).isTrue();
verify(clientFramesRead).goAway(2147483647, ErrorCode.NO_ERROR, ByteString.EMPTY);
Expand All @@ -1229,12 +1230,16 @@ private void verifyGracefulShutdown(int lastStreamId)
clientFrameWriter.flush();
assertThat(clientFrameReader.nextFrame(clientFramesRead)).isTrue();
verify(clientFramesRead).goAway(lastStreamId, ErrorCode.NO_ERROR, ByteString.EMPTY);
if (gracePeriodNanos != null) {
assertThat(fakeClock.forwardNanos(gracePeriodNanos)).isEqualTo(1);
assertThat(socket.isClosed()).isTrue();
}
}

private void shutdownAndTerminate(int lastStreamId) throws IOException {
assertThat(serverTransport.getActiveStreams().length).isEqualTo(0);
serverTransport.shutdown();
verifyGracefulShutdown(lastStreamId);
verifyGracefulShutdown(lastStreamId, null);
assertThat(clientFrameReader.nextFrame(clientFramesRead)).isFalse();
verify(transportListener, timeout(TIME_OUT_MS)).transportTerminated();
}
Expand Down Expand Up @@ -1369,6 +1374,7 @@ public synchronized void close() throws IOException {
// PipedInputStream can only be woken by PipedOutputStream, so PipedOutputStream.close() is
// a better imitation of Socket.close().
inputStreamSource.close();
super.close();
}
}

Expand Down