// 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/types.h>

#include <stddef.h>                         // size_t

#include <gflags/gflags.h>

#include "butil/compat.h"                   // OS_MACOSX

#include "butil/macros.h"                   // ARRAY_SIZE

#include "butil/scoped_lock.h"              // BAIDU_SCOPED_LOCK

#include "butil/fast_rand.h"

#include "butil/unique_ptr.h"

#include "butil/third_party/murmurhash3/murmurhash3.h" // fmix64

#include "butil/reloadable_flags.h"

#include "bthread/errno.h"                  // ESTOP

#include "bthread/butex.h"                  // butex_*

#include "bthread/sys_futex.h"              // futex_wake_private

#include "bthread/processor.h"              // cpu_relax

#include "bthread/task_control.h"

#include "bthread/task_group.h"

#include "bthread/timer_thread.h"

#include "bthread/bthread.h"

#ifdef __x86_64__

#include <x86intrin.h>

#endif // __x86_64__

#ifdef __ARM_NEON

#include <arm_neon.h>

#endif // __ARM_NEON

namespace bthread {

// Global span function pointers for bthread lifecycle tracing.

// These are set by brpc layer via bthread_set_span_funcs().

void* (*g_create_bthread_span)() = NULL;

void (*g_rpcz_parent_span_dtor)(void*) = NULL;

void (*g_end_bthread_span)() = NULL;

static const bthread_attr_t BTHREAD_ATTR_TASKGROUP = {

    BTHREAD_STACKTYPE_UNKNOWN, 0, NULL, BTHREAD_TAG_INVALID, {0} };

DEFINE_bool(show_bthread_creation_in_vars, false, "When this flags is on, The time "

            "from bthread creation to first run will be recorded and shown in /vars");

BUTIL_VALIDATE_GFLAG(show_bthread_creation_in_vars, butil::PassValidate);

DEFINE_bool(show_per_worker_usage_in_vars, false,

            "Show per-worker usage in /vars/bthread_per_worker_usage_<tid>");

BUTIL_VALIDATE_GFLAG(show_per_worker_usage_in_vars, butil::PassValidate);

DEFINE_bool(bthread_enable_cpu_clock_stat, false,

            "Enable CPU clock statistics for bthread");

BUTIL_VALIDATE_GFLAG(bthread_enable_cpu_clock_stat, butil::PassValidate);

BAIDU_VOLATILE_THREAD_LOCAL(TaskGroup*, tls_task_group, NULL);

// Sync with TaskMeta::local_storage when a bthread is created or destroyed.

// During running, the two fields may be inconsistent, use tls_bls as the

// groundtruth.

BAIDU_VOLATILE_THREAD_LOCAL(LocalStorage, tls_bls, BTHREAD_LOCAL_STORAGE_INITIALIZER);

// defined in bthread/key.cpp

extern void return_keytable(bthread_keytable_pool_t*, KeyTable*);

// [Hacky] This is a special TLS set by bthread-rpc privately... to save

// overhead of creation keytable, may be removed later.

BAIDU_VOLATILE_THREAD_LOCAL(void*, tls_unique_user_ptr, NULL);

const TaskStatistics EMPTY_STAT = { 0, 0, 0 };

AtomicInteger128::Value AtomicInteger128::load() const {

#ifdef __x86_64__

    (void)_mutex;

    (void)_seq;

    __m128i value = _mm_load_si128(reinterpret_cast<const __m128i*>(&_value));

    return {value[0], value[1]};

#elif defined(__ARM_NEON)

    (void)_mutex;

    (void)_seq;

    int64x2_t value = vld1q_s64(reinterpret_cast<const int64_t*>(&_value));

    return {value[0], value[1]};

#elif defined(__riscv) && __riscv_xlen == 64

    (void)_mutex;

    // RISC-V: Seqlock-based atomic 128-bit load.

    int64_t v1, v2;

    uint64_t seq0, seq1;

    do {

        __asm__ volatile(

            "ld %0, %1\n\t"

            : "=r"(seq0)

            : "m"(_seq)

            : "memory"

        );

        if (seq0 & 1) continue;

        __asm__ volatile("fence r, rw\n\t" ::: "memory");

        __asm__ volatile(

            "ld %0, %2\n\t"

            "ld %1, %3\n\t"

            : "=r"(v1), "=r"(v2)

            : "m"(_value.v1), "m"(_value.v2)

            : "memory"

        );

        __asm__ volatile("fence r, rw\n\t" ::: "memory");

        __asm__ volatile(

            "ld %0, %1\n\t"

            : "=r"(seq1)

            : "m"(_seq)

            : "memory"

        );

    } while (seq0 != seq1);

    return {v1, v2};

#else

    BAIDU_SCOPED_LOCK(const_cast<FastPthreadMutex&>(_mutex));

    return _value;

#endif

}

void AtomicInteger128::store(Value value) {

#ifdef __x86_64__

    (void)_seq;

    __m128i v = _mm_load_si128(reinterpret_cast<__m128i*>(&value));

    _mm_store_si128(reinterpret_cast<__m128i*>(&_value), v);

#elif defined(__ARM_NEON)

    (void)_seq;

    int64x2_t v = vld1q_s64(reinterpret_cast<int64_t*>(&value));

    vst1q_s64(reinterpret_cast<int64_t*>(&_value), v);

#elif defined(__riscv) && __riscv_xlen == 64

    (void)_mutex;

    // RISC-V: Seqlock-based atomic 128-bit store.

    uint64_t old_seq;

    __asm__ volatile(

        "ld %0, %1\n\t"

        : "=r"(old_seq)

        : "m"(_seq)

        : "memory"

    );

    uint64_t new_seq = old_seq + 1;

    __asm__ volatile(

        "fence w, w\n\t"

        "sd %1, %0\n\t"

        : "=m"(_seq)

        : "r"(new_seq)

        : "memory"

    );

    __asm__ volatile("fence w, w\n\t" ::: "memory");

    __asm__ volatile(

        "sd %2, %0\n\t"

        "sd %3, %1\n\t"

        : "=m"(_value.v1), "=m"(_value.v2)

        : "r"(value.v1), "r"(value.v2)

        : "memory"

    );

    __asm__ volatile("fence w, w\n\t" ::: "memory");

    new_seq++;

    __asm__ volatile(

        "sd %1, %0\n\t"

        : "=m"(_seq)

        : "r"(new_seq)

        : "memory"

    );

#else

    BAIDU_SCOPED_LOCK(const_cast<FastPthreadMutex&>(_mutex));

    _value = value;

#endif

}

int TaskGroup::get_attr(bthread_t tid, bthread_attr_t* out) {

    TaskMeta* const m = address_meta(tid);

    if (m != NULL) {

        const uint32_t given_ver = get_version(tid);

        BAIDU_SCOPED_LOCK(m->version_lock);

        if (given_ver == *m->version_butex) {

            *out = m->attr;

            return 0;

        }

    }

    errno = EINVAL;

    return -1;

}

void TaskGroup::set_stopped(bthread_t tid) {

    TaskMeta* const m = address_meta(tid);

    if (m != NULL) {

        const uint32_t given_ver = get_version(tid);

        BAIDU_SCOPED_LOCK(m->version_lock);

        if (given_ver == *m->version_butex) {

            m->stop = true;

        }

    }

}

bool TaskGroup::is_stopped(bthread_t tid) {

    TaskMeta* const m = address_meta(tid);

    if (m != NULL) {

        const uint32_t given_ver = get_version(tid);

        BAIDU_SCOPED_LOCK(m->version_lock);

        if (given_ver == *m->version_butex) {

            return m->stop;

        }

    }

    // If the tid does not exist or version does not match, it's intuitive

    // to treat the thread as "stopped".

    return true;

}

bool TaskGroup::wait_task(bthread_t* tid) {

    do {

#ifndef BTHREAD_DONT_SAVE_PARKING_STATE

        if (_last_pl_state.stopped()) {

            return false;

        }

        _pl->wait(_last_pl_state);

        if (steal_task(tid)) {

            return true;

        }

#else

        const ParkingLot::State st = _pl->get_state();

        if (st.stopped()) {

            return false;

        }

        if (steal_task(tid)) {

            return true;

        }

        _pl->wait(st);

#endif

    } while (true);

}

static double get_cumulated_cputime_from_this(void* arg) {

    return static_cast<TaskGroup*>(arg)->cumulated_cputime_ns() / 1000000000.0;

}

int64_t TaskGroup::cumulated_cputime_ns() const {

    CPUTimeStat cpu_time_stat = _cpu_time_stat.load();

    // Add the elapsed time of running bthread.

    int64_t cumulated_cputime_ns = cpu_time_stat.cumulated_cputime_ns();

    if (!cpu_time_stat.is_main_task()) {

        cumulated_cputime_ns += butil::cpuwide_time_ns() - cpu_time_stat.last_run_ns();

    }

    return cumulated_cputime_ns;

}

void TaskGroup::run_main_task() {

    bvar::PassiveStatus<double> cumulated_cputime(

        get_cumulated_cputime_from_this, this);

    std::unique_ptr<bvar::PerSecond<bvar::PassiveStatus<double> > > usage_bvar;

    TaskGroup* dummy = this;

    bthread_t tid;

    while (wait_task(&tid)) {

        sched_to(&dummy, tid);

        DCHECK_EQ(this, dummy);

        DCHECK_EQ(_cur_meta->stack, _main_stack);

        if (_cur_meta->tid != _main_tid) {

            task_runner(1/*skip remained*/);

        }

        if (FLAGS_show_per_worker_usage_in_vars && !usage_bvar) {

            char name[32];

#if defined(OS_MACOSX)

            snprintf(name, sizeof(name), "bthread_worker_usage_%" PRIu64,

                     pthread_numeric_id());

#else

            snprintf(name, sizeof(name), "bthread_worker_usage_%ld",

                     (long)syscall(SYS_gettid));

#endif

            usage_bvar.reset(new bvar::PerSecond<bvar::PassiveStatus<double> >

                             (name, &cumulated_cputime, 1));

        }

    }

    // Don't forget to add elapse of last wait_task.

    current_task()->stat.cputime_ns +=

        butil::cpuwide_time_ns() - _cpu_time_stat.load_unsafe().last_run_ns();

}

TaskGroup::TaskGroup(TaskControl* c)

