chenBright commented on code in PR #3633:
URL: https://github.com/apache/brpc/pull/3633#discussion_r4436113448
##########
src/bthread/task_group.cpp:
##########
@@ -848,13 +851,16 @@ void TaskGroup::ready_to_run(TaskMeta* meta, bool
nosignal) {
_control->_task_tracer.set_status(TASK_STATUS_READY, meta);
#endif // BRPC_BTHREAD_TRACER
push_rq(meta->tid);
+ // push_rq() may switch context and the current task may be resumed by
+ // another group. Use the current group for all subsequent accounting.
+ TaskGroup* g = BAIDU_GET_VOLATILE_THREAD_LOCAL(tls_task_group);
Review Comment:
push_rq() can now suspend the caller and resume it on another worker.
Updating the local pointer in ready_to_run() does not update its callers: after
butex_wake_except() wakes many joiners, task_runner() still uses its old g to
install _release_last_context and call ending_sched().
##########
src/bthread/task_group_inl.h:
##########
@@ -83,22 +83,138 @@ inline void TaskGroup::sched_to(TaskGroup** pg, bthread_t
next_tid) {
}
inline void TaskGroup::push_rq(bthread_t tid) {
- while (!_rq.push(tid)) {
+ TaskGroup* g = this;
+ while (!g->_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
// brpc)
- flush_nosignal_tasks();
- LOG_EVERY_SECOND(ERROR) << "_rq is full, capacity=" << _rq.capacity();
- // 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.
- ::usleep(1000);
+ g->flush_nosignal_tasks();
+ LOG_EVERY_SECOND(ERROR) << "_rq is full, capacity=" <<
g->_rq.capacity();
+ const auto pop_task = [g](bthread_t* t) -> bool {
+#ifndef BTHREAD_FAIR_WSQ
+ return g->_rq.pop(t);
+#else
+ return g->_rq.steal(t);
+#endif
+ };
+ if (g->_sched_recursive_guard != 0) {
+ // Remained callbacks run with the guard held, either in
+ // sched_to()'s remained loop or at task_runner() entry.
Re-entering
+ // sched_to() here would be recursive, while waiting here can
+ // livelock if every worker is running a remained callback. Make
+ // room for |tid| and requeue the popped task after the guard is
+ // balanced.
+ CHECK(g->_pending_rq_tid == INVALID_BTHREAD);
+ bthread_t pending_tid;
+ if (!pop_task(&pending_tid)) {
+ continue;
+ }
+ g->_pending_rq_tid = pending_tid;
+ const bool pushed = g->_rq.push(tid);
+ CHECK(pushed) << "push_rq: failed to push after popping a task";
+ return;
+ }
+ if (g->is_current_pthread_task() && !g->is_current_main_task()) {
+ // A pthread-mode bthread runs on _main_stack and cannot be
+ // scheduled out, fall back to waiting.
+ ::usleep(1000);
+ continue;
+ }
+ bthread_t run_tid = 0;
+ if (!pop_task(&run_tid)) {
+ // The queue was drained by stealers between the failed push and
+ // the pop, retrying the push will succeed.
+ continue;
+ }
+ if (g->is_current_main_task() &&
+ address_meta(run_tid)->attr.stack_type ==
BTHREAD_STACKTYPE_PTHREAD) {
+ // pthread-mode tasks are run inline on _main_stack by the
+ // dispatch loops (run_main_task/task_runner). Switching to one
+ // from the main task itself causes no stack jump and would
+ // silently drop it, so it cannot be run here. But pushing it
+ // straight back makes the LIFO pop() return this same task
+ // forever, hiding any runnable task sitting deeper in the
+ // queue. Hold pthread-mode tasks aside and keep popping until
+ // a runnable one shows up.
+ // Bounded budget: the array must live on the worker's stack
+ // (no allocation on a resource-exhaustion path), and held
+ // tasks are invisible to stealers, so keeping the list short
+ // limits how long they can be delayed. 16 * sizeof(bthread_t)
+ // == 128 bytes == two cache lines.
+ bthread_t pending[16];
+ int npending = 0;
+ pending[npending++] = run_tid;
+ bool found = false;
+ while (npending < 16 && pop_task(&run_tid)) {
+ if (address_meta(run_tid)->attr.stack_type !=
+ BTHREAD_STACKTYPE_PTHREAD) {
+ found = true;
+ break;
+ }
+ pending[npending++] = run_tid;
+ }
+ // Push the held tasks back. The pops above freed at least
+ // |npending| slots, _rq has a single producer (this worker)
+ // and stealers never push, so these pushes must succeed.
+ // Reverse order restores their relative order in the queue.
+ for (int i = npending - 1; i >= 0; --i) {
+ const bool pushed = g->_rq.push(pending[i]);
+ CHECK(pushed) << "push_rq: failed to put back held task";
+ }
+ if (!found) {
+ // Everything within reach is pthread-mode: wait for
+ // stealers to drain the queue.
+ ::usleep(1000);
+ continue;
+ }
+ // Fall through: run_tid is a runnable task, run it inline
+ // below to free a slot for |tid|.
+ }
+ // Run the popped bthread in place instead of spinning, otherwise
+ // all workers may spin here and no one drains the queues any more.
+ // The pop also frees one slot, which guarantees that the deferred
+ // push of the current task succeeds: _rq has a single producer (the
+ // owning worker) and stealers never push.
+ // |args| lives on the stack that sched_to() is about to freeze and is
+ // read by the remained-callback after the switch, so it must stay in
+ // this scope: if it were declared inside the branch below, its slot
+ // could be reused before the callback has finished reading it.
+ ReadyToRunArgs args = { g->tag(), nullptr, false };
+ if (!g->is_current_main_task()) {
+ // Preserve the task being scheduled out. The push is deferred to
+ // a remained-callback: publishing a still-running task into _rq
+ // lets a stealer jump into it and run the same context
+ // concurrently on another worker. The callback runs on the stack
+ // of |run_tid| after sched_to() has frozen the current context,
+ // same as exchange(). Capture the meta now, current_tid() is
+ // |run_tid| by the time the callback runs.
+ // The callback may re-enter push_rq() while _rq is still full;
+ // the guard branch above makes room without switching and defers
+ // the popped task until the guard is balanced.
+ args.meta = g->_cur_meta;
+ g->set_remained(ready_to_run_in_worker_ignoresignal, &args);
+ }
+ // The main task is never queued; ending_sched() resumes it via
+ // _main_tid when the queue drains. Jumping to a pthread-mode task
+ // from a normal bthread resumes the dispatch loop on _main_stack
+ // which runs it inline, same as ending_sched() does.
+ sched_to(&g, run_tid);
Review Comment:
This switch makes bthread_start_background() an implicit scheduling point.
Existing callers, including ExecutionQueueBase::start_execute(), deliberately
use background submission because they may hold a pthread mutex.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
