if (g_worker_startfn) {

g_worker_startfn();

}

// NOTE: all fileds must be initialized before the vars.

: _ngroup(0)

, _groups((TaskGroup**)calloc(BTHREAD_MAX_CONCURRENCY, sizeof(TaskGroup*)))

, _stop(false)

, _concurrency(0)

, _nworkers("bthread_worker_count")

, _pending_time(NULL)

// Delay exposure of following two vars because they rely on TC which

// is not initialized yet.

, _cumulated_worker_time(get_cumulated_worker_time_from_this, this)

, _worker_usage_second(&_cumulated_worker_time, 1)

, _cumulated_switch_count(get_cumulated_switch_count_from_this, this)

, _switch_per_second(&_cumulated_switch_count)

, _cumulated_signal_count(get_cumulated_signal_count_from_this, this)

, _signal_per_second(&_cumulated_signal_count)

, _status(print_rq_sizes_in_the_tc, this)

, _nbthreads("bthread_count")

{

// calloc shall set memory to zero

CHECK(_groups) << "Fail to create array of groups";

if (_concurrency != 0) {

LOG(ERROR) << "Already initialized";

return -1;

}

if (concurrency <= 0) {

LOG(ERROR) << "Invalid concurrency=" << concurrency;

return -1;

}

_concurrency = concurrency;

// Make sure TimerThread is ready.

if (get_or_create_global_timer_thread() == NULL) {

LOG(ERROR) << "Fail to get global_timer_thread";

return -1;

}

_workers.resize(_concurrency);

for (int i = 0; i < _concurrency; ++i) {

const int rc = pthread_create(&_workers[i], NULL, worker_thread, this);

if (rc) {

LOG(ERROR) << "Fail to create _workers[" << i << "], " << berror(rc);

return -1;

}

}

_worker_usage_second.expose("bthread_worker_usage");

_switch_per_second.expose("bthread_switch_second");

_signal_per_second.expose("bthread_signal_second");

_status.expose("bthread_group_status");

// Wait for at least one group is added so that choose_one_group()

// never returns NULL.

// TODO: Handle the case that worker quits before add_group

while (_ngroup == 0) {

usleep(100); // TODO: Elaborate

}

return 0;

if (num <= 0) {

return 0;

}

try {

_workers.resize(_concurrency + num);

} catch (...) {

return 0;

}

const int old_concurency = _concurrency.load(butil::memory_order_relaxed);

for (int i = 0; i < num; ++i) {

// Worker will add itself to _idle_workers, so we have to add

// _concurrency before create a worker.

_concurrency.fetch_add(1);

const int rc = pthread_create(

&_workers[i + old_concurency], NULL, worker_thread, this);

if (rc) {

LOG(WARNING) << "Fail to create _workers[" << i + old_concurency

<< "], " << berror(rc);

_concurrency.fetch_sub(1, butil::memory_order_release);

break;

}

}

// Cannot fail

_workers.resize(_concurrency.load(butil::memory_order_relaxed));

return _concurrency.load(butil::memory_order_relaxed) - old_concurency;

// Close epoll threads so that worker threads are not waiting on epoll(

// which cannot be woken up by signal_task below)

CHECK_EQ(0, stop_and_join_epoll_threads());

// Stop workers

{

BAIDU_SCOPED_LOCK(_modify_group_mutex);

_stop = true;

_ngroup.exchange(0, butil::memory_order_relaxed);

}

for (int i = 0; i < PARKING_LOT_NUM; ++i) {

_pl[i].stop();

}

// Interrupt blocking operations.

for (size_t i = 0; i < _workers.size(); ++i) {

interrupt_pthread(_workers[i]);

}

// Join workers

for (size_t i = 0; i < _workers.size(); ++i) {

pthread_join(_workers[i], NULL);

}

if (__builtin_expect(NULL == g, 0)) {

return -1;

}

std::unique_lock<butil::Mutex> mu(_modify_group_mutex);

if (_stop) {

return -1;

}

size_t ngroup = _ngroup.load(butil::memory_order_relaxed);

if (ngroup < (size_t)BTHREAD_MAX_CONCURRENCY) {

_groups[ngroup] = g;

_ngroup.store(ngroup + 1, butil::memory_order_release);

}

mu.unlock();

// See the comments in _destroy_group

// TODO: Not needed anymore since non-worker pthread cannot have TaskGroup

signal_task(65536);

return 0;

if (NULL == g) {

LOG(ERROR) << "Param[g] is NULL";

return -1;

}

if (g->_control != this) {

LOG(ERROR) << "TaskGroup=" << g

<< " does not belong to this TaskControl=" << this;

return -1;

}

bool erased = false;

{

BAIDU_SCOPED_LOCK(_modify_group_mutex);

const size_t ngroup = _ngroup.load(butil::memory_order_relaxed);

for (size_t i = 0; i < ngroup; ++i) {

if (_groups[i] == g) {

// No need for atomic_thread_fence because lock did it.

_groups[i] = _groups[ngroup - 1];

// Change _ngroup and keep _groups unchanged at last so that:

// - If steal_task sees the newest _ngroup, it would not touch

// _groups[ngroup -1]

// - If steal_task sees old _ngroup and is still iterating on

// _groups, it would not miss _groups[ngroup - 1] which was

// swapped to _groups[i]. Although adding new group would

// overwrite it, since we do signal_task in _add_group(),

// we think the pending tasks of _groups[ngroup - 1] would

// not miss.

_ngroup.store(ngroup - 1, butil::memory_order_release);

//_groups[ngroup - 1] = NULL;

erased = true;

break;

}

}

}

// Can't delete g immediately because for performance consideration,

// we don't lock _modify_group_mutex in steal_task which may

// access the removed group concurrently. We use simple strategy here:

// Schedule a function which deletes the TaskGroup after

// FLAGS_task_group_delete_delay seconds

if (erased) {

get_global_timer_thread()->schedule(

delete_task_group, g,

butil::microseconds_from_now(FLAGS_task_group_delete_delay * 1000000L));

}

return 0;

// 1: Acquiring fence is paired with releasing fence in _add_group to

// avoid accessing uninitialized slot of _groups.

const size_t ngroup = _ngroup.load(butil::memory_order_acquire/*1*/);

if (0 == ngroup) {

return false;

}

// NOTE: Don't return inside `for' iteration since we need to update |seed|

bool stolen = false;

size_t s = *seed;

for (size_t i = 0; i < ngroup; ++i, s += offset) {

TaskGroup* g = _groups[s % ngroup];

// g is possibly NULL because of concurrent _destroy_group

if (g) {

if (g->_rq.steal(tid)) {

stolen = true;

break;

}

if (g->_remote_rq.pop(tid)) {

stolen = true;

break;

}

}

}

*seed = s;

return stolen;

if (num_task <= 0) {

return;

}

// TODO(gejun): Current algorithm does not guarantee enough threads will

// be created to match caller's requests. But in another side, there's also

// many useless signalings according to current impl. Capping the concurrency

// is a good balance between performance and timeliness of scheduling.

if (num_task > 2) {

num_task = 2;

}

int start_index = butil::fmix64(pthread_self()) % PARKING_LOT_NUM;

num_task -= _pl[start_index].signal(1);

if (num_task > 0) {

for (int i = 1; i < PARKING_LOT_NUM && num_task > 0; ++i) {

if (++start_index >= PARKING_LOT_NUM) {

start_index = 0;

}

num_task -= _pl[start_index].signal(1);

}

}

const size_t ngroup = _ngroup.load(butil::memory_order_relaxed);

DEFINE_SMALL_ARRAY(int, nums, ngroup, 128);

{

BAIDU_SCOPED_LOCK(_modify_group_mutex);

// ngroup > _ngroup: nums[_ngroup ... ngroup-1] = 0

// ngroup < _ngroup: just ignore _groups[_ngroup ... ngroup-1]

for (size_t i = 0; i < ngroup; ++i) {

nums[i] = (_groups[i] ? _groups[i]->_rq.volatile_size() : 0);

}

}

for (size_t i = 0; i < ngroup; ++i) {

os << nums[i] << ' ';

}

}

