Merge branch 'master' of https://github.com/grpc/grpc into cpp_bazelness

pull/9780/head
Nicolas "Pixel" Noble 8 years ago
commit 0dac09e724
  1. 25
      examples/ruby/errors_and_cancellation/README.md
  2. 117
      examples/ruby/errors_and_cancellation/error_examples_client.rb
  3. 76
      examples/ruby/errors_and_cancellation/error_examples_server.rb
  4. 247
      include/grpc++/impl/codegen/async_stream.h
  5. 48
      include/grpc++/impl/codegen/call.h
  6. 15
      include/grpc++/impl/codegen/client_context.h
  7. 106
      include/grpc++/impl/codegen/sync_stream.h
  8. 21
      include/grpc/impl/codegen/grpc_types.h
  9. 2
      src/compiler/README.md
  10. 2
      src/core/README.md
  11. 2
      src/core/ext/census/gen/README.md
  12. 83
      src/core/ext/transport/chttp2/transport/chttp2_transport.c
  13. 20
      src/core/ext/transport/chttp2/transport/frame_ping.c
  14. 3
      src/core/ext/transport/chttp2/transport/frame_ping.h
  15. 7
      src/core/ext/transport/chttp2/transport/internal.h
  16. 8
      src/core/ext/transport/chttp2/transport/writing.c
  17. 14
      src/core/lib/iomgr/resource_quota.c
  18. 2
      src/core/lib/surface/call.c
  19. 16
      src/cpp/README.md
  20. 3
      src/cpp/client/client_context.cc
  21. 58
      src/objective-c/GRPCClient/GRPCCall.m
  22. 61
      src/objective-c/GRPCClient/internal_testing/GRPCCall+InternalTests.h
  23. 52
      src/objective-c/GRPCClient/internal_testing/GRPCCall+InternalTests.m
  24. 59
      src/objective-c/GRPCClient/private/GRPCOpBatchLog.h
  25. 72
      src/objective-c/GRPCClient/private/GRPCOpBatchLog.m
  26. 8
      src/objective-c/GRPCClient/private/GRPCWrappedCall.m
  27. 50
      src/objective-c/RxLibrary/GRXImmediateSingleWriter.h
  28. 89
      src/objective-c/RxLibrary/GRXImmediateSingleWriter.m
  29. 3
      src/objective-c/RxLibrary/GRXWriter+Immediate.m
  30. 1
      src/objective-c/tests/GRPCClientTests.m
  31. 50
      src/objective-c/tests/InteropTests.m
  32. 1
      src/objective-c/tests/InteropTestsLocalCleartext.m
  33. 1
      src/objective-c/tests/InteropTestsLocalSSL.m
  34. 1
      src/objective-c/tests/InteropTestsRemote.m
  35. 4
      src/objective-c/tests/InteropTestsRemoteWithCronet/InteropTestsRemoteWithCronet.m
  36. 7
      src/objective-c/tests/Podfile
  37. 241
      src/objective-c/tests/Tests.xcodeproj/project.pbxproj
  38. 6
      src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme
  39. 6
      src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/InteropTestsLocalCleartext.xcscheme
  40. 6
      src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/InteropTestsLocalSSL.xcscheme
  41. 6
      src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/InteropTestsRemote.xcscheme
  42. 2
      src/php/README.md
  43. 2
      src/php/lib/Grpc/AbstractCall.php
  44. 2
      src/php/tests/unit_tests/ServerTest.php
  45. 2
      templates/tools/dockerfile/interoptest/grpc_interop_go/Dockerfile.template
  46. 2
      templates/tools/dockerfile/interoptest/grpc_interop_http2/Dockerfile.template
  47. 2
      templates/tools/dockerfile/stress_test/grpc_interop_stress_go/Dockerfile.template
  48. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/crash-59a56fa18034a104fb9f16cd58071b6ff93b8756
  49. BIN
      test/core/end2end/fuzzers/api_fuzzer_corpus/poc-c726ee220e980ed6ad17809fd9efe2844ee61555ac08e4f88afd8901cc2dd53a
  50. 20
      test/core/end2end/tests/keepalive_timeout.c
  51. 18
      test/core/end2end/tests/ping.c
  52. 52
      test/core/iomgr/resource_quota_test.c
  53. 319
      test/cpp/end2end/async_end2end_test.cc
  54. 85
      test/cpp/end2end/end2end_test.cc
  55. 2
      test/cpp/end2end/mock_test.cc
  56. 22
      test/cpp/end2end/test_service_impl.cc
  57. 2
      test/cpp/end2end/test_service_impl.h
  58. 191
      test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc
  59. 2
      tools/dockerfile/interoptest/grpc_interop_go/Dockerfile
  60. 2
      tools/dockerfile/interoptest/grpc_interop_http2/Dockerfile
  61. 2
      tools/dockerfile/stress_test/grpc_interop_stress_go/Dockerfile
  62. 46
      tools/run_tests/generated/tests.json

@ -0,0 +1,25 @@
#Errors and Cancelletion code samples for grpc-ruby
The examples in this directory show use of grpc errors.
On the server side, errors are returned from service
implementations by raising a certain `GRPC::BadStatus` exception.
On the client side, GRPC errors get raised when either:
* the call completes (unary and client-streaming call types)
* the response `Enumerable` is iterated through (server-streaming and
bidi call types).
## To run the examples here:
Start the server:
```
> ruby error_examples_server.rb
```
Then run the client:
```
> ruby error_examples_client.rb
```

@ -0,0 +1,117 @@
#!/usr/bin/env ruby
# Copyright 2015, Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
# * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
# Sample app that connects to an error-throwing implementation of
# Route Guide service.
#
# Usage: $ path/to/route_guide_client.rb
this_dir = File.expand_path(File.dirname(__FILE__))
lib_dir = File.join(File.dirname(this_dir), 'lib')
$LOAD_PATH.unshift(lib_dir) unless $LOAD_PATH.include?(lib_dir)
require 'grpc'
require 'route_guide_services_pb'
include Routeguide
def run_get_feature_expect_error(stub)
resp = stub.get_feature(Point.new)
end
def run_list_features_expect_error(stub)
resps = stub.list_features(Rectangle.new)
# NOOP iteration to pick up error
resps.each { }
end
def run_record_route_expect_error(stub)
stub.record_route([])
end
def run_route_chat_expect_error(stub)
resps = stub.route_chat([])
# NOOP iteration to pick up error
resps.each { }
end
def main
stub = RouteGuide::Stub.new('localhost:50051', :this_channel_is_insecure)
begin
run_get_feature_expect_error(stub)
rescue GRPC::BadStatus => e
puts "===== GetFeature exception: ====="
puts e.inspect
puts "e.code: #{e.code}"
puts "e.details: #{e.details}"
puts "e.metadata: #{e.metadata}"
puts "================================="
end
begin
run_list_features_expect_error(stub)
rescue GRPC::BadStatus => e
error = true
puts "===== ListFeatures exception: ====="
puts e.inspect
puts "e.code: #{e.code}"
puts "e.details: #{e.details}"
puts "e.metadata: #{e.metadata}"
puts "================================="
end
begin
run_route_chat_expect_error(stub)
rescue GRPC::BadStatus => e
puts "==== RouteChat exception: ===="
puts e.inspect
puts "e.code: #{e.code}"
puts "e.details: #{e.details}"
puts "e.metadata: #{e.metadata}"
puts "================================="
end
begin
run_record_route_expect_error(stub)
rescue GRPC::BadStatus => e
puts "==== RecordRoute exception: ===="
puts e.inspect
puts "e.code: #{e.code}"
puts "e.details: #{e.details}"
puts "e.metadata: #{e.metadata}"
puts "================================="
end
end
main

@ -0,0 +1,76 @@
#!/usr/bin/env ruby
# -*- coding: utf-8 -*-
# Copyright 2015, Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
# * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
# Error-throwing implementation of Route Guide service.
#
# Usage: $ path/to/route_guide_server.rb
this_dir = File.expand_path(File.dirname(__FILE__))
lib_dir = File.join(File.dirname(this_dir), 'lib')
$LOAD_PATH.unshift(lib_dir) unless $LOAD_PATH.include?(lib_dir)
require 'grpc'
require 'route_guide_services_pb'
include Routeguide
include GRPC::Core::StatusCodes
# CanellingandErrorReturningServiceImpl provides an implementation of the RouteGuide service.
class CancellingAndErrorReturningServerImpl < RouteGuide::Service
# def get_feature
# Note get_feature isn't implemented in this subclass, so the server
# will get a gRPC UNIMPLEMENTED error when it's called.
def list_features(rectangle, _call)
raise "string appears on the client in the 'details' field of a 'GRPC::Unknown' exception"
end
def record_route(call)
raise GRPC::BadStatus.new_status_exception(CANCELLED)
end
def route_chat(notes)
raise GRPC::BadStatus.new_status_exception(ABORTED, details = 'arbitrary', metadata = {somekey: 'val'})
end
end
def main
port = '0.0.0.0:50051'
s = GRPC::RpcServer.new
s.add_http2_port(port, :this_port_is_insecure)
GRPC.logger.info("... running insecurely on #{port}")
s.handle(CancellingAndErrorReturningServerImpl.new)
s.run_till_terminated
end
main

@ -101,6 +101,39 @@ class AsyncWriterInterface {
/// \param[in] msg The message to be written.
/// \param[in] tag The tag identifying the operation.
virtual void Write(const W& msg, void* tag) = 0;
/// Request the writing of \a msg using WriteOptions \a options with
/// identifying tag \a tag.
///
/// Only one write may be outstanding at any given time. This means that
/// after calling Write, one must wait to receive \a tag from the completion
/// queue BEFORE calling Write again.
/// WriteOptions \a options is used to set the write options of this message.
/// This is thread-safe with respect to \a Read
///
/// \param[in] msg The message to be written.
/// \param[in] options The WriteOptions to be used to write this message.
/// \param[in] tag The tag identifying the operation.
virtual void Write(const W& msg, WriteOptions options, void* tag) = 0;
/// Request the writing of \a msg and coalesce it with the writing
/// of trailing metadata, using WriteOptions \a options with identifying tag
/// \a tag.
///
/// For client, WriteLast is equivalent of performing Write and WritesDone in
/// a single step.
/// For server, WriteLast buffers the \a msg. The writing of \a msg is held
/// until Finish is called, where \a msg and trailing metadata are coalesced
/// and write is initiated. Note that WriteLast can only buffer \a msg up to
/// the flow control window size. If \a msg size is larger than the window
/// size, it will be sent on wire without buffering.
///
/// \param[in] msg The message to be written.
/// \param[in] options The WriteOptions to be used to write this message.
/// \param[in] tag The tag identifying the operation.
void WriteLast(const W& msg, WriteOptions options, void* tag) {
Write(msg, options.set_last_message(), tag);
}
};
template <class R>
@ -183,11 +216,17 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
: context_(context), call_(channel->CreateCall(method, context, cq)) {
finish_ops_.RecvMessage(response);
finish_ops_.AllowNoMessage();
init_ops_.set_output_tag(tag);
init_ops_.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
call_.PerformOps(&init_ops_);
// if corked bit is set in context, we buffer up the initial metadata to
// coalesce with later message to be sent. No op is performed.
if (context_->initial_metadata_corked_) {
write_ops_.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
} else {
write_ops_.set_output_tag(tag);
write_ops_.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
call_.PerformOps(&write_ops_);
}
}
void ReadInitialMetadata(void* tag) override {
@ -205,10 +244,21 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
call_.PerformOps(&write_ops_);
}
void Write(const W& msg, WriteOptions options, void* tag) override {
write_ops_.set_output_tag(tag);
if (options.is_last_message()) {
options.set_buffer_hint();
write_ops_.ClientSendClose();
}
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
call_.PerformOps(&write_ops_);
}
void WritesDone(void* tag) override {
writes_done_ops_.set_output_tag(tag);
writes_done_ops_.ClientSendClose();
call_.PerformOps(&writes_done_ops_);
write_ops_.set_output_tag(tag);
write_ops_.ClientSendClose();
call_.PerformOps(&write_ops_);
}
void Finish(Status* status, void* tag) override {
@ -223,10 +273,9 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
private:
ClientContext* context_;
Call call_;
CallOpSet<CallOpSendInitialMetadata> init_ops_;
CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
CallOpSet<CallOpSendMessage> write_ops_;
CallOpSet<CallOpClientSendClose> writes_done_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
write_ops_;
CallOpSet<CallOpRecvInitialMetadata, CallOpGenericRecvMessage,
CallOpClientRecvStatus>
finish_ops_;
@ -253,10 +302,17 @@ class ClientAsyncReaderWriter final
const RpcMethod& method, ClientContext* context,
void* tag)
: context_(context), call_(channel->CreateCall(method, context, cq)) {
init_ops_.set_output_tag(tag);
init_ops_.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
call_.PerformOps(&init_ops_);
if (context_->initial_metadata_corked_) {
// if corked bit is set in context, we buffer up the initial metadata to
// coalesce with later message to be sent. No op is performed.
write_ops_.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
} else {
write_ops_.set_output_tag(tag);
write_ops_.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
call_.PerformOps(&write_ops_);
}
}
void ReadInitialMetadata(void* tag) override {
@ -283,10 +339,21 @@ class ClientAsyncReaderWriter final
call_.PerformOps(&write_ops_);
}
void Write(const W& msg, WriteOptions options, void* tag) override {
write_ops_.set_output_tag(tag);
if (options.is_last_message()) {
options.set_buffer_hint();
write_ops_.ClientSendClose();
}
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
call_.PerformOps(&write_ops_);
}
void WritesDone(void* tag) override {
writes_done_ops_.set_output_tag(tag);
writes_done_ops_.ClientSendClose();
call_.PerformOps(&writes_done_ops_);
write_ops_.set_output_tag(tag);
write_ops_.ClientSendClose();
call_.PerformOps(&write_ops_);
}
void Finish(Status* status, void* tag) override {
@ -301,11 +368,10 @@ class ClientAsyncReaderWriter final
private:
ClientContext* context_;
Call call_;
CallOpSet<CallOpSendInitialMetadata> init_ops_;
CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> read_ops_;
CallOpSet<CallOpSendMessage> write_ops_;
CallOpSet<CallOpClientSendClose> writes_done_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
write_ops_;
CallOpSet<CallOpRecvInitialMetadata, CallOpClientRecvStatus> finish_ops_;
};
@ -395,6 +461,20 @@ class ServerAsyncWriterInterface : public ServerAsyncStreamingInterface,
public AsyncWriterInterface<W> {
public:
virtual void Finish(const Status& status, void* tag) = 0;
/// Request the writing of \a msg and coalesce it with trailing metadata which
/// contains \a status, using WriteOptions options with identifying tag \a
/// tag.
///
/// WriteAndFinish is equivalent of performing WriteLast and Finish in a
/// single step.
///
/// \param[in] msg The message to be written.
/// \param[in] options The WriteOptions to be used to write this message.
/// \param[in] status The Status that server returns to client.
/// \param[in] tag The tag identifying the operation.
virtual void WriteAndFinish(const W& msg, WriteOptions options,
const Status& status, void* tag) = 0;
};
template <class W>
@ -418,29 +498,37 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> {
void Write(const W& msg, void* tag) override {
write_ops_.set_output_tag(tag);
if (!ctx_->sent_initial_metadata_) {
write_ops_.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
write_ops_.set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
}
EnsureInitialMetadataSent(&write_ops_);
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
call_.PerformOps(&write_ops_);
}
void Write(const W& msg, WriteOptions options, void* tag) override {
write_ops_.set_output_tag(tag);
if (options.is_last_message()) {
options.set_buffer_hint();
}
EnsureInitialMetadataSent(&write_ops_);
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
call_.PerformOps(&write_ops_);
}
void WriteAndFinish(const W& msg, WriteOptions options, const Status& status,
void* tag) override {
write_ops_.set_output_tag(tag);
EnsureInitialMetadataSent(&write_ops_);
options.set_buffer_hint();
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
write_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
call_.PerformOps(&write_ops_);
}
void Finish(const Status& status, void* tag) override {
finish_ops_.set_output_tag(tag);
if (!ctx_->sent_initial_metadata_) {
finish_ops_.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
finish_ops_.set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
}
EnsureInitialMetadataSent(&finish_ops_);
finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
call_.PerformOps(&finish_ops_);
}
@ -448,10 +536,24 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> {
private:
void BindCall(Call* call) override { call_ = *call; }
template <class T>
void EnsureInitialMetadataSent(T* ops) {
if (!ctx_->sent_initial_metadata_) {
ops->SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
ops->set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
}
}
Call call_;
ServerContext* ctx_;
CallOpSet<CallOpSendInitialMetadata> meta_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> write_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpServerSendStatus>
write_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> finish_ops_;
};
@ -462,6 +564,20 @@ class ServerAsyncReaderWriterInterface : public ServerAsyncStreamingInterface,
public AsyncReaderInterface<R> {
public:
virtual void Finish(const Status& status, void* tag) = 0;
/// Request the writing of \a msg and coalesce it with trailing metadata which
/// contains \a status, using WriteOptions options with identifying tag \a
/// tag.
///
/// WriteAndFinish is equivalent of performing WriteLast and Finish in a
/// single step.
///
/// \param[in] msg The message to be written.
/// \param[in] options The WriteOptions to be used to write this message.
/// \param[in] status The Status that server returns to client.
/// \param[in] tag The tag identifying the operation.
virtual void WriteAndFinish(const W& msg, WriteOptions options,
const Status& status, void* tag) = 0;
};
template <class W, class R>
@ -492,29 +608,36 @@ class ServerAsyncReaderWriter final
void Write(const W& msg, void* tag) override {
write_ops_.set_output_tag(tag);
if (!ctx_->sent_initial_metadata_) {
write_ops_.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
write_ops_.set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
}
EnsureInitialMetadataSent(&write_ops_);
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
call_.PerformOps(&write_ops_);
}
void Write(const W& msg, WriteOptions options, void* tag) override {
write_ops_.set_output_tag(tag);
if (options.is_last_message()) {
options.set_buffer_hint();
}
EnsureInitialMetadataSent(&write_ops_);
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
call_.PerformOps(&write_ops_);
}
void WriteAndFinish(const W& msg, WriteOptions options, const Status& status,
void* tag) override {
write_ops_.set_output_tag(tag);
EnsureInitialMetadataSent(&write_ops_);
options.set_buffer_hint();
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
write_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
call_.PerformOps(&write_ops_);
}
void Finish(const Status& status, void* tag) override {
finish_ops_.set_output_tag(tag);
if (!ctx_->sent_initial_metadata_) {
finish_ops_.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
finish_ops_.set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
}
EnsureInitialMetadataSent(&finish_ops_);
finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
call_.PerformOps(&finish_ops_);
}
@ -524,11 +647,25 @@ class ServerAsyncReaderWriter final
void BindCall(Call* call) override { call_ = *call; }
template <class T>
void EnsureInitialMetadataSent(T* ops) {
if (!ctx_->sent_initial_metadata_) {
ops->SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
ops->set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
}
}
Call call_;
ServerContext* ctx_;
CallOpSet<CallOpSendInitialMetadata> meta_ops_;
CallOpSet<CallOpRecvMessage<R>> read_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> write_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpServerSendStatus>
write_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> finish_ops_;
};

