Merge pull request #7190 from dgquintas/bb_reader_error

Return success status of grpc_byte_buffer_reader_init
pull/7268/head^2
Nicolas Noble 9 years ago committed by GitHub
commit e705dc44a1
  1. 4
      include/grpc++/impl/codegen/core_codegen.h
  2. 5
      include/grpc++/impl/codegen/core_codegen_interface.h
  3. 17
      include/grpc++/impl/codegen/proto_utils.h
  4. 2
      include/grpc++/support/byte_buffer.h
  5. 9
      include/grpc/impl/codegen/byte_buffer.h
  6. 10
      src/core/lib/surface/byte_buffer_reader.c
  7. 6
      src/cpp/common/core_codegen.cc
  8. 10
      src/cpp/util/byte_buffer.cc
  9. 2
      src/csharp/ext/grpc_csharp_ext.c
  10. 5
      src/node/ext/byte_buffer.cc
  11. 10
      src/objective-c/GRPCClient/private/NSData+GRPC.m
  12. 6
      src/php/ext/grpc/byte_buffer.c
  13. 4
      src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi
  14. 6
      src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi
  15. 2
      src/python/grpcio/grpc/_cython/imports.generated.h
  16. 5
      src/ruby/ext/grpc/rb_byte_buffer.c
  17. 2
      src/ruby/ext/grpc/rb_grpc_imports.generated.h
  18. 3
      test/core/end2end/cq_verifier.c
  19. 36
      test/core/surface/byte_buffer_reader_test.c

@ -54,8 +54,8 @@ class CoreCodegen : public CoreCodegenInterface {
void grpc_byte_buffer_destroy(grpc_byte_buffer* bb) GRPC_OVERRIDE;
void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader* reader,
grpc_byte_buffer* buffer) GRPC_OVERRIDE;
int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader* reader,
grpc_byte_buffer* buffer) GRPC_OVERRIDE;
void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader* reader)
GRPC_OVERRIDE;
int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader* reader,

@ -65,8 +65,9 @@ class CoreCodegenInterface {
virtual void grpc_byte_buffer_destroy(grpc_byte_buffer* bb) = 0;
virtual void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader* reader,
grpc_byte_buffer* buffer) = 0;
virtual int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader* reader,
grpc_byte_buffer* buffer)
GRPC_MUST_USE_RESULT = 0;
virtual void grpc_byte_buffer_reader_destroy(
grpc_byte_buffer_reader* reader) = 0;
virtual int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader* reader,