    :  _control(c) {

    CHECK(c);

}

TaskGroup::~TaskGroup() {

    if (_main_tid) {

        TaskMeta* m = address_meta(_main_tid);

        CHECK(_main_stack == m->stack);

#ifdef BUTIL_USE_ASAN

        _main_stack->storage.bottom = NULL;

        _main_stack->storage.stacksize = 0;

#endif // BUTIL_USE_ASAN

        return_stack(m->release_stack());

        return_resource(get_slot(_main_tid));

        _main_tid = 0;

    }

}

#ifdef BUTIL_USE_ASAN

// Returns the **highest** address of the calling pthread's stack and its

// total size, matching brpc's `StackStorage::bottom` convention (see comment

// in bthread/stack.h: "Assume stack grows upwards"). Note that on Linux

// `pthread_attr_getstack(3)` returns the lowest address of the region, so

// we have to translate it; on macOS `pthread_get_stackaddr_np(3)` already

// returns the stack base (highest address), so we use it as-is.

int PthreadAttrGetStack(void*& stack_addr, size_t& stack_size) {

#if defined(OS_MACOSX)

    stack_addr = pthread_get_stackaddr_np(pthread_self());

    stack_size = pthread_get_stacksize_np(pthread_self());

    return 0;

#else

    pthread_attr_t attr;

    int rc = pthread_getattr_np(pthread_self(), &attr);

    if (0 != rc) {

        LOG(ERROR) << "Fail to get pthread attributes: " << berror(rc);

        return rc;

    }

    void* stack_lowest = NULL;

    rc = pthread_attr_getstack(&attr, &stack_lowest, &stack_size);

    if (0 != rc) {

        LOG(ERROR) << "Fail to get pthread stack: " << berror(rc);

    } else {

        // Translate lowest -> highest to match StackStorage::bottom.

        stack_addr = (char*)stack_lowest + stack_size;

    }

    pthread_attr_destroy(&attr);

    return rc;

#endif // OS_MACOSX

}

#endif // BUTIL_USE_ASAN

int TaskGroup::init(size_t runqueue_capacity) {

    if (_rq.init(runqueue_capacity) != 0) {

        LOG(FATAL) << "Fail to init _rq";

        return -1;

    }

    if (_remote_rq.init(runqueue_capacity / 2) != 0) {

        LOG(FATAL) << "Fail to init _remote_rq";

        return -1;

    }

#ifdef BUTIL_USE_ASAN

    void* stack_addr = NULL;

    size_t stack_size = 0;

    if (0 != PthreadAttrGetStack(stack_addr, stack_size)) {

        return -1;

    }

#endif // BUTIL_USE_ASAN

    ContextualStack* stk = get_stack(STACK_TYPE_MAIN, NULL);

    if (NULL == stk) {

        LOG(FATAL) << "Fail to get main stack container";

        return -1;

    }

    butil::ResourceId<TaskMeta> slot;

    TaskMeta* m = butil::get_resource<TaskMeta>(&slot);

    if (NULL == m) {

        LOG(FATAL) << "Fail to get TaskMeta";

        return -1;

    }

    m->sleep_failed = false;

    m->stop = false;

    m->interrupted = false;

    m->about_to_quit = false;

    m->fn = NULL;

    m->arg = NULL;

    m->local_storage = LOCAL_STORAGE_INIT;

    m->cpuwide_start_ns = butil::cpuwide_time_ns();

    m->stat = EMPTY_STAT;

    m->attr = BTHREAD_ATTR_TASKGROUP;

    m->tid = make_tid(*m->version_butex, slot);

    m->set_stack(stk);

#ifdef BUTIL_USE_ASAN

    stk->storage.bottom = stack_addr;

    stk->storage.stacksize = stack_size;

    // No guard size required for ASan.

#endif // BUTIL_USE_ASAN

    _cur_meta = m;

    _main_tid = m->tid;

    _main_stack = stk;

    CPUTimeStat cpu_time_stat;

    cpu_time_stat.set_last_run_ns(m->cpuwide_start_ns, true);

    _cpu_time_stat.store(cpu_time_stat);

    _last_cpu_clock_ns = 0;

    return 0;

}

#ifdef BUTIL_USE_ASAN

void TaskGroup::asan_task_runner(intptr_t) {

    // This is a new thread, and it doesn't have the fake stack yet. ASan will

    // create it lazily, for now just pass NULL.

    internal::FinishSwitchFiber(NULL);

    task_runner(0);

}

#endif // BUTIL_USE_ASAN

void TaskGroup::task_runner(intptr_t skip_remained) {

    // NOTE: tls_task_group is volatile since tasks are moved around

    //       different groups.

    TaskGroup* g = BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_task_group);

#ifdef BRPC_BTHREAD_TRACER

