mirror of https://github.com/grpc/grpc.git
Add endpoint binder pool for client channel creation (#27755)
The pool serves as a buffer for interaction between C++ and Java. The buffer can let us avoid calling into Java code in channel connector implementation (which has not been merged yet), simplifies interaction between C++ and Java. Temporary changes are made to channel_create.cc to keep the example apps working but they will be rewrite after we start creating client channel instead of direct channel.pull/27530/head
Ming-Chuan
3 years ago
committed by
GitHub
parent
3181f4e530
commit
4d61638857
15 changed files with 600 additions and 49 deletions
@ -0,0 +1,108 @@ |
||||
// Copyright 2021 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 <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/ext/transport/binder/client/endpoint_binder_pool.h" |
||||
|
||||
#include "src/core/ext/transport/binder/client/jni_utils.h" |
||||
|
||||
#ifdef GPR_SUPPORT_BINDER_TRANSPORT |
||||
|
||||
#include <jni.h> |
||||
|
||||
#include "src/core/ext/transport/binder/wire_format/binder_android.h" |
||||
|
||||
extern "C" { |
||||
// Adds endpoint binder to binder pool when Java notify us that the endpoint
|
||||
// binder is ready. This is called from GrpcBinderConnection.java
|
||||
JNIEXPORT void JNICALL |
||||
Java_io_grpc_binder_cpp_GrpcBinderConnection_notifyConnected__Ljava_lang_String_2Landroid_os_IBinder_2( |
||||
JNIEnv* jni_env, jobject, jstring conn_id_jstring, jobject ibinder) { |
||||
jboolean isCopy; |
||||
const char* conn_id = jni_env->GetStringUTFChars(conn_id_jstring, &isCopy); |
||||
gpr_log(GPR_ERROR, "%s called with conn_id = %s", __func__, conn_id); |
||||
GPR_ASSERT(ibinder != nullptr); |
||||
ndk::SpAIBinder aibinder = grpc_binder::FromJavaBinder(jni_env, ibinder); |
||||
gpr_log(GPR_ERROR, "aibinder = %p", aibinder.get()); |
||||
auto b = absl::make_unique<grpc_binder::BinderAndroid>(aibinder); |
||||
GPR_ASSERT(b != nullptr); |
||||
grpc_binder::GetEndpointBinderPool()->AddEndpointBinder(conn_id, |
||||
std::move(b)); |
||||
if (isCopy == JNI_TRUE) { |
||||
jni_env->ReleaseStringUTFChars(conn_id_jstring, conn_id); |
||||
} |
||||
} |
||||
} |
||||
|
||||
#endif // GPR_SUPPORT_BINDER_TRANSPORT
|
||||
|
||||
namespace grpc_binder { |
||||
|
||||
void EndpointBinderPool ::GetEndpointBinder( |
||||
std::string conn_id, |
||||
std::function<void(std::unique_ptr<grpc_binder::Binder>)> cb) { |
||||
gpr_log(GPR_ERROR, "GetEndpointBinder %s", conn_id.c_str()); |
||||
std::unique_ptr<grpc_binder::Binder> b; |
||||
{ |
||||
grpc_core::MutexLock l(&m_); |
||||
if (binder_map_.count(conn_id)) { |
||||
b = std::move(binder_map_[conn_id]); |
||||
binder_map_.erase(conn_id); |
||||
GPR_ASSERT(b != nullptr); |
||||
} else { |
||||
if (pending_requests_.count(conn_id) != 0) { |
||||
gpr_log(GPR_ERROR, "Duplicate GetEndpointBinder request. conn_id = %s", |
||||
conn_id.c_str()); |
||||
return; |
||||
} |
||||
pending_requests_[conn_id] = std::move(cb); |
||||
return; |
||||
} |
||||
} |
||||
GPR_ASSERT(b != nullptr); |
||||
cb(std::move(b)); |
||||
} |
||||
|
||||
void EndpointBinderPool::AddEndpointBinder( |
||||
std::string conn_id, std::unique_ptr<grpc_binder::Binder> b) { |
||||
gpr_log(GPR_ERROR, "AddEndpointBinder %s", conn_id.c_str()); |
||||
GPR_ASSERT(b != nullptr); |
||||
// cb will be set in the following block if there is a pending callback
|
||||
std::function<void(std::unique_ptr<grpc_binder::Binder>)> cb = nullptr; |
||||
{ |
||||
grpc_core::MutexLock l(&m_); |
||||
if (binder_map_.count(conn_id) != 0) { |
||||
gpr_log(GPR_ERROR, "EndpointBinder already in the pool. conn_id = %s", |
||||
conn_id.c_str()); |
||||
return; |
||||
} |
||||
if (pending_requests_.count(conn_id)) { |
||||
cb = std::move(pending_requests_[conn_id]); |
||||
pending_requests_.erase(conn_id); |
||||
} else { |
||||
binder_map_[conn_id] = std::move(b); |
||||
b = nullptr; |
||||
} |
||||
} |
||||
if (cb != nullptr) { |
||||
cb(std::move(b)); |
||||
} |
||||
} |
||||
|
||||
EndpointBinderPool* GetEndpointBinderPool() { |
||||
static EndpointBinderPool* p = new EndpointBinderPool(); |
||||
return p; |
||||
} |
||||
} // namespace grpc_binder
|
||||
@ -0,0 +1,65 @@ |
||||
// Copyright 2021 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.
|
||||
|
||||
#ifndef GRPC_CORE_EXT_TRANSPORT_BINDER_CLIENT_ENDPOINT_BINDER_POOL_H |
||||
#define GRPC_CORE_EXT_TRANSPORT_BINDER_CLIENT_ENDPOINT_BINDER_POOL_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <functional> |
||||
#include <string> |
||||
|
||||
#include "absl/container/flat_hash_map.h" |
||||
|
||||
#include "src/core/ext/transport/binder/wire_format/binder.h" |
||||
#include "src/core/lib/gprpp/sync.h" |
||||
|
||||
namespace grpc_binder { |
||||
|
||||
// This class serves as a buffer of endpoint binders between C++ and
|
||||
// Java. `AddEndpointBinder` will be indirectly invoked by Java code, and
|
||||
// `GetEndpointBinder` is for C++ code to register callback to get endpoint
|
||||
// binder when become available. This simplifies JNI related threading issues
|
||||
// since both side only need to interact with this buffer in non-blocking
|
||||
// manner and avoids cross-language callbacks.
|
||||
class EndpointBinderPool { |
||||
public: |
||||
// Invokes the callback when the binder corresponding to the conn_id become
|
||||
// available. If the binder is already available, invokes the callback
|
||||
// immediately.
|
||||
// Ownership of the endpoint binder will be transferred to the callback
|
||||
// function and it will be removed from the pool
|
||||
void GetEndpointBinder( |
||||
std::string conn_id, |
||||
std::function<void(std::unique_ptr<grpc_binder::Binder>)> cb); |
||||
|
||||
// Add an endpoint binder to the pool
|
||||
void AddEndpointBinder(std::string conn_id, |
||||
std::unique_ptr<grpc_binder::Binder> b); |
||||
|
||||
private: |
||||
grpc_core::Mutex m_; |
||||
absl::flat_hash_map<std::string, std::unique_ptr<grpc_binder::Binder>> |
||||
binder_map_ ABSL_GUARDED_BY(m_); |
||||
absl::flat_hash_map<std::string, |
||||
std::function<void(std::unique_ptr<grpc_binder::Binder>)>> |
||||
pending_requests_ ABSL_GUARDED_BY(m_); |
||||
}; |
||||
|
||||
// Returns the singleton
|
||||
EndpointBinderPool* GetEndpointBinderPool(); |
||||
|
||||
} // namespace grpc_binder
|
||||
|
||||
#endif // GRPC_CORE_EXT_TRANSPORT_BINDER_CLIENT_ENDPOINT_BINDER_POOL_H
|
||||
@ -0,0 +1,76 @@ |
||||
// Copyright 2021 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 "src/core/ext/transport/binder/client/endpoint_binder_pool.h" |
||||
|
||||
#include <cassert> |
||||
#include <string> |
||||
#include <utility> |
||||
#include <vector> |
||||
|
||||
#include <gmock/gmock.h> |
||||
#include <gtest/gtest.h> |
||||
|
||||
#include "absl/memory/memory.h" |
||||
|
||||
#include "src/core/ext/transport/binder/client/endpoint_binder_pool.h" |
||||
#include "test/core/transport/binder/mock_objects.h" |
||||
#include "test/core/util/test_config.h" |
||||
|
||||
namespace grpc_binder { |
||||
|
||||
class CallbackChecker { |
||||
public: |
||||
MOCK_METHOD(void, Cb, (std::unique_ptr<grpc_binder::Binder>), ()); |
||||
}; |
||||
|
||||
TEST(EndpointBinderPoolTest, AddBeforeGet) { |
||||
EndpointBinderPool pool; |
||||
auto b = absl::make_unique<grpc_binder::MockBinder>(); |
||||
CallbackChecker cc; |
||||
pool.AddEndpointBinder("test", std::move(b)); |
||||
// TODO(mingcl): Use pointer matcher to verify it is `b` being passed back
|
||||
// here. It is only available in newer gtest version
|
||||
EXPECT_CALL(cc, Cb(testing::_)); |
||||
pool.GetEndpointBinder( |
||||
"test", std::bind(&CallbackChecker::Cb, &cc, std::placeholders::_1)); |
||||
} |
||||
|
||||
TEST(EndpointBinderPoolTest, GetBeforeAdd) { |
||||
EndpointBinderPool pool; |
||||
auto b = absl::make_unique<grpc_binder::MockBinder>(); |
||||
CallbackChecker cc; |
||||
EXPECT_CALL(cc, Cb(testing::_)).Times(0); |
||||
pool.GetEndpointBinder( |
||||
"test", std::bind(&CallbackChecker::Cb, &cc, std::placeholders::_1)); |
||||
EXPECT_CALL(cc, Cb(testing::_)).Times(1); |
||||
pool.AddEndpointBinder("test", std::move(b)); |
||||
} |
||||
|
||||
TEST(EndpointBinderPoolTest, ExpectNotCalled) { |
||||
EndpointBinderPool pool; |
||||
auto b = absl::make_unique<grpc_binder::MockBinder>(); |
||||
CallbackChecker cc; |
||||
EXPECT_CALL(cc, Cb(testing::_)).Times(0); |
||||
pool.GetEndpointBinder( |
||||
"test", std::bind(&CallbackChecker::Cb, &cc, std::placeholders::_1)); |
||||
} |
||||
|
||||
} // namespace grpc_binder
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
grpc::testing::TestEnvironment env(argc, argv); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
Loading…
Reference in new issue