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

#include "concurrency/cross_thread_signal.hpp"

/* We destroy the machinery if there have been no changefeeds for this many seconds */
static const int machinery_expiration_secs = 60;

void cfeed_artificial_table_backend_t::machinery_t::send_all_change(
        const store_key_t &key,
        const ql::datum_t &old_val,
        const ql::datum_t &new_val) {
    ql::changefeed::msg_t::change_t change;
    change.pkey = key;
    change.old_val = old_val;
    change.new_val = new_val;
    send_all(ql::changefeed::msg_t(change));
}

void cfeed_artificial_table_backend_t::machinery_t::send_all_stop() {
    send_all(ql::changefeed::msg_t(ql::changefeed::msg_t::stop_t()));
}

void cfeed_artificial_table_backend_t::machinery_t::maybe_remove() {
    assert_thread();
    last_subscriber_time = current_microtime();
    /* The `cfeed_artificial_table_backend_t` has a repeating timer that will eventually
    clean us up */
}

cfeed_artificial_table_backend_t::cfeed_artificial_table_backend_t() :
    begin_destruction_was_called(false),
    remove_machinery_timer(
        machinery_expiration_secs * THOUSAND,
        [this]() { maybe_remove_machinery(); })
    { }

cfeed_artificial_table_backend_t::~cfeed_artificial_table_backend_t() {
    /* Catch subclasses that forget to call `begin_changefeed_destruction()`. This isn't
    strictly necessary (some hypothetical subclasses might not have any reason to call
    it) but it's far more likely that the subclass should have called it but forgot to.
    */
    guarantee(begin_destruction_was_called);
}

bool cfeed_artificial_table_backend_t::read_changes(
    ql::env_t *env,
    const ql::protob_t<const Backtrace> &bt,
    ql::changefeed::keyspec_t::spec_t &&spec,
    signal_t *interruptor,
    counted_t<ql::datum_stream_t> *cfeed_out,
    std::string *error_out) {

    guarantee(!begin_destruction_was_called);
    class visitor_t : public boost::static_visitor<bool> {
    public:
        bool operator()(const ql::changefeed::keyspec_t::limit_t &) const {
            return false;
        }
        bool operator()(const ql::changefeed::keyspec_t::range_t &) const {
            return true;
        }
        bool operator()(const ql::changefeed::keyspec_t::point_t &) const {
            return true;
        }
    };
    if (!boost::apply_visitor(visitor_t(), spec)) {
        *error_out = "System tables don't support changefeeds on `.limit()`.";
        return false;
    }
    threadnum_t request_thread = get_thread_id();
    cross_thread_signal_t interruptor2(interruptor, home_thread());
    on_thread_t thread_switcher(home_thread());
    new_mutex_in_line_t mutex_lock(&mutex);
    wait_interruptible(mutex_lock.acq_signal(), &interruptor2);
    if (!machinery.has()) {
        machinery = construct_changefeed_machinery(&interruptor2);
    }
    /* We construct two `on_thread_t`s for a total of four thread switches. This is
    necessary because we have to call `subscribe()` on the client thread, but we don't
    want to release the lock on the home thread until `subscribe()` returns. */
    on_thread_t thread_switcher_2(request_thread);
    *cfeed_out = machinery->subscribe(env, std::move(spec), this, bt);
    return true;
}

