<commit_before>#include "rdb_protocol/changefeed.hpp"

#include <queue>

#include "concurrency/cross_thread_signal.hpp"
#include "concurrency/interruptor.hpp"
#include "containers/archive/boost_types.hpp"
#include "rdb_protocol/btree.hpp"
#include "rdb_protocol/env.hpp"
#include "rdb_protocol/protocol.hpp"
#include "rdb_protocol/val.hpp"
#include "rpc/mailbox/typed.hpp"

#include "debug.hpp"

namespace ql {

namespace changefeed {

server_t::server_t(mailbox_manager_t *_manager)
    : uuid(generate_uuid()),
      manager(_manager),
      stop_mailbox(manager, std::bind(&server_t::stop_mailbox_cb, this, ph::_1)) { }

server_t::~server_t() { }

void server_t::stop_mailbox_cb(client_t::addr_t addr) {
    auto_drainer_t::lock_t lock(&drainer);
    rwlock_in_line_t spot(&clients_lock, access_t::read);
    spot.read_signal()->wait_lazily_unordered();
    auto it = clients.find(addr);
    // The client might have already been removed from e.g. a peer disconnect or
    // drainer destruction.  (Also, if we have multiple shards per btree this
    // will be called twice, and the second time it should be a no-op.)
    if (it != clients.end()) {
        it->second.cond->pulse_if_not_already_pulsed();
    }
}

void server_t::add_client(const client_t::addr_t &addr) {
    auto_drainer_t::lock_t lock(&drainer);
    rwlock_in_line_t spot(&clients_lock, access_t::write);
    spot.write_signal()->wait_lazily_unordered();
    auto info = &clients[addr];
    // The entry might already exist if we have multiple shards per btree, but
    // that's fine.
    if (!info->cond.has()) {
        info->stamp = 0;
        cond_t *stopped = new cond_t();
        info->cond.init(stopped);
        // We spawn now so the auto drainer lock is acquired immediately.
        // Passing the raw pointer `stopped` is safe because `add_client_cb` is
        // the only function which can remove an entry from the map.
        coro_t::spawn_now_dangerously(
            std::bind(&server_t::add_client_cb, this, stopped, addr));
    }
}

void server_t::add_client_cb(signal_t *stopped, client_t::addr_t addr) {
    auto_drainer_t::lock_t coro_lock(&drainer);
    disconnect_watcher_t disconnect(
        manager->get_connectivity_service(), addr.get_peer());
    {
        wait_any_t wait_any(
            &disconnect, stopped, coro_lock.get_drain_signal());
        wait_any.wait_lazily_unordered();
    }
    send_all_with_lock(coro_lock, msg_t(msg_t::stop_t()));
    rwlock_in_line_t coro_spot(&clients_lock, access_t::write);
    coro_spot.write_signal()->wait_lazily_unordered();
    size_t erased = clients.erase(addr);
    // This is true even if we have multiple shards per btree because
    // `add_client` only spawns one of us.
    guarantee(erased == 1);
}

struct stamped_msg_t {
    stamped_msg_t() { }
    stamped_msg_t(uuid_u _server_uuid, uint64_t _stamp, msg_t _submsg)
        : server_uuid(std::move(_server_uuid)),
          stamp(_stamp),
          submsg(std::move(_submsg)) { }
    uuid_u server_uuid;
    uint64_t stamp;
    msg_t submsg;
};

RDB_MAKE_SERIALIZABLE_3(stamped_msg_t, server_uuid, stamp, submsg);

// This function takes a `lock_t` to make sure you have one.  (We can't just
// always ackquire a drainer lock before sending because we sometimes send a
// `stop_t` during destruction, and you can't acquire a drain lock on a draining
// `auto_drainer_t`.)
void server_t::send_all_with_lock(const auto_drainer_t::lock_t &, msg_t msg) {
    rwlock_in_line_t spot(&clients_lock, access_t::read);
    spot.read_signal()->wait_lazily_unordered();
    for (auto it = clients.begin(); it != clients.end(); ++it) {
        uint64_t stamp;
        {
            // We don't need a write lock as long as we make sure the coroutine
            // doesn't block between reading and updating the stamp.
            ASSERT_NO_CORO_WAITING;
            stamp = it->second.stamp++;
        }
        send(manager, it->first, stamped_msg_t(uuid, stamp, msg));
    }
}

void server_t::send_all(msg_t msg) {
    auto_drainer_t::lock_t lock(&drainer);
    send_all_with_lock(lock, std::move(msg));
}

void server_t::stop_all() {
    auto_drainer_t::lock_t lock(&drainer);
    rwlock_in_line_t spot(&clients_lock, access_t::read);
    spot.read_signal()->wait_lazily_unordered();
    for (auto it = clients.begin(); it != clients.end(); ++it) {
        it->second.cond->pulse_if_not_already_pulsed();
    }
}

server_t::addr_t server_t::get_stop_addr() {
    return stop_mailbox.get_address();
}

uint64_t server_t::get_stamp(const client_t::addr_t &addr) {
    auto_drainer_t::lock_t lock(&drainer);
    rwlock_in_line_t spot(&clients_lock, access_t::read);
    spot.read_signal()->wait_lazily_unordered();
    auto it = clients.find(addr);
    if (it == clients.end()) {
        // The client was removed, so no future messages are coming.
        return std::numeric_limits<uint64_t>::max();
    } else {
        return it->second.stamp;
    }
}

uuid_u server_t::get_uuid() {
    return uuid;
}

msg_t::msg_t(msg_t &&msg) : op(std::move(msg.op)) { }
msg_t::msg_t(stop_t &&_op) : op(std::move(_op)) { }
msg_t::msg_t(change_t &&_op) : op(std::move(_op)) { }

msg_t::change_t::change_t() { }
msg_t::change_t::change_t(counted_t<const datum_t> _old_val,
                          counted_t<const datum_t> _new_val)
    : old_val(std::move(_old_val)), new_val(std::move(_new_val)) { }
msg_t::change_t::~change_t() { }

RDB_IMPL_SERIALIZABLE_1_SINCE_v1_13(msg_t, op);
RDB_IMPL_ME_SERIALIZABLE_2_SINCE_v1_13(
    msg_t::change_t, empty_ok(old_val), empty_ok(new_val));
RDB_IMPL_SERIALIZABLE_0_SINCE_v1_13(msg_t::stop_t);

enum class detach_t { NO, YES };

// Uses the home thread of the subscriber, not the client.
class subscription_t : public home_thread_mixin_t {
public:
    virtual ~subscription_t();
    std::vector<counted_t<const datum_t> >
    get_els(batcher_t *batcher, const signal_t *interruptor);
    virtual void add_el(
        const uuid_u &uuid , uint64_t stamp, counted_t<const datum_t> d) = 0;
    virtual void start(
        env_t *env, namespace_interface_t *nif, client_t::addr_t *addr) = 0;
    void stop(const std::string &msg, detach_t should_detach);
protected:
    explicit subscription_t(feed_t *_feed);
    void maybe_signal_cond() THROWS_NOTHING;
    void destructor_cleanup(std::function<void()> del_sub) THROWS_NOTHING;
    // If an error occurs, we're detached and `exc` is set to an exception to rethrow.
    std::exception_ptr exc;
    // If we exceed the array size limit, elements are evicted from `els` and
    // `skipped` is incremented appropriately.  If `skipped` is non-0, we send
    // an error object to the user with the number of skipped elements before
    // continuing.
    size_t skipped;
    // The feed we're subscribed to.
    feed_t *feed;
private:
    virtual bool has_el() = 0;
    virtual counted_t<const datum_t> pop_el() = 0;
    // Used to block on more changes.  NULL unless we're waiting.
    cond_t *cond;
    auto_drainer_t drainer;
    DISABLE_COPYING(subscription_t);
};

class feed_t : public home_thread_mixin_t, public slow_atomic_countable_t<feed_t> {
public:
    feed_t(client_t *client,
           mailbox_manager_t *manager,
           base_namespace_repo_t *ns_repo,
           uuid_u uuid,
           std::string pkey,
           signal_t *interruptor);
    ~feed_t();

