Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

epoll bthread deal first #2819

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/brpc/event_dispatcher.cpp
Original file line number Diff line number Diff line change
@@ -55,6 +55,7 @@ void InitializeGlobalDispatchers() {
FLAGS_usercode_in_pthread ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL;
attr.tag = (BTHREAD_TAG_DEFAULT + i) % FLAGS_task_group_ntags;
CHECK_EQ(0, g_edisp[i * FLAGS_event_dispatcher_num + j].Start(&attr));
bthread_epoll_tid_set(i, g_edisp[i].Tid());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这里应该是 g_edisp[i * FLAGS_event_dispatcher_num + j] 吧?

Copy link
Contributor Author

@zhengJade zhengJade Dec 9, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

是的,我本地二次修改的时候没提交上来

}
}
// This atexit is will be run before g_task_control.stop() because above
2 changes: 2 additions & 0 deletions src/brpc/event_dispatcher.h
Original file line number Diff line number Diff line change
@@ -133,6 +133,8 @@ template <typename T> friend class IOEvent;
// Returns 0 on success, -1 otherwise and errno is set
int UnregisterEvent(IOEventDataId event_data_id, int fd, bool pollin);

bthread_t Tid() const { return _tid; }
zhengJade marked this conversation as resolved.
Show resolved Hide resolved

private:
DISALLOW_COPY_AND_ASSIGN(EventDispatcher);

6 changes: 6 additions & 0 deletions src/bthread/bthread.cpp
Original file line number Diff line number Diff line change
@@ -637,4 +637,10 @@ uint64_t bthread_cpu_clock_ns(void) {
return 0;
}

void bthread_epoll_tid_set(bthread_tag_t tag, bthread_t tid) {
CHECK(tag >= BTHREAD_TAG_DEFAULT && tag < FLAGS_task_group_ntags);
auto c = bthread::get_task_control();
return c->set_group_epoll_tid(tag, tid);
}

} // extern "C"
3 changes: 3 additions & 0 deletions src/bthread/bthread.h
Original file line number Diff line number Diff line change
@@ -401,6 +401,9 @@ extern void* bthread_getspecific(bthread_key_t key);
// Return current bthread tag
extern bthread_tag_t bthread_self_tag(void);

// set task_groups epoll tid by tag
extern void bthread_epoll_tid_set(bthread_tag_t tag, bthread_t tid);

// The first call to bthread_once() by any thread in a process, with a given
// once_control, will call the init_routine() with no arguments. Subsequent
// calls of bthread_once() with the same once_control will not call the
27 changes: 27 additions & 0 deletions src/bthread/parking_lot.cpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
// 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: Thu Nov 14 13:40:57 CST 2024
#include "parking_lot.h"

