Method grab_thread

src/bvar/collector.cpp:156–284

static CollectorSpeedLimit g_null_speed_limit = BVAR_COLLECTOR_SPEED_LIMIT_INITIALIZER;

void Collector::grab_thread() {
    _last_active_cpuwide_us = butil::cpuwide_time_us();
    int64_t last_before_update_sl = _last_active_cpuwide_us;

    // This is the thread for collecting TLS submissions. User's callbacks are
    // called inside the separate _dump_thread to prevent a slow callback
    // (caused by busy disk generally) from blocking collecting code too long
    // that pending requests may explode memory.
    CHECK_EQ(0, pthread_create(&_dump_thread, NULL, run_dump_thread, this));

    // vars
    bvar::PassiveStatus<int64_t> pending_sampled_data(
        "bvar_collector_pending_samples", get_pending_count, this);
    double busy_seconds = 0;
    bvar::PassiveStatus<double> busy_seconds_var(deref_value<double>, &busy_seconds);
    bvar::PerSecond<bvar::PassiveStatus<double> > busy_seconds_second(
        "bvar_collector_grab_thread_usage", &busy_seconds_var);

    bvar::PassiveStatus<int64_t> ngrab_var(deref_value<int64_t>, &_ngrab);
    bvar::PerSecond<bvar::PassiveStatus<int64_t> > ngrab_second(
        "bvar_collector_grab_second", &ngrab_var);

    // Maps for calculating speed limit.
    typedef std::map<CollectorSpeedLimit*, size_t> GrapMap;
    GrapMap last_ngrab_map;
    GrapMap ngrab_map;
    // Map for group samples by preprocessors.
    typedef std::map<CollectorPreprocessor*, std::vector<Collected*> >
        PreprocessorMap;
    PreprocessorMap prep_map;

    // The main loop.
    while (!_stop) {
        const int64_t abstime = _last_active_cpuwide_us + COLLECTOR_GRAB_INTERVAL_US;

        // Clear and reuse vectors in prep_map, don't clear prep_map directly.
        for (PreprocessorMap::iterator it = prep_map.begin(); it != prep_map.end();
             ++it) {
            it->second.clear();
        }

        // Collect TLS submissions and give them to dump_thread.
        butil::LinkNode<Collected>* head = this->reset();
        if (head) {
            butil::LinkNode<Collected> tmp_root;
            head->InsertBeforeAsList(&tmp_root);
            head = NULL;
            
            // Group samples by preprocessors.
            for (butil::LinkNode<Collected>* p = tmp_root.next(); p != &tmp_root;) {
                butil::LinkNode<Collected>* saved_next = p->next();
                p->RemoveFromList();
                CollectorPreprocessor* prep = p->value()->preprocessor();
                prep_map[prep].push_back(p->value());
                p = saved_next;
            }
            // Iterate prep_map
            butil::LinkNode<Collected> root;