farmer|cmds: Improve remote harvester info (#6979)

* farmer|cmds: List harvester by IP/peer_node_id instead of IP/port * farmer: Cache harvester responses * test: Trigger `update_cached_plots` and adjust `get_plots` format * farmer|cmds|test: Adjust RPC output of `get_plots` * rename: Farmer RPC `get_plots` to `get_harvesters` + rename related code * farmer: Fix remove loop indentation * farmer: Catch exceptions in `_periodically_clear_cache_and_refresh_task` * cmds: Distinguish between local and remote harvesters * cmds: Explicitly check against 0 * cmds: Rename `print_harvesters` -> `process_harvesters`
2026-08-24 10:05:29 -05:00 · 2021-07-06 20:20:21 -07:00
parent 8940a7a007
commit af30ce78a2
5 changed files with 141 additions and 64 deletions
@@ -12,18 +12,19 @@ from chia.util.default_root import DEFAULT_ROOT_PATH
 from chia.util.ints import uint16
 from chia.util.misc import format_bytes
 from chia.util.misc import format_minutes
+from chia.util.network import is_localhost

 SECONDS_PER_BLOCK = (24 * 3600) / 4608


-async def get_plots(farmer_rpc_port: int) -> Optional[Dict[str, Any]]:
+async def get_harvesters(farmer_rpc_port: int) -> Optional[Dict[str, Any]]:
    try:
        config = load_config(DEFAULT_ROOT_PATH, "config.yaml")
        self_hostname = config["self_hostname"]
        if farmer_rpc_port is None:
            farmer_rpc_port = config["farmer"]["rpc_port"]
        farmer_client = await FarmerRpcClient.create(self_hostname, uint16(farmer_rpc_port), DEFAULT_ROOT_PATH, config)
-        plots = await farmer_client.get_plots()
+        plots = await farmer_client.get_harvesters()
    except Exception as e:
        if isinstance(e, aiohttp.ClientConnectorError):
            print(f"Connection error. Check if farmer is running at {farmer_rpc_port}")
@@ -178,7 +179,7 @@ async def challenges(farmer_rpc_port: int, limit: int) -> None:


 async def summary(rpc_port: int, wallet_rpc_port: int, harvester_rpc_port: int, farmer_rpc_port: int) -> None:
-    all_plots = await get_plots(farmer_rpc_port)
+    all_harvesters = await get_harvesters(farmer_rpc_port)
    blockchain_state = await get_blockchain_state(rpc_port)
    farmer_running = await is_farmer_running(farmer_rpc_port)

@@ -211,23 +212,40 @@ async def summary(rpc_port: int, wallet_rpc_port: int, harvester_rpc_port: int,
        print(f"Block rewards: {(amounts['farmer_reward_amount'] + amounts['pool_reward_amount']) / units['chia']}")
        print(f"Last height farmed: {amounts['last_height_farmed']}")

-    total_plot_size = 0
-    total_plots = 0
-    if all_plots is not None:
-        for harvester_ip, plots in all_plots.items():
-            if harvester_ip == "success":
-                # This key is just "success": True
-                continue
-            total_plot_size_harvester = sum(map(lambda x: x["file_size"], plots["plots"]))
-            total_plot_size += total_plot_size_harvester
-            total_plots += len(plots["plots"])
-            print(f"Harvester {harvester_ip}:")
-            print(f"   {len(plots['plots'])} plots of size: {format_bytes(total_plot_size_harvester)}")
+    class PlotStats:
+        total_plot_size = 0
+        total_plots = 0

-        print(f"Plot count for all harvesters: {total_plots}")
+    if all_harvesters is not None:
+        harvesters_local: dict = {}
+        harvesters_remote: dict = {}
+        for harvester in all_harvesters["harvesters"]:
+            ip = harvester["connection"]["host"]
+            if is_localhost(ip):
+                harvesters_local[harvester["connection"]["node_id"]] = harvester
+            else:
+                if ip not in harvesters_remote:
+                    harvesters_remote[ip] = {}
+                harvesters_remote[ip][harvester["connection"]["node_id"]] = harvester
+
+        def process_harvesters(harvester_peers_in: dict):
+            for harvester_peer_id, plots in harvester_peers_in.items():
+                total_plot_size_harvester = sum(map(lambda x: x["file_size"], plots["plots"]))
+                PlotStats.total_plot_size += total_plot_size_harvester
+                PlotStats.total_plots += len(plots["plots"])
+                print(f"   {len(plots['plots'])} plots of size: {format_bytes(total_plot_size_harvester)}")
+
+        if len(harvesters_local) > 0:
+            print(f"Local Harvester{'s' if len(harvesters_local) > 1 else ''}")
+            process_harvesters(harvesters_local)
+        for harvester_ip, harvester_peers in harvesters_remote.items():
+            print(f"Remote Harvester{'s' if len(harvester_peers) > 1 else ''} for IP: {harvester_ip}")
+            process_harvesters(harvester_peers)
+
+        print(f"Plot count for all harvesters: {PlotStats.total_plots}")

        print("Total size of plots: ", end="")
-        print(format_bytes(total_plot_size))
+        print(format_bytes(PlotStats.total_plot_size))
    else:
        print("Plot count: Unknown")
        print("Total size of plots: Unknown")
@@ -239,11 +257,11 @@ async def summary(rpc_port: int, wallet_rpc_port: int, harvester_rpc_port: int,
        print("Estimated network space: Unknown")

    minutes = -1
-    if blockchain_state is not None and all_plots is not None:
-        proportion = total_plot_size / blockchain_state["space"] if blockchain_state["space"] else -1
+    if blockchain_state is not None and all_harvesters is not None:
+        proportion = PlotStats.total_plot_size / blockchain_state["space"] if blockchain_state["space"] else -1
        minutes = int((await get_average_block_time(rpc_port) / 60) / proportion) if proportion else -1

-    if all_plots is not None and total_plots == 0:
+    if all_harvesters is not None and PlotStats.total_plots == 0:
        print("Expected time to win: Never (no plots)")
    else:
        print("Expected time to win: " + format_minutes(minutes))
@@ -50,6 +50,7 @@ log = logging.getLogger(__name__)

 UPDATE_POOL_INFO_INTERVAL: int = 3600
 UPDATE_POOL_FARMER_INFO_INTERVAL: int = 300
+UPDATE_HARVESTER_CACHE_INTERVAL: int = 60

 """
 HARVESTER PROTOCOL (FARMER <-> HARVESTER)
@@ -131,6 +132,8 @@ class Farmer:
        # Last time we updated pool_state based on the config file
        self.last_config_access_time: uint64 = uint64(0)

+        self.harvester_cache: Dict[str, Dict[str, Tuple[Dict, float]]] = {}
+
    async def _start(self):
        self.update_pool_state_task = asyncio.create_task(self._periodically_update_pool_state_task())
        self.cache_clear_task = asyncio.create_task(self._periodically_clear_cache_and_refresh_task())
@@ -490,24 +493,71 @@ class Farmer:

        return None

-    async def get_plots(self) -> Dict:
-        rpc_response = {}
+    async def update_cached_harvesters(self):
+        # First remove outdated cache entries
+        remove_hosts = []
+        for host, host_cache in self.harvester_cache.items():
+            remove_peers = []
+            for peer_id, peer_cache in host_cache.items():
+                _, last_update = peer_cache
+                # If the peer cache hasn't been updated for 10x interval, drop it since the harvester doesn't respond
+                if time.time() - last_update > UPDATE_HARVESTER_CACHE_INTERVAL * 10:
+                    remove_peers.append(peer_id)
+            for key in remove_peers:
+                del host_cache[key]
+            if len(host_cache) == 0:
+                remove_hosts.append(host)
+        for key in remove_hosts:
+            del self.harvester_cache[key]
+        # Now query each harvester and update caches
        for connection in self.server.get_connections():
-            if connection.connection_type == NodeType.HARVESTER:
-                peer_host = connection.peer_host
-                peer_port = connection.peer_port
-                peer_full = f"{peer_host}:{peer_port}"
+            if connection.connection_type != NodeType.HARVESTER:
+                continue
+            cache_entry = await self.get_cached_harvesters(connection)
+            if cache_entry is None or time.time() - cache_entry[1] > UPDATE_HARVESTER_CACHE_INTERVAL:
                response = await connection.request_plots(harvester_protocol.RequestPlots(), timeout=5)
-                if response is None:
+                if response is not None:
+                    if isinstance(response, harvester_protocol.RespondPlots):
+                        if connection.peer_host not in self.harvester_cache:
+                            self.harvester_cache[connection.peer_host] = {}
+
+                        self.harvester_cache[connection.peer_host][connection.peer_node_id.hex()] = (
+                            response.to_json_dict(),
+                            time.time(),
+                        )
+                    else:
+                        self.log.error(
+                            f"Invalid response from harvester:"
+                            f"peer_host {connection.peer_host}, peer_node_id {connection.peer_node_id}"
+                        )
+                else:
                    self.log.error(
                        "Harvester did not respond. You might need to update harvester to the latest version"
                    )
-                    continue
-                if not isinstance(response, harvester_protocol.RespondPlots):
-                    self.log.error(f"Invalid response from harvester: {peer_host}:{peer_port}")
-                    continue
-                rpc_response[peer_full] = response.to_json_dict()
-        return rpc_response
+
+    async def get_cached_harvesters(self, connection: WSChiaConnection) -> Optional[Tuple[Dict, float]]:
+        host_cache = self.harvester_cache.get(connection.peer_host)
+        if host_cache is None:
+            return None
+        return host_cache.get(connection.peer_node_id.hex())
+
+    async def get_harvesters(self) -> Dict:
+        harvesters: List = []
+        for connection in self.server.get_connections():
+            if connection.connection_type != NodeType.HARVESTER:
+                continue
+
+            cache_entry = await self.get_cached_harvesters(connection)
+            if cache_entry is not None:
+                harvester_object: dict = dict(cache_entry[0])
+                harvester_object["connection"] = {
+                    "node_id": connection.peer_node_id.hex(),
+                    "host": connection.peer_host,
+                    "port": connection.peer_port,
+                }
+                harvesters.append(harvester_object)
+
+        return {"harvesters": harvesters}

    async def _periodically_update_pool_state_task(self):
        time_slept: uint64 = uint64(0)
@@ -531,27 +581,34 @@ class Farmer:
        time_slept: uint64 = uint64(0)
        refresh_slept = 0
        while not self._shut_down:
-            if time_slept > self.constants.SUB_SLOT_TIME_TARGET:
-                now = time.time()
-                removed_keys: List[bytes32] = []
-                for key, add_time in self.cache_add_time.items():
-                    if now - float(add_time) > self.constants.SUB_SLOT_TIME_TARGET * 3:
-                        self.sps.pop(key, None)
-                        self.proofs_of_space.pop(key, None)
-                        self.quality_str_to_identifiers.pop(key, None)
-                        self.number_of_responses.pop(key, None)
-                        removed_keys.append(key)
-                for key in removed_keys:
-                    self.cache_add_time.pop(key, None)
-                time_slept = uint64(0)
-                log.debug(
-                    f"Cleared farmer cache. Num sps: {len(self.sps)} {len(self.proofs_of_space)} "
-                    f"{len(self.quality_str_to_identifiers)} {len(self.number_of_responses)}"
-                )
-            time_slept += 1
-            refresh_slept += 1
-            # Periodically refresh GUI to show the correct download/upload rate.
-            if refresh_slept >= 30:
-                self.state_changed("add_connection", {})
-                refresh_slept = 0
+            try:
+                if time_slept > self.constants.SUB_SLOT_TIME_TARGET:
+                    now = time.time()
+                    removed_keys: List[bytes32] = []
+                    for key, add_time in self.cache_add_time.items():
+                        if now - float(add_time) > self.constants.SUB_SLOT_TIME_TARGET * 3:
+                            self.sps.pop(key, None)
+                            self.proofs_of_space.pop(key, None)
+                            self.quality_str_to_identifiers.pop(key, None)
+                            self.number_of_responses.pop(key, None)
+                            removed_keys.append(key)
+                    for key in removed_keys:
+                        self.cache_add_time.pop(key, None)
+                    time_slept = uint64(0)
+                    log.debug(
+                        f"Cleared farmer cache. Num sps: {len(self.sps)} {len(self.proofs_of_space)} "
+                        f"{len(self.quality_str_to_identifiers)} {len(self.number_of_responses)}"
+                    )
+                time_slept += 1
+                refresh_slept += 1
+                # Periodically refresh GUI to show the correct download/upload rate.
+                if refresh_slept >= 30:
+                    self.state_changed("add_connection", {})
+                    refresh_slept = 0
+
+                # Handles harvester plots cache cleanup and updates
+                await self.update_cached_harvesters()
+            except Exception:
+                log.error(f"_periodically_clear_cache_and_refresh_task failed: {traceback.print_exc()}")
+
            await asyncio.sleep(1)
@@ -19,7 +19,7 @@ class FarmerRpcApi:
            "/set_reward_targets": self.set_reward_targets,
            "/get_pool_state": self.get_pool_state,
            "/set_payout_instructions": self.set_payout_instructions,
-            "/get_plots": self.get_plots,
+            "/get_harvesters": self.get_harvesters,
            "/get_pool_login_link": self.get_pool_login_link,
        }

@@ -112,8 +112,8 @@ class FarmerRpcApi:
        await self.service.set_payout_instructions(launcher_id, request["payout_instructions"])
        return {}

-    async def get_plots(self, _: Dict):
-        return await self.service.get_plots()
+    async def get_harvesters(self, _: Dict):
+        return await self.service.get_harvesters()

    async def get_pool_login_link(self, request: Dict) -> Dict:
        launcher_id: bytes32 = bytes32(hexstr_to_bytes(request["launcher_id"]))
@@ -49,8 +49,8 @@ class FarmerRpcClient(RpcClient):
        request = {"launcher_id": launcher_id.hex(), "payout_instructions": payout_instructions}
        return await self.fetch("set_payout_instructions", request)

-    async def get_plots(self) -> Dict[str, Any]:
-        return await self.fetch("get_plots", {})
+    async def get_harvesters(self) -> Dict[str, Any]:
+        return await self.fetch("get_harvesters", {})

    async def get_pool_login_link(self, launcher_id: bytes32) -> Optional[str]:
        try:
@@ -165,9 +165,11 @@ class TestRpc:
            res_2 = await client_2.get_plots()
            assert len(res_2["plots"]) == num_plots

-            # Test farmer get_plots
-            farmer_res = await client.get_plots()
-            assert len(list(farmer_res.values())[0]["plots"]) == num_plots
+            # Test farmer get_harvesters
+            await farmer_rpc_api.service.update_cached_harvesters()
+            farmer_res = await client.get_harvesters()
+            assert len(list(farmer_res["harvesters"])) == 1
+            assert len(list(farmer_res["harvesters"][0]["plots"])) == num_plots

            assert len(await client_2.get_plot_directories()) == 1