Merge branch 'buffer_pools_for_realsies' into flow_control_v2

include/grpc++/ext/reflection.grpc.pb.h

@@ -176,6 +176,8 @@ class ServerReflection GRPC_FINAL {
     }
   };
   typedef Service StreamedUnaryService;
+  typedef Service SplitStreamedService;
+  typedef Service StreamedService;
 };
 
 }  // namespace v1alpha

include/grpc++/impl/codegen/method_handler_impl.h

@@ -236,6 +236,19 @@ class StreamedUnaryHandler
             ServerUnaryStreamer<RequestType, ResponseType>, true>(func) {}
 };
 
+template <class RequestType, class ResponseType>
+class SplitServerStreamingHandler
+    : public TemplatedBidiStreamingHandler<
+          ServerSplitStreamer<RequestType, ResponseType>, false> {
+ public:
+  explicit SplitServerStreamingHandler(
+      std::function<Status(ServerContext*,
+                           ServerSplitStreamer<RequestType, ResponseType>*)>
+          func)
+      : TemplatedBidiStreamingHandler<
+            ServerSplitStreamer<RequestType, ResponseType>, false>(func) {}
+};
+
 // Handle unknown method by returning UNIMPLEMENTED error.
 class UnknownMethodHandler : public MethodHandler {
  public:

include/grpc++/impl/codegen/server_context.h

@@ -44,7 +44,6 @@
 #include <grpc++/impl/codegen/time.h>
 #include <grpc/impl/codegen/compression_types.h>
 
-struct gpr_timespec;
 struct grpc_metadata;
 struct grpc_call;
 struct census_context;

include/grpc++/impl/codegen/service_type.h

@@ -147,14 +147,15 @@ class Service {
     methods_[index].reset();
   }
 
-  void MarkMethodStreamedUnary(int index,
-                               MethodHandler* streamed_unary_method) {
+  void MarkMethodStreamed(int index, MethodHandler* streamed_method) {
     GPR_CODEGEN_ASSERT(methods_[index] && methods_[index]->handler() &&
-                       "Cannot mark an async or generic method Streamed Unary");
-    methods_[index]->SetHandler(streamed_unary_method);
+                       "Cannot mark an async or generic method Streamed");
+    methods_[index]->SetHandler(streamed_method);
 
     // From the server's point of view, streamed unary is a special
-    // case of BIDI_STREAMING that has 1 read and 1 write, in that order.
+    // case of BIDI_STREAMING that has 1 read and 1 write, in that order,
+    // and split server-side streaming is BIDI_STREAMING with 1 read and
+    // any number of writes, in that order
     methods_[index]->SetMethodType(::grpc::RpcMethod::BIDI_STREAMING);
   }
 

include/grpc++/impl/codegen/sync_stream.h

@@ -538,7 +538,7 @@ class ServerReaderWriter GRPC_FINAL : public ServerReaderWriterInterface<W, R> {
 /// the \a NextMessageSize method to determine an upper-bound on the size of
 /// the message.
 /// A key difference relative to streaming: ServerUnaryStreamer
-///  must have exactly 1 Read and exactly 1 Write, in that order, to function
+/// must have exactly 1 Read and exactly 1 Write, in that order, to function
 /// correctly. Otherwise, the RPC is in error.
 template <class RequestType, class ResponseType>
 class ServerUnaryStreamer GRPC_FINAL
@@ -577,6 +577,43 @@ class ServerUnaryStreamer GRPC_FINAL
   bool write_done_;
 };
 
+/// A class to represent a flow-controlled server-side streaming call.
+/// This is something of a hybrid between server-side and bidi streaming.
+/// This is invoked through a server-side streaming call on the client side,
+/// but the server responds to it as though it were a bidi streaming call that
+/// must first have exactly 1 Read and then any number of Writes.
+template <class RequestType, class ResponseType>
+class ServerSplitStreamer GRPC_FINAL
+    : public ServerReaderWriterInterface<ResponseType, RequestType> {
+ public:
+  ServerSplitStreamer(Call* call, ServerContext* ctx)
+      : body_(call, ctx), read_done_(false) {}
+
+  void SendInitialMetadata() GRPC_OVERRIDE { body_.SendInitialMetadata(); }
+
+  bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE {
+    return body_.NextMessageSize(sz);
+  }
+
+  bool Read(RequestType* request) GRPC_OVERRIDE {
+    if (read_done_) {
+      return false;
+    }
+    read_done_ = true;
+    return body_.Read(request);
+  }
+
+  using WriterInterface<ResponseType>::Write;
+  bool Write(const ResponseType& response,
+             const WriteOptions& options) GRPC_OVERRIDE {
+    return read_done_ && body_.Write(response, options);
+  }
+
+ private:
+  internal::ServerReaderWriterBody<ResponseType, RequestType> body_;
+  bool read_done_;
+};
+
 }  // namespace grpc
 
 #endif  // GRPCXX_IMPL_CODEGEN_SYNC_STREAM_H

