<commit_before>/*
 *
 * 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 <grpc/grpc.h>
#include <grpc/grpc_security.h>

#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include "test/core/util/grpc_profiler.h"
#include "test/core/util/test_config.h"
#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/util/port.h"
#include "test/core/end2end/data/ssl_test_data.h"

static grpc_completion_queue *cq;
static grpc_server *server;
static grpc_call *call;
static grpc_call_details call_details;
static grpc_metadata_array request_metadata_recv;
static grpc_metadata_array initial_metadata_send;
static grpc_byte_buffer *payload_buffer = NULL;
/* Used to drain the terminal read in unary calls. */
static grpc_byte_buffer *terminal_buffer = NULL;

static grpc_op read_op;
static grpc_op metadata_send_op;
static grpc_op write_op;
static grpc_op status_op[2];
static int was_cancelled = 2;
static grpc_op unary_ops[6];
static int got_sigint = 0;

static void *tag(gpr_intptr t) { return (void *)t; }

typedef enum {
  FLING_SERVER_NEW_REQUEST = 1,
  FLING_SERVER_READ_FOR_UNARY,
  FLING_SERVER_BATCH_OPS_FOR_UNARY,
  FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING,
  FLING_SERVER_READ_FOR_STREAMING,
  FLING_SERVER_WRITE_FOR_STREAMING,
  FLING_SERVER_SEND_STATUS_FOR_STREAMING
} fling_server_tags;

typedef struct {
  gpr_refcount pending_ops;
  gpr_uint32 flags;
} call_state;

static void request_call(void) {
  grpc_metadata_array_init(&request_metadata_recv);
  grpc_server_request_call(server, &call, &call_details, &request_metadata_recv,
                           cq, tag(FLING_SERVER_NEW_REQUEST));
}

static void handle_unary_method(void) {
  grpc_op *op;

  grpc_metadata_array_init(&initial_metadata_send);

  op = unary_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(call, unary_ops, op - unary_ops,
                                   tag(FLING_SERVER_BATCH_OPS_FOR_UNARY)));
}

static void send_initial_metadata(void) {
  grpc_metadata_array_init(&initial_metadata_send);
  metadata_send_op.op = GRPC_OP_SEND_INITIAL_METADATA;
  metadata_send_op.data.send_initial_metadata.count = 0;
  GPR_ASSERT(GRPC_CALL_OK ==
             grpc_call_start_batch(
                 call, &metadata_send_op, 1,
                 tag(FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING)));
}

static void start_read_op(int t) {
  /* Starting read at server */
  read_op.op = GRPC_OP_RECV_MESSAGE;
  read_op.data.recv_message = &payload_buffer;
  GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call, &read_op, 1, tag(t)));
}

static void start_write_op(void) {
  /* Starting write at server */
  write_op.op = GRPC_OP_SEND_MESSAGE;
  if (payload_buffer == NULL) {
    gpr_log(GPR_INFO, "NULL payload buffer !!!");
  }
  write_op.data.send_message = payload_buffer;
  GPR_ASSERT(GRPC_CALL_OK ==
             grpc_call_start_batch(call, &write_op, 1,
                                   tag(FLING_SERVER_WRITE_FOR_STREAMING)));
}

static void start_send_status(void) {
  status_op[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER;
  status_op[0].data.send_status_from_server.status = GRPC_STATUS_OK;
  status_op[0].data.send_status_from_server.trailing_metadata_count = 0;
  status_op[0].data.send_status_from_server.status_details = "";
  status_op[1].op = GRPC_OP_RECV_CLOSE_ON_SERVER;
  status_op[1].data.recv_close_on_server.cancelled = &was_cancelled;

  GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(
                                 call, status_op, 2,
                                 tag(FLING_SERVER_SEND_STATUS_FOR_STREAMING)));
}

static void sigint_handler(int x) { got_sigint = 1; }

int main(int argc, char **argv) {
  grpc_event *ev;
  call_state *s;
  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);
