<commit_before>#ifndef RPC_CONNECTIVITY_CLUSTER_HPP_
#define RPC_CONNECTIVITY_CLUSTER_HPP_

#include <map>
#include <set>
#include <utility>

#include "arch/types.hpp"
#include "concurrency/auto_drainer.hpp"
#include "concurrency/one_per_thread.hpp"
#include "containers/archive/tcp_conn_stream.hpp"
#include "containers/map_sentries.hpp"
#include "perfmon/perfmon.hpp"
#include "rpc/connectivity/connectivity.hpp"
#include "rpc/connectivity/messages.hpp"
#include "containers/uuid.hpp"

namespace boost {
template <class> class optional;
template <class> class scoped_ptr;
template <class> class function;
}

extern const char *const cluster_proto_header;

class peer_address_t {
public:
    peer_address_t(ip_address_t i, int p) : ip(i), port(p) { }
    peer_address_t() : ip(), port(0) { } // For deserialization
    ip_address_t ip;
    int port;

    bool operator==(const peer_address_t &a) const {
        return ip == a.ip && port == a.port;
    }
    bool operator!=(const peer_address_t &a) const {
        return ip != a.ip || port != a.port;
    }
    bool operator<(const peer_address_t &a) const {
        return ip < a.ip || (ip == a.ip && port < a.port);
    }

private:
    RDB_MAKE_ME_SERIALIZABLE_2(ip, port);
};

void debug_print(append_only_printf_buffer_t *buf, const peer_address_t &address);