src/compiler/cpp_generator.cc

@@ -624,7 +624,7 @@ void PrintHeaderServerMethodStreamedUnary(
     printer->Indent();
     printer->Print(*vars,
                    "WithStreamedUnaryMethod_$Method$() {\n"
-                   "  ::grpc::Service::MarkMethodStreamedUnary($Idx$,\n"
+                   "  ::grpc::Service::MarkMethodStreamed($Idx$,\n"
                    "    new ::grpc::StreamedUnaryHandler< $Request$, "
                    "$Response$>(std::bind"
                    "(&WithStreamedUnaryMethod_$Method$<BaseClass>::"
@@ -656,6 +656,58 @@ void PrintHeaderServerMethodStreamedUnary(
   }
 }
 
+void PrintHeaderServerMethodSplitStreaming(
+    Printer *printer, const Method *method,
+    std::map<grpc::string, grpc::string> *vars) {
+  (*vars)["Method"] = method->name();
+  (*vars)["Request"] = method->input_type_name();
+  (*vars)["Response"] = method->output_type_name();
+  if (method->ServerOnlyStreaming()) {
+    printer->Print(*vars, "template <class BaseClass>\n");
+    printer->Print(*vars,
+                   "class WithSplitStreamingMethod_$Method$ : "
+                   "public BaseClass {\n");
+    printer->Print(
+        " private:\n"
+        "  void BaseClassMustBeDerivedFromService(const Service *service) "
+        "{}\n");
+    printer->Print(" public:\n");
+    printer->Indent();
+    printer->Print(*vars,
+                   "WithSplitStreamingMethod_$Method$() {\n"
+                   "  ::grpc::Service::MarkMethodStreamed($Idx$,\n"
+                   "    new ::grpc::SplitServerStreamingHandler< $Request$, "
+                   "$Response$>(std::bind"
+                   "(&WithSplitStreamingMethod_$Method$<BaseClass>::"
+                   "Streamed$Method$, this, std::placeholders::_1, "
+                   "std::placeholders::_2)));\n"
+                   "}\n");
+    printer->Print(*vars,
+                   "~WithSplitStreamingMethod_$Method$() GRPC_OVERRIDE {\n"
+                   "  BaseClassMustBeDerivedFromService(this);\n"
+                   "}\n");
+    printer->Print(
+        *vars,
+        "// disable regular version of this method\n"
+        "::grpc::Status $Method$("
+        "::grpc::ServerContext* context, const $Request$* request, "
+        "::grpc::ServerWriter< $Response$>* writer) GRPC_FINAL GRPC_OVERRIDE "
+        "{\n"
+        "  abort();\n"
+        "  return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n"
+        "}\n");
+    printer->Print(*vars,
+                   "// replace default version of method with split streamed\n"
+                   "virtual ::grpc::Status Streamed$Method$("
+                   "::grpc::ServerContext* context, "
+                   "::grpc::ServerSplitStreamer< "
+                   "$Request$,$Response$>* server_split_streamer)"
+                   " = 0;\n");
+    printer->Outdent();
+    printer->Print(*vars, "};\n");
+  }
+}
+
 void PrintHeaderServerMethodGeneric(
     Printer *printer, const Method *method,
     std::map<grpc::string, grpc::string> *vars) {
@@ -844,6 +896,48 @@ void PrintHeaderService(Printer *printer, const Service *service,
   }
   printer->Print(" StreamedUnaryService;\n");
 
+  // Server side - controlled server-side streaming
+  for (int i = 0; i < service->method_count(); ++i) {
+    (*vars)["Idx"] = as_string(i);
+    PrintHeaderServerMethodSplitStreaming(printer, service->method(i).get(),
+                                          vars);
+  }
+
+  printer->Print("typedef ");
+  for (int i = 0; i < service->method_count(); ++i) {
+    (*vars)["method_name"] = service->method(i).get()->name();
+    if (service->method(i)->ServerOnlyStreaming()) {
+      printer->Print(*vars, "WithSplitStreamingMethod_$method_name$<");
+    }
+  }
+  printer->Print("Service");
+  for (int i = 0; i < service->method_count(); ++i) {
+    if (service->method(i)->ServerOnlyStreaming()) {
+      printer->Print(" >");
+    }
+  }
+  printer->Print(" SplitStreamedService;\n");
+
+  // Server side - typedef for controlled both unary and server-side streaming
+  printer->Print("typedef ");
+  for (int i = 0; i < service->method_count(); ++i) {
+    (*vars)["method_name"] = service->method(i).get()->name();
+    if (service->method(i)->ServerOnlyStreaming()) {
+      printer->Print(*vars, "WithSplitStreamingMethod_$method_name$<");
+    }
+    if (service->method(i)->NoStreaming()) {
+      printer->Print(*vars, "WithStreamedUnaryMethod_$method_name$<");
+    }
+  }
+  printer->Print("Service");
+  for (int i = 0; i < service->method_count(); ++i) {
+    if (service->method(i)->NoStreaming() ||
+        service->method(i)->ServerOnlyStreaming()) {
+      printer->Print(" >");
+    }
+  }
+  printer->Print(" StreamedService;\n");
+
   printer->Outdent();
   printer->Print("};\n");
   printer->Print(service->GetTrailingComments().c_str());

src/csharp/Grpc.IntegrationTesting/InteropClient.cs

@@ -583,7 +583,6 @@ namespace Grpc.IntegrationTesting
             var e = Assert.Throws<RpcException>(() => client.UnimplementedCall(new Empty()));
 
             Assert.AreEqual(StatusCode.Unimplemented, e.Status.StatusCode);
-            Assert.AreEqual("", e.Status.Detail);
             Console.WriteLine("Passed!");
         }
 

src/proto/grpc/testing/duplicate/echo_duplicate.proto

@@ -38,4 +38,5 @@ package grpc.testing.duplicate;
 
 service EchoTestService {
   rpc Echo(grpc.testing.EchoRequest) returns (grpc.testing.EchoResponse);
+  rpc ResponseStream(EchoRequest) returns (stream EchoResponse);
 }

src/proto/grpc/testing/test.proto

@@ -77,7 +77,7 @@ service TestService {
 
   // The test server will not implement this method. It will be used
   // to test the behavior when clients call unimplemented methods.
-  rpc UnimplementedMethod(grpc.testing.Empty) returns (grpc.testing.Empty);
+  rpc UnimplementedCall(grpc.testing.Empty) returns (grpc.testing.Empty);
 }
 
 // A simple service NOT implemented at servers so clients can test for

test/cpp/codegen/compiler_test_golden

@@ -213,7 +213,7 @@ class ServiceA GRPC_FINAL {
     void BaseClassMustBeDerivedFromService(const Service *service) {}
    public:
     WithStreamedUnaryMethod_MethodA1() {
-      ::grpc::Service::MarkMethodStreamedUnary(0,
+      ::grpc::Service::MarkMethodStreamed(0,
         new ::grpc::StreamedUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithStreamedUnaryMethod_MethodA1<BaseClass>::StreamedMethodA1, this, std::placeholders::_1, std::placeholders::_2)));
     }
     ~WithStreamedUnaryMethod_MethodA1() GRPC_OVERRIDE {
@@ -228,6 +228,8 @@ class ServiceA GRPC_FINAL {
     virtual ::grpc::Status StreamedMethodA1(::grpc::ServerContext* context, ::grpc::ServerUnaryStreamer< ::grpc::testing::Request,::grpc::testing::Response>* server_unary_streamer) = 0;
   };
   typedef WithStreamedUnaryMethod_MethodA1<Service > StreamedUnaryService;
+  typedef Service SplitStreamedService;
+  typedef WithStreamedUnaryMethod_MethodA1<Service > StreamedService;
 };
 
 // ServiceB leading comment 1
@@ -312,7 +314,7 @@ class ServiceB GRPC_FINAL {
     void BaseClassMustBeDerivedFromService(const Service *service) {}
    public:
     WithStreamedUnaryMethod_MethodB1() {
-      ::grpc::Service::MarkMethodStreamedUnary(0,
+      ::grpc::Service::MarkMethodStreamed(0,
         new ::grpc::StreamedUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithStreamedUnaryMethod_MethodB1<BaseClass>::StreamedMethodB1, this, std::placeholders::_1, std::placeholders::_2)));
     }
     ~WithStreamedUnaryMethod_MethodB1() GRPC_OVERRIDE {
@@ -327,6 +329,8 @@ class ServiceB GRPC_FINAL {
     virtual ::grpc::Status StreamedMethodB1(::grpc::ServerContext* context, ::grpc::ServerUnaryStreamer< ::grpc::testing::Request,::grpc::testing::Response>* server_unary_streamer) = 0;
   };
   typedef WithStreamedUnaryMethod_MethodB1<Service > StreamedUnaryService;
+  typedef Service SplitStreamedService;
+  typedef WithStreamedUnaryMethod_MethodB1<Service > StreamedService;
 };
 // ServiceB trailing comment 1
 

test/cpp/end2end/hybrid_end2end_test.cc

@@ -320,6 +320,29 @@ class HybridEnd2endTest : public ::testing::Test {
     EXPECT_TRUE(s.ok());
   }
 
+  void SendSimpleServerStreamingToDupService() {
+    std::shared_ptr<Channel> channel =
+        CreateChannel(server_address_.str(), InsecureChannelCredentials());
+    auto stub = grpc::testing::duplicate::EchoTestService::NewStub(channel);
+    EchoRequest request;
+    EchoResponse response;
+    ClientContext context;
+    context.set_wait_for_ready(true);
+    request.set_message("hello");
+
+    auto stream = stub->ResponseStream(&context, request);
+    EXPECT_TRUE(stream->Read(&response));
+    EXPECT_EQ(response.message(), request.message() + "0_dup");
+    EXPECT_TRUE(stream->Read(&response));
+    EXPECT_EQ(response.message(), request.message() + "1_dup");
+    EXPECT_TRUE(stream->Read(&response));
+    EXPECT_EQ(response.message(), request.message() + "2_dup");
+    EXPECT_FALSE(stream->Read(&response));
+
+    Status s = stream->Finish();
+    EXPECT_TRUE(s.ok());
+  }
+
   void SendBidiStreaming() {
     EchoRequest request;
     EchoResponse response;
@@ -498,6 +521,140 @@ TEST_F(HybridEnd2endTest,
   request_stream_handler_thread.join();
 }
 
+// Add a second service with one sync split server streaming method.
+class SplitResponseStreamDupPkg
+    : public duplicate::EchoTestService::
+          WithSplitStreamingMethod_ResponseStream<TestServiceImplDupPkg> {
+ public:
+  Status StreamedResponseStream(
+      ServerContext* context,
+      ServerSplitStreamer<EchoRequest, EchoResponse>* stream) GRPC_OVERRIDE {
+    EchoRequest req;
+    EchoResponse resp;
+    uint32_t next_msg_sz;
+    stream->NextMessageSize(&next_msg_sz);
+    gpr_log(GPR_INFO, "Split Streamed Next Message Size is %u", next_msg_sz);
+    GPR_ASSERT(stream->Read(&req));
+    for (int i = 0; i < kNumResponseStreamsMsgs; i++) {
+      resp.set_message(req.message() + grpc::to_string(i) + "_dup");
+      GPR_ASSERT(stream->Write(resp));
+    }
+    return Status::OK;
+  }
+};
+
+TEST_F(HybridEnd2endTest,
+       AsyncRequestStreamResponseStream_SyncSplitStreamedDupService) {
+  typedef EchoTestService::WithAsyncMethod_RequestStream<
+      EchoTestService::WithAsyncMethod_ResponseStream<TestServiceImpl>>
+      SType;
+  SType service;
+  SplitResponseStreamDupPkg dup_service;
+  SetUpServer(&service, &dup_service, nullptr, 8192);
+  ResetStub();
+  std::thread response_stream_handler_thread(HandleServerStreaming<SType>,
+                                             &service, cqs_[0].get());
+  std::thread request_stream_handler_thread(HandleClientStreaming<SType>,
+                                            &service, cqs_[1].get());
+  TestAllMethods();
+  SendSimpleServerStreamingToDupService();
+  response_stream_handler_thread.join();
+  request_stream_handler_thread.join();
+}
+
+// Add a second service that is fully split server streamed
+class FullySplitStreamedDupPkg
+    : public duplicate::EchoTestService::SplitStreamedService {
+ public:
+  Status StreamedResponseStream(
+      ServerContext* context,
+      ServerSplitStreamer<EchoRequest, EchoResponse>* stream) GRPC_OVERRIDE {
+    EchoRequest req;
+    EchoResponse resp;
+    uint32_t next_msg_sz;
+    stream->NextMessageSize(&next_msg_sz);
+    gpr_log(GPR_INFO, "Split Streamed Next Message Size is %u", next_msg_sz);
+    GPR_ASSERT(stream->Read(&req));
+    for (int i = 0; i < kNumResponseStreamsMsgs; i++) {
+      resp.set_message(req.message() + grpc::to_string(i) + "_dup");
+      GPR_ASSERT(stream->Write(resp));
+    }
+    return Status::OK;
+  }
+};
+
+TEST_F(HybridEnd2endTest,
+       AsyncRequestStreamResponseStream_FullySplitStreamedDupService) {
+  typedef EchoTestService::WithAsyncMethod_RequestStream<
+      EchoTestService::WithAsyncMethod_ResponseStream<TestServiceImpl>>
+      SType;
+  SType service;
+  FullySplitStreamedDupPkg dup_service;
+  SetUpServer(&service, &dup_service, nullptr, 8192);
+  ResetStub();
+  std::thread response_stream_handler_thread(HandleServerStreaming<SType>,
+                                             &service, cqs_[0].get());
+  std::thread request_stream_handler_thread(HandleClientStreaming<SType>,
+                                            &service, cqs_[1].get());
+  TestAllMethods();
+  SendSimpleServerStreamingToDupService();
+  response_stream_handler_thread.join();
+  request_stream_handler_thread.join();
+}
+
+// Add a second service that is fully server streamed
+class FullyStreamedDupPkg : public duplicate::EchoTestService::StreamedService {
+ public:
+  Status StreamedEcho(ServerContext* context,
+                      ServerUnaryStreamer<EchoRequest, EchoResponse>* stream)
+      GRPC_OVERRIDE {
+    EchoRequest req;
+    EchoResponse resp;
+    uint32_t next_msg_sz;
+    stream->NextMessageSize(&next_msg_sz);
+    gpr_log(GPR_INFO, "Streamed Unary Next Message Size is %u", next_msg_sz);
+    GPR_ASSERT(stream->Read(&req));
+    resp.set_message(req.message() + "_dup");
+    GPR_ASSERT(stream->Write(resp));
+    return Status::OK;
+  }
+  Status StreamedResponseStream(
+      ServerContext* context,
+      ServerSplitStreamer<EchoRequest, EchoResponse>* stream) GRPC_OVERRIDE {
+    EchoRequest req;
+    EchoResponse resp;
+    uint32_t next_msg_sz;
+    stream->NextMessageSize(&next_msg_sz);
+    gpr_log(GPR_INFO, "Split Streamed Next Message Size is %u", next_msg_sz);
+    GPR_ASSERT(stream->Read(&req));
+    for (int i = 0; i < kNumResponseStreamsMsgs; i++) {
+      resp.set_message(req.message() + grpc::to_string(i) + "_dup");
+      GPR_ASSERT(stream->Write(resp));
+    }
+    return Status::OK;
+  }
+};
+
+TEST_F(HybridEnd2endTest,
+       AsyncRequestStreamResponseStream_FullyStreamedDupService) {
+  typedef EchoTestService::WithAsyncMethod_RequestStream<
+      EchoTestService::WithAsyncMethod_ResponseStream<TestServiceImpl>>
+      SType;
+  SType service;
+  FullyStreamedDupPkg dup_service;
+  SetUpServer(&service, &dup_service, nullptr, 8192);
+  ResetStub();
+  std::thread response_stream_handler_thread(HandleServerStreaming<SType>,
+                                             &service, cqs_[0].get());
+  std::thread request_stream_handler_thread(HandleClientStreaming<SType>,
+                                            &service, cqs_[1].get());
+  TestAllMethods();
+  SendEchoToDupService();
+  SendSimpleServerStreamingToDupService();
+  response_stream_handler_thread.join();
+  request_stream_handler_thread.join();
+}
+
 // Add a second service with one async method.
 TEST_F(HybridEnd2endTest, AsyncRequestStreamResponseStream_AsyncDupService) {
   typedef EchoTestService::WithAsyncMethod_RequestStream<

test/cpp/interop/client.cc

@@ -80,7 +80,7 @@ DEFINE_string(test_case, "large_unary",
               "slow client consumer;\n"
               "status_code_and_message: verify status code & message;\n"
               "timeout_on_sleeping_server: deadline exceeds on stream;\n"
-              "unimplemented_method: client calls an unimplemented_method;\n");
+              "unimplemented_method: client calls an unimplemented method;\n");
 DEFINE_string(default_service_account, "",
               "Email of GCE default service account");
 DEFINE_string(service_account_key_file, "",

test/cpp/interop/interop_client.cc

@@ -1010,7 +1010,7 @@ bool InteropClient::DoUnimplementedMethod() {
   ClientContext context;
 
   Status s =
-      serviceStub_.Get()->UnimplementedMethod(&context, request, &response);
+      serviceStub_.Get()->UnimplementedCall(&context, request, &response);
 
   if (!AssertStatusCode(s, StatusCode::UNIMPLEMENTED)) {
     return false;

test/cpp/qps/server_async.cc

@@ -58,7 +58,7 @@ namespace testing {
 
 template <class RequestType, class ResponseType, class ServiceType,
           class ServerContextType>
-class AsyncQpsServerTest : public Server {
+class AsyncQpsServerTest GRPC_FINAL : public grpc::testing::Server {
  public:
   AsyncQpsServerTest(
       const ServerConfig &config,

test/cpp/qps/server_sync.cc

@@ -31,8 +31,6 @@
  *
  */
 
-#include <thread>
-
 #include <grpc++/resource_quota.h>
 #include <grpc++/security/server_credentials.h>
 #include <grpc++/server.h>

tools/run_tests/filter_pull_request_tests.py

@@ -36,18 +36,16 @@ from subprocess import call, check_output
 
 class TestSuite:
   """
-  Contains tag to identify job as belonging to this test suite and
+  Contains label to identify job as belonging to this test suite and
   triggers to identify if changed files are relevant
   """
-  def __init__(self, tags):
+  def __init__(self, labels):
     """
-    Build TestSuite to group tests by their tags
-    :param tag: string used to identify if a job belongs to this TestSuite
-    todo(mattkwong): Change the use of tag because do not want to depend on
-    job.shortname to identify what suite a test belongs to
+    Build TestSuite to group tests based on labeling
+    :param label: strings that should match a jobs's platform, config, language, or test group
     """
     self.triggers = []
-    self.tags = tags
+    self.labels = labels
 
   def add_trigger(self, trigger):
     """
@@ -56,46 +54,75 @@ class TestSuite:
     """
     self.triggers.append(trigger)
 
+
 # Create test suites
-_core_test_suite = TestSuite(['_c_'])
-_cpp_test_suite = TestSuite(['_c++_'])
-_csharp_test_suite = TestSuite(['_csharp_'])
-_node_test_suite = TestSuite(['_node_'])
-_objc_test_suite = TestSuite(['_objc_'])
-_php_test_suite = TestSuite(['_php_', '_php7_'])
-_python_test_suite = TestSuite(['_python_'])
-_ruby_test_suite = TestSuite(['_ruby'])
-_all_test_suites = [_core_test_suite, _cpp_test_suite, _csharp_test_suite,
-                    _node_test_suite, _objc_test_suite, _php_test_suite,
-                    _python_test_suite, _ruby_test_suite]
+_SANITY_TEST_SUITE = TestSuite(['sanity'])
+_CORE_TEST_SUITE = TestSuite(['c'])
+_CPP_TEST_SUITE = TestSuite(['c++'])
+_CSHARP_TEST_SUITE = TestSuite(['csharp'])
+_NODE_TEST_SUITE = TestSuite(['node'])
+_OBJC_TEST_SUITE = TestSuite(['objc'])
+_PHP_TEST_SUITE = TestSuite(['php', 'php7'])
+_PYTHON_TEST_SUITE = TestSuite(['python'])
+_RUBY_TEST_SUITE = TestSuite(['ruby'])
+_LINUX_TEST_SUITE = TestSuite(['linux'])
+_WINDOWS_TEST_SUITE = TestSuite(['windows'])
+_MACOS_TEST_SUITE = TestSuite(['macos'])
+_ALL_TEST_SUITES = [_SANITY_TEST_SUITE, _CORE_TEST_SUITE, _CPP_TEST_SUITE,
+                    _CSHARP_TEST_SUITE, _NODE_TEST_SUITE, _OBJC_TEST_SUITE,
+                    _PHP_TEST_SUITE, _PYTHON_TEST_SUITE, _RUBY_TEST_SUITE,
+                    _LINUX_TEST_SUITE, _WINDOWS_TEST_SUITE, _MACOS_TEST_SUITE]
 
 # Dictionary of whitelistable files where the key is a regex matching changed files
 # and the value is a list of tests that should be run. An empty list means that
 # the changed files should not trigger any tests. Any changed file that does not
 # match any of these regexes will trigger all tests
 _WHITELIST_DICT = {
-  '^templates/.*': [],
-  '^doc/.*': [],
-  '^examples/.*': [],
-  '^summerofcode/.*': [],
-  '.*README.md$': [],
-  '.*LICENSE$': [],
-  '^src/cpp.*': [_cpp_test_suite],
-  '^src/csharp.*': [_csharp_test_suite],
-  '^src/node.*': [_node_test_suite],
-  '^src/objective-c.*': [_objc_test_suite],
-  '^src/php.*': [_php_test_suite],
-  '^src/python.*': [_python_test_suite],
-  '^src/ruby.*': [_ruby_test_suite],
-  '^test/core.*': [_core_test_suite],
-  '^test/cpp.*': [_cpp_test_suite],
-  '^test/distrib/cpp.*': [_cpp_test_suite],
-  '^test/distrib/csharp.*': [_csharp_test_suite],
-  '^test/distrib/node.*': [_node_test_suite],
-  '^test/distrib/php.*': [_php_test_suite],
-  '^test/distrib/python.*': [_python_test_suite],
-  '^test/distrib/ruby.*': [_ruby_test_suite]
+  '^doc/': [],
+  '^examples/': [],
+  '^include/grpc\+\+/': [_CPP_TEST_SUITE],
+  '^summerofcode/': [],
+  '^src/cpp/': [_CPP_TEST_SUITE],
+  '^src/csharp/': [_CSHARP_TEST_SUITE],
+  '^src/node/': [_NODE_TEST_SUITE],
+  '^src/objective\-c/': [_OBJC_TEST_SUITE],
+  '^src/php/': [_PHP_TEST_SUITE],
+  '^src/python/': [_PYTHON_TEST_SUITE],
+  '^src/ruby/': [_RUBY_TEST_SUITE],
+  '^templates/': [_SANITY_TEST_SUITE],
+  '^test/core/': [_CORE_TEST_SUITE],
+  '^test/cpp/': [_CPP_TEST_SUITE],
+  '^test/distrib/cpp/': [_CPP_TEST_SUITE],
+  '^test/distrib/csharp/': [_CSHARP_TEST_SUITE],
+  '^test/distrib/node/': [_NODE_TEST_SUITE],
+  '^test/distrib/php/': [_PHP_TEST_SUITE],
+  '^test/distrib/python/': [_PYTHON_TEST_SUITE],
+  '^test/distrib/ruby/': [_RUBY_TEST_SUITE],
+  '^vsprojects/': [_WINDOWS_TEST_SUITE],
+  'binding\.gyp$': [_NODE_TEST_SUITE],
+  'composer\.json$': [_PHP_TEST_SUITE],
+  'config\.m4$': [_PHP_TEST_SUITE],
+  'CONTRIBUTING\.md$': [],
+  'Gemfile$': [_RUBY_TEST_SUITE],
+  'grpc.def$': [_WINDOWS_TEST_SUITE],
+  'grpc\.gemspec$': [_RUBY_TEST_SUITE],
+  'gRPC\.podspec$': [_OBJC_TEST_SUITE],
+  'gRPC\-Core\.podspec$': [_OBJC_TEST_SUITE],
+  'gRPC\-ProtoRPC\.podspec$': [_OBJC_TEST_SUITE],
+  'gRPC\-RxLibrary\.podspec$': [_OBJC_TEST_SUITE],
+  'INSTALL\.md$': [],
+  'LICENSE$': [],
+  'MANIFEST\.md$': [],
+  'package\.json$': [_PHP_TEST_SUITE],
+  'package\.xml$': [_PHP_TEST_SUITE],
+  'PATENTS$': [],
+  'PYTHON\-MANIFEST\.in$': [_PYTHON_TEST_SUITE],
+  'README\.md$': [],
+  'requirements\.txt$': [_PYTHON_TEST_SUITE],
+  'setup\.cfg$': [_PYTHON_TEST_SUITE],
+  'setup\.py$': [_PYTHON_TEST_SUITE]
 }
+
 # Add all triggers to their respective test suites
 for trigger, test_suites in _WHITELIST_DICT.iteritems():
   for test_suite in test_suites:
@@ -106,10 +133,6 @@ def _get_changed_files(base_branch):
   """
   Get list of changed files between current branch and base of target merge branch
   """
-  # git fetch might need to be called on Jenkins slave
-  # todo(mattkwong): remove or uncomment below after seeing if Jenkins needs this
-  # call(['git', 'fetch'])
-
   # Get file changes between branch and merge-base of specified branch
   # Not combined to be Windows friendly
   base_commit = check_output(["git", "merge-base", base_branch, "HEAD"]).rstrip()
@@ -129,27 +152,17 @@ def _can_skip_tests(file_names, triggers):
   return True
 
 
-def _remove_irrelevant_tests(tests, tag):
+def _remove_irrelevant_tests(tests, skippable_labels):
   """
   Filters out tests by config or language - will not remove sanitizer tests
   :param tests: list of all tests generated by run_tests_matrix.py
-  :param tag: string representing language or config to filter - "_(language)_" or "_(config)"
-  :return: list of relevant tests
-  """
-  # todo(mattkwong): find a more reliable way to filter tests - don't use shortname
-  return [test for test in tests if tag not in test.shortname or
-          any(san_tag in test.shortname for san_tag in ['_asan', '_tsan', '_msan'])]
-
-
-def _remove_sanitizer_tests(tests):
-  """
-  Filters out sanitizer tests
-  :param tests: list of all tests generated by run_tests_matrix.py
+  :param skippable_labels: list of languages and platforms with skippable tests
   :return: list of relevant tests
   """
-  # todo(mattkwong): find a more reliable way to filter tests - don't use shortname
-  return [test for test in tests if
-          all(san_tag not in test.shortname for san_tag in ['_asan', '_tsan', '_msan'])]
+  # test.labels[0] is platform and test.labels[2] is language
+  # We skip a test if both are considered safe to skip
+  return [test for test in tests if test.labels[0] not in skippable_labels or \
+          test.labels[2] not in skippable_labels]
 
 
 def filter_tests(tests, base_branch):
@@ -158,7 +171,7 @@ def filter_tests(tests, base_branch):
   :param tests: list of all tests generated by run_tests_matrix.py
   :return: list of relevant tests
   """
-  print("Finding file differences between %s repo and current branch...\n" % base_branch)
+  print("Finding file differences between gRPC %s branch and pull request...\n" % base_branch)
   changed_files = _get_changed_files(base_branch)
   for changed_file in changed_files:
     print(changed_file)
@@ -170,15 +183,13 @@ def filter_tests(tests, base_branch):
   for changed_file in changed_files:
     if not re.match(all_triggers, changed_file):
       return(tests)
-  # Filter out tests by language
-  for test_suite in _all_test_suites:
+  # Figure out which language and platform tests to run
+  skippable_labels = []
+  for test_suite in _ALL_TEST_SUITES:
     if _can_skip_tests(changed_files, test_suite.triggers):
-      for tag in test_suite.tags:
-        print("  Filtering %s tests" % tag)
-        tests = _remove_irrelevant_tests(tests, tag)
-  # Sanitizer tests skipped if core and c++ are skipped
-  if _can_skip_tests(changed_files, _cpp_test_suite.triggers + _core_test_suite.triggers):
-    print("  Filtering Sanitizer tests")
-    tests = _remove_sanitizer_tests(tests)
+      for label in test_suite.labels:
+        print("  Filtering %s tests" % label)
+        skippable_labels.append(label)
 
+  tests = _remove_irrelevant_tests(tests, skippable_labels)
   return tests