    TaskTracer::set_running_status(g->tid(), g->_cur_meta);

#endif // BRPC_BTHREAD_TRACER

    if (!skip_remained) {

        while (g->_last_context_remained) {

            RemainedFn fn = g->_last_context_remained;

            g->_last_context_remained = NULL;

            fn(g->_last_context_remained_arg);

            g = BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_task_group);

        }

#ifndef NDEBUG

        --g->_sched_recursive_guard;

#endif

    }

    do {

        // A task can be stopped before it gets running, in which case

        // we may skip user function, but that may confuse user:

        // Most tasks have variables to remember running result of the task,

        // which is often initialized to values indicating success. If an

        // user function is never called, the variables will be unchanged

        // however they'd better reflect failures because the task is stopped

        // abnormally.

        // Meta and identifier of the task is persistent in this run.

        TaskMeta* const m = g->_cur_meta;

        if (FLAGS_show_bthread_creation_in_vars) {

            // NOTE: the thread triggering exposure of pending time may spend

            // considerable time because a single bvar::LatencyRecorder

            // contains many bvar.

            g->_control->exposed_pending_time() <<

                (butil::cpuwide_time_ns() - m->cpuwide_start_ns) / 1000L;

        }

        // Not catch exceptions except ExitException which is for implementing

        // bthread_exit(). User code is intended to crash when an exception is

        // not caught explicitly. This is consistent with other threading

        // libraries.

        void* thread_return;

        try {

            thread_return = m->fn(m->arg);

        } catch (ExitException& e) {

            thread_return = e.value();

        }

        if (m->attr.flags & BTHREAD_INHERIT_SPAN) {

            if (g_end_bthread_span) {

                g_end_bthread_span();

            }

        }

        // TODO: Save thread_return

        (void)thread_return;

        // Logging must be done before returning the keytable, since the logging lib

        // use bthread local storage internally, or will cause memory leak.

        // FIXME: the time from quiting fn to here is not counted into cputime

        if (m->attr.flags & BTHREAD_LOG_START_AND_FINISH) {

            LOG(INFO) << "Finished bthread " << m->tid << ", cputime="

                      << m->stat.cputime_ns / 1000000.0 << "ms";

        }

        // Clean up span if it exists. This must be done before keytable cleanup

        // because span cleanup may use bthread local storage (e.g. logging,

        // which allocates bthread-local stream arrays via bthread_setspecific).

        // If span cleanup ran after keytable cleanup, such allocations would

        // re-populate the keytable and never be reclaimed, causing memory leak.

        LocalStorage* tls_bls_ptr = bthread::tls_bls_ptr();

        if (tls_bls_ptr->rpcz_parent_span && g_rpcz_parent_span_dtor) {

            g_rpcz_parent_span_dtor(tls_bls_ptr->rpcz_parent_span);

            tls_bls_ptr = bthread::tls_bls_ptr();

            tls_bls_ptr->rpcz_parent_span = NULL;

            m->local_storage.rpcz_parent_span = NULL;

        }

        // Clean tls variables, must be done before changing version_butex

        // otherwise another thread just joined this thread may not see side

        // effects of destructing tls variables.

        KeyTable* kt = tls_bls_ptr->keytable;

        if (kt != NULL) {

            return_keytable(m->attr.keytable_pool, kt);

            // After deletion: tls may be set during deletion.

            tls_bls_ptr = bthread::tls_bls_ptr();

            tls_bls_ptr->keytable = NULL;

            m->local_storage.keytable = NULL; // optional

        }

        // During running the function in TaskMeta and deleting the KeyTable in

        // return_KeyTable, the group is probably changed.

        g =  BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_task_group);

        // Increase the version and wake up all joiners, if resulting version

        // is 0, change it to 1 to make bthread_t never be 0. Any access

        // or join to the bthread after changing version will be rejected.

        // The spinlock is for visibility of TaskGroup::get_attr.

