To implement a resilient PyTorch Lightning–style callback that logs per-batch metrics to a remote HTTP endpoint with exponential backoff and local buffering, use a background sender thread that reads a thread-safe queue, posts data with retries/backoff, stores failed payloads locally (in-memory queue plus optional disk file), and exposes flush/close for graceful shutdown.1) Brief approach- on_batch_end enqueues metric payloads (non-blocking).- A sender thread consumes the queue, attempts HTTP POST with exponential backoff on errors.- On repeated failures, payloads are buffered locally (queue remains) and optionally appended to a local file for persistence.- flush() blocks until queue and retry buffer empty; close() triggers flush and stops thread.2) Implementation (outline, simplified)python
import threading
import time
import json
import requests
from queue import Queue, Empty
class RemoteMonitorCallback:
def __init__(self, endpoint, max_retry=5, backoff_base=1.0, max_backoff=60.0, buffer_size=10000, persist_file=None):
self.endpoint = endpoint
self.max_retry = max_retry
self.backoff_base = backoff_base
self.max_backoff = max_backoff
self.queue = Queue(maxsize=buffer_size)
self.persist_file = persist_file
self._stop = threading.Event()
self._sender = threading.Thread(target=self._sender_loop, daemon=True)
self._sender.start()
# Called by training loop
def on_batch_end(self, batch_idx, logs):
payload = {"batch": batch_idx, "metrics": logs, "ts": time.time()}
try:
self.queue.put_nowait(payload)
except Exception:
# Queue full: drop or persist to disk
if self.persist_file:
with open(self.persist_file, "a") as f:
f.write(json.dumps(payload) + "\n")
def _post_with_backoff(self, payload):
attempt = 0
while attempt < self.max_retry and not self._stop.is_set():
try:
resp = requests.post(self.endpoint, json=payload, timeout=5)
if 200 <= resp.status_code < 300:
return True
# treat 4xx as non-retriable except 429
if 400 <= resp.status_code < 500 and resp.status_code != 429:
return False
except requests.RequestException:
pass
# exponential backoff with jitter
backoff = min(self.max_backoff, self.backoff_base * (2 ** attempt))
jitter = backoff * 0.1
sleep = backoff + (jitter * (0.5 - time.time() % 1))
time.sleep(sleep)
attempt += 1
return False
def _sender_loop(self):
while not self._stop.is_set() or not self.queue.empty():
try:
payload = self.queue.get(timeout=1)
except Empty:
# try to reload persisted buffer if available
if self.persist_file:
try:
with open(self.persist_file, "r") as f:
lines = f.readlines()
if lines:
# reload and truncate file
with open(self.persist_file, "w") as f:
f.truncate(0)
for line in lines:
try:
p = json.loads(line)
self.queue.put_nowait(p)
except Exception:
pass
except FileNotFoundError:
pass
continue
success = self._post_with_backoff(payload)
if not success:
# on permanent failure re-persist to disk if configured, else requeue with backoff delay
if self.persist_file:
with open(self.persist_file, "a") as f:
f.write(json.dumps(payload) + "\n")
else:
# sleep a bit before requeuing to avoid tight loop
time.sleep(self.backoff_base)
try:
self.queue.put_nowait(payload)
except Exception:
pass
self.queue.task_done()
# flush and stop
def flush(self, timeout=None):
start = time.time()
while not self.queue.empty():
if timeout and (time.time() - start) > timeout:
break
time.sleep(0.1)
# attempt to reload persisted file into queue and wait again
if self.persist_file:
try:
with open(self.persist_file, "r") as f:
lines = f.readlines()
if lines:
for line in lines:
try:
self.queue.put_nowait(json.loads(line))
except Exception:
pass
with open(self.persist_file, "w") as f:
f.truncate(0)
except FileNotFoundError:
pass
# wait for sender to empty
self.queue.join()
def close(self, timeout=30):
# signal stop, flush and join thread
self._stop.set()
self.flush(timeout=timeout)
self._sender.join(timeout=timeout)