<commit_msg>Stopgap fix for fling_stream_test timing out<commit_after>/*
 *
 * 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 <grpc/grpc.h>
#include <grpc/grpc_security.h>

#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include "test/core/util/grpc_profiler.h"
#include "test/core/util/test_config.h"
#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/util/port.h"
#include "test/core/end2end/data/ssl_test_data.h"

static grpc_completion_queue *cq;
static grpc_server *server;
static grpc_call *call;
static grpc_call_details call_details;
static grpc_metadata_array request_metadata_recv;
static grpc_metadata_array initial_metadata_send;
static grpc_byte_buffer *payload_buffer = NULL;
/* Used to drain the terminal read in unary calls. */
static grpc_byte_buffer *terminal_buffer = NULL;

static grpc_op read_op;
static grpc_op metadata_send_op;
static grpc_op write_op;
static grpc_op status_op[2];
static int was_cancelled = 2;
static grpc_op unary_ops[6];
static int got_sigint = 0;

static void *tag(gpr_intptr t) { return (void *)t; }

typedef enum {
  FLING_SERVER_NEW_REQUEST = 1,
  FLING_SERVER_READ_FOR_UNARY,
  FLING_SERVER_BATCH_OPS_FOR_UNARY,
  FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING,
  FLING_SERVER_READ_FOR_STREAMING,
  FLING_SERVER_WRITE_FOR_STREAMING,
  FLING_SERVER_SEND_STATUS_FOR_STREAMING
} fling_server_tags;

typedef struct {
  gpr_refcount pending_ops;
  gpr_uint32 flags;
} call_state;

static void request_call(void) {
  grpc_metadata_array_init(&request_metadata_recv);
  grpc_server_request_call(server, &call, &call_details, &request_metadata_recv,
                           cq, tag(FLING_SERVER_NEW_REQUEST));
}

static void handle_unary_method(void) {
  grpc_op *op;

  grpc_metadata_array_init(&initial_metadata_send);

  op = unary_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(call, unary_ops, op - unary_ops,
                                   tag(FLING_SERVER_BATCH_OPS_FOR_UNARY)));
}

static void send_initial_metadata(void) {
  grpc_metadata_array_init(&initial_metadata_send);
  metadata_send_op.op = GRPC_OP_SEND_INITIAL_METADATA;
  metadata_send_op.data.send_initial_metadata.count = 0;
  GPR_ASSERT(GRPC_CALL_OK ==
             grpc_call_start_batch(
                 call, &metadata_send_op, 1,
                 tag(FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING)));
}

static void start_read_op(int t) {
  /* Starting read at server */
  read_op.op = GRPC_OP_RECV_MESSAGE;
  read_op.data.recv_message = &payload_buffer;
  GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call, &read_op, 1, tag(t)));
}

static void start_write_op(void) {
  /* Starting write at server */
  write_op.op = GRPC_OP_SEND_MESSAGE;
  if (payload_buffer == NULL) {
    gpr_log(GPR_INFO, "NULL payload buffer !!!");
  }
  write_op.data.send_message = payload_buffer;
  GPR_ASSERT(GRPC_CALL_OK ==
             grpc_call_start_batch(call, &write_op, 1,
                                   tag(FLING_SERVER_WRITE_FOR_STREAMING)));
}

static void start_send_status(void) {
  status_op[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER;
  status_op[0].data.send_status_from_server.status = GRPC_STATUS_OK;
  status_op[0].data.send_status_from_server.trailing_metadata_count = 0;
  status_op[0].data.send_status_from_server.status_details = "";
  status_op[1].op = GRPC_OP_RECV_CLOSE_ON_SERVER;
  status_op[1].data.recv_close_on_server.cancelled = &was_cancelled;

  GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(
                                 call, status_op, 2,
                                 tag(FLING_SERVER_SEND_STATUS_FOR_STREAMING)));
}

static void sigint_handler(int x) { _exit(0); }

int main(int argc, char **argv) {
  grpc_event *ev;
  call_state *s;
  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);

{'fraction_non_alphanumeric': 0.0906508875739645, 'fraction_numerical': 0.005522682445759369, 'mean_word_length': 4.378022910479423, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 54, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2817308', 'n_tokens_mistral': 4484, 'n_tokens_neox': 4311, 'n_words': 1022}