| 66 | } |
| 67 | |
| 68 | void ExecutionQueueBase::start_execute(TaskNode* node) { |
| 69 | node->next = TaskNode::UNCONNECTED; |
| 70 | node->status = TaskNode::UNEXECUTED; |
| 71 | node->iterated = false; |
| 72 | if (node->high_priority) { |
| 73 | // Add _high_priority_tasks before pushing this task into queue to |
| 74 | // make sure that _execute_tasks sees the newest number when this |
| 75 | // task is in the queue. Although there might be some useless for |
| 76 | // loops in _execute_tasks if this thread is scheduled out at this |
| 77 | // point, we think it's just fine. |
| 78 | _high_priority_tasks.fetch_add(1, butil::memory_order_relaxed); |
| 79 | } |
| 80 | TaskNode* const prev_head = _head.exchange(node, butil::memory_order_release); |
| 81 | if (prev_head != NULL) { |
| 82 | node->next = prev_head; |
| 83 | return; |
| 84 | } |
| 85 | // Get the right to execute the task, start a bthread to avoid deadlock |
| 86 | // or stack overflow |
| 87 | node->next = NULL; |
| 88 | node->q = this; |
| 89 | |
| 90 | ExecutionQueueVars* const vars = get_execq_vars(); |
| 91 | vars->execq_active_count << 1; |
| 92 | if (node->in_place) { |
| 93 | int niterated = 0; |
| 94 | _execute(node, node->high_priority, &niterated); |
| 95 | TaskNode* tmp = node; |
| 96 | // return if no more |
| 97 | if (node->high_priority) { |
| 98 | _high_priority_tasks.fetch_sub(niterated, butil::memory_order_relaxed); |
| 99 | } |
| 100 | if (!_more_tasks(tmp, &tmp, !node->iterated)) { |
| 101 | vars->execq_active_count << -1; |
| 102 | return_task_node(node); |
| 103 | return; |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | if (nullptr == _options.executor) { |
| 108 | if (_options.use_pthread) { |
| 109 | if (_pthread_started) { |
| 110 | BAIDU_SCOPED_LOCK(_mutex); |
| 111 | _current_head = node; |
| 112 | _cond.Signal(); |
| 113 | } else { |
| 114 | // Start the execution bthread in background once. |
| 115 | if (pthread_create(&_pid, NULL, |
| 116 | _execute_tasks_pthread, |
| 117 | node) != 0) { |
| 118 | PLOG(FATAL) << "Fail to create pthread"; |
| 119 | _execute_tasks(node); |
| 120 | } |
| 121 | _pthread_started = true; |
| 122 | } |
| 123 | } else { |
| 124 | bthread_t tid; |
| 125 | // We start the execution bthread in background instead of foreground as |
nothing calls this directly
no test coverage detected