From 25d02d56370bfa0ead2a79203c522ce02cbe9394 Mon Sep 17 00:00:00 2001 From: David Garcia Quintas Date: Thu, 4 Jun 2015 17:40:54 -0700 Subject: [PATCH] Made grpc_byte_buffer_reader able to decompress msgs. --- include/grpc/byte_buffer.h | 61 ++++++++++++++++- include/grpc/byte_buffer_reader.h | 3 +- include/grpc/grpc.h | 29 +------- src/core/surface/byte_buffer.c | 18 ++++- src/core/surface/byte_buffer_reader.c | 59 ++++++++++++---- src/core/surface/call.c | 3 + test/core/surface/byte_buffer_reader_test.c | 75 +++++++++++++++++++++ 7 files changed, 203 insertions(+), 45 deletions(-) diff --git a/include/grpc/byte_buffer.h b/include/grpc/byte_buffer.h index 0ff494cdec9..eb9608a14e4 100644 --- a/include/grpc/byte_buffer.h +++ b/include/grpc/byte_buffer.h @@ -34,10 +34,17 @@ #ifndef GRPC_BYTE_BUFFER_H #define GRPC_BYTE_BUFFER_H -#include #include -typedef enum { GRPC_BB_SLICE_BUFFER } grpc_byte_buffer_type; +typedef enum { + GRPC_BB_SLICE_BUFFER, + + /* Keep the GRPC_BB_COMPRESSED_* entries in the same order as the + * grpc_compression_algorithm enum entries. */ + GRPC_BB_COMPRESSED_NONE, /* for overriding otherwise compressed channels */ + GRPC_BB_COMPRESSED_DEFLATE, + GRPC_BB_COMPRESSED_GZIP +} grpc_byte_buffer_type; /* byte buffers are containers for messages passed in from the public api's */ struct grpc_byte_buffer { @@ -46,5 +53,55 @@ struct grpc_byte_buffer { gpr_slice_buffer slice_buffer; } data; }; +typedef struct grpc_byte_buffer grpc_byte_buffer; + +/** Returns the grpc_compression_algorithm enum value for a + * grpc_byte_buffer_type GRPC_BB_COMPRESSED_* value. */ +#define GRPC_COMPRESS_ALGORITHM_FROM_BB_TYPE(bb_type) \ + (bb_type - GRPC_BB_COMPRESSED_NONE) + +/** Returns a byte buffer instance over the given slices (up to \a nslices) of + * \a type type. + * + * Increases the reference count for all \a slices processed. + * + * The user is responsible for invoking grpc_byte_buffer_destroy on + * the returned instance. */ +grpc_byte_buffer *grpc_byte_buffer_typed_create(gpr_slice *slices, + size_t nslices, + grpc_byte_buffer_type type); +/** Convenience method for creating GRPC_BB_SLICE_BUFFER byte buffers. \sa + * grpc_byte_buffer_typed_create */ +grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices); + +/** Copies input byte buffer \a bb. + * + * Increases the reference count of all the source slices. The user is + * responsible for calling grpc_byte_buffer_destroy over the returned copy. */ +grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb); + +/** Returns the size of the given byte buffer, in bytes. */ +size_t grpc_byte_buffer_length(grpc_byte_buffer *bb); + +/** Destroys \a byte_buffer deallocating all its memory. */ +void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer); + + +/** Reader for byte buffers. Iterates over slices in the 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 */ +void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader, + grpc_byte_buffer *buffer); + +/** Cleanup and destroy \a reader */ +void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader); + +/** Updates \a slice with the next piece of data from from \a reader and returns + * 1. Returns 0 at the end of the stream. Caller is responsible for calling + * gpr_slice_unref on the result. */ +int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader, + gpr_slice *slice); #endif /* GRPC_BYTE_BUFFER_H */ diff --git a/include/grpc/byte_buffer_reader.h b/include/grpc/byte_buffer_reader.h index cb757cf6420..3f186da4f53 100644 --- a/include/grpc/byte_buffer_reader.h +++ b/include/grpc/byte_buffer_reader.h @@ -38,7 +38,8 @@ #include struct grpc_byte_buffer_reader { - grpc_byte_buffer *buffer; + grpc_byte_buffer *buffer_in; + grpc_byte_buffer *buffer_out; /* Different current objects correspond to different types of byte buffers */ union { /* Index into a slice buffer's array of slices */ diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h index 57bf2fad5ab..25c59501582 100644 --- a/include/grpc/grpc.h +++ b/include/grpc/grpc.h @@ -37,6 +37,7 @@ #include #include +#include #include #include @@ -155,34 +156,6 @@ typedef enum grpc_call_error { (start_write/add_metadata). Illegal on invoke/accept. */ #define GRPC_WRITE_NO_COMPRESS (0x00000002u) -/* A buffer of bytes */ -struct grpc_byte_buffer; -typedef struct grpc_byte_buffer grpc_byte_buffer; - -/* Sample helpers to obtain byte buffers (these will certainly move - someplace else) */ -grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices); -grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb); -size_t grpc_byte_buffer_length(grpc_byte_buffer *bb); -void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer); - -/* Reader for byte buffers. Iterates over slices in the 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 */ -void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader, - grpc_byte_buffer *buffer); - -/** Cleanup and destroy \a reader */ -void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader); - -/* At the end of the stream, returns 0. Otherwise, returns 1 and sets slice to - be the returned slice. Caller is responsible for calling gpr_slice_unref on - the result. */ -int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader, - gpr_slice *slice); - /* A single metadata element */ typedef struct grpc_metadata { const char *key; diff --git a/src/core/surface/byte_buffer.c b/src/core/surface/byte_buffer.c index 12244f6644e..184c3da26e8 100644 --- a/src/core/surface/byte_buffer.c +++ b/src/core/surface/byte_buffer.c @@ -36,10 +36,16 @@ #include grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices) { + return grpc_byte_buffer_typed_create(slices, nslices, GRPC_BB_SLICE_BUFFER); +} + +grpc_byte_buffer *grpc_byte_buffer_typed_create(gpr_slice *slices, + size_t nslices, + grpc_byte_buffer_type type) { size_t i; grpc_byte_buffer *bb = malloc(sizeof(grpc_byte_buffer)); - bb->type = GRPC_BB_SLICE_BUFFER; + bb->type = type; gpr_slice_buffer_init(&bb->data.slice_buffer); for (i = 0; i < nslices; i++) { gpr_slice_ref(slices[i]); @@ -49,9 +55,13 @@ grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices) { return bb; } + grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb) { switch (bb->type) { case GRPC_BB_SLICE_BUFFER: + case GRPC_BB_COMPRESSED_NONE: + case GRPC_BB_COMPRESSED_DEFLATE: + case GRPC_BB_COMPRESSED_GZIP: return grpc_byte_buffer_create(bb->data.slice_buffer.slices, bb->data.slice_buffer.count); } @@ -64,6 +74,9 @@ void grpc_byte_buffer_destroy(grpc_byte_buffer *bb) { if (!bb) return; switch (bb->type) { case GRPC_BB_SLICE_BUFFER: + case GRPC_BB_COMPRESSED_NONE: + case GRPC_BB_COMPRESSED_DEFLATE: + case GRPC_BB_COMPRESSED_GZIP: gpr_slice_buffer_destroy(&bb->data.slice_buffer); break; } @@ -73,6 +86,9 @@ void grpc_byte_buffer_destroy(grpc_byte_buffer *bb) { size_t grpc_byte_buffer_length(grpc_byte_buffer *bb) { switch (bb->type) { case GRPC_BB_SLICE_BUFFER: + case GRPC_BB_COMPRESSED_NONE: + case GRPC_BB_COMPRESSED_DEFLATE: + case GRPC_BB_COMPRESSED_GZIP: return bb->data.slice_buffer.length; } gpr_log(GPR_ERROR, "should never reach here"); diff --git a/src/core/surface/byte_buffer_reader.c b/src/core/surface/byte_buffer_reader.c index 41ad7002748..0754a172263 100644 --- a/src/core/surface/byte_buffer_reader.c +++ b/src/core/surface/byte_buffer_reader.c @@ -39,35 +39,68 @@ #include #include +#include "src/core/compression/algorithm.h" +#include "src/core/compression/message_compress.h" + void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader, grpc_byte_buffer *buffer) { - reader->buffer = buffer; + grpc_compression_algorithm compress_algo; + gpr_slice_buffer decompressed_slices_buffer; + reader->buffer_in = buffer; switch (buffer->type) { + case GRPC_BB_COMPRESSED_DEFLATE: + case GRPC_BB_COMPRESSED_GZIP: + compress_algo = + GRPC_COMPRESS_ALGORITHM_FROM_BB_TYPE(reader->buffer_in->type); + gpr_slice_buffer_init(&decompressed_slices_buffer); + grpc_msg_decompress(compress_algo, &reader->buffer_in->data.slice_buffer, + &decompressed_slices_buffer); + /* the output buffer is a regular GRPC_BB_SLICE_BUFFER */ + reader->buffer_out = grpc_byte_buffer_create( + decompressed_slices_buffer.slices, + decompressed_slices_buffer.count); + gpr_slice_buffer_destroy(&decompressed_slices_buffer); + /* fallthrough */ case GRPC_BB_SLICE_BUFFER: + case GRPC_BB_COMPRESSED_NONE: reader->current.index = 0; } } void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader) { - /* no-op: the user is responsible for memory deallocation. - * Other cleanup operations would go here if needed. */ + switch (reader->buffer_in->type) { + case GRPC_BB_COMPRESSED_DEFLATE: + case GRPC_BB_COMPRESSED_GZIP: + grpc_byte_buffer_destroy(reader->buffer_out); + break; + case GRPC_BB_SLICE_BUFFER: + case GRPC_BB_COMPRESSED_NONE: + ; /* no-op */ + } } int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader, gpr_slice *slice) { - grpc_byte_buffer *buffer = reader->buffer; gpr_slice_buffer *slice_buffer; - switch (buffer->type) { + grpc_byte_buffer *buffer = NULL; + + /* Pick the right buffer based on the input type */ + switch (reader->buffer_in->type) { case GRPC_BB_SLICE_BUFFER: - slice_buffer = &buffer->data.slice_buffer; - if (reader->current.index < slice_buffer->count) { - *slice = gpr_slice_ref(slice_buffer->slices[reader->current.index]); - reader->current.index += 1; - return 1; - } else { - return 0; - } + case GRPC_BB_COMPRESSED_NONE: + buffer = reader->buffer_in; break; + case GRPC_BB_COMPRESSED_DEFLATE: + case GRPC_BB_COMPRESSED_GZIP: + buffer = reader->buffer_out; + break; + } + GPR_ASSERT(buffer); + slice_buffer = &buffer->data.slice_buffer; + if (reader->current.index < slice_buffer->count) { + *slice = gpr_slice_ref(slice_buffer->slices[reader->current.index]); + reader->current.index += 1; + return 1; } return 0; } diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 88ff5cfbce3..8810dd3d513 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -806,6 +806,9 @@ static void copy_byte_buffer_to_stream_ops(grpc_byte_buffer *byte_buffer, switch (byte_buffer->type) { case GRPC_BB_SLICE_BUFFER: + case GRPC_BB_COMPRESSED_NONE: + case GRPC_BB_COMPRESSED_DEFLATE: + case GRPC_BB_COMPRESSED_GZIP: for (i = 0; i < byte_buffer->data.slice_buffer.count; i++) { gpr_slice slice = byte_buffer->data.slice_buffer.slices[i]; gpr_slice_ref(slice); diff --git a/test/core/surface/byte_buffer_reader_test.c b/test/core/surface/byte_buffer_reader_test.c index c2f5fc2eb74..f594bdcd627 100644 --- a/test/core/surface/byte_buffer_reader_test.c +++ b/test/core/surface/byte_buffer_reader_test.c @@ -42,6 +42,8 @@ #include #include "test/core/util/test_config.h" +#include "src/core/compression/message_compress.h" + #include #define LOG_TEST(x) gpr_log(GPR_INFO, "%s", x) @@ -89,9 +91,82 @@ static void test_read_one_slice_malloc(void) { grpc_byte_buffer_destroy(buffer); } +static void test_read_none_compressed_slice(void) { + gpr_slice slice; + grpc_byte_buffer *buffer; + grpc_byte_buffer_reader reader; + gpr_slice first_slice, second_slice; + int first_code, second_code; + + LOG_TEST("test_read_none_compressed_slice"); + slice = gpr_slice_from_copied_string("test"); + buffer = grpc_byte_buffer_typed_create(&slice, 1, GRPC_BB_COMPRESSED_NONE); + gpr_slice_unref(slice); + grpc_byte_buffer_reader_init(&reader, buffer); + 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); + gpr_slice_unref(first_slice); + second_code = grpc_byte_buffer_reader_next(&reader, &second_slice); + GPR_ASSERT(second_code == 0); + grpc_byte_buffer_destroy(buffer); +} + +static void read_compressed_slice(grpc_compression_algorithm algorithm, + int input_size) { + gpr_slice input_slice; + gpr_slice_buffer sliceb_in; + gpr_slice_buffer sliceb_out; + grpc_byte_buffer *buffer; + grpc_byte_buffer_reader reader; + gpr_slice read_slice; + int read_count = 0; + + gpr_slice_buffer_init(&sliceb_in); + gpr_slice_buffer_init(&sliceb_out); + + input_slice = gpr_slice_malloc(input_size); + memset(GPR_SLICE_START_PTR(input_slice), 'a', input_size); + gpr_slice_buffer_add(&sliceb_in, input_slice); /* takes ownership */ + GPR_ASSERT(grpc_msg_compress(algorithm, &sliceb_in, &sliceb_out)); + + buffer = grpc_byte_buffer_typed_create(sliceb_out.slices, sliceb_out.count, + GRPC_BB_COMPRESSED_NONE + algorithm); + grpc_byte_buffer_reader_init(&reader, buffer); + + while (grpc_byte_buffer_reader_next(&reader, &read_slice)) { + GPR_ASSERT(memcmp(GPR_SLICE_START_PTR(read_slice), + GPR_SLICE_START_PTR(input_slice) + read_count, + GPR_SLICE_LENGTH(read_slice)) == 0); + read_count += GPR_SLICE_LENGTH(read_slice); + gpr_slice_unref(read_slice); + } + GPR_ASSERT(read_count == input_size); + grpc_byte_buffer_reader_destroy(&reader); + grpc_byte_buffer_destroy(buffer); + gpr_slice_buffer_destroy(&sliceb_out); + gpr_slice_buffer_destroy(&sliceb_in); +} + +static void test_read_gzip_compressed_slice(void) { + const int INPUT_SIZE = 2048; + LOG_TEST("test_read_gzip_compressed_slice"); + read_compressed_slice(GRPC_COMPRESS_GZIP, INPUT_SIZE); +} + +static void test_read_deflate_compressed_slice(void) { + const int INPUT_SIZE = 2048; + LOG_TEST("test_read_deflate_compressed_slice"); + read_compressed_slice(GRPC_COMPRESS_DEFLATE, INPUT_SIZE); +} + int main(int argc, char **argv) { grpc_test_init(argc, argv); test_read_one_slice(); test_read_one_slice_malloc(); + test_read_none_compressed_slice(); + test_read_gzip_compressed_slice(); + test_read_deflate_compressed_slice(); + return 0; }