def pull_tasks(self, kill_event):
        """ Pulls tasks from the incoming tasks 0mq pipe onto the internal
        pending task queue

        Parameters:
        -----------
        kill_event : threading.Event
              Event to let the thread know when it is time to die.
        """
        logger.info("[TASK PULL THREAD] starting")
        poller = zmq.Poller()
        poller.register(self.task_incoming, zmq.POLLIN)

        # Send a registration message
        msg = self.create_reg_message()
        logger.debug("Sending registration message: {}".format(msg))
        self.task_incoming.send(msg)
        last_beat = time.time()
        last_interchange_contact = time.time()
        task_recv_counter = 0

        poll_timer = 1

        while not kill_event.is_set():
            time.sleep(LOOP_SLOWDOWN)
            ready_worker_count = self.ready_worker_queue.qsize()
            pending_task_count = self.pending_task_queue.qsize()

            logger.debug("[TASK_PULL_THREAD] ready workers:{}, pending tasks:{}".format(ready_worker_count,
                                                                                        pending_task_count))

            if time.time() > last_beat + self.heartbeat_period:
                self.heartbeat()
                last_beat = time.time()

            if pending_task_count < self.max_queue_size and ready_worker_count > 0:
                logger.debug("[TASK_PULL_THREAD] Requesting tasks: {}".format(ready_worker_count))
                msg = ((ready_worker_count).to_bytes(4, "little"))
                self.task_incoming.send(msg)

            socks = dict(poller.poll(timeout=poll_timer))

            if self.task_incoming in socks and socks[self.task_incoming] == zmq.POLLIN:
                _, pkl_msg = self.task_incoming.recv_multipart()
                tasks = pickle.loads(pkl_msg)
                last_interchange_contact = time.time()

                if tasks == 'STOP':
                    logger.critical("[TASK_PULL_THREAD] Received stop request")
                    kill_event.set()
                    break

                elif tasks == HEARTBEAT_CODE:
                    logger.debug("Got heartbeat from interchange")

                else:
                    # Reset timer on receiving message
                    poll_timer = 1
                    task_recv_counter += len(tasks)
                    logger.debug("[TASK_PULL_THREAD] Got tasks: {} of {}".format([t['task_id'] for t in tasks],
                                                                                 task_recv_counter))
                    for task in tasks:
                        self.pending_task_queue.put(task)
            else:
                logger.debug("[TASK_PULL_THREAD] No incoming tasks")
                # Limit poll duration to heartbeat_period
                # heartbeat_period is in s vs poll_timer in ms
                poll_timer = min(self.heartbeat_period * 1000, poll_timer * 2)

                # Only check if no messages were received.
                if time.time() > last_interchange_contact + self.heartbeat_threshold:
                    logger.critical("[TASK_PULL_THREAD] Missing contact with interchange beyond heartbeat_threshold")
                    kill_event.set()
                    logger.critical("[TASK_PULL_THREAD] Exiting")
                    break
def pull_tasks(self, kill_event):
        
        logger.info("[TASK PULL THREAD] starting")
        poller = zmq.Poller()
        poller.register(self.task_incoming, zmq.POLLIN)

        
        msg = self.create_reg_message()
        logger.debug("Sending registration message: {}".format(msg))
        self.task_incoming.send(msg)
        last_beat = time.time()
        last_interchange_contact = time.time()
        task_recv_counter = 0

        poll_timer = 1

        while not kill_event.is_set():
            time.sleep(LOOP_SLOWDOWN)
            ready_worker_count = self.ready_worker_queue.qsize()
            pending_task_count = self.pending_task_queue.qsize()

            logger.debug("[TASK_PULL_THREAD] ready workers:{}, pending tasks:{}".format(ready_worker_count,
                                                                                        pending_task_count))

            if time.time() > last_beat + self.heartbeat_period:
                self.heartbeat()
                last_beat = time.time()

            if pending_task_count < self.max_queue_size and ready_worker_count > 0:
                logger.debug("[TASK_PULL_THREAD] Requesting tasks: {}".format(ready_worker_count))
                msg = ((ready_worker_count).to_bytes(4, "little"))
                self.task_incoming.send(msg)

            socks = dict(poller.poll(timeout=poll_timer))

            if self.task_incoming in socks and socks[self.task_incoming] == zmq.POLLIN:
                _, pkl_msg = self.task_incoming.recv_multipart()
                tasks = pickle.loads(pkl_msg)
                last_interchange_contact = time.time()

                if tasks == 'STOP':
                    logger.critical("[TASK_PULL_THREAD] Received stop request")
                    kill_event.set()
                    break

                elif tasks == HEARTBEAT_CODE:
                    logger.debug("Got heartbeat from interchange")

                else:
                    
                    poll_timer = 1
                    task_recv_counter += len(tasks)
                    logger.debug("[TASK_PULL_THREAD] Got tasks: {} of {}".format([t['task_id'] for t in tasks],
                                                                                 task_recv_counter))
                    for task in tasks:
                        self.pending_task_queue.put(task)
            else:
                logger.debug("[TASK_PULL_THREAD] No incoming tasks")
                
                
                poll_timer = min(self.heartbeat_period * 1000, poll_timer * 2)

                
                if time.time() > last_interchange_contact + self.heartbeat_threshold:
                    logger.critical("[TASK_PULL_THREAD] Missing contact with interchange beyond heartbeat_threshold")
                    kill_event.set()
                    logger.critical("[TASK_PULL_THREAD] Exiting")
                    break
