| 250 | } |
| 251 | |
| 252 | Status TaskSchedulerImpl::ExecuteMore(size_t thread_id, int num_tasks_to_execute, |
| 253 | bool execute_all) { |
| 254 | num_tasks_to_execute = std::max(1, num_tasks_to_execute); |
| 255 | |
| 256 | int last_id = 0; |
| 257 | for (;;) { |
| 258 | if (aborted_.value.load()) { |
| 259 | return Status::Cancelled("Scheduler cancelled"); |
| 260 | } |
| 261 | |
| 262 | // Pick next bundle of tasks |
| 263 | const auto& tasks = PickTasks(num_tasks_to_execute, last_id); |
| 264 | if (tasks.empty()) { |
| 265 | break; |
| 266 | } |
| 267 | last_id = tasks.back().first; |
| 268 | |
| 269 | // Execute picked tasks immediately |
| 270 | for (size_t i = 0; i < tasks.size(); ++i) { |
| 271 | int group_id = tasks[i].first; |
| 272 | int64_t task_id = tasks[i].second; |
| 273 | bool task_group_finished = false; |
| 274 | Status status = ExecuteTask(thread_id, group_id, task_id, &task_group_finished); |
| 275 | if (!status.ok()) { |
| 276 | // Mark the current and remaining picked tasks as finished |
| 277 | for (size_t j = i; j < tasks.size(); ++j) { |
| 278 | if (PostExecuteTask(thread_id, tasks[j].first)) { |
| 279 | bool all_task_groups_finished = false; |
| 280 | RETURN_NOT_OK( |
| 281 | OnTaskGroupFinished(thread_id, group_id, &all_task_groups_finished)); |
| 282 | if (all_task_groups_finished) { |
| 283 | return Status::OK(); |
| 284 | } |
| 285 | } |
| 286 | } |
| 287 | return status; |
| 288 | } else { |
| 289 | if (task_group_finished) { |
| 290 | bool all_task_groups_finished = false; |
| 291 | RETURN_NOT_OK( |
| 292 | OnTaskGroupFinished(thread_id, group_id, &all_task_groups_finished)); |
| 293 | if (all_task_groups_finished) { |
| 294 | return Status::OK(); |
| 295 | } |
| 296 | } |
| 297 | } |
| 298 | } |
| 299 | |
| 300 | if (!execute_all) { |
| 301 | num_tasks_to_execute -= static_cast<int>(tasks.size()); |
| 302 | if (num_tasks_to_execute == 0) { |
| 303 | break; |
| 304 | } |
| 305 | } |
| 306 | } |
| 307 | |
| 308 | return Status::OK(); |
| 309 | } |