Update information on tracker.
(self, period_update=False)
| 345 | return new_keys |
| 346 | |
| 347 | def _update_tracker(self, period_update=False): |
| 348 | """Update information on tracker.""" |
| 349 | try: |
| 350 | if self._tracker_conn is None: |
| 351 | self._tracker_conn = socket.socket( |
| 352 | base.get_addr_family(self._tracker_addr), socket.SOCK_STREAM |
| 353 | ) |
| 354 | self._tracker_conn.connect(self._tracker_addr) |
| 355 | self._tracker_conn.sendall(struct.pack("<i", base.RPC_TRACKER_MAGIC)) |
| 356 | magic = struct.unpack("<i", base.recvall(self._tracker_conn, 4))[0] |
| 357 | if magic != base.RPC_TRACKER_MAGIC: |
| 358 | self.loop.stop() |
| 359 | raise RuntimeError(f"{self._tracker_addr} is not RPC Tracker") |
| 360 | # just connect to tracker, need to update all keys |
| 361 | self._tracker_pending_puts = self._server_pool.keys() |
| 362 | |
| 363 | if self._tracker_conn and period_update: |
| 364 | # periodically update tracker information |
| 365 | # regenerate key if the key is not in tracker anymore |
| 366 | # and there is no in-coming connection after timeout_alloc |
| 367 | base.sendjson(self._tracker_conn, [TrackerCode.GET_PENDING_MATCHKEYS]) |
| 368 | pending_keys = set(base.recvjson(self._tracker_conn)) |
| 369 | update_keys = [] |
| 370 | for k, v in self._server_pool.items(): |
| 371 | if k not in pending_keys: |
| 372 | if v.alloc_time is None: |
| 373 | v.alloc_time = time.time() |
| 374 | elif time.time() - v.alloc_time > self.timeout_alloc: |
| 375 | update_keys.append(k) |
| 376 | v.alloc_time = None |
| 377 | if update_keys: |
| 378 | logging.info( |
| 379 | "RPCProxy: No incoming conn on %s, regenerate keys...", str(update_keys) |
| 380 | ) |
| 381 | new_keys = self._regenerate_server_keys(update_keys) |
| 382 | self._tracker_pending_puts += new_keys |
| 383 | |
| 384 | need_update_info = False |
| 385 | # report new connections |
| 386 | for key in self._tracker_pending_puts: |
| 387 | rpc_key, _ = base.split_random_key(key) |
| 388 | base.sendjson( |
| 389 | self._tracker_conn, [TrackerCode.PUT, rpc_key, (self._listen_port, key), None] |
| 390 | ) |
| 391 | assert base.recvjson(self._tracker_conn) == TrackerCode.SUCCESS |
| 392 | if rpc_key not in self._key_set: |
| 393 | self._key_set.add(rpc_key) |
| 394 | need_update_info = True |
| 395 | |
| 396 | if need_update_info: |
| 397 | keylist = "[" + ",".join(self._key_set) + "]" |
| 398 | cinfo = {"key": "server:proxy" + keylist, "addr": [None, self._listen_port]} |
| 399 | base.sendjson(self._tracker_conn, [TrackerCode.UPDATE_INFO, cinfo]) |
| 400 | assert base.recvjson(self._tracker_conn) == TrackerCode.SUCCESS |
| 401 | self._tracker_pending_puts = [] |
| 402 | except OSError as err: |
| 403 | logging.info( |
| 404 | "Lost tracker connection: %s, try reconnect in %g sec", |