• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1 /*
2  *
3  * Copyright 2017 gRPC authors.
4  *
5  * Licensed under the Apache License, Version 2.0 (the "License");
6  * you may not use this file except in compliance with the License.
7  * You may obtain a copy of the License at
8  *
9  *     http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  *
17  */
18 
19 #include <grpc/support/port_platform.h>
20 
21 #include <grpc/support/alloc.h>
22 #include <grpc/support/log.h>
23 
24 #include "src/core/lib/compression/stream_compression_gzip.h"
25 #include "src/core/lib/iomgr/exec_ctx.h"
26 #include "src/core/lib/slice/slice_internal.h"
27 
28 #define OUTPUT_BLOCK_SIZE (1024)
29 
30 typedef struct grpc_stream_compression_context_gzip {
31   grpc_stream_compression_context base;
32 
33   z_stream zs;
34   int (*flate)(z_stream* zs, int flush);
35 } grpc_stream_compression_context_gzip;
36 
gzip_flate(grpc_stream_compression_context_gzip * ctx,grpc_slice_buffer * in,grpc_slice_buffer * out,size_t * output_size,size_t max_output_size,int flush,bool * end_of_context)37 static bool gzip_flate(grpc_stream_compression_context_gzip* ctx,
38                        grpc_slice_buffer* in, grpc_slice_buffer* out,
39                        size_t* output_size, size_t max_output_size, int flush,
40                        bool* end_of_context) {
41   GPR_ASSERT(flush == 0 || flush == Z_SYNC_FLUSH || flush == Z_FINISH);
42   /* Full flush is not allowed when inflating. */
43   GPR_ASSERT(!(ctx->flate == inflate && (flush == Z_FINISH)));
44 
45   grpc_core::ExecCtx exec_ctx;
46   int r;
47   bool eoc = false;
48   size_t original_max_output_size = max_output_size;
49   while (max_output_size > 0 && (in->length > 0 || flush) && !eoc) {
50     size_t slice_size = max_output_size < OUTPUT_BLOCK_SIZE ? max_output_size
51                                                             : OUTPUT_BLOCK_SIZE;
52     grpc_slice slice_out = GRPC_SLICE_MALLOC(slice_size);
53     ctx->zs.avail_out = static_cast<uInt>(slice_size);
54     ctx->zs.next_out = GRPC_SLICE_START_PTR(slice_out);
55     while (ctx->zs.avail_out > 0 && in->length > 0 && !eoc) {
56       grpc_slice slice = grpc_slice_buffer_take_first(in);
57       ctx->zs.avail_in = static_cast<uInt> GRPC_SLICE_LENGTH(slice);
58       ctx->zs.next_in = GRPC_SLICE_START_PTR(slice);
59       r = ctx->flate(&ctx->zs, Z_NO_FLUSH);
60       if (r < 0 && r != Z_BUF_ERROR) {
61         gpr_log(GPR_ERROR, "zlib error (%d)", r);
62         grpc_slice_unref_internal(slice_out);
63 
64         return false;
65       } else if (r == Z_STREAM_END && ctx->flate == inflate) {
66         eoc = true;
67       }
68       if (ctx->zs.avail_in > 0) {
69         grpc_slice_buffer_undo_take_first(
70             in,
71             grpc_slice_sub(slice, GRPC_SLICE_LENGTH(slice) - ctx->zs.avail_in,
72                            GRPC_SLICE_LENGTH(slice)));
73       }
74       grpc_slice_unref_internal(slice);
75     }
76     if (flush != 0 && ctx->zs.avail_out > 0 && !eoc) {
77       GPR_ASSERT(in->length == 0);
78       r = ctx->flate(&ctx->zs, flush);
79       if (flush == Z_SYNC_FLUSH) {
80         switch (r) {
81           case Z_OK:
82             /* Maybe flush is not complete; just made some partial progress. */
83             if (ctx->zs.avail_out > 0) {
84               flush = 0;
85             }
86             break;
87           case Z_BUF_ERROR:
88           case Z_STREAM_END:
89             flush = 0;
90             break;
91           default:
92             gpr_log(GPR_ERROR, "zlib error (%d)", r);
93             grpc_slice_unref_internal(slice_out);
94 
95             return false;
96         }
97       } else if (flush == Z_FINISH) {
98         switch (r) {
99           case Z_OK:
100           case Z_BUF_ERROR:
101             /* Wait for the next loop to assign additional output space. */
102             GPR_ASSERT(ctx->zs.avail_out == 0);
103             break;
104           case Z_STREAM_END:
105             flush = 0;
106             break;
107           default:
108             gpr_log(GPR_ERROR, "zlib error (%d)", r);
109             grpc_slice_unref_internal(slice_out);
110 
111             return false;
112         }
113       }
114     }
115 
116     if (ctx->zs.avail_out == 0) {
117       grpc_slice_buffer_add(out, slice_out);
118     } else if (ctx->zs.avail_out < slice_size) {
119       size_t len = GRPC_SLICE_LENGTH(slice_out);
120       GRPC_SLICE_SET_LENGTH(slice_out, len - ctx->zs.avail_out);
121       grpc_slice_buffer_add(out, slice_out);
122     } else {
123       grpc_slice_unref_internal(slice_out);
124     }
125     max_output_size -= (slice_size - ctx->zs.avail_out);
126   }
127 
128   if (end_of_context) {
129     *end_of_context = eoc;
130   }
131   if (output_size) {
132     *output_size = original_max_output_size - max_output_size;
133   }
134   return true;
135 }
136 
grpc_stream_compress_gzip(grpc_stream_compression_context * ctx,grpc_slice_buffer * in,grpc_slice_buffer * out,size_t * output_size,size_t max_output_size,grpc_stream_compression_flush flush)137 static bool grpc_stream_compress_gzip(grpc_stream_compression_context* ctx,
138                                       grpc_slice_buffer* in,
139                                       grpc_slice_buffer* out,
140                                       size_t* output_size,
141                                       size_t max_output_size,
142                                       grpc_stream_compression_flush flush) {
143   if (ctx == nullptr) {
144     return false;
145   }
146   grpc_stream_compression_context_gzip* gzip_ctx =
147       reinterpret_cast<grpc_stream_compression_context_gzip*>(ctx);
148   GPR_ASSERT(gzip_ctx->flate == deflate);
149   int gzip_flush;
150   switch (flush) {
151     case GRPC_STREAM_COMPRESSION_FLUSH_NONE:
152       gzip_flush = 0;
153       break;
154     case GRPC_STREAM_COMPRESSION_FLUSH_SYNC:
155       gzip_flush = Z_SYNC_FLUSH;
156       break;
157     case GRPC_STREAM_COMPRESSION_FLUSH_FINISH:
158       gzip_flush = Z_FINISH;
159       break;
160     default:
161       gzip_flush = 0;
162   }
163   return gzip_flate(gzip_ctx, in, out, output_size, max_output_size, gzip_flush,
164                     nullptr);
165 }
166 
grpc_stream_decompress_gzip(grpc_stream_compression_context * ctx,grpc_slice_buffer * in,grpc_slice_buffer * out,size_t * output_size,size_t max_output_size,bool * end_of_context)167 static bool grpc_stream_decompress_gzip(grpc_stream_compression_context* ctx,
168                                         grpc_slice_buffer* in,
169                                         grpc_slice_buffer* out,
170                                         size_t* output_size,
171                                         size_t max_output_size,
172                                         bool* end_of_context) {
173   if (ctx == nullptr) {
174     return false;
175   }
176   grpc_stream_compression_context_gzip* gzip_ctx =
177       reinterpret_cast<grpc_stream_compression_context_gzip*>(ctx);
178   GPR_ASSERT(gzip_ctx->flate == inflate);
179   return gzip_flate(gzip_ctx, in, out, output_size, max_output_size,
180                     Z_SYNC_FLUSH, end_of_context);
181 }
182 
183 static grpc_stream_compression_context*
grpc_stream_compression_context_create_gzip(grpc_stream_compression_method method)184 grpc_stream_compression_context_create_gzip(
185     grpc_stream_compression_method method) {
186   GPR_ASSERT(method == GRPC_STREAM_COMPRESSION_GZIP_COMPRESS ||
187              method == GRPC_STREAM_COMPRESSION_GZIP_DECOMPRESS);
188   grpc_stream_compression_context_gzip* gzip_ctx =
189       static_cast<grpc_stream_compression_context_gzip*>(
190           gpr_zalloc(sizeof(grpc_stream_compression_context_gzip)));
191   int r;
192   if (gzip_ctx == nullptr) {
193     return nullptr;
194   }
195   if (method == GRPC_STREAM_COMPRESSION_GZIP_DECOMPRESS) {
196     r = inflateInit2(&gzip_ctx->zs, 0x1F);
197     gzip_ctx->flate = inflate;
198   } else {
199     r = deflateInit2(&gzip_ctx->zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED, 0x1F, 8,
200                      Z_DEFAULT_STRATEGY);
201     gzip_ctx->flate = deflate;
202   }
203   if (r != Z_OK) {
204     gpr_free(gzip_ctx);
205     return nullptr;
206   }
207 
208   gzip_ctx->base.vtable = &grpc_stream_compression_gzip_vtable;
209   return reinterpret_cast<grpc_stream_compression_context*>(gzip_ctx);
210 }
211 
grpc_stream_compression_context_destroy_gzip(grpc_stream_compression_context * ctx)212 static void grpc_stream_compression_context_destroy_gzip(
213     grpc_stream_compression_context* ctx) {
214   if (ctx == nullptr) {
215     return;
216   }
217   grpc_stream_compression_context_gzip* gzip_ctx =
218       reinterpret_cast<grpc_stream_compression_context_gzip*>(ctx);
219   if (gzip_ctx->flate == inflate) {
220     inflateEnd(&gzip_ctx->zs);
221   } else {
222     deflateEnd(&gzip_ctx->zs);
223   }
224   gpr_free(ctx);
225 }
226 
227 const grpc_stream_compression_vtable grpc_stream_compression_gzip_vtable = {
228     grpc_stream_compress_gzip, grpc_stream_decompress_gzip,
229     grpc_stream_compression_context_create_gzip,
230     grpc_stream_compression_context_destroy_gzip};
231