(self)
| 299 | self.idx = idx |
| 300 | |
| 301 | def run(self): |
| 302 | enable_death_signal(_warn=self.idx == 0) |
| 303 | self.ds.reset_state() |
| 304 | itr = _repeat_iter(lambda: self.ds) |
| 305 | |
| 306 | context = zmq.Context() |
| 307 | socket = context.socket(zmq.PUSH) |
| 308 | socket.set_hwm(self.hwm) |
| 309 | socket.connect(self.conn_name) |
| 310 | try: |
| 311 | while True: |
| 312 | try: |
| 313 | dp = next(itr) |
| 314 | socket.send(dumps(dp), copy=False) |
| 315 | except Exception: |
| 316 | dp = _ExceptionWrapper(sys.exc_info()).pack() |
| 317 | socket.send(dumps(dp), copy=False) |
| 318 | raise |
| 319 | # sigint could still propagate here, e.g. when nested |
| 320 | except KeyboardInterrupt: |
| 321 | pass |
| 322 | finally: |
| 323 | socket.close(0) |
| 324 | context.destroy(0) |
| 325 | |
| 326 | def __init__(self, ds, num_proc=1, hwm=50): |
| 327 | """ |
nothing calls this directly
no test coverage detected