Merge remote-tracking branch 'upstream/master' into fix-authorizer

Author: Muxi Yan

examples/BUILD

@@ -101,7 +101,8 @@ cc_binary(
 
 cc_binary(
     name = "keyvaluestore_client",
-    srcs = ["cpp/keyvaluestore/client.cc"],
+    srcs = ["cpp/keyvaluestore/caching_interceptor.h",
+            "cpp/keyvaluestore/client.cc"],    
     defines = ["BAZEL_BUILD"],
     deps = [":keyvaluestore", "//:grpc++"],
 )

examples/cpp/keyvaluestore/caching_interceptor.h

@@ -0,0 +1,134 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include <map>
+
+#include <grpcpp/support/client_interceptor.h>
+
+#ifdef BAZEL_BUILD
+#include "examples/protos/keyvaluestore.grpc.pb.h"
+#else
+#include "keyvaluestore.grpc.pb.h"
+#endif
+
+// This is a naive implementation of a cache. A new cache is for each call. For
+// each new key request, the key is first searched in the map and if found, the
+// interceptor fills in the return value without making a request to the server.
+// Only if the key is not found in the cache do we make a request.
+class CachingInterceptor : public grpc::experimental::Interceptor {
+ public:
+  CachingInterceptor(grpc::experimental::ClientRpcInfo* info) {}
+
+  void Intercept(
+      ::grpc::experimental::InterceptorBatchMethods* methods) override {
+    bool hijack = false;
+    if (methods->QueryInterceptionHookPoint(
+            grpc::experimental::InterceptionHookPoints::
+                PRE_SEND_INITIAL_METADATA)) {
+      // Hijack all calls
+      hijack = true;
+      // Create a stream on which this interceptor can make requests
+      stub_ = keyvaluestore::KeyValueStore::NewStub(
+          methods->GetInterceptedChannel());
+      stream_ = stub_->GetValues(&context_);
+    }
+    if (methods->QueryInterceptionHookPoint(
+            grpc::experimental::InterceptionHookPoints::PRE_SEND_MESSAGE)) {
+      // We know that clients perform a Read and a Write in a loop, so we don't
+      // need to maintain a list of the responses.
+      std::string requested_key;
+      const keyvaluestore::Request* req_msg =
+          static_cast<const keyvaluestore::Request*>(methods->GetSendMessage());
+      if (req_msg != nullptr) {
+        requested_key = req_msg->key();
+      } else {
+        // The non-serialized form would not be available in certain scenarios,
+        // so add a fallback
+        keyvaluestore::Request req_msg;
+        auto* buffer = methods->GetSerializedSendMessage();
+        auto copied_buffer = *buffer;
+        GPR_ASSERT(
+            grpc::SerializationTraits<keyvaluestore::Request>::Deserialize(
+                &copied_buffer, &req_msg)
+                .ok());
+        requested_key = req_msg.key();
+      }
+
+      // Check if the key is present in the map
+      auto search = cached_map_.find(requested_key);
+      if (search != cached_map_.end()) {
+        std::cout << "Key " << requested_key << "found in map";
+        response_ = search->second;
+      } else {
+        std::cout << "Key " << requested_key << "not found in cache";
+        // Key was not found in the cache, so make a request
+        keyvaluestore::Request req;
+        req.set_key(requested_key);
+        stream_->Write(req);
+        keyvaluestore::Response resp;
+        stream_->Read(&resp);
+        response_ = resp.value();
+        // Insert the pair in the cache for future requests
+        cached_map_.insert({requested_key, response_});
+      }
+    }
+    if (methods->QueryInterceptionHookPoint(
+            grpc::experimental::InterceptionHookPoints::PRE_SEND_CLOSE)) {
+      stream_->WritesDone();
+    }
+    if (methods->QueryInterceptionHookPoint(
+            grpc::experimental::InterceptionHookPoints::PRE_RECV_MESSAGE)) {
+      keyvaluestore::Response* resp =
+          static_cast<keyvaluestore::Response*>(methods->GetRecvMessage());
+      resp->set_value(response_);
+    }
+    if (methods->QueryInterceptionHookPoint(
+            grpc::experimental::InterceptionHookPoints::PRE_RECV_STATUS)) {
+      auto* status = methods->GetRecvStatus();
+      *status = grpc::Status::OK;
+    }
+    // One of Hijack or Proceed always needs to be called to make progress.
+    if (hijack) {
+      // Hijack is called only once when PRE_SEND_INITIAL_METADATA is present in
+      // the hook points
+      methods->Hijack();
+    } else {
+      // Proceed is an indicator that the interceptor is done intercepting the
+      // batch.
+      methods->Proceed();
+    }
+  }
+
+ private:
+  grpc::ClientContext context_;
+  std::unique_ptr<keyvaluestore::KeyValueStore::Stub> stub_;
+  std::unique_ptr<
+      grpc::ClientReaderWriter<keyvaluestore::Request, keyvaluestore::Response>>
+      stream_;
+  std::map<std::string, std::string> cached_map_;
+  std::string response_;
+};
+
+class CachingInterceptorFactory
+    : public grpc::experimental::ClientInterceptorFactoryInterface {
+ public:
+  grpc::experimental::Interceptor* CreateClientInterceptor(
+      grpc::experimental::ClientRpcInfo* info) override {
+    return new CachingInterceptor(info);
+  }
+};

examples/cpp/keyvaluestore/client.cc

@@ -23,6 +23,8 @@
 
 #include <grpcpp/grpcpp.h>
 
+#include "caching_interceptor.h"
+
 #ifdef BAZEL_BUILD
 #include "examples/protos/keyvaluestore.grpc.pb.h"
 #else
@@ -77,9 +79,20 @@ int main(int argc, char** argv) {
   // are created. This channel models a connection to an endpoint (in this case,
   // localhost at port 50051). We indicate that the channel isn't authenticated
   // (use of InsecureChannelCredentials()).
-  KeyValueStoreClient client(grpc::CreateChannel(
-      "localhost:50051", grpc::InsecureChannelCredentials()));
-  std::vector<std::string> keys = {"key1", "key2", "key3", "key4", "key5"};
+  // In this example, we are using a cache which has been added in as an
+  // interceptor.
+  grpc::ChannelArguments args;
+  std::vector<
+      std::unique_ptr<grpc::experimental::ClientInterceptorFactoryInterface>>
+      interceptor_creators;
+  interceptor_creators.push_back(std::unique_ptr<CachingInterceptorFactory>(
+      new CachingInterceptorFactory()));
+  auto channel = grpc::experimental::CreateCustomChannelWithInterceptors(
+      "localhost:50051", grpc::InsecureChannelCredentials(), args,
+      std::move(interceptor_creators));
+  KeyValueStoreClient client(channel);
+  std::vector<std::string> keys = {"key1", "key2", "key3", "key4",
+                                   "key5", "key1", "key2", "key4"};
   client.GetValues(keys);
 
   return 0;

src/core/ext/transport/chttp2/transport/chttp2_transport.cc

@@ -968,19 +968,19 @@ static grpc_closure_scheduler* write_scheduler(grpc_chttp2_transport* t,
      get better latency overall if we switch writing work elsewhere and continue
      with application work above */
   if (!t->is_first_write_in_batch) {
-    return grpc_executor_scheduler(GRPC_EXECUTOR_SHORT);
+    return grpc_core::Executor::Scheduler(grpc_core::ExecutorJobType::SHORT);
   }
   /* equivalently, if it's a partial write, we *know* we're going to be taking a
      thread jump to write it because of the above, may as well do so
      immediately */
   if (partial_write) {
-    return grpc_executor_scheduler(GRPC_EXECUTOR_SHORT);
+    return grpc_core::Executor::Scheduler(grpc_core::ExecutorJobType::SHORT);
   }
   switch (t->opt_target) {
     case GRPC_CHTTP2_OPTIMIZE_FOR_THROUGHPUT:
       /* executor gives us the largest probability of being able to batch a
        * write with others on this transport */
-      return grpc_executor_scheduler(GRPC_EXECUTOR_SHORT);
+      return grpc_core::Executor::Scheduler(grpc_core::ExecutorJobType::SHORT);
     case GRPC_CHTTP2_OPTIMIZE_FOR_LATENCY:
       return grpc_schedule_on_exec_ctx;
   }

src/core/lib/iomgr/combiner.cc

@@ -83,8 +83,9 @@ grpc_combiner* grpc_combiner_create(void) {
   gpr_atm_no_barrier_store(&lock->state, STATE_UNORPHANED);
   gpr_mpscq_init(&lock->queue);
   grpc_closure_list_init(&lock->final_list);
-  GRPC_CLOSURE_INIT(&lock->offload, offload, lock,
-                    grpc_executor_scheduler(GRPC_EXECUTOR_SHORT));
+  GRPC_CLOSURE_INIT(
+      &lock->offload, offload, lock,
+      grpc_core::Executor::Scheduler(grpc_core::ExecutorJobType::SHORT));
   GRPC_COMBINER_TRACE(gpr_log(GPR_INFO, "C:%p create", lock));
   return lock;
 }
@@ -235,7 +236,7 @@ bool grpc_combiner_continue_exec_ctx() {
   // 3. the DEFAULT executor is threaded
   // 4. the current thread is not a worker for any background poller
   if (contended && grpc_core::ExecCtx::Get()->IsReadyToFinish() &&
-      grpc_executor_is_threaded() &&
+      grpc_core::Executor::IsThreadedDefault() &&
       !grpc_iomgr_is_any_background_poller_thread()) {
     GPR_TIMER_MARK("offload_from_finished_exec_ctx", 0);
     // this execution context wants to move on: schedule remaining work to be