    void add_point_sub(subscription_t *sub,
                       const counted_t<const datum_t> &key) THROWS_NOTHING;
    void del_point_sub(subscription_t *sub,
                       const counted_t<const datum_t> &key) THROWS_NOTHING;

    void add_table_sub(subscription_t *sub) THROWS_NOTHING;
    void del_table_sub(subscription_t *sub) THROWS_NOTHING;

    void each_table_sub(const std::function<void(subscription_t *)> &f) THROWS_NOTHING;
    void each_point_sub(const std::function<void(subscription_t *)> &f) THROWS_NOTHING;
    void each_sub(const std::function<void(subscription_t *)> &f) THROWS_NOTHING;
    void on_point_sub(counted_t<const datum_t> key,
                      const std::function<void(subscription_t *)> &f) THROWS_NOTHING;


    bool can_be_removed();
    client_t::addr_t get_addr() const;

    const std::string pkey;
private:
    void each_sub_in_vec(
        const std::vector<std::set<subscription_t *> > &vec,
        rwlock_in_line_t *spot,
        const std::function<void(subscription_t *)> &f) THROWS_NOTHING;
    void each_sub_in_vec_cb(const std::function<void(subscription_t *)> &f,
                            const std::vector<std::set<subscription_t *> > &vec,
                            const std::vector<int> &subscription_threads,
                            int i);
    void each_point_sub_cb(const std::function<void(subscription_t *)> &f, int i);
    void mailbox_cb(stamped_msg_t msg);
    void constructor_cb();

