<commit_before>// Copyright 2010-2014 RethinkDB, all rights reserved.
#include "protob/protob.hpp"

#include <google/protobuf/stubs/common.h>

#include <set>
#include <string>
#include <limits>

#include "errors.hpp"
#include <boost/lexical_cast.hpp>

#include "arch/arch.hpp"
#include "arch/io/network.hpp"
#include "clustering/administration/metadata.hpp"
#include "concurrency/cross_thread_signal.hpp"
#include "containers/auth_key.hpp"
#include "protob/json_shim.hpp"
#include "rdb_protocol/env.hpp"
#include "rpc/semilattice/joins/vclock.hpp"
#include "rpc/semilattice/view.hpp"
#include "utils.hpp"

#include "rdb_protocol/ql2.pb.h"
#include "rdb_protocol/query_server.hpp"

struct protob_server_exc_t : public std::exception {
public:
    explicit protob_server_exc_t(const std::string& data) : info(data) { }
    ~protob_server_exc_t() throw () { }
    const char *what() const throw () { return info.c_str(); }
private:
    std::string info;
};

const uint32_t MAX_QUERY_SIZE = 64 * MEGABYTE;
const size_t MAX_RESPONSE_SIZE = std::numeric_limits<uint32_t>::max();

class json_protocol_t {
public:
    static bool parse_query(tcp_conn_t *conn,
                            signal_t *interruptor,
                            query_handler_t *handler,
                            ql::protob_t<Query> *query_out) {
        int64_t token;
        uint32_t size;
        conn->read(&token, sizeof(token), interruptor);
        conn->read(&size, sizeof(size), interruptor);

        if (size > MAX_QUERY_SIZE) {
            Response error_response;
            handler->unparseable_query(token, &error_response,
                                       strprintf("Payload size (%" PRIu32 ") greater than maximum (%" PRIu32 ").",
                                                 size, MAX_QUERY_SIZE));
            send_response(error_response, handler, conn, interruptor);
            throw tcp_conn_read_closed_exc_t();
        } else {
            scoped_array_t<char> data(size + 1);
            conn->read(data.data(), size, interruptor);
            data[size] = 0; // Null terminate the string, which the json parser requires

            if (!json_shim::parse_json_pb(query_out->get(), token, data.data())) {
                Response error_response;
                handler->unparseable_query(token, &error_response,
                                           "Client is buggy (failed to deserialize query).");
                send_response(error_response, handler, conn, interruptor);
                return false;
            }
        }
        return true;
    }

    static void send_response(const Response &response,
                              query_handler_t *handler,
                              tcp_conn_t *conn,
                              signal_t *interruptor) {
        int64_t token = response.token();
        uint32_t size;
        std::string str;

        json_shim::write_json_pb(response, &str);
        if (str.size() > MAX_RESPONSE_SIZE) {
            Response error_response;
            handler->unparseable_query(token, &error_response,
                strprintf("Response size (%zu) is greater than maximum (%zu).",
                          str.size(), MAX_RESPONSE_SIZE));
            send_response(error_response, handler, conn, interruptor);
            return;
        }
        size = str.size();

        conn->write(&token, sizeof(token), interruptor);
        conn->write(&size, sizeof(size), interruptor);
        conn->write(str.data(), size, interruptor);
    }
};

class protobuf_protocol_t {
public:
    static bool parse_query(tcp_conn_t *conn,
                            signal_t *interruptor,
                            query_handler_t *handler,
                            ql::protob_t<Query> *query_out) {
        uint32_t size;
        conn->read(&size, sizeof(size), interruptor);

        if (size > MAX_QUERY_SIZE) {
            Response error_response;
            handler->unparseable_query(0, &error_response,
                                       strprintf("Payload size (%" PRIu32 ") greater than maximum (%" PRIu32 ").",
                                                 size, MAX_QUERY_SIZE));
            send_response(error_response, handler, conn, interruptor);
            return false;
        } else {
            scoped_array_t<char> data(size);
            conn->read(data.data(), size, interruptor);

            if (!query_out->get()->ParseFromArray(data.data(), size)) {
                Response error_response;
                int64_t token = query_out->get()->has_token() ? query_out->get()->token() : 0;
                handler->unparseable_query(token, &error_response,
                                           "Client is buggy (failed to deserialize query).");
                send_response(error_response, handler, conn, interruptor);
                return false;
            }
        }
        return true;
    }

