Merge pull request #1464 from yang-g/init

Expose max message size at the server side.
pull/1476/head
Craig Tiller 10 years ago
commit d53d87cbd3
  1. 218
      Makefile
  2. 5
      include/grpc++/config.h
  3. 10
      include/grpc++/impl/call.h
  4. 7
      include/grpc++/server.h
  5. 6
      include/grpc++/server_builder.h
  6. 6
      src/core/profiling/stap_timers.c
  7. 4
      src/core/support/cpu_windows.c
  8. 31
      src/core/surface/call.c
  9. 16
      src/cpp/common/call.cc
  10. 9
      src/cpp/proto/proto_utils.cc
  11. 3
      src/cpp/proto/proto_utils.h
  12. 37
      src/cpp/server/server.cc
  13. 5
      src/cpp/server/server_builder.cc
  14. 14
      src/python/src/grpc/_adapter/_tag.h
  15. 4
      test/build/systemtap.c
  16. 1
      test/core/end2end/gen_build_json.py
  17. 210
      test/core/end2end/tests/max_message_length.c
  18. 24
      test/cpp/end2end/end2end_test.cc
  19. 99
      tools/run_tests/tests.json
  20. 34
      vsprojects/Grpc.mak

File diff suppressed because one or more lines are too long

@ -93,13 +93,17 @@
#endif
#ifndef GRPC_CUSTOM_ZEROCOPYOUTPUTSTREAM
#include <google/protobuf/io/coded_stream.h>
#include <google/protobuf/io/zero_copy_stream.h>
#define GRPC_CUSTOM_ZEROCOPYOUTPUTSTREAM \
::google::protobuf::io::ZeroCopyOutputStream
#define GRPC_CUSTOM_ZEROCOPYINPUTSTREAM \
::google::protobuf::io::ZeroCopyInputStream
#define GRPC_CUSTOM_CODEDINPUTSTREAM \
::google::protobuf::io::CodedInputStream
#endif
#ifdef GRPC_CXX0X_NO_NULLPTR
#include <memory>
const class {
@ -126,6 +130,7 @@ typedef GRPC_CUSTOM_PROTOBUF_INT64 int64;
namespace io {
typedef GRPC_CUSTOM_ZEROCOPYOUTPUTSTREAM ZeroCopyOutputStream;
typedef GRPC_CUSTOM_ZEROCOPYINPUTSTREAM ZeroCopyInputStream;
typedef GRPC_CUSTOM_CODEDINPUTSTREAM CodedInputStream;
} // namespace io
} // namespace protobuf

@ -80,6 +80,10 @@ class CallOpBuffer : public CompletionQueueTag {
// Called by completion queue just prior to returning from Next() or Pluck()
bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE;
void set_max_message_size(int max_message_size) {
max_message_size_ = max_message_size;
}
bool got_message;
private:
@ -99,6 +103,7 @@ class CallOpBuffer : public CompletionQueueTag {
grpc::protobuf::Message* recv_message_;
ByteBuffer* recv_message_buffer_;
grpc_byte_buffer* recv_buf_;
int max_message_size_;
// Client send close
bool client_send_close_;
// Client recv status
@ -130,16 +135,21 @@ class Call GRPC_FINAL {
public:
/* call is owned by the caller */
Call(grpc_call* call, CallHook* call_hook_, CompletionQueue* cq);
Call(grpc_call* call, CallHook* call_hook_, CompletionQueue* cq,
int max_message_size);
void PerformOps(CallOpBuffer* buffer);
grpc_call* call() { return call_; }
CompletionQueue* cq() { return cq_; }
int max_message_size() { return max_message_size_; }
private:
CallHook* call_hook_;
CompletionQueue* cq_;
grpc_call* call_;
int max_message_size_;
};
} // namespace grpc

@ -79,7 +79,8 @@ class Server GRPC_FINAL : public GrpcLibrary,
class AsyncRequest;
// ServerBuilder use only
Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned);
Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
int max_message_size);
// Register a service. This call does not take ownership of the service.
// The service must exist for the lifetime of the Server instance.
bool RegisterService(RpcService* service);
@ -106,6 +107,8 @@ class Server GRPC_FINAL : public GrpcLibrary,
ServerAsyncStreamingInterface* stream,
CompletionQueue* cq, void* tag);
const int max_message_size_;
// Completion queue.
CompletionQueue cq_;
@ -126,7 +129,7 @@ class Server GRPC_FINAL : public GrpcLibrary,
// Whether the thread pool is created and owned by the server.
bool thread_pool_owned_;
private:
Server() : server_(NULL) { abort(); }
Server() : max_message_size_(-1), server_(NULL) { abort(); }
};
} // namespace grpc

