task_group_inl.h 3.91 KB
Newer Older
gejun's avatar
gejun committed
1
// bthread - A M:N threading library to make applications more concurrent.
gejun's avatar
gejun committed
2
// Copyright (c) 2012 Baidu, Inc.
gejun's avatar
gejun committed
3 4 5 6 7 8 9 10 11 12 13 14
// 
// Licensed 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.
gejun's avatar
gejun committed
15 16 17 18

// Author: Ge,Jun (gejun@baidu.com)
// Date: Tue Jul 10 17:40:58 CST 2012

gejun's avatar
gejun committed
19 20
#ifndef BTHREAD_TASK_GROUP_INL_H
#define BTHREAD_TASK_GROUP_INL_H
gejun's avatar
gejun committed
21 22 23 24

namespace bthread {

// Utilities to manipulate bthread_t
25
inline bthread_t make_tid(uint32_t version, butil::ResourceId<TaskMeta> slot) {
gejun's avatar
gejun committed
26 27 28
    return (((bthread_t)version) << 32) | (bthread_t)slot.value;
}

29 30
inline butil::ResourceId<TaskMeta> get_slot(bthread_t tid) {
    butil::ResourceId<TaskMeta> id = { (tid & 0xFFFFFFFFul) };
gejun's avatar
gejun committed
31 32 33 34 35 36
    return id;
}
inline uint32_t get_version(bthread_t tid) {
    return (uint32_t)((tid >> 32) & 0xFFFFFFFFul);
}

37
inline TaskMeta* TaskGroup::address_meta(bthread_t tid) {
gejun's avatar
gejun committed
38 39 40 41 42 43 44 45
    // TaskMeta * m = address_resource<TaskMeta>(get_slot(tid));
    // if (m != NULL && m->version == get_version(tid)) {
    //     return m;
    // }
    // return NULL;
    return address_resource(get_slot(tid));
}

46
inline void TaskGroup::exchange(TaskGroup** pg, bthread_t next_tid) {
gejun's avatar
gejun committed
47 48 49 50
    TaskGroup* g = *pg;
    if (g->is_current_pthread_task()) {
        return g->ready_to_run(next_tid);
    }
51
    ReadyToRunArgs args = { g->current_tid(), false };
52 53 54
    g->set_remained((g->current_task()->about_to_quit
                     ? ready_to_run_in_worker_ignoresignal
                     : ready_to_run_in_worker),
55
                    &args);
gejun's avatar
gejun committed
56 57 58
    TaskGroup::sched_to(pg, next_tid);
}

59
inline void TaskGroup::sched_to(TaskGroup** pg, bthread_t next_tid) {
gejun's avatar
gejun committed
60
    TaskMeta* next_meta = address_meta(next_tid);
gejun's avatar
gejun committed
61 62 63 64
    if (next_meta->stack == NULL) {
        ContextualStack* stk = get_stack(next_meta->stack_type(), task_runner);
        if (stk) {
            next_meta->set_stack(stk);
gejun's avatar
gejun committed
65 66 67 68 69 70
        } 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;
gejun's avatar
gejun committed
71
            next_meta->set_stack((*pg)->_main_stack);
gejun's avatar
gejun committed
72 73 74 75 76 77
        }
    }
    // Update now_ns only when wait_task did yield.
    sched_to(pg, next_meta);
}

78 79 80 81 82 83 84 85
inline void TaskGroup::push_rq(bthread_t tid) {
    while (!_rq.push(tid)) {
        // Created too many bthreads: a promising approach is to insert the
        // task into another TaskGroup, but we don't use it because:
        // * There're already many bthreads to run, inserting the bthread
        //   into other TaskGroup does not help.
        // * Insertions into other TaskGroups perform worse when all workers
        //   are busy at creating bthreads (proved by test_input_messenger in
gejun's avatar
gejun committed
86
        //   brpc)
87 88
        flush_nosignal_tasks();
        LOG_EVERY_SECOND(ERROR) << "_rq is full, capacity=" << _rq.capacity();
89 90 91 92
        // TODO(gejun): May cause deadlock when all workers are spinning here.
        // A better solution is to pop and run existing bthreads, however which
        // make set_remained()-callbacks do context switches and need extensive
        // reviews on related code.
93 94
        ::usleep(1000);
    }
gejun's avatar
gejun committed
95 96
}

97 98 99 100 101
inline void TaskGroup::flush_nosignal_tasks_remote() {
    if (_remote_num_nosignal) {
        _remote_rq._mutex.lock();
        flush_nosignal_tasks_remote_locked(_remote_rq._mutex);
    }
gejun's avatar
gejun committed
102 103 104 105
}

}  // namespace bthread

gejun's avatar
gejun committed
106
#endif  // BTHREAD_TASK_GROUP_INL_H