#ifdef BRPC_BTHREAD_TRACER

        bool tracing = false;

#endif // BRPC_BTHREAD_TRACER

        {

            BAIDU_SCOPED_LOCK(m->version_lock);

#ifdef BRPC_BTHREAD_TRACER

            tracing = TaskTracer::set_end_status_unsafe(m);

#endif // BRPC_BTHREAD_TRACER

            if (0 == ++*m->version_butex) {

                ++*m->version_butex;

            }

        }

        butex_wake_except(m->version_butex, 0);

#ifdef BRPC_BTHREAD_TRACER

        if (tracing) {

            // Wait for tracing completion.

            g->_control->_task_tracer.WaitForTracing(m);

        }

        g->_control->_task_tracer.set_status(TASK_STATUS_UNKNOWN, m);

#endif // BRPC_BTHREAD_TRACER

        g->_control->_nbthreads << -1;

        g->_control->tag_nbthreads(g->tag()) << -1;

        g->set_remained(_release_last_context, m);

        ending_sched(&g);

    } while (g->_cur_meta->tid != g->_main_tid);

    // Was called from a pthread and we don't have BTHREAD_STACKTYPE_PTHREAD

    // tasks to run, quit for more tasks.

}

void TaskGroup::_release_last_context(void* arg) {

    TaskMeta* m = static_cast<TaskMeta*>(arg);

    if (m->stack_type() != STACK_TYPE_PTHREAD) {

        return_stack(m->release_stack()/*may be NULL*/);

    } else {

        // it's _main_stack, don't return.

        m->set_stack(NULL);

    }

    return_resource(get_slot(m->tid));

}