    static void send_response(const Response &response,
                              query_handler_t *handler,
                              tcp_conn_t *conn,
                              signal_t *interruptor) {
        scoped_array_t<char> scoped_array;
        const char *data;
        uint32_t size;

        if (static_cast<uint64_t>(response.ByteSize()) > MAX_RESPONSE_SIZE) {
            Response error_response;
            handler->unparseable_query(response.token(), &error_response,
                strprintf("Response size (%d) is greater than maximum (%zu).",
                          response.ByteSize(), MAX_RESPONSE_SIZE));
            send_response(error_response, handler, conn, interruptor);
            return;
        }
        size = response.ByteSize();
        scoped_array.init(size);
        response.SerializeToArray(scoped_array.data(), size);
        data = scoped_array.data();
        conn->write(&size, sizeof(size), interruptor);
        conn->write(data, size, interruptor);
    }
};

query_server_t::query_server_t(rdb_context_t *_rdb_ctx,
                               const std::set<ip_address_t> &local_addresses,
                               int port,
                               query_handler_t *_handler,
                               boost::shared_ptr<semilattice_readwrite_view_t<auth_semilattice_metadata_t> > _auth_metadata) :
        rdb_ctx(_rdb_ctx),
        handler(_handler),
        auth_metadata(_auth_metadata),
        shutting_down_conds(get_num_threads()),
        pulse_sdc_on_shutdown(&main_shutting_down_cond),
        next_thread(0)
{
    rassert(rdb_ctx != NULL);
    for (int i = 0; i < get_num_threads(); ++i) {
        cross_thread_signal_t *s =
            new cross_thread_signal_t(&main_shutting_down_cond, threadnum_t(i));
        shutting_down_conds.push_back(s);
        rassert(s == &shutting_down_conds[i]);<commit_msg>Removed reserve parameter of shutting_down_conds construction in protob.cc.<commit_after>// Copyright 2010-2014 RethinkDB, all rights reserved.
#include "protob/protob.hpp"

#include <google/protobuf/stubs/common.h>

#include <set>
#include <string>
#include <limits>

#include "errors.hpp"
#include <boost/lexical_cast.hpp>

#include "arch/arch.hpp"
#include "arch/io/network.hpp"
#include "clustering/administration/metadata.hpp"
#include "concurrency/cross_thread_signal.hpp"
#include "containers/auth_key.hpp"
#include "protob/json_shim.hpp"
#include "rdb_protocol/env.hpp"
#include "rpc/semilattice/joins/vclock.hpp"
#include "rpc/semilattice/view.hpp"
#include "utils.hpp"

#include "rdb_protocol/ql2.pb.h"
#include "rdb_protocol/query_server.hpp"

struct protob_server_exc_t : public std::exception {
public:
    explicit protob_server_exc_t(const std::string& data) : info(data) { }
    ~protob_server_exc_t() throw () { }
    const char *what() const throw () { return info.c_str(); }
private:
    std::string info;
};

const uint32_t MAX_QUERY_SIZE = 64 * MEGABYTE;
const size_t MAX_RESPONSE_SIZE = std::numeric_limits<uint32_t>::max();

class json_protocol_t {
public:
    static bool parse_query(tcp_conn_t *conn,
                            signal_t *interruptor,
                            query_handler_t *handler,
                            ql::protob_t<Query> *query_out) {
        int64_t token;
        uint32_t size;
        conn->read(&token, sizeof(token), interruptor);
        conn->read(&size, sizeof(size), interruptor);

        if (size > MAX_QUERY_SIZE) {
            Response error_response;
            handler->unparseable_query(token, &error_response,
                                       strprintf("Payload size (%" PRIu32 ") greater than maximum (%" PRIu32 ").",
                                                 size, MAX_QUERY_SIZE));
            send_response(error_response, handler, conn, interruptor);
            throw tcp_conn_read_closed_exc_t();
        } else {
            scoped_array_t<char> data(size + 1);
            conn->read(data.data(), size, interruptor);
            data[size] = 0; // Null terminate the string, which the json parser requires

            if (!json_shim::parse_json_pb(query_out->get(), token, data.data())) {
                Response error_response;
                handler->unparseable_query(token, &error_response,
                                           "Client is buggy (failed to deserialize query).");
                send_response(error_response, handler, conn, interruptor);
                return false;
            }
        }
        return true;
    }

    static void send_response(const Response &response,
                              query_handler_t *handler,
                              tcp_conn_t *conn,
                              signal_t *interruptor) {
        int64_t token = response.token();
        uint32_t size;
        std::string str;

        json_shim::write_json_pb(response, &str);
        if (str.size() > MAX_RESPONSE_SIZE) {
            Response error_response;
            handler->unparseable_query(token, &error_response,
                strprintf("Response size (%zu) is greater than maximum (%zu).",
                          str.size(), MAX_RESPONSE_SIZE));
            send_response(error_response, handler, conn, interruptor);
            return;
        }
        size = str.size();

        conn->write(&token, sizeof(token), interruptor);
        conn->write(&size, sizeof(size), interruptor);
        conn->write(str.data(), size, interruptor);
    }
};

class protobuf_protocol_t {
public:
    static bool parse_query(tcp_conn_t *conn,
                            signal_t *interruptor,
                            query_handler_t *handler,
                            ql::protob_t<Query> *query_out) {
        uint32_t size;
        conn->read(&size, sizeof(size), interruptor);

        if (size > MAX_QUERY_SIZE) {
            Response error_response;
            handler->unparseable_query(0, &error_response,
                                       strprintf("Payload size (%" PRIu32 ") greater than maximum (%" PRIu32 ").",
                                                 size, MAX_QUERY_SIZE));
            send_response(error_response, handler, conn, interruptor);
            return false;
        } else {
            scoped_array_t<char> data(size);
            conn->read(data.data(), size, interruptor);

            if (!query_out->get()->ParseFromArray(data.data(), size)) {
                Response error_response;
                int64_t token = query_out->get()->has_token() ? query_out->get()->token() : 0;
                handler->unparseable_query(token, &error_response,
                                           "Client is buggy (failed to deserialize query).");
                send_response(error_response, handler, conn, interruptor);
                return false;
            }
        }
        return true;
    }

    static void send_response(const Response &response,
                              query_handler_t *handler,
                              tcp_conn_t *conn,
                              signal_t *interruptor) {
        scoped_array_t<char> scoped_array;
        const char *data;
        uint32_t size;

        if (static_cast<uint64_t>(response.ByteSize()) > MAX_RESPONSE_SIZE) {
            Response error_response;
            handler->unparseable_query(response.token(), &error_response,
                strprintf("Response size (%d) is greater than maximum (%zu).",
                          response.ByteSize(), MAX_RESPONSE_SIZE));
            send_response(error_response, handler, conn, interruptor);
            return;
        }
        size = response.ByteSize();
        scoped_array.init(size);
        response.SerializeToArray(scoped_array.data(), size);
        data = scoped_array.data();
        conn->write(&size, sizeof(size), interruptor);
        conn->write(data, size, interruptor);
    }
};

query_server_t::query_server_t(rdb_context_t *_rdb_ctx,
                               const std::set<ip_address_t> &local_addresses,
                               int port,
                               query_handler_t *_handler,
                               boost::shared_ptr<semilattice_readwrite_view_t<auth_semilattice_metadata_t> > _auth_metadata) :
        rdb_ctx(_rdb_ctx),
        handler(_handler),
        auth_metadata(_auth_metadata),
        shutting_down_conds(),
        pulse_sdc_on_shutdown(&main_shutting_down_cond),
        next_thread(0)
{
    rassert(rdb_ctx != NULL);
    for (int i = 0; i < get_num_threads(); ++i) {
        cross_thread_signal_t *s =
            new cross_thread_signal_t(&main_shutting_down_cond, threadnum_t(i));
        shutting_down_conds.push_back(s);
        rassert(s == &shutting_down_conds[i]);
{'fraction_non_alphanumeric': 0.10684229784120014, 'fraction_numerical': 0.006586169045005488, 'mean_word_length': 2.0422974176313446, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 87, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2317826', 'n_tokens_mistral': 3978, 'n_tokens_neox': 3862, 'n_words': 756}