Args: ds (DataFlow): input DataFlow. num_proc (int): number of processes to use. hwm (int): the zmq "high-water mark" (queue size) for both sender and receiver.
(self, ds, num_proc=1, hwm=50)
| 324 | context.destroy(0) |
| 325 | |
| 326 | def __init__(self, ds, num_proc=1, hwm=50): |
| 327 | """ |
| 328 | Args: |
| 329 | ds (DataFlow): input DataFlow. |
| 330 | num_proc (int): number of processes to use. |
| 331 | hwm (int): the zmq "high-water mark" (queue size) for both sender and receiver. |
| 332 | """ |
| 333 | super(MultiProcessRunnerZMQ, self).__init__() |
| 334 | |
| 335 | self.ds = ds |
| 336 | self.num_proc = num_proc |
| 337 | self._hwm = hwm |
| 338 | |
| 339 | if num_proc > 1: |
| 340 | logger.info("[MultiProcessRunnerZMQ] Will fork a dataflow more than one times. " |
| 341 | "This assumes the datapoints are i.i.d.") |
| 342 | try: |
| 343 | self._size = ds.__len__() |
| 344 | except NotImplementedError: |
| 345 | self._size = -1 |
| 346 | |
| 347 | def _recv(self): |
| 348 | ret = loads(self.socket.recv(copy=False)) |