GitHub Viewer
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
// bthread - An M:N threading library to make applications more concurrent.
#include // heap functions
#include
#include "butil/scoped_lock.h"
#include "butil/logging.h"
#include "butil/third_party/murmurhash3/murmurhash3.h" // fmix64
#include "butil/resource_pool.h"
#include "butil/threading/platform_thread.h"
#include "bvar/bvar.h"
#include "bthread/sys_futex.h"
#include "bthread/timer_thread.h"
#include "bthread/log.h"
namespace bthread {
DEFINE_uint32(brpc_timer_num_buckets, 13, "brpc timer num buckets");
// Tasks unscheduled after being pulled into the timer thread's min-heap are
// only recycled when popped at their run_time, which can be far in the future
// for large timeouts. To bound the memory they occupy (~ qps * timeout), the
// timer thread periodically sweeps the heap and drops unscheduled tasks. The
// sweep only kicks in once the heap grows beyond this size, so that small
// heaps (where the retained memory is negligible) never pay the O(N) cost.
DEFINE_uint32(brpc_timer_heap_sweep_min_size, 4096,
"The timer thread sweeps unscheduled tasks out of its internal "
"heap only when the heap has at least this many tasks");
// The timer thread only consumes buckets and reclaims unscheduled tasks when
// it wakes up, which normally happens at the nearest task's run_time. If every
// pending task has a far-future run_time (e.g. minutes away), the thread would
// sleep that whole time while newly scheduled-then-unscheduled tasks pile up
// in the buckets, occupying pooled slots for the entire duration. Capping the
// sleep makes the thread wake up periodically to drain the buckets and sweep
// the heap, bounding that latency regardless of the timeout distribution.
// 0 (the default) disables the cap: sleep until the nearest run_time, the
// legacy behavior. Set it to a positive value to bound reclaim latency when
// tasks may have far-future run_times.
DEFINE_uint32(brpc_timer_max_wakeup_interval_ms, 0,
"The timer thread wakes up at least this often (in milliseconds) "
"to reclaim unscheduled tasks even when all pending tasks are far "
"in the future; 0 means no periodic wakeup");
// Defined in task_control.cpp
void run_worker_startfn();
const TimerThread::TaskId TimerThread::INVALID_TASK_ID = 0;
TimerThreadOptions::TimerThreadOptions()
: num_buckets(13) {
}
// A task contains the necessary information for running fn(arg).
// Tasks are created in Bucket::schedule and destroyed in TimerThread::run
struct BAIDU_CACHELINE_ALIGNMENT TimerThread::Task {
Task* next; // For linking tasks in a Bucket.
int64_t run_time; // run the task at this realtime
void (*fn)(void*); // the fn(arg) to run
void* arg;
// Current TaskId, checked against version in TimerThread::run to test
// if this task is unscheduled.
TaskId task_id;
// initial_version: not run yet
// initial_version + 1: running
// initial_version + 2: removed (also the version of next Task reused
// this struct)
butil::atomic version;
Task() : version(2/*skip 0*/) {}
// Run this task and delete this struct.
// Returns true if fn(arg) did run.
bool run_and_delete();
// Delete this struct if this task was unscheduled.
// Returns true on deletion.
bool try_delete();
};
// Timer tasks are sharded into different Buckets to reduce contentions.
class BAIDU_CACHELINE_ALIGNMENT TimerThread::Bucket {
public:
Bucket()
: _nearest_run_time(std::numeric_limits::max())
, _task_head(nullptr) {
}
~Bucket() {}
struct ScheduleResult {
TimerThread::TaskId task_id;
bool earlier;
};
// Schedule a task into this bucket.
// Returns the TaskId and if it has the nearest run time.
ScheduleResult schedule(void (*fn)(void*), void* arg,
const timespec& abstime);
// Pull all scheduled tasks.
// This function is called in timer thread.
Task* consume_tasks();
private:
FastPthreadMutex _mutex;
int64_t _nearest_run_time;
Task* _task_head;
};
// Utilies for making and extracting TaskId.
inline TimerThread::TaskId make_task_id(
butil::ResourceId slot, uint32_t version) {
return TimerThread::TaskId((((uint64_t)version) > 32);
}
inline bool task_greater(const TimerThread::Task* a, const TimerThread::Task* b) {
return a->run_time > b->run_time;
}
void* TimerThread::run_this(void* arg) {
butil::PlatformThread::SetNameSimple("brpc_timer");
static_cast(arg)->run();
return nullptr;
}
TimerThread::TimerThread()
: _started(false)
, _stop(false)
, _buckets(nullptr)
, _nearest_run_time(std::numeric_limits::max())
, _nsignals(0)
, _npending(0)
, _thread(0) {
}
TimerThread::~TimerThread() {
stop_and_join();
delete [] _buckets;
_buckets = nullptr;
}
int TimerThread::start(const TimerThreadOptions* options_in) {
if (_started) {
return 0;
}
if (options_in) {
_options = *options_in;
}
if (_options.num_buckets == 0) {
LOG(ERROR) 1024) {
LOG(ERROR) arg = arg;
task->run_time = butil::timespec_to_microseconds(abstime);
uint32_t version = task->version.load(butil::memory_order_relaxed);
if (version == 0) { // skip 0.
task->version.fetch_add(2, butil::memory_order_relaxed);
version = 2;
}
const TaskId id = make_task_id(slot_id, version);
task->task_id = id;
bool earlier = false;
{
BAIDU_SCOPED_LOCK(_mutex);
task->next = _task_head;
_task_head = task;
if (task->run_time < _nearest_run_time) {
_nearest_run_time = task->run_time;
earlier = true;
}
}
ScheduleResult result = { id, earlier };
return result;
}
TimerThread::TaskId TimerThread::schedule(
void (*fn)(void*), void* arg, const timespec& abstime) {
if (_stop.load(butil::memory_order_relaxed) || !_started) {
// Not add tasks when TimerThread is about to stop.
return INVALID_TASK_ID;
}
// Hashing by pthread id is better for cache locality.
const Bucket::ScheduleResult result =
_buckets[butil::fmix64(pthread_numeric_id()) % _options.num_buckets]
.schedule(fn, arg, abstime);
if (result.earlier) {
bool earlier = false;
const int64_t run_time = butil::timespec_to_microseconds(abstime);
{
BAIDU_SCOPED_LOCK(_mutex);
if (run_time < _nearest_run_time) {
_nearest_run_time = run_time;
++_nsignals;
earlier = true;
}
}
if (earlier) {
futex_wake_private(&_nsignals, 1);
}
}
return result.task_id;
}
// Notice that we don't recycle the Task in this function, let TimerThread::run
// do it. The side effect is that we may allocate many unscheduled tasks before
// TimerThread wakes up. The number is approximately qps * timeout_s. Under the
// precondition that ResourcePool caches 128K for each thread, with some
// further calculations, we can conclude that in a RPC scenario:
// when timeout / latency < 2730 (128K / sizeof(Task))
// unscheduled tasks do not occupy additional memory. 2730 is a large ratio
// between timeout and latency in most RPC scenarios, this is why we don't
// try to reuse tasks right now inside unschedule() with more complicated code.
int TimerThread::unschedule(TaskId task_id) {
const butil::ResourceId slot_id = slot_of_task_id(task_id);
Task* const task = butil::address_resource(slot_id);
if (task == nullptr) {
LOG(ERROR) try_delete()) { // still scheduled, keep it
tasks[j++] = task;
}
}
if (j != tasks.size()) {
tasks.resize(j);
std::make_heap(tasks.begin(), tasks.end(), task_greater);
}
last_sweep_size = tasks.size();
}
bool pull_again = false;
while (!tasks.empty()) {
Task* task1 = tasks[0]; // the about-to-run task
if (butil::gettimeofday_us() < task1->run_time) { // not ready yet.
break;
}
// Each time before we run the earliest task (that we think),
// check the globally shared _nearest_run_time. If a task earlier
// than task1 was scheduled during pulling from buckets, we'll
// know. In RPC scenarios, _nearest_run_time is not often changed by
// threads because the task needs to be the earliest in its bucket,
// since run_time of scheduled tasks are often in ascending order,
// most tasks are unlikely to be "earliest". (If run_time of tasks
// are in descending orders, all tasks are "earliest" after every
// insertion, and they'll grab _mutex and change _nearest_run_time
// frequently, fortunately this is not true at most of time).
{
BAIDU_SCOPED_LOCK(_mutex);
if (task1->run_time > _nearest_run_time) {
// a task is earlier than task1. We need to check buckets.
pull_again = true;
break;
}
}
std::pop_heap(tasks.begin(), tasks.end(), task_greater);
tasks.pop_back();
if (task1->run_and_delete()) {
++ntriggered;
}
}
// Publish the heap size before possibly looping back on pull_again,
// so the observability counter doesn't go stale during the retry spin.
_npending.store((int64_t)tasks.size(), butil::memory_order_relaxed);
if (pull_again) {
BT_VLOG