yandextank/plugins/Bfg/worker.py
def _feed(self):
        """
        A feeder that runs in distinct thread in main process.
        """
        self.plan = StpdReader(self.stpd_filename)
        if self.cached_stpd:
            self.plan = list(self.plan)
        for task in self.plan:
            if self.quit.is_set():
                logger.info("Stop feeding: gonna quit")
                return
            # try putting a task to a queue unless there is a quit flag
            # or all workers have exited
            while True:
                try:
                    self.task_queue.put(task, timeout=1)
                    break
                except Full:
                    if self.quit.is_set() or self.workers_finished:
                        return
                    else:
                        continue
        workers_count = self.instances
        logger.info(
            "Feeded all data. Publishing %d killer tasks" % (workers_count))
        retry_delay = 1
        for _ in range(5):
            try:
                [
                    self.task_queue.put(None, timeout=1)
                    for _ in xrange(0, workers_count)
                ]
                break
            except Full:
                logger.debug(
                    "Couldn't post killer tasks"
                    " because queue is full. Retrying in %ss", retry_delay)
                time.sleep(retry_delay)
                retry_delay *= 2

        try:
            logger.info("Waiting for workers")
            map(lambda x: x.join(), self.pool)
            logger.info("All workers exited.")
            self.workers_finished = True
        except (KeyboardInterrupt, SystemExit):
            self.task_queue.close()
            self.results.close()
            self.quit.set()
            logger.info("Going to quit. Waiting for workers")
            map(lambda x: x.join(), self.pool)
            self.workers_finished = True
def _feed(self):
        self.plan = StpdReader(self.stpd_filename)
        if self.cached_stpd:
            self.plan = list(self.plan)
        for task in self.plan:
            if self.quit.is_set():
                logger.info("Stop feeding: gonna quit")
                return
            # try putting a task to a queue unless there is a quit flag
            # or all workers have exited
            while True:
                try:
                    self.task_queue.put(task, timeout=1)
                    break
                except Full:
                    if self.quit.is_set() or self.workers_finished:
                        return
                    else:
                        continue
        workers_count = self.instances
        logger.info(
            "Feeded all data. Publishing %d killer tasks" % (workers_count))
        retry_delay = 1
        for _ in range(5):
            try:
                [
                    self.task_queue.put(None, timeout=1)
                    for _ in xrange(0, workers_count)
                ]
                break
            except Full:
                logger.debug(
                    "Couldn't post killer tasks"
                    " because queue is full. Retrying in %ss", retry_delay)
                time.sleep(retry_delay)
                retry_delay *= 2

        try:
            logger.info("Waiting for workers")
            map(lambda x: x.join(), self.pool)
            logger.info("All workers exited.")
            self.workers_finished = True
        except (KeyboardInterrupt, SystemExit):
            self.task_queue.close()
            self.results.close()
            self.quit.set()
            logger.info("Going to quit. Waiting for workers")
            map(lambda x: x.join(), self.pool)
            self.workers_finished = True
def _feed ( self ) : self . plan = StpdReader ( self . stpd_filename ) if self . cached_stpd : self . plan = list ( self . plan ) for task in self . plan : if self . quit . is_set ( ) : logger . info ( "Stop feeding: gonna quit" ) return # try putting a task to a queue unless there is a quit flag # or all workers have exited while True : try : self . task_queue . put ( task , timeout = 1 ) break except Full : if self . quit . is_set ( ) or self . workers_finished : return else : continue workers_count = self . instances logger . info ( "Feeded all data. Publishing %d killer tasks" % ( workers_count ) ) retry_delay = 1 for _ in range ( 5 ) : try : [ self . task_queue . put ( None , timeout = 1 ) for _ in xrange ( 0 , workers_count ) ] break except Full : logger . debug ( "Couldn't post killer tasks" " because queue is full. Retrying in %ss" , retry_delay ) time . sleep ( retry_delay ) retry_delay *= 2 try : logger . info ( "Waiting for workers" ) map ( lambda x : x . join ( ) , self . pool ) logger . info ( "All workers exited." ) self . workers_finished = True except ( KeyboardInterrupt , SystemExit ) : self . task_queue . close ( ) self . results . close ( ) self . quit . set ( ) logger . info ( "Going to quit. Waiting for workers" ) map ( lambda x : x . join ( ) , self . pool ) self . workers_finished = True
A feeder that runs in distinct thread in main process.
A feeder that runs in distinct thread in main process .