@ -84,8 +84,9 @@ inline grpc_metadata* FillMetadataArray(
/// Per-message write options.
class WriteOptions {
public:
WriteOptions() : flags_(0) {}
WriteOptions(const WriteOptions& other) : flags_(other.flags_) {}
WriteOptions() : flags_(0), last_message_(false) {}
WriteOptions(const WriteOptions& other)
: flags_(other.flags_), last_message_(other.last_message_) {}
/// Clear all flags.
inline void Clear() { flags_ = 0; }
@ -141,6 +142,43 @@ class WriteOptions {
/// \sa GRPC_WRITE_BUFFER_HINT
inline bool get_buffer_hint() const { return GetBit(GRPC_WRITE_BUFFER_HINT); }
/// corked bit: aliases set_buffer_hint currently, with the intent that
/// set_buffer_hint will be removed in the future
inline WriteOptions& set_corked() {
SetBit(GRPC_WRITE_BUFFER_HINT);
return *this;
}
inline WriteOptions& clear_corked() {
ClearBit(GRPC_WRITE_BUFFER_HINT);
return *this;
}
inline bool is_corked() const { return GetBit(GRPC_WRITE_BUFFER_HINT); }
/// last-message bit: indicates this is the last message in a stream
/// client-side: makes Write the equivalent of performing Write, WritesDone
/// in a single step
/// server-side: hold the Write until the service handler returns (sync api)
/// or until Finish is called (async api)
inline WriteOptions& set_last_message() {
last_message_ = true;
return *this;
}
/// Clears flag indicating that this is the last message in a stream,
/// disabling coalescing.
inline WriteOptions& clear_last_messsage() {
last_message_ = false;
return *this;
}
/// Get value for the flag indicating that this is the last message, and
/// should be coalesced with trailing metadata.
///
/// \sa GRPC_WRITE_LAST_MESSAGE
bool is_last_message() const { return last_message_; }
WriteOptions& operator=(const WriteOptions& rhs) {
flags_ = rhs.flags_;
return *this;
@ -154,6 +192,7 @@ class WriteOptions {
bool GetBit(const uint32_t mask) const { return (flags_ & mask) != 0; }
uint32_t flags_;
bool last_message_;
};
/// Default argument for CallOpSet. I is unused by the class, but can be
@ -224,7 +263,7 @@ class CallOpSendMessage {
/// after use.
template <class M>
Status SendMessage(const M& message,
const WriteOptions& options) GRPC_MUST_USE_RESULT;
WriteOptions options) GRPC_MUST_USE_RESULT;
template <class M>
Status SendMessage(const M& message) GRPC_MUST_USE_RESULT;
@ -252,8 +291,7 @@ class CallOpSendMessage {
};
template <class M>
Status CallOpSendMessage::SendMessage(const M& message,
const WriteOptions& options) {
Status CallOpSendMessage::SendMessage(const M& message, WriteOptions options) {
write_options_ = options;
return SerializationTraits<M>::Serialize(message, &send_buf_, &own_buf_);
}

@ -281,6 +281,17 @@ class ClientContext {
/// \param algorithm The compression algorithm used for the client call.
void set_compression_algorithm(grpc_compression_algorithm algorithm);
/// Flag whether the initial metadata should be \a corked
///
/// If \a corked is true, then the initial metadata will be colasced with the
/// write of first message in the stream.
///
/// \param corked The flag indicating whether the initial metadata is to be
/// corked or not.
void set_initial_metadata_corked(bool corked) {
initial_metadata_corked_ = corked;
}
/// Return the peer uri in a string.
///
/// \warning This value is never authenticated or subject to any security
@ -357,7 +368,8 @@ class ClientContext {
(cacheable_ ? GRPC_INITIAL_METADATA_CACHEABLE_REQUEST : 0) |
(wait_for_ready_explicitly_set_
? GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET
: 0);
: 0) |
(initial_metadata_corked_ ? GRPC_INITIAL_METADATA_CORKED : 0);
}
grpc::string authority() { return authority_; }
@ -384,6 +396,7 @@ class ClientContext {
PropagationOptions propagation_options_;
grpc_compression_algorithm compression_algorithm_;
bool initial_metadata_corked_;
};
} // namespace grpc

@ -100,22 +100,40 @@ class WriterInterface {
public:
virtual ~WriterInterface() {}
/// Blocking write \a msg to the stream with options.
/// Blocking write \a msg to the stream with WriteOptions \a options.
/// This is thread-safe with respect to \a Read
///
/// \param msg The message to be written to the stream.
/// \param options Options affecting the write operation.
/// \param options The WriteOptions affecting the write operation.
///
/// \return \a true on success, \a false when the stream has been closed.
virtual bool Write(const W& msg, const WriteOptions& options) = 0;
virtual bool Write(const W& msg, WriteOptions options) = 0;
/// Blocking write \a msg to the stream with default options.
/// Blocking write \a msg to the stream with default write options.
/// This is thread-safe with respect to \a Read
///
/// \param msg The message to be written to the stream.
///
/// \return \a true on success, \a false when the stream has been closed.
inline bool Write(const W& msg) { return Write(msg, WriteOptions()); }
/// Write \a msg and coalesce it with the writing of trailing metadata, using
/// WriteOptions \a options.
///
/// For client, WriteLast is equivalent of performing Write and WritesDone in
/// a single step. \a msg and trailing metadata are coalesced and sent on wire
/// by calling this function.
/// For server, WriteLast buffers the \a msg. The writing of \a msg is held
/// until the service handler returns, where \a msg and trailing metadata are
/// coalesced and sent on wire. Note that WriteLast can only buffer \a msg up
/// to the flow control window size. If \a msg size is larger than the window
/// size, it will be sent on wire without buffering.
///
/// \param[in] msg The message to be written to the stream.
/// \param[in] options The WriteOptions to be used to write this message.
void WriteLast(const W& msg, WriteOptions options) {
Write(msg, options.set_last_message());
}
};
/// Client-side interface for streaming reads of message of type \a R.
@ -213,11 +231,13 @@ class ClientWriter : public ClientWriterInterface<W> {
finish_ops_.RecvMessage(response);
finish_ops_.AllowNoMessage();
CallOpSet<CallOpSendInitialMetadata> ops;
ops.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
call_.PerformOps(&ops);
cq_.Pluck(&ops);
if (!context_->initial_metadata_corked_) {
CallOpSet<CallOpSendInitialMetadata> ops;
ops.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
call_.PerformOps(&ops);
cq_.Pluck(&ops);
}
}
void WaitForInitialMetadata() {
@ -230,11 +250,24 @@ class ClientWriter : public ClientWriterInterface<W> {
}
using WriterInterface<W>::Write;
bool Write(const W& msg, const WriteOptions& options) override {
CallOpSet<CallOpSendMessage> ops;
bool Write(const W& msg, WriteOptions options) override {
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpClientSendClose>
ops;
if (options.is_last_message()) {
options.set_buffer_hint();
ops.ClientSendClose();
}
if (context_->initial_metadata_corked_) {
ops.SendInitialMetadata(context_->send_initial_metadata_,
context_->initial_metadata_flags());
context_->set_initial_metadata_corked(false);
}
if (!ops.SendMessage(msg, options).ok()) {
return false;
}
call_.PerformOps(&ops);
return cq_.Pluck(&ops);
}
@ -293,11 +326,13 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
ClientReaderWriter(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context)
: context_(context), call_(channel->CreateCall(method, context, &cq_)) {
CallOpSet<CallOpSendInitialMetadata> ops;
ops.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
call_.PerformOps(&ops);
cq_.Pluck(&ops);
if (!context_->initial_metadata_corked_) {
CallOpSet<CallOpSendInitialMetadata> ops;
ops.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
call_.PerformOps(&ops);
cq_.Pluck(&ops);
}
}
void WaitForInitialMetadata() override {
@ -325,9 +360,24 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
}
using WriterInterface<W>::Write;
bool Write(const W& msg, const WriteOptions& options) override {
CallOpSet<CallOpSendMessage> ops;
if (!ops.SendMessage(msg, options).ok()) return false;
bool Write(const W& msg, WriteOptions options) override {
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpClientSendClose>
ops;
if (options.is_last_message()) {
options.set_buffer_hint();
ops.ClientSendClose();
}
if (context_->initial_metadata_corked_) {
ops.SendInitialMetadata(context_->send_initial_metadata_,
context_->initial_metadata_flags());
context_->set_initial_metadata_corked(false);
}
if (!ops.SendMessage(msg, options).ok()) {
return false;
}
call_.PerformOps(&ops);
return cq_.Pluck(&ops);
}
@ -423,7 +473,10 @@ class ServerWriter final : public ServerWriterInterface<W> {
}
using WriterInterface<W>::Write;
bool Write(const W& msg, const WriteOptions& options) override {
bool Write(const W& msg, WriteOptions options) override {
if (options.is_last_message()) {
options.set_buffer_hint();
}
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> ops;
if (!ops.SendMessage(msg, options).ok()) {
return false;
@ -485,7 +538,10 @@ class ServerReaderWriterBody final {
return call_->cq()->Pluck(&ops) && ops.got_message;
}
bool Write(const W& msg, const WriteOptions& options) {
bool Write(const W& msg, WriteOptions options) {
if (options.is_last_message()) {
options.set_buffer_hint();
}
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> ops;
if (!ops.SendMessage(msg, options).ok()) {
return false;
@ -523,7 +579,7 @@ class ServerReaderWriter final : public ServerReaderWriterInterface<W, R> {
bool Read(R* msg) override { return body_.Read(msg); }
using WriterInterface<W>::Write;
bool Write(const W& msg, const WriteOptions& options) override {
bool Write(const W& msg, WriteOptions options) override {
return body_.Write(msg, options);
}
@ -562,8 +618,7 @@ class ServerUnaryStreamer final
}
using WriterInterface<ResponseType>::Write;
bool Write(const ResponseType& response,
const WriteOptions& options) override {
bool Write(const ResponseType& response, WriteOptions options) override {
if (write_done_ || !read_done_) {
return false;
}
@ -604,8 +659,7 @@ class ServerSplitStreamer final
}
using WriterInterface<ResponseType>::Write;
bool Write(const ResponseType& response,
const WriteOptions& options) override {
bool Write(const ResponseType& response, WriteOptions options) override {
return read_done_ && body_.Write(response, options);
}

@ -198,14 +198,14 @@ typedef struct {
#define GRPC_ARG_HTTP2_WRITE_BUFFER_SIZE "grpc.http2.write_buffer_size"
/** After a duration of this time the client pings the server to see if the
transport is still alive. Int valued, seconds. */
#define GRPC_ARG_HTTP2_KEEPALIVE_TIME "grpc.http2.keepalive_time"
#define GRPC_ARG_CLIENT_KEEPALIVE_TIME_S "grpc.client_keepalive_time"
/** After waiting for a duration of this time, if the client does not receive
the ping ack, it will close the transport. Int valued, seconds. */
#define GRPC_ARG_HTTP2_KEEPALIVE_TIMEOUT "grpc.http2.keepalive_timeout"
#define GRPC_ARG_CLIENT_KEEPALIVE_TIMEOUT_S "grpc.client_keepalive_timeout"
/** Is it permissible to send keepalive pings without any outstanding streams.
Int valued, 0(false)/1(true). */
#define GRPC_ARG_HTTP2_KEEPALIVE_PERMIT_WITHOUT_CALLS \
"grpc.http2.keepalive_permit_without_calls"
#define GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS \
"grpc.keepalive_permit_without_calls"
/** Default authority to pass if none specified on call construction. A string.
* */
#define GRPC_ARG_DEFAULT_AUTHORITY "grpc.default_authority"
@ -316,13 +316,16 @@ typedef enum grpc_call_error {
/** Signal that GRPC_INITIAL_METADATA_WAIT_FOR_READY was explicitly set
by the calling application. */
#define GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET (0x00000080u)
/** Signal that the initial metadata should be corked */
#define GRPC_INITIAL_METADATA_CORKED (0x00000100u)
/** Mask of all valid flags */
#define GRPC_INITIAL_METADATA_USED_MASK \
(GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST | \
GRPC_INITIAL_METADATA_WAIT_FOR_READY | \
GRPC_INITIAL_METADATA_CACHEABLE_REQUEST | \
GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET)
#define GRPC_INITIAL_METADATA_USED_MASK \
(GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST | \
GRPC_INITIAL_METADATA_WAIT_FOR_READY | \
GRPC_INITIAL_METADATA_CACHEABLE_REQUEST | \
GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET | \
GRPC_INITIAL_METADATA_CORKED)
/** A single metadata element */
typedef struct grpc_metadata {

@ -1,4 +1,4 @@
#Overview
# Overview
This directory contains source code for gRPC protocol buffer compiler (*protoc*) plugins. Along with `protoc`,
these plugins are used to generate gRPC client and server stubs from `.proto` files.

@ -1,4 +1,4 @@
#Overview
# Overview
This directory contains source code for C library (a.k.a the *gRPC C core*) that provides all gRPC's core functionality through a low level API. Libraries in other languages in this repository (C++, Ruby,
Python, PHP, NodeJS, Objective-C) are layered on top of this library.

@ -1,6 +1,6 @@
Files generated for use by Census stats and trace recording subsystem.
#Files
# Files
* census.pb.{h,c} - Generated from src/core/ext/census/census.proto, using the
script `tools/codegen/core/gen_nano_proto.sh src/proto/census/census.proto
$PWD/src/core/ext/census/gen src/core/ext/census/gen`

@ -69,10 +69,16 @@
#define MAX_WRITE_BUFFER_SIZE (64 * 1024 * 1024)
#define DEFAULT_MAX_HEADER_LIST_SIZE (16 * 1024)
#define DEFAULT_KEEPALIVE_TIME_SECOND INT_MAX
#define DEFAULT_KEEPALIVE_TIMEOUT_SECOND 20
#define DEFAULT_CLIENT_KEEPALIVE_TIME_S INT_MAX
#define DEFAULT_CLIENT_KEEPALIVE_TIMEOUT_S 20
#define DEFAULT_KEEPALIVE_PERMIT_WITHOUT_CALLS false
static int g_default_client_keepalive_time_s = DEFAULT_CLIENT_KEEPALIVE_TIME_S;
static int g_default_client_keepalive_timeout_s =
DEFAULT_CLIENT_KEEPALIVE_TIMEOUT_S;
static bool g_default_keepalive_permit_without_calls =
DEFAULT_KEEPALIVE_PERMIT_WITHOUT_CALLS;
#define MAX_CLIENT_STREAM_ID 0x7fffffffu
int grpc_http_trace = 0;
int grpc_flowctl_trace = 0;
@ -142,6 +148,8 @@ static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
grpc_chttp2_ping_type ping_type,
grpc_closure *on_initiate,
grpc_closure *on_complete);
static void retry_initiate_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
grpc_error *error);
#define DEFAULT_MIN_TIME_BETWEEN_PINGS_MS 0
#define DEFAULT_MAX_PINGS_BETWEEN_DATA 3
@ -267,6 +275,8 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
grpc_closure_init(&t->destructive_reclaimer_locked,
destructive_reclaimer_locked, t,
grpc_combiner_scheduler(t->combiner, false));
grpc_closure_init(&t->retry_initiate_ping_locked, retry_initiate_ping_locked,
t, grpc_combiner_scheduler(t->combiner, false));
grpc_closure_init(&t->start_bdp_ping_locked, start_bdp_ping_locked, t,
grpc_combiner_scheduler(t->combiner, false));
grpc_closure_init(&t->finish_bdp_ping_locked, finish_bdp_ping_locked, t,
@ -345,15 +355,16 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
/* client-side keepalive setting */
t->keepalive_time =
DEFAULT_KEEPALIVE_TIME_SECOND == INT_MAX
g_default_client_keepalive_time_s == INT_MAX
? gpr_inf_future(GPR_TIMESPAN)
: gpr_time_from_seconds(DEFAULT_KEEPALIVE_TIME_SECOND, GPR_TIMESPAN);
: gpr_time_from_seconds(g_default_client_keepalive_time_s,
GPR_TIMESPAN);
t->keepalive_timeout =
DEFAULT_KEEPALIVE_TIMEOUT_SECOND == INT_MAX
g_default_client_keepalive_timeout_s == INT_MAX
? gpr_inf_future(GPR_TIMESPAN)
: gpr_time_from_seconds(DEFAULT_KEEPALIVE_TIMEOUT_SECOND,
: gpr_time_from_seconds(g_default_client_keepalive_timeout_s,
GPR_TIMESPAN);
t->keepalive_permit_without_calls = DEFAULT_KEEPALIVE_PERMIT_WITHOUT_CALLS;
t->keepalive_permit_without_calls = g_default_keepalive_permit_without_calls;
if (channel_args) {
for (i = 0; i < channel_args->num_args; i++) {
@ -403,24 +414,25 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
t->enable_bdp_probe = grpc_channel_arg_get_integer(
&channel_args->args[i], (grpc_integer_options){1, 0, 1});
} else if (0 == strcmp(channel_args->args[i].key,
GRPC_ARG_HTTP2_KEEPALIVE_TIME)) {
GRPC_ARG_CLIENT_KEEPALIVE_TIME_S)) {
const int value = grpc_channel_arg_get_integer(
&channel_args->args[i],
(grpc_integer_options){DEFAULT_KEEPALIVE_TIME_SECOND, 1, INT_MAX});
(grpc_integer_options){g_default_client_keepalive_time_s, 1,
INT_MAX});
t->keepalive_time = value == INT_MAX
? gpr_inf_future(GPR_TIMESPAN)
: gpr_time_from_seconds(value, GPR_TIMESPAN);
} else if (0 == strcmp(channel_args->args[i].key,
GRPC_ARG_HTTP2_KEEPALIVE_TIMEOUT)) {
GRPC_ARG_CLIENT_KEEPALIVE_TIMEOUT_S)) {
const int value = grpc_channel_arg_get_integer(
&channel_args->args[i],
(grpc_integer_options){DEFAULT_KEEPALIVE_TIMEOUT_SECOND, 0,
(grpc_integer_options){g_default_client_keepalive_timeout_s, 0,
INT_MAX});
t->keepalive_timeout = value == INT_MAX
? gpr_inf_future(GPR_TIMESPAN)
: gpr_time_from_seconds(value, GPR_TIMESPAN);
} else if (0 == strcmp(channel_args->args[i].key,
GRPC_ARG_HTTP2_KEEPALIVE_PERMIT_WITHOUT_CALLS)) {
GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS)) {
t->keepalive_permit_without_calls =
(uint32_t)grpc_channel_arg_get_integer(
&channel_args->args[i], (grpc_integer_options){0, 0, 1});
@ -474,6 +486,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
t->ping_state.pings_before_data_required =
t->ping_policy.max_pings_without_data;
t->ping_state.is_delayed_ping_timer_set = false;
/** Start client-side keepalive pings */
if (t->is_client) {
@ -1210,8 +1223,13 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
}
} else {
GPR_ASSERT(s->id != 0);
grpc_chttp2_become_writable(exec_ctx, t, s,
GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED,
grpc_chttp2_stream_write_type write_type =
GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED;
if (op->send_message != NULL &&
(op->send_message->flags & GRPC_WRITE_BUFFER_HINT)) {
write_type = GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK;
}
grpc_chttp2_become_writable(exec_ctx, t, s, write_type,
"op.send_initial_metadata");
}
} else {
@ -1394,6 +1412,13 @@ static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
}
}
static void retry_initiate_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
grpc_error *error) {
grpc_chttp2_transport *t = tp;
t->ping_state.is_delayed_ping_timer_set = false;
grpc_chttp2_initiate_write(exec_ctx, t, false, "retry_send_ping");
}
void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
uint64_t id) {
grpc_chttp2_ping_queue *pq =
@ -2103,6 +2128,32 @@ static void finish_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "bdp_ping");
}
void grpc_chttp2_config_default_keepalive_args(grpc_channel_args *args) {
size_t i;
if (args) {
for (i = 0; i < args->num_args; i++) {
if (0 == strcmp(args->args[i].key, GRPC_ARG_CLIENT_KEEPALIVE_TIME_S)) {
g_default_client_keepalive_time_s = grpc_channel_arg_get_integer(
&args->args[i], (grpc_integer_options){
g_default_client_keepalive_time_s, 1, INT_MAX});
} else if (0 == strcmp(args->args[i].key,
GRPC_ARG_CLIENT_KEEPALIVE_TIMEOUT_S)) {
g_default_client_keepalive_timeout_s = grpc_channel_arg_get_integer(
&args->args[i],
(grpc_integer_options){g_default_client_keepalive_timeout_s, 0,
INT_MAX});
} else if (0 == strcmp(args->args[i].key,
GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS)) {
g_default_keepalive_permit_without_calls =
(uint32_t)grpc_channel_arg_get_integer(
&args->args[i],
(grpc_integer_options){g_default_keepalive_permit_without_calls,
0, 1});
}
}
}
}
static void init_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error) {
grpc_chttp2_transport *t = arg;
@ -2146,9 +2197,7 @@ static void finish_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg,
grpc_timer_init(
exec_ctx, &t->keepalive_ping_timer,
gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), t->keepalive_time),
grpc_closure_create(init_keepalive_ping_locked, t,
grpc_combiner_scheduler(t->combiner, false)),
gpr_now(GPR_CLOCK_MONOTONIC));
&t->init_keepalive_ping_locked, gpr_now(GPR_CLOCK_MONOTONIC));
}
}
GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "keepalive ping end");

@ -40,6 +40,8 @@
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
static bool g_disable_ping_ack = false;
grpc_slice grpc_chttp2_ping_create(uint8_t ack, uint64_t opaque_8bytes) {
grpc_slice slice = grpc_slice_malloc(9 + 8);
uint8_t *p = GRPC_SLICE_START_PTR(slice);
@ -101,15 +103,21 @@ grpc_error *grpc_chttp2_ping_parser_parse(grpc_exec_ctx *exec_ctx, void *parser,
if (p->is_ack) {
grpc_chttp2_ack_ping(exec_ctx, t, p->opaque_8bytes);
} else {
if (t->ping_ack_count == t->ping_ack_capacity) {
t->ping_ack_capacity = GPR_MAX(t->ping_ack_capacity * 3 / 2, 3);
t->ping_acks = gpr_realloc(
t->ping_acks, t->ping_ack_capacity * sizeof(*t->ping_acks));
if (!g_disable_ping_ack) {
if (t->ping_ack_count == t->ping_ack_capacity) {
t->ping_ack_capacity = GPR_MAX(t->ping_ack_capacity * 3 / 2, 3);
t->ping_acks = gpr_realloc(
t->ping_acks, t->ping_ack_capacity * sizeof(*t->ping_acks));
}
t->ping_acks[t->ping_ack_count++] = p->opaque_8bytes;
grpc_chttp2_initiate_write(exec_ctx, t, false, "ping response");
}
t->ping_acks[t->ping_ack_count++] = p->opaque_8bytes;
grpc_chttp2_initiate_write(exec_ctx, t, false, "ping response");
}
}
return GRPC_ERROR_NONE;
}
void grpc_set_disable_ping_ack(bool disable_ping_ack) {
g_disable_ping_ack = disable_ping_ack;
}

@ -53,4 +53,7 @@ grpc_error *grpc_chttp2_ping_parser_parse(grpc_exec_ctx *exec_ctx, void *parser,
grpc_chttp2_stream *s,
grpc_slice slice, int is_last);
/* Test-only function for disabling ping ack */
void grpc_set_disable_ping_ack(bool disable_ping_ack);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FRAME_PING_H */

@ -102,6 +102,8 @@ typedef struct {
typedef struct {
gpr_timespec last_ping_sent_time;
int pings_before_data_required;
grpc_timer delayed_ping_timer;
bool is_delayed_ping_timer_set;
} grpc_chttp2_repeated_ping_state;
/* deframer state for the overall http2 stream of bytes */
@ -308,6 +310,7 @@ struct grpc_chttp2_transport {
grpc_chttp2_repeated_ping_policy ping_policy;
grpc_chttp2_repeated_ping_state ping_state;
uint64_t ping_ctr; /* unique id for pings */
grpc_closure retry_initiate_ping_locked;
/** ping acks */
size_t ping_ack_count;
@ -827,4 +830,8 @@ void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx,
uint32_t grpc_chttp2_target_incoming_window(grpc_chttp2_transport *t);
/** Set the default keepalive configurations, must only be called at
initialization */
void grpc_chttp2_config_default_keepalive_args(grpc_channel_args *args);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */

@ -101,6 +101,14 @@ static void maybe_initiate_ping(grpc_exec_ctx *exec_ctx,
"Ping delayed [%p]: not enough time elapsed since last ping",
t->peer_string);
}
if (!t->ping_state.is_delayed_ping_timer_set) {
t->ping_state.is_delayed_ping_timer_set = true;
grpc_timer_init(exec_ctx, &t->ping_state.delayed_ping_timer,
gpr_time_add(t->ping_state.last_ping_sent_time,
t->ping_policy.min_time_between_pings),
&t->retry_initiate_ping_locked,
gpr_now(GPR_CLOCK_MONOTONIC));
}
return;
}
/* coalesce equivalent pings into this one */

@ -279,11 +279,17 @@ static void rq_step_sched(grpc_exec_ctx *exec_ctx,
/* update the atomically available resource estimate - use no barriers since
timeliness of delivery really doesn't matter much */
static void rq_update_estimate(grpc_resource_quota *resource_quota) {
gpr_atm memory_usage_estimation = MEMORY_USAGE_ESTIMATION_MAX;
if (resource_quota->size != 0) {
memory_usage_estimation =
GPR_CLAMP((gpr_atm)((1.0 -
((double)resource_quota->free_pool) /
((double)resource_quota->size)) *
MEMORY_USAGE_ESTIMATION_MAX),
0, MEMORY_USAGE_ESTIMATION_MAX);
}
gpr_atm_no_barrier_store(&resource_quota->memory_usage_estimation,
(gpr_atm)((1.0 -
((double)resource_quota->free_pool) /
((double)resource_quota->size)) *
MEMORY_USAGE_ESTIMATION_MAX));
memory_usage_estimation);
}
/* returns true if all allocations are completed */

@ -625,7 +625,7 @@ static bool get_final_status_from(
void (*set_value)(grpc_status_code code, void *user_data),
void *set_value_user_data, grpc_slice *details) {
grpc_status_code code;
grpc_slice slice;
grpc_slice slice = grpc_empty_slice();
grpc_error_get_status(error, call->send_deadline, &code, &slice, NULL);
if (code == GRPC_STATUS_OK && !allow_ok_status) {
return false;

@ -1,17 +1,17 @@
#Overview
# Overview
This directory contains source code for C++ implementation of gRPC.
#Pre-requisites
# Pre-requisites
##Linux
## Linux
```sh
$ [sudo] apt-get install build-essential autoconf libtool
```
##Mac OSX
## Mac OSX
For a Mac system, git is not available by default. You will first need to
install Xcode from the Mac AppStore and then run the following command from a
@ -21,7 +21,7 @@ terminal:
$ [sudo] xcode-select --install
```
##Protoc
## Protoc
By default gRPC uses [protocol buffers](https://github.com/google/protobuf),
you will need the `protoc` compiler to generate stub server and client code.
@ -39,12 +39,12 @@ $ sudo make install # 'make' should have been run by core grpc
Alternatively, you can download `protoc` binaries from
[the protocol buffers Github repository](https://github.com/google/protobuf/releases).
#Installation
# Installation
Currently to install gRPC for C++, you need to build from source as described
below.
#Build from Source
# Build from Source
```sh
$ git clone -b $(curl -L http://grpc.io/release) https://github.com/grpc/grpc
@ -54,7 +54,7 @@ below.
$ [sudo] make install
```
#Documentation
# Documentation
You can find out how to build and run our simplest gRPC C++ example in our
[C++ quick start](../../examples/cpp).

@ -67,7 +67,8 @@ ClientContext::ClientContext()
call_canceled_(false),
deadline_(gpr_inf_future(GPR_CLOCK_REALTIME)),
census_context_(nullptr),
propagate_from_call_(nullptr) {
propagate_from_call_(nullptr),
initial_metadata_corked_(false) {
g_client_callbacks->DefaultConstructor(this);
}

@ -36,6 +36,7 @@
#include <grpc/grpc.h>
#include <grpc/support/time.h>
#import <RxLibrary/GRXConcurrentWriteable.h>
#import <RxLibrary/GRXImmediateSingleWriter.h>
#import "private/GRPCConnectivityMonitor.h"
#import "private/GRPCHost.h"
@ -45,6 +46,11 @@
#import "private/NSDictionary+GRPC.h"
#import "private/NSError+GRPC.h"
// At most 6 ops can be in an op batch for a client: SEND_INITIAL_METADATA,
// SEND_MESSAGE, SEND_CLOSE_FROM_CLIENT, RECV_INITIAL_METADATA, RECV_MESSAGE,
// and RECV_STATUS_ON_CLIENT.
NSInteger kMaxClientBatch = 6;
NSString * const kGRPCHeadersKey = @"io.grpc.HeadersKey";
NSString * const kGRPCTrailersKey = @"io.grpc.TrailersKey";
static NSMutableDictionary *callFlags;
@ -100,6 +106,13 @@ static NSMutableDictionary *callFlags;
GRPCCall *_retainSelf;
GRPCRequestHeaders *_requestHeaders;
// In the case that the call is a unary call (i.e. the writer to GRPCCall is of type
// GRXImmediateSingleWriter), GRPCCall will delay sending ops (not send them to C core
// immediately) and buffer them into a batch _unaryOpBatch. The batch is sent to C core when
// the SendClose op is added.
BOOL _unaryCall;
NSMutableArray *_unaryOpBatch;
}
@synthesize state = _state;
@ -157,6 +170,11 @@ static NSMutableDictionary *callFlags;
_requestWriter = requestWriter;
_requestHeaders = [[GRPCRequestHeaders alloc] initWithCall:self];
if ([requestWriter isKindOfClass:[GRXImmediateSingleWriter class]]) {
_unaryCall = YES;
_unaryOpBatch = [NSMutableArray arrayWithCapacity:kMaxClientBatch];
}
}
return self;
}
@ -165,6 +183,9 @@ static NSMutableDictionary *callFlags;
- (void)finishWithError:(NSError *)errorOrNil {
@synchronized(self) {
if (_state == GRXWriterStateFinished) {
return;
}
_state = GRXWriterStateFinished;
}
@ -254,15 +275,22 @@ static NSMutableDictionary *callFlags;
- (void)sendHeaders:(NSDictionary *)headers {
// TODO(jcanizales): Add error handlers for async failures
[_wrappedCall startBatchWithOperations:@[[[GRPCOpSendMetadata alloc] initWithMetadata:headers
flags:[GRPCCall callFlagsForHost:_host path:_path]
handler:nil]]];
GRPCOpSendMetadata *op = [[GRPCOpSendMetadata alloc] initWithMetadata:headers
flags:[GRPCCall callFlagsForHost:_host path:_path]
handler:nil]; // No clean-up needed after SEND_INITIAL_METADATA
if (!_unaryCall) {
[_wrappedCall startBatchWithOperations:@[op]];
} else {
[_unaryOpBatch addObject:op];
}
}
#pragma mark GRXWriteable implementation
// Only called from the call queue. The error handler will be called from the
// network queue if the write didn't succeed.
// If the call is a unary call, parameter \a errorHandler will be ignored and
// the error handler of GRPCOpSendClose will be executed in case of error.
- (void)writeMessage:(NSData *)message withErrorHandler:(void (^)())errorHandler {
__weak GRPCCall *weakSelf = self;
@ -275,9 +303,17 @@ static NSMutableDictionary *callFlags;
}
}
};
[_wrappedCall startBatchWithOperations:@[[[GRPCOpSendMessage alloc] initWithMessage:message
handler:resumingHandler]]
errorHandler:errorHandler];
GRPCOpSendMessage *op = [[GRPCOpSendMessage alloc] initWithMessage:message
handler:resumingHandler];
if (!_unaryCall) {
[_wrappedCall startBatchWithOperations:@[op]
errorHandler:errorHandler];
} else {
// Ignored errorHandler since it is the same as the one for GRPCOpSendClose.
// TODO (mxyan): unify the error handlers of all Ops into a single closure.
[_unaryOpBatch addObject:op];
}
}
- (void)writeValue:(id)value {
@ -302,8 +338,14 @@ static NSMutableDictionary *callFlags;
// Only called from the call queue. The error handler will be called from the
// network queue if the requests stream couldn't be closed successfully.
- (void)finishRequestWithErrorHandler:(void (^)())errorHandler {
[_wrappedCall startBatchWithOperations:@[[[GRPCOpSendClose alloc] init]]
errorHandler:errorHandler];
if (!_unaryCall) {
[_wrappedCall startBatchWithOperations:@[[[GRPCOpSendClose alloc] init]]
errorHandler:errorHandler];
} else {
[_unaryOpBatch addObject:[[GRPCOpSendClose alloc] init]];
[_wrappedCall startBatchWithOperations:_unaryOpBatch
errorHandler:errorHandler];
}
}
- (void)writesFinishedWithError:(NSError *)errorOrNil {

@ -0,0 +1,61 @@
/*
*
* Copyright 2017, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#ifdef GRPC_TEST_OBJC
#import "../GRPCCall.h"
/**
* Methods used for gRPC internal tests. DO NOT USE.
*/
@interface GRPCCall (InternalTests)
/**
* Enables logging of op batches. Memory consumption increases as more ops are logged.
*
* This function is for internal testing of gRPC only. It is not part of gRPC's public interface.
* Do not use in production. To enable, set the preprocessor flag GRPC_TEST_OBJC.
*/
+ (void)enableOpBatchLog:(BOOL)enabled;
/**
* Obtain the logged op batches. Invoking this method will clean the log.
*
* This function is for internal testing of gRPC only. It is not part of gRPC's public interface.
* Do not use in production. To enable, set the preprocessor flag GRPC_TEST_OBJC.
*/
+ (NSArray *)obtainAndCleanOpBatchLog;
@end
#endif

@ -0,0 +1,52 @@
/*
*
* Copyright 2017, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#ifdef GRPC_TEST_OBJC
#import "GRPCCall+InternalTests.h"
#import "../private/GRPCOpBatchLog.h"
@implementation GRPCCall (InternalTests)
+ (void)enableOpBatchLog:(BOOL)enabled {
[GRPCOpBatchLog enableOpBatchLog:enabled];
}
+ (NSArray *)obtainAndCleanOpBatchLog {
return [GRPCOpBatchLog obtainAndCleanOpBatchLog];
}
@end
#endif

@ -0,0 +1,59 @@
/*
*
* Copyright 2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#ifdef GRPC_TEST_OBJC
/**
* Logs the op batches of a client. Used for testing.
*/
@interface GRPCOpBatchLog : NSObject
/**
* Enables logging of op batches. Memory consumption increases as more ops are logged.
*/
+ (void)enableOpBatchLog:(BOOL)enabled;
/**
* Add an op batch to log.
*/
+ (void)addOpBatchToLog:(NSArray *)batch;
/**
* Obtain the logged op batches. Invoking this method will clean the log.
*/
+ (NSArray *)obtainAndCleanOpBatchLog;
@end
#endif

@ -0,0 +1,72 @@
/*
*
* Copyright 2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#ifdef GRPC_TEST_OBJC
#import "GRPCOpBatchLog.h"
static NSMutableArray *opBatchLog = nil;
@implementation GRPCOpBatchLog
+ (void)enableOpBatchLog:(BOOL)enabled {
@synchronized (opBatchLog) {
if (enabled) {
if (!opBatchLog) {
opBatchLog = [NSMutableArray array];
}
} else {
if (opBatchLog) {
opBatchLog = nil;
}
}
}
}
+ (void)addOpBatchToLog:(NSArray *)batch {
@synchronized (opBatchLog) {
[opBatchLog addObject:batch];
}
}
+ (NSArray *)obtainAndCleanOpBatchLog {
@synchronized (opBatchLog) {
NSArray *out = opBatchLog;
opBatchLog = [NSMutableArray array];
return out;
}
}
@end
#endif

@ -44,6 +44,8 @@
#import "NSData+GRPC.h"
#import "NSError+GRPC.h"
#import "GRPCOpBatchLog.h"
@implementation GRPCOperation {
@protected
// Most operation subclasses don't set any flags in the grpc_op, and rely on the flag member being
@ -274,6 +276,12 @@
}
- (void)startBatchWithOperations:(NSArray *)operations errorHandler:(void (^)())errorHandler {
// Keep logs of op batches when we are running tests. Disabled when in production for improved
// performance.
#ifdef GRPC_TEST_OBJC
[GRPCOpBatchLog addOpBatchToLog:operations];
#endif
size_t nops = operations.count;
grpc_op *ops_array = gpr_malloc(nops * sizeof(grpc_op));
size_t i = 0;

@ -0,0 +1,50 @@
/*
*
* Copyright 2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#import <Foundation/Foundation.h>
#import "GRXImmediateWriter.h"
/**
* Utility to construct GRXWriter instances from values that are immediately available when
* required.
*/
@interface GRXImmediateSingleWriter : GRXImmediateWriter
/**
* Returns a writer that sends the passed value to its writeable and then finishes (releasing the
* value).
*/
+ (GRXWriter *)writerWithValue:(id)value;
@end

@ -0,0 +1,89 @@
/*
*
* Copyright 2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#import "GRXImmediateSingleWriter.h"
@implementation GRXImmediateSingleWriter {
id _value;
id<GRXWriteable> _writeable;
}
@synthesize state = _state;
- (instancetype)initWithValue:(id)value {
if (self = [super init]) {
_value = value;
_state = GRXWriterStateNotStarted;
}
return self;
}
+ (GRXWriter *)writerWithValue:(id)value {
return [[self alloc] initWithValue:value];
}
- (void)startWithWriteable:(id<GRXWriteable>)writeable {
_state = GRXWriterStateStarted;
_writeable = writeable;
[writeable writeValue:_value];
[self finish];
}
- (void)finish {
_state = GRXWriterStateFinished;
_value = nil;
id<GRXWriteable> writeable = _writeable;
_writeable = nil;
[writeable writesFinishedWithError:nil];
}
// Overwrite the setter to disallow manual state transition. The getter
// of _state is synthesized.
- (void)setState:(GRXWriterState)newState {
// Manual state transition is not allowed
return;
}
// Overrides [requestWriter(Transformations):map:] for Protocol Buffers
// encoding.
// We need the return value of this map to be a GRXImmediateSingleWriter but
// the original \a map function returns a new Writer of another type. So we
// need to override this function here.
- (GRXWriter *)map:(id (^)(id))map {
// Since _value is available when creating the object, we can simply
// apply the map and store the output.
_value = map(_value);
return self;
}
@end

@ -34,6 +34,7 @@
#import "GRXWriter+Immediate.h"
#import "GRXImmediateWriter.h"
#import "GRXImmediateSingleWriter.h"
@implementation GRXWriter (Immediate)
@ -50,7 +51,7 @@
}
+ (instancetype)writerWithValue:(id)value {
return [GRXImmediateWriter writerWithValue:value];
return [GRXImmediateSingleWriter writerWithValue:value];
}
+ (instancetype)writerWithError:(NSError *)error {

@ -38,6 +38,7 @@
#import <GRPCClient/GRPCCall+ChannelArg.h>
#import <GRPCClient/GRPCCall+OAuth2.h>
#import <GRPCClient/GRPCCall+Tests.h>
#import <GRPCClient/internal_testing/GRPCCall+InternalTests.h>
#import <ProtoRPC/ProtoMethod.h>
#import <RemoteTest/Messages.pbobjc.h>
#import <RxLibrary/GRXWriteable.h>

@ -38,6 +38,7 @@
#import <Cronet/Cronet.h>
#import <GRPCClient/GRPCCall+ChannelArg.h>
#import <GRPCClient/GRPCCall+Tests.h>
#import <GRPCClient/internal_testing/GRPCCall+InternalTests.h>
#import <GRPCClient/GRPCCall+Cronet.h>
#import <ProtoRPC/ProtoRPC.h>
#import <RemoteTest/Messages.pbobjc.h>
@ -45,6 +46,8 @@
#import <RemoteTest/Test.pbrpc.h>
#import <RxLibrary/GRXBufferedPipe.h>
#import <RxLibrary/GRXWriter+Immediate.h>
#import <grpc/support/log.h>
#import <grpc/grpc.h>
#define TEST_TIMEOUT 32
@ -97,15 +100,6 @@
return 0;
}
+ (void)setUp {
#ifdef GRPC_COMPILE_WITH_CRONET
// Cronet setup
[Cronet setHttp2Enabled:YES];
[Cronet start];
[GRPCCall useCronetWithEngine:[Cronet getGlobalEngine]];
#endif
}
- (void)setUp {
self.continueAfterFailure = NO;
@ -155,6 +149,44 @@
[self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
}
- (void)testPacketCoalescing {
XCTAssertNotNil(self.class.host);
__weak XCTestExpectation *expectation = [self expectationWithDescription:@"LargeUnary"];
RMTSimpleRequest *request = [RMTSimpleRequest message];
request.responseType = RMTPayloadType_Compressable;
request.responseSize = 10;
request.payload.body = [NSMutableData dataWithLength:10];
[GRPCCall enableOpBatchLog:YES];
[_service unaryCallWithRequest:request handler:^(RMTSimpleResponse *response, NSError *error) {
XCTAssertNil(error, @"Finished with unexpected error: %@", error);
RMTSimpleResponse *expectedResponse = [RMTSimpleResponse message];
expectedResponse.payload.type = RMTPayloadType_Compressable;
expectedResponse.payload.body = [NSMutableData dataWithLength:10];
XCTAssertEqualObjects(response, expectedResponse);
// The test is a success if there is a batch of exactly 3 ops (SEND_INITIAL_METADATA,
// SEND_MESSAGE, SEND_CLOSE_FROM_CLIENT). Without packet coalescing each batch of ops contains
// only one op.
NSArray *opBatches = [GRPCCall obtainAndCleanOpBatchLog];
const NSInteger kExpectedOpBatchSize = 3;
for (NSObject *o in opBatches) {
if ([o isKindOfClass:[NSArray class]]) {
NSArray *batch = (NSArray *)o;
if ([batch count] == kExpectedOpBatchSize) {
[expectation fulfill];
break;
}
}
}
}];
[self waitForExpectationsWithTimeout:16 handler:nil];
[GRPCCall enableOpBatchLog:NO];
}
- (void)test4MBResponsesAreAccepted {
XCTAssertNotNil(self.class.host);
__weak XCTestExpectation *expectation = [self expectationWithDescription:@"MaxResponseSize"];

@ -32,6 +32,7 @@
*/
#import <GRPCClient/GRPCCall+Tests.h>
#import <GRPCClient/internal_testing/GRPCCall+InternalTests.h>
#import "InteropTests.h"

@ -32,6 +32,7 @@
*/
#import <GRPCClient/GRPCCall+Tests.h>
#import <GRPCClient/internal_testing/GRPCCall+InternalTests.h>
#import "InteropTests.h"

@ -32,6 +32,7 @@
*/
#import <GRPCClient/GRPCCall+Tests.h>
#import <GRPCClient/internal_testing/GRPCCall+InternalTests.h>
#import "InteropTests.h"

@ -32,6 +32,10 @@
*/
#import <GRPCClient/GRPCCall+Tests.h>
#import <GRPCClient/internal_testing/GRPCCall+InternalTests.h>
#import <Cronet/Cronet.h>
#import <GRPCClient/GRPCCall+Cronet.h>
#import "InteropTests.h"

@ -97,15 +97,20 @@ post_install do |installer|
# GPR_UNREACHABLE_CODE causes "Control may reach end of non-void
# function" warning
config.build_settings['GCC_WARN_ABOUT_RETURN_TYPE'] = 'NO'
config.build_settings['GCC_PREPROCESSOR_DEFINITIONS'] = '$(inherited) COCOAPODS=1 GRPC_CRONET_WITH_PACKET_COALESCING=1'
end
end
# Activate Cronet for the dedicated build configuration 'Cronet', which will be used solely by
# the test target 'InteropTestsRemoteWithCronet'
# Activate GRPCCall+InternalTests functions for the dedicated build configuration 'Test', which will
# be used by all test targets using it.
if target.name == 'gRPC'
target.build_configurations.each do |config|
if config.name == 'Cronet'
config.build_settings['GCC_PREPROCESSOR_DEFINITIONS'] = '$(inherited) COCOAPODS=1 GRPC_COMPILE_WITH_CRONET=1'
config.build_settings['GCC_PREPROCESSOR_DEFINITIONS'] = '$(inherited) COCOAPODS=1 GRPC_COMPILE_WITH_CRONET=1 GRPC_TEST_OBJC=1'
elsif config.name == 'Test'
config.build_settings['GCC_PREPROCESSOR_DEFINITIONS'] = '$(inherited) COCOAPODS=1 GRPC_TEST_OBJC=1'
end
end
end

@ -125,8 +125,10 @@
0A4F89D9C90E9C30990218F0 /* Pods.release.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = Pods.release.xcconfig; path = "Pods/Target Support Files/Pods/Pods.release.xcconfig"; sourceTree = "<group>"; };
0D2284C3DF7E57F0ED504E39 /* Pods-CoreCronetEnd2EndTests.debug.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-CoreCronetEnd2EndTests.debug.xcconfig"; path = "Pods/Target Support Files/Pods-CoreCronetEnd2EndTests/Pods-CoreCronetEnd2EndTests.debug.xcconfig"; sourceTree = "<group>"; };
14B09A58FEE53A7A6B838920 /* Pods-InteropTestsLocalSSL.cronet.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-InteropTestsLocalSSL.cronet.xcconfig"; path = "Pods/Target Support Files/Pods-InteropTestsLocalSSL/Pods-InteropTestsLocalSSL.cronet.xcconfig"; sourceTree = "<group>"; };
1588C85DEAF7FC0ACDEA4C02 /* Pods-InteropTestsLocalCleartext.test.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-InteropTestsLocalCleartext.test.xcconfig"; path = "Pods/Target Support Files/Pods-InteropTestsLocalCleartext/Pods-InteropTestsLocalCleartext.test.xcconfig"; sourceTree = "<group>"; };
17F60BF2871F6AF85FB3FA12 /* Pods-InteropTestsRemoteWithCronet.debug.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-InteropTestsRemoteWithCronet.debug.xcconfig"; path = "Pods/Target Support Files/Pods-InteropTestsRemoteWithCronet/Pods-InteropTestsRemoteWithCronet.debug.xcconfig"; sourceTree = "<group>"; };
20DFF2F3C97EF098FE5A3171 /* libPods-Tests.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = "libPods-Tests.a"; sourceTree = BUILT_PRODUCTS_DIR; };
2B89F3037963E6EDDD48D8C3 /* Pods-InteropTestsRemoteWithCronet.test.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-InteropTestsRemoteWithCronet.test.xcconfig"; path = "Pods/Target Support Files/Pods-InteropTestsRemoteWithCronet/Pods-InteropTestsRemoteWithCronet.test.xcconfig"; sourceTree = "<group>"; };
35F2B6BF3BAE8F0DC4AFD76E /* libPods.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = libPods.a; sourceTree = BUILT_PRODUCTS_DIR; };
386712AEACF7C2190C4B8B3F /* Pods-CronetUnitTests.cronet.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-CronetUnitTests.cronet.xcconfig"; path = "Pods/Target Support Files/Pods-CronetUnitTests/Pods-CronetUnitTests.cronet.xcconfig"; sourceTree = "<group>"; };
3B0861FC805389C52DB260D4 /* Pods-RxLibraryUnitTests.release.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-RxLibraryUnitTests.release.xcconfig"; path = "Pods/Target Support Files/Pods-RxLibraryUnitTests/Pods-RxLibraryUnitTests.release.xcconfig"; sourceTree = "<group>"; };
@ -162,15 +164,22 @@
63E240CD1B6C4E2B005F3B0E /* InteropTestsLocalSSL.m */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.objc; path = InteropTestsLocalSSL.m; sourceTree = "<group>"; };
63E240CF1B6C63DC005F3B0E /* TestCertificates.bundle */ = {isa = PBXFileReference; lastKnownFileType = "wrapper.plug-in"; path = TestCertificates.bundle; sourceTree = "<group>"; };
64F68A9A6A63CC930DD30A6E /* Pods-CronetUnitTests.debug.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-CronetUnitTests.debug.xcconfig"; path = "Pods/Target Support Files/Pods-CronetUnitTests/Pods-CronetUnitTests.debug.xcconfig"; sourceTree = "<group>"; };
6793C9D019CB268C5BB491A2 /* Pods-CoreCronetEnd2EndTests.test.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-CoreCronetEnd2EndTests.test.xcconfig"; path = "Pods/Target Support Files/Pods-CoreCronetEnd2EndTests/Pods-CoreCronetEnd2EndTests.test.xcconfig"; sourceTree = "<group>"; };
781089FAE980F51F88A3BE0B /* Pods-RxLibraryUnitTests.test.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-RxLibraryUnitTests.test.xcconfig"; path = "Pods/Target Support Files/Pods-RxLibraryUnitTests/Pods-RxLibraryUnitTests.test.xcconfig"; sourceTree = "<group>"; };
79C68EFFCB5533475D810B79 /* Pods-RxLibraryUnitTests.cronet.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-RxLibraryUnitTests.cronet.xcconfig"; path = "Pods/Target Support Files/Pods-RxLibraryUnitTests/Pods-RxLibraryUnitTests.cronet.xcconfig"; sourceTree = "<group>"; };
7A2E97E3F469CC2A758D77DE /* Pods-InteropTestsLocalSSL.release.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-InteropTestsLocalSSL.release.xcconfig"; path = "Pods/Target Support Files/Pods-InteropTestsLocalSSL/Pods-InteropTestsLocalSSL.release.xcconfig"; sourceTree = "<group>"; };
9E9444C764F0FFF64A7EB58E /* libPods-InteropTestsRemoteWithCronet.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = "libPods-InteropTestsRemoteWithCronet.a"; sourceTree = BUILT_PRODUCTS_DIR; };
A0361771A855917162911180 /* Pods-Tests.test.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-Tests.test.xcconfig"; path = "Pods/Target Support Files/Pods-Tests/Pods-Tests.test.xcconfig"; sourceTree = "<group>"; };
A58BE6DF1C62D1739EBB2C78 /* libPods-RxLibraryUnitTests.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = "libPods-RxLibraryUnitTests.a"; sourceTree = BUILT_PRODUCTS_DIR; };
A6F832FCEFA6F6881E620F12 /* Pods-InteropTestsRemote.test.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-InteropTestsRemote.test.xcconfig"; path = "Pods/Target Support Files/Pods-InteropTestsRemote/Pods-InteropTestsRemote.test.xcconfig"; sourceTree = "<group>"; };
AA7CB64B4DD9915AE7C03163 /* Pods-InteropTestsLocalCleartext.cronet.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-InteropTestsLocalCleartext.cronet.xcconfig"; path = "Pods/Target Support Files/Pods-InteropTestsLocalCleartext/Pods-InteropTestsLocalCleartext.cronet.xcconfig"; sourceTree = "<group>"; };
AC414EF7A6BF76ED02B6E480 /* Pods-InteropTestsRemoteWithCronet.release.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-InteropTestsRemoteWithCronet.release.xcconfig"; path = "Pods/Target Support Files/Pods-InteropTestsRemoteWithCronet/Pods-InteropTestsRemoteWithCronet.release.xcconfig"; sourceTree = "<group>"; };
B226619DC4E709E0FFFF94B8 /* Pods-CronetUnitTests.test.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-CronetUnitTests.test.xcconfig"; path = "Pods/Target Support Files/Pods-CronetUnitTests/Pods-CronetUnitTests.test.xcconfig"; sourceTree = "<group>"; };
B94C27C06733CF98CE1B2757 /* Pods-AllTests.debug.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-AllTests.debug.xcconfig"; path = "Pods/Target Support Files/Pods-AllTests/Pods-AllTests.debug.xcconfig"; sourceTree = "<group>"; };
C6134277D2EB8B380862A03F /* libPods-CronetUnitTests.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = "libPods-CronetUnitTests.a"; sourceTree = BUILT_PRODUCTS_DIR; };
CAE086D5B470DA367D415AB0 /* libPods-AllTests.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = "libPods-AllTests.a"; sourceTree = BUILT_PRODUCTS_DIR; };
D13BEC8181B8E678A1B52F54 /* Pods-InteropTestsLocalSSL.test.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-InteropTestsLocalSSL.test.xcconfig"; path = "Pods/Target Support Files/Pods-InteropTestsLocalSSL/Pods-InteropTestsLocalSSL.test.xcconfig"; sourceTree = "<group>"; };
DB1F4391AF69D20D38D74B67 /* Pods-AllTests.test.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-AllTests.test.xcconfig"; path = "Pods/Target Support Files/Pods-AllTests/Pods-AllTests.test.xcconfig"; sourceTree = "<group>"; };
DBE059B4AC7A51919467EEC0 /* libPods-InteropTestsRemote.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = "libPods-InteropTestsRemote.a"; sourceTree = BUILT_PRODUCTS_DIR; };
DBEDE45BDA60DF1E1C8950C0 /* libPods-InteropTestsLocalSSL.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = "libPods-InteropTestsLocalSSL.a"; sourceTree = BUILT_PRODUCTS_DIR; };
DC3CA1D948F068E76957A861 /* Pods-InteropTestsRemote.debug.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-InteropTestsRemote.debug.xcconfig"; path = "Pods/Target Support Files/Pods-InteropTestsRemote/Pods-InteropTestsRemote.debug.xcconfig"; sourceTree = "<group>"; };
@ -317,6 +326,15 @@
64F68A9A6A63CC930DD30A6E /* Pods-CronetUnitTests.debug.xcconfig */,
386712AEACF7C2190C4B8B3F /* Pods-CronetUnitTests.cronet.xcconfig */,
02192CF1FF9534E3D18C65FC /* Pods-CronetUnitTests.release.xcconfig */,
DB1F4391AF69D20D38D74B67 /* Pods-AllTests.test.xcconfig */,
6793C9D019CB268C5BB491A2 /* Pods-CoreCronetEnd2EndTests.test.xcconfig */,
B226619DC4E709E0FFFF94B8 /* Pods-CronetUnitTests.test.xcconfig */,
1588C85DEAF7FC0ACDEA4C02 /* Pods-InteropTestsLocalCleartext.test.xcconfig */,
D13BEC8181B8E678A1B52F54 /* Pods-InteropTestsLocalSSL.test.xcconfig */,
A6F832FCEFA6F6881E620F12 /* Pods-InteropTestsRemote.test.xcconfig */,
2B89F3037963E6EDDD48D8C3 /* Pods-InteropTestsRemoteWithCronet.test.xcconfig */,
781089FAE980F51F88A3BE0B /* Pods-RxLibraryUnitTests.test.xcconfig */,
A0361771A855917162911180 /* Pods-Tests.test.xcconfig */,
);
name = Pods;
sourceTree = "<group>";
@ -1237,6 +1255,210 @@
/* End PBXTargetDependency section */
/* Begin XCBuildConfiguration section */
5E1228981E4D400F00E8504F /* Test */ = {
isa = XCBuildConfiguration;
buildSettings = {
ALWAYS_SEARCH_USER_PATHS = NO;
CLANG_CXX_LANGUAGE_STANDARD = "gnu++0x";
CLANG_CXX_LIBRARY = "libc++";
CLANG_ENABLE_MODULES = YES;
CLANG_ENABLE_OBJC_ARC = YES;
CLANG_WARN_BOOL_CONVERSION = YES;
CLANG_WARN_CONSTANT_CONVERSION = YES;
CLANG_WARN_DIRECT_OBJC_ISA_USAGE = YES_ERROR;
CLANG_WARN_EMPTY_BODY = YES;
CLANG_WARN_ENUM_CONVERSION = YES;
CLANG_WARN_INT_CONVERSION = YES;
CLANG_WARN_OBJC_ROOT_CLASS = YES_ERROR;
CLANG_WARN_UNREACHABLE_CODE = YES;
CLANG_WARN__DUPLICATE_METHOD_MATCH = YES;
COPY_PHASE_STRIP = NO;
DEBUG_INFORMATION_FORMAT = "dwarf-with-dsym";
ENABLE_STRICT_OBJC_MSGSEND = YES;
GCC_C_LANGUAGE_STANDARD = gnu99;
GCC_DYNAMIC_NO_PIC = NO;
GCC_NO_COMMON_BLOCKS = YES;
GCC_OPTIMIZATION_LEVEL = 0;
GCC_PREPROCESSOR_DEFINITIONS = (
"DEBUG=1",
"$(inherited)",
);
GCC_SYMBOLS_PRIVATE_EXTERN = NO;
GCC_TREAT_WARNINGS_AS_ERRORS = YES;
GCC_WARN_64_TO_32_BIT_CONVERSION = YES;
GCC_WARN_ABOUT_RETURN_TYPE = YES_ERROR;
GCC_WARN_UNDECLARED_SELECTOR = YES;
GCC_WARN_UNINITIALIZED_AUTOS = YES_AGGRESSIVE;
GCC_WARN_UNUSED_FUNCTION = YES;
GCC_WARN_UNUSED_VARIABLE = YES;
IPHONEOS_DEPLOYMENT_TARGET = 8.3;
MTL_ENABLE_DEBUG_INFO = YES;
ONLY_ACTIVE_ARCH = YES;
SDKROOT = iphoneos;
};
name = Test;
};
5E1228991E4D400F00E8504F /* Test */ = {
isa = XCBuildConfiguration;
baseConfigurationReference = A0361771A855917162911180 /* Pods-Tests.test.xcconfig */;
buildSettings = {
GCC_TREAT_WARNINGS_AS_ERRORS = YES;
PRODUCT_NAME = "$(TARGET_NAME)";
SKIP_INSTALL = YES;
};
name = Test;
};
5E12289A1E4D400F00E8504F /* Test */ = {
isa = XCBuildConfiguration;
baseConfigurationReference = DB1F4391AF69D20D38D74B67 /* Pods-AllTests.test.xcconfig */;
buildSettings = {
FRAMEWORK_SEARCH_PATHS = (
"$(SDKROOT)/Developer/Library/Frameworks",
"$(inherited)",
);
GCC_PREPROCESSOR_DEFINITIONS = (
"DEBUG=1",
"$(inherited)",
"GRPC_TEST_OBJC=1",
);
INFOPLIST_FILE = Info.plist;
LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
PRODUCT_NAME = "$(TARGET_NAME)";
};
name = Test;
};
5E12289B1E4D400F00E8504F /* Test */ = {
isa = XCBuildConfiguration;
baseConfigurationReference = 781089FAE980F51F88A3BE0B /* Pods-RxLibraryUnitTests.test.xcconfig */;
buildSettings = {
DEBUG_INFORMATION_FORMAT = dwarf;
ENABLE_TESTABILITY = YES;
INFOPLIST_FILE = Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.0;
LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
PRODUCT_BUNDLE_IDENTIFIER = io.grpc.RxLibraryUnitTests;
PRODUCT_NAME = "$(TARGET_NAME)";
};
name = Test;
};
5E12289C1E4D400F00E8504F /* Test */ = {
isa = XCBuildConfiguration;
baseConfigurationReference = A6F832FCEFA6F6881E620F12 /* Pods-InteropTestsRemote.test.xcconfig */;
buildSettings = {
DEBUG_INFORMATION_FORMAT = dwarf;
ENABLE_TESTABILITY = YES;
GCC_PREPROCESSOR_DEFINITIONS = (
"$(inherited)",
"COCOAPODS=1",
"$(inherited)",
"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
"GRPC_TEST_OBJC=1",
);
INFOPLIST_FILE = Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.0;
LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
PRODUCT_BUNDLE_IDENTIFIER = io.grpc.InteropTests;
PRODUCT_NAME = "$(TARGET_NAME)";
};
name = Test;
};
5E12289D1E4D400F00E8504F /* Test */ = {
isa = XCBuildConfiguration;
baseConfigurationReference = D13BEC8181B8E678A1B52F54 /* Pods-InteropTestsLocalSSL.test.xcconfig */;
buildSettings = {
DEBUG_INFORMATION_FORMAT = dwarf;
ENABLE_TESTABILITY = YES;
GCC_PREPROCESSOR_DEFINITIONS = (
"$(inherited)",
"COCOAPODS=1",
"$(inherited)",
"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
"GRPC_TEST_OBJC=1",
);
INFOPLIST_FILE = Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.0;
LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
PRODUCT_BUNDLE_IDENTIFIER = io.grpc.InteropTestsLocalSSL;
PRODUCT_NAME = "$(TARGET_NAME)";
};
name = Test;
};
5E12289E1E4D400F00E8504F /* Test */ = {
isa = XCBuildConfiguration;
baseConfigurationReference = 1588C85DEAF7FC0ACDEA4C02 /* Pods-InteropTestsLocalCleartext.test.xcconfig */;
buildSettings = {
DEBUG_INFORMATION_FORMAT = dwarf;
ENABLE_TESTABILITY = YES;
GCC_PREPROCESSOR_DEFINITIONS = (
"$(inherited)",
"COCOAPODS=1",
"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
"GRPC_TEST_OBJC=1",
);
INFOPLIST_FILE = Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.0;
LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
PRODUCT_BUNDLE_IDENTIFIER = io.grpc.InteropTestsLocalCleartext;
PRODUCT_NAME = "$(TARGET_NAME)";
};
name = Test;
};
5E12289F1E4D400F00E8504F /* Test */ = {
isa = XCBuildConfiguration;
baseConfigurationReference = 6793C9D019CB268C5BB491A2 /* Pods-CoreCronetEnd2EndTests.test.xcconfig */;
buildSettings = {
CLANG_ANALYZER_NONNULL = YES;
"CODE_SIGN_IDENTITY[sdk=iphoneos*]" = "iPhone Developer";
DEBUG_INFORMATION_FORMAT = dwarf;
ENABLE_TESTABILITY = YES;
INFOPLIST_FILE = CoreCronetEnd2EndTests/Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.3;
LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
PRODUCT_BUNDLE_IDENTIFIER = io.grpc.CoreCronetEnd2EndTests;
PRODUCT_NAME = "$(TARGET_NAME)";
USER_HEADER_SEARCH_PATHS = "$(inherited) \"${PODS_ROOT}/../../../..\"";
};
name = Test;
};
5E1228A01E4D400F00E8504F /* Test */ = {
isa = XCBuildConfiguration;
baseConfigurationReference = 2B89F3037963E6EDDD48D8C3 /* Pods-InteropTestsRemoteWithCronet.test.xcconfig */;
buildSettings = {
CLANG_ANALYZER_NONNULL = YES;
"CODE_SIGN_IDENTITY[sdk=iphoneos*]" = "iPhone Developer";
DEBUG_INFORMATION_FORMAT = dwarf;
ENABLE_TESTABILITY = YES;
GCC_PREPROCESSOR_DEFINITIONS = (
"$(inherited)",
"COCOAPODS=1",
"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
"GRPC_TEST_OBJC=1",
);
INFOPLIST_FILE = InteropTestsRemoteWithCronet/Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.3;
LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
PRODUCT_BUNDLE_IDENTIFIER = io.grpc.InteropTestsRemoteWithCronet;
PRODUCT_NAME = "$(TARGET_NAME)";
};
name = Test;
};
5E1228A11E4D400F00E8504F /* Test */ = {
isa = XCBuildConfiguration;
baseConfigurationReference = B226619DC4E709E0FFFF94B8 /* Pods-CronetUnitTests.test.xcconfig */;
buildSettings = {
CLANG_ANALYZER_NONNULL = YES;
"CODE_SIGN_IDENTITY[sdk=iphoneos*]" = "iPhone Developer";
DEBUG_INFORMATION_FORMAT = dwarf;
ENABLE_TESTABILITY = YES;
INFOPLIST_FILE = CronetUnitTests/Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.3;
LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
PRODUCT_BUNDLE_IDENTIFIER = io.grpc.CronetUnitTests;
PRODUCT_NAME = "$(TARGET_NAME)";
USER_HEADER_SEARCH_PATHS = "\"${PODS_ROOT}/../../../..\" $(inherited)";
};
name = Test;
};
5E8A5DAC1D3840B4000F8BC4 /* Debug */ = {
isa = XCBuildConfiguration;
baseConfigurationReference = 0D2284C3DF7E57F0ED504E39 /* Pods-CoreCronetEnd2EndTests.debug.xcconfig */;
@ -1407,6 +1629,12 @@
buildSettings = {
DEBUG_INFORMATION_FORMAT = dwarf;
ENABLE_TESTABILITY = YES;
GCC_PREPROCESSOR_DEFINITIONS = (
"$(inherited)",
"COCOAPODS=1",
"$(inherited)",
"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
);
INFOPLIST_FILE = Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.0;
LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
@ -1471,10 +1699,10 @@
GCC_PREPROCESSOR_DEFINITIONS = (
"$(inherited)",
"COCOAPODS=1",
"$(inherited)",
"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
"GRPC_COMPILE_WITH_CRONET=1",
"GRPC_CRONET_WITH_PACKET_COALESCING=1",
"GRPC_TEST_OBJC=1",
);
INFOPLIST_FILE = InteropTestsRemoteWithCronet/Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.3;
@ -1495,7 +1723,6 @@
GCC_PREPROCESSOR_DEFINITIONS = (
"$(inherited)",
"COCOAPODS=1",
"$(inherited)",
"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
);
INFOPLIST_FILE = InteropTestsRemoteWithCronet/Info.plist;
@ -1770,6 +1997,7 @@
isa = XCConfigurationList;
buildConfigurations = (
5E8A5DAC1D3840B4000F8BC4 /* Debug */,
5E12289F1E4D400F00E8504F /* Test */,
5EC3C7A71D4FC18C000330E2 /* Cronet */,
5E8A5DAD1D3840B4000F8BC4 /* Release */,
);
@ -1780,6 +2008,7 @@
isa = XCConfigurationList;
buildConfigurations = (
5EAD6D2C1E27047400002378 /* Debug */,
5E1228A11E4D400F00E8504F /* Test */,
5EAD6D2D1E27047400002378 /* Cronet */,
5EAD6D2E1E27047400002378 /* Release */,
);
@ -1790,6 +2019,7 @@
isa = XCConfigurationList;
buildConfigurations = (
5EE84BF91D4717E40050C6CC /* Debug */,
5E1228A01E4D400F00E8504F /* Test */,
5EC3C7A81D4FC18C000330E2 /* Cronet */,
5EE84BFA1D4717E40050C6CC /* Release */,
);
@ -1800,6 +2030,7 @@
isa = XCConfigurationList;
buildConfigurations = (
63423F4E1B150A5F006CF63C /* Debug */,
5E12289A1E4D400F00E8504F /* Test */,
5EC3C7A21D4FC18C000330E2 /* Cronet */,
63423F4F1B150A5F006CF63C /* Release */,
);
@ -1810,6 +2041,7 @@
isa = XCConfigurationList;
buildConfigurations = (
635697D91B14FC11007A7283 /* Debug */,
5E1228981E4D400F00E8504F /* Test */,
5EC3C7A01D4FC18C000330E2 /* Cronet */,
635697DA1B14FC11007A7283 /* Release */,
);
@ -1820,6 +2052,7 @@
isa = XCConfigurationList;
buildConfigurations = (
635697DC1B14FC11007A7283 /* Debug */,
5E1228991E4D400F00E8504F /* Test */,
5EC3C7A11D4FC18C000330E2 /* Cronet */,
635697DD1B14FC11007A7283 /* Release */,
);
@ -1830,6 +2063,7 @@
isa = XCConfigurationList;
buildConfigurations = (
63DC841C1BE15179000708E8 /* Debug */,
5E12289B1E4D400F00E8504F /* Test */,
5EC3C7A31D4FC18C000330E2 /* Cronet */,
63DC841D1BE15179000708E8 /* Release */,
);
@ -1840,6 +2074,7 @@
isa = XCConfigurationList;
buildConfigurations = (
63DC842C1BE15267000708E8 /* Debug */,
5E12289C1E4D400F00E8504F /* Test */,
5EC3C7A41D4FC18C000330E2 /* Cronet */,
63DC842D1BE15267000708E8 /* Release */,
);
@ -1850,6 +2085,7 @@
isa = XCConfigurationList;
buildConfigurations = (
63DC843D1BE15294000708E8 /* Debug */,
5E12289D1E4D400F00E8504F /* Test */,
5EC3C7A51D4FC18C000330E2 /* Cronet */,
63DC843E1BE15294000708E8 /* Release */,
);
@ -1860,6 +2096,7 @@
isa = XCConfigurationList;
buildConfigurations = (
63DC844C1BE152B5000708E8 /* Debug */,
5E12289E1E4D400F00E8504F /* Test */,
5EC3C7A61D4FC18C000330E2 /* Cronet */,
63DC844D1BE152B5000708E8 /* Release */,
);

@ -23,7 +23,7 @@
</BuildActionEntries>
</BuildAction>
<TestAction
buildConfiguration = "Debug"
buildConfiguration = "Test"
selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
shouldUseLaunchSchemeArgsEnv = "YES">
@ -63,7 +63,7 @@
</AdditionalOptions>
</TestAction>
<LaunchAction
buildConfiguration = "Debug"
buildConfiguration = "Test"
selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
launchStyle = "0"
@ -101,7 +101,7 @@
</MacroExpansion>
</ProfileAction>
<AnalyzeAction
buildConfiguration = "Debug">
buildConfiguration = "Test">
</AnalyzeAction>
<ArchiveAction
buildConfiguration = "Release"

@ -23,7 +23,7 @@
</BuildActionEntries>
</BuildAction>
<TestAction
buildConfiguration = "Debug"
buildConfiguration = "Test"
selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
shouldUseLaunchSchemeArgsEnv = "YES">
@ -54,7 +54,7 @@
</AdditionalOptions>
</TestAction>
<LaunchAction
buildConfiguration = "Debug"
buildConfiguration = "Test"
selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
launchStyle = "0"
@ -92,7 +92,7 @@
</MacroExpansion>
</ProfileAction>
<AnalyzeAction
buildConfiguration = "Debug">
buildConfiguration = "Test">
</AnalyzeAction>
<ArchiveAction
buildConfiguration = "Release"

@ -23,7 +23,7 @@
</BuildActionEntries>
</BuildAction>
<TestAction
buildConfiguration = "Debug"
buildConfiguration = "Test"
selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
shouldUseLaunchSchemeArgsEnv = "YES">
@ -57,7 +57,7 @@
</AdditionalOptions>
</TestAction>
<LaunchAction
buildConfiguration = "Debug"
buildConfiguration = "Test"
selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
launchStyle = "0"
@ -86,7 +86,7 @@
debugDocumentVersioning = "YES">
</ProfileAction>
<AnalyzeAction
buildConfiguration = "Debug">
buildConfiguration = "Test">
</AnalyzeAction>
<ArchiveAction
buildConfiguration = "Release"

@ -23,7 +23,7 @@
</BuildActionEntries>
</BuildAction>
<TestAction
buildConfiguration = "Debug"
buildConfiguration = "Test"
selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
shouldUseLaunchSchemeArgsEnv = "YES">
@ -57,7 +57,7 @@
</AdditionalOptions>
</TestAction>
<LaunchAction
buildConfiguration = "Debug"
buildConfiguration = "Test"
selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
launchStyle = "0"
@ -86,7 +86,7 @@
debugDocumentVersioning = "YES">
</ProfileAction>
<AnalyzeAction
buildConfiguration = "Debug">
buildConfiguration = "Test">
</AnalyzeAction>
<ArchiveAction
buildConfiguration = "Release"

@ -1,5 +1,5 @@
#Overview
# Overview
This directory contains source code for PHP implementation of gRPC layered on
shared C library.

@ -131,7 +131,7 @@ abstract class AbstractCall
// Proto3 implementation
if (method_exists($data, 'encode')) {
return $data->encode();
} else if (method_exists($data, 'serializeToString')) {
} elseif (method_exists($data, 'serializeToString')) {
return $data->serializeToString();
}

@ -69,7 +69,7 @@ class ServerTest extends PHPUnit_Framework_TestCase
$this->server = new Grpc\Server();
$port = $this->server->addHttp2Port('0.0.0.0:0');
$this->server->start();
$channel = new Grpc\Channel('localhost:' . $port,
$channel = new Grpc\Channel('localhost:'.$port,
['credentials' => Grpc\ChannelCredentials::createInsecure()]);
$deadline = Grpc\Timeval::infFuture();

@ -29,7 +29,7 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
FROM golang:1.5
FROM golang:latest
<%include file="../../go_path.include"/>
<%include file="../../python_deps.include"/>

@ -29,7 +29,7 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
FROM golang:1.5
FROM golang:latest
<%include file="../../go_path.include"/>
<%include file="../../python_deps.include"/>

@ -29,7 +29,7 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
FROM golang:1.5
FROM golang:latest
<%include file="../../gcp_api_libraries.include"/>
<%include file="../../python_deps.include"/>

@ -41,6 +41,7 @@
#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include <grpc/support/useful.h>
#include "src/core/ext/transport/chttp2/transport/frame_ping.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/support/env.h"
@ -109,13 +110,15 @@ static void test_keepalive_timeout(grpc_end2end_test_config config) {
grpc_raw_byte_buffer_create(&response_payload_slice, 1);
gpr_timespec deadline = five_seconds_time();
grpc_arg keepalive_args[2];
keepalive_args[0].type = GRPC_ARG_INTEGER;
keepalive_args[0].key = GRPC_ARG_HTTP2_KEEPALIVE_TIME;
keepalive_args[0].value.integer = 2;
keepalive_args[1].type = GRPC_ARG_INTEGER;
keepalive_args[1].key = GRPC_ARG_HTTP2_KEEPALIVE_TIMEOUT;
keepalive_args[1].value.integer = 0;
grpc_arg keepalive_args[] = {{.type = GRPC_ARG_INTEGER,
.key = GRPC_ARG_CLIENT_KEEPALIVE_TIME_S,
.value.integer = 2},
{.type = GRPC_ARG_INTEGER,
.key = GRPC_ARG_CLIENT_KEEPALIVE_TIMEOUT_S,
.value.integer = 0},
{.type = GRPC_ARG_INTEGER,
.key = GRPC_ARG_HTTP2_BDP_PROBE,
.value.integer = 1}};
grpc_channel_args *client_args = NULL;
client_args = grpc_channel_args_copy_and_add(client_args, keepalive_args, 2);
@ -134,6 +137,9 @@ static void test_keepalive_timeout(grpc_end2end_test_config config) {
grpc_call_error error;
grpc_slice details;
/* Disable ping ack to trigger the keepalive timeout */
grpc_set_disable_ping_ack(true);
c = grpc_channel_create_call(
f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq,
grpc_slice_from_static_string("/foo"),

@ -41,9 +41,12 @@
#include "test/core/end2end/cq_verifier.h"
#define PING_NUM 5
static void *tag(intptr_t t) { return (void *)t; }
static void test_ping(grpc_end2end_test_config config) {
static void test_ping(grpc_end2end_test_config config,
int min_time_between_pings_ms) {
grpc_end2end_test_fixture f = config.create_fixture(NULL, NULL);
cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_connectivity_state state = GRPC_CHANNEL_IDLE;
@ -51,7 +54,7 @@ static void test_ping(grpc_end2end_test_config config) {
grpc_arg a[] = {{.type = GRPC_ARG_INTEGER,
.key = GRPC_ARG_HTTP2_MIN_TIME_BETWEEN_PINGS_MS,
.value.integer = 0},
.value.integer = min_time_between_pings_ms},
{.type = GRPC_ARG_INTEGER,
.key = GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA,
.value.integer = 20}};
@ -70,7 +73,11 @@ static void test_ping(grpc_end2end_test_config config) {
READY is reached */
while (state != GRPC_CHANNEL_READY) {
grpc_channel_watch_connectivity_state(
f.client, state, grpc_timeout_seconds_to_deadline(3), f.cq, tag(99));
f.client, state,
gpr_time_add(grpc_timeout_seconds_to_deadline(3),
gpr_time_from_millis(min_time_between_pings_ms * PING_NUM,
GPR_TIMESPAN)),
f.cq, tag(99));
CQ_EXPECT_COMPLETION(cqv, tag(99), 1);
cq_verify(cqv);
state = grpc_channel_check_connectivity_state(f.client, 0);
@ -79,7 +86,7 @@ static void test_ping(grpc_end2end_test_config config) {
state == GRPC_CHANNEL_TRANSIENT_FAILURE);
}
for (i = 1; i <= 5; i++) {
for (i = 1; i <= PING_NUM; i++) {
grpc_channel_ping(f.client, f.cq, tag(i), NULL);
CQ_EXPECT_COMPLETION(cqv, tag(i), 1);
cq_verify(cqv);
@ -102,7 +109,8 @@ static void test_ping(grpc_end2end_test_config config) {
void ping(grpc_end2end_test_config config) {
GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION);
test_ping(config);
test_ping(config, 0);
test_ping(config, 100);
}
void ping_pre_init(void) {}

@ -682,6 +682,56 @@ static void test_one_slice_deleted_late(void) {
}
}
static void test_resize_to_zero(void) {
gpr_log(GPR_INFO, "** test_resize_to_zero **");
grpc_resource_quota *q = grpc_resource_quota_create("test_resize_to_zero");
grpc_resource_quota_resize(q, 0);
grpc_resource_quota_unref(q);
}
static void test_negative_rq_free_pool(void) {
gpr_log(GPR_INFO, "** test_negative_rq_free_pool **");
grpc_resource_quota *q =
grpc_resource_quota_create("test_negative_rq_free_pool");
grpc_resource_quota_resize(q, 1024);
grpc_resource_user *usr = grpc_resource_user_create(q, "usr");
grpc_resource_user_slice_allocator alloc;
int num_allocs = 0;
grpc_resource_user_slice_allocator_init(&alloc, usr, inc_int_cb, &num_allocs);
grpc_slice_buffer buffer;
grpc_slice_buffer_init(&buffer);
{
const int start_allocs = num_allocs;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_resource_user_alloc_slices(&exec_ctx, &alloc, 1024, 1, &buffer);
grpc_exec_ctx_finish(&exec_ctx);
GPR_ASSERT(num_allocs == start_allocs + 1);
}
grpc_resource_quota_resize(q, 512);
double eps = 0.0001;
GPR_ASSERT(grpc_resource_quota_get_memory_pressure(q) < 1 + eps);
GPR_ASSERT(grpc_resource_quota_get_memory_pressure(q) > 1 - eps);
{
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_resource_user_unref(&exec_ctx, usr);
grpc_exec_ctx_finish(&exec_ctx);
}
grpc_resource_quota_unref(q);
{
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_slice_buffer_destroy_internal(&exec_ctx, &buffer);
grpc_exec_ctx_finish(&exec_ctx);
}
}
int main(int argc, char **argv) {
grpc_test_init(argc, argv);
grpc_init();
@ -705,6 +755,8 @@ int main(int argc, char **argv) {
test_reclaimers_can_be_posted_repeatedly();
test_one_slice();
test_one_slice_deleted_late();
test_resize_to_zero();
test_negative_rq_free_pool();
grpc_shutdown();
return 0;
}

@ -484,6 +484,81 @@ TEST_P(AsyncEnd2endTest, SimpleClientStreaming) {
EXPECT_TRUE(recv_status.ok());
}
// Two pings and a final pong.
TEST_P(AsyncEnd2endTest, SimpleClientStreamingWithCoalescingApi) {
ResetStub();
EchoRequest send_request;
EchoRequest recv_request;
EchoResponse send_response;
EchoResponse recv_response;
Status recv_status;
ClientContext cli_ctx;
ServerContext srv_ctx;
ServerAsyncReader<EchoResponse, EchoRequest> srv_stream(&srv_ctx);
send_request.set_message(GetParam().message_content);
cli_ctx.set_initial_metadata_corked(true);
// tag:1 never comes up since no op is performed
std::unique_ptr<ClientAsyncWriter<EchoRequest>> cli_stream(
stub_->AsyncRequestStream(&cli_ctx, &recv_response, cq_.get(), tag(1)));
service_.RequestRequestStream(&srv_ctx, &srv_stream, cq_.get(), cq_.get(),
tag(2));
cli_stream->Write(send_request, tag(3));
// 65536(64KB) is the default flow control window size. Should change this
// number when default flow control window size changes. For the write of
// send_request larger than the flow control window size, tag:3 will not come
// up until server read is initiated. For write of send_request smaller than
// the flow control window size, the request can take the free ride with
// initial metadata due to coalescing, thus write tag:3 will come up here.
if (GetParam().message_content.length() < 65536) {
Verifier(GetParam().disable_blocking)
.Expect(2, true)
.Expect(3, true)
.Verify(cq_.get());
} else {
Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
}
srv_stream.Read(&recv_request, tag(4));
if (GetParam().message_content.length() < 65536) {
Verifier(GetParam().disable_blocking).Expect(4, true).Verify(cq_.get());
} else {
Verifier(GetParam().disable_blocking)
.Expect(3, true)
.Expect(4, true)
.Verify(cq_.get());
}
EXPECT_EQ(send_request.message(), recv_request.message());
cli_stream->WriteLast(send_request, WriteOptions(), tag(5));
srv_stream.Read(&recv_request, tag(6));
Verifier(GetParam().disable_blocking)
.Expect(5, true)
.Expect(6, true)
.Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
srv_stream.Read(&recv_request, tag(7));
Verifier(GetParam().disable_blocking).Expect(7, false).Verify(cq_.get());
send_response.set_message(recv_request.message());
srv_stream.Finish(send_response, Status::OK, tag(8));
cli_stream->Finish(&recv_status, tag(9));
Verifier(GetParam().disable_blocking)
.Expect(8, true)
.Expect(9, true)
.Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
EXPECT_TRUE(recv_status.ok());
}
// One ping, two pongs.
TEST_P(AsyncEnd2endTest, SimpleServerStreaming) {
ResetStub();
@ -540,6 +615,112 @@ TEST_P(AsyncEnd2endTest, SimpleServerStreaming) {
EXPECT_TRUE(recv_status.ok());
}
// One ping, two pongs. Using WriteAndFinish API
TEST_P(AsyncEnd2endTest, SimpleServerStreamingWithCoalescingApiWAF) {
ResetStub();
EchoRequest send_request;
EchoRequest recv_request;
EchoResponse send_response;
EchoResponse recv_response;
Status recv_status;
ClientContext cli_ctx;
ServerContext srv_ctx;
ServerAsyncWriter<EchoResponse> srv_stream(&srv_ctx);
send_request.set_message(GetParam().message_content);
std::unique_ptr<ClientAsyncReader<EchoResponse>> cli_stream(
stub_->AsyncResponseStream(&cli_ctx, send_request, cq_.get(), tag(1)));
service_.RequestResponseStream(&srv_ctx, &recv_request, &srv_stream,
cq_.get(), cq_.get(), tag(2));
Verifier(GetParam().disable_blocking)
.Expect(1, true)
.Expect(2, true)
.Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
send_response.set_message(recv_request.message());
srv_stream.Write(send_response, tag(3));
cli_stream->Read(&recv_response, tag(4));
Verifier(GetParam().disable_blocking)
.Expect(3, true)
.Expect(4, true)
.Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
srv_stream.WriteAndFinish(send_response, WriteOptions(), Status::OK, tag(5));
cli_stream->Read(&recv_response, tag(6));
Verifier(GetParam().disable_blocking)
.Expect(5, true)
.Expect(6, true)
.Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
cli_stream->Read(&recv_response, tag(7));
Verifier(GetParam().disable_blocking).Expect(7, false).Verify(cq_.get());
cli_stream->Finish(&recv_status, tag(8));
Verifier(GetParam().disable_blocking).Expect(8, true).Verify(cq_.get());
EXPECT_TRUE(recv_status.ok());
}
// One ping, two pongs. Using WriteLast API
TEST_P(AsyncEnd2endTest, SimpleServerStreamingWithCoalescingApiWL) {
ResetStub();
EchoRequest send_request;
EchoRequest recv_request;
EchoResponse send_response;
EchoResponse recv_response;
Status recv_status;
ClientContext cli_ctx;
ServerContext srv_ctx;
ServerAsyncWriter<EchoResponse> srv_stream(&srv_ctx);
send_request.set_message(GetParam().message_content);
std::unique_ptr<ClientAsyncReader<EchoResponse>> cli_stream(
stub_->AsyncResponseStream(&cli_ctx, send_request, cq_.get(), tag(1)));
service_.RequestResponseStream(&srv_ctx, &recv_request, &srv_stream,
cq_.get(), cq_.get(), tag(2));
Verifier(GetParam().disable_blocking)
.Expect(1, true)
.Expect(2, true)
.Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
send_response.set_message(recv_request.message());
srv_stream.Write(send_response, tag(3));
cli_stream->Read(&recv_response, tag(4));
Verifier(GetParam().disable_blocking)
.Expect(3, true)
.Expect(4, true)
.Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
srv_stream.WriteLast(send_response, WriteOptions(), tag(5));
cli_stream->Read(&recv_response, tag(6));
srv_stream.Finish(Status::OK, tag(7));
Verifier(GetParam().disable_blocking)
.Expect(5, true)
.Expect(6, true)
.Expect(7, true)
.Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
cli_stream->Read(&recv_response, tag(8));
Verifier(GetParam().disable_blocking).Expect(8, false).Verify(cq_.get());
cli_stream->Finish(&recv_status, tag(9));
Verifier(GetParam().disable_blocking).Expect(9, true).Verify(cq_.get());
EXPECT_TRUE(recv_status.ok());
}
// One ping, one pong.
TEST_P(AsyncEnd2endTest, SimpleBidiStreaming) {
ResetStub();
@ -599,6 +780,144 @@ TEST_P(AsyncEnd2endTest, SimpleBidiStreaming) {
EXPECT_TRUE(recv_status.ok());
}
// One ping, one pong. Using server:WriteAndFinish api
TEST_P(AsyncEnd2endTest, SimpleBidiStreamingWithCoalescingApiWAF) {
ResetStub();
EchoRequest send_request;
EchoRequest recv_request;
EchoResponse send_response;
EchoResponse recv_response;
Status recv_status;
ClientContext cli_ctx;
ServerContext srv_ctx;
ServerAsyncReaderWriter<EchoResponse, EchoRequest> srv_stream(&srv_ctx);
send_request.set_message(GetParam().message_content);
cli_ctx.set_initial_metadata_corked(true);
std::unique_ptr<ClientAsyncReaderWriter<EchoRequest, EchoResponse>>
cli_stream(stub_->AsyncBidiStream(&cli_ctx, cq_.get(), tag(1)));
service_.RequestBidiStream(&srv_ctx, &srv_stream, cq_.get(), cq_.get(),
tag(2));
cli_stream->WriteLast(send_request, WriteOptions(), tag(3));
// 65536(64KB) is the default flow control window size. Should change this
// number when default flow control window size changes. For the write of
// send_request larger than the flow control window size, tag:3 will not come
// up until server read is initiated. For write of send_request smaller than
// the flow control window size, the request can take the free ride with
// initial metadata due to coalescing, thus write tag:3 will come up here.
if (GetParam().message_content.length() < 65536) {
Verifier(GetParam().disable_blocking)
.Expect(2, true)
.Expect(3, true)
.Verify(cq_.get());
} else {
Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
}
srv_stream.Read(&recv_request, tag(4));
if (GetParam().message_content.length() < 65536) {
Verifier(GetParam().disable_blocking).Expect(4, true).Verify(cq_.get());
} else {
Verifier(GetParam().disable_blocking)
.Expect(3, true)
.Expect(4, true)
.Verify(cq_.get());
}
EXPECT_EQ(send_request.message(), recv_request.message());
srv_stream.Read(&recv_request, tag(5));
Verifier(GetParam().disable_blocking).Expect(5, false).Verify(cq_.get());
send_response.set_message(recv_request.message());
srv_stream.WriteAndFinish(send_response, WriteOptions(), Status::OK, tag(6));
cli_stream->Read(&recv_response, tag(7));
Verifier(GetParam().disable_blocking)
.Expect(6, true)
.Expect(7, true)
.Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
cli_stream->Finish(&recv_status, tag(8));
Verifier(GetParam().disable_blocking).Expect(8, true).Verify(cq_.get());
EXPECT_TRUE(recv_status.ok());
}
// One ping, one pong. Using server:WriteLast api
TEST_P(AsyncEnd2endTest, SimpleBidiStreamingWithCoalescingApiWL) {
ResetStub();
EchoRequest send_request;
EchoRequest recv_request;
EchoResponse send_response;
EchoResponse recv_response;
Status recv_status;
ClientContext cli_ctx;
ServerContext srv_ctx;
ServerAsyncReaderWriter<EchoResponse, EchoRequest> srv_stream(&srv_ctx);
send_request.set_message(GetParam().message_content);
cli_ctx.set_initial_metadata_corked(true);
std::unique_ptr<ClientAsyncReaderWriter<EchoRequest, EchoResponse>>
cli_stream(stub_->AsyncBidiStream(&cli_ctx, cq_.get(), tag(1)));
service_.RequestBidiStream(&srv_ctx, &srv_stream, cq_.get(), cq_.get(),
tag(2));
cli_stream->WriteLast(send_request, WriteOptions(), tag(3));
// 65536(64KB) is the default flow control window size. Should change this
// number when default flow control window size changes. For the write of
// send_request larger than the flow control window size, tag:3 will not come
// up until server read is initiated. For write of send_request smaller than
// the flow control window size, the request can take the free ride with
// initial metadata due to coalescing, thus write tag:3 will come up here.
if (GetParam().message_content.length() < 65536) {
Verifier(GetParam().disable_blocking)
.Expect(2, true)
.Expect(3, true)
.Verify(cq_.get());
} else {
Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
}
srv_stream.Read(&recv_request, tag(4));
if (GetParam().message_content.length() < 65536) {
Verifier(GetParam().disable_blocking).Expect(4, true).Verify(cq_.get());
} else {
Verifier(GetParam().disable_blocking)
.Expect(3, true)
.Expect(4, true)
.Verify(cq_.get());
}
EXPECT_EQ(send_request.message(), recv_request.message());
srv_stream.Read(&recv_request, tag(5));
Verifier(GetParam().disable_blocking).Expect(5, false).Verify(cq_.get());
send_response.set_message(recv_request.message());
srv_stream.WriteLast(send_response, WriteOptions(), tag(6));
srv_stream.Finish(Status::OK, tag(7));
cli_stream->Read(&recv_response, tag(8));
Verifier(GetParam().disable_blocking)
.Expect(6, true)
.Expect(7, true)
.Expect(8, true)
.Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
cli_stream->Finish(&recv_status, tag(9));
Verifier(GetParam().disable_blocking).Expect(9, true).Verify(cq_.get());
EXPECT_TRUE(recv_status.ok());
}
// Metadata tests
TEST_P(AsyncEnd2endTest, ClientInitialMetadataRpc) {
ResetStub();

@ -702,6 +702,21 @@ TEST_P(End2endTest, RequestStreamOneRequest) {
EXPECT_TRUE(s.ok());
}
TEST_P(End2endTest, RequestStreamOneRequestWithCoalescingApi) {
ResetStub();
EchoRequest request;
EchoResponse response;
ClientContext context;
context.set_initial_metadata_corked(true);
auto stream = stub_->RequestStream(&context, &response);
request.set_message("hello");
stream->WriteLast(request, WriteOptions());
Status s = stream->Finish();
EXPECT_EQ(response.message(), request.message());
EXPECT_TRUE(s.ok());
}
TEST_P(End2endTest, RequestStreamTwoRequests) {
ResetStub();
EchoRequest request;
@ -718,6 +733,22 @@ TEST_P(End2endTest, RequestStreamTwoRequests) {
EXPECT_TRUE(s.ok());
}
TEST_P(End2endTest, RequestStreamTwoRequestsWithCoalescingApi) {
ResetStub();
EchoRequest request;
EchoResponse response;
ClientContext context;
context.set_initial_metadata_corked(true);
auto stream = stub_->RequestStream(&context, &response);
request.set_message("hello");
EXPECT_TRUE(stream->Write(request));
stream->WriteLast(request, WriteOptions());
Status s = stream->Finish();
EXPECT_EQ(response.message(), "hellohello");
EXPECT_TRUE(s.ok());
}
TEST_P(End2endTest, ResponseStream) {
ResetStub();
EchoRequest request;
@ -738,6 +769,27 @@ TEST_P(End2endTest, ResponseStream) {
EXPECT_TRUE(s.ok());
}
TEST_P(End2endTest, ResponseStreamWithCoalescingApi) {
ResetStub();
EchoRequest request;
EchoResponse response;
ClientContext context;
request.set_message("hello");
context.AddMetadata(kServerUseCoalescingApi, "1");
auto stream = stub_->ResponseStream(&context, request);
EXPECT_TRUE(stream->Read(&response));
EXPECT_EQ(response.message(), request.message() + "0");
EXPECT_TRUE(stream->Read(&response));
EXPECT_EQ(response.message(), request.message() + "1");
EXPECT_TRUE(stream->Read(&response));
EXPECT_EQ(response.message(), request.message() + "2");
EXPECT_FALSE(stream->Read(&response));
Status s = stream->Finish();
EXPECT_TRUE(s.ok());
}
TEST_P(End2endTest, BidiStream) {
ResetStub();
EchoRequest request;
@ -770,6 +822,39 @@ TEST_P(End2endTest, BidiStream) {
EXPECT_TRUE(s.ok());
}
TEST_P(End2endTest, BidiStreamWithCoalescingApi) {
ResetStub();
EchoRequest request;
EchoResponse response;
ClientContext context;
context.AddMetadata(kServerFinishAfterNReads, "3");
context.set_initial_metadata_corked(true);
grpc::string msg("hello");
auto stream = stub_->BidiStream(&context);
request.set_message(msg + "0");
EXPECT_TRUE(stream->Write(request));
EXPECT_TRUE(stream->Read(&response));
EXPECT_EQ(response.message(), request.message());
request.set_message(msg + "1");
EXPECT_TRUE(stream->Write(request));
EXPECT_TRUE(stream->Read(&response));
EXPECT_EQ(response.message(), request.message());
request.set_message(msg + "2");
stream->WriteLast(request, WriteOptions());
EXPECT_TRUE(stream->Read(&response));
EXPECT_EQ(response.message(), request.message());
EXPECT_FALSE(stream->Read(&response));
EXPECT_FALSE(stream->Read(&response));
Status s = stream->Finish();
EXPECT_TRUE(s.ok());
}
// Talk to the two services with the same name but different package names.
// The two stubs are created on the same channel.
TEST_P(End2endTest, DiffPackageServices) {

@ -89,7 +89,7 @@ class MockClientReaderWriter<EchoRequest, EchoResponse> final
return true;
}
bool Write(const EchoRequest& msg, const WriteOptions& options) override {
bool Write(const EchoRequest& msg, WriteOptions options) override {
gpr_log(GPR_INFO, "mock recv msg %s", msg.message().c_str());
last_message_ = msg.message();
return true;

@ -246,6 +246,9 @@ Status TestServiceImpl::ResponseStream(ServerContext* context,
int server_try_cancel = GetIntValueFromMetadata(
kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL);
int server_coalescing_api = GetIntValueFromMetadata(
kServerUseCoalescingApi, context->client_metadata(), 0);
if (server_try_cancel == CANCEL_BEFORE_PROCESSING) {
ServerTryCancel(context);
return Status::CANCELLED;
@ -260,7 +263,11 @@ Status TestServiceImpl::ResponseStream(ServerContext* context,
for (int i = 0; i < kNumResponseStreamsMsgs; i++) {
response.set_message(request->message() + grpc::to_string(i));
writer->Write(response);
if (i == kNumResponseStreamsMsgs - 1 && server_coalescing_api != 0) {
writer->WriteLast(response, WriteOptions());
} else {
writer->Write(response);
}
}
if (server_try_cancel_thd != nullptr) {
@ -305,10 +312,21 @@ Status TestServiceImpl::BidiStream(
new std::thread(&TestServiceImpl::ServerTryCancel, this, context);
}
// kServerFinishAfterNReads suggests after how many reads, the server should
// write the last message and send status (coalesced using WriteLast)
int server_write_last = GetIntValueFromMetadata(
kServerFinishAfterNReads, context->client_metadata(), 0);
int read_counts = 0;
while (stream->Read(&request)) {
read_counts++;
gpr_log(GPR_INFO, "recv msg %s", request.message().c_str());
response.set_message(request.message());
stream->Write(response);
if (read_counts == server_write_last) {
stream->WriteLast(response, WriteOptions());
} else {
stream->Write(response);
}
}
if (server_try_cancel_thd != nullptr) {

@ -48,6 +48,8 @@ const int kNumResponseStreamsMsgs = 3;
const char* const kServerCancelAfterReads = "cancel_after_reads";
const char* const kServerTryCancelRequest = "server_try_cancel";
const char* const kDebugInfoTrailerKey = "debug-info-bin";
const char* const kServerFinishAfterNReads = "server_finish_after_n_reads";
const char* const kServerUseCoalescingApi = "server_use_coalescing_api";
typedef enum {
DO_NOT_CANCEL = 0,

@ -239,6 +239,173 @@ static void BM_StreamingPingPongMsgs(benchmark::State& state) {
state.SetBytesProcessed(msg_size * state.iterations() * 2);
}
// Repeatedly makes Streaming Bidi calls (exchanging a configurable number of
// messages in each call) in a loop on a single channel. Different from
// BM_StreamingPingPong we are using stream coalescing api, e.g. WriteLast,
// WriteAndFinish, set_initial_metadata_corked. These apis aim at saving
// sendmsg syscalls for streaming by coalescing 1. initial metadata with first
// message; 2. final streaming message with trailing metadata.
//
// First parmeter (i.e state.range(0)): Message size (in bytes) to use
// Second parameter (i.e state.range(1)): Number of ping pong messages.
// Note: One ping-pong means two messages (one from client to server and
// the other from server to client):
// Third parameter (i.e state.range(2)): Switch between using WriteAndFinish
// API and WriteLast API for server.
template <class Fixture, class ClientContextMutator, class ServerContextMutator>
static void BM_StreamingPingPongWithCoalescingApi(benchmark::State& state) {
const int msg_size = state.range(0);
const int max_ping_pongs = state.range(1);
// This options is used to test out server API: WriteLast and WriteAndFinish
// respectively, since we can not use both of them on server side at the same
// time. Value 1 means we are testing out the WriteAndFinish API, and
// otherwise we are testing out the WriteLast API.
const int write_and_finish = state.range(2);
EchoTestService::AsyncService service;
std::unique_ptr<Fixture> fixture(new Fixture(&service));
{
EchoResponse send_response;
EchoResponse recv_response;
EchoRequest send_request;
EchoRequest recv_request;
if (msg_size > 0) {
send_request.set_message(std::string(msg_size, 'a'));
send_response.set_message(std::string(msg_size, 'b'));
}
std::unique_ptr<EchoTestService::Stub> stub(
EchoTestService::NewStub(fixture->channel()));
while (state.KeepRunning()) {
ServerContext svr_ctx;
ServerContextMutator svr_ctx_mut(&svr_ctx);
ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx);
service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(),
fixture->cq(), tag(0));
ClientContext cli_ctx;
ClientContextMutator cli_ctx_mut(&cli_ctx);
cli_ctx.set_initial_metadata_corked(true);
// tag:1 here will never comes up, since we are not performing any op due
// to initial metadata coalescing.
auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1));
void* t;
bool ok;
int need_tags;
// Send 'max_ping_pongs' number of ping pong messages
int ping_pong_cnt = 0;
while (ping_pong_cnt < max_ping_pongs) {
if (ping_pong_cnt == max_ping_pongs - 1) {
request_rw->WriteLast(send_request, WriteOptions(), tag(2));
} else {
request_rw->Write(send_request, tag(2)); // Start client send
}
need_tags = (1 << 2) | (1 << 3) | (1 << 4) | (1 << 5);
if (ping_pong_cnt == 0) {
// wait for the server call structure (call_hook, etc.) to be
// initialized (async stream between client side and server side
// established). It is necessary when client init metadata is
// coalesced
GPR_ASSERT(fixture->cq()->Next(&t, &ok));
while ((int)(intptr_t)t != 0) {
// In some cases tag:2 comes before tag:0 (write tag comes out
// first), this while loop is to make sure get tag:0.
int i = (int)(intptr_t)t;
GPR_ASSERT(need_tags & (1 << i));
need_tags &= ~(1 << i);
GPR_ASSERT(fixture->cq()->Next(&t, &ok));
}
}
response_rw.Read(&recv_request, tag(3)); // Start server recv
request_rw->Read(&recv_response, tag(4)); // Start client recv
while (need_tags) {
GPR_ASSERT(fixture->cq()->Next(&t, &ok));
GPR_ASSERT(ok);
int i = (int)(intptr_t)t;
// If server recv is complete, start the server send operation
if (i == 3) {
if (ping_pong_cnt == max_ping_pongs - 1) {
if (write_and_finish == 1) {
response_rw.WriteAndFinish(send_response, WriteOptions(),
Status::OK, tag(5));
} else {
response_rw.WriteLast(send_response, WriteOptions(), tag(5));
// WriteLast buffers the write, so neither server write op nor
// client read op will finish inside the while loop.
need_tags &= ~(1 << 4);
need_tags &= ~(1 << 5);
}
} else {
response_rw.Write(send_response, tag(5));
}
}
GPR_ASSERT(need_tags & (1 << i));
need_tags &= ~(1 << i);
}
ping_pong_cnt++;
}
if (max_ping_pongs == 0) {
need_tags = (1 << 6) | (1 << 7) | (1 << 8);
} else {
if (write_and_finish == 1) {
need_tags = (1 << 8);
} else {
// server's buffered write and the client's read of the buffered write
// tags should come up.
need_tags = (1 << 4) | (1 << 5) | (1 << 7) | (1 << 8);
}
}
// No message write or initial metadata write happened yet.
if (max_ping_pongs == 0) {
request_rw->WritesDone(tag(6));
// wait for server call data structure(call_hook, etc.) to be
// initialized, since initial metadata is corked.
GPR_ASSERT(fixture->cq()->Next(&t, &ok));
while ((int)(intptr_t)t != 0) {
int i = (int)(intptr_t)t;
GPR_ASSERT(need_tags & (1 << i));
need_tags &= ~(1 << i);
GPR_ASSERT(fixture->cq()->Next(&t, &ok));
}
response_rw.Finish(Status::OK, tag(7));
} else {
if (write_and_finish != 1) {
response_rw.Finish(Status::OK, tag(7));
}
}
Status recv_status;
request_rw->Finish(&recv_status, tag(8));
while (need_tags) {
GPR_ASSERT(fixture->cq()->Next(&t, &ok));
int i = (int)(intptr_t)t;
GPR_ASSERT(need_tags & (1 << i));
need_tags &= ~(1 << i);
}
GPR_ASSERT(recv_status.ok());
}
}
fixture->Finish(state);
fixture.reset();
state.SetBytesProcessed(msg_size * state.iterations() * max_ping_pongs * 2);
}
/*******************************************************************************
* CONFIGURATIONS
*/
@ -269,6 +436,30 @@ BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, InProcessCHTTP2, NoOpMutator,
BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, TCP, NoOpMutator, NoOpMutator)
->Range(0, 128 * 1024 * 1024);
// Generate Args for StreamingPingPongWithCoalescingApi benchmarks. Currently
// generates args for only "small streams" (i.e streams with 0, 1 or 2 messages)
static void StreamingPingPongWithCoalescingApiArgs(
benchmark::internal::Benchmark* b) {
int msg_size = 0;
b->Args(
{0, 0, 0}); // spl case: 0 ping-pong msgs (msg_size doesn't matter here)
b->Args(
{0, 0, 1}); // spl case: 0 ping-pong msgs (msg_size doesn't matter here)
for (msg_size = 0; msg_size <= 128 * 1024 * 1024;
msg_size == 0 ? msg_size++ : msg_size *= 8) {
b->Args({msg_size, 1, 0});
b->Args({msg_size, 2, 0});
b->Args({msg_size, 1, 1});
b->Args({msg_size, 2, 1});
}
}
BENCHMARK_TEMPLATE(BM_StreamingPingPongWithCoalescingApi, InProcessCHTTP2,
NoOpMutator, NoOpMutator)
->Apply(StreamingPingPongWithCoalescingApiArgs);
} // namespace testing
} // namespace grpc

@ -27,7 +27,7 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
FROM golang:1.5
FROM golang:latest
# Using login shell removes Go from path, so we add it.
RUN ln -s /usr/local/go/bin/go /usr/local/bin

@ -27,7 +27,7 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
FROM golang:1.5
FROM golang:latest
# Using login shell removes Go from path, so we add it.
RUN ln -s /usr/local/go/bin/go /usr/local/bin

@ -27,7 +27,7 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
FROM golang:1.5
FROM golang:latest
# Google Cloud platform API libraries
RUN apt-get update && apt-get install -y python-pip && apt-get clean

@ -83123,6 +83123,29 @@
],
"uses_polling": false
},
{
"args": [
"test/core/end2end/fuzzers/api_fuzzer_corpus/crash-59a56fa18034a104fb9f16cd58071b6ff93b8756"
],
"ci_platforms": [
"linux"
],
"cpu_cost": 0.1,
"exclude_configs": [
"tsan"
],
"exclude_iomgrs": [
"uv"
],
"flaky": false,
"language": "c",
"name": "api_fuzzer_one_entry",
"platforms": [
"mac",
"linux"
],
"uses_polling": false
},
{
"args": [
"test/core/end2end/fuzzers/api_fuzzer_corpus/crash-59b587d15c0bcdb985417cd7a133cecfcc232698"
@ -94209,6 +94232,29 @@
],
"uses_polling": false
},
{
"args": [
"test/core/end2end/fuzzers/api_fuzzer_corpus/poc-c726ee220e980ed6ad17809fd9efe2844ee61555ac08e4f88afd8901cc2dd53a"
],
"ci_platforms": [
"linux"
],
"cpu_cost": 0.1,
"exclude_configs": [
"tsan"
],
"exclude_iomgrs": [
"uv"
],
"flaky": false,
"language": "c",
"name": "api_fuzzer_one_entry",
"platforms": [
"mac",
"linux"
],
"uses_polling": false
},
{
"args": [
"test/core/end2end/fuzzers/api_fuzzer_corpus/timeout-0fa0559576ad2a45b06d0bfb84115963d7d48206"

Loading…
Cancel
Save