aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/core/memory_usage
diff options
context:
space:
mode:
Diffstat (limited to 'test/core/memory_usage')
-rw-r--r--test/core/memory_usage/client.c314
-rw-r--r--test/core/memory_usage/memory_usage_test.c93
-rw-r--r--test/core/memory_usage/server.c321
3 files changed, 728 insertions, 0 deletions
diff --git a/test/core/memory_usage/client.c b/test/core/memory_usage/client.c
new file mode 100644
index 0000000000..9fc122b4c3
--- /dev/null
+++ b/test/core/memory_usage/client.c
@@ -0,0 +1,314 @@
+/*
+ *
+ * Copyright 2016, 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 <grpc/grpc.h>
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/byte_buffer_reader.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/cmdline.h>
+#include <grpc/support/log.h>
+#include <grpc/support/time.h>
+#include <grpc/support/useful.h>
+#include "src/core/lib/support/string.h"
+#include "test/core/util/memory_counters.h"
+#include "test/core/util/test_config.h"
+
+static grpc_channel *channel;
+static grpc_completion_queue *cq;
+static grpc_op metadata_ops[2];
+static grpc_op status_ops[2];
+static grpc_op snapshot_ops[6];
+static grpc_op *op;
+
+typedef struct {
+ grpc_call *call;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_status_code status;
+ char *details;
+ size_t details_capacity;
+ grpc_metadata_array trailing_metadata_recv;
+} fling_call;
+
+// Statically allocate call data structs. Enough to accomodate 10000 ping-pong
+// calls and 1 extra for the snapshot calls.
+static fling_call calls[10001];
+
+static void *tag(intptr_t t) { return (void *)t; }
+
+// A call is intentionally divided into two steps. First step is to initiate a
+// call (i.e send and recv metadata). A call is outstanding after we initated,
+// so we can measure the call memory usage.
+static void init_ping_pong_request(int call_idx) {
+ grpc_metadata_array_init(&calls[call_idx].initial_metadata_recv);
+
+ memset(metadata_ops, 0, sizeof(metadata_ops));
+ op = metadata_ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata = &calls[call_idx].initial_metadata_recv;
+ op++;
+
+ calls[call_idx].call = grpc_channel_create_call(
+ channel, NULL, GRPC_PROPAGATE_DEFAULTS, cq, "/Reflector/reflectUnary",
+ "localhost", gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+
+ GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(calls[call_idx].call,
+ metadata_ops,
+ (size_t)(op - metadata_ops),
+ tag(call_idx), NULL));
+ grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+}
+
+// Second step is to finish the call (i.e recv status) and destroy the call.
+static void finish_ping_pong_request(int call_idx) {
+ grpc_metadata_array_init(&calls[call_idx].trailing_metadata_recv);
+
+ memset(status_ops, 0, sizeof(status_ops));
+ op = status_ops;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata =
+ &calls[call_idx].trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &calls[call_idx].status;
+ op->data.recv_status_on_client.status_details = &calls[call_idx].details;
+ op->data.recv_status_on_client.status_details_capacity =
+ &calls[call_idx].details_capacity;
+ op++;
+
+ GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(calls[call_idx].call,
+ status_ops,
+ (size_t)(op - status_ops),
+ tag(call_idx), NULL));
+ grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+ grpc_metadata_array_destroy(&calls[call_idx].initial_metadata_recv);
+ grpc_metadata_array_destroy(&calls[call_idx].trailing_metadata_recv);
+ gpr_free(calls[call_idx].details);
+ grpc_call_destroy(calls[call_idx].call);
+ calls[call_idx].call = NULL;
+}
+
+static struct grpc_memory_counters send_snapshot_request(
+ int call_idx, const char *call_type) {
+ grpc_metadata_array_init(&calls[call_idx].initial_metadata_recv);
+ grpc_metadata_array_init(&calls[call_idx].trailing_metadata_recv);
+
+ grpc_byte_buffer *response_payload_recv = NULL;
+ memset(snapshot_ops, 0, sizeof(snapshot_ops));
+ op = snapshot_ops;
+
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata = &calls[call_idx].initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata =
+ &calls[call_idx].trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &calls[call_idx].status;
+ op->data.recv_status_on_client.status_details = &calls[call_idx].details;
+ op->data.recv_status_on_client.status_details_capacity =
+ &calls[call_idx].details_capacity;
+ op++;
+
+ calls[call_idx].call = grpc_channel_create_call(
+ channel, NULL, GRPC_PROPAGATE_DEFAULTS, cq, call_type, "localhost",
+ gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+ GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(
+ calls[call_idx].call, snapshot_ops,
+ (size_t)(op - snapshot_ops), (void *)0, NULL));
+ grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+
+ grpc_byte_buffer_reader reader;
+ grpc_byte_buffer_reader_init(&reader, response_payload_recv);
+ grpc_slice response = grpc_byte_buffer_reader_readall(&reader);
+
+ struct grpc_memory_counters snapshot;
+ snapshot.total_size_absolute =
+ ((struct grpc_memory_counters *)GRPC_SLICE_START_PTR(response))
+ ->total_size_absolute;
+ snapshot.total_allocs_absolute =
+ ((struct grpc_memory_counters *)GRPC_SLICE_START_PTR(response))
+ ->total_allocs_absolute;
+ snapshot.total_size_relative =
+ ((struct grpc_memory_counters *)GRPC_SLICE_START_PTR(response))
+ ->total_size_relative;
+ snapshot.total_allocs_relative =
+ ((struct grpc_memory_counters *)GRPC_SLICE_START_PTR(response))
+ ->total_allocs_relative;
+
+ grpc_metadata_array_destroy(&calls[call_idx].initial_metadata_recv);
+ grpc_metadata_array_destroy(&calls[call_idx].trailing_metadata_recv);
+ grpc_slice_unref(response);
+ grpc_byte_buffer_reader_destroy(&reader);
+ grpc_byte_buffer_destroy(response_payload_recv);
+ gpr_free(calls[call_idx].details);
+ calls[call_idx].details = NULL;
+ calls[call_idx].details_capacity = 0;
+ grpc_call_destroy(calls[call_idx].call);
+ calls[call_idx].call = NULL;
+
+ return snapshot;
+}
+
+int main(int argc, char **argv) {
+ grpc_memory_counters_init();
+ grpc_slice slice = grpc_slice_from_copied_string("x");
+ char *fake_argv[1];
+
+ char *target = "localhost:443";
+ gpr_cmdline *cl;
+ grpc_event event;
+
+ grpc_init();
+
+ GPR_ASSERT(argc >= 1);
+ fake_argv[0] = argv[0];
+ grpc_test_init(1, fake_argv);
+
+ int warmup_iterations = 100;
+ int benchmark_iterations = 1000;
+
+ cl = gpr_cmdline_create("memory profiling client");
+ gpr_cmdline_add_string(cl, "target", "Target host:port", &target);
+ gpr_cmdline_add_int(cl, "warmup", "Warmup iterations", &warmup_iterations);
+ gpr_cmdline_add_int(cl, "benchmark", "Benchmark iterations",
+ &benchmark_iterations);
+ gpr_cmdline_parse(cl, argc, argv);
+ gpr_cmdline_destroy(cl);
+
+ for (int k = 0; k < (int)(sizeof(calls) / sizeof(fling_call)); k++) {
+ calls[k].details = NULL;
+ calls[k].details_capacity = 0;
+ }
+
+ cq = grpc_completion_queue_create(NULL);
+
+ struct grpc_memory_counters client_channel_start =
+ grpc_memory_counters_snapshot();
+ channel = grpc_insecure_channel_create(target, NULL, NULL);
+
+ int call_idx = 0;
+
+ struct grpc_memory_counters before_server_create =
+ send_snapshot_request(0, "Reflector/GetBeforeSvrCreation");
+ struct grpc_memory_counters after_server_create =
+ send_snapshot_request(0, "Reflector/GetAfterSvrCreation");
+
+ // warmup period
+ for (call_idx = 0; call_idx < warmup_iterations; ++call_idx) {
+ init_ping_pong_request(call_idx + 1);
+ }
+
+ struct grpc_memory_counters server_benchmark_calls_start =
+ send_snapshot_request(0, "Reflector/SimpleSnapshot");
+
+ struct grpc_memory_counters client_benchmark_calls_start =
+ grpc_memory_counters_snapshot();
+
+ // benchmark period
+ for (; call_idx < warmup_iterations + benchmark_iterations; ++call_idx) {
+ init_ping_pong_request(call_idx + 1);
+ }
+
+ struct grpc_memory_counters client_calls_inflight =
+ grpc_memory_counters_snapshot();
+
+ struct grpc_memory_counters server_calls_inflight =
+ send_snapshot_request(0, "Reflector/DestroyCalls");
+
+ do {
+ event = grpc_completion_queue_next(
+ cq, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
+ gpr_time_from_micros(10000, GPR_TIMESPAN)),
+ NULL);
+ } while (event.type != GRPC_QUEUE_TIMEOUT);
+
+ // second step - recv status and destroy call
+ for (call_idx = 0; call_idx < warmup_iterations + benchmark_iterations;
+ ++call_idx) {
+ finish_ping_pong_request(call_idx + 1);
+ }
+
+ struct grpc_memory_counters server_calls_end =
+ send_snapshot_request(0, "Reflector/SimpleSnapshot");
+
+ struct grpc_memory_counters client_channel_end =
+ grpc_memory_counters_snapshot();
+
+ grpc_channel_destroy(channel);
+ grpc_completion_queue_shutdown(cq);
+
+ do {
+ event = grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME),
+ NULL);
+ } while (event.type != GRPC_QUEUE_SHUTDOWN);
+ grpc_slice_unref(slice);
+
+ grpc_completion_queue_destroy(cq);
+ grpc_shutdown();
+
+ gpr_log(GPR_INFO, "---------client stats--------");
+ gpr_log(GPR_INFO, "client call memory usage: %f bytes per call",
+ (double)(client_calls_inflight.total_size_relative -
+ client_benchmark_calls_start.total_size_relative) /
+ benchmark_iterations);
+ gpr_log(GPR_INFO, "client channel memory usage %zi bytes",
+ client_channel_end.total_size_relative -
+ client_channel_start.total_size_relative);
+
+ gpr_log(GPR_INFO, "---------server stats--------");
+ gpr_log(GPR_INFO, "server create: %zi bytes",
+ after_server_create.total_size_relative -
+ before_server_create.total_size_relative);
+ gpr_log(GPR_INFO, "server call memory usage: %f bytes per call",
+ (double)(server_calls_inflight.total_size_relative -
+ server_benchmark_calls_start.total_size_relative) /
+ benchmark_iterations);
+ gpr_log(GPR_INFO, "server channel memory usage %zi bytes",
+ server_calls_end.total_size_relative -
+ after_server_create.total_size_relative);
+
+ grpc_memory_counters_destroy();
+ return 0;
+}
diff --git a/test/core/memory_usage/memory_usage_test.c b/test/core/memory_usage/memory_usage_test.c
new file mode 100644
index 0000000000..7e7a9d050d
--- /dev/null
+++ b/test/core/memory_usage/memory_usage_test.c
@@ -0,0 +1,93 @@
+/*
+ *
+ * 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 <stdio.h>
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/host_port.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/subprocess.h>
+#include "src/core/lib/support/string.h"
+#include "test/core/util/port.h"
+
+int main(int argc, char **argv) {
+ char *me = argv[0];
+ char *lslash = strrchr(me, '/');
+ char root[1024];
+ int port = grpc_pick_unused_port_or_die();
+ char *args[10];
+ int status;
+ gpr_subprocess *svr, *cli;
+ /* figure out where we are */
+ if (lslash) {
+ memcpy(root, me, (size_t)(lslash - me));
+ root[lslash - me] = 0;
+ } else {
+ strcpy(root, ".");
+ }
+ /* start the server */
+ gpr_asprintf(&args[0], "%s/memory_profile_server%s", root,
+ gpr_subprocess_binary_extension());
+ args[1] = "--bind";
+ gpr_join_host_port(&args[2], "::", port);
+ args[3] = "--no-secure";
+ svr = gpr_subprocess_create(4, (const char **)args);
+ gpr_free(args[0]);
+ gpr_free(args[2]);
+
+ /* start the client */
+ gpr_asprintf(&args[0], "%s/memory_profile_client%s", root,
+ gpr_subprocess_binary_extension());
+ args[1] = "--target";
+ gpr_join_host_port(&args[2], "127.0.0.1", port);
+ args[3] = "--warmup=1000";
+ args[4] = "--benchmark=9000";
+ cli = gpr_subprocess_create(5, (const char **)args);
+ gpr_free(args[0]);
+ gpr_free(args[2]);
+
+ /* wait for completion */
+ printf("waiting for client\n");
+ if ((status = gpr_subprocess_join(cli))) {
+ gpr_subprocess_destroy(cli);
+ gpr_subprocess_destroy(svr);
+ return status;
+ }
+ gpr_subprocess_destroy(cli);
+
+ gpr_subprocess_interrupt(svr);
+ status = gpr_subprocess_join(svr);
+ gpr_subprocess_destroy(svr);
+ return status;
+}
diff --git a/test/core/memory_usage/server.c b/test/core/memory_usage/server.c
new file mode 100644
index 0000000000..c0710930b0
--- /dev/null
+++ b/test/core/memory_usage/server.c
@@ -0,0 +1,321 @@
+/*
+ *
+ * Copyright 2016, 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 <grpc/grpc.h>
+#include <grpc/grpc_security.h>
+
+#include <signal.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#ifndef _WIN32
+/* This is for _exit() below, which is temporary. */
+#include <unistd.h>
+#endif
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/cmdline.h>
+#include <grpc/support/host_port.h>
+#include <grpc/support/log.h>
+#include <grpc/support/time.h>
+#include "test/core/end2end/data/ssl_test_data.h"
+#include "test/core/util/memory_counters.h"
+#include "test/core/util/port.h"
+#include "test/core/util/test_config.h"
+
+static grpc_completion_queue *cq;
+static grpc_server *server;
+static grpc_op metadata_ops[2];
+static grpc_op snapshot_ops[5];
+static grpc_op status_op;
+static int got_sigint = 0;
+static grpc_byte_buffer *payload_buffer = NULL;
+static grpc_byte_buffer *terminal_buffer = NULL;
+static int was_cancelled = 2;
+
+static void *tag(intptr_t t) { return (void *)t; }
+
+typedef enum {
+ FLING_SERVER_NEW_REQUEST = 1,
+ FLING_SERVER_SEND_INIT_METADATA,
+ FLING_SERVER_WAIT_FOR_DESTROY,
+ FLING_SERVER_SEND_STATUS_FLING_CALL,
+ FLING_SERVER_SEND_STATUS_SNAPSHOT,
+ FLING_SERVER_BATCH_SEND_STATUS_FLING_CALL
+} fling_server_tags;
+
+typedef struct {
+ fling_server_tags state;
+ grpc_call *call;
+ grpc_call_details call_details;
+ grpc_metadata_array request_metadata_recv;
+ grpc_metadata_array initial_metadata_send;
+} fling_call;
+
+// hold up to 10000 calls and 6 snaphost calls
+static fling_call calls[100006];
+
+static void request_call_unary(int call_idx) {
+ if (call_idx == (int)(sizeof(calls) / sizeof(fling_call))) {
+ gpr_log(GPR_INFO, "Used all call slots (10000) on server. Server exit.");
+ _exit(0);
+ }
+ grpc_metadata_array_init(&calls[call_idx].request_metadata_recv);
+ grpc_server_request_call(
+ server, &calls[call_idx].call, &calls[call_idx].call_details,
+ &calls[call_idx].request_metadata_recv, cq, cq, &calls[call_idx]);
+}
+
+static void send_initial_metadata_unary(void *tag) {
+ grpc_metadata_array_init(&(*(fling_call *)tag).initial_metadata_send);
+ metadata_ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
+ metadata_ops[0].data.send_initial_metadata.count = 0;
+
+ GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch((*(fling_call *)tag).call,
+ metadata_ops, 1, tag, NULL));
+}
+
+static void send_status(void *tag) {
+ status_op.op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ status_op.data.send_status_from_server.status = GRPC_STATUS_OK;
+ status_op.data.send_status_from_server.trailing_metadata_count = 0;
+ status_op.data.send_status_from_server.status_details = "";
+
+ GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch((*(fling_call *)tag).call,
+ &status_op, 1, tag, NULL));
+}
+
+static void send_snapshot(void *tag, struct grpc_memory_counters *snapshot) {
+ grpc_op *op;
+
+ grpc_slice snapshot_slice =
+ grpc_slice_new(snapshot, sizeof(*snapshot), gpr_free);
+ payload_buffer = grpc_raw_byte_buffer_create(&snapshot_slice, 1);
+ grpc_metadata_array_init(&(*(fling_call *)tag).initial_metadata_send);
+
+ op = snapshot_ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message = &terminal_buffer;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ if (payload_buffer == NULL) {
+ gpr_log(GPR_INFO, "NULL payload buffer !!!");
+ }
+ op->data.send_message = payload_buffer;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.status = GRPC_STATUS_OK;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status_details = "";
+ op++;
+ 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((*(fling_call *)tag).call, snapshot_ops,
+ (size_t)(op - snapshot_ops), tag, NULL));
+}
+/* We have some sort of deadlock, so let's not exit gracefully for now.
+ When that is resolved, please remove the #include <unistd.h> above. */
+static void sigint_handler(int x) { _exit(0); }
+
+int main(int argc, char **argv) {
+ grpc_memory_counters_init();
+ grpc_event ev;
+ char *addr_buf = NULL;
+ gpr_cmdline *cl;
+ int shutdown_started = 0;
+ int shutdown_finished = 0;
+
+ int secure = 0;
+ char *addr = NULL;
+
+ char *fake_argv[1];
+
+ GPR_ASSERT(argc >= 1);
+ fake_argv[0] = argv[0];
+ grpc_test_init(1, fake_argv);
+
+ grpc_init();
+ srand((unsigned)clock());
+
+ cl = gpr_cmdline_create("fling server");
+ gpr_cmdline_add_string(cl, "bind", "Bind host:port", &addr);
+ gpr_cmdline_add_flag(cl, "secure", "Run with security?", &secure);
+ gpr_cmdline_parse(cl, argc, argv);
+ gpr_cmdline_destroy(cl);
+
+ if (addr == NULL) {
+ gpr_join_host_port(&addr_buf, "::", grpc_pick_unused_port_or_die());
+ addr = addr_buf;
+ }
+ gpr_log(GPR_INFO, "creating server on: %s", addr);
+
+ cq = grpc_completion_queue_create(NULL);
+
+ struct grpc_memory_counters before_server_create =
+ grpc_memory_counters_snapshot();
+ if (secure) {
+ grpc_ssl_pem_key_cert_pair pem_key_cert_pair = {test_server1_key,
+ test_server1_cert};
+ grpc_server_credentials *ssl_creds = grpc_ssl_server_credentials_create(
+ NULL, &pem_key_cert_pair, 1, 0, NULL);
+ server = grpc_server_create(NULL, NULL);
+ GPR_ASSERT(grpc_server_add_secure_http2_port(server, addr, ssl_creds));
+ grpc_server_credentials_release(ssl_creds);
+ } else {
+ server = grpc_server_create(NULL, NULL);
+ GPR_ASSERT(grpc_server_add_insecure_http2_port(server, addr));
+ }
+
+ grpc_server_register_completion_queue(server, cq, NULL);
+ grpc_server_start(server);
+
+ struct grpc_memory_counters after_server_create =
+ grpc_memory_counters_snapshot();
+
+ gpr_free(addr_buf);
+ addr = addr_buf = NULL;
+
+ // initialize call instances
+ for (int i = 0; i < (int)(sizeof(calls) / sizeof(fling_call)); i++) {
+ grpc_call_details_init(&calls[i].call_details);
+ calls[i].state = FLING_SERVER_NEW_REQUEST;
+ }
+
+ int next_call_idx = 0;
+ struct grpc_memory_counters current_snapshot;
+
+ request_call_unary(next_call_idx);
+
+ signal(SIGINT, sigint_handler);
+
+ while (!shutdown_finished) {
+ if (got_sigint && !shutdown_started) {
+ gpr_log(GPR_INFO, "Shutting down due to SIGINT");
+ grpc_server_shutdown_and_notify(server, cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(
+ cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5), NULL)
+ .type == GRPC_OP_COMPLETE);
+ grpc_completion_queue_shutdown(cq);
+ shutdown_started = 1;
+ }
+ ev = grpc_completion_queue_next(
+ cq, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
+ gpr_time_from_micros(1000000, GPR_TIMESPAN)),
+ NULL);
+ fling_call *s = ev.tag;
+ switch (ev.type) {
+ case GRPC_OP_COMPLETE:
+ switch (s->state) {
+ case FLING_SERVER_NEW_REQUEST:
+ request_call_unary(++next_call_idx);
+ if (0 ==
+ strcmp(s->call_details.method, "/Reflector/reflectUnary")) {
+ s->state = FLING_SERVER_SEND_INIT_METADATA;
+ send_initial_metadata_unary(s);
+ } else if (0 == strcmp(s->call_details.method,
+ "Reflector/GetBeforeSvrCreation")) {
+ s->state = FLING_SERVER_SEND_STATUS_SNAPSHOT;
+ send_snapshot(s, &before_server_create);
+ } else if (0 == strcmp(s->call_details.method,
+ "Reflector/GetAfterSvrCreation")) {
+ s->state = FLING_SERVER_SEND_STATUS_SNAPSHOT;
+ send_snapshot(s, &after_server_create);
+ } else if (0 == strcmp(s->call_details.method,
+ "Reflector/SimpleSnapshot")) {
+ s->state = FLING_SERVER_SEND_STATUS_SNAPSHOT;
+ current_snapshot = grpc_memory_counters_snapshot();
+ send_snapshot(s, &current_snapshot);
+ } else if (0 == strcmp(s->call_details.method,
+ "Reflector/DestroyCalls")) {
+ s->state = FLING_SERVER_BATCH_SEND_STATUS_FLING_CALL;
+ current_snapshot = grpc_memory_counters_snapshot();
+ send_snapshot(s, &current_snapshot);
+ } else {
+ gpr_log(GPR_ERROR, "Wrong call method");
+ }
+ break;
+ case FLING_SERVER_SEND_INIT_METADATA:
+ s->state = FLING_SERVER_WAIT_FOR_DESTROY;
+ break;
+ case FLING_SERVER_WAIT_FOR_DESTROY:
+ break;
+ case FLING_SERVER_SEND_STATUS_FLING_CALL:
+ grpc_call_destroy(s->call);
+ grpc_call_details_destroy(&s->call_details);
+ grpc_metadata_array_destroy(&s->initial_metadata_send);
+ grpc_metadata_array_destroy(&s->request_metadata_recv);
+ break;
+ case FLING_SERVER_BATCH_SEND_STATUS_FLING_CALL:
+ for (int k = 0; k < (int)(sizeof(calls) / sizeof(fling_call));
+ ++k) {
+ if (calls[k].state == FLING_SERVER_WAIT_FOR_DESTROY) {
+ calls[k].state = FLING_SERVER_SEND_STATUS_FLING_CALL;
+ send_status(&calls[k]);
+ }
+ }
+ // no break here since we want to continue to case
+ // FLING_SERVER_SEND_STATUS_SNAPSHOT to destroy the snapshot call
+ case FLING_SERVER_SEND_STATUS_SNAPSHOT:
+ grpc_byte_buffer_destroy(payload_buffer);
+ grpc_byte_buffer_destroy(terminal_buffer);
+ grpc_call_destroy(s->call);
+ grpc_call_details_destroy(&s->call_details);
+ grpc_metadata_array_destroy(&s->initial_metadata_send);
+ grpc_metadata_array_destroy(&s->request_metadata_recv);
+ terminal_buffer = NULL;
+ payload_buffer = NULL;
+ break;
+ }
+ break;
+ case GRPC_QUEUE_SHUTDOWN:
+ GPR_ASSERT(shutdown_started);
+ shutdown_finished = 1;
+ break;
+ case GRPC_QUEUE_TIMEOUT:
+ break;
+ }
+ }
+
+ grpc_server_destroy(server);
+ grpc_completion_queue_destroy(cq);
+ grpc_shutdown();
+ grpc_memory_counters_destroy();
+ return 0;
+}
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-09 03:19:35 +0000