@ -68,6 +68,11 @@ class ServerBuilder {
// Register a generic service.
void RegisterAsyncGenericService(AsyncGenericService* service);
// Set max message size in bytes.
void SetMaxMessageSize(int max_message_size) {
max_message_size_ = max_message_size;
}
// Add a listening port. Can be called multiple times.
void AddListeningPort(const grpc::string& addr,
std::shared_ptr<ServerCredentials> creds,
@ -87,6 +92,7 @@ class ServerBuilder {
int* selected_port;
};
int max_message_size_;
std::vector<RpcService*> services_;
std::vector<AsynchronousService*> async_services_;
std::vector<Port> ports_;

@ -42,15 +42,15 @@
#include "src/core/profiling/stap_probes.h"
/* Latency profiler API implementation. */
void grpc_timer_add_mark(int tag, void* id, const char *file, int line) {
void grpc_timer_add_mark(int tag, void* id, const char* file, int line) {
_STAP_ADD_MARK(tag);
}
void grpc_timer_begin(int tag, void* id, const char *file, int line) {
void grpc_timer_begin(int tag, void* id, const char* file, int line) {
_STAP_TIMING_NS_BEGIN(tag);
}
void grpc_timer_end(int tag, void* id, const char *file, int line) {
void grpc_timer_end(int tag, void* id, const char* file, int line) {
_STAP_TIMING_NS_END(tag);
}

@ -43,8 +43,6 @@ unsigned gpr_cpu_num_cores(void) {
return si.dwNumberOfProcessors;
}
unsigned gpr_cpu_current_cpu(void) {
return GetCurrentProcessorNumber();
}
unsigned gpr_cpu_current_cpu(void) { return GetCurrentProcessorNumber(); }
#endif /* GPR_WIN32 */

@ -247,6 +247,9 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op);
static void execute_op(grpc_call *call, grpc_transport_op *op);
static void recv_metadata(grpc_call *call, grpc_metadata_batch *metadata);
static void finish_read_ops(grpc_call *call);
static grpc_call_error cancel_with_status(
grpc_call *c, grpc_status_code status, const char *description,
gpr_uint8 locked);
grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
const void *server_transport_data,
@ -628,7 +631,7 @@ static int begin_message(grpc_call *call, grpc_begin_message msg) {
gpr_asprintf(
&message, "Message terminated early; read %d bytes, expected %d",
(int)call->incoming_message.length, (int)call->incoming_message_length);
grpc_call_cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message);
cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message, 1);
gpr_free(message);
return 0;
}
@ -639,7 +642,7 @@ static int begin_message(grpc_call *call, grpc_begin_message msg) {
&message,
"Maximum message length of %d exceeded by a message of length %d",
grpc_channel_get_max_message_length(call->channel), msg.length);
grpc_call_cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message);
cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message, 1);
gpr_free(message);
return 0;
} else if (msg.length > 0) {
@ -659,9 +662,9 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) {
}
/* we have to be reading a message to know what to do here */
if (!call->reading_message) {
grpc_call_cancel_with_status(
cancel_with_status(
call, GRPC_STATUS_INVALID_ARGUMENT,
"Received payload data while not reading a message");
"Received payload data while not reading a message", 1);
return 0;
}
/* append the slice to the incoming buffer */
@ -672,7 +675,7 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) {
gpr_asprintf(
&message, "Receiving message overflow; read %d bytes, expected %d",
(int)call->incoming_message.length, (int)call->incoming_message_length);
grpc_call_cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message);
cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message, 1);
gpr_free(message);
return 0;
} else if (call->incoming_message.length == call->incoming_message_length) {
@ -999,6 +1002,12 @@ grpc_call_error grpc_call_cancel(grpc_call *call) {
grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
grpc_status_code status,
const char *description) {
return cancel_with_status(c, status, description, 0);
}
static grpc_call_error cancel_with_status(
grpc_call *c, grpc_status_code status, const char *description,
gpr_uint8 locked) {
grpc_transport_op op;
grpc_mdstr *details =
description ? grpc_mdstr_from_string(c->metadata_context, description)
@ -1006,10 +1015,14 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
memset(&op, 0, sizeof(op));
op.cancel_with_status = status;
lock(c);
if (locked == 0) {
lock(c);
}
set_status_code(c, STATUS_FROM_API_OVERRIDE, status);
set_status_details(c, STATUS_FROM_API_OVERRIDE, details);
unlock(c);
if (locked == 0) {
unlock(c);
}
execute_op(c, &op);
@ -1030,8 +1043,8 @@ static void call_alarm(void *arg, int success) {
grpc_call *call = arg;
if (success) {
if (call->is_client) {
grpc_call_cancel_with_status(call, GRPC_STATUS_DEADLINE_EXCEEDED,
"Deadline Exceeded");
cancel_with_status(call, GRPC_STATUS_DEADLINE_EXCEEDED,
"Deadline Exceeded", 0);
} else {
grpc_call_cancel(call);
}

@ -55,6 +55,7 @@ CallOpBuffer::CallOpBuffer()
recv_message_(nullptr),
recv_message_buffer_(nullptr),
recv_buf_(nullptr),
max_message_size_(-1),
client_send_close_(false),
recv_trailing_metadata_(nullptr),
recv_status_(nullptr),
@ -311,7 +312,8 @@ bool CallOpBuffer::FinalizeResult(void** tag, bool* status) {
got_message = *status;
if (recv_message_) {
GRPC_TIMER_BEGIN(GRPC_PTAG_PROTO_DESERIALIZE, 0);
*status = *status && DeserializeProto(recv_buf_, recv_message_);
*status = *status &&
DeserializeProto(recv_buf_, recv_message_, max_message_size_);
grpc_byte_buffer_destroy(recv_buf_);
GRPC_TIMER_END(GRPC_PTAG_PROTO_DESERIALIZE, 0);
} else {
@ -338,9 +340,19 @@ bool CallOpBuffer::FinalizeResult(void** tag, bool* status) {
}
Call::Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq)
: call_hook_(call_hook), cq_(cq), call_(call) {}
: call_hook_(call_hook), cq_(cq), call_(call), max_message_size_(-1) {}
Call::Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq,
int max_message_size)
: call_hook_(call_hook),
cq_(cq),
call_(call),
max_message_size_(max_message_size) {}
void Call::PerformOps(CallOpBuffer* buffer) {
if (max_message_size_ > 0) {
buffer->set_max_message_size(max_message_size_);
}
call_hook_->PerformOpsOnCall(buffer, this);
}

@ -158,10 +158,15 @@ bool SerializeProto(const grpc::protobuf::Message& msg, grpc_byte_buffer** bp) {
return msg.SerializeToZeroCopyStream(&writer);
}
bool DeserializeProto(grpc_byte_buffer* buffer, grpc::protobuf::Message* msg) {
bool DeserializeProto(grpc_byte_buffer* buffer, grpc::protobuf::Message* msg,
int max_message_size) {
if (!buffer) return false;
GrpcBufferReader reader(buffer);
return msg->ParseFromZeroCopyStream(&reader);
::grpc::protobuf::io::CodedInputStream decoder(&reader);
if (max_message_size > 0) {
decoder.SetTotalBytesLimit(max_message_size, max_message_size);
}
return msg->ParseFromCodedStream(&decoder) && decoder.ConsumedEntireMessage();
}
} // namespace grpc

@ -47,7 +47,8 @@ bool SerializeProto(const grpc::protobuf::Message& msg,
grpc_byte_buffer** buffer);
// The caller keeps ownership of buffer and msg.
bool DeserializeProto(grpc_byte_buffer* buffer, grpc::protobuf::Message* msg);
bool DeserializeProto(grpc_byte_buffer* buffer, grpc::protobuf::Message* msg,
int max_message_size);
} // namespace grpc

@ -100,7 +100,7 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
public:
explicit CallData(Server* server, SyncRequest* mrd)
: cq_(mrd->cq_),
call_(mrd->call_, server, &cq_),
call_(mrd->call_, server, &cq_, server->max_message_size_),
ctx_(mrd->deadline_, mrd->request_metadata_.metadata,
mrd->request_metadata_.count),
has_request_payload_(mrd->has_request_payload_),
@ -126,8 +126,11 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
if (has_request_payload_) {
GRPC_TIMER_BEGIN(GRPC_PTAG_PROTO_DESERIALIZE, call_.call());
req.reset(method_->AllocateRequestProto());
if (!DeserializeProto(request_payload_, req.get())) {
abort(); // for now
if (!DeserializeProto(request_payload_, req.get(),
call_.max_message_size())) {
// FIXME(yangg) deal with deserialization failure
cq_.Shutdown();
return;
}
GRPC_TIMER_END(GRPC_PTAG_PROTO_DESERIALIZE, call_.call());
}
@ -176,12 +179,27 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
grpc_completion_queue* cq_;
};
Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned)
: started_(false),
grpc_server* CreateServer(grpc_completion_queue* cq, int max_message_size) {
if (max_message_size > 0) {
grpc_arg arg;
arg.type = GRPC_ARG_INTEGER;
arg.key = const_cast<char*>(GRPC_ARG_MAX_MESSAGE_LENGTH);
arg.value.integer = max_message_size;
grpc_channel_args args = {1, &arg};
return grpc_server_create(cq, &args);
} else {
return grpc_server_create(cq, nullptr);
}
}
Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
int max_message_size)
: max_message_size_(max_message_size),
started_(false),
shutdown_(false),
num_running_cb_(0),
sync_methods_(new std::list<SyncRequest>),
server_(grpc_server_create(cq_.cq(), nullptr)),
server_(CreateServer(cq_.cq(), max_message_size)),
thread_pool_(thread_pool),
thread_pool_owned_(thread_pool_owned) {}
@ -220,7 +238,7 @@ bool Server::RegisterAsyncService(AsynchronousService* service) {
GPR_ASSERT(service->dispatch_impl_ == nullptr &&
"Can only register an asynchronous service against one server.");
service->dispatch_impl_ = this;
service->request_args_ = new void* [service->method_count_];
service->request_args_ = new void*[service->method_count_];
for (size_t i = 0; i < service->method_count_; ++i) {
void* tag =
grpc_server_register_method(server_, service->method_names_[i], nullptr,
@ -347,7 +365,8 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
if (*status && request_) {
if (payload_) {
GRPC_TIMER_BEGIN(GRPC_PTAG_PROTO_DESERIALIZE, call_);
*status = DeserializeProto(payload_, request_);
*status =
DeserializeProto(payload_, request_, server_->max_message_size_);
GRPC_TIMER_END(GRPC_PTAG_PROTO_DESERIALIZE, call_);
} else {
*status = false;
@ -374,7 +393,7 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
}
ctx->call_ = call_;
ctx->cq_ = cq_;
Call call(call_, server_, cq_);
Call call(call_, server_, cq_, server_->max_message_size_);
if (orig_status && call_) {
ctx->BeginCompletionOp(&call);
}

@ -42,7 +42,7 @@
namespace grpc {
ServerBuilder::ServerBuilder()
: generic_service_(nullptr), thread_pool_(nullptr) {}
: max_message_size_(-1), generic_service_(nullptr), thread_pool_(nullptr) {}
void ServerBuilder::RegisterService(SynchronousService* service) {
services_.push_back(service->service());
@ -86,7 +86,8 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
thread_pool_ = new ThreadPool(cores);
thread_pool_owned = true;
}
std::unique_ptr<Server> server(new Server(thread_pool_, thread_pool_owned));
std::unique_ptr<Server> server(
new Server(thread_pool_, thread_pool_owned, max_message_size_));
for (auto service = services_.begin(); service != services_.end();
service++) {
if (!server->RegisterService(*service)) {

@ -44,14 +44,14 @@
replacement for its descriptive functionality until Python can move its whole
C and C adapter stack to more closely resemble the core batching API. */
typedef enum {
PYGRPC_SERVER_RPC_NEW = 0,
PYGRPC_INITIAL_METADATA = 1,
PYGRPC_READ = 2,
PYGRPC_WRITE_ACCEPTED = 3,
PYGRPC_FINISH_ACCEPTED = 4,
PYGRPC_SERVER_RPC_NEW = 0,
PYGRPC_INITIAL_METADATA = 1,
PYGRPC_READ = 2,
PYGRPC_WRITE_ACCEPTED = 3,
PYGRPC_FINISH_ACCEPTED = 4,
PYGRPC_CLIENT_METADATA_READ = 5,
PYGRPC_FINISHED_CLIENT = 6,
PYGRPC_FINISHED_SERVER = 7
PYGRPC_FINISHED_CLIENT = 6,
PYGRPC_FINISHED_SERVER = 7
} pygrpc_tag_type;
typedef struct {

@ -37,6 +37,4 @@
#error "_SYS_SDT_H not defined, despite <sys/sdt.h> being present."
#endif
int main() {
return 0;
}
int main() { return 0; }

@ -62,6 +62,7 @@ END2END_TESTS = {
'graceful_server_shutdown': True,
'invoke_large_request': False,
'max_concurrent_streams': True,
'max_message_length': True,
'no_op': True,
'ping_pong_streaming': True,
'request_response_with_binary_metadata_and_payload': True,

@ -0,0 +1,210 @@
/*
*
* 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.
*
*/
#include "test/core/end2end/end2end_tests.h"
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <grpc/byte_buffer.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include <grpc/support/useful.h>
#include "test/core/end2end/cq_verifier.h"
enum { TIMEOUT = 200000 };
static void *tag(gpr_intptr t) { return (void *)t; }
static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
const char *test_name,
grpc_channel_args *client_args,
grpc_channel_args *server_args) {
grpc_end2end_test_fixture f;
gpr_log(GPR_INFO, "%s/%s", test_name, config.name);
f = config.create_fixture(client_args, server_args);
config.init_client(&f, client_args);
config.init_server(&f, server_args);
return f;
}
static gpr_timespec n_seconds_time(int n) {
return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n);
}
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
grpc_event *ev;
grpc_completion_type type;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
GPR_ASSERT(ev);
type = ev->type;
grpc_event_finish(ev);
} while (type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return;
grpc_server_shutdown(f->server);
grpc_server_destroy(f->server);
f->server = NULL;
}
static void shutdown_client(grpc_end2end_test_fixture *f) {
if (!f->client) return;
grpc_channel_destroy(f->client);
f->client = NULL;
}
static void end_test(grpc_end2end_test_fixture *f) {
shutdown_server(f);
shutdown_client(f);
grpc_completion_queue_shutdown(f->server_cq);
drain_cq(f->server_cq);
grpc_completion_queue_destroy(f->server_cq);
grpc_completion_queue_shutdown(f->client_cq);
drain_cq(f->client_cq);
grpc_completion_queue_destroy(f->client_cq);
}
static void test_max_message_length(grpc_end2end_test_config config) {
grpc_end2end_test_fixture f;
grpc_arg server_arg;
grpc_channel_args server_args;
grpc_call *c;
grpc_call *s;
cq_verifier *v_client;
cq_verifier *v_server;
grpc_op ops[6];
grpc_op *op;
gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
grpc_byte_buffer *request_payload =
grpc_byte_buffer_create(&request_payload_slice, 1);
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
grpc_call_details call_details;
grpc_status_code status;
char *details = NULL;
size_t details_capacity = 0;
int was_cancelled = 2;
server_arg.key = GRPC_ARG_MAX_MESSAGE_LENGTH;
server_arg.type = GRPC_ARG_INTEGER;
server_arg.value.integer = 5;
server_args.num_args = 1;
server_args.args = &server_arg;
f = begin_test(config, __FUNCTION__, NULL, &server_args);
v_client = cq_verifier_create(f.client_cq);
v_server = cq_verifier_create(f.server_cq);
c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
"foo.test.google.fr:1234", gpr_inf_future);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
grpc_metadata_array_init(&trailing_metadata_recv);
grpc_metadata_array_init(&request_metadata_recv);
grpc_call_details_init(&call_details);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op++;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message = request_payload;
op++;
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
op++;
op->op = GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata = &initial_metadata_recv;
op++;
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
op->data.recv_status_on_client.status = &status;
op->data.recv_status_on_client.status_details = &details;
op->data.recv_status_on_client.status_details_capacity = &details_capacity;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
op = ops;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
cq_expect_completion(v_server, tag(102), GRPC_OP_OK);
cq_verify(v_server);
cq_expect_completion(v_client, tag(1), GRPC_OP_OK);
cq_verify(v_client);
GPR_ASSERT(status == GRPC_STATUS_CANCELLED);
GPR_ASSERT(0 == strcmp(details, "Cancelled"));
GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234"));
GPR_ASSERT(was_cancelled == 1);
gpr_free(details);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
grpc_call_destroy(c);
grpc_call_destroy(s);
cq_verifier_destroy(v_client);
cq_verifier_destroy(v_server);
end_test(&f);
config.tear_down_data(&f);
}
void grpc_end2end_tests(grpc_end2end_test_config config) {
test_max_message_length(config);
}

@ -172,7 +172,7 @@ class TestServiceImplDupPkg
class End2endTest : public ::testing::Test {
protected:
End2endTest() : thread_pool_(2) {}
End2endTest() : kMaxMessageSize_(8192), thread_pool_(2) {}
void SetUp() GRPC_OVERRIDE {
int port = grpc_pick_unused_port_or_die();
@ -182,6 +182,8 @@ class End2endTest : public ::testing::Test {
builder.AddListeningPort(server_address_.str(),
InsecureServerCredentials());
builder.RegisterService(&service_);
builder.SetMaxMessageSize(
kMaxMessageSize_); // For testing max message size.
builder.RegisterService(&dup_pkg_service_);
builder.SetThreadPool(&thread_pool_);
server_ = builder.BuildAndStart();
@ -198,6 +200,7 @@ class End2endTest : public ::testing::Test {
std::unique_ptr<grpc::cpp::test::util::TestService::Stub> stub_;
std::unique_ptr<Server> server_;
std::ostringstream server_address_;
const int kMaxMessageSize_;
TestServiceImpl service_;
TestServiceImplDupPkg dup_pkg_service_;
ThreadPool thread_pool_;
@ -426,8 +429,7 @@ TEST_F(End2endTest, DiffPackageServices) {
// rpc and stream should fail on bad credentials.
TEST_F(End2endTest, BadCredentials) {
std::unique_ptr<Credentials> bad_creds =
ServiceAccountCredentials("", "", 1);
std::unique_ptr<Credentials> bad_creds = ServiceAccountCredentials("", "", 1);
EXPECT_EQ(nullptr, bad_creds.get());
std::shared_ptr<ChannelInterface> channel =
CreateChannel(server_address_.str(), bad_creds, ChannelArguments());
@ -501,14 +503,13 @@ TEST_F(End2endTest, ClientCancelsRequestStream) {
auto stream = stub_->RequestStream(&context, &response);
EXPECT_TRUE(stream->Write(request));
EXPECT_TRUE(stream->Write(request));
context.TryCancel();
Status s = stream->Finish();
EXPECT_EQ(grpc::StatusCode::CANCELLED, s.code());
EXPECT_EQ(response.message(), "");
EXPECT_EQ(response.message(), "");
}
// Client cancels server stream after sending some messages
@ -588,6 +589,17 @@ TEST_F(End2endTest, ThreadStress) {
}
}
TEST_F(End2endTest, RpcMaxMessageSize) {
ResetStub();
EchoRequest request;
EchoResponse response;
request.set_message(string(kMaxMessageSize_ * 2, 'a'));
ClientContext context;
Status s = stub_->Echo(&context, request, &response);
EXPECT_FALSE(s.IsOk());
}
} // namespace testing
} // namespace grpc

@ -810,6 +810,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fake_security_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1035,6 +1044,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1260,6 +1278,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_uds_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1485,6 +1512,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_simple_ssl_fullstack_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1710,6 +1746,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_simple_ssl_with_oauth2_fullstack_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1935,6 +1980,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2160,6 +2214,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_one_byte_at_a_time_max_message_length_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2385,6 +2448,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_max_message_length_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2610,6 +2682,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_uds_max_message_length_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2835,6 +2916,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_max_message_length_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -3060,6 +3150,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_one_byte_at_a_time_max_message_length_unsecure_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",

File diff suppressed because one or more lines are too long
Loading…
Cancel
Save