<commit_before>/*
 *
 * Copyright 2016 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 "test/core/end2end/fixtures/http_proxy_fixture.h"

#include "src/core/lib/iomgr/sockaddr.h"

#include <string.h>

#include <grpc/grpc.h>
#include <grpc/slice_buffer.h>
#include <grpc/support/alloc.h>
#include <grpc/support/atm.h>
#include <grpc/support/host_port.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include <grpc/support/sync.h>
#include <grpc/support/thd.h>
#include <grpc/support/useful.h>

#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/http/parser.h"
#include "src/core/lib/iomgr/closure.h"
#include "src/core/lib/iomgr/combiner.h"
#include "src/core/lib/iomgr/endpoint.h"
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/iomgr/pollset.h"
#include "src/core/lib/iomgr/pollset_set.h"
#include "src/core/lib/iomgr/resolve_address.h"
#include "src/core/lib/iomgr/sockaddr_utils.h"
#include "src/core/lib/iomgr/tcp_client.h"
#include "src/core/lib/iomgr/tcp_server.h"
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/slice/b64.h"
#include "src/core/lib/slice/slice_internal.h"
#include "src/core/lib/support/string.h"
#include "test/core/util/port.h"

struct grpc_end2end_http_proxy {
  char* proxy_name;
  gpr_thd_id thd;
  grpc_tcp_server* server;
  grpc_channel_args* channel_args;
  gpr_mu* mu;
  grpc_pollset* pollset;
  gpr_refcount users;

  grpc_combiner* combiner;
};

//
// Connection handling
//

// proxy_connection structure is only accessed in the closures which are all
// scheduled under the same combiner lock. So there is is no need for a mutex to
// protect this structure.
typedef struct proxy_connection {
  grpc_end2end_http_proxy* proxy;

  grpc_endpoint* client_endpoint;
  grpc_endpoint* server_endpoint;

  gpr_refcount refcount;

  grpc_pollset_set* pollset_set;

  // NOTE: All the closures execute under proxy->combiner lock. Which means
  // there will not be any data-races between the closures
  grpc_closure on_read_request_done;
  grpc_closure on_server_connect_done;
  grpc_closure on_write_response_done;
  grpc_closure on_client_read_done;
  grpc_closure on_client_write_done;
  grpc_closure on_server_read_done;
  grpc_closure on_server_write_done;

  bool client_read_failed : 1;
  bool client_write_failed : 1;
  bool client_shutdown : 1;
  bool server_read_failed : 1;
  bool server_write_failed : 1;
  bool server_shutdown : 1;

  grpc_slice_buffer client_read_buffer;
  grpc_slice_buffer client_deferred_write_buffer;
  bool client_is_writing;
  grpc_slice_buffer client_write_buffer;
  grpc_slice_buffer server_read_buffer;
  grpc_slice_buffer server_deferred_write_buffer;
  bool server_is_writing;
  grpc_slice_buffer server_write_buffer;

  grpc_http_parser http_parser;
  grpc_http_request http_request;
} proxy_connection;

static void proxy_connection_ref(proxy_connection* conn, const char* reason) {
  gpr_ref(&conn->refcount);
}

// Helper function to destroy the proxy connection.
static void proxy_connection_unref(grpc_exec_ctx* exec_ctx,
                                   proxy_connection* conn, const char* reason) {
  if (gpr_unref(&conn->refcount)) {
    gpr_log(GPR_DEBUG, "endpoints: %p %p", conn->client_endpoint,
            conn->server_endpoint);
    grpc_endpoint_destroy(exec_ctx, conn->client_endpoint);
    if (conn->server_endpoint != nullptr) {
      grpc_endpoint_destroy(exec_ctx, conn->server_endpoint);
    }
    grpc_pollset_set_destroy(exec_ctx, conn->pollset_set);
    grpc_slice_buffer_destroy_internal(exec_ctx, &conn->client_read_buffer);
    grpc_slice_buffer_destroy_internal(exec_ctx,
                                       &conn->client_deferred_write_buffer);
    grpc_slice_buffer_destroy_internal(exec_ctx, &conn->client_write_buffer);
    grpc_slice_buffer_destroy_internal(exec_ctx, &conn->server_read_buffer);
    grpc_slice_buffer_destroy_internal(exec_ctx,
                                       &conn->server_deferred_write_buffer);
    grpc_slice_buffer_destroy_internal(exec_ctx, &conn->server_write_buffer);
    grpc_http_parser_destroy(&conn->http_parser);
    grpc_http_request_destroy(&conn->http_request);
    gpr_unref(&conn->proxy->users);
    gpr_free(conn);
  }
}

enum failure_type {
  SETUP_FAILED,  // To be used before we start proxying.
  CLIENT_READ_FAILED,
  CLIENT_WRITE_FAILED,
  SERVER_READ_FAILED,
  SERVER_WRITE_FAILED,
};

// Helper function to shut down the proxy connection.
static void proxy_connection_failed(grpc_exec_ctx* exec_ctx,
                                    proxy_connection* conn,
                                    failure_type failure, const char* prefix,
                                    grpc_error* error) {
  gpr_log(GPR_INFO, "%s: %s", prefix, grpc_error_string(error));
  // Decide whether we should shut down the client and server.
  bool shutdown_client = false;
  bool shutdown_server = false;
  if (failure == SETUP_FAILED) {
    shutdown_client = true;
    shutdown_server = true;
  } else {
    if ((failure == CLIENT_READ_FAILED && conn->client_write_failed) ||
        (failure == CLIENT_WRITE_FAILED && conn->client_read_failed) ||
        (failure == SERVER_READ_FAILED && !conn->client_is_writing)) {
      shutdown_client = true;
    }
    if ((failure == SERVER_READ_FAILED && conn->server_write_failed) ||
        (failure == SERVER_WRITE_FAILED && conn->server_read_failed) ||
        (failure == CLIENT_READ_FAILED && !conn->server_is_writing)) {
      shutdown_server = true;
    }
  }
  // If we decided to shut down either one and have not yet done so, do so.
  if (shutdown_client && !conn->client_shutdown) {
    grpc_endpoint_shutdown(exec_ctx, conn->client_endpoint,
                           GRPC_ERROR_REF(error));
    conn->client_shutdown = true;
  }
  if (shutdown_server && !conn->server_shutdown) {
    grpc_endpoint_shutdown(exec_ctx, conn->server_endpoint,
                           GRPC_ERROR_REF(error));
    conn->server_shutdown = true;
  }
  // Unref the connection.
  proxy_connection_unref(exec_ctx, conn, "conn_failed");
  GRPC_ERROR_UNREF(error);
}

// Callback for writing proxy data to the client.
static void on_client_write_done(grpc_exec_ctx* exec_ctx, void* arg,
                                 grpc_error* error) {
  proxy_connection* conn = (proxy_connection*)arg;
  conn->client_is_writing = false;
  if (error != GRPC_ERROR_NONE) {
    proxy_connection_failed(exec_ctx, conn, CLIENT_WRITE_FAILED,
                            "HTTP proxy client write", GRPC_ERROR_REF(error));
    return;
  }
  // Clear write buffer (the data we just wrote).
  grpc_slice_buffer_reset_and_unref(&conn->client_write_buffer);
  // If more data was read from the server since we started this write,
  // write that data now.
  if (conn->client_deferred_write_buffer.length > 0) {
    grpc_slice_buffer_move_into(&conn->client_deferred_write_buffer,
                                &conn->client_write_buffer);
    conn->client_is_writing = true;
    grpc_endpoint_write(exec_ctx, conn->client_endpoint,
                        &conn->client_write_buffer,
                        &conn->on_client_write_done);
  } else {
    // No more writes.  Unref the connection.
    proxy_connection_unref(exec_ctx, conn, "write_done");
  }
}

// Callback for writing proxy data to the backend server.
static void on_server_write_done(grpc_exec_ctx* exec_ctx, void* arg,
                                 grpc_error* error) {
  proxy_connection* conn = (proxy_connection*)arg;
  conn->server_is_writing = false;
  if (error != GRPC_ERROR_NONE) {
    proxy_connection_failed(exec_ctx, conn, SERVER_WRITE_FAILED,
                            "HTTP proxy server write", GRPC_ERROR_REF(error));
    return;
  }
  // Clear write buffer (the data we just wrote).
  grpc_slice_buffer_reset_and_unref(&conn->server_write_buffer);
  // If more data was read from the client since we started this write,
  // write that data now.
  if (conn->server_deferred_write_buffer.length > 0) {
    grpc_slice_buffer_move_into(&conn->server_deferred_write_buffer,
                                &conn->server_write_buffer);
    conn->server_is_writing = true;
    grpc_endpoint_write(exec_ctx, conn->server_endpoint,
                        &conn->server_write_buffer,
                        &conn->on_server_write_done);
  } else {
    // No more writes.  Unref the connection.
    proxy_connection_unref(exec_ctx, conn, "server_write");
  }
}

// Callback for reading data from the client, which will be proxied to
// the backend server.
static void on_client_read_done(grpc_exec_ctx* exec_ctx, void* arg,
                                grpc_error* error) {
  proxy_connection* conn = (proxy_connection*)arg;
  if (error != GRPC_ERROR_NONE) {
    // Report a read failure on the client endpoint. If there is no pending
    // server write, then shutdown the server endpoint as well.
    proxy_connection_failed(exec_ctx, conn, CLIENT_READ_FAILED,
                            "HTTP proxy client read", GRPC_ERROR_REF(error));
    return;
  }
  // If there is already a pending write (i.e., server_write_buffer is
  // not empty), then move the read data into server_deferred_write_buffer,
  // and the next write will be requested in on_server_write_done(), when
  // the current write is finished.
  //
  // Otherwise, move the read data into the write buffer and write it.
  if (conn->server_is_writing) {
    grpc_slice_buffer_move_into(&conn->client_read_buffer,
                                &conn->server_deferred_write_buffer);
  } else {
    grpc_slice_buffer_move_into(&conn->client_read_buffer,
                                &conn->server_write_buffer);
    proxy_connection_ref(conn, "client_read");
    conn->server_is_writing = true;
    grpc_endpoint_write(exec_ctx, conn->server_endpoint,
                        &conn->server_write_buffer,
                        &conn->on_server_write_done);
  }
  // Read more data.
  grpc_endpoint_read(exec_ctx, conn->client_endpoint, &conn->client_read_buffer,
                     &conn->on_client_read_done);
}

// Callback for reading data from the backend server, which will be
// proxied to the client.
static void on_server_read_done(grpc_exec_ctx* exec_ctx, void* arg,
                                grpc_error* error) {
  proxy_connection* conn = (proxy_connection*)arg;
  if (error != GRPC_ERROR_NONE) {
    // Report a read failure on the server end point. If there is no pending
    // write to the client, then shutdown the client endpoint as well.
    proxy_connection_failed(exec_ctx, conn, SERVER_READ_FAILED,
                            "HTTP proxy server read", GRPC_ERROR_REF(error));
    return;
  }
  // If there is already a pending write (i.e., client_write_buffer is
  // not empty), then move the read data into client_deferred_write_buffer,
  // and the next write will be requested in on_client_write_done(), when
  // the current write is finished.
  //
  // Otherwise, move the read data into the write buffer and write it.
  if (conn->client_is_writing) {
    grpc_slice_buffer_move_into(&conn->server_read_buffer,
                                &conn->client_deferred_write_buffer);
  } else {
    grpc_slice_buffer_move_into(&conn->server_read_buffer,
                                &conn->client_write_buffer);
    proxy_connection_ref(conn, "server_read");
    conn->client_is_writing = true;
    grpc_endpoint_write(exec_ctx, conn->client_endpoint,
                        &conn->client_write_buffer,
                        &conn->on_client_write_done);
  }
  // Read more data.
  grpc_endpoint_read(exec_ctx, conn->server_endpoint, &conn->server_read_buffer,
                     &conn->on_server_read_done);
}
<commit_msg>Delete outdated comments<commit_after>/*
 *
 * Copyright 2016 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 "test/core/end2end/fixtures/http_proxy_fixture.h"

#include "src/core/lib/iomgr/sockaddr.h"

#include <string.h>

#include <grpc/grpc.h>
#include <grpc/slice_buffer.h>
#include <grpc/support/alloc.h>
#include <grpc/support/atm.h>
#include <grpc/support/host_port.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include <grpc/support/sync.h>
#include <grpc/support/thd.h>
#include <grpc/support/useful.h>

#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/http/parser.h"
#include "src/core/lib/iomgr/closure.h"
#include "src/core/lib/iomgr/combiner.h"
#include "src/core/lib/iomgr/endpoint.h"
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/iomgr/pollset.h"
#include "src/core/lib/iomgr/pollset_set.h"
#include "src/core/lib/iomgr/resolve_address.h"
#include "src/core/lib/iomgr/sockaddr_utils.h"
#include "src/core/lib/iomgr/tcp_client.h"
#include "src/core/lib/iomgr/tcp_server.h"
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/slice/b64.h"
#include "src/core/lib/slice/slice_internal.h"
#include "src/core/lib/support/string.h"
#include "test/core/util/port.h"

struct grpc_end2end_http_proxy {
  char* proxy_name;
  gpr_thd_id thd;
  grpc_tcp_server* server;
  grpc_channel_args* channel_args;
  gpr_mu* mu;
  grpc_pollset* pollset;
  gpr_refcount users;

  grpc_combiner* combiner;
};

//
// Connection handling
//

// proxy_connection structure is only accessed in the closures which are all
// scheduled under the same combiner lock. So there is is no need for a mutex to
// protect this structure.
typedef struct proxy_connection {
  grpc_end2end_http_proxy* proxy;

  grpc_endpoint* client_endpoint;
  grpc_endpoint* server_endpoint;

  gpr_refcount refcount;

  grpc_pollset_set* pollset_set;

  // NOTE: All the closures execute under proxy->combiner lock. Which means
  // there will not be any data-races between the closures
  grpc_closure on_read_request_done;
  grpc_closure on_server_connect_done;
  grpc_closure on_write_response_done;
  grpc_closure on_client_read_done;
  grpc_closure on_client_write_done;
  grpc_closure on_server_read_done;
  grpc_closure on_server_write_done;

  bool client_read_failed : 1;
  bool client_write_failed : 1;
  bool client_shutdown : 1;
  bool server_read_failed : 1;
  bool server_write_failed : 1;
  bool server_shutdown : 1;

  grpc_slice_buffer client_read_buffer;
  grpc_slice_buffer client_deferred_write_buffer;
  bool client_is_writing;
  grpc_slice_buffer client_write_buffer;
  grpc_slice_buffer server_read_buffer;
  grpc_slice_buffer server_deferred_write_buffer;
  bool server_is_writing;
  grpc_slice_buffer server_write_buffer;

  grpc_http_parser http_parser;
  grpc_http_request http_request;
} proxy_connection;

static void proxy_connection_ref(proxy_connection* conn, const char* reason) {
  gpr_ref(&conn->refcount);
}

// Helper function to destroy the proxy connection.
static void proxy_connection_unref(grpc_exec_ctx* exec_ctx,
                                   proxy_connection* conn, const char* reason) {
  if (gpr_unref(&conn->refcount)) {
    gpr_log(GPR_DEBUG, "endpoints: %p %p", conn->client_endpoint,
            conn->server_endpoint);
    grpc_endpoint_destroy(exec_ctx, conn->client_endpoint);
    if (conn->server_endpoint != nullptr) {
      grpc_endpoint_destroy(exec_ctx, conn->server_endpoint);
    }
    grpc_pollset_set_destroy(exec_ctx, conn->pollset_set);
    grpc_slice_buffer_destroy_internal(exec_ctx, &conn->client_read_buffer);
    grpc_slice_buffer_destroy_internal(exec_ctx,
                                       &conn->client_deferred_write_buffer);
    grpc_slice_buffer_destroy_internal(exec_ctx, &conn->client_write_buffer);
    grpc_slice_buffer_destroy_internal(exec_ctx, &conn->server_read_buffer);
    grpc_slice_buffer_destroy_internal(exec_ctx,
                                       &conn->server_deferred_write_buffer);
    grpc_slice_buffer_destroy_internal(exec_ctx, &conn->server_write_buffer);
    grpc_http_parser_destroy(&conn->http_parser);
    grpc_http_request_destroy(&conn->http_request);
    gpr_unref(&conn->proxy->users);
    gpr_free(conn);
  }
}

enum failure_type {
  SETUP_FAILED,  // To be used before we start proxying.
  CLIENT_READ_FAILED,
  CLIENT_WRITE_FAILED,
  SERVER_READ_FAILED,
  SERVER_WRITE_FAILED,
};

// Helper function to shut down the proxy connection.
static void proxy_connection_failed(grpc_exec_ctx* exec_ctx,
                                    proxy_connection* conn,
                                    failure_type failure, const char* prefix,
                                    grpc_error* error) {
  gpr_log(GPR_INFO, "%s: %s", prefix, grpc_error_string(error));
  // Decide whether we should shut down the client and server.
  bool shutdown_client = false;
  bool shutdown_server = false;
  if (failure == SETUP_FAILED) {
    shutdown_client = true;
    shutdown_server = true;
  } else {
    if ((failure == CLIENT_READ_FAILED && conn->client_write_failed) ||
        (failure == CLIENT_WRITE_FAILED && conn->client_read_failed) ||
        (failure == SERVER_READ_FAILED && !conn->client_is_writing)) {
      shutdown_client = true;
    }
    if ((failure == SERVER_READ_FAILED && conn->server_write_failed) ||
        (failure == SERVER_WRITE_FAILED && conn->server_read_failed) ||
        (failure == CLIENT_READ_FAILED && !conn->server_is_writing)) {
      shutdown_server = true;
    }
  }
  // If we decided to shut down either one and have not yet done so, do so.
  if (shutdown_client && !conn->client_shutdown) {
    grpc_endpoint_shutdown(exec_ctx, conn->client_endpoint,
                           GRPC_ERROR_REF(error));
    conn->client_shutdown = true;
  }
  if (shutdown_server && !conn->server_shutdown) {
    grpc_endpoint_shutdown(exec_ctx, conn->server_endpoint,
                           GRPC_ERROR_REF(error));
    conn->server_shutdown = true;
  }
  // Unref the connection.
  proxy_connection_unref(exec_ctx, conn, "conn_failed");
  GRPC_ERROR_UNREF(error);
}

// Callback for writing proxy data to the client.
static void on_client_write_done(grpc_exec_ctx* exec_ctx, void* arg,
                                 grpc_error* error) {
  proxy_connection* conn = (proxy_connection*)arg;
  conn->client_is_writing = false;
  if (error != GRPC_ERROR_NONE) {
    proxy_connection_failed(exec_ctx, conn, CLIENT_WRITE_FAILED,
                            "HTTP proxy client write", GRPC_ERROR_REF(error));
    return;
  }
  // Clear write buffer (the data we just wrote).
  grpc_slice_buffer_reset_and_unref(&conn->client_write_buffer);
  // If more data was read from the server since we started this write,
  // write that data now.
  if (conn->client_deferred_write_buffer.length > 0) {
    grpc_slice_buffer_move_into(&conn->client_deferred_write_buffer,
                                &conn->client_write_buffer);
    conn->client_is_writing = true;
    grpc_endpoint_write(exec_ctx, conn->client_endpoint,
                        &conn->client_write_buffer,
                        &conn->on_client_write_done);
  } else {
    // No more writes.  Unref the connection.
    proxy_connection_unref(exec_ctx, conn, "write_done");
  }
}

// Callback for writing proxy data to the backend server.
static void on_server_write_done(grpc_exec_ctx* exec_ctx, void* arg,
                                 grpc_error* error) {
  proxy_connection* conn = (proxy_connection*)arg;
  conn->server_is_writing = false;
  if (error != GRPC_ERROR_NONE) {
    proxy_connection_failed(exec_ctx, conn, SERVER_WRITE_FAILED,
                            "HTTP proxy server write", GRPC_ERROR_REF(error));
    return;
  }
  // Clear write buffer (the data we just wrote).
  grpc_slice_buffer_reset_and_unref(&conn->server_write_buffer);
  // If more data was read from the client since we started this write,
  // write that data now.
  if (conn->server_deferred_write_buffer.length > 0) {
    grpc_slice_buffer_move_into(&conn->server_deferred_write_buffer,
                                &conn->server_write_buffer);
    conn->server_is_writing = true;
    grpc_endpoint_write(exec_ctx, conn->server_endpoint,
                        &conn->server_write_buffer,
                        &conn->on_server_write_done);
  } else {
    // No more writes.  Unref the connection.
    proxy_connection_unref(exec_ctx, conn, "server_write");
  }
}

// Callback for reading data from the client, which will be proxied to
// the backend server.
static void on_client_read_done(grpc_exec_ctx* exec_ctx, void* arg,
                                grpc_error* error) {
  proxy_connection* conn = (proxy_connection*)arg;
  if (error != GRPC_ERROR_NONE) {
    proxy_connection_failed(exec_ctx, conn, CLIENT_READ_FAILED,
                            "HTTP proxy client read", GRPC_ERROR_REF(error));
    return;
  }
  // If there is already a pending write (i.e., server_write_buffer is
  // not empty), then move the read data into server_deferred_write_buffer,
  // and the next write will be requested in on_server_write_done(), when
  // the current write is finished.
  //
  // Otherwise, move the read data into the write buffer and write it.
  if (conn->server_is_writing) {
    grpc_slice_buffer_move_into(&conn->client_read_buffer,
                                &conn->server_deferred_write_buffer);
  } else {
    grpc_slice_buffer_move_into(&conn->client_read_buffer,
                                &conn->server_write_buffer);
    proxy_connection_ref(conn, "client_read");
    conn->server_is_writing = true;
    grpc_endpoint_write(exec_ctx, conn->server_endpoint,
                        &conn->server_write_buffer,
                        &conn->on_server_write_done);
  }
  // Read more data.
  grpc_endpoint_read(exec_ctx, conn->client_endpoint, &conn->client_read_buffer,
                     &conn->on_client_read_done);
}

// Callback for reading data from the backend server, which will be
// proxied to the client.
static void on_server_read_done(grpc_exec_ctx* exec_ctx, void* arg,
                                grpc_error* error) {
  proxy_connection* conn = (proxy_connection*)arg;
  if (error != GRPC_ERROR_NONE) {
    proxy_connection_failed(exec_ctx, conn, SERVER_READ_FAILED,
                            "HTTP proxy server read", GRPC_ERROR_REF(error));
    return;
  }
  // If there is already a pending write (i.e., client_write_buffer is
  // not empty), then move the read data into client_deferred_write_buffer,
  // and the next write will be requested in on_client_write_done(), when
  // the current write is finished.
  //
  // Otherwise, move the read data into the write buffer and write it.
  if (conn->client_is_writing) {
    grpc_slice_buffer_move_into(&conn->server_read_buffer,
                                &conn->client_deferred_write_buffer);
  } else {
    grpc_slice_buffer_move_into(&conn->server_read_buffer,
                                &conn->client_write_buffer);
    proxy_connection_ref(conn, "server_read");
    conn->client_is_writing = true;
    grpc_endpoint_write(exec_ctx, conn->client_endpoint,
                        &conn->client_write_buffer,
                        &conn->on_client_write_done);
  }
  // Read more data.
  grpc_endpoint_read(exec_ctx, conn->server_endpoint, &conn->server_read_buffer,
                     &conn->on_server_read_done);
}

{'fraction_non_alphanumeric': 0.0954413031882155, 'fraction_numerical': 0.001723359730827623, 'mean_word_length': 3.2519190509420794, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 175, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7052308', 'n_tokens_mistral': 7795, 'n_tokens_neox': 7727, 'n_words': 1723}