@ -111,14 +111,21 @@ class GrpcBufferReader GRPC_FINAL
: public ::grpc::protobuf::io::ZeroCopyInputStream {
public:
explicit GrpcBufferReader(grpc_byte_buffer* buffer)
: byte_count_(0), backup_count_(0) {
g_core_codegen_interface->grpc_byte_buffer_reader_init(&reader_, buffer);
: byte_count_(0), backup_count_(0), status_() {
if (!g_core_codegen_interface->grpc_byte_buffer_reader_init(&reader_,
buffer)) {
status_ = Status(StatusCode::INTERNAL,
"Couldn't initialize byte buffer reader");
}
}
~GrpcBufferReader() GRPC_OVERRIDE {
g_core_codegen_interface->grpc_byte_buffer_reader_destroy(&reader_);
}
bool Next(const void** data, int* size) GRPC_OVERRIDE {
if (!status_.ok()) {
return false;
}
if (backup_count_ > 0) {
*data = GPR_SLICE_START_PTR(slice_) + GPR_SLICE_LENGTH(slice_) -
backup_count_;
@ -139,6 +146,8 @@ class GrpcBufferReader GRPC_FINAL
return true;
}
Status status() const { return status_; }
void BackUp(int count) GRPC_OVERRIDE { backup_count_ = count; }
bool Skip(int count) GRPC_OVERRIDE {
@ -165,6 +174,7 @@ class GrpcBufferReader GRPC_FINAL
int64_t backup_count_;
grpc_byte_buffer_reader reader_;
gpr_slice slice_;
Status status_;
};
} // namespace internal
@ -202,6 +212,9 @@ class SerializationTraits<T, typename std::enable_if<std::is_base_of<
Status result = g_core_codegen_interface->ok();
{
internal::GrpcBufferReader reader(buffer);
if (!reader.status().ok()) {
return reader.status();
}
::grpc::protobuf::io::CodedInputStream decoder(&reader);
if (max_message_size > 0) {
decoder.SetTotalBytesLimit(max_message_size, max_message_size);

@ -64,7 +64,7 @@ class ByteBuffer GRPC_FINAL {
ByteBuffer& operator=(const ByteBuffer&);
/// Dump (read) the buffer contents into \a slices.
void Dump(std::vector<Slice>* slices) const;
Status Dump(std::vector<Slice>* slices) const;
/// Remove all data.
void Clear();

@ -1,6 +1,6 @@
/*
*
* Copyright 2015, Google Inc.
* Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@ -93,9 +93,10 @@ GRPCAPI void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer);
struct grpc_byte_buffer_reader;
typedef struct grpc_byte_buffer_reader grpc_byte_buffer_reader;
/** Initialize \a reader to read over \a buffer */
GRPCAPI void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
grpc_byte_buffer *buffer);
/** Initialize \a reader to read over \a buffer.
* Returns 1 upon success, 0 otherwise. */
GRPCAPI int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
grpc_byte_buffer *buffer);
/** Cleanup and destroy \a reader */
GRPCAPI void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader);

@ -54,8 +54,8 @@ static int is_compressed(grpc_byte_buffer *buffer) {
return 1 /* GPR_TRUE */;
}
void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
grpc_byte_buffer *buffer) {
int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
grpc_byte_buffer *buffer) {
gpr_slice_buffer decompressed_slices_buffer;
reader->buffer_in = buffer;
switch (reader->buffer_in->type) {
@ -67,9 +67,10 @@ void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
&decompressed_slices_buffer) == 0) {
gpr_log(GPR_ERROR,
"Unexpected error decompressing data for algorithm with enum "
"value '%d'. Reading data as if it were uncompressed.",
"value '%d'.",
reader->buffer_in->data.raw.compression);
reader->buffer_out = reader->buffer_in;
memset(reader, 0, sizeof(*reader));
return 0;
} else { /* all fine */
reader->buffer_out =
grpc_raw_byte_buffer_create(decompressed_slices_buffer.slices,
@ -82,6 +83,7 @@ void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
reader->current.index = 0;
break;
}
return 1;
}
void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader) {

@ -74,9 +74,9 @@ void CoreCodegen::grpc_byte_buffer_destroy(grpc_byte_buffer* bb) {
::grpc_byte_buffer_destroy(bb);
}
void CoreCodegen::grpc_byte_buffer_reader_init(grpc_byte_buffer_reader* reader,
grpc_byte_buffer* buffer) {
::grpc_byte_buffer_reader_init(reader, buffer);
int CoreCodegen::grpc_byte_buffer_reader_init(grpc_byte_buffer_reader* reader,
grpc_byte_buffer* buffer) {
return ::grpc_byte_buffer_reader_init(reader, buffer);
}
void CoreCodegen::grpc_byte_buffer_reader_destroy(

@ -58,18 +58,22 @@ void ByteBuffer::Clear() {
}
}
void ByteBuffer::Dump(std::vector<Slice>* slices) const {
Status ByteBuffer::Dump(std::vector<Slice>* slices) const {
slices->clear();
if (!buffer_) {
return;
return Status(StatusCode::FAILED_PRECONDITION, "Buffer not initialized");
}
grpc_byte_buffer_reader reader;
grpc_byte_buffer_reader_init(&reader, buffer_);
if (!grpc_byte_buffer_reader_init(&reader, buffer_)) {
return Status(StatusCode::INTERNAL,
"Couldn't initialize byte buffer reader");
}
gpr_slice s;
while (grpc_byte_buffer_reader_next(&reader, &s)) {
slices->push_back(Slice(s, Slice::STEAL_REF));
}
grpc_byte_buffer_reader_destroy(&reader);
return Status::OK;
}
size_t ByteBuffer::Length() const {

@ -253,6 +253,7 @@ GPR_EXPORT intptr_t GPR_CALLTYPE grpcsharp_batch_context_recv_message_length(
if (!ctx->recv_message) {
return -1;
}
/* TODO(issue:#7206): check return value of grpc_byte_buffer_reader_init. */
grpc_byte_buffer_reader_init(&reader, ctx->recv_message);
return (intptr_t)grpc_byte_buffer_length(reader.buffer_out);
}
@ -267,6 +268,7 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_batch_context_recv_message_to_buffer(
gpr_slice slice;
size_t offset = 0;
/* TODO(issue:#7206): check return value of grpc_byte_buffer_reader_init. */
grpc_byte_buffer_reader_init(&reader, ctx->recv_message);
while (grpc_byte_buffer_reader_next(&reader, &slice)) {

@ -73,7 +73,10 @@ Local<Value> ByteBufferToBuffer(grpc_byte_buffer *buffer) {
return scope.Escape(Nan::Null());
}
grpc_byte_buffer_reader reader;
grpc_byte_buffer_reader_init(&reader, buffer);
if (!grpc_byte_buffer_reader_init(&reader, buffer)) {
Nan::ThrowError("Error initializing byte buffer reader.");
return scope.Escape(Nan::Undefined());
}
gpr_slice slice = grpc_byte_buffer_reader_readall(&reader);
size_t length = GPR_SLICE_LENGTH(slice);
char *result = new char[length];

@ -42,7 +42,15 @@
static void MallocAndCopyByteBufferToCharArray(grpc_byte_buffer *buffer,
size_t *length, char **array) {
grpc_byte_buffer_reader reader;
grpc_byte_buffer_reader_init(&reader, buffer);
if (!grpc_byte_buffer_reader_init(&reader, buffer)) {
// grpc_byte_buffer_reader_init can fail if the data sent by the server
// could not be decompressed for any reason. This is an issue with the data
// coming from the server and thus we want the RPC to fail with error code
// INTERNAL.
*array = NULL;
*length = 0;
return;
}
// The slice contains uncompressed data even if compressed data was received
// because the reader takes care of automatically decompressing it
gpr_slice slice = grpc_byte_buffer_reader_readall(&reader);

@ -58,7 +58,9 @@ grpc_byte_buffer *string_to_byte_buffer(char *string, size_t length) {
void byte_buffer_to_string(grpc_byte_buffer *buffer, char **out_string,
size_t *out_length) {
if (buffer == NULL) {
grpc_byte_buffer_reader reader;
if (buffer == NULL || !grpc_byte_buffer_reader_init(&reader, buffer)) {
/* TODO(dgq): distinguish between the error cases. */
*out_string = NULL;
*out_length = 0;
return;
@ -66,8 +68,6 @@ void byte_buffer_to_string(grpc_byte_buffer *buffer, char **out_string,
size_t length = grpc_byte_buffer_length(buffer);
char *string = ecalloc(length + 1, sizeof(char));
size_t offset = 0;
grpc_byte_buffer_reader reader;
grpc_byte_buffer_reader_init(&reader, buffer);
gpr_slice next;
while (grpc_byte_buffer_reader_next(&reader, &next) != 0) {
memcpy(string + offset, GPR_SLICE_START_PTR(next), GPR_SLICE_LENGTH(next));

@ -132,8 +132,8 @@ cdef extern from "grpc/_cython/loader.h":
size_t grpc_byte_buffer_length(grpc_byte_buffer *bb) nogil
void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer) nogil
void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
grpc_byte_buffer *buffer) nogil
int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
grpc_byte_buffer *buffer) nogil
int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader,
gpr_slice *slice) nogil
void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader) nogil

@ -252,9 +252,13 @@ cdef class ByteBuffer:
cdef gpr_slice data_slice
cdef size_t data_slice_length
cdef void *data_slice_pointer
cdef bint reader_status
if self.c_byte_buffer != NULL:
with nogil:
grpc_byte_buffer_reader_init(&reader, self.c_byte_buffer)
reader_status = grpc_byte_buffer_reader_init(
&reader, self.c_byte_buffer)
if not reader_status:
return None
result = bytearray()
with nogil:
while grpc_byte_buffer_reader_next(&reader, &data_slice):

@ -470,7 +470,7 @@ extern grpc_byte_buffer_length_type grpc_byte_buffer_length_import;
typedef void(*grpc_byte_buffer_destroy_type)(grpc_byte_buffer *byte_buffer);
extern grpc_byte_buffer_destroy_type grpc_byte_buffer_destroy_import;
#define grpc_byte_buffer_destroy grpc_byte_buffer_destroy_import
typedef void(*grpc_byte_buffer_reader_init_type)(grpc_byte_buffer_reader *reader, grpc_byte_buffer *buffer);
typedef int(*grpc_byte_buffer_reader_init_type)(grpc_byte_buffer_reader *reader, grpc_byte_buffer *buffer);
extern grpc_byte_buffer_reader_init_type grpc_byte_buffer_reader_init_import;
#define grpc_byte_buffer_reader_init grpc_byte_buffer_reader_init_import
typedef void(*grpc_byte_buffer_reader_destroy_type)(grpc_byte_buffer_reader *reader);

@ -56,7 +56,10 @@ VALUE grpc_rb_byte_buffer_to_s(grpc_byte_buffer *buffer) {
return Qnil;
}
rb_string = rb_str_buf_new(grpc_byte_buffer_length(buffer));
grpc_byte_buffer_reader_init(&reader, buffer);
if (!grpc_byte_buffer_reader_init(&reader, buffer)) {
rb_raise(rb_eRuntimeError, "Error initializing byte buffer reader.");
return Qnil;
}
while (grpc_byte_buffer_reader_next(&reader, &next) != 0) {
rb_str_cat(rb_string, (const char *) GPR_SLICE_START_PTR(next),
GPR_SLICE_LENGTH(next));

@ -470,7 +470,7 @@ extern grpc_byte_buffer_length_type grpc_byte_buffer_length_import;
typedef void(*grpc_byte_buffer_destroy_type)(grpc_byte_buffer *byte_buffer);
extern grpc_byte_buffer_destroy_type grpc_byte_buffer_destroy_import;
#define grpc_byte_buffer_destroy grpc_byte_buffer_destroy_import
typedef void(*grpc_byte_buffer_reader_init_type)(grpc_byte_buffer_reader *reader, grpc_byte_buffer *buffer);
typedef int(*grpc_byte_buffer_reader_init_type)(grpc_byte_buffer_reader *reader, grpc_byte_buffer *buffer);
extern grpc_byte_buffer_reader_init_type grpc_byte_buffer_reader_init_import;
#define grpc_byte_buffer_reader_init grpc_byte_buffer_reader_init_import
typedef void(*grpc_byte_buffer_reader_destroy_type)(grpc_byte_buffer_reader *reader);

@ -149,7 +149,8 @@ int byte_buffer_eq_string(grpc_byte_buffer *bb, const char *str) {
grpc_byte_buffer *rbb;
int res;
grpc_byte_buffer_reader_init(&reader, bb);
GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, bb) &&
"Couldn't init byte buffer reader");
rbb = grpc_raw_byte_buffer_from_reader(&reader);
res = byte_buffer_eq_slice(rbb, gpr_slice_from_copied_string(str));
grpc_byte_buffer_reader_destroy(&reader);

@ -59,7 +59,8 @@ static void test_read_one_slice(void) {
slice = gpr_slice_from_copied_string("test");
buffer = grpc_raw_byte_buffer_create(&slice, 1);
gpr_slice_unref(slice);
grpc_byte_buffer_reader_init(&reader, buffer);
GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, buffer) &&
"Couldn't init byte buffer reader");
first_code = grpc_byte_buffer_reader_next(&reader, &first_slice);
GPR_ASSERT(first_code != 0);
GPR_ASSERT(memcmp(GPR_SLICE_START_PTR(first_slice), "test", 4) == 0);
@ -81,7 +82,8 @@ static void test_read_one_slice_malloc(void) {
memcpy(GPR_SLICE_START_PTR(slice), "test", 4);
buffer = grpc_raw_byte_buffer_create(&slice, 1);
gpr_slice_unref(slice);
grpc_byte_buffer_reader_init(&reader, buffer);
GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, buffer) &&
"Couldn't init byte buffer reader");
first_code = grpc_byte_buffer_reader_next(&reader, &first_slice);
GPR_ASSERT(first_code != 0);
GPR_ASSERT(memcmp(GPR_SLICE_START_PTR(first_slice), "test", 4) == 0);
@ -102,7 +104,8 @@ static void test_read_none_compressed_slice(void) {
slice = gpr_slice_from_copied_string("test");
buffer = grpc_raw_byte_buffer_create(&slice, 1);
gpr_slice_unref(slice);
grpc_byte_buffer_reader_init(&reader, buffer);
GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, buffer) &&
"Couldn't init byte buffer reader");
first_code = grpc_byte_buffer_reader_next(&reader, &first_slice);
GPR_ASSERT(first_code != 0);
GPR_ASSERT(memcmp(GPR_SLICE_START_PTR(first_slice), "test", 4) == 0);
@ -112,6 +115,20 @@ static void test_read_none_compressed_slice(void) {
grpc_byte_buffer_destroy(buffer);
}
static void test_read_corrupted_slice(void) {
gpr_slice slice;
grpc_byte_buffer *buffer;
grpc_byte_buffer_reader reader;
LOG_TEST("test_read_corrupted_slice");
slice = gpr_slice_from_copied_string("test");
buffer = grpc_raw_byte_buffer_create(&slice, 1);
buffer->data.raw.compression = GRPC_COMPRESS_GZIP; /* lies! */
gpr_slice_unref(slice);
GPR_ASSERT(!grpc_byte_buffer_reader_init(&reader, buffer));
grpc_byte_buffer_destroy(buffer);
}
static void read_compressed_slice(grpc_compression_algorithm algorithm,
size_t input_size) {
gpr_slice input_slice;
@ -132,7 +149,8 @@ static void read_compressed_slice(grpc_compression_algorithm algorithm,
buffer = grpc_raw_compressed_byte_buffer_create(sliceb_out.slices,
sliceb_out.count, algorithm);
grpc_byte_buffer_reader_init(&reader, buffer);
GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, buffer) &&
"Couldn't init byte buffer reader");
while (grpc_byte_buffer_reader_next(&reader, &read_slice)) {
GPR_ASSERT(memcmp(GPR_SLICE_START_PTR(read_slice),
@ -170,7 +188,8 @@ static void test_byte_buffer_from_reader(void) {
memcpy(GPR_SLICE_START_PTR(slice), "test", 4);
buffer = grpc_raw_byte_buffer_create(&slice, 1);
gpr_slice_unref(slice);
grpc_byte_buffer_reader_init(&reader, buffer);
GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, buffer) &&
"Couldn't init byte buffer reader");
buffer_from_reader = grpc_raw_byte_buffer_from_reader(&reader);
GPR_ASSERT(buffer->type == buffer_from_reader->type);
@ -206,7 +225,8 @@ static void test_readall(void) {
gpr_slice_unref(slices[0]);
gpr_slice_unref(slices[1]);
grpc_byte_buffer_reader_init(&reader, buffer);
GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, buffer) &&
"Couldn't init byte buffer reader");
slice_out = grpc_byte_buffer_reader_readall(&reader);
GPR_ASSERT(GPR_SLICE_LENGTH(slice_out) == 512 + 1024);
@ -241,7 +261,8 @@ static void test_byte_buffer_copy(void) {
gpr_slice_unref(slices[1]);
copied_buffer = grpc_byte_buffer_copy(buffer);
grpc_byte_buffer_reader_init(&reader, copied_buffer);
GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, buffer) &&
"Couldn't init byte buffer reader");
slice_out = grpc_byte_buffer_reader_readall(&reader);
GPR_ASSERT(GPR_SLICE_LENGTH(slice_out) == 512 + 1024);
@ -260,6 +281,7 @@ int main(int argc, char **argv) {
test_read_none_compressed_slice();
test_read_gzip_compressed_slice();
test_read_deflate_compressed_slice();
test_read_corrupted_slice();
test_byte_buffer_from_reader();
test_byte_buffer_copy();
test_readall();

Loading…
Cancel
Save