double TaskControl::get_cumulated_worker_time() {

int64_t cputime_ns = 0;

BAIDU_SCOPED_LOCK(_modify_group_mutex);

const size_t ngroup = _ngroup.load(butil::memory_order_relaxed);

for (size_t i = 0; i < ngroup; ++i) {

if (_groups[i]) {

cputime_ns += _groups[i]->_cumulated_cputime_ns;

}

}

return cputime_ns / 1000000000.0;

}

int64_t TaskControl::get_cumulated_switch_count() {

int64_t c = 0;

BAIDU_SCOPED_LOCK(_modify_group_mutex);

const size_t ngroup = _ngroup.load(butil::memory_order_relaxed);

for (size_t i = 0; i < ngroup; ++i) {

if (_groups[i]) {

c += _groups[i]->_nswitch;

}

}

return c;

}

int64_t TaskControl::get_cumulated_signal_count() {

int64_t c = 0;

BAIDU_SCOPED_LOCK(_modify_group_mutex);

const size_t ngroup = _ngroup.load(butil::memory_order_relaxed);

for (size_t i = 0; i < ngroup; ++i) {

TaskGroup* g = _groups[i];

if (g) {

c += g->_nsignaled + g->_remote_nsignaled;

}

}

return c;

}

bvar::LatencyRecorder* TaskControl::create_exposed_pending_time() {

bool is_creator = false;

_pending_time_mutex.lock();

bvar::LatencyRecorder* pt = _pending_time.load(butil::memory_order_consume);

if (!pt) {

pt = new bvar::LatencyRecorder;

_pending_time.store(pt, butil::memory_order_release);

is_creator = true;

}

_pending_time_mutex.unlock();

if (is_creator) {

pt->expose("bthread_creation");

}

return pt;

}