Method SetRemoteConsumed

src/brpc/stream.cpp:362–401

}

void Stream::SetRemoteConsumed(size_t new_remote_consumed) {
    CHECK(_cur_buf_size > 0);
    bthread_id_list_t tmplist;
    bthread_id_list_init(&tmplist, 0, 0);
    bthread_mutex_lock(&_congestion_control_mutex);
    if (_remote_consumed >= new_remote_consumed) {
        bthread_mutex_unlock(&_congestion_control_mutex);
        return;
    }
    const bool was_full = _produced >= _remote_consumed + _cur_buf_size;

    if (FLAGS_socket_max_streams_unconsumed_bytes > 0) {
        _host_socket->_total_streams_unconsumed_size -= new_remote_consumed - _remote_consumed;
        if (_host_socket->_total_streams_unconsumed_size > FLAGS_socket_max_streams_unconsumed_bytes) {
            if (_options.min_buf_size > 0) {
                _cur_buf_size = _options.min_buf_size;
            } else {
                _cur_buf_size /= 2;
            }
            LOG(INFO) << "stream consumers on socket " << _host_socket->id() << " is crowded, " <<  "cut stream " << id() << " buffer to " << _cur_buf_size;
        } else if (_produced >= new_remote_consumed + _cur_buf_size && (_options.max_buf_size <= 0 || _cur_buf_size < (size_t)_options.max_buf_size)) {
            if (_options.max_buf_size > 0 && _cur_buf_size * 2 > (size_t)_options.max_buf_size) {
                _cur_buf_size = _options.max_buf_size;
            } else {
                _cur_buf_size *= 2;
            }
        }
    }

    _remote_consumed = new_remote_consumed;
    const bool is_full = _produced >= _remote_consumed + _cur_buf_size;
    if (was_full && !is_full) {
        bthread_id_list_swap(&tmplist, &_writable_wait_list);
    }
    bthread_mutex_unlock(&_congestion_control_mutex);

    // broadcast
    bthread_id_list_reset(&tmplist, 0);
    bthread_id_list_destroy(&tmplist);
}

void* Stream::RunOnWritable(void* arg) {
    WritableMeta *wm = (WritableMeta*)arg;