aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core
diff options
context:
space:
mode:
Diffstat (limited to 'src/core')
-rw-r--r--src/core/ext/filters/client_channel/OWNERS (renamed from src/core/OWNERS)3
-rw-r--r--src/core/ext/filters/client_channel/subchannel.c1
-rw-r--r--src/core/ext/transport/chttp2/transport/chttp2_transport.c1
-rw-r--r--src/core/ext/transport/chttp2/transport/internal.h1
-rw-r--r--src/core/lib/compression/stream_compression.c191
-rw-r--r--src/core/lib/compression/stream_compression.h90
-rw-r--r--src/core/lib/iomgr/OWNERS1
-rw-r--r--src/core/lib/iomgr/ev_epoll_limited_pollers_linux.c25
-rw-r--r--src/core/lib/iomgr/ev_epollsig_linux.c25
9 files changed, 309 insertions, 29 deletions
diff --git a/src/core/OWNERS b/src/core/ext/filters/client_channel/OWNERS
index 986c9a1c3c..773bc73179 100644
--- a/src/core/OWNERS
+++ b/src/core/ext/filters/client_channel/OWNERS
@@ -1,5 +1,4 @@
set noparent
-@ctiller
@markdroth
@dgquintas
-
+@ctiller
diff --git a/src/core/ext/filters/client_channel/subchannel.c b/src/core/ext/filters/client_channel/subchannel.c
index 88157ed738..5788819331 100644
--- a/src/core/ext/filters/client_channel/subchannel.c
+++ b/src/core/ext/filters/client_channel/subchannel.c
@@ -188,6 +188,7 @@ static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg,
grpc_connector_unref(exec_ctx, c->connector);
grpc_pollset_set_destroy(exec_ctx, c->pollset_set);
grpc_subchannel_key_destroy(exec_ctx, c->key);
+ gpr_mu_destroy(&c->mu);
gpr_free(c);
}
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index bd25f69a1d..731ebf400f 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -33,6 +33,7 @@
#include "src/core/ext/transport/chttp2/transport/internal.h"
#include "src/core/ext/transport/chttp2/transport/varint.h"
#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/compression/stream_compression.h"
#include "src/core/lib/http/parser.h"
#include "src/core/lib/iomgr/executor.h"
#include "src/core/lib/iomgr/timer.h"
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index a5b67e5aba..4563b78e75 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -33,6 +33,7 @@
#include "src/core/ext/transport/chttp2/transport/hpack_parser.h"
#include "src/core/ext/transport/chttp2/transport/incoming_metadata.h"
#include "src/core/ext/transport/chttp2/transport/stream_map.h"
+#include "src/core/lib/compression/stream_compression.h"
#include "src/core/lib/iomgr/combiner.h"
#include "src/core/lib/iomgr/endpoint.h"
#include "src/core/lib/iomgr/timer.h"
diff --git a/src/core/lib/compression/stream_compression.c b/src/core/lib/compression/stream_compression.c
new file mode 100644
index 0000000000..df13d53e06
--- /dev/null
+++ b/src/core/lib/compression/stream_compression.c
@@ -0,0 +1,191 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+
+#include "src/core/lib/compression/stream_compression.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/slice/slice_internal.h"
+
+#define OUTPUT_BLOCK_SIZE (1024)
+
+static bool gzip_flate(grpc_stream_compression_context *ctx,
+ grpc_slice_buffer *in, grpc_slice_buffer *out,
+ size_t *output_size, size_t max_output_size, int flush,
+ bool *end_of_context) {
+ GPR_ASSERT(flush == 0 || flush == Z_SYNC_FLUSH || flush == Z_FINISH);
+ /* Full flush is not allowed when inflating. */
+ GPR_ASSERT(!(ctx->flate == inflate && (flush == Z_FINISH)));
+
+ grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+ int r;
+ bool eoc = false;
+ size_t original_max_output_size = max_output_size;
+ while (max_output_size > 0 && (in->length > 0 || flush) && !eoc) {
+ size_t slice_size = max_output_size < OUTPUT_BLOCK_SIZE ? max_output_size
+ : OUTPUT_BLOCK_SIZE;
+ grpc_slice slice_out = GRPC_SLICE_MALLOC(slice_size);
+ ctx->zs.avail_out = (uInt)slice_size;
+ ctx->zs.next_out = GRPC_SLICE_START_PTR(slice_out);
+ while (ctx->zs.avail_out > 0 && in->length > 0 && !eoc) {
+ grpc_slice slice = grpc_slice_buffer_take_first(in);
+ ctx->zs.avail_in = (uInt)GRPC_SLICE_LENGTH(slice);
+ ctx->zs.next_in = GRPC_SLICE_START_PTR(slice);
+ r = ctx->flate(&ctx->zs, Z_NO_FLUSH);
+ if (r < 0 && r != Z_BUF_ERROR) {
+ gpr_log(GPR_ERROR, "zlib error (%d)", r);
+ grpc_slice_unref_internal(&exec_ctx, slice_out);
+ grpc_exec_ctx_finish(&exec_ctx);
+ return false;
+ } else if (r == Z_STREAM_END && ctx->flate == inflate) {
+ eoc = true;
+ }
+ if (ctx->zs.avail_in > 0) {
+ grpc_slice_buffer_undo_take_first(
+ in,
+ grpc_slice_sub(slice, GRPC_SLICE_LENGTH(slice) - ctx->zs.avail_in,
+ GRPC_SLICE_LENGTH(slice)));
+ }
+ grpc_slice_unref_internal(&exec_ctx, slice);
+ }
+ if (flush != 0 && ctx->zs.avail_out > 0 && !eoc) {
+ GPR_ASSERT(in->length == 0);
+ r = ctx->flate(&ctx->zs, flush);
+ if (flush == Z_SYNC_FLUSH) {
+ switch (r) {
+ case Z_OK:
+ /* Maybe flush is not complete; just made some partial progress. */
+ if (ctx->zs.avail_out > 0) {
+ flush = 0;
+ }
+ break;
+ case Z_BUF_ERROR:
+ case Z_STREAM_END:
+ flush = 0;
+ break;
+ default:
+ gpr_log(GPR_ERROR, "zlib error (%d)", r);
+ grpc_slice_unref_internal(&exec_ctx, slice_out);
+ grpc_exec_ctx_finish(&exec_ctx);
+ return false;
+ }
+ } else if (flush == Z_FINISH) {
+ switch (r) {
+ case Z_OK:
+ case Z_BUF_ERROR:
+ /* Wait for the next loop to assign additional output space. */
+ GPR_ASSERT(ctx->zs.avail_out == 0);
+ break;
+ case Z_STREAM_END:
+ flush = 0;
+ break;
+ default:
+ gpr_log(GPR_ERROR, "zlib error (%d)", r);
+ grpc_slice_unref_internal(&exec_ctx, slice_out);
+ grpc_exec_ctx_finish(&exec_ctx);
+ return false;
+ }
+ }
+ }
+
+ if (ctx->zs.avail_out == 0) {
+ grpc_slice_buffer_add(out, slice_out);
+ } else if (ctx->zs.avail_out < slice_size) {
+ slice_out.data.refcounted.length -= ctx->zs.avail_out;
+ grpc_slice_buffer_add(out, slice_out);
+ } else {
+ grpc_slice_unref_internal(&exec_ctx, slice_out);
+ }
+ max_output_size -= (slice_size - ctx->zs.avail_out);
+ }
+ grpc_exec_ctx_finish(&exec_ctx);
+ if (end_of_context) {
+ *end_of_context = eoc;
+ }
+ if (output_size) {
+ *output_size = original_max_output_size - max_output_size;
+ }
+ return true;
+}
+
+bool grpc_stream_compress(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) {
+ GPR_ASSERT(ctx->flate == deflate);
+ int gzip_flush;
+ switch (flush) {
+ case GRPC_STREAM_COMPRESSION_FLUSH_NONE:
+ gzip_flush = 0;
+ break;
+ case GRPC_STREAM_COMPRESSION_FLUSH_SYNC:
+ gzip_flush = Z_SYNC_FLUSH;
+ break;
+ case GRPC_STREAM_COMPRESSION_FLUSH_FINISH:
+ gzip_flush = Z_FINISH;
+ break;
+ default:
+ gzip_flush = 0;
+ }
+ return gzip_flate(ctx, in, out, output_size, max_output_size, gzip_flush,
+ NULL);
+}
+
+bool grpc_stream_decompress(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) {
+ GPR_ASSERT(ctx->flate == inflate);
+ return gzip_flate(ctx, in, out, output_size, max_output_size, Z_SYNC_FLUSH,
+ end_of_context);
+}
+
+grpc_stream_compression_context *grpc_stream_compression_context_create(
+ grpc_stream_compression_method method) {
+ grpc_stream_compression_context *ctx =
+ gpr_zalloc(sizeof(grpc_stream_compression_context));
+ int r;
+ if (ctx == NULL) {
+ return NULL;
+ }
+ if (method == GRPC_STREAM_COMPRESSION_DECOMPRESS) {
+ r = inflateInit2(&ctx->zs, 0x1F);
+ ctx->flate = inflate;
+ } else {
+ r = deflateInit2(&ctx->zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED, 0x1F, 8,
+ Z_DEFAULT_STRATEGY);
+ ctx->flate = deflate;
+ }
+ if (r != Z_OK) {
+ gpr_free(ctx);
+ return NULL;
+ }
+
+ return ctx;
+}
+
+void grpc_stream_compression_context_destroy(
+ grpc_stream_compression_context *ctx) {
+ if (ctx->flate == inflate) {
+ inflateEnd(&ctx->zs);
+ } else {
+ deflateEnd(&ctx->zs);
+ }
+ gpr_free(ctx);
+}
diff --git a/src/core/lib/compression/stream_compression.h b/src/core/lib/compression/stream_compression.h
new file mode 100644
index 0000000000..844dff81a3
--- /dev/null
+++ b/src/core/lib/compression/stream_compression.h
@@ -0,0 +1,90 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPC_CORE_LIB_COMPRESSION_STREAM_COMPRESSION_H
+#define GRPC_CORE_LIB_COMPRESSION_STREAM_COMPRESSION_H
+
+#include <stdbool.h>
+
+#include <grpc/slice_buffer.h>
+#include <zlib.h>
+
+/* Stream compression/decompression context */
+typedef struct grpc_stream_compression_context {
+ z_stream zs;
+ int (*flate)(z_stream *zs, int flush);
+} grpc_stream_compression_context;
+
+typedef enum grpc_stream_compression_method {
+ GRPC_STREAM_COMPRESSION_COMPRESS = 0,
+ GRPC_STREAM_COMPRESSION_DECOMPRESS,
+ GRPC_STREAM_COMPRESSION_METHOD_COUNT
+} grpc_stream_compression_method;
+
+typedef enum grpc_stream_compression_flush {
+ GRPC_STREAM_COMPRESSION_FLUSH_NONE = 0,
+ GRPC_STREAM_COMPRESSION_FLUSH_SYNC,
+ GRPC_STREAM_COMPRESSION_FLUSH_FINISH,
+ GRPC_STREAM_COMPRESSION_FLUSH_COUNT
+} grpc_stream_compression_flush;
+
+/**
+ * Compress bytes provided in \a in with a given context, with an optional flush
+ * at the end of compression. Emits at most \a max_output_size compressed bytes
+ * into \a out. If all the bytes in input buffer \a in are depleted and \a flush
+ * is not GRPC_STREAM_COMPRESSION_FLUSH_NONE, the corresponding flush method is
+ * executed. The total number of bytes emitted is outputed in \a output_size.
+ *
+ * A SYNC flush indicates that the entire messages in \a in can be decompressed
+ * from \a out. A FINISH flush implies a SYNC flush, and that any further
+ * compression will not be dependent on the state of the current context and any
+ * previous compressed bytes. It allows corresponding decompression context to
+ * be dropped when reaching this boundary.
+ */
+bool grpc_stream_compress(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);
+
+/**
+ * Decompress bytes provided in \a in with a given context. Emits at most \a
+ * max_output_size decompressed bytes into \a out. If decompression process
+ * reached the end of a gzip stream, \a end_of_context is set to true; otherwise
+ * it is set to false. The total number of bytes emitted is outputed in \a
+ * output_size.
+ */
+bool grpc_stream_decompress(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);
+
+/**
+ * Creates a stream compression context. \a pending_bytes_buffer is the input
+ * buffer for compression/decompression operations. \a method specifies whether
+ * the context is for compression or decompression.
+ */
+grpc_stream_compression_context *grpc_stream_compression_context_create(
+ grpc_stream_compression_method method);
+
+/**
+ * Destroys a stream compression context.
+ */
+void grpc_stream_compression_context_destroy(
+ grpc_stream_compression_context *ctx);
+
+#endif
diff --git a/src/core/lib/iomgr/OWNERS b/src/core/lib/iomgr/OWNERS
deleted file mode 100644
index 8ee86c2618..0000000000
--- a/src/core/lib/iomgr/OWNERS
+++ /dev/null
@@ -1 +0,0 @@
-@murgatroid99 *_uv.c *_uv.h
diff --git a/src/core/lib/iomgr/ev_epoll_limited_pollers_linux.c b/src/core/lib/iomgr/ev_epoll_limited_pollers_linux.c
index 9f82c480bc..27b4892d1d 100644
--- a/src/core/lib/iomgr/ev_epoll_limited_pollers_linux.c
+++ b/src/core/lib/iomgr/ev_epoll_limited_pollers_linux.c
@@ -932,24 +932,12 @@ static int fd_wrapped_fd(grpc_fd *fd) {
static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
grpc_closure *on_done, int *release_fd,
const char *reason) {
- bool is_fd_closed = false;
grpc_error *error = GRPC_ERROR_NONE;
polling_island *unref_pi = NULL;
gpr_mu_lock(&fd->po.mu);
fd->on_done_closure = on_done;
- /* If release_fd is not NULL, we should be relinquishing control of the file
- descriptor fd->fd (but we still own the grpc_fd structure). */
- if (release_fd != NULL) {
- *release_fd = fd->fd;
- } else {
- close(fd->fd);
- is_fd_closed = true;
- }
-
- fd->orphaned = true;
-
/* Remove the active status but keep referenced. We want this grpc_fd struct
to be alive (and not added to freelist) until the end of this function */
REF_BY(fd, 1, reason);
@@ -964,13 +952,24 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
before doing this.) */
if (fd->po.pi != NULL) {
polling_island *pi_latest = polling_island_lock(fd->po.pi);
- polling_island_remove_fd_locked(pi_latest, fd, is_fd_closed, &error);
+ polling_island_remove_fd_locked(pi_latest, fd, false /* is_fd_closed */,
+ &error);
gpr_mu_unlock(&pi_latest->mu);
unref_pi = fd->po.pi;
fd->po.pi = NULL;
}
+ /* If release_fd is not NULL, we should be relinquishing control of the file
+ descriptor fd->fd (but we still own the grpc_fd structure). */
+ if (release_fd != NULL) {
+ *release_fd = fd->fd;
+ } else {
+ close(fd->fd);
+ }
+
+ fd->orphaned = true;
+
GRPC_CLOSURE_SCHED(exec_ctx, fd->on_done_closure, GRPC_ERROR_REF(error));
gpr_mu_unlock(&fd->po.mu);
diff --git a/src/core/lib/iomgr/ev_epollsig_linux.c b/src/core/lib/iomgr/ev_epollsig_linux.c
index 3c4ca9c7c5..0d969dccce 100644
--- a/src/core/lib/iomgr/ev_epollsig_linux.c
+++ b/src/core/lib/iomgr/ev_epollsig_linux.c
@@ -855,24 +855,12 @@ static int fd_wrapped_fd(grpc_fd *fd) {
static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
grpc_closure *on_done, int *release_fd,
const char *reason) {
- bool is_fd_closed = false;
grpc_error *error = GRPC_ERROR_NONE;
polling_island *unref_pi = NULL;
gpr_mu_lock(&fd->po.mu);
fd->on_done_closure = on_done;
- /* If release_fd is not NULL, we should be relinquishing control of the file
- descriptor fd->fd (but we still own the grpc_fd structure). */
- if (release_fd != NULL) {
- *release_fd = fd->fd;
- } else {
- close(fd->fd);
- is_fd_closed = true;
- }
-
- fd->orphaned = true;
-
/* Remove the active status but keep referenced. We want this grpc_fd struct
to be alive (and not added to freelist) until the end of this function */
REF_BY(fd, 1, reason);
@@ -887,13 +875,24 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
before doing this.) */
if (fd->po.pi != NULL) {
polling_island *pi_latest = polling_island_lock(fd->po.pi);
- polling_island_remove_fd_locked(pi_latest, fd, is_fd_closed, &error);
+ polling_island_remove_fd_locked(pi_latest, fd, false /* is_fd_closed */,
+ &error);
gpr_mu_unlock(&pi_latest->mu);
unref_pi = fd->po.pi;
fd->po.pi = NULL;
}
+ /* If release_fd is not NULL, we should be relinquishing control of the file
+ descriptor fd->fd (but we still own the grpc_fd structure). */
+ if (release_fd != NULL) {
+ *release_fd = fd->fd;
+ } else {
+ close(fd->fd);
+ }
+
+ fd->orphaned = true;
+
GRPC_CLOSURE_SCHED(exec_ctx, fd->on_done_closure, GRPC_ERROR_REF(error));
gpr_mu_unlock(&fd->po.mu);
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-06 10:03:50 +0000