| … | … | @@ -27,22 +27,37 @@ SEND_TIMEOUT = 1.0 |
| 27 | 27 | BACKOFF_START = 1.0 |
| 28 | 28 | BACKOFF_MAX = 15.0 |
| 29 | 29 | |
| 30 | # The firmware drops a client that has sent nothing for CLIENT_IDLE_TIMEOUT_MS |
| 31 | # (300 s, esp324xlight.ino:95). Keyframe fade mode deliberately stays silent |
| 32 | # between keyframes, so a static look WILL cross that threshold and be dropped. |
| 33 | # Poking the connection well inside the window prevents it. |
| 34 | KEEPALIVE_INTERVAL = 60.0 |
| 35 | |
| 36 | # "hostname" with no argument only prints the current hostname (cmdHostname's |
| 37 | # else branch) -- it is the one command that produces traffic without touching |
| 38 | # pin state, so it cannot disturb a pwm2 fade in flight. A pwm/pwm2 keepalive |
| 39 | # would restart or cancel the fade. |
| 40 | KEEPALIVE_PAYLOAD = "hostname\n" |
| 41 | |
| 30 | 42 | |
| 31 | 43 | class _Target: |
| 32 | | __slots__ = ("host", "port", "sock", "pending", "last_sent", "last_send_time", |
| 33 | | "next_retry", "backoff", "status", "error") |
| 44 | __slots__ = ("host", "port", "sock", "desired", "dirty", "force", |
| 45 | "last_sent", "last_tx_time", "next_retry", "backoff", |
| 46 | "status", "error") |
| 34 | 47 | |
| 35 | 48 | def __init__(self, host, port): |
| 36 | 49 | self.host = host |
| 37 | 50 | self.port = port |
| 38 | 51 | self.sock = None |
| 39 | | self.pending = None |
| 40 | | self.last_sent = None |
| 41 | | self.last_send_time = 0.0 |
| 52 | self.desired = None # latest state the fixture SHOULD hold |
| 53 | self.dirty = False # desired differs from what reached the wire |
| 54 | self.force = False # bypass dedup and rate limit once |
| 55 | self.last_sent = None # what actually reached the wire |
| 56 | self.last_tx_time = 0.0 |
| 42 | 57 | self.next_retry = 0.0 |
| 43 | 58 | self.backoff = BACKOFF_START |
| 44 | 59 | self.status = "idle" |
| 45 | | self.error = "" |
| 60 | self.error = '' |
| 46 | 61 | |
| 47 | 62 | |
| 48 | 63 | class Sender: |
| … | … | @@ -76,7 +91,7 @@ class Sender: |
| 76 | 91 | self._targets.clear() |
| 77 | 92 | |
| 78 | 93 | def send(self, host, port, payload, force=False): |
| 79 | | """Queue the latest payload for a host. Never blocks.""" |
| 94 | """Record the latest desired state for a host. Never blocks.""" |
| 80 | 95 | if not host or not payload: |
| 81 | 96 | return |
| 82 | 97 | key = (host, port) |
| … | … | @@ -85,7 +100,12 @@ class Sender: |
| 85 | 100 | if tgt is None: |
| 86 | 101 | tgt = _Target(host, port) |
| 87 | 102 | self._targets[key] = tgt |
| 88 | | tgt.pending = (payload, force) |
| 103 | tgt.desired = payload |
| 104 | if force: |
| 105 | tgt.force = True |
| 106 | tgt.dirty = True |
| 107 | elif payload != tgt.last_sent: |
| 108 | tgt.dirty = True |
| 89 | 109 | self._wake.set() |
| 90 | 110 | |
| 91 | 111 | def forget(self, host, port): |
| … | … | @@ -118,41 +138,93 @@ class Sender: |
| 118 | 138 | tgt.error = repr(exc) |
| 119 | 139 | |
| 120 | 140 | def _service(self, tgt, now): |
| 121 | | with self._lock: |
| 122 | | item = tgt.pending |
| 123 | | tgt.pending = None |
| 124 | | if item is None: |
| 125 | | return |
| 126 | | payload, force = item |
| 127 | | |
| 128 | | if not force: |
| 129 | | if payload == tgt.last_sent: |
| 130 | | return |
| 131 | | if now - tgt.last_send_time < self.min_interval: |
| 132 | | # Put it back; we'll pick it up on a later tick. |
| 133 | | with self._lock: |
| 134 | | if tgt.pending is None: |
| 135 | | tgt.pending = item |
| 136 | | return |
| 141 | if tgt.desired is None: |
| 142 | return # nothing has ever been asked of this fixture |
| 137 | 143 | |
| 144 | # (1) Reconnect on our own schedule. Previously a reconnect was only |
| 145 | # attempted when a new payload happened to arrive, so a fixture dropped |
| 146 | # while idle stayed dark until the next scene change. |
| 138 | 147 | if tgt.sock is None: |
| 139 | 148 | if now < tgt.next_retry: |
| 140 | 149 | return |
| 141 | 150 | if not self._connect(tgt): |
| 142 | 151 | return |
| 152 | # The fixture may have rebooted or been dropped mid-look; it cannot |
| 153 | # be assumed to still hold last_sent. Re-assert the desired state. |
| 154 | tgt.last_sent = None |
| 155 | tgt.dirty = True |
| 156 | |
| 157 | # (2) Read whatever the firmware sent back. This serves two purposes: |
| 158 | # it stops command acks accumulating until a full receive buffer stalls |
| 159 | # writes, and -- more importantly -- recv() returning b"" is the peer |
| 160 | # having closed. That detects a drop with no traffic of our own, on the |
| 161 | # next 0.25 s tick, instead of waiting for the keepalive to fail. |
| 162 | if not self._drain(tgt): |
| 163 | tgt.status = "lost" |
| 164 | tgt.error = "peer closed" |
| 165 | self._close(tgt) |
| 166 | tgt.dirty = True # re-assert state once reconnected |
| 167 | tgt.next_retry = now + tgt.backoff |
| 168 | tgt.backoff = min(tgt.backoff * 2.0, BACKOFF_MAX) |
| 169 | return |
| 143 | 170 | |
| 171 | # (3) Push state. |
| 172 | if tgt.dirty: |
| 173 | if not tgt.force and now - tgt.last_tx_time < self.min_interval: |
| 174 | return # rate limited; desired is retained, retry next tick |
| 175 | payload = tgt.desired |
| 176 | if self._write(tgt, payload, now): |
| 177 | tgt.last_sent = payload |
| 178 | tgt.dirty = False |
| 179 | tgt.force = False |
| 180 | return |
| 181 | |
| 182 | # (4) Keepalive, so the firmware's idle timer never fires on a look we |
| 183 | # are deliberately not resending. |
| 184 | if now - tgt.last_tx_time >= KEEPALIVE_INTERVAL: |
| 185 | self._write(tgt, KEEPALIVE_PAYLOAD, now) # never becomes last_sent |
| 186 | |
| 187 | def _write(self, tgt, payload, now): |
| 188 | """Write bytes; on failure keep the state queued for the retry.""" |
| 144 | 189 | try: |
| 145 | 190 | tgt.sock.sendall(payload.encode("utf-8")) |
| 146 | | tgt.last_sent = payload |
| 147 | | tgt.last_send_time = now |
| 191 | tgt.last_tx_time = now |
| 148 | 192 | tgt.status = "connected" |
| 149 | 193 | tgt.error = "" |
| 194 | tgt.backoff = BACKOFF_START |
| 195 | return True |
| 150 | 196 | except (socket.timeout, OSError) as exc: |
| 151 | 197 | tgt.error = str(exc) |
| 152 | 198 | tgt.status = "lost" |
| 153 | 199 | self._close(tgt) |
| 200 | # Do NOT drop the payload -- the previous version discarded it here, |
| 201 | # so the first change after a silent drop was swallowed entirely. |
| 202 | tgt.dirty = True |
| 154 | 203 | tgt.next_retry = now + tgt.backoff |
| 155 | 204 | tgt.backoff = min(tgt.backoff * 2.0, BACKOFF_MAX) |
| 205 | return False |
| 206 | |
| 207 | @staticmethod |
| 208 | def _drain(tgt): |
| 209 | """Consume pending acks. False means the peer has closed the socket.""" |
| 210 | alive = True |
| 211 | try: |
| 212 | tgt.sock.setblocking(False) |
| 213 | while True: |
| 214 | if not tgt.sock.recv(4096): |
| 215 | alive = False # orderly close from the firmware |
| 216 | break |
| 217 | except (BlockingIOError, socket.timeout): |
| 218 | pass # nothing more to read; still connected |
| 219 | except OSError: |
| 220 | alive = False |
| 221 | finally: |
| 222 | try: |
| 223 | tgt.sock.setblocking(True) |
| 224 | tgt.sock.settimeout(SEND_TIMEOUT) |
| 225 | except OSError: |
| 226 | pass |
| 227 | return alive |
| 156 | 228 | |
| 157 | 229 | def _connect(self, tgt): |
| 158 | 230 | tgt.status = "connecting" |
| … | … | @@ -195,6 +267,8 @@ class Sender: |
| 195 | 267 | except OSError: |
| 196 | 268 | pass |
| 197 | 269 | tgt.sock = None |
| 270 | # Forget what the wire holds, but keep `desired` -- that is what gets |
| 271 | # re-asserted once the fixture comes back. |
| 198 | 272 | tgt.last_sent = None |
| 199 | 273 | |
| 200 | 274 | |