/src/brpc/src/bthread/bthread.cpp

Source (jump to first uncovered line)
// 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.

// Date: Tue Jul 10 17:40:58 CST 2012

#include <sys/syscall.h>
#include <gflags/gflags.h>
#include "butil/macros.h"                       // BAIDU_CASSERT
#include "butil/logging.h"
#include "butil/thread_local.h"
#include "butil/reloadable_flags.h"
#include "bthread/task_group.h"                // TaskGroup
#include "bthread/task_control.h"              // TaskControl
#include "bthread/timer_thread.h"
#include "bthread/list_of_abafree_id.h"
#include "bthread/bthread.h"

namespace bthread {

static bool validate_bthread_concurrency(const char*, int32_t val) {
    // bthread_setconcurrency sets the flag on success path which should
    // not be strictly in a validator. But it's OK for a int flag.
    return bthread_setconcurrency(val) == 0;
}
static bool validate_bthread_min_concurrency(const char*, int32_t val);
static bool validate_bthread_current_tag(const char*, int32_t val);
static bool validate_bthread_concurrency_by_tag(const char*, int32_t val);

DEFINE_int32(bthread_concurrency, 8 + BTHREAD_EPOLL_THREAD_NUM,
             "Number of pthread workers");
BUTIL_VALIDATE_GFLAG(bthread_concurrency, validate_bthread_concurrency);

DEFINE_int32(bthread_min_concurrency, 0,
            "Initial number of pthread workers which will be added on-demand."
            " The laziness is disabled when this value is non-positive,"
            " and workers will be created eagerly according to -bthread_concurrency and bthread_setconcurrency(). ");
BUTIL_VALIDATE_GFLAG(bthread_min_concurrency, validate_bthread_min_concurrency);

DEFINE_int32(bthread_current_tag, BTHREAD_TAG_INVALID, "Set bthread concurrency for this tag");
BUTIL_VALIDATE_GFLAG(bthread_current_tag, validate_bthread_current_tag);

DEFINE_int32(bthread_concurrency_by_tag, 8 + BTHREAD_EPOLL_THREAD_NUM,
             "Number of pthread workers of FLAGS_bthread_current_tag");
BUTIL_VALIDATE_GFLAG(bthread_concurrency_by_tag, validate_bthread_concurrency_by_tag);

static bool never_set_bthread_concurrency = true;

BAIDU_CASSERT(sizeof(TaskControl*) == sizeof(butil::atomic<TaskControl*>), atomic_size_match);

pthread_mutex_t g_task_control_mutex = PTHREAD_MUTEX_INITIALIZER;
// Referenced in rpc, needs to be extern.
// Notice that we can't declare the variable as atomic<TaskControl*> which
// are not constructed before main().
TaskControl* g_task_control = NULL;

extern BAIDU_THREAD_LOCAL TaskGroup* tls_task_group;
extern void (*g_worker_startfn)();
extern void (*g_tagged_worker_startfn)(bthread_tag_t);
extern void* (*g_create_span_func)();

inline TaskControl* get_task_control() {
    return g_task_control;
}

inline TaskControl* get_or_new_task_control() {
    butil::atomic<TaskControl*>* p = (butil::atomic<TaskControl*>*)&g_task_control;
    TaskControl* c = p->load(butil::memory_order_consume);
    if (c != NULL) {
        return c;
    }
    BAIDU_SCOPED_LOCK(g_task_control_mutex);
    c = p->load(butil::memory_order_consume);
    if (c != NULL) {
        return c;
    }
    c = new (std::nothrow) TaskControl;
    if (NULL == c) {
        return NULL;
    }
    int concurrency = FLAGS_bthread_min_concurrency > 0 ?
        FLAGS_bthread_min_concurrency :
        FLAGS_bthread_concurrency;
    if (c->init(concurrency) != 0) {
        LOG(ERROR) << "Fail to init g_task_control";
        delete c;
        return NULL;
    }
    p->store(c, butil::memory_order_release);
    return c;
}

#ifdef BRPC_BTHREAD_TRACER
BAIDU_THREAD_LOCAL TaskMeta* pthread_fake_meta = NULL;

bthread_t init_for_pthread_stack_trace() {
    if (NULL != pthread_fake_meta) {
        return pthread_fake_meta->tid;
    }

    TaskControl* c = get_task_control();
    if (NULL == c) {
        LOG(ERROR) << "TaskControl has not been created, "
                      "please use bthread_start_xxx before call this function";
        return INVALID_BTHREAD;
    }

    butil::ResourceId<TaskMeta> slot;
    pthread_fake_meta = butil::get_resource(&slot);
    if (BAIDU_UNLIKELY(NULL == pthread_fake_meta)) {
        LOG(ERROR) << "Fail to get TaskMeta";
        return INVALID_BTHREAD;
    }

    pthread_fake_meta->attr = BTHREAD_ATTR_PTHREAD;
    pthread_fake_meta->tid = make_tid(*pthread_fake_meta->version_butex, slot);
    // Make TaskTracer use signal trace mode for pthread.
    c->_task_tracer.set_running_status(syscall(SYS_gettid), pthread_fake_meta);

    // Release the TaskMeta at exit of pthread.
    butil::thread_atexit([]() {
        // Similar to TaskGroup::task_runner.
        bool tracing;
        {
            BAIDU_SCOPED_LOCK(pthread_fake_meta->version_lock);
            tracing = TaskTracer::set_end_status_unsafe(pthread_fake_meta);
            // If resulting version is 0,
            // change it to 1 to make bthread_t never be 0.
            if (0 == ++*pthread_fake_meta->version_butex) {
                ++*pthread_fake_meta->version_butex;
            }
        }

        if (tracing) {
            // Wait for tracing completion.
            get_task_control()->_task_tracer.WaitForTracing(pthread_fake_meta);
        }
        get_task_control()->_task_tracer.set_status(
            TASK_STATUS_UNKNOWN, pthread_fake_meta);

        butil::return_resource(get_slot(pthread_fake_meta->tid));
        pthread_fake_meta = NULL;
    });

    return pthread_fake_meta->tid;
}

void stack_trace(std::ostream& os, bthread_t tid) {
    TaskControl* c = get_task_control();
    if (NULL == c) {
        os << "TaskControl has not been created";
        return;
    }
    c->stack_trace(os, tid);
}

std::string stack_trace(bthread_t tid) {
    TaskControl* c = get_task_control();
    if (NULL == c) {
        return "TaskControl has not been created";
    }
    return c->stack_trace(tid);
}
#endif // BRPC_BTHREAD_TRACER

static int add_workers_for_each_tag(int num) {
    int added = 0;
    auto c = get_task_control();
    for (auto i = 0; i < num; ++i) {
        added += c->add_workers(1, i % FLAGS_task_group_ntags);
    }
    return added;
}

static bool validate_bthread_min_concurrency(const char*, int32_t val) {
    if (val <= 0) {
        return true;
    }
    if (val < BTHREAD_MIN_CONCURRENCY || val > FLAGS_bthread_concurrency) {
        return false;
    }
    TaskControl* c = get_task_control();
    if (!c) {
        return true;
    }
    BAIDU_SCOPED_LOCK(g_task_control_mutex);
    int concurrency = c->concurrency();
    if (val > concurrency) {
        int added = bthread::add_workers_for_each_tag(val - concurrency);
        return added == (val - concurrency);
    } else {
        return true;
    }
}

static bool validate_bthread_current_tag(const char*, int32_t val) {
    if (val == BTHREAD_TAG_INVALID) {
        return true;
    } else if (val < BTHREAD_TAG_DEFAULT || val >= FLAGS_task_group_ntags) {
        return false;
    }
    BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
    auto c = bthread::get_task_control();
    if (c == NULL) {
        FLAGS_bthread_concurrency_by_tag = 8 + BTHREAD_EPOLL_THREAD_NUM;
        return true;
    }
    FLAGS_bthread_concurrency_by_tag = c->concurrency(val);
    return true;
}

static bool validate_bthread_concurrency_by_tag(const char*, int32_t val) {
    return bthread_setconcurrency_by_tag(val, FLAGS_bthread_current_tag) == 0;
}

__thread TaskGroup* tls_task_group_nosignal = NULL;

BUTIL_FORCE_INLINE int
start_from_non_worker(bthread_t* __restrict tid,
                      const bthread_attr_t* __restrict attr,
                      void* (*fn)(void*),
                      void* __restrict arg) {
    TaskControl* c = get_or_new_task_control();
    if (NULL == c) {
        return ENOMEM;
    }
    auto tag = BTHREAD_TAG_DEFAULT;
    if (attr != NULL && attr->tag != BTHREAD_TAG_INVALID) {
        tag = attr->tag;
    }
    if (attr != NULL && (attr->flags & BTHREAD_NOSIGNAL)) {
        // Remember the TaskGroup to insert NOSIGNAL tasks for 2 reasons:
        // 1. NOSIGNAL is often for creating many bthreads in batch,
        //    inserting into the same TaskGroup maximizes the batch.
        // 2. bthread_flush() needs to know which TaskGroup to flush.
        auto g = tls_task_group_nosignal;
        if (NULL == g) {
            g = c->choose_one_group(tag);
            tls_task_group_nosignal = g;
        }
        return g->start_background<true>(tid, attr, fn, arg);
    }
    return c->choose_one_group(tag)->start_background<true>(tid, attr, fn, arg);
}

// Meet one of the three conditions, can run in thread local
// attr is nullptr
// tag equal to thread local
// tag equal to BTHREAD_TAG_INVALID
BUTIL_FORCE_INLINE bool can_run_thread_local(const bthread_attr_t* __restrict attr) {
    return attr == nullptr || attr->tag == bthread::tls_task_group->tag() ||
           attr->tag == BTHREAD_TAG_INVALID;
}

struct TidTraits {
    static const size_t BLOCK_SIZE = 63;
    static const size_t MAX_ENTRIES = 65536;
    static const size_t INIT_GC_SIZE = 65536;
    static const bthread_t ID_INIT;
    static bool exists(bthread_t id) { return bthread::TaskGroup::exists(id); }
};
const bthread_t TidTraits::ID_INIT = INVALID_BTHREAD;

typedef ListOfABAFreeId<bthread_t, TidTraits> TidList;

struct TidStopper {
    void operator()(bthread_t id) const { bthread_stop(id); }
};
struct TidJoiner {
    void operator()(bthread_t & id) const {
        bthread_join(id, NULL);
        id = INVALID_BTHREAD;
    }
};

}  // namespace bthread

