def start(self, poll_period=None):
logger.info("<STR_LIT>")<EOL>if poll_period is None:<EOL><INDENT>poll_period = self.poll_period<EOL><DEDENT>start = time.time()<EOL>count = <NUM_LIT:0><EOL>self._kill_event = threading.Event()<EOL>self._task_puller_thread = threading.Thread(target=self.migrate_tasks_to_internal,<EOL>args=(self._kill_event,))<EOL>self._task_puller_thread.start()<EOL>self._command_thread = threading.Thread(target=self._command_server,<EOL>args=(self._kill_event,))<EOL>self._command_thread.start()<EOL>poller = zmq.Poller()<EOL>poller.register(self.task_outgoing, zmq.POLLIN)<EOL>poller.register(self.results_incoming, zmq.POLLIN)<EOL>interesting_managers = set()<EOL>while not self._kill_event.is_set():<EOL><INDENT>self.socks = dict(poller.poll(timeout=poll_period))<EOL>if self.task_outgoing in self.socks and self.socks[self.task_outgoing] == zmq.POLLIN:<EOL><INDENT>logger.debug("<STR_LIT>")<EOL>message = self.task_outgoing.recv_multipart()<EOL>manager = message[<NUM_LIT:0>]<EOL>if manager not in self._ready_manager_queue:<EOL><INDENT>reg_flag = False<EOL>try:<EOL><INDENT>msg = json.loads(message[<NUM_LIT:1>].decode('<STR_LIT:utf-8>'))<EOL>reg_flag = True<EOL><DEDENT>except Exception:<EOL><INDENT>logger.warning("<STR_LIT>".format(<EOL>manager))<EOL>logger.debug("<STR_LIT>".format(message[<NUM_LIT:0>]))<EOL><DEDENT>self._ready_manager_queue[manager] = {'<STR_LIT>': time.time(),<EOL>'<STR_LIT>': <NUM_LIT:0>,<EOL>'<STR_LIT>': None,<EOL>'<STR_LIT>': <NUM_LIT:0>,<EOL>'<STR_LIT>': True,<EOL>'<STR_LIT>': []}<EOL>if reg_flag is True:<EOL><INDENT>interesting_managers.add(manager)<EOL>logger.info("<STR_LIT>".format(manager))<EOL>self._ready_manager_queue[manager].update(msg)<EOL>logger.info("<STR_LIT>".format(manager, msg))<EOL>if (msg['<STR_LIT>'].rsplit("<STR_LIT:.>", <NUM_LIT:1>)[<NUM_LIT:0>] != self.current_platform['<STR_LIT>'].rsplit("<STR_LIT:.>", <NUM_LIT:1>)[<NUM_LIT:0>] or<EOL>msg['<STR_LIT>'] != self.current_platform['<STR_LIT>']):<EOL><INDENT>logger.warn("<STR_LIT>".format(manager))<EOL>if self.suppress_failure is False:<EOL><INDENT>logger.debug("<STR_LIT>")<EOL>self._kill_event.set()<EOL>e = ManagerLost(manager)<EOL>result_package = {'<STR_LIT>': -<NUM_LIT:1>, '<STR_LIT>': serialize_object(e)}<EOL>pkl_package = pickle.dumps(result_package)<EOL>self.results_outgoing.send(pkl_package)<EOL>logger.warning("<STR_LIT>")<EOL><DEDENT>else:<EOL><INDENT>logger.debug("<STR_LIT>")<EOL><DEDENT><DEDENT>else:<EOL><INDENT>logger.info("<STR_LIT>".format(manager, msg['<STR_LIT>']))<EOL>logger.info("<STR_LIT>".format(manager,<EOL>msg['<STR_LIT>'].rsplit("<STR_LIT:.>", <NUM_LIT:1>)[<NUM_LIT:0>]))<EOL><DEDENT><DEDENT>else:<EOL><INDENT>if self.suppress_failure is False:<EOL><INDENT>self._kill_event.set()<EOL>e = BadRegistration(manager, critical=True)<EOL>result_package = {'<STR_LIT>': -<NUM_LIT:1>, '<STR_LIT>': serialize_object(e)}<EOL>pkl_package = pickle.dumps(result_package)<EOL>self.results_outgoing.send(pkl_package)<EOL><DEDENT>else:<EOL><INDENT>logger.debug("<STR_LIT>".format(<EOL>manager))<EOL><DEDENT><DEDENT><DEDENT>else:<EOL><INDENT>tasks_requested = int.from_bytes(message[<NUM_LIT:1>], "<STR_LIT>")<EOL>self._ready_manager_queue[manager]['<STR_LIT>'] = time.time()<EOL>if tasks_requested == HEARTBEAT_CODE:<EOL><INDENT>logger.debug("<STR_LIT>".format(manager))<EOL>self.task_outgoing.send_multipart([manager, b'<STR_LIT>', PKL_HEARTBEAT_CODE])<EOL><DEDENT>else:<EOL><INDENT>logger.debug("<STR_LIT>".format(manager, tasks_requested))<EOL>self._ready_manager_queue[manager]['<STR_LIT>'] = tasks_requested<EOL>interesting_managers.add(manager)<EOL><DEDENT><DEDENT>logger.debug("<STR_LIT>")<EOL><DEDENT>logger.debug("<STR_LIT>".format(len(self._ready_manager_queue),<EOL>len(interesting_managers)))<EOL>if interesting_managers and not self.pending_task_queue.empty():<EOL><INDENT>shuffled_managers = list(interesting_managers)<EOL>random.shuffle(shuffled_managers)<EOL>while shuffled_managers and not self.pending_task_queue.empty():  <EOL><INDENT>manager = shuffled_managers.pop()<EOL>tasks_inflight = len(self._ready_manager_queue[manager]['<STR_LIT>'])<EOL>real_capacity = min(self._ready_manager_queue[manager]['<STR_LIT>'],<EOL>self._ready_manager_queue[manager]['<STR_LIT>'] - tasks_inflight)<EOL>if (real_capacity and self._ready_manager_queue[manager]['<STR_LIT>']):<EOL><INDENT>tasks = self.get_tasks(real_capacity)<EOL>if tasks:<EOL><INDENT>self.task_outgoing.send_multipart([manager, b'<STR_LIT>', pickle.dumps(tasks)])<EOL>task_count = len(tasks)<EOL>count += task_count<EOL>tids = [t['<STR_LIT>'] for t in tasks]<EOL>self._ready_manager_queue[manager]['<STR_LIT>'] -= task_count<EOL>self._ready_manager_queue[manager]['<STR_LIT>'].extend(tids)<EOL>logger.debug("<STR_LIT>".format(tids, manager))<EOL>if self._ready_manager_queue[manager]['<STR_LIT>'] > <NUM_LIT:0>:<EOL><INDENT>logger.debug("<STR_LIT>".format(manager, self._ready_manager_queue[manager]['<STR_LIT>']))<EOL><DEDENT>else:<EOL><INDENT>logger.debug("<STR_LIT>".format(manager))<EOL>interesting_managers.remove(manager)<EOL><DEDENT><DEDENT><DEDENT>else:<EOL><INDENT>interesting_managers.remove(manager)<EOL><DEDENT><DEDENT>logger.debug("<STR_LIT>".format(len(interesting_managers)))<EOL><DEDENT>else:<EOL><INDENT>logger.debug("<STR_LIT>")<EOL><DEDENT>if self.results_incoming in self.socks and self.socks[self.results_incoming] == zmq.POLLIN:<EOL><INDENT>logger.debug("<STR_LIT>")<EOL>manager, *b_messages = self.results_incoming.recv_multipart()<EOL>if manager not in self._ready_manager_queue:<EOL><INDENT>logger.warning("<STR_LIT>".format(manager))<EOL><DEDENT>else:<EOL><INDENT>logger.debug("<STR_LIT>".format(len(b_messages)))<EOL>for b_message in b_messages:<EOL><INDENT>r = pickle.loads(b_message)<EOL>self._ready_manager_queue[manager]['<STR_LIT>'].remove(r['<STR_LIT>'])<EOL><DEDENT>self.results_outgoing.send_multipart(b_messages)<EOL>logger.debug("<STR_LIT>".format(self._ready_manager_queue[manager]['<STR_LIT>']))<EOL><DEDENT>logger.debug("<STR_LIT>")<EOL><DEDENT>logger.debug("<STR_LIT>")<EOL>bad_managers = [manager for manager in self._ready_manager_queue if<EOL>time.time() - self._ready_manager_queue[manager]['<STR_LIT>'] > self.heartbeat_threshold]<EOL>for manager in bad_managers:<EOL><INDENT>logger.debug("<STR_LIT>".format(self._ready_manager_queue[manager]['<STR_LIT>'], time.time()))<EOL>logger.warning("<STR_LIT>".format(manager))<EOL>for tid in self._ready_manager_queue[manager]['<STR_LIT>']:<EOL><INDENT>try:<EOL><INDENT>raise ManagerLost(manager)<EOL><DEDENT>except Exception:<EOL><INDENT>result_package = {'<STR_LIT>': tid, '<STR_LIT>': serialize_object(RemoteExceptionWrapper(*sys.exc_info()))}<EOL>pkl_package = pickle.dumps(result_package)<EOL>self.results_outgoing.send(pkl_package)<EOL>logger.warning("<STR_LIT>")<EOL><DEDENT><DEDENT>self._ready_manager_queue.pop(manager, '<STR_LIT:None>')<EOL><DEDENT>logger.debug("<STR_LIT>")<EOL>logger.debug("<STR_LIT>")<EOL><DEDENT>delta = time.time() - start<EOL>logger.info("<STR_LIT>".format(count, delta))<EOL>logger.warning("<STR_LIT>")<EOL>
Start the NeedNameQeueu

        Parameters:
        ----------

        TODO: Move task receiving to a thread