From fbaf8722d1a4f956e99233ac176422b03034ff9c Mon Sep 17 00:00:00 2001 From: DevTekVE Date: Thu, 13 Jun 2024 09:56:54 +0200 Subject: [PATCH] Refactor athenad and sunnylinkd python code Several changes have been made to improve the code structure of the athenad and sunnylinkd scripts. The while loop in athenad now uses end_event.wait(5) for cleaner execution control. In sunnylinkd.py, there have been additions to manage exceptions and refinements in the handling of network checks. The final major change is the inclusion of a new constant for sunnylink reconnection timeout that plays major role in various parts of the script. --- system/athena/athenad.py | 6 ++---- system/athena/sunnylinkd.py | 36 ++++++++++++++++++++---------------- 2 files changed, 22 insertions(+), 20 deletions(-) diff --git a/system/athena/athenad.py b/system/athena/athenad.py index e8ef682026..71678d534e 100755 --- a/system/athena/athenad.py +++ b/system/athena/athenad.py @@ -634,6 +634,7 @@ def log_handler(end_event: threading.Event, log_attr_name=LOG_ATTR_NAME) -> None is_sunnylink = log_attr_name != LOG_ATTR_NAME if PC: cloudlog.debug("athena.log_handler: Not supported on PC") + time.sleep(1) return log_files = [] @@ -808,7 +809,7 @@ def ws_manage(ws: WebSocket, end_event: threading.Event) -> None: onroad_prev = None sock = ws.sock - while True: + while not end_event.wait(5): onroad = params.get_bool("IsOnroad") if onroad != onroad_prev: onroad_prev = onroad @@ -826,9 +827,6 @@ def ws_manage(ws: WebSocket, end_event: threading.Event) -> None: sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 7 if onroad else 10) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 2 if onroad else 3) - if end_event.wait(5): - break - def backoff(retries: int) -> int: return random.randrange(0, min(128, int(2 ** retries))) diff --git a/system/athena/sunnylinkd.py b/system/athena/sunnylinkd.py index 4150e8a475..67ae8a95af 100755 --- a/system/athena/sunnylinkd.py +++ b/system/athena/sunnylinkd.py @@ -1,17 +1,15 @@ #!/usr/bin/env python3 -#TODO: Add this to files_common to allow release to public from __future__ import annotations import base64 import gzip -import json import os import threading import time from openpilot.system.athena.athenad import ws_send, jsonrpc_handler, \ - recv_queue, RECONNECT_TIMEOUT_S, UploadQueueCache, upload_queue, cur_upload_items, backoff, ws_manage, log_handler + recv_queue, UploadQueueCache, upload_queue, cur_upload_items, backoff, ws_manage, log_handler from jsonrpc import dispatcher from websocket import (ABNF, WebSocket, WebSocketException, WebSocketTimeoutException, create_connection) @@ -26,6 +24,7 @@ SUNNYLINK_ATHENA_HOST = os.getenv('SUNNYLINK_ATHENA_HOST', 'wss://ws.stg.api.sun HANDLER_THREADS = int(os.getenv('HANDLER_THREADS', "4")) LOCAL_PORT_WHITELIST = {8022} SUNNYLINK_LOG_ATTR_NAME = "user.sunny.upload" +SUNNYLINK_RECONNECT_TIMEOUT_S = 70 # FYI changing this will also would require a change on sidebar.cc params = Params() sunnylink_api = SunnylinkApi(params.get("SunnylinkDongleId", encoding='utf-8')) @@ -58,24 +57,22 @@ def handle_long_poll(ws: WebSocket, exit_event: threading.Event | None) -> None: end_event.set() comma_prime_cellular_end_event.set() - prime_type = params.get("PrimeType", encoding='utf-8') + prime_type = params.get("PrimeType", encoding='utf-8') or 0 metered = sm['deviceState'].networkMetered - if int(prime_type) > 2 and metered: + if metered and int(prime_type) > 2: cloudlog.debug(f"sunnylinkd.handle_long_poll: PrimeType({prime_type}) > 2 and networkMetered({metered})") comma_prime_cellular_end_event.set() elif comma_prime_cellular_end_event.is_set(): cloudlog.debug(f"sunnylinkd.handle_long_poll: comma_prime_cellular_end_event is set and not PrimeType({prime_type}) > 2 or not networkMetered({metered})") comma_prime_cellular_end_event.clear() - - except (KeyboardInterrupt, SystemExit): + finally: end_event.set() comma_prime_cellular_end_event.set() - raise - finally: for thread in threads: cloudlog.debug(f"sunnylinkd athena.joining {thread.name}") thread.join() + cloudlog.debug(f"sunnylinkd athena.joined {thread.name}") def ws_recv(ws: WebSocket, end_event: threading.Event) -> None: @@ -89,12 +86,12 @@ def ws_recv(ws: WebSocket, end_event: threading.Event) -> None: recv_queue.put_nowait(data) cloudlog.debug(f"sunnylinkd.ws_recv.recv {data}") elif opcode in (ABNF.OPCODE_PING, ABNF.OPCODE_PONG): - cloudlog.debug(f"sunnylinkd.ws_recv.pong {opcode}") + cloudlog.debug(f"sunnylinkd.ws_recv.pong") last_ping = int(time.monotonic() * 1e9) Params().put("LastSunnylinkPingTime", str(last_ping)) except WebSocketTimeoutException: ns_since_last_ping = int(time.monotonic() * 1e9) - last_ping - if ns_since_last_ping > RECONNECT_TIMEOUT_S * 1e9: + if ns_since_last_ping > SUNNYLINK_RECONNECT_TIMEOUT_S * 1e9: cloudlog.exception("sunnylinkd.ws_recv.timeout") end_event.set() except Exception: @@ -103,15 +100,15 @@ def ws_recv(ws: WebSocket, end_event: threading.Event) -> None: def ws_ping(ws: WebSocket, end_event: threading.Event) -> None: - # last_ping = int(time.monotonic() * 1e9) - while not end_event.is_set(): + ws.ping() # Send the first ping + while not end_event.wait(SUNNYLINK_RECONNECT_TIMEOUT_S * 0.7): # Sleep about 70% before a timeout try: ws.ping() cloudlog.debug(f"sunnylinkd.ws_recv.ws_ping: Pinging") except Exception: cloudlog.exception("sunnylinkd.ws_ping.exception") end_event.set() - time.sleep(RECONNECT_TIMEOUT_S * 0.7) # Sleep about 70% before a timeout + cloudlog.debug(f"sunnylinkd.ws_ping.end_event is set, exiting ws_ping thread") def ws_queue(end_event: threading.Event) -> None: resume_requested = False @@ -129,7 +126,13 @@ def ws_queue(end_event: threading.Event) -> None: resume_requested = False tries += 1 time.sleep(backoff(tries)) # Wait for the backoff time before the next attempt - cloudlog.debug("Resume requested or end_event is set, exiting ws_queue thread") + + if end_event.is_set(): + cloudlog.debug("end_event is set, exiting ws_queue thread") + elif resume_requested: + cloudlog.debug(f"Resume requested to server after {tries} tries") + else: + cloudlog.error(f"Reached end of ws_queue while end_event is not set and resume_requested is {resume_requested}") def sunny_log_handler(end_event: threading.Event, comma_prime_cellular_end_event: threading.Event) -> None: @@ -195,7 +198,8 @@ def main(exit_event: threading.Event = None): cloudlog.event("sunnylinkd.main.connecting_ws", ws_uri=ws_uri, retries=conn_retries) ws = create_connection(ws_uri, cookie="jwt=" + sunnylink_api.get_token(), - enable_multithread=True) + enable_multithread=True, + timeout=SUNNYLINK_RECONNECT_TIMEOUT_S) cloudlog.event("sunnylinkd.main.connected_ws", ws_uri=ws_uri, retries=conn_retries, duration=time.monotonic() - conn_start) conn_start = None