extern "C" {

int bthread_start_urgent(bthread_t* __restrict tid,
                         const bthread_attr_t* __restrict attr,
                         void * (*fn)(void*),
                         void* __restrict arg) {
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (g) {
        // if attribute is null use thread local task group
        if (bthread::can_run_thread_local(attr)) {
            return bthread::TaskGroup::start_foreground(&g, tid, attr, fn, arg);
        }
    }
    return bthread::start_from_non_worker(tid, attr, fn, arg);
}

int bthread_start_background(bthread_t* __restrict tid,
                             const bthread_attr_t* __restrict attr,
                             void * (*fn)(void*),
                             void* __restrict arg) {
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (g) {
        // if attribute is null use thread local task group
        if (bthread::can_run_thread_local(attr)) {
            return g->start_background<false>(tid, attr, fn, arg);
        }
    }
    return bthread::start_from_non_worker(tid, attr, fn, arg);
}

void bthread_flush() {
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (g) {
        return g->flush_nosignal_tasks();
    }
    g = bthread::tls_task_group_nosignal;
    if (g) {
        // NOSIGNAL tasks were created in this non-worker.
        bthread::tls_task_group_nosignal = NULL;
        return g->flush_nosignal_tasks_remote();
    }
}

int bthread_interrupt(bthread_t tid, bthread_tag_t tag) {
    return bthread::TaskGroup::interrupt(tid, bthread::get_task_control(), tag);
}

int bthread_stop(bthread_t tid) {
    bthread::TaskGroup::set_stopped(tid);
    return bthread_interrupt(tid);
}

int bthread_stopped(bthread_t tid) {
    return (int)bthread::TaskGroup::is_stopped(tid);
}

bthread_t bthread_self(void) {
    bthread::TaskGroup* g = bthread::tls_task_group;
    // note: return 0 for main tasks now, which include main thread and
    // all work threads. So that we can identify main tasks from logs
    // more easily. This is probably questionable in future.
    if (g != NULL && !g->is_current_main_task()/*note*/) {
        return g->current_tid();
    }
    return INVALID_BTHREAD;
}

int bthread_equal(bthread_t t1, bthread_t t2) {
    return t1 == t2;
}

void bthread_exit(void* retval) {
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (g != NULL && !g->is_current_main_task()) {
        throw bthread::ExitException(retval);
    } else {
        pthread_exit(retval);
    }
}

int bthread_join(bthread_t tid, void** thread_return) {
    return bthread::TaskGroup::join(tid, thread_return);
}

int bthread_attr_init(bthread_attr_t* a) {
    *a = BTHREAD_ATTR_NORMAL;
    return 0;
}

int bthread_attr_destroy(bthread_attr_t*) {
    return 0;
}

int bthread_getattr(bthread_t tid, bthread_attr_t* attr) {
    return bthread::TaskGroup::get_attr(tid, attr);
}

int bthread_getconcurrency(void) {
    return bthread::FLAGS_bthread_concurrency;
}

int bthread_setconcurrency(int num) {
    if (num < BTHREAD_MIN_CONCURRENCY || num > BTHREAD_MAX_CONCURRENCY) {
        LOG(ERROR) << "Invalid concurrency=" << num;
        return EINVAL;
    }
    if (bthread::FLAGS_bthread_min_concurrency > 0) {
        if (num < bthread::FLAGS_bthread_min_concurrency) {
            return EINVAL;
        }
        if (bthread::never_set_bthread_concurrency) {
            bthread::never_set_bthread_concurrency = false;
        }
        bthread::FLAGS_bthread_concurrency = num;
        return 0;
    }
    bthread::TaskControl* c = bthread::get_task_control();
    if (c != NULL) {
        if (num < c->concurrency()) {
            return EPERM;
        } else if (num == c->concurrency()) {
            return 0;
        }
    }
    BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
    c = bthread::get_task_control();
    if (c == NULL) {
        if (bthread::never_set_bthread_concurrency) {
            bthread::never_set_bthread_concurrency = false;
            bthread::FLAGS_bthread_concurrency = num;
        } else if (num > bthread::FLAGS_bthread_concurrency) {
            bthread::FLAGS_bthread_concurrency = num;
        }
        return 0;
    }
    if (bthread::FLAGS_bthread_concurrency != c->concurrency()) {
        LOG(ERROR) << "CHECK failed: bthread_concurrency="
                   << bthread::FLAGS_bthread_concurrency
                   << " != tc_concurrency=" << c->concurrency();
        bthread::FLAGS_bthread_concurrency = c->concurrency();
    }
    if (num > bthread::FLAGS_bthread_concurrency) {
        // Create more workers if needed.
        auto added = bthread::add_workers_for_each_tag(num - bthread::FLAGS_bthread_concurrency);
        bthread::FLAGS_bthread_concurrency += added;
    }
    return (num == bthread::FLAGS_bthread_concurrency ? 0 : EPERM);
}

int bthread_getconcurrency_by_tag(bthread_tag_t tag) {
    BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
    auto c = bthread::get_task_control();
    if (c == NULL) {
        return EPERM;
    }
    return c->concurrency(tag);
}

int bthread_setconcurrency_by_tag(int num, bthread_tag_t tag) {
    if (tag == BTHREAD_TAG_INVALID) {
        return 0;
    } else if (tag < BTHREAD_TAG_DEFAULT || tag >= FLAGS_task_group_ntags) {
        return EINVAL;
    }
    if (num < BTHREAD_MIN_CONCURRENCY || num > BTHREAD_MAX_CONCURRENCY) {
        LOG(ERROR) << "Invalid concurrency_by_tag=" << num;
        return EINVAL;
    }
    auto c = bthread::get_or_new_task_control();
    BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
    auto tag_ngroup = c->concurrency(tag);
    auto add = num - tag_ngroup;

    if (add >= 0) {
        auto added = c->add_workers(add, tag);
        bthread::FLAGS_bthread_concurrency += added;
        return (add == added ? 0 : EPERM);
    } else {
        LOG(WARNING) << "Fail to set concurrency by tag: " << tag
                     << ", tag concurrency should be larger than old oncurrency. old concurrency: "
                     << tag_ngroup << ", new concurrency: " << num;
        return EPERM;
    }
}

int bthread_about_to_quit() {
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (g != NULL) {
        bthread::TaskMeta* current_task = g->current_task();
        if(!(current_task->attr.flags & BTHREAD_NEVER_QUIT)) {
            current_task->about_to_quit = true;
        }
        return 0;
    }
    return EPERM;
}

int bthread_timer_add(bthread_timer_t* id, timespec abstime,
                      void (*on_timer)(void*), void* arg) {
    bthread::TaskControl* c = bthread::get_or_new_task_control();
    if (c == NULL) {
        return ENOMEM;
    }
    bthread::TimerThread* tt = bthread::get_or_create_global_timer_thread();
    if (tt == NULL) {
        return ENOMEM;
    }
    bthread_timer_t tmp = tt->schedule(on_timer, arg, abstime);
    if (tmp != 0) {
        *id = tmp;
        return 0;
    }
    return ESTOP;
}

int bthread_timer_del(bthread_timer_t id) {
    bthread::TaskControl* c = bthread::get_task_control();
    if (c != NULL) {
        bthread::TimerThread* tt = bthread::get_global_timer_thread();
        if (tt == NULL) {
            return EINVAL;
        }
        const int state = tt->unschedule(id);
        if (state >= 0) {
            return state;
        }
    }
    return EINVAL;
}

int bthread_usleep(uint64_t microseconds) {
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (NULL != g && !g->is_current_pthread_task()) {
        return bthread::TaskGroup::usleep(&g, microseconds);
    }
    return ::usleep(microseconds);
}

int bthread_yield(void) {
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (NULL != g && !g->is_current_pthread_task()) {
        bthread::TaskGroup::yield(&g);
        return 0;
    }
    // pthread_yield is not available on MAC
    return sched_yield();
}

int bthread_set_worker_startfn(void (*start_fn)()) {
    if (start_fn == NULL) {
        return EINVAL;
    }
    bthread::g_worker_startfn = start_fn;
    return 0;
}

int bthread_set_tagged_worker_startfn(void (*start_fn)(bthread_tag_t)) {
    if (start_fn == NULL) {
        return EINVAL;
    }
    bthread::g_tagged_worker_startfn = start_fn;
    return 0;
}

int bthread_set_create_span_func(void* (*func)()) {
    if (func == NULL) {
        return EINVAL;
    }
    bthread::g_create_span_func = func;
    return 0;
}

void bthread_stop_world() {
    bthread::TaskControl* c = bthread::get_task_control();
    if (c != NULL) {
        c->stop_and_join();
    }
}

int bthread_list_init(bthread_list_t* list,
                      unsigned /*size*/,
                      unsigned /*conflict_size*/) {
    list->impl = new (std::nothrow) bthread::TidList;
    if (NULL == list->impl) {
        return ENOMEM;
    }
    // Set unused fields to zero as well.
    list->head = 0;
    list->size = 0;
    list->conflict_head = 0;
    list->conflict_size = 0;
    return 0;
}

void bthread_list_destroy(bthread_list_t* list) {
    delete static_cast<bthread::TidList*>(list->impl);
    list->impl = NULL;
}

int bthread_list_add(bthread_list_t* list, bthread_t id) {
    if (list->impl == NULL) {
        return EINVAL;
    }
    return static_cast<bthread::TidList*>(list->impl)->add(id);
}

int bthread_list_stop(bthread_list_t* list) {
    if (list->impl == NULL) {
        return EINVAL;
    }
    static_cast<bthread::TidList*>(list->impl)->apply(bthread::TidStopper());
    return 0;
}

int bthread_list_join(bthread_list_t* list) {
    if (list->impl == NULL) {
        return EINVAL;
    }
    static_cast<bthread::TidList*>(list->impl)->apply(bthread::TidJoiner());
    return 0;
}

bthread_tag_t bthread_self_tag(void) {
    return bthread::tls_task_group != nullptr ? bthread::tls_task_group->tag()
                                              : BTHREAD_TAG_DEFAULT;
}

uint64_t bthread_cpu_clock_ns(void) {
     bthread::TaskGroup* g = bthread::tls_task_group;
    if (g != NULL && !g->is_current_main_task()) {
        return g->current_task_cpu_clock_ns();
    }
    return 0;
}

}  // extern "C"

