Skip to content

Commit

Permalink
all: use PerfMark.traceTask (#9950)
Browse files Browse the repository at this point in the history 
* all: use PerfMark.traceTask

* make linter happy
  • Loading branch information
@carl-mastrangelo
carl-mastrangelo authored Apr 18, 2023
1 parent 2e5cc84 commit 4e3ee44
Show file tree
Hide file tree
Showing 13 changed files with 291 additions and 447 deletions.
13 changes: 4 additions & 9 deletions core/src/main/java/io/grpc/internal/AbstractStream.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import io.grpc.Decompressor;
import io.perfmark.Link;
import io.perfmark.PerfMark;
import io.perfmark.TaskCloseable;
import java.io.InputStream;
import javax.annotation.concurrent.GuardedBy;

Expand Down Expand Up @@ -219,25 +220,19 @@ protected final void deframe(final ReadableBuffer frame) {
*/
private void requestMessagesFromDeframer(final int numMessages) {
if (deframer instanceof ThreadOptimizedDeframer) {
PerfMark.startTask("AbstractStream.request");
try {
try (TaskCloseable ignore = PerfMark.traceTask("AbstractStream.request")) {
deframer.request(numMessages);
} finally {
PerfMark.stopTask("AbstractStream.request");
}
return;
}
final Link link = PerfMark.linkOut();
class RequestRunnable implements Runnable {
@Override public void run() {
PerfMark.startTask("AbstractStream.request");
PerfMark.linkIn(link);
try {
try (TaskCloseable ignore = PerfMark.traceTask("AbstractStream.request")) {
PerfMark.linkIn(link);
deframer.request(numMessages);
} catch (Throwable t) {
deframeFailed(t);
} finally {
PerfMark.stopTask("AbstractStream.request");
}
}
}
Expand Down
223 changes: 98 additions & 125 deletions core/src/main/java/io/grpc/internal/ClientCallImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import io.perfmark.Link;
import io.perfmark.PerfMark;
import io.perfmark.Tag;
import io.perfmark.TaskCloseable;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.util.Locale;
Expand Down Expand Up @@ -187,11 +188,9 @@ static void prepareHeaders(

@Override
public void start(Listener<RespT> observer, Metadata headers) {
PerfMark.startTask("ClientCall.start", tag);
try {
try (TaskCloseable ignore = PerfMark.traceTask("ClientCall.start")) {
PerfMark.attachTag(tag);
startInternal(observer, headers);
} finally {
PerfMark.stopTask("ClientCall.start", tag);
}
}

Expand Down Expand Up @@ -446,23 +445,19 @@ private static boolean isFirstMin(@Nullable Deadline deadline0, @Nullable Deadli

@Override
public void request(int numMessages) {
PerfMark.startTask("ClientCall.request", tag);
try {
try (TaskCloseable ignore = PerfMark.traceTask("ClientCall.request")) {
PerfMark.attachTag(tag);
checkState(stream != null, "Not started");
checkArgument(numMessages >= 0, "Number requested must be non-negative");
stream.request(numMessages);
} finally {
PerfMark.stopTask("ClientCall.request", tag);
}
}

@Override
public void cancel(@Nullable String message, @Nullable Throwable cause) {
PerfMark.startTask("ClientCall.cancel", tag);
try {
try (TaskCloseable ignore = PerfMark.traceTask("ClientCall.cancel")) {
PerfMark.attachTag(tag);
cancelInternal(message, cause);
} finally {
PerfMark.stopTask("ClientCall.cancel", tag);
}
}

Expand Down Expand Up @@ -497,11 +492,9 @@ private void cancelInternal(@Nullable String message, @Nullable Throwable cause)

@Override
public void halfClose() {
PerfMark.startTask("ClientCall.halfClose", tag);
try {
try (TaskCloseable ignore = PerfMark.traceTask("ClientCall.halfClose")) {
PerfMark.attachTag(tag);
halfCloseInternal();
} finally {
PerfMark.stopTask("ClientCall.halfClose", tag);
}
}

Expand All @@ -515,11 +508,9 @@ private void halfCloseInternal() {

@Override
public void sendMessage(ReqT message) {
PerfMark.startTask("ClientCall.sendMessage", tag);
try {
try (TaskCloseable ignore = PerfMark.traceTask("ClientCall.sendMessage")) {
PerfMark.attachTag(tag);
sendMessageInternal(message);
} finally {
PerfMark.stopTask("ClientCall.sendMessage", tag);
}
}

Expand Down Expand Up @@ -603,104 +594,93 @@ private void exceptionThrown(Status status) {

@Override
public void headersRead(final Metadata headers) {
PerfMark.startTask("ClientStreamListener.headersRead", tag);
final Link link = PerfMark.linkOut();

final class HeadersRead extends ContextRunnable {
HeadersRead() {
super(context);
}

@Override
public void runInContext() {
PerfMark.startTask("ClientCall$Listener.headersRead", tag);
PerfMark.linkIn(link);
try {
runInternal();
} finally {
PerfMark.stopTask("ClientCall$Listener.headersRead", tag);
try (TaskCloseable ignore = PerfMark.traceTask("ClientStreamListener.headersRead")) {
PerfMark.attachTag(tag);
final Link link = PerfMark.linkOut();
final class HeadersRead extends ContextRunnable {
HeadersRead() {
super(context);
}
}

private void runInternal() {
if (exceptionStatus != null) {
return;
@Override
public void runInContext() {
try (TaskCloseable ignore = PerfMark.traceTask("ClientCall$Listener.headersRead")) {
PerfMark.attachTag(tag);
PerfMark.linkIn(link);
runInternal();
}
}
try {
observer.onHeaders(headers);
} catch (Throwable t) {
exceptionThrown(
Status.CANCELLED.withCause(t).withDescription("Failed to read headers"));

private void runInternal() {
if (exceptionStatus != null) {
return;
}
try {
observer.onHeaders(headers);
} catch (Throwable t) {
exceptionThrown(
Status.CANCELLED.withCause(t).withDescription("Failed to read headers"));
}
}
}
}

try {
callExecutor.execute(new HeadersRead());
} finally {
PerfMark.stopTask("ClientStreamListener.headersRead", tag);
}
}

@Override
public void messagesAvailable(final MessageProducer producer) {
PerfMark.startTask("ClientStreamListener.messagesAvailable", tag);
final Link link = PerfMark.linkOut();

final class MessagesAvailable extends ContextRunnable {
MessagesAvailable() {
super(context);
}

@Override
public void runInContext() {
PerfMark.startTask("ClientCall$Listener.messagesAvailable", tag);
PerfMark.linkIn(link);
try {
runInternal();
} finally {
PerfMark.stopTask("ClientCall$Listener.messagesAvailable", tag);
try (TaskCloseable ignore = PerfMark.traceTask("ClientStreamListener.messagesAvailable")) {
PerfMark.attachTag(tag);
final Link link = PerfMark.linkOut();
final class MessagesAvailable extends ContextRunnable {
MessagesAvailable() {
super(context);
}
}

private void runInternal() {
if (exceptionStatus != null) {
GrpcUtil.closeQuietly(producer);
return;
@Override
public void runInContext() {
try (TaskCloseable ignore =
PerfMark.traceTask("ClientCall$Listener.messagesAvailable")) {
PerfMark.attachTag(tag);
PerfMark.linkIn(link);
runInternal();
}
}
try {
InputStream message;
while ((message = producer.next()) != null) {
try {
observer.onMessage(method.parseResponse(message));
} catch (Throwable t) {
GrpcUtil.closeQuietly(message);
throw t;

private void runInternal() {
if (exceptionStatus != null) {
GrpcUtil.closeQuietly(producer);
return;
}
try {
InputStream message;
while ((message = producer.next()) != null) {
try {
observer.onMessage(method.parseResponse(message));
} catch (Throwable t) {
GrpcUtil.closeQuietly(message);
throw t;
}
message.close();
}
message.close();
} catch (Throwable t) {
GrpcUtil.closeQuietly(producer);
exceptionThrown(
Status.CANCELLED.withCause(t).withDescription("Failed to read message."));
}
} catch (Throwable t) {
GrpcUtil.closeQuietly(producer);
exceptionThrown(
Status.CANCELLED.withCause(t).withDescription("Failed to read message."));
}
}
}

try {
callExecutor.execute(new MessagesAvailable());
} finally {
PerfMark.stopTask("ClientStreamListener.messagesAvailable", tag);
}
}

@Override
public void closed(Status status, RpcProgress rpcProgress, Metadata trailers) {
PerfMark.startTask("ClientStreamListener.closed", tag);
try {
try (TaskCloseable ignore = PerfMark.traceTask("ClientStreamListener.closed")) {
PerfMark.attachTag(tag);
closedInternal(status, rpcProgress, trailers);
} finally {
PerfMark.stopTask("ClientStreamListener.closed", tag);
}
}

Expand Down Expand Up @@ -730,12 +710,10 @@ final class StreamClosed extends ContextRunnable {

@Override
public void runInContext() {
PerfMark.startTask("ClientCall$Listener.onClose", tag);
PerfMark.linkIn(link);
try {
try (TaskCloseable ignore = PerfMark.traceTask("ClientCall$Listener.onClose")) {
PerfMark.attachTag(tag);
PerfMark.linkIn(link);
runInternal();
} finally {
PerfMark.stopTask("ClientCall$Listener.onClose", tag);
}
}

Expand Down Expand Up @@ -770,43 +748,38 @@ public void onReady() {
if (method.getType().clientSendsOneMessage()) {
return;
}
try (TaskCloseable ignore = PerfMark.traceTask("ClientStreamListener.onReady")) {
PerfMark.attachTag(tag);
final Link link = PerfMark.linkOut();

PerfMark.startTask("ClientStreamListener.onReady", tag);
final Link link = PerfMark.linkOut();

final class StreamOnReady extends ContextRunnable {
StreamOnReady() {
super(context);
}

@Override
public void runInContext() {
PerfMark.startTask("ClientCall$Listener.onReady", tag);
PerfMark.linkIn(link);
try {
runInternal();
} finally {
PerfMark.stopTask("ClientCall$Listener.onReady", tag);
final class StreamOnReady extends ContextRunnable {
StreamOnReady() {
super(context);
}
}

private void runInternal() {
if (exceptionStatus != null) {
return;
@Override
public void runInContext() {
try (TaskCloseable ignore = PerfMark.traceTask("ClientCall$Listener.onReady")) {
PerfMark.attachTag(tag);
PerfMark.linkIn(link);
runInternal();
}
}
try {
observer.onReady();
} catch (Throwable t) {
exceptionThrown(
Status.CANCELLED.withCause(t).withDescription("Failed to call onReady."));

private void runInternal() {
if (exceptionStatus != null) {
return;
}
try {
observer.onReady();
} catch (Throwable t) {
exceptionThrown(
Status.CANCELLED.withCause(t).withDescription("Failed to call onReady."));
}
}
}
}

try {
callExecutor.execute(new StreamOnReady());
} finally {
PerfMark.stopTask("ClientStreamListener.onReady", tag);
}
}
}
Expand Down
Loading

0 comments on commit 4e3ee44

Please sign in to comment.