diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 72a6b91836..777d7514a3 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -1018,8 +1018,11 @@ void TaskGroup::priority_to_run(void* args_in) { if (args->meta->priority_index < 0) { return g->push_rq(args->meta->tid); } - return g->control()->push_ed_priority_queue( + g->control()->push_ed_priority_queue( args->tag, args->meta->priority_index, args->meta->tid); + + ++g->_nsignaled; + g->control()->signal_task(1, args->tag); } struct SleepArgs { diff --git a/test/bthread_priority_queue_unittest.cpp b/test/bthread_priority_queue_unittest.cpp index d6c9e43a98..42f6dd85db 100644 --- a/test/bthread_priority_queue_unittest.cpp +++ b/test/bthread_priority_queue_unittest.cpp @@ -138,8 +138,9 @@ TEST_F(PriorityQueueTest, start_foreground_priority_to_run) { struct EDSimArg { int n_tasks; + int error_code; }; - EDSimArg ed_arg{N}; + EDSimArg ed_arg{N, 0}; auto ed_fn = [](void* arg) -> void* { EDSimArg* ea = static_cast(arg); @@ -147,10 +148,21 @@ TEST_F(PriorityQueueTest, start_foreground_priority_to_run) { bthread::TaskGroup::address_meta(bthread_self()); meta->priority_index = 0; + std::vector children; + children.reserve(ea->n_tasks); for (int i = 0; i < ea->n_tasks; ++i) { TaskArg* ta = new TaskArg{i}; bthread_t child; - bthread_start_urgent(&child, NULL, priority_task_fn, ta); + int rc = bthread_start_urgent(&child, NULL, priority_task_fn, ta); + if (rc != 0) { + delete ta; + ea->error_code = rc; + break; + } + children.push_back(child); + } + for (auto child : children) { + bthread_join(child, NULL); } return NULL; }; @@ -161,7 +173,8 @@ TEST_F(PriorityQueueTest, start_foreground_priority_to_run) { bthread_t ed_tid; ASSERT_EQ(0, bthread_start_background(&ed_tid, &priority_attr, ed_fn, &ed_arg)); - bthread_join(ed_tid, NULL); + ASSERT_EQ(0, bthread_join(ed_tid, NULL)); + ASSERT_EQ(0, ed_arg.error_code); ASSERT_EQ(N, g_priority_count.load()); std::lock_guard lk(g_tid_mutex); @@ -180,6 +193,7 @@ TEST_F(PriorityQueueTest, multiple_eds_concurrent_preempt) { int ed_index; int n_children; std::atomic* resume_count; + int error_code; }; auto ed_fn = [](void* arg) -> void* { @@ -194,7 +208,13 @@ TEST_F(PriorityQueueTest, multiple_eds_concurrent_preempt) { int id = ea->ed_index * ea->n_children + i; TaskArg* ta = new TaskArg{id}; bthread_t child; - bthread_start_urgent(&child, NULL, priority_task_fn, ta); + const int rc = bthread_start_urgent( + &child, NULL, priority_task_fn, ta); + if (rc != 0) { + delete ta; + ea->error_code = rc; + break; + } children.push_back(child); ea->resume_count->fetch_add(1, std::memory_order_relaxed); } @@ -210,13 +230,14 @@ TEST_F(PriorityQueueTest, multiple_eds_concurrent_preempt) { std::vector ed_args(NUM_EDS); std::vector ed_tids(NUM_EDS); for (int i = 0; i < NUM_EDS; ++i) { - ed_args[i] = {i, TASKS_PER_ED, &resume_count}; + ed_args[i] = {i, TASKS_PER_ED, &resume_count, 0}; ASSERT_EQ(0, bthread_start_background(&ed_tids[i], &priority_attr, ed_fn, &ed_args[i])); } for (int i = 0; i < NUM_EDS; ++i) { - bthread_join(ed_tids[i], NULL); + ASSERT_EQ(0, bthread_join(ed_tids[i], NULL)); + ASSERT_EQ(0, ed_args[i].error_code); } ASSERT_EQ(TOTAL, g_priority_count.load());