gRPC improvements:

- Differentiate between the server terminating abruptly and it erroneously not returning an exit code.
- Do not assume that the message buffer stays untouched on the last call to Read() that eventually returns false.

There is a small chance that the channel will be shut down because it's idle between Read() returning false and us checking the state. Unfortunately, I haven't found a way to set the idle timeout from C++ :(

--
PiperOrigin-RevId: 147321404
MOS_MIGRATED_REVID=147321404

1 files changed, 23 insertions, 2 deletions

diff --git a/src/main/cpp/blaze.cc b/src/main/cpp/blaze.cc
index b0de6b3aa8..4240be51aa 100644
--- a/src/main/cpp/blaze.cc
+++ b/src/main/cpp/blaze.cc
@@ -1641,7 +1641,16 @@ unsigned int GrpcBlazeServer::Communicate() {
   std::thread cancel_thread(&GrpcBlazeServer::CancelThread, this);
   bool command_id_set = false;
   bool pipe_broken = false;
+  int exit_code = -1;
+  bool finished = false;
+  bool finished_warning_emitted = false;
+
   while (reader->Read(&response)) {
+    if (finished && !finished_warning_emitted) {
+      fprintf(stderr, "\nServer returned messages after reporting exit code\n");
+      finished_warning_emitted = true;
+    }
+
     if (response.cookie() != response_cookie_) {
       fprintf(stderr, "\nServer response cookie invalid, exiting\n");
       return blaze_exit_code::INTERNAL_ERROR;
@@ -1650,6 +1659,11 @@ unsigned int GrpcBlazeServer::Communicate() {
     bool pipe_broken_now = false;
     const char* broken_pipe_name;
 
+    if (response.finished()) {
+      exit_code = response.exit_code();
+      finished = true;
+    }
+
     if (!response.standard_output().empty()) {
       size_t size = response.standard_output().size();
       size_t r = fwrite(response.standard_output().c_str(), 1, size, stdout);
@@ -1685,7 +1699,14 @@ unsigned int GrpcBlazeServer::Communicate() {
   SendAction(CancelThreadAction::JOIN);
   cancel_thread.join();
 
-  if (!response.finished()) {
+  grpc::Status status = reader->Finish();
+  if (!status.ok()) {
+    fprintf(stderr, "\nServer terminated abruptly "
+            "(error code: %d, error message: '%s', log file: '%s')\n\n",
+            status.error_code(), status.error_message().c_str(),
+            globals->jvm_log_file.c_str());
+    return GetExitCodeForAbruptExit(*globals);
+  } else if (!finished) {
     fprintf(stderr, "\nServer finished RPC without an explicit exit code "
             "(log file: '%s')\n\n", globals->jvm_log_file.c_str());
     return GetExitCodeForAbruptExit(*globals);
@@ -1694,7 +1715,7 @@ unsigned int GrpcBlazeServer::Communicate() {
   // We'll exit with exit code SIGPIPE on Unixes due to PropagateSignalOnExit()
   return pipe_broken
       ? blaze_exit_code::LOCAL_ENVIRONMENTAL_ERROR
-      : response.exit_code();
+      : exit_code;
 }
 
 void GrpcBlazeServer::Disconnect() {