class connectivity_cluster_t :
    public connectivity_service_t,
    public message_service_t,
    public home_thread_mixin_debug_only_t
{
public:
    class run_t {
    public:
        run_t(connectivity_cluster_t *parent,
            int port,
            message_handler_t *message_handler,
            int client_port = 0) THROWS_ONLY(address_in_use_exc_t);

        ~run_t();

        /* Attaches the cluster this node is part of to another existing
        cluster. May only be called on home thread. Returns immediately (it does
        its work in the background). */
        void join(peer_address_t) THROWS_NOTHING;
        int get_port();

    private:
        friend class connectivity_cluster_t;

        class connection_entry_t : public home_thread_mixin_debug_only_t {
        public:
            /* The constructor registers us in every thread's `connection_map`;
            the destructor deregisters us. Both also notify all subscribers. */
            connection_entry_t(run_t *, peer_id_t, tcp_conn_stream_t *, peer_address_t) THROWS_NOTHING;
            ~connection_entry_t() THROWS_NOTHING;

            /* NULL for our "connection" to ourself */
            tcp_conn_stream_t *conn;

            /* `connection_t` contains a `peer_address_t` so that we can call
            `get_peers_list()` on any thread. Otherwise, we would have to go
            cross-thread to access the routing table. */
            peer_address_t address;

            /* Unused for our connection to ourself */
            mutex_t send_mutex;

            uuid_t session_id;

            perfmon_collection_t pm_collection;
            perfmon_sampler_t pm_bytes_sent;
            perfmon_membership_t pm_collection_membership, pm_bytes_sent_membership;

        private:
            /* We only hold this information so we can deregister ourself */
            run_t *parent;
            peer_id_t peer;

            struct entry_installation_t {
                auto_drainer_t drainer_;
                connection_entry_t *that_;

                explicit entry_installation_t(connection_entry_t *that);
                ~entry_installation_t();
            };

            /* These are one-per-thread; we destroy them as we deregister
            ourselves on each thread. Things that want to send a message via
            this `connection_entry_t` acquire the drainer for their thread when
            they look us up in `thread_info_t::connection_map`. */
            scoped_ptr_t<one_per_thread_t<entry_installation_t> > entries;
        };

        /* Sets a variable to a value in its constructor; sets it to NULL in its
        destructor. This is kind of silly. The reason we need it is that we need
        the variable to be set before the constructors for some other fields of
        the `run_t` are constructed. */
        class variable_setter_t {
        public:
            variable_setter_t(run_t **var, run_t *val)
                : variable(var)
#ifndef NDEBUG
                , value(val)
#endif
            {
                guarantee_reviewed(*variable == NULL);
                *variable = value;
            }

            ~variable_setter_t() THROWS_NOTHING {
                guarantee_reviewed(*variable == value);
                *variable = NULL;
            }
        private:
            run_t **variable;
#ifndef NDEBUG
            run_t *value;
#endif
            DISABLE_COPYING(variable_setter_t);
        };

        void on_new_connection(const scoped_ptr_t<tcp_conn_descriptor_t> &nconn, auto_drainer_t::lock_t lock) THROWS_NOTHING;

        /* `connectivity_cluster_t::join_blocking()` is spawned in a new
        coroutine by `connectivity_cluster_t::join()`. It's also run by
        `connectivity_cluster_t::handle()` when we hear about a new peer from a
        peer we are connected to. */
        void join_blocking(peer_address_t address, boost::optional<peer_id_t>, auto_drainer_t::lock_t) THROWS_NOTHING;<commit_msg>Made things complie in release mode by making variables not be debug-only in cluster.hpp.<commit_after>#ifndef RPC_CONNECTIVITY_CLUSTER_HPP_
#define RPC_CONNECTIVITY_CLUSTER_HPP_

#include <map>
#include <set>
#include <utility>

#include "arch/types.hpp"
#include "concurrency/auto_drainer.hpp"
#include "concurrency/one_per_thread.hpp"
#include "containers/archive/tcp_conn_stream.hpp"
#include "containers/map_sentries.hpp"
#include "perfmon/perfmon.hpp"
#include "rpc/connectivity/connectivity.hpp"
#include "rpc/connectivity/messages.hpp"
#include "containers/uuid.hpp"

namespace boost {
template <class> class optional;
template <class> class scoped_ptr;
template <class> class function;
}

extern const char *const cluster_proto_header;

class peer_address_t {
public:
    peer_address_t(ip_address_t i, int p) : ip(i), port(p) { }
    peer_address_t() : ip(), port(0) { } // For deserialization
    ip_address_t ip;
    int port;

    bool operator==(const peer_address_t &a) const {
        return ip == a.ip && port == a.port;
    }
    bool operator!=(const peer_address_t &a) const {
        return ip != a.ip || port != a.port;
    }
    bool operator<(const peer_address_t &a) const {
        return ip < a.ip || (ip == a.ip && port < a.port);
    }

private:
    RDB_MAKE_ME_SERIALIZABLE_2(ip, port);
};

void debug_print(append_only_printf_buffer_t *buf, const peer_address_t &address);

class connectivity_cluster_t :
    public connectivity_service_t,
    public message_service_t,
    public home_thread_mixin_debug_only_t
{
public:
    class run_t {
    public:
        run_t(connectivity_cluster_t *parent,
            int port,
            message_handler_t *message_handler,
            int client_port = 0) THROWS_ONLY(address_in_use_exc_t);

        ~run_t();

        /* Attaches the cluster this node is part of to another existing
        cluster. May only be called on home thread. Returns immediately (it does
        its work in the background). */
        void join(peer_address_t) THROWS_NOTHING;
        int get_port();

    private:
        friend class connectivity_cluster_t;

        class connection_entry_t : public home_thread_mixin_debug_only_t {
        public:
            /* The constructor registers us in every thread's `connection_map`;
            the destructor deregisters us. Both also notify all subscribers. */
            connection_entry_t(run_t *, peer_id_t, tcp_conn_stream_t *, peer_address_t) THROWS_NOTHING;
            ~connection_entry_t() THROWS_NOTHING;

            /* NULL for our "connection" to ourself */
            tcp_conn_stream_t *conn;

            /* `connection_t` contains a `peer_address_t` so that we can call
            `get_peers_list()` on any thread. Otherwise, we would have to go
            cross-thread to access the routing table. */
            peer_address_t address;

            /* Unused for our connection to ourself */
            mutex_t send_mutex;

            uuid_t session_id;

            perfmon_collection_t pm_collection;
            perfmon_sampler_t pm_bytes_sent;
            perfmon_membership_t pm_collection_membership, pm_bytes_sent_membership;

        private:
            /* We only hold this information so we can deregister ourself */
            run_t *parent;
            peer_id_t peer;

            struct entry_installation_t {
                auto_drainer_t drainer_;
                connection_entry_t *that_;

                explicit entry_installation_t(connection_entry_t *that);
                ~entry_installation_t();
            };

            /* These are one-per-thread; we destroy them as we deregister
            ourselves on each thread. Things that want to send a message via
            this `connection_entry_t` acquire the drainer for their thread when
            they look us up in `thread_info_t::connection_map`. */
            scoped_ptr_t<one_per_thread_t<entry_installation_t> > entries;
        };

        /* Sets a variable to a value in its constructor; sets it to NULL in its
        destructor. This is kind of silly. The reason we need it is that we need
        the variable to be set before the constructors for some other fields of
        the `run_t` are constructed. */
        class variable_setter_t {
        public:
            variable_setter_t(run_t **var, run_t *val) : variable(var) , value(val) {
                guarantee_reviewed(*variable == NULL);
                *variable = value;
            }

            ~variable_setter_t() THROWS_NOTHING {
                guarantee_reviewed(*variable == value);
                *variable = NULL;
            }
        private:
            run_t **variable;
            run_t *value;
            DISABLE_COPYING(variable_setter_t);
        };

        void on_new_connection(const scoped_ptr_t<tcp_conn_descriptor_t> &nconn, auto_drainer_t::lock_t lock) THROWS_NOTHING;

        /* `connectivity_cluster_t::join_blocking()` is spawned in a new
        coroutine by `connectivity_cluster_t::join()`. It's also run by
        `connectivity_cluster_t::handle()` when we hear about a new peer from a
        peer we are connected to. */
        void join_blocking(peer_address_t address, boost::optional<peer_id_t>, auto_drainer_t::lock_t) THROWS_NOTHING;
{'fraction_non_alphanumeric': 0.0793064241046049, 'fraction_numerical': 0.0005685048322910744, 'mean_word_length': 2.574331188621741, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3930530', 'n_tokens_mistral': 3214, 'n_tokens_neox': 3133, 'n_words': 864}