int TaskGroup::start_foreground(TaskGroup** pg,

                                bthread_t* __restrict th,

                                const bthread_attr_t* __restrict attr,

                                void * (*fn)(void*),

                                void* __restrict arg) {

    if (__builtin_expect(!fn, 0)) {

        return EINVAL;

    }

    const int64_t start_ns = butil::cpuwide_time_ns();

    const bthread_attr_t using_attr = (attr ? *attr : BTHREAD_ATTR_NORMAL);

    butil::ResourceId<TaskMeta> slot;

    TaskMeta* m = butil::get_resource(&slot);

    if (BAIDU_UNLIKELY(NULL == m)) {

        return ENOMEM;

    }

    CHECK(m->current_waiter.load(butil::memory_order_relaxed) == NULL);

    m->sleep_failed = false;

    m->stop = false;

    m->interrupted = false;

    m->about_to_quit = false;

    m->fn = fn;

    m->arg = arg;

    CHECK(m->stack == NULL);

    m->attr = using_attr;

    m->local_storage = LOCAL_STORAGE_INIT;

    if (using_attr.flags & BTHREAD_INHERIT_SPAN) {

        if (g_create_bthread_span) {

            m->local_storage.rpcz_parent_span = g_create_bthread_span();

        } else {

            m->local_storage.rpcz_parent_span = BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_bls).rpcz_parent_span;

        }

    }

    m->cpuwide_start_ns = start_ns;

    m->stat = EMPTY_STAT;

    m->tid = make_tid(*m->version_butex, slot);

    TaskGroup* g = *pg;

    m->priority_index = g->_cur_meta->priority_index;

    m->attr.tag = g->tag();

    *th = m->tid;

    if (using_attr.flags & BTHREAD_LOG_START_AND_FINISH) {

        LOG(INFO) << "Started bthread " << m->tid;

    }

    g->_control->_nbthreads << 1;

    g->_control->tag_nbthreads(g->tag()) << 1;

#ifdef BRPC_BTHREAD_TRACER

    g->_control->_task_tracer.set_status(TASK_STATUS_CREATED, m);