void cfeed_artificial_table_backend_t::begin_changefeed_destruction() {
    new_mutex_in_line_t mutex_lock(&mutex);
    mutex_lock.acq_signal()->wait_lazily_unordered();
    guarantee(!begin_destruction_was_called);
    if (machinery.has()) {
        /* All changefeeds should be closed before we start destruction */<commit_msg>Added handling for QL exceptions thrown by `subscribe`.<commit_after>// Copyright 2010-2014 RethinkDB, all rights reserved.
#include "rdb_protocol/artificial_table/cfeed_backend.hpp"

#include "concurrency/cross_thread_signal.hpp"

/* We destroy the machinery if there have been no changefeeds for this many seconds */
static const int machinery_expiration_secs = 60;

void cfeed_artificial_table_backend_t::machinery_t::send_all_change(
        const store_key_t &key,
        const ql::datum_t &old_val,
        const ql::datum_t &new_val) {
    ql::changefeed::msg_t::change_t change;
    change.pkey = key;
    change.old_val = old_val;
    change.new_val = new_val;
    send_all(ql::changefeed::msg_t(change));
}

void cfeed_artificial_table_backend_t::machinery_t::send_all_stop() {
    send_all(ql::changefeed::msg_t(ql::changefeed::msg_t::stop_t()));
}

void cfeed_artificial_table_backend_t::machinery_t::maybe_remove() {
    assert_thread();
    last_subscriber_time = current_microtime();
    /* The `cfeed_artificial_table_backend_t` has a repeating timer that will eventually
    clean us up */
}

cfeed_artificial_table_backend_t::cfeed_artificial_table_backend_t() :
    begin_destruction_was_called(false),
    remove_machinery_timer(
        machinery_expiration_secs * THOUSAND,
        [this]() { maybe_remove_machinery(); })
    { }

cfeed_artificial_table_backend_t::~cfeed_artificial_table_backend_t() {
    /* Catch subclasses that forget to call `begin_changefeed_destruction()`. This isn't
    strictly necessary (some hypothetical subclasses might not have any reason to call
    it) but it's far more likely that the subclass should have called it but forgot to.
    */
    guarantee(begin_destruction_was_called);
}

bool cfeed_artificial_table_backend_t::read_changes(
    ql::env_t *env,
    const ql::protob_t<const Backtrace> &bt,
    ql::changefeed::keyspec_t::spec_t &&spec,
    signal_t *interruptor,
    counted_t<ql::datum_stream_t> *cfeed_out,
    std::string *error_out) {

    guarantee(!begin_destruction_was_called);
    class visitor_t : public boost::static_visitor<bool> {
    public:
        bool operator()(const ql::changefeed::keyspec_t::limit_t &) const {
            return false;
        }
        bool operator()(const ql::changefeed::keyspec_t::range_t &) const {
            return true;
        }
        bool operator()(const ql::changefeed::keyspec_t::point_t &) const {
            return true;
        }
    };
    if (!boost::apply_visitor(visitor_t(), spec)) {
        *error_out = "System tables don't support changefeeds on `.limit()`.";
        return false;
    }
    threadnum_t request_thread = get_thread_id();
    cross_thread_signal_t interruptor2(interruptor, home_thread());
    on_thread_t thread_switcher(home_thread());
    new_mutex_in_line_t mutex_lock(&mutex);
    wait_interruptible(mutex_lock.acq_signal(), &interruptor2);
    if (!machinery.has()) {
        machinery = construct_changefeed_machinery(&interruptor2);
    }
    /* We construct two `on_thread_t`s for a total of four thread switches. This is
    necessary because we have to call `subscribe()` on the client thread, but we don't
    want to release the lock on the home thread until `subscribe()` returns. */
    on_thread_t thread_switcher_2(request_thread);
    try {
        *cfeed_out = machinery->subscribe(env, std::move(spec), this, bt);
    } catch (const ql::base_exc_t &e) {
        *error_out = e.what();
        return false;
    }
    return true;
}

void cfeed_artificial_table_backend_t::begin_changefeed_destruction() {
    new_mutex_in_line_t mutex_lock(&mutex);
    mutex_lock.acq_signal()->wait_lazily_unordered();
    guarantee(!begin_destruction_was_called);
    if (machinery.has()) {
        /* All changefeeds should be closed before we start destruction */
{'fraction_non_alphanumeric': 0.10465271297160379, 'fraction_numerical': 0.0037328356219170776, 'mean_word_length': 4.224233983286908, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, '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': '6196491', 'n_tokens_mistral': 2468, 'n_tokens_neox': 2445, 'n_words': 516}