    client_t *client;
    uuid_u uuid;
    mailbox_manager_t *manager;
    mailbox_t<void(stamped_msg_t)> mailbox;
    std::vector<server_t::addr_t> stop_addrs;

    std::vector<scoped_ptr_t<disconnect_watcher_t> > disconnect_watchers;
    wait_any_t any_disconnect;

    struct queue_t {
        rwlock_t lock;
        uint64_t next;
        struct lt_t {
            bool operator()(const stamped_msg_t &left, const stamped_msg_t &right) {
                return left.stamp > right.stamp; // We want the min val to be on top.
            }
        };
        std::priority_queue<stamped_msg_t, std::vector<stamped_msg_t>, lt_t> map;
    };
    // Maps from a `server_t`'s uuid_u.  We don't need a lock for this because
    // the set of `uuid_u`s never changes after it's initialized.
    std::map<uuid_u, scoped_ptr_t<queue_t> > queues;
    cond_t queues_ready;

    typedef std::vector<std::set<subscription_t *> > subvec_t;
    subvec_t table_subs;
    std::map<counted_t<const datum_t>, subvec_t> point_subs;
    rwlock_t table_subs_lock, point_subs_lock;
    int64_t num_subs;

    bool detached;

    auto_drainer_t drainer;
};

class point_sub_t : public subscription_t {
public:
    // Throws QL exceptions.
    point_sub_t(feed_t *feed, counted_t<const datum_t> _key)
        : subscription_t(feed), key(std::move(_key)), stamp(0) {<commit_msg>Removed typedef from changefeed code.<commit_after>#include "rdb_protocol/changefeed.hpp"

#include <queue>

#include "concurrency/cross_thread_signal.hpp"
#include "concurrency/interruptor.hpp"
#include "containers/archive/boost_types.hpp"
#include "rdb_protocol/btree.hpp"
#include "rdb_protocol/env.hpp"
#include "rdb_protocol/protocol.hpp"
#include "rdb_protocol/val.hpp"
#include "rpc/mailbox/typed.hpp"

#include "debug.hpp"

namespace ql {

namespace changefeed {

server_t::server_t(mailbox_manager_t *_manager)
    : uuid(generate_uuid()),
      manager(_manager),
      stop_mailbox(manager, std::bind(&server_t::stop_mailbox_cb, this, ph::_1)) { }

server_t::~server_t() { }

void server_t::stop_mailbox_cb(client_t::addr_t addr) {
    auto_drainer_t::lock_t lock(&drainer);
    rwlock_in_line_t spot(&clients_lock, access_t::read);
    spot.read_signal()->wait_lazily_unordered();
    auto it = clients.find(addr);
    // The client might have already been removed from e.g. a peer disconnect or
    // drainer destruction.  (Also, if we have multiple shards per btree this
    // will be called twice, and the second time it should be a no-op.)
    if (it != clients.end()) {
        it->second.cond->pulse_if_not_already_pulsed();
    }
}

void server_t::add_client(const client_t::addr_t &addr) {
    auto_drainer_t::lock_t lock(&drainer);
    rwlock_in_line_t spot(&clients_lock, access_t::write);
    spot.write_signal()->wait_lazily_unordered();
    auto info = &clients[addr];
    // The entry might already exist if we have multiple shards per btree, but
    // that's fine.
    if (!info->cond.has()) {
        info->stamp = 0;
        cond_t *stopped = new cond_t();
        info->cond.init(stopped);
        // We spawn now so the auto drainer lock is acquired immediately.
        // Passing the raw pointer `stopped` is safe because `add_client_cb` is
        // the only function which can remove an entry from the map.
        coro_t::spawn_now_dangerously(
            std::bind(&server_t::add_client_cb, this, stopped, addr));
    }
}

void server_t::add_client_cb(signal_t *stopped, client_t::addr_t addr) {
    auto_drainer_t::lock_t coro_lock(&drainer);
    disconnect_watcher_t disconnect(
        manager->get_connectivity_service(), addr.get_peer());
    {
        wait_any_t wait_any(
            &disconnect, stopped, coro_lock.get_drain_signal());
        wait_any.wait_lazily_unordered();
    }
    send_all_with_lock(coro_lock, msg_t(msg_t::stop_t()));
    rwlock_in_line_t coro_spot(&clients_lock, access_t::write);
    coro_spot.write_signal()->wait_lazily_unordered();
    size_t erased = clients.erase(addr);
    // This is true even if we have multiple shards per btree because
    // `add_client` only spawns one of us.
    guarantee(erased == 1);
}

struct stamped_msg_t {
    stamped_msg_t() { }
    stamped_msg_t(uuid_u _server_uuid, uint64_t _stamp, msg_t _submsg)
        : server_uuid(std::move(_server_uuid)),
          stamp(_stamp),
          submsg(std::move(_submsg)) { }
    uuid_u server_uuid;
    uint64_t stamp;
    msg_t submsg;
};

RDB_MAKE_SERIALIZABLE_3(stamped_msg_t, server_uuid, stamp, submsg);

// This function takes a `lock_t` to make sure you have one.  (We can't just
// always ackquire a drainer lock before sending because we sometimes send a
// `stop_t` during destruction, and you can't acquire a drain lock on a draining
// `auto_drainer_t`.)
void server_t::send_all_with_lock(const auto_drainer_t::lock_t &, msg_t msg) {
    rwlock_in_line_t spot(&clients_lock, access_t::read);
    spot.read_signal()->wait_lazily_unordered();
    for (auto it = clients.begin(); it != clients.end(); ++it) {
        uint64_t stamp;
        {
            // We don't need a write lock as long as we make sure the coroutine
            // doesn't block between reading and updating the stamp.
            ASSERT_NO_CORO_WAITING;
            stamp = it->second.stamp++;
        }
        send(manager, it->first, stamped_msg_t(uuid, stamp, msg));
    }
}

void server_t::send_all(msg_t msg) {
    auto_drainer_t::lock_t lock(&drainer);
    send_all_with_lock(lock, std::move(msg));
}

void server_t::stop_all() {
    auto_drainer_t::lock_t lock(&drainer);
    rwlock_in_line_t spot(&clients_lock, access_t::read);
    spot.read_signal()->wait_lazily_unordered();
    for (auto it = clients.begin(); it != clients.end(); ++it) {
        it->second.cond->pulse_if_not_already_pulsed();
    }
}

server_t::addr_t server_t::get_stop_addr() {
    return stop_mailbox.get_address();
}

uint64_t server_t::get_stamp(const client_t::addr_t &addr) {
    auto_drainer_t::lock_t lock(&drainer);
    rwlock_in_line_t spot(&clients_lock, access_t::read);
    spot.read_signal()->wait_lazily_unordered();
    auto it = clients.find(addr);
    if (it == clients.end()) {
        // The client was removed, so no future messages are coming.
        return std::numeric_limits<uint64_t>::max();
    } else {
        return it->second.stamp;
    }
}

uuid_u server_t::get_uuid() {
    return uuid;
}

msg_t::msg_t(msg_t &&msg) : op(std::move(msg.op)) { }
msg_t::msg_t(stop_t &&_op) : op(std::move(_op)) { }
msg_t::msg_t(change_t &&_op) : op(std::move(_op)) { }

msg_t::change_t::change_t() { }
msg_t::change_t::change_t(counted_t<const datum_t> _old_val,
                          counted_t<const datum_t> _new_val)
    : old_val(std::move(_old_val)), new_val(std::move(_new_val)) { }
msg_t::change_t::~change_t() { }

RDB_IMPL_SERIALIZABLE_1_SINCE_v1_13(msg_t, op);
RDB_IMPL_ME_SERIALIZABLE_2_SINCE_v1_13(
    msg_t::change_t, empty_ok(old_val), empty_ok(new_val));
RDB_IMPL_SERIALIZABLE_0_SINCE_v1_13(msg_t::stop_t);

enum class detach_t { NO, YES };

// Uses the home thread of the subscriber, not the client.
class subscription_t : public home_thread_mixin_t {
public:
    virtual ~subscription_t();
    std::vector<counted_t<const datum_t> >
    get_els(batcher_t *batcher, const signal_t *interruptor);
    virtual void add_el(
        const uuid_u &uuid , uint64_t stamp, counted_t<const datum_t> d) = 0;
    virtual void start(
        env_t *env, namespace_interface_t *nif, client_t::addr_t *addr) = 0;
    void stop(const std::string &msg, detach_t should_detach);
protected:
    explicit subscription_t(feed_t *_feed);
    void maybe_signal_cond() THROWS_NOTHING;
    void destructor_cleanup(std::function<void()> del_sub) THROWS_NOTHING;
    // If an error occurs, we're detached and `exc` is set to an exception to rethrow.
    std::exception_ptr exc;
    // If we exceed the array size limit, elements are evicted from `els` and
    // `skipped` is incremented appropriately.  If `skipped` is non-0, we send
    // an error object to the user with the number of skipped elements before
    // continuing.
    size_t skipped;
    // The feed we're subscribed to.
    feed_t *feed;
private:
    virtual bool has_el() = 0;
    virtual counted_t<const datum_t> pop_el() = 0;
    // Used to block on more changes.  NULL unless we're waiting.
    cond_t *cond;
    auto_drainer_t drainer;
    DISABLE_COPYING(subscription_t);
};

class feed_t : public home_thread_mixin_t, public slow_atomic_countable_t<feed_t> {
public:
    feed_t(client_t *client,
           mailbox_manager_t *manager,
           base_namespace_repo_t *ns_repo,
           uuid_u uuid,
           std::string pkey,
           signal_t *interruptor);
    ~feed_t();

    void add_point_sub(subscription_t *sub,
                       const counted_t<const datum_t> &key) THROWS_NOTHING;
    void del_point_sub(subscription_t *sub,
                       const counted_t<const datum_t> &key) THROWS_NOTHING;

    void add_table_sub(subscription_t *sub) THROWS_NOTHING;
    void del_table_sub(subscription_t *sub) THROWS_NOTHING;

    void each_table_sub(const std::function<void(subscription_t *)> &f) THROWS_NOTHING;
    void each_point_sub(const std::function<void(subscription_t *)> &f) THROWS_NOTHING;
    void each_sub(const std::function<void(subscription_t *)> &f) THROWS_NOTHING;
    void on_point_sub(counted_t<const datum_t> key,
                      const std::function<void(subscription_t *)> &f) THROWS_NOTHING;


    bool can_be_removed();
    client_t::addr_t get_addr() const;

    const std::string pkey;
private:
    void each_sub_in_vec(
        const std::vector<std::set<subscription_t *> > &vec,
        rwlock_in_line_t *spot,
        const std::function<void(subscription_t *)> &f) THROWS_NOTHING;
    void each_sub_in_vec_cb(const std::function<void(subscription_t *)> &f,
                            const std::vector<std::set<subscription_t *> > &vec,
                            const std::vector<int> &subscription_threads,
                            int i);
    void each_point_sub_cb(const std::function<void(subscription_t *)> &f, int i);
    void mailbox_cb(stamped_msg_t msg);
    void constructor_cb();

    client_t *client;
    uuid_u uuid;
    mailbox_manager_t *manager;
    mailbox_t<void(stamped_msg_t)> mailbox;
    std::vector<server_t::addr_t> stop_addrs;

    std::vector<scoped_ptr_t<disconnect_watcher_t> > disconnect_watchers;
    wait_any_t any_disconnect;

    struct queue_t {
        rwlock_t lock;
        uint64_t next;
        struct lt_t {
            bool operator()(const stamped_msg_t &left, const stamped_msg_t &right) {
                return left.stamp > right.stamp; // We want the min val to be on top.
            }
        };
        std::priority_queue<stamped_msg_t, std::vector<stamped_msg_t>, lt_t> map;
    };
    // Maps from a `server_t`'s uuid_u.  We don't need a lock for this because
    // the set of `uuid_u`s never changes after it's initialized.
    std::map<uuid_u, scoped_ptr_t<queue_t> > queues;
    cond_t queues_ready;

    std::vector<std::set<subscription_t *> > table_subs;
    std::map<counted_t<const datum_t>,
             std::vector<std::set<subscription_t *> > > point_subs;
    rwlock_t table_subs_lock, point_subs_lock;
    int64_t num_subs;

    bool detached;

    auto_drainer_t drainer;
};

class point_sub_t : public subscription_t {
public:
    // Throws QL exceptions.
    point_sub_t(feed_t *feed, counted_t<const datum_t> _key)
        : subscription_t(feed), key(std::move(_key)), stamp(0) {
{'fraction_non_alphanumeric': 0.11592568991716093, 'fraction_numerical': 0.0037253075829616197, 'mean_word_length': 3.5775185102086606, 'pattern_counts': {'":': 0, '<': 81, '<?xml version=': 0, '>': 117, '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': '5754313', 'n_tokens_mistral': 7325, 'n_tokens_neox': 7259, 'n_words': 1557}