| 1734 | } |
| 1735 | |
| 1736 | void Scheduler::AssignmentCtx::SelectExecutorOnHost( |
| 1737 | const IpAddr& executor_ip, BackendDescriptorPB* executor) { |
| 1738 | DCHECK(executor_group_.LookUpExecutorIp(executor_ip, nullptr)); |
| 1739 | const ExecutorGroup::Executors& executors_on_host = |
| 1740 | executor_group_.GetExecutorsForHost(executor_ip); |
| 1741 | DCHECK(executors_on_host.size() > 0); |
| 1742 | if (executors_on_host.size() == 1) { |
| 1743 | *executor = *executors_on_host.begin(); |
| 1744 | } else { |
| 1745 | ExecutorGroup::Executors::const_iterator* next_executor_on_host; |
| 1746 | next_executor_on_host = |
| 1747 | FindOrInsert(&next_executor_per_host_, executor_ip, executors_on_host.begin()); |
| 1748 | auto eq = [next_executor_on_host](auto& elem) { |
| 1749 | const BackendDescriptorPB& next_executor = **next_executor_on_host; |
| 1750 | // The IP addresses must already match, so it is sufficient to check the port. |
| 1751 | DCHECK_EQ(next_executor.ip_address(), elem.ip_address()); |
| 1752 | return next_executor.address().port() == elem.address().port(); |
| 1753 | }; |
| 1754 | DCHECK(find_if(executors_on_host.begin(), executors_on_host.end(), eq) |
| 1755 | != executors_on_host.end()); |
| 1756 | *executor = **next_executor_on_host; |
| 1757 | // Rotate |
| 1758 | ++(*next_executor_on_host); |
| 1759 | if (*next_executor_on_host == executors_on_host.end()) { |
| 1760 | *next_executor_on_host = executors_on_host.begin(); |
| 1761 | } |
| 1762 | } |
| 1763 | } |
| 1764 | |
| 1765 | void TScanRangeToScanRangePB(const TScanRange& tscan_range, ScanRangePB* scan_range_pb) { |
| 1766 | if (tscan_range.__isset.hdfs_file_split) { |
no test coverage detected