#endif // BRPC_BTHREAD_TRACER

    if (g->is_current_pthread_task()) {

        // never create foreground task in pthread.

        g->ready_to_run(m, using_attr.flags & BTHREAD_NOSIGNAL);

    } else {

        // NOSIGNAL affects current task, not the new task.

        RemainedFn fn = NULL;

        auto& cur_attr = g->_cur_meta->attr;

        if (g->_control->_enable_priority_queue && cur_attr.flags & BTHREAD_GLOBAL_PRIORITY) {

            fn = priority_to_run;

        } else if (g->current_task()->about_to_quit) {

            fn = ready_to_run_in_worker_ignoresignal;

        } else {

            fn = ready_to_run_in_worker;

        }

        ReadyToRunArgs args = {

            g->tag(), g->_cur_meta, (bool)(using_attr.flags & BTHREAD_NOSIGNAL)

        };

        g->set_remained(fn, &args);

        sched_to(pg, m->tid);

    }

    return 0;

}

template <bool REMOTE>

int TaskGroup::start_background(bthread_t* __restrict th,

                                const bthread_attr_t* __restrict attr,

                                void * (*fn)(void*),

                                void* __restrict arg) {

    if (__builtin_expect(!fn, 0)) {

        return EINVAL;

    }

    const int64_t start_ns = butil::cpuwide_time_ns();

    const bthread_attr_t using_attr = (attr ? *attr : BTHREAD_ATTR_NORMAL);

    butil::ResourceId<TaskMeta> slot;

    TaskMeta* m = butil::get_resource(&slot);

    if (BAIDU_UNLIKELY(NULL == m)) {

        return ENOMEM;

    }

    CHECK(m->current_waiter.load(butil::memory_order_relaxed) == NULL);

    m->sleep_failed = false;

    m->stop = false;

    m->interrupted = false;

    m->about_to_quit = false;

    m->fn = fn;

    m->arg = arg;

    CHECK(m->stack == NULL);

    m->attr = using_attr;

    m->local_storage = LOCAL_STORAGE_INIT;

    if (using_attr.flags & BTHREAD_INHERIT_SPAN) {

        if (g_create_bthread_span) {

            m->local_storage.rpcz_parent_span = g_create_bthread_span();

        } else {

            m->local_storage.rpcz_parent_span = BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_bls).rpcz_parent_span;

        }

    }

    m->cpuwide_start_ns = start_ns;

    m->stat = EMPTY_STAT;

    m->tid = make_tid(*m->version_butex, slot);

    m->priority_index = _cur_meta->priority_index;

    *th = m->tid;

    if (using_attr.flags & BTHREAD_LOG_START_AND_FINISH) {

        LOG(INFO) << "Started bthread " << m->tid;

    }

    m->attr.tag = tag();

    _control->_nbthreads << 1;

    _control->tag_nbthreads(tag()) << 1;

#ifdef BRPC_BTHREAD_TRACER

    _control->_task_tracer.set_status(TASK_STATUS_CREATED, m);

#endif // BRPC_BTHREAD_TRACER

    if (REMOTE) {

        ready_to_run_remote(m, (using_attr.flags & BTHREAD_NOSIGNAL));

    } else {

        ready_to_run(m, (using_attr.flags & BTHREAD_NOSIGNAL));

    }

    return 0;

}

Unexecuted instantiation: int bthread::TaskGroup::start_background<true>(unsigned long*, bthread_attr_t const*, void* (*)(void*), void*)

Unexecuted instantiation: int bthread::TaskGroup::start_background<false>(unsigned long*, bthread_attr_t const*, void* (*)(void*), void*)

// Explicit instantiations.

template int

TaskGroup::start_background<true>(bthread_t* __restrict th,

                                  const bthread_attr_t* __restrict attr,

                                  void * (*fn)(void*),

                                  void* __restrict arg);

template int

TaskGroup::start_background<false>(bthread_t* __restrict th,

                                   const bthread_attr_t* __restrict attr,

                                   void * (*fn)(void*),

                                   void* __restrict arg);

