| 1186 | previous_counters=previous_counters, |
| 1187 | current_ts=current_ts, |
| 1188 | current_counters=current_counters, |
| 1189 | ) |
| 1190 | previous_ts = current_ts |
| 1191 | previous_counters = current_counters |
| 1192 | |
| 1193 | try: |
| 1194 | current_ts, current_counters = self._read_snapshot() |
| 1195 | except Exception as exc: |
| 1196 | self._error = str(exc) |
| 1197 | logger.warning("⚠️ 网络采样最终读取失败: target=%s err=%s", self.target, exc) |
| 1198 | return |
| 1199 | self._record_delta( |
| 1200 | previous_ts=previous_ts, |
| 1201 | previous_counters=previous_counters, |
| 1202 | current_ts=current_ts, |
| 1203 | current_counters=current_counters, |
| 1204 | ) |
| 1205 | |
| 1206 | |
| 1207 | class BenchmarkWorkerStop(RuntimeError): |
| 1208 | """Worker exits because the benchmark window is closed and useful work is exhausted.""" |
| 1209 | |
| 1210 | |
| 1211 | class BenchmarkNode: |
| 1212 | def __init__(self): |
| 1213 | self.test_config: Optional[Dict[str, Any]] = None |
| 1214 | self.node_id: str = f"node_{uuid.uuid4().hex[:8]}" |
| 1215 | self.kv_store: Optional[Any] = None |
| 1216 | self.fluxon_client: Optional[KvClient] = None |
| 1217 | self.channel_id: Optional[str] = None # Channel ID |
| 1218 | # Coordinator address must be provided by CLI; keep unset until main() assigns. |
| 1219 | self.coordinator_host: str = "" |
| 1220 | self.coordinator_port: int = 0 |
| 1221 | self.operation_results: List[OperationResult] = [] |
| 1222 | self.start_time: Optional[float] = None |
| 1223 | self.end_time: Optional[float] = None |
| 1224 | self.key_prefix: Optional[str] = None |
| 1225 | self.instance_key: Optional[str] = None |
| 1226 | # MQ/channel state is encapsulated in MQState. |
| 1227 | self.mq_state = MQState() |
| 1228 | self.chan_config: Dict[str, Any] = CHAN_CONFIG.copy() |
| 1229 | self.mq_unique_id: str = "" |
| 1230 | # Optional: simulate MQ consumer handling time (ms range) |
| 1231 | # Shape: (min_ms, max_ms), assigned by coordinator. |
| 1232 | self.consumer_sim_handle_ms_range = None |
| 1233 | self.value_size_mode: str = ValueSizeMode.FIXED.value |
| 1234 | self.value_size_weighted_set: List[Tuple[int, float]] = [] |
| 1235 | self._payload_pool_by_size: Dict[int, Tuple[bytes, ...]] = {} |
| 1236 | self._payload_pool_by_size: Dict[int, Tuple[bytes, ...]] = {} |
| 1237 | |
| 1238 | # Reuse self.end_time as the metrics window end: |
| 1239 | # - KV mode: set to deadline_ts |
| 1240 | # - MPMC mode: set to the time when the main thread broadcasts stop intent |
| 1241 | |
| 1242 | self._inflight_lock = threading.Lock() |
| 1243 | self._inflight_requests = 0 |
| 1244 | |
| 1245 | # Progress snapshot updated by worker threads; read by heartbeat thread. |