def pull_tasks ( self , kill_event ) : logger . info ( "[TASK PULL THREAD] starting" ) poller = zmq . Poller ( ) poller . register ( self . task_incoming , zmq . POLLIN ) # Send a registration message msg = self . create_reg_message ( ) logger . debug ( "Sending registration message: {}" . format ( msg ) ) self . task_incoming . send ( msg ) last_beat = time . time ( ) last_interchange_contact = time . time ( ) task_recv_counter = 0 poll_timer = 1 while not kill_event . is_set ( ) : time . sleep ( LOOP_SLOWDOWN ) ready_worker_count = self . ready_worker_queue . qsize ( ) pending_task_count = self . pending_task_queue . qsize ( ) logger . debug ( "[TASK_PULL_THREAD] ready workers:{}, pending tasks:{}" . format ( ready_worker_count , pending_task_count ) ) if time . time ( ) > last_beat + self . heartbeat_period : self . heartbeat ( ) last_beat = time . time ( ) if pending_task_count < self . max_queue_size and ready_worker_count > 0 : logger . debug ( "[TASK_PULL_THREAD] Requesting tasks: {}" . format ( ready_worker_count ) ) msg = ( ( ready_worker_count ) . to_bytes ( 4 , "little" ) ) self . task_incoming . send ( msg ) socks = dict ( poller . poll ( timeout = poll_timer ) ) if self . task_incoming in socks and socks [ self . task_incoming ] == zmq . POLLIN : _ , pkl_msg = self . task_incoming . recv_multipart ( ) tasks = pickle . loads ( pkl_msg ) last_interchange_contact = time . time ( ) if tasks == 'STOP' : logger . critical ( "[TASK_PULL_THREAD] Received stop request" ) kill_event . set ( ) break elif tasks == HEARTBEAT_CODE : logger . debug ( "Got heartbeat from interchange" ) else : # Reset timer on receiving message poll_timer = 1 task_recv_counter += len ( tasks ) logger . debug ( "[TASK_PULL_THREAD] Got tasks: {} of {}" . format ( [ t [ 'task_id' ] for t in tasks ] , task_recv_counter ) ) for task in tasks : self . pending_task_queue . put ( task ) else : logger . debug ( "[TASK_PULL_THREAD] No incoming tasks" ) # Limit poll duration to heartbeat_period # heartbeat_period is in s vs poll_timer in ms poll_timer = min ( self . heartbeat_period * 1000 , poll_timer * 2 ) # Only check if no messages were received. if time . time ( ) > last_interchange_contact + self . heartbeat_threshold : logger . critical ( "[TASK_PULL_THREAD] Missing contact with interchange beyond heartbeat_threshold" ) kill_event . set ( ) logger . critical ( "[TASK_PULL_THREAD] Exiting" ) break
Pulls tasks from the incoming tasks 0mq pipe onto the internal
        pending task queue

        Parameters:
        -----------
        kill_event : threading.Event
              Event to let the thread know when it is time to die.
Pulls tasks from the incoming tasks 0mq pipe onto the internal pending task queue