def worker_loop(self):
    '\n        Child process worker loop. Manages an underlying thread pool, pulls and\n        sends back results to the main pool.\n        '
    from testplan.runners.pools.communication import Message
    message = Message(**self.metadata)
    try:
        self._pre_loop_setup(message)
    except Exception:
        print('_pre_loop_setup failed')
        self._transport.send_and_receive(message.make(message.SetupFailed, data=traceback.format_exc()), expect=message.Ack)
        return
    with self._child_pool():
        message = Message(**self.metadata)
        next_possible_request = time.time()
        next_heartbeat = time.time()
        request_delay = self._pool_cfg.active_loop_sleep
        while True:
            now = time.time()
            if (self._pool_cfg.worker_heartbeat and (now > next_heartbeat)):
                hb_resp = self._transport.send_and_receive(message.make(message.Heartbeat, data=time.time()))
                if (hb_resp is None):
                    self.logger.critical('Pool seems dead, child exits1.')
                    self.exit_loop()
                    break
                else:
                    self.logger.debug('Pool heartbeat response: {} at {} before {}s.'.format(hb_resp.cmd, hb_resp.data, (time.time() - hb_resp.data)))
                next_heartbeat = (now + self._pool_cfg.worker_heartbeat)
            if self._pool.results:
                task_results = []
                for uid in list(self._pool.results.keys()):
                    task_results.append(self._pool.results[uid])
                    self.logger.debug('Sending back result for {}'.format(self._pool.results[uid].task))
                    del self._pool.results[uid]
                self._transport.send_and_receive(message.make(message.TaskResults, data=task_results), expect=message.Ack)
            demand = (self._pool.workers_requests() - self._pool.unassigned.qsize())
            if ((demand > 0) and (time.time() > next_possible_request)):
                received = self._transport.send_and_receive(message.make(message.TaskPullRequest, data=demand))
                if ((received is None) or (received.cmd == Message.Stop)):
                    self.logger.critical('Pool seems dead or stopping, child exits.')
                    self.exit_loop()
                    break
                elif (received.cmd == Message.TaskSending):
                    next_possible_request = time.time()
                    request_delay = 0
                    for task in received.data:
                        self.logger.debug('Added {} to local pool'.format(task))
                        self._pool.add(task, task.uid())
                    for worker in self._pool._workers:
                        worker.requesting = 0
                elif (received.cmd == Message.Ack):
                    request_delay = min(((request_delay + 0.2) * 1.5), self._pool_cfg.max_active_loop_sleep)
                    next_possible_request = (time.time() + request_delay)
                    pass
            time.sleep(self._pool_cfg.active_loop_sleep)
    self.logger.info('Local pool {} stopped.'.format(self._pool))
Child process worker loop. Manages an underlying thread pool, pulls and
sends back results to the main pool.
def worker_loop(self):
    '\n        Child process worker loop. Manages an underlying thread pool, pulls and\n        sends back results to the main pool.\n        '
    from testplan.runners.pools.communication import Message
    message = Message(**self.metadata)
    try:
        self._pre_loop_setup(message)
    except Exception:
        print('_pre_loop_setup failed')
        self._transport.send_and_receive(message.make(message.SetupFailed, data=traceback.format_exc()), expect=message.Ack)
        return
    with self._child_pool():
        message = Message(**self.metadata)
        next_possible_request = time.time()
        next_heartbeat = time.time()
        request_delay = self._pool_cfg.active_loop_sleep
        while True:
            now = time.time()
            if (self._pool_cfg.worker_heartbeat and (now > next_heartbeat)):
                hb_resp = self._transport.send_and_receive(message.make(message.Heartbeat, data=time.time()))
                if (hb_resp is None):
                    self.logger.critical('Pool seems dead, child exits1.')
                    self.exit_loop()
                    break
                else:
                    self.logger.debug('Pool heartbeat response: {} at {} before {}s.'.format(hb_resp.cmd, hb_resp.data, (time.time() - hb_resp.data)))
                next_heartbeat = (now + self._pool_cfg.worker_heartbeat)
            if self._pool.results:
                task_results = []
                for uid in list(self._pool.results.keys()):
                    task_results.append(self._pool.results[uid])
                    self.logger.debug('Sending back result for {}'.format(self._pool.results[uid].task))
                    del self._pool.results[uid]
                self._transport.send_and_receive(message.make(message.TaskResults, data=task_results), expect=message.Ack)
            demand = (self._pool.workers_requests() - self._pool.unassigned.qsize())
            if ((demand > 0) and (time.time() > next_possible_request)):
                received = self._transport.send_and_receive(message.make(message.TaskPullRequest, data=demand))
                if ((received is None) or (received.cmd == Message.Stop)):
                    self.logger.critical('Pool seems dead or stopping, child exits.')
                    self.exit_loop()
                    break
                elif (received.cmd == Message.TaskSending):
                    next_possible_request = time.time()
                    request_delay = 0
                    for task in received.data:
                        self.logger.debug('Added {} to local pool'.format(task))
                        self._pool.add(task, task.uid())
                    for worker in self._pool._workers:
                        worker.requesting = 0
                elif (received.cmd == Message.Ack):
                    request_delay = min(((request_delay + 0.2) * 1.5), self._pool_cfg.max_active_loop_sleep)
                    next_possible_request = (time.time() + request_delay)
                    pass
            time.sleep(self._pool_cfg.active_loop_sleep)
    self.logger.info('Local pool {} stopped.'.format(self._pool))