Coverage Report

Created: 2025-03-11 06:06

Line	Count	Source (jump to first uncovered line)
1		// Licensed to the Apache Software Foundation (ASF) under one
2		// or more contributor license agreements. See the NOTICE file
3		// distributed with this work for additional information
4		// regarding copyright ownership. The ASF licenses this file
5		// to you under the Apache License, Version 2.0 (the
6		// "License"); you may not use this file except in compliance
7		// with the License. You may obtain a copy of the License at
8		//
9		// http://www.apache.org/licenses/LICENSE-2.0
10		//
11		// Unless required by applicable law or agreed to in writing,
12		// software distributed under the License is distributed on an
13		// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14		// KIND, either express or implied. See the License for the
15		// specific language governing permissions and limitations
16		// under the License.
17
18		// bthread - An M:N threading library to make applications more concurrent.
19
20		// Date: Tue Jul 10 17:40:58 CST 2012
21
22		#include <sys/syscall.h>
23		#include <gflags/gflags.h>
24		#include "butil/macros.h" // BAIDU_CASSERT
25		#include "butil/logging.h"
26		#include "butil/thread_local.h"
27		#include "butil/reloadable_flags.h"
28		#include "bthread/task_group.h" // TaskGroup
29		#include "bthread/task_control.h" // TaskControl
30		#include "bthread/timer_thread.h"
31		#include "bthread/list_of_abafree_id.h"
32		#include "bthread/bthread.h"
33
34		namespace bthread {
35
36	0	static bool validate_bthread_concurrency(const char*, int32_t val) {
37		// bthread_setconcurrency sets the flag on success path which should
38		// not be strictly in a validator. But it's OK for a int flag.
39	0	return bthread_setconcurrency(val) == 0;
40	0	}
41		static bool validate_bthread_min_concurrency(const char*, int32_t val);
42		static bool validate_bthread_current_tag(const char*, int32_t val);
43		static bool validate_bthread_concurrency_by_tag(const char*, int32_t val);
44
45		DEFINE_int32(bthread_concurrency, 8 + BTHREAD_EPOLL_THREAD_NUM,
46		"Number of pthread workers");
47		BUTIL_VALIDATE_GFLAG(bthread_concurrency, validate_bthread_concurrency);
48
49		DEFINE_int32(bthread_min_concurrency, 0,
50		"Initial number of pthread workers which will be added on-demand."
51		" The laziness is disabled when this value is non-positive,"
52		" and workers will be created eagerly according to -bthread_concurrency and bthread_setconcurrency(). ");
53		BUTIL_VALIDATE_GFLAG(bthread_min_concurrency, validate_bthread_min_concurrency);
54
55		DEFINE_int32(bthread_current_tag, BTHREAD_TAG_INVALID, "Set bthread concurrency for this tag");
56		BUTIL_VALIDATE_GFLAG(bthread_current_tag, validate_bthread_current_tag);
57
58		DEFINE_int32(bthread_concurrency_by_tag, 8 + BTHREAD_EPOLL_THREAD_NUM,
59		"Number of pthread workers of FLAGS_bthread_current_tag");
60		BUTIL_VALIDATE_GFLAG(bthread_concurrency_by_tag, validate_bthread_concurrency_by_tag);
61
62		static bool never_set_bthread_concurrency = true;
63
64		BAIDU_CASSERT(sizeof(TaskControl) == sizeof(butil::atomic<TaskControl>), atomic_size_match);
65
66		pthread_mutex_t g_task_control_mutex = PTHREAD_MUTEX_INITIALIZER;
67		// Referenced in rpc, needs to be extern.
68		// Notice that we can't declare the variable as atomic<TaskControl*> which
69		// are not constructed before main().
70		TaskControl* g_task_control = NULL;
71
72		extern BAIDU_THREAD_LOCAL TaskGroup* tls_task_group;
73		extern void (*g_worker_startfn)();
74		extern void (*g_tagged_worker_startfn)(bthread_tag_t);
75		extern void* (*g_create_span_func)();
76
77	0	inline TaskControl* get_task_control() {
78	0	return g_task_control;
79	0	}
80
81	0	inline TaskControl* get_or_new_task_control() {
82	0	butil::atomic<TaskControl> p = (butil::atomic<TaskControl>)&g_task_control;
83	0	TaskControl* c = p->load(butil::memory_order_consume);
84	0	if (c != NULL) {
85	0	return c;
86	0	}
87	0	BAIDU_SCOPED_LOCK(g_task_control_mutex);
88	0	c = p->load(butil::memory_order_consume);
89	0	if (c != NULL) {
90	0	return c;
91	0	}
92	0	c = new (std::nothrow) TaskControl;
93	0	if (NULL == c) {
94	0	return NULL;
95	0	}
96	0	int concurrency = FLAGS_bthread_min_concurrency > 0 ?
97	0	FLAGS_bthread_min_concurrency :
98	0	FLAGS_bthread_concurrency;
99	0	if (c->init(concurrency) != 0) {
100	0	LOG(ERROR) << "Fail to init g_task_control";
101	0	delete c;
102	0	return NULL;
103	0	}
104	0	p->store(c, butil::memory_order_release);
105	0	return c;
106	0	}
107
108		#ifdef BRPC_BTHREAD_TRACER
109		BAIDU_THREAD_LOCAL TaskMeta* pthread_fake_meta = NULL;
110
111		bthread_t init_for_pthread_stack_trace() {
112		if (NULL != pthread_fake_meta) {
113		return pthread_fake_meta->tid;
114		}
115
116		TaskControl* c = get_task_control();
117		if (NULL == c) {
118		LOG(ERROR) << "TaskControl has not been created, "
119		"please use bthread_start_xxx before call this function";
120		return INVALID_BTHREAD;
121		}
122
123		butil::ResourceId<TaskMeta> slot;
124		pthread_fake_meta = butil::get_resource(&slot);
125		if (BAIDU_UNLIKELY(NULL == pthread_fake_meta)) {
126		LOG(ERROR) << "Fail to get TaskMeta";
127		return INVALID_BTHREAD;
128		}
129
130		pthread_fake_meta->attr = BTHREAD_ATTR_PTHREAD;
131		pthread_fake_meta->tid = make_tid(*pthread_fake_meta->version_butex, slot);
132		// Make TaskTracer use signal trace mode for pthread.
133		c->_task_tracer.set_running_status(syscall(SYS_gettid), pthread_fake_meta);
134
135		// Release the TaskMeta at exit of pthread.
136		butil::thread_atexit([]() {
137		// Similar to TaskGroup::task_runner.
138		bool tracing;
139		{
140		BAIDU_SCOPED_LOCK(pthread_fake_meta->version_lock);
141		tracing = TaskTracer::set_end_status_unsafe(pthread_fake_meta);
142		// If resulting version is 0,
143		// change it to 1 to make bthread_t never be 0.
144		if (0 == ++*pthread_fake_meta->version_butex) {
145		++*pthread_fake_meta->version_butex;
146		}
147		}
148
149		if (tracing) {
150		// Wait for tracing completion.
151		get_task_control()->_task_tracer.WaitForTracing(pthread_fake_meta);
152		}
153		get_task_control()->_task_tracer.set_status(
154		TASK_STATUS_UNKNOWN, pthread_fake_meta);
155
156		butil::return_resource(get_slot(pthread_fake_meta->tid));
157		pthread_fake_meta = NULL;
158		});
159
160		return pthread_fake_meta->tid;
161		}
162
163		void stack_trace(std::ostream& os, bthread_t tid) {
164		TaskControl* c = get_task_control();
165		if (NULL == c) {
166		os << "TaskControl has not been created";
167		return;
168		}
169		c->stack_trace(os, tid);
170		}
171
172		std::string stack_trace(bthread_t tid) {
173		TaskControl* c = get_task_control();
174		if (NULL == c) {
175		return "TaskControl has not been created";
176		}
177		return c->stack_trace(tid);
178		}
179		#endif // BRPC_BTHREAD_TRACER
180
181	0	static int add_workers_for_each_tag(int num) {
182	0	int added = 0;
183	0	auto c = get_task_control();
184	0	for (auto i = 0; i < num; ++i) {
185	0	added += c->add_workers(1, i % FLAGS_task_group_ntags);
186	0	}
187	0	return added;
188	0	}
189
190	0	static bool validate_bthread_min_concurrency(const char*, int32_t val) {
191	0	if (val <= 0) {
192	0	return true;
193	0	}
194	0	if (val < BTHREAD_MIN_CONCURRENCY \|\| val > FLAGS_bthread_concurrency) {
195	0	return false;
196	0	}
197	0	TaskControl* c = get_task_control();
198	0	if (!c) {
199	0	return true;
200	0	}
201	0	BAIDU_SCOPED_LOCK(g_task_control_mutex);
202	0	int concurrency = c->concurrency();
203	0	if (val > concurrency) {
204	0	int added = bthread::add_workers_for_each_tag(val - concurrency);
205	0	return added == (val - concurrency);
206	0	} else {
207	0	return true;
208	0	}
209	0	}
210
211	0	static bool validate_bthread_current_tag(const char*, int32_t val) {
212	0	if (val == BTHREAD_TAG_INVALID) {
213	0	return true;
214	0	} else if (val < BTHREAD_TAG_DEFAULT \|\| val >= FLAGS_task_group_ntags) {
215	0	return false;
216	0	}
217	0	BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
218	0	auto c = bthread::get_task_control();
219	0	if (c == NULL) {
220	0	FLAGS_bthread_concurrency_by_tag = 8 + BTHREAD_EPOLL_THREAD_NUM;
221	0	return true;
222	0	}
223	0	FLAGS_bthread_concurrency_by_tag = c->concurrency(val);
224	0	return true;
225	0	}
226
227	0	static bool validate_bthread_concurrency_by_tag(const char*, int32_t val) {
228	0	return bthread_setconcurrency_by_tag(val, FLAGS_bthread_current_tag) == 0;
229	0	}
230
231		__thread TaskGroup* tls_task_group_nosignal = NULL;
232
233		BUTIL_FORCE_INLINE int
234		start_from_non_worker(bthread_t* __restrict tid,
235		const bthread_attr_t* __restrict attr,
236		void* (fn)(void),
237	0	void* __restrict arg) {
238	0	TaskControl* c = get_or_new_task_control();
239	0	if (NULL == c) {
240	0	return ENOMEM;
241	0	}
242	0	auto tag = BTHREAD_TAG_DEFAULT;
243	0	if (attr != NULL && attr->tag != BTHREAD_TAG_INVALID) {
244	0	tag = attr->tag;
245	0	}
246	0	if (attr != NULL && (attr->flags & BTHREAD_NOSIGNAL)) {
247		// Remember the TaskGroup to insert NOSIGNAL tasks for 2 reasons:
248		// 1. NOSIGNAL is often for creating many bthreads in batch,
249		// inserting into the same TaskGroup maximizes the batch.
250		// 2. bthread_flush() needs to know which TaskGroup to flush.
251	0	auto g = tls_task_group_nosignal;
252	0	if (NULL == g) {
253	0	g = c->choose_one_group(tag);
254	0	tls_task_group_nosignal = g;
255	0	}
256	0	return g->start_background<true>(tid, attr, fn, arg);
257	0	}
258	0	return c->choose_one_group(tag)->start_background<true>(tid, attr, fn, arg);
259	0	}
260
261		// Meet one of the three conditions, can run in thread local
262		// attr is nullptr
263		// tag equal to thread local
264		// tag equal to BTHREAD_TAG_INVALID
265	0	BUTIL_FORCE_INLINE bool can_run_thread_local(const bthread_attr_t* __restrict attr) {
266	0	return attr == nullptr \|\| attr->tag == bthread::tls_task_group->tag() \|\|
267	0	attr->tag == BTHREAD_TAG_INVALID;
268	0	}
269
270		struct TidTraits {
271		static const size_t BLOCK_SIZE = 63;
272		static const size_t MAX_ENTRIES = 65536;
273		static const size_t INIT_GC_SIZE = 65536;
274		static const bthread_t ID_INIT;
275	0	static bool exists(bthread_t id) { return bthread::TaskGroup::exists(id); }
276		};
277		const bthread_t TidTraits::ID_INIT = INVALID_BTHREAD;
278
279		typedef ListOfABAFreeId<bthread_t, TidTraits> TidList;
280
281		struct TidStopper {
282	0	void operator()(bthread_t id) const { bthread_stop(id); }
283		};
284		struct TidJoiner {
285	0	void operator()(bthread_t & id) const {
286	0	bthread_join(id, NULL);
287	0	id = INVALID_BTHREAD;
288	0	}
289		};
290
291		} // namespace bthread
292
293		extern "C" {
294
295		int bthread_start_urgent(bthread_t* __restrict tid,
296		const bthread_attr_t* __restrict attr,
297		void * (fn)(void),
298	0	void* __restrict arg) {
299	0	bthread::TaskGroup* g = bthread::tls_task_group;
300	0	if (g) {
301		// if attribute is null use thread local task group
302	0	if (bthread::can_run_thread_local(attr)) {
303	0	return bthread::TaskGroup::start_foreground(&g, tid, attr, fn, arg);
304	0	}
305	0	}
306	0	return bthread::start_from_non_worker(tid, attr, fn, arg);
307	0	}
308
309		int bthread_start_background(bthread_t* __restrict tid,
310		const bthread_attr_t* __restrict attr,
311		void * (fn)(void),
312	0	void* __restrict arg) {
313	0	bthread::TaskGroup* g = bthread::tls_task_group;
314	0	if (g) {
315		// if attribute is null use thread local task group
316	0	if (bthread::can_run_thread_local(attr)) {
317	0	return g->start_background<false>(tid, attr, fn, arg);
318	0	}
319	0	}
320	0	return bthread::start_from_non_worker(tid, attr, fn, arg);
321	0	}
322
323	0	void bthread_flush() {
324	0	bthread::TaskGroup* g = bthread::tls_task_group;
325	0	if (g) {
326	0	return g->flush_nosignal_tasks();
327	0	}
328	0	g = bthread::tls_task_group_nosignal;
329	0	if (g) {
330		// NOSIGNAL tasks were created in this non-worker.
331	0	bthread::tls_task_group_nosignal = NULL;
332	0	return g->flush_nosignal_tasks_remote();
333	0	}
334	0	}
335
336	0	int bthread_interrupt(bthread_t tid, bthread_tag_t tag) {
337	0	return bthread::TaskGroup::interrupt(tid, bthread::get_task_control(), tag);
338	0	}
339
340	0	int bthread_stop(bthread_t tid) {
341	0	bthread::TaskGroup::set_stopped(tid);
342	0	return bthread_interrupt(tid);
343	0	}
344
345	0	int bthread_stopped(bthread_t tid) {
346	0	return (int)bthread::TaskGroup::is_stopped(tid);
347	0	}
348
349	23.5k	bthread_t bthread_self(void) {
350	23.5k	bthread::TaskGroup* g = bthread::tls_task_group;
351		// note: return 0 for main tasks now, which include main thread and
352		// all work threads. So that we can identify main tasks from logs
353		// more easily. This is probably questionable in future.
354	23.5k	if (g != NULL && !g->is_current_main_task()/note/) {
355	0	return g->current_tid();
356	0	}
357	23.5k	return INVALID_BTHREAD;
358	23.5k	}
359
360	0	int bthread_equal(bthread_t t1, bthread_t t2) {
361	0	return t1 == t2;
362	0	}
363
364	0	void bthread_exit(void* retval) {
365	0	bthread::TaskGroup* g = bthread::tls_task_group;
366	0	if (g != NULL && !g->is_current_main_task()) {
367	0	throw bthread::ExitException(retval);
368	0	} else {
369	0	pthread_exit(retval);
370	0	}
371	0	}
372
373	0	int bthread_join(bthread_t tid, void** thread_return) {
374	0	return bthread::TaskGroup::join(tid, thread_return);
375	0	}
376
377	0	int bthread_attr_init(bthread_attr_t* a) {
378	0	*a = BTHREAD_ATTR_NORMAL;
379	0	return 0;
380	0	}
381
382	0	int bthread_attr_destroy(bthread_attr_t*) {
383	0	return 0;
384	0	}
385
386	0	int bthread_getattr(bthread_t tid, bthread_attr_t* attr) {
387	0	return bthread::TaskGroup::get_attr(tid, attr);
388	0	}
389
390	0	int bthread_getconcurrency(void) {
391	0	return bthread::FLAGS_bthread_concurrency;
392	0	}
393
394	0	int bthread_setconcurrency(int num) {
395	0	if (num < BTHREAD_MIN_CONCURRENCY \|\| num > BTHREAD_MAX_CONCURRENCY) {
396	0	LOG(ERROR) << "Invalid concurrency=" << num;
397	0	return EINVAL;
398	0	}
399	0	if (bthread::FLAGS_bthread_min_concurrency > 0) {
400	0	if (num < bthread::FLAGS_bthread_min_concurrency) {
401	0	return EINVAL;
402	0	}
403	0	if (bthread::never_set_bthread_concurrency) {
404	0	bthread::never_set_bthread_concurrency = false;
405	0	}
406	0	bthread::FLAGS_bthread_concurrency = num;
407	0	return 0;
408	0	}
409	0	bthread::TaskControl* c = bthread::get_task_control();
410	0	if (c != NULL) {
411	0	if (num < c->concurrency()) {
412	0	return EPERM;
413	0	} else if (num == c->concurrency()) {
414	0	return 0;
415	0	}
416	0	}
417	0	BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
418	0	c = bthread::get_task_control();
419	0	if (c == NULL) {
420	0	if (bthread::never_set_bthread_concurrency) {
421	0	bthread::never_set_bthread_concurrency = false;
422	0	bthread::FLAGS_bthread_concurrency = num;
423	0	} else if (num > bthread::FLAGS_bthread_concurrency) {
424	0	bthread::FLAGS_bthread_concurrency = num;
425	0	}
426	0	return 0;
427	0	}
428	0	if (bthread::FLAGS_bthread_concurrency != c->concurrency()) {
429	0	LOG(ERROR) << "CHECK failed: bthread_concurrency="
430	0	<< bthread::FLAGS_bthread_concurrency
431	0	<< " != tc_concurrency=" << c->concurrency();
432	0	bthread::FLAGS_bthread_concurrency = c->concurrency();
433	0	}
434	0	if (num > bthread::FLAGS_bthread_concurrency) {
435		// Create more workers if needed.
436	0	auto added = bthread::add_workers_for_each_tag(num - bthread::FLAGS_bthread_concurrency);
437	0	bthread::FLAGS_bthread_concurrency += added;
438	0	}
439	0	return (num == bthread::FLAGS_bthread_concurrency ? 0 : EPERM);
440	0	}
441
442	0	int bthread_getconcurrency_by_tag(bthread_tag_t tag) {
443	0	BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
444	0	auto c = bthread::get_task_control();
445	0	if (c == NULL) {
446	0	return EPERM;
447	0	}
448	0	return c->concurrency(tag);
449	0	}
450
451	0	int bthread_setconcurrency_by_tag(int num, bthread_tag_t tag) {
452	0	if (tag == BTHREAD_TAG_INVALID) {
453	0	return 0;
454	0	} else if (tag < BTHREAD_TAG_DEFAULT \|\| tag >= FLAGS_task_group_ntags) {
455	0	return EINVAL;
456	0	}
457	0	if (num < BTHREAD_MIN_CONCURRENCY \|\| num > BTHREAD_MAX_CONCURRENCY) {
458	0	LOG(ERROR) << "Invalid concurrency_by_tag=" << num;
459	0	return EINVAL;
460	0	}
461	0	auto c = bthread::get_or_new_task_control();
462	0	BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
463	0	auto tag_ngroup = c->concurrency(tag);
464	0	auto add = num - tag_ngroup;
465
466	0	if (add >= 0) {
467	0	auto added = c->add_workers(add, tag);
468	0	bthread::FLAGS_bthread_concurrency += added;
469	0	return (add == added ? 0 : EPERM);
470	0	} else {
471	0	LOG(WARNING) << "Fail to set concurrency by tag: " << tag
472	0	<< ", tag concurrency should be larger than old oncurrency. old concurrency: "
473	0	<< tag_ngroup << ", new concurrency: " << num;
474	0	return EPERM;
475	0	}
476	0	}
477
478	0	int bthread_about_to_quit() {
479	0	bthread::TaskGroup* g = bthread::tls_task_group;
480	0	if (g != NULL) {
481	0	bthread::TaskMeta* current_task = g->current_task();
482	0	if(!(current_task->attr.flags & BTHREAD_NEVER_QUIT)) {
483	0	current_task->about_to_quit = true;
484	0	}
485	0	return 0;
486	0	}
487	0	return EPERM;
488	0	}
489
490		int bthread_timer_add(bthread_timer_t* id, timespec abstime,
491	0	void (on_timer)(void), void* arg) {
492	0	bthread::TaskControl* c = bthread::get_or_new_task_control();
493	0	if (c == NULL) {
494	0	return ENOMEM;
495	0	}
496	0	bthread::TimerThread* tt = bthread::get_or_create_global_timer_thread();
497	0	if (tt == NULL) {
498	0	return ENOMEM;
499	0	}
500	0	bthread_timer_t tmp = tt->schedule(on_timer, arg, abstime);
501	0	if (tmp != 0) {
502	0	*id = tmp;
503	0	return 0;
504	0	}
505	0	return ESTOP;
506	0	}
507
508	0	int bthread_timer_del(bthread_timer_t id) {
509	0	bthread::TaskControl* c = bthread::get_task_control();
510	0	if (c != NULL) {
511	0	bthread::TimerThread* tt = bthread::get_global_timer_thread();
512	0	if (tt == NULL) {
513	0	return EINVAL;
514	0	}
515	0	const int state = tt->unschedule(id);
516	0	if (state >= 0) {
517	0	return state;
518	0	}
519	0	}
520	0	return EINVAL;
521	0	}
522
523	0	int bthread_usleep(uint64_t microseconds) {
524	0	bthread::TaskGroup* g = bthread::tls_task_group;
525	0	if (NULL != g && !g->is_current_pthread_task()) {
526	0	return bthread::TaskGroup::usleep(&g, microseconds);
527	0	}
528	0	return ::usleep(microseconds);
529	0	}
530
531	0	int bthread_yield(void) {
532	0	bthread::TaskGroup* g = bthread::tls_task_group;
533	0	if (NULL != g && !g->is_current_pthread_task()) {
534	0	bthread::TaskGroup::yield(&g);
535	0	return 0;
536	0	}
537		// pthread_yield is not available on MAC
538	0	return sched_yield();
539	0	}
540
541	0	int bthread_set_worker_startfn(void (*start_fn)()) {
542	0	if (start_fn == NULL) {
543	0	return EINVAL;
544	0	}
545	0	bthread::g_worker_startfn = start_fn;
546	0	return 0;
547	0	}
548
549	0	int bthread_set_tagged_worker_startfn(void (*start_fn)(bthread_tag_t)) {
550	0	if (start_fn == NULL) {
551	0	return EINVAL;
552	0	}
553	0	bthread::g_tagged_worker_startfn = start_fn;
554	0	return 0;
555	0	}
556
557	0	int bthread_set_create_span_func(void* (*func)()) {
558	0	if (func == NULL) {
559	0	return EINVAL;
560	0	}
561	0	bthread::g_create_span_func = func;
562	0	return 0;
563	0	}
564
565	0	void bthread_stop_world() {
566	0	bthread::TaskControl* c = bthread::get_task_control();
567	0	if (c != NULL) {
568	0	c->stop_and_join();
569	0	}
570	0	}
571
572		int bthread_list_init(bthread_list_t* list,
573		unsigned /size/,
574	0	unsigned /conflict_size/) {
575	0	list->impl = new (std::nothrow) bthread::TidList;
576	0	if (NULL == list->impl) {
577	0	return ENOMEM;
578	0	}
579		// Set unused fields to zero as well.
580	0	list->head = 0;
581	0	list->size = 0;
582	0	list->conflict_head = 0;
583	0	list->conflict_size = 0;
584	0	return 0;
585	0	}
586
587	0	void bthread_list_destroy(bthread_list_t* list) {
588	0	delete static_cast<bthread::TidList*>(list->impl);
589	0	list->impl = NULL;
590	0	}
591
592	0	int bthread_list_add(bthread_list_t* list, bthread_t id) {
593	0	if (list->impl == NULL) {
594	0	return EINVAL;
595	0	}
596	0	return static_cast<bthread::TidList*>(list->impl)->add(id);
597	0	}
598
599	0	int bthread_list_stop(bthread_list_t* list) {
600	0	if (list->impl == NULL) {
601	0	return EINVAL;
602	0	}
603	0	static_cast<bthread::TidList*>(list->impl)->apply(bthread::TidStopper());
604	0	return 0;
605	0	}
606
607	0	int bthread_list_join(bthread_list_t* list) {
608	0	if (list->impl == NULL) {
609	0	return EINVAL;
610	0	}
611	0	static_cast<bthread::TidList*>(list->impl)->apply(bthread::TidJoiner());
612	0	return 0;
613	0	}
614
615	0	bthread_tag_t bthread_self_tag(void) {
616	0	return bthread::tls_task_group != nullptr ? bthread::tls_task_group->tag()
617	0	: BTHREAD_TAG_DEFAULT;
618	0	}
619
620	0	uint64_t bthread_cpu_clock_ns(void) {
621	0	bthread::TaskGroup* g = bthread::tls_task_group;
622	0	if (g != NULL && !g->is_current_main_task()) {
623	0	return g->current_task_cpu_clock_ns();
624	0	}
625	0	return 0;
626	0	}
627
628		} // extern "C"