(self, start_port, count, random_start=True)
| 102 | filelock.release() |
| 103 | |
| 104 | def get_port_range(self, start_port, count, random_start=True): |
| 105 | assert count > 0 |
| 106 | if start_port and self._no_random_ports(): |
| 107 | return start_port |
| 108 | |
| 109 | candidates = [] |
| 110 | |
| 111 | def drop_candidates(): |
| 112 | for port in candidates: |
| 113 | self._release_port_no_lock(port) |
| 114 | candidates[:] = [] |
| 115 | |
| 116 | with self._lock: |
| 117 | for attempts in six.moves.range(128): |
| 118 | for left, right in self._valid_range: |
| 119 | if right - left < count: |
| 120 | continue |
| 121 | |
| 122 | if random_start: |
| 123 | start = random.randint(left, right - ((right - left) // 2)) |
| 124 | else: |
| 125 | start = left |
| 126 | for probe_port in six.moves.range(start, right): |
| 127 | if self._capture_port_no_lock(probe_port, socket.SOCK_STREAM): |
| 128 | candidates.append(probe_port) |
| 129 | else: |
| 130 | drop_candidates() |
| 131 | |
| 132 | if len(candidates) == count: |
| 133 | return candidates[0] |
| 134 | # Can't find required number of ports without gap in the current range |
| 135 | drop_candidates() |
| 136 | |
| 137 | raise PortManagerException( |
| 138 | "Failed to find valid port range (start_port: {} count: {}) (range: {} used: {})".format( |
| 139 | start_port, count, self._valid_range, self._filelocks |
| 140 | ) |
| 141 | ) |
| 142 | |
| 143 | def _count_valid_ports(self): |
| 144 | res = 0 |
nothing calls this directly
no test coverage detected