| 1239 | return Void(); |
| 1240 | } |
| 1241 | when(std::pair<StorageServerInterface, ProcessClass> newInterface = wait(interfaceChanged)) { |
| 1242 | auto const& lastKnownInterface = server->getLastKnownInterface(); |
| 1243 | bool restartRecruiting = newInterface.first.waitFailure.getEndpoint().getPrimaryAddress() != |
| 1244 | lastKnownInterface.waitFailure.getEndpoint().getPrimaryAddress(); |
| 1245 | bool localityChanged = lastKnownInterface.locality != newInterface.first.locality; |
| 1246 | bool machineLocalityChanged = |
| 1247 | lastKnownInterface.locality.zoneId().get() != newInterface.first.locality.zoneId().get(); |
| 1248 | TraceEvent("StorageServerInterfaceChanged", self->distributorId) |
| 1249 | .detail("ServerID", server->getId()) |
| 1250 | .detail("NewWaitFailureToken", newInterface.first.waitFailure.getEndpoint().token) |
| 1251 | .detail("OldWaitFailureToken", lastKnownInterface.waitFailure.getEndpoint().token) |
| 1252 | .detail("LocalityChanged", localityChanged) |
| 1253 | .detail("MachineLocalityChanged", machineLocalityChanged); |
| 1254 | |
| 1255 | server->updateLastKnown(newInterface.first, newInterface.second); |
| 1256 | if (localityChanged && !isTss) { |
| 1257 | CODE_PROBE(true, "Server locality changed"); |
| 1258 | |
| 1259 | // The locality change of a server will affect machine teams related to the server if |
| 1260 | // the server's machine locality is changed |
| 1261 | if (machineLocalityChanged) { |
| 1262 | // First handle the impact on the machine of the server on the old locality |
| 1263 | Reference<TCMachineInfo> machine = server->machine; |
| 1264 | ASSERT_GE(machine->serversOnMachine.size(), 1); |
| 1265 | if (machine->serversOnMachine.size() == 1) { |
| 1266 | // When server is the last server on the machine, |
| 1267 | // remove the machine and the related machine team |
| 1268 | self->removeMachine(machine); |
| 1269 | server->machine = Reference<TCMachineInfo>(); |
| 1270 | } else { |
| 1271 | // we remove the server from the machine, and |
| 1272 | // update locality entry for the machine and the global machineLocalityMap |
| 1273 | int serverIndex = -1; |
| 1274 | for (int i = 0; i < machine->serversOnMachine.size(); ++i) { |
| 1275 | if (machine->serversOnMachine[i].getPtr() == server) { |
| 1276 | // NOTE: now the machine's locality is wrong. Need update it whenever uses |
| 1277 | // it. |
| 1278 | serverIndex = i; |
| 1279 | machine->serversOnMachine[i] = machine->serversOnMachine.back(); |
| 1280 | machine->serversOnMachine.pop_back(); |
| 1281 | break; // Invariant: server only appear on the machine once |
| 1282 | } |
| 1283 | } |
| 1284 | ASSERT(serverIndex != -1); |
| 1285 | // NOTE: we do not update the machine's locality map even when |
| 1286 | // its representative server is changed. |
| 1287 | } |
| 1288 | |
| 1289 | // Second handle the impact on the destination machine where the server's new locality |
| 1290 | // is; If the destination machine is new, create one; otherwise, add server to an |
| 1291 | // existing one Update server's machine reference to the destination machine |
| 1292 | Reference<TCMachineInfo> destMachine = |
| 1293 | self->checkAndCreateMachine(self->server_info[server->getId()]); |
| 1294 | ASSERT(destMachine.isValid()); |
| 1295 | } |
| 1296 | |
| 1297 | // Ensure the server's server team belong to a machine team, and |
| 1298 | // Get the newBadTeams due to the locality change |
no test coverage detected