int TaskGroup::join(bthread_t tid, void** return_value) {

    if (__builtin_expect(!tid, 0)) {  // tid of bthread is never 0.

        return EINVAL;

    }

    TaskMeta* m = address_meta(tid);

    if (BAIDU_UNLIKELY(NULL == m)) {

        // The bthread is not created yet, this join is definitely wrong.

        return EINVAL;

    }

    TaskGroup* g = BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_task_group);

    if (g != NULL && g->current_tid() == tid) {

        // joining self causes indefinite waiting.

        return EINVAL;

    }

    const uint32_t expected_version = get_version(tid);

    while (*m->version_butex == expected_version) {

        if (butex_wait(m->version_butex, expected_version, NULL) < 0 &&

            errno != EWOULDBLOCK && errno != EINTR) {

            return errno;

        }

    }

    // Ensure all memory writes made by the joined bthread are visible to

    // the joining thread after join returns. This matches the semantic

    // guarantee provided by pthread_join() across supported architectures.

    butil::atomic_thread_fence(butil::memory_order_acquire);

    if (return_value) {

        *return_value = NULL;

    }

    return 0;

}

bool TaskGroup::exists(bthread_t tid) {

    if (tid != 0) {  // tid of bthread is never 0.

        TaskMeta* m = address_meta(tid);

        if (m != NULL) {

            return (*m->version_butex == get_version(tid));

        }

    }

    return false;

}

TaskStatistics TaskGroup::main_stat() const {

    TaskMeta* m = address_meta(_main_tid);

    return m ? m->stat : EMPTY_STAT;

}

void TaskGroup::ending_sched(TaskGroup** pg) {

    TaskGroup* g = *pg;

    bthread_t next_tid = 0;

    // Find next task to run, if none, switch to idle thread of the group.

#ifndef BTHREAD_FAIR_WSQ

    // When BTHREAD_FAIR_WSQ is defined, profiling shows that cpu cost of

    // WSQ::steal() in example/multi_threaded_echo_c++ changes from 1.9%

    // to 2.9%

    const bool popped = g->_rq.pop(&next_tid);

#else

    const bool popped = g->_rq.steal(&next_tid);

#endif

    if (!popped && !g->steal_task(&next_tid)) {

        // Jump to main task if there's no task to run.

        next_tid = g->_main_tid;

    }

    TaskMeta* const cur_meta = g->_cur_meta;

    TaskMeta* next_meta = address_meta(next_tid);

    if (next_meta->stack == NULL) {

        if (next_meta->stack_type() == cur_meta->stack_type()) {

            // Reuse the stack of the current ending task.

            //

            // also works with pthread_task scheduling to pthread_task, the

            // transfered stack is just _main_stack.

            next_meta->set_stack(cur_meta->release_stack());

        } else {

#ifdef BUTIL_USE_ASAN

            ContextualStack* stk = get_stack(

                next_meta->stack_type(), asan_task_runner);

#else

            ContextualStack* stk = get_stack(next_meta->stack_type(), task_runner);

#endif // BUTIL_USE_ASAN

            if (stk) {

                next_meta->set_stack(stk);

            } else {

                // stack_type is BTHREAD_STACKTYPE_PTHREAD or out of memory,

                // In latter case, attr is forced to be BTHREAD_STACKTYPE_PTHREAD.

                // This basically means that if we can't allocate stack, run

                // the task in pthread directly.

                next_meta->attr.stack_type = BTHREAD_STACKTYPE_PTHREAD;

                next_meta->set_stack(g->_main_stack);

            }

        }

    }

    sched_to(pg, next_meta);

}

void TaskGroup::sched(TaskGroup** pg) {

    TaskGroup* g = *pg;

    bthread_t next_tid = 0;

    // Find next task to run, if none, switch to idle thread of the group.

#ifndef BTHREAD_FAIR_WSQ

    const bool popped = g->_rq.pop(&next_tid);

#else

    const bool popped = g->_rq.steal(&next_tid);

#endif

    if (!popped && !g->steal_task(&next_tid)) {

        // Jump to main task if there's no task to run.

        next_tid = g->_main_tid;

    }

    sched_to(pg, next_tid);

}

extern void CheckBthreadScheSafety();

