Broadcast messages at regular intervals.
(method, size, delay)
| 69 | |
| 70 | |
| 71 | async def broadcast_messages(method, size, delay): |
| 72 | """Broadcast messages at regular intervals.""" |
| 73 | if method == "pubsub": |
| 74 | global PUBSUB |
| 75 | PUBSUB = PubSub() |
| 76 | load_average = 0 |
| 77 | time_average = 0 |
| 78 | pc1, pt1 = time.perf_counter_ns(), time.process_time_ns() |
| 79 | await asyncio.sleep(delay) |
| 80 | while True: |
| 81 | print(f"clients = {len(CLIENTS)}") |
| 82 | pc0, pt0 = time.perf_counter_ns(), time.process_time_ns() |
| 83 | load_average = 0.9 * load_average + 0.1 * (pt0 - pt1) / (pc0 - pc1) |
| 84 | print( |
| 85 | f"load = {(pt0 - pt1) / (pc0 - pc1) * 100:.1f}% / " |
| 86 | f"average = {load_average * 100:.1f}%, " |
| 87 | f"late = {(pc0 - pc1 - delay * 1e9) / 1e6:.1f} ms" |
| 88 | ) |
| 89 | pc1, pt1 = pc0, pt0 |
| 90 | |
| 91 | assert size > 20 |
| 92 | message = str(time.time_ns()).encode() + b" " + os.urandom(size - 20) |
| 93 | |
| 94 | if method == "default": |
| 95 | broadcast(CLIENTS, message) |
| 96 | elif method == "naive": |
| 97 | # Since the loop can yield control, make a copy of CLIENTS |
| 98 | # to avoid: RuntimeError: Set changed size during iteration |
| 99 | for websocket in CLIENTS.copy(): |
| 100 | await send(websocket, message) |
| 101 | elif method == "task": |
| 102 | for websocket in CLIENTS: |
| 103 | asyncio.create_task(send(websocket, message)) |
| 104 | elif method == "wait": |
| 105 | if CLIENTS: # asyncio.wait doesn't accept an empty list |
| 106 | await asyncio.wait( |
| 107 | [ |
| 108 | asyncio.create_task(send(websocket, message)) |
| 109 | for websocket in CLIENTS |
| 110 | ] |
| 111 | ) |
| 112 | elif method == "queue": |
| 113 | for queue in CLIENTS: |
| 114 | queue.put_nowait(message) |
| 115 | elif method == "pubsub": |
| 116 | PUBSUB.publish(message) |
| 117 | else: |
| 118 | raise NotImplementedError(f"unsupported method: {method}") |
| 119 | |
| 120 | pc2 = time.perf_counter_ns() |
| 121 | wait = delay + (pc1 - pc2) / 1e9 |
| 122 | time_average = 0.9 * time_average + 0.1 * (pc2 - pc1) |
| 123 | print( |
| 124 | f"broadcast = {(pc2 - pc1) / 1e6:.1f}ms / " |
| 125 | f"average = {time_average / 1e6:.1f}ms, " |
| 126 | f"wait = {wait * 1e3:.1f}ms" |
| 127 | ) |
| 128 | await asyncio.sleep(wait) |