| 102 | write_stats = None |
| 103 | |
| 104 | def on_start(self): |
| 105 | forking_enable(self.forking_enable) |
| 106 | Pool = (self.BlockingPool if self.options.get('threads', True) |
| 107 | else self.Pool) |
| 108 | proc_alive_timeout = ( |
| 109 | self.app.conf.worker_proc_alive_timeout if self.app |
| 110 | else None |
| 111 | ) |
| 112 | P = self._pool = Pool(processes=self.limit, |
| 113 | initializer=process_initializer, |
| 114 | on_process_exit=process_destructor, |
| 115 | enable_timeouts=True, |
| 116 | synack=False, |
| 117 | proc_alive_timeout=proc_alive_timeout, |
| 118 | **self.options) |
| 119 | |
| 120 | # Create proxy methods |
| 121 | self.on_apply = P.apply_async |
| 122 | self.maintain_pool = P.maintain_pool |
| 123 | self.terminate_job = P.terminate_job |
| 124 | self.grow = P.grow |
| 125 | self.shrink = P.shrink |
| 126 | self.flush = getattr(P, 'flush', None) # FIXME add to billiard |
| 127 | |
| 128 | def restart(self): |
| 129 | self._pool.restart() |