void TaskGroup::sched_to(TaskGroup** pg, TaskMeta* next_meta) {

    TaskGroup* g = *pg;

#ifndef NDEBUG

    if ((++g->_sched_recursive_guard) > 1) {

        LOG(FATAL) << "Recursively(" << g->_sched_recursive_guard - 1

                   << ") call sched_to(" << g << ")";

    }

#endif

    // Save errno so that errno is bthread-specific.

    int saved_errno = errno;

    void* saved_unique_user_ptr = BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_unique_user_ptr);

    TaskMeta* const cur_meta = g->_cur_meta;

    int64_t now = butil::cpuwide_time_ns();

    CPUTimeStat cpu_time_stat = g->_cpu_time_stat.load_unsafe();

    int64_t elp_ns = now - cpu_time_stat.last_run_ns();

    cur_meta->stat.cputime_ns += elp_ns;

    // Update cpu_time_stat.

    cpu_time_stat.set_last_run_ns(now, is_main_task(g, next_meta->tid));

    cpu_time_stat.add_cumulated_cputime_ns(elp_ns, is_main_task(g, cur_meta->tid));

    g->_cpu_time_stat.store(cpu_time_stat);

    if (FLAGS_bthread_enable_cpu_clock_stat) {

        const int64_t cpu_thread_time = butil::cputhread_time_ns();

        if (g->_last_cpu_clock_ns != 0) {

            cur_meta->stat.cpu_usage_ns += cpu_thread_time - g->_last_cpu_clock_ns;

        }

        g->_last_cpu_clock_ns = cpu_thread_time;

    } else {

        g->_last_cpu_clock_ns = 0;

    }

    ++cur_meta->stat.nswitch;

    ++ g->_nswitch;

    // Switch to the task

    if (__builtin_expect(next_meta != cur_meta, 1)) {

        g->_cur_meta = next_meta;

        // Switch tls_bls

        cur_meta->local_storage = BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_bls);

        BAIDU_SET_VOLATILE_THREAD_LOCAL(tls_bls, next_meta->local_storage);

        // Logging must be done after switching the local storage, since the logging lib

        // use bthread local storage internally, or will cause memory leak.

        if ((cur_meta->attr.flags & BTHREAD_LOG_CONTEXT_SWITCH) ||

            (next_meta->attr.flags & BTHREAD_LOG_CONTEXT_SWITCH)) {

            LOG(INFO) << "Switch bthread: " << cur_meta->tid << " -> "

                      << next_meta->tid;

        }

        if (cur_meta->stack != NULL) {

            if (next_meta->stack != cur_meta->stack) {

                CheckBthreadScheSafety();

#ifdef BRPC_BTHREAD_TRACER

                g->_control->_task_tracer.set_status(TASK_STATUS_JUMPING, cur_meta);

                g->_control->_task_tracer.set_status(TASK_STATUS_JUMPING, next_meta);

#endif // BRPC_BTHREAD_TRACER

                {

                    BTHREAD_SCOPED_ASAN_FIBER_SWITCHER(next_meta->stack->storage);

                    jump_stack(cur_meta->stack, next_meta->stack);

                }

                // probably went to another group, need to assign g again.

                g = BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_task_group);

#ifdef BRPC_BTHREAD_TRACER

                TaskTracer::set_running_status(g->tid(), g->_cur_meta);

#endif // BRPC_BTHREAD_TRACER

            }

#ifndef NDEBUG

            else {

                // else pthread_task is switching to another pthread_task, sc

                // can only equal when they're both _main_stack

                CHECK(cur_meta->stack == g->_main_stack);

            }

#endif

        } /* else because of ending_sched(including pthread_task->pthread_task). */

#ifdef BRPC_BTHREAD_TRACER

        else {

            // _cur_meta: TASK_STATUS_FIRST_READY -> TASK_STATUS_RUNNING.

            TaskTracer::set_running_status(g->tid(), g->_cur_meta);

        }

#endif // BRPC_BTHREAD_TRACER

    } else {

        LOG(FATAL) << "bthread=" << g->current_tid() << " sched_to itself!";

    }

    while (g->_last_context_remained) {

        RemainedFn fn = g->_last_context_remained;

        g->_last_context_remained = NULL;

        fn(g->_last_context_remained_arg);

        g = BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_task_group);

    }

    // Restore errno

    errno = saved_errno;

    // tls_unique_user_ptr probably changed.

    BAIDU_SET_VOLATILE_THREAD_LOCAL(tls_unique_user_ptr, saved_unique_user_ptr);

#ifndef NDEBUG

    --g->_sched_recursive_guard;

#endif

    *pg = g;

}

void TaskGroup::destroy_self() {

    if (_control) {

        _control->_destroy_group(this);

        _control = NULL;

    } else {

        CHECK(false);

    }

}

void TaskGroup::ready_to_run(TaskMeta* meta, bool nosignal) {

#ifdef BRPC_BTHREAD_TRACER

    _control->_task_tracer.set_status(TASK_STATUS_READY, meta);

#endif // BRPC_BTHREAD_TRACER

    push_rq(meta->tid);

    if (nosignal) {

        ++_num_nosignal;

    } else {

        const int additional_signal = _num_nosignal;

        _num_nosignal = 0;

        _nsignaled += 1 + additional_signal;

        _control->signal_task(1 + additional_signal, _tag);

    }

}