mirror of https://github.com/grpc/grpc.git
Merge pull request #16492 from vjpai/client_callback
EXPERIMENTAL: C++ generic client-side unary callback APIpull/16541/head^2
Vijay Pai
7 years ago
committed by
GitHub
commit
3bc10c0f44
28 changed files with 858 additions and 10 deletions
@ -0,0 +1,103 @@ |
||||
/*
|
||||
* |
||||
* 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. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H |
||||
#define GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H |
||||
|
||||
#include <functional> |
||||
|
||||
#include <grpcpp/impl/codegen/call.h> |
||||
#include <grpcpp/impl/codegen/channel_interface.h> |
||||
#include <grpcpp/impl/codegen/config.h> |
||||
#include <grpcpp/impl/codegen/core_codegen_interface.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
// Forward declarations
|
||||
namespace grpc_core { |
||||
class CQCallbackInterface; |
||||
}; |
||||
|
||||
namespace grpc { |
||||
namespace internal { |
||||
|
||||
class CallbackWithStatusTag { |
||||
public: |
||||
// always allocated against a call arena, no memory free required
|
||||
static void operator delete(void* ptr, std::size_t size) { |
||||
assert(size == sizeof(CallbackWithStatusTag)); |
||||
} |
||||
|
||||
// This operator should never be called as the memory should be freed as part
|
||||
// of the arena destruction. It only exists to provide a matching operator
|
||||
// delete to the operator new so that some compilers will not complain (see
|
||||
// https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
|
||||
// there are no tests catching the compiler warning.
|
||||
static void operator delete(void*, void*) { assert(0); } |
||||
|
||||
CallbackWithStatusTag(grpc_call* call, std::function<void(Status)> f, |
||||
CompletionQueueTag* ops); |
||||
~CallbackWithStatusTag() {} |
||||
void* tag() { return static_cast<void*>(impl_); } |
||||
Status* status_ptr() { return status_; } |
||||
CompletionQueueTag* ops() { return ops_; } |
||||
|
||||
// force_run can not be performed on a tag if operations using this tag
|
||||
// have been sent to PerformOpsOnCall. It is intended for error conditions
|
||||
// that are detected before the operations are internally processed.
|
||||
void force_run(Status s); |
||||
|
||||
private: |
||||
grpc_core::CQCallbackInterface* impl_; |
||||
Status* status_; |
||||
CompletionQueueTag* ops_; |
||||
}; |
||||
|
||||
class CallbackWithSuccessTag { |
||||
public: |
||||
// always allocated against a call arena, no memory free required
|
||||
static void operator delete(void* ptr, std::size_t size) { |
||||
assert(size == sizeof(CallbackWithSuccessTag)); |
||||
} |
||||
|
||||
// This operator should never be called as the memory should be freed as part
|
||||
// of the arena destruction. It only exists to provide a matching operator
|
||||
// delete to the operator new so that some compilers will not complain (see
|
||||
// https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
|
||||
// there are no tests catching the compiler warning.
|
||||
static void operator delete(void*, void*) { assert(0); } |
||||
|
||||
CallbackWithSuccessTag(grpc_call* call, std::function<void(bool)> f, |
||||
CompletionQueueTag* ops); |
||||
|
||||
void* tag() { return static_cast<void*>(impl_); } |
||||
CompletionQueueTag* ops() { return ops_; } |
||||
|
||||
// force_run can not be performed on a tag if operations using this tag
|
||||
// have been sent to PerformOpsOnCall. It is intended for error conditions
|
||||
// that are detected before the operations are internally processed.
|
||||
void force_run(bool ok); |
||||
|
||||
private: |
||||
grpc_core::CQCallbackInterface* impl_; |
||||
CompletionQueueTag* ops_; |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H
|
||||
@ -0,0 +1,95 @@ |
||||
/*
|
||||
* |
||||
* 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. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H |
||||
#define GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H |
||||
|
||||
#include <functional> |
||||
|
||||
#include <grpcpp/impl/codegen/call.h> |
||||
#include <grpcpp/impl/codegen/callback_common.h> |
||||
#include <grpcpp/impl/codegen/channel_interface.h> |
||||
#include <grpcpp/impl/codegen/config.h> |
||||
#include <grpcpp/impl/codegen/core_codegen_interface.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
namespace grpc { |
||||
|
||||
class Channel; |
||||
class ClientContext; |
||||
class CompletionQueue; |
||||
|
||||
namespace internal { |
||||
class RpcMethod; |
||||
|
||||
/// Perform a callback-based unary call
|
||||
/// TODO(vjpai): Combine as much as possible with the blocking unary call code
|
||||
template <class InputMessage, class OutputMessage> |
||||
void CallbackUnaryCall(ChannelInterface* channel, const RpcMethod& method, |
||||
ClientContext* context, const InputMessage* request, |
||||
OutputMessage* result, |
||||
std::function<void(Status)> on_completion) { |
||||
CallbackUnaryCallImpl<InputMessage, OutputMessage> x( |
||||
channel, method, context, request, result, on_completion); |
||||
} |
||||
|
||||
template <class InputMessage, class OutputMessage> |
||||
class CallbackUnaryCallImpl { |
||||
public: |
||||
CallbackUnaryCallImpl(ChannelInterface* channel, const RpcMethod& method, |
||||
ClientContext* context, const InputMessage* request, |
||||
OutputMessage* result, |
||||
std::function<void(Status)> on_completion) { |
||||
CompletionQueue* cq = channel->CallbackCQ(); |
||||
GPR_CODEGEN_ASSERT(cq != nullptr); |
||||
Call call(channel->CreateCall(method, context, cq)); |
||||
|
||||
using FullCallOpSet = |
||||
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, |
||||
CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>, |
||||
CallOpClientSendClose, CallOpClientRecvStatus>; |
||||
|
||||
auto* ops = new (g_core_codegen_interface->grpc_call_arena_alloc( |
||||
call.call(), sizeof(FullCallOpSet))) FullCallOpSet; |
||||
|
||||
auto* tag = new (g_core_codegen_interface->grpc_call_arena_alloc( |
||||
call.call(), sizeof(CallbackWithStatusTag))) |
||||
CallbackWithStatusTag(call.call(), on_completion, ops); |
||||
|
||||
// TODO(vjpai): Unify code with sync API as much as possible
|
||||
Status s = ops->SendMessage(*request); |
||||
if (!s.ok()) { |
||||
tag->force_run(s); |
||||
return; |
||||
} |
||||
ops->SendInitialMetadata(context->send_initial_metadata_, |
||||
context->initial_metadata_flags()); |
||||
ops->RecvInitialMetadata(context); |
||||
ops->RecvMessage(result); |
||||
ops->AllowNoMessage(); |
||||
ops->ClientSendClose(); |
||||
ops->ClientRecvStatus(context, tag->status_ptr()); |
||||
ops->set_cq_tag(tag->tag()); |
||||
call.PerformOps(ops); |
||||
} |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H
|
||||
@ -0,0 +1,24 @@ |
||||
/*
|
||||
* |
||||
* 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. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPCPP_SUPPORT_CLIENT_CALLBACK_H |
||||
#define GRPCPP_SUPPORT_CLIENT_CALLBACK_H |
||||
|
||||
#include <grpcpp/impl/codegen/client_callback.h> |
||||
|
||||
#endif // GRPCPP_SUPPORT_CLIENT_CALLBACK_H
|
||||
@ -0,0 +1,131 @@ |
||||
/*
|
||||
* |
||||
* 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 <functional> |
||||
|
||||
#include <grpcpp/impl/codegen/callback_common.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
#include "src/core/lib/gprpp/memory.h" |
||||
#include "src/core/lib/surface/completion_queue.h" |
||||
|
||||
namespace grpc { |
||||
namespace internal { |
||||
|
||||
namespace { |
||||
class CallbackWithSuccessImpl : public grpc_core::CQCallbackInterface { |
||||
public: |
||||
static void operator delete(void* ptr, std::size_t size) { |
||||
assert(size == sizeof(CallbackWithSuccessImpl)); |
||||
} |
||||
|
||||
// This operator should never be called as the memory should be freed as part
|
||||
// of the arena destruction. It only exists to provide a matching operator
|
||||
// delete to the operator new so that some compilers will not complain (see
|
||||
// https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
|
||||
// there are no tests catching the compiler warning.
|
||||
static void operator delete(void*, void*) { assert(0); } |
||||
|
||||
CallbackWithSuccessImpl(grpc_call* call, CallbackWithSuccessTag* parent, |
||||
std::function<void(bool)> f) |
||||
: call_(call), parent_(parent), func_(std::move(f)) { |
||||
grpc_call_ref(call); |
||||
} |
||||
|
||||
void Run(bool ok) override { |
||||
void* ignored = parent_->ops(); |
||||
bool new_ok = ok; |
||||
GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &new_ok)); |
||||
GPR_ASSERT(ignored == parent_->ops()); |
||||
func_(ok); |
||||
func_ = nullptr; // release the function
|
||||
grpc_call_unref(call_); |
||||
} |
||||
|
||||
private: |
||||
grpc_call* call_; |
||||
CallbackWithSuccessTag* parent_; |
||||
std::function<void(bool)> func_; |
||||
}; |
||||
|
||||
class CallbackWithStatusImpl : public grpc_core::CQCallbackInterface { |
||||
public: |
||||
static void operator delete(void* ptr, std::size_t size) { |
||||
assert(size == sizeof(CallbackWithStatusImpl)); |
||||
} |
||||
|
||||
// This operator should never be called as the memory should be freed as part
|
||||
// of the arena destruction. It only exists to provide a matching operator
|
||||
// delete to the operator new so that some compilers will not complain (see
|
||||
// https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
|
||||
// there are no tests catching the compiler warning.
|
||||
static void operator delete(void*, void*) { assert(0); } |
||||
|
||||
CallbackWithStatusImpl(grpc_call* call, CallbackWithStatusTag* parent, |
||||
std::function<void(Status)> f) |
||||
: call_(call), parent_(parent), func_(std::move(f)), status_() { |
||||
grpc_call_ref(call); |
||||
} |
||||
|
||||
void Run(bool ok) override { |
||||
void* ignored = parent_->ops(); |
||||
|
||||
GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &ok)); |
||||
GPR_ASSERT(ignored == parent_->ops()); |
||||
|
||||
func_(status_); |
||||
func_ = nullptr; // release the function
|
||||
grpc_call_unref(call_); |
||||
} |
||||
Status* status_ptr() { return &status_; } |
||||
|
||||
private: |
||||
grpc_call* call_; |
||||
CallbackWithStatusTag* parent_; |
||||
std::function<void(Status)> func_; |
||||
Status status_; |
||||
}; |
||||
|
||||
} // namespace
|
||||
|
||||
CallbackWithSuccessTag::CallbackWithSuccessTag(grpc_call* call, |
||||
std::function<void(bool)> f, |
||||
CompletionQueueTag* ops) |
||||
: impl_(new (grpc_call_arena_alloc(call, sizeof(CallbackWithSuccessImpl))) |
||||
CallbackWithSuccessImpl(call, this, std::move(f))), |
||||
ops_(ops) {} |
||||
|
||||
void CallbackWithSuccessTag::force_run(bool ok) { impl_->Run(ok); } |
||||
|
||||
CallbackWithStatusTag::CallbackWithStatusTag(grpc_call* call, |
||||
std::function<void(Status)> f, |
||||
CompletionQueueTag* ops) |
||||
: ops_(ops) { |
||||
auto* impl = new (grpc_call_arena_alloc(call, sizeof(CallbackWithStatusImpl))) |
||||
CallbackWithStatusImpl(call, this, std::move(f)); |
||||
impl_ = impl; |
||||
status_ = impl->status_ptr(); |
||||
} |
||||
|
||||
void CallbackWithStatusTag::force_run(Status s) { |
||||
*status_ = std::move(s); |
||||
impl_->Run(true); |
||||
} |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,126 @@ |
||||
/*
|
||||
* |
||||
* 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 <functional> |
||||
#include <mutex> |
||||
|
||||
#include <grpcpp/channel.h> |
||||
#include <grpcpp/client_context.h> |
||||
#include <grpcpp/create_channel.h> |
||||
#include <grpcpp/generic/generic_stub.h> |
||||
#include <grpcpp/impl/codegen/proto_utils.h> |
||||
#include <grpcpp/server.h> |
||||
#include <grpcpp/server_builder.h> |
||||
#include <grpcpp/server_context.h> |
||||
#include <grpcpp/support/client_callback.h> |
||||
|
||||
#include "src/proto/grpc/testing/echo.grpc.pb.h" |
||||
#include "test/core/util/test_config.h" |
||||
#include "test/cpp/end2end/test_service_impl.h" |
||||
#include "test/cpp/util/byte_buffer_proto_helper.h" |
||||
|
||||
#include <gtest/gtest.h> |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
namespace { |
||||
|
||||
class ClientCallbackEnd2endTest : public ::testing::Test { |
||||
protected: |
||||
ClientCallbackEnd2endTest() {} |
||||
|
||||
void SetUp() override { |
||||
ServerBuilder builder; |
||||
builder.RegisterService(&service_); |
||||
|
||||
server_ = builder.BuildAndStart(); |
||||
is_server_started_ = true; |
||||
} |
||||
|
||||
void ResetStub() { |
||||
ChannelArguments args; |
||||
channel_ = server_->InProcessChannel(args); |
||||
stub_.reset(new GenericStub(channel_)); |
||||
} |
||||
|
||||
void TearDown() override { |
||||
if (is_server_started_) { |
||||
server_->Shutdown(); |
||||
} |
||||
} |
||||
|
||||
void SendRpcs(int num_rpcs) { |
||||
const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo"); |
||||
grpc::string test_string(""); |
||||
for (int i = 0; i < num_rpcs; i++) { |
||||
EchoRequest request; |
||||
std::unique_ptr<ByteBuffer> send_buf; |
||||
ByteBuffer recv_buf; |
||||
ClientContext cli_ctx; |
||||
|
||||
test_string += "Hello world. "; |
||||
request.set_message(test_string); |
||||
send_buf = SerializeToByteBuffer(&request); |
||||
|
||||
std::mutex mu; |
||||
std::condition_variable cv; |
||||
bool done = false; |
||||
stub_->experimental().UnaryCall( |
||||
&cli_ctx, kMethodName, send_buf.get(), &recv_buf, |
||||
[&request, &recv_buf, &done, &mu, &cv](Status s) { |
||||
GPR_ASSERT(s.ok()); |
||||
|
||||
EchoResponse response; |
||||
EXPECT_TRUE(ParseFromByteBuffer(&recv_buf, &response)); |
||||
EXPECT_EQ(request.message(), response.message()); |
||||
std::lock_guard<std::mutex> l(mu); |
||||
done = true; |
||||
cv.notify_one(); |
||||
}); |
||||
std::unique_lock<std::mutex> l(mu); |
||||
while (!done) { |
||||
cv.wait(l); |
||||
} |
||||
} |
||||
} |
||||
bool is_server_started_; |
||||
std::shared_ptr<Channel> channel_; |
||||
std::unique_ptr<grpc::GenericStub> stub_; |
||||
TestServiceImpl service_; |
||||
std::unique_ptr<Server> server_; |
||||
}; |
||||
|
||||
TEST_F(ClientCallbackEnd2endTest, SimpleRpc) { |
||||
ResetStub(); |
||||
SendRpcs(1); |
||||
} |
||||
|
||||
TEST_F(ClientCallbackEnd2endTest, SequentialRpcs) { |
||||
ResetStub(); |
||||
SendRpcs(10); |
||||
} |
||||
|
||||
} // namespace
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
|
||||
int main(int argc, char** argv) { |
||||
grpc_test_init(argc, argv); |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
Loading…
Reference in new issue