| 142 | } |
| 143 | |
| 144 | void* ExecutionQueueBase::_execute_tasks(void* arg) { |
| 145 | ExecutionQueueVars* vars = get_execq_vars(); |
| 146 | TaskNode* head = (TaskNode*)arg; |
| 147 | ExecutionQueueBase* m = (ExecutionQueueBase*)head->q; |
| 148 | TaskNode* cur_tail = NULL; |
| 149 | bool destroy_queue = false; |
| 150 | for (;;) { |
| 151 | if (head->iterated) { |
| 152 | CHECK(head->next != NULL); |
| 153 | TaskNode* saved_head = head; |
| 154 | head = head->next; |
| 155 | m->return_task_node(saved_head); |
| 156 | } |
| 157 | int rc = 0; |
| 158 | if (m->_high_priority_tasks.load(butil::memory_order_relaxed) > 0) { |
| 159 | int nexecuted = 0; |
| 160 | // Don't care the return value |
| 161 | rc = m->_execute(head, true, &nexecuted); |
| 162 | m->_high_priority_tasks.fetch_sub( |
| 163 | nexecuted, butil::memory_order_relaxed); |
| 164 | if (nexecuted == 0) { |
| 165 | // Some high_priority tasks are not in queue |
| 166 | sched_yield(); |
| 167 | } |
| 168 | } else { |
| 169 | rc = m->_execute(head, false, NULL); |
| 170 | } |
| 171 | if (rc == ESTOP) { |
| 172 | destroy_queue = true; |
| 173 | } |
| 174 | // Release TaskNode until uniterated task or last task |
| 175 | while (head->next != NULL && head->iterated) { |
| 176 | TaskNode* saved_head = head; |
| 177 | head = head->next; |
| 178 | m->return_task_node(saved_head); |
| 179 | } |
| 180 | if (cur_tail == NULL) { |
| 181 | for (cur_tail = head; cur_tail->next != NULL; |
| 182 | cur_tail = cur_tail->next) {} |
| 183 | } |
| 184 | // break when no more tasks and head has been executed |
| 185 | if (!m->_more_tasks(cur_tail, &cur_tail, !head->iterated)) { |
| 186 | CHECK_EQ(cur_tail, head); |
| 187 | CHECK(head->iterated); |
| 188 | m->return_task_node(head); |
| 189 | break; |
| 190 | } |
| 191 | } |
| 192 | if (destroy_queue) { |
| 193 | CHECK(m->_head.load(butil::memory_order_relaxed) == NULL); |
| 194 | CHECK(m->_stopped); |
| 195 | // Add _join_butex by 2 to make it equal to the next version of the |
| 196 | // ExecutionQueue from the same slot so that join with old id would |
| 197 | // return immediately. |
| 198 | // |
| 199 | // 1: release fence to make join sees the newest changes when it sees |
| 200 | // the newest _join_butex |
| 201 | m->_join_butex->fetch_add(2, butil::memory_order_release/*1*/); |
nothing calls this directly
no test coverage detected