namespace bthread {

butil::atomic<int> ParkingLot::_waiting_count{0};

} // namespace bthread
4 changes: 4 additions & 0 deletions src/bthread/parking_lot.h
Original file line number Diff line number Diff line change
@@ -30,6 +30,7 @@ namespace bthread {
// Park idle workers.
class BAIDU_CACHELINE_ALIGNMENT ParkingLot {
public:
static butil::atomic<int> _waiting_count;
zhengJade marked this conversation as resolved.
Show resolved Hide resolved
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这里 _waiting_count 作为全局静态变量感觉不能有效减少 signal 的次数,因为即使在 _waiting_count != 0 的场景,仍然存在parking_lot list中有的pl存在waiting thread,有的pl不存在。在 signal_task 中是先随机选择一个pl来signal,选择的 pl 不一定存在 waiting thread。

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这里减少的是都繁忙情况下不要 signal,而不是说能够准确的 signal 阻塞的 pl,在比较繁忙的情况是实际上是不存在部分 wait,部分 wake 的,而且 sigal 的时候有返回值的,如果返回失败会接着 signal 另一个 pl

class State {
public:
State(): val(0) {}
@@ -57,14 +58,17 @@ class BAIDU_CACHELINE_ALIGNMENT ParkingLot {
// Wait for tasks.
// If the `expected_state' does not match, wait() may finish directly.
void wait(const State& expected_state) {
_waiting_count++;
futex_wait_private(&_pending_signal, expected_state.val, NULL);
_waiting_count--;
}

// Wakeup suspended wait() and make them unwaitable ever.
void stop() {
_pending_signal.fetch_or(1);
futex_wake_private(&_pending_signal, 10000);
}

private:
// higher 31 bits for signalling, LSB for stopping.
butil::atomic<int> _pending_signal;
26 changes: 24 additions & 2 deletions src/bthread/task_control.cpp
Original file line number Diff line number Diff line change
@@ -183,6 +183,7 @@ TaskControl::TaskControl()
, _signal_per_second(&_cumulated_signal_count)
, _status(print_rq_sizes_in_the_tc, this)
, _nbthreads("bthread_count")
, _priority_qs(FLAGS_task_group_ntags)
, _pl(FLAGS_task_group_ntags)
{}

@@ -207,6 +208,10 @@ int TaskControl::init(int concurrency) {
_tagged_worker_usage_second.push_back(new bvar::PerSecond<bvar::PassiveStatus<double>>(
"bthread_worker_usage", tag_str, _tagged_cumulated_worker_time[i], 1));
_tagged_nbthreads.push_back(new bvar::Adder<int64_t>("bthread_count", tag_str));
if (_priority_qs[i].init(BTHREAD_MAX_CONCURRENCY) != 0) {
LOG(FATAL) << "Fail to init _priority_q";
return -1;
}
}

// Make sure TimerThread is ready.
@@ -430,6 +435,11 @@ int TaskControl::_destroy_group(TaskGroup* g) {

bool TaskControl::steal_task(bthread_t* tid, size_t* seed, size_t offset) {
auto tag = tls_task_group->tag();

if (_priority_qs[tag].steal(tid)) {
return true;
}

// 1: Acquiring fence is paired with releasing fence in _add_group to
// avoid accessing uninitialized slot of _groups.
const size_t ngroup = tag_ngroup(tag).load(butil::memory_order_acquire/*1*/);
@@ -472,13 +482,18 @@ void TaskControl::signal_task(int num_task, bthread_tag_t tag) {
}
auto& pl = tag_pl(tag);
int start_index = butil::fmix64(pthread_numeric_id()) % PARKING_LOT_NUM;
num_task -= pl[start_index].signal(1);
// WARNING: This allow some bad case happen when wait_count is not accurente.
if (ParkingLot::_waiting_count.load(butil::memory_order_relaxed) > 0) {
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);
if (ParkingLot::_waiting_count.load(butil::memory_order_relaxed) > 0) {
num_task -= pl[start_index].signal(1);
}
}
}
if (num_task > 0 &&
@@ -575,4 +590,11 @@ bvar::LatencyRecorder* TaskControl::create_exposed_pending_time() {
return pt;
}

void TaskControl::set_group_epoll_tid(bthread_tag_t tag, bthread_t tid) {
auto groups = tag_group(tag);
const size_t ngroup = tag_ngroup(tag).load(butil::memory_order_acquire);
for (size_t i = 0; i < ngroup; i++) {
groups[i]->add_epoll_tid(tid);
}
}
} // namespace bthread
9 changes: 8 additions & 1 deletion src/bthread/task_control.h
Original file line number Diff line number Diff line change
@@ -30,6 +30,7 @@
#include <vector>
#include <array>
#include <memory>
#include <unordered_map>
#include "butil/atomicops.h" // butil::atomic
#include "bvar/bvar.h" // bvar::PassiveStatus
#include "bthread/task_tracer.h"
@@ -96,6 +97,12 @@ friend bthread_t init_for_pthread_stack_trace();
void stack_trace(std::ostream& os, bthread_t tid);
std::string stack_trace(bthread_t tid);
#endif // BRPC_BTHREAD_TRACER
// Only deal once when init epoll bthread.
void set_group_epoll_tid(bthread_tag_t tag, bthread_t tid);

void push_priority_q(bthread_tag_t tag, bthread_t tid) {
_priority_qs[tag].push(tid);
}

private:
typedef std::array<TaskGroup*, BTHREAD_MAX_CONCURRENCY> TaggedGroups;
@@ -153,13 +160,13 @@ friend bthread_t init_for_pthread_stack_trace();
std::vector<bvar::PassiveStatus<double>*> _tagged_cumulated_worker_time;
std::vector<bvar::PerSecond<bvar::PassiveStatus<double>>*> _tagged_worker_usage_second;
std::vector<bvar::Adder<int64_t>*> _tagged_nbthreads;
std::vector<WorkStealingQueue<bthread_t>> _priority_qs;

std::vector<TaggedParkingLot> _pl;

#ifdef BRPC_BTHREAD_TRACER
TaskTracer _task_tracer;
#endif // BRPC_BTHREAD_TRACER

};

inline bvar::LatencyRecorder& TaskControl::exposed_pending_time() {
18 changes: 14 additions & 4 deletions src/bthread/task_group.cpp
Original file line number Diff line number Diff line change
@@ -435,12 +435,18 @@ int TaskGroup::start_foreground(TaskGroup** pg,
} else {
// NOSIGNAL affects current task, not the new task.
RemainedFn fn = NULL;
if (g->current_task()->about_to_quit) {
if (g->cur_epoll_tid()) {
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->_cur_meta, (bool)(using_attr.flags & BTHREAD_NOSIGNAL) };
ReadyToRunArgs args = {
g->tag(),
g->_cur_meta,
(bool)(using_attr.flags & BTHREAD_NOSIGNAL)
};
g->set_remained(fn, &args);
TaskGroup::sched_to(pg, m->tid);
}
@@ -565,7 +571,6 @@ void TaskGroup::ending_sched(TaskGroup** pg) {
// 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) {
@@ -804,6 +809,11 @@ void TaskGroup::ready_to_run_in_worker_ignoresignal(void* args_in) {
return tls_task_group->push_rq(args->meta->tid);
}

void TaskGroup::priority_to_run(void* args_in) {
ReadyToRunArgs* args = static_cast<ReadyToRunArgs*>(args_in);
return tls_task_group->control()->push_priority_q(args->tag, args->meta->tid);
}

struct SleepArgs {
uint64_t timeout_us;
bthread_t tid;
@@ -978,7 +988,7 @@ int TaskGroup::interrupt(bthread_t tid, TaskControl* c, bthread_tag_t tag) {

void TaskGroup::yield(TaskGroup** pg) {
TaskGroup* g = *pg;
ReadyToRunArgs args = { g->_cur_meta, false };
ReadyToRunArgs args = { g->tag(), g->_cur_meta, false };
g->set_remained(ready_to_run_in_worker, &args);
sched(pg);
}
8 changes: 8 additions & 0 deletions src/bthread/task_group.h
Original file line number Diff line number Diff line change
@@ -22,6 +22,7 @@
#ifndef BTHREAD_TASK_GROUP_H
#define BTHREAD_TASK_GROUP_H

#include <unordered_set>
#include "butil/time.h" // cpuwide_time_ns
#include "bthread/task_control.h"
#include "bthread/task_meta.h" // bthread_t, TaskMeta
@@ -199,6 +200,10 @@ class TaskGroup {
total_ns += butil::cputhread_time_ns() - _last_cpu_clock_ns;
return total_ns;
}
// Thread Unsafe
void add_epoll_tid(bthread_t tid) { _epoll_tids.emplace(tid); }

bool cur_epoll_tid() { return _epoll_tids.count(current_tid()) > 0; }
zhengJade marked this conversation as resolved.
Show resolved Hide resolved

private:
friend class TaskControl;
@@ -218,11 +223,13 @@ friend class TaskControl;
static void _release_last_context(void*);
static void _add_sleep_event(void*);
struct ReadyToRunArgs {
bthread_tag_t tag;
TaskMeta* meta;
bool nosignal;
};
static void ready_to_run_in_worker(void*);
static void ready_to_run_in_worker_ignoresignal(void*);
static void priority_to_run(void*);

// Wait for a task to run.
// Returns true on success, false is treated as permanent error and the
@@ -278,6 +285,7 @@ friend class TaskControl;

// Worker thread id.
pid_t _tid;
std::unordered_set<bthread_t> _epoll_tids;
};

} // namespace bthread
2 changes: 1 addition & 1 deletion src/bthread/task_group_inl.h
Original file line number Diff line number Diff line change
@@ -51,7 +51,7 @@ inline void TaskGroup::exchange(TaskGroup** pg, TaskMeta* next_meta) {
if (g->is_current_pthread_task()) {
return g->ready_to_run(next_meta);
}
ReadyToRunArgs args = { g->_cur_meta, false };
ReadyToRunArgs args = { g->tag(), g->_cur_meta, false };
g->set_remained((g->current_task()->about_to_quit
? ready_to_run_in_worker_ignoresignal
: ready_to_run_in_worker),
1 change: 0 additions & 1 deletion test/bthread_setconcurrency_unittest.cpp
Original file line number Diff line number Diff line change
@@ -214,7 +214,6 @@ int concurrency_by_tag(int num) {

TEST(BthreadTest, concurrency_by_tag) {
ASSERT_EQ(concurrency_by_tag(1), false);
auto tag_con = bthread_getconcurrency_by_tag(0);
auto con = bthread_getconcurrency();
ASSERT_EQ(concurrency_by_tag(con), true);
ASSERT_EQ(concurrency_by_tag(con + 1), true);
Loading