| 182 | |
| 183 | @staticmethod |
| 184 | def send_worker(port, send_queue, worker_alive): |
| 185 | timing = AccumDict() |
| 186 | log = Logger('./var/log/send_worker.log', verbose=opt.verbose) |
| 187 | |
| 188 | ctx = SerializingContext() |
| 189 | socket = ctx.socket(zmq.PUSH) |
| 190 | socket.bind(f"tcp://*:{port}") |
| 191 | |
| 192 | log(f'Sending on port {port}', important=True) |
| 193 | |
| 194 | try: |
| 195 | while worker_alive.value: |
| 196 | tt = TicToc() |
| 197 | |
| 198 | try: |
| 199 | method, data = send_queue.get(timeout=GET_TIMEOUT) |
| 200 | except queue.Empty: |
| 201 | log("send queue empty") |
| 202 | continue |
| 203 | |
| 204 | # get the latest non-critical request from the queue |
| 205 | # don't skip critical request |
| 206 | while not send_queue.empty() and not method['critical']: |
| 207 | log(f"skip {method}") |
| 208 | method, data = send_queue.get() |
| 209 | |
| 210 | log("sending", method) |
| 211 | |
| 212 | tt.tic() |
| 213 | socket.send_data(method, data) |
| 214 | timing.add('SEND', tt.toc()) |
| 215 | |
| 216 | Once(timing, log, per=1) |
| 217 | except KeyboardInterrupt: |
| 218 | log("predictor_worker: user interrupt", important=True) |
| 219 | |
| 220 | worker_alive.value = 0 |
| 221 | log("send_worker exit", important=True) |
| 222 | |
| 223 | |
| 224 | def run_worker(in_port=None, out_port=None): |