polymer/Polymer.py
def supervise(self):
        """If not in a hot_loop, call supervise() to start the tasks"""
        self.retval = set([])
        stats = TaskMgrStats(
            worker_count=self.worker_count,
            log_interval=self.log_interval,
            hot_loop=self.hot_loop,
        )

        hot_loop = self.hot_loop
        if self.log_level >= 2:
            logmsg = "TaskMgr.supervise() started {0} workers".format(self.worker_count)
            self.log.info(logmsg)
        self.workers = self.spawn_workers()

        ## Add work
        self.num_tasks = 0
        if not hot_loop:
            if self.log_level >= 2:
                logmsg = "TaskMgr.supervise() received {0} tasks".format(
                    len(self.work_todo)
                )
                self.log.info(logmsg)
            for task in self.work_todo:
                self.num_tasks += 1
                if self.log_level >= 2:
                    logmsg = "TaskMgr.supervise() queued task: {0}".format(task)
                    self.log.info(logmsg)
                self.queue_task(task)

        finished = False
        while not finished:
            try:
                if hot_loop:
                    # Calculate the adaptive loop delay
                    delay = self.calc_wait_time(stats.exec_times)
                    self.queue_tasks_from_controller(delay=delay)  # queue tasks
                    time.sleep(delay)

                r_msg = self.r_q.get_nowait()  # __ACK__ or __FINISHED__
                task = r_msg.get("task")
                w_id = r_msg.get("w_id")
                state = r_msg.get("state", "")
                if state == "__ACK__":
                    self.worker_assignments[w_id] = task
                    self.work_todo.remove(task)
                    if self.log_level >= 3:
                        self.log.debug("r_msg: {0}".format(r_msg))
                    if self.log_level >= 3:
                        self.log.debug("w_id={0} received task={1}".format(w_id, task))
                elif state == "__FINISHED__":
                    now = time.time()
                    task_exec_time = task.task_stop - task.task_start
                    task_queue_time = now - task.queue_time - task_exec_time
                    stats.exec_times.append(task_exec_time)
                    stats.queue_times.append(task_queue_time)

                    if self.log_level >= 1:
                        self.log.debug(
                            "TaskMgr.work_todo: {0} tasks left".format(
                                len(self.work_todo)
                            )
                        )
                    if self.log_level >= 3:
                        self.log.debug("TaskMgr.work_todo: {0}".format(self.work_todo))
                        self.log.debug("r_msg: {0}".format(r_msg))

                    if not hot_loop:
                        self.retval.add(task)  # Add result to retval
                        self.worker_assignments.pop(w_id)  # Delete the key
                        finished = self.is_finished()
                    else:
                        self.controller.to_q.put(task)  # Send to the controller
                        self.worker_assignments.pop(w_id)  # Delete the key

                elif state == "__ERROR__":

                    now = time.time()
                    task_exec_time = task.task_stop - task.task_start
                    task_queue_time = now - task.queue_time - task_exec_time
                    stats.exec_times.append(task_exec_time)
                    stats.queue_times.append(task_queue_time)

                    if self.log_level >= 1:
                        self.log.error("r_msg: {0}".format(r_msg))
                        self.log.error("".join(r_msg.get("error")))
                        self.log.debug(
                            "TaskMgr.work_todo: {0} tasks left".format(
                                len(self.work_todo)
                            )
                        )
                    if self.log_level >= 3:
                        self.log.debug("TaskMgr.work_todo: {0}".format(self.work_todo))

                    if not hot_loop:
                        if not self.resubmit_on_error:
                            # If task is in work_todo, delete it
                            for tt in self.work_todo:
                                if tt == task:
                                    self.work_todo.remove(task)  # Remove task...

                            try:
                                # Delete the worker assignment...
                                self.worker_assignments.pop(w_id)
                            except:
                                pass
                            self.retval.add(task)  # Add result to retval

                    self.respawn_dead_workers()

            except Empty:
                state = "__EMPTY__"
            except Exception as e:
                tb_str = "".join(tb.format_exception(*(sys.exc_info())))
                raise e(tb_str)

            if stats.log_time:
                if self.log_level >= 0:
                    self.log.info(stats.log_message)

            # Adaptive loop delay
            delay = self.calc_wait_time(stats.exec_times)
            time.sleep(delay)

            self.respawn_dead_workers()
            finished = self.is_finished()

        if not hot_loop:
            self.kill_workers()
            for w_id, p in self.workers.items():
                p.join()

            ## Log a final stats summary...
            if self.log_level > 0:
                self.log.info(stats.log_message)
            return self.retval
def supervise(self):
        """If not in a hot_loop, call supervise() to start the tasks"""
        self.retval = set([])
        stats = TaskMgrStats(
            worker_count=self.worker_count,
            log_interval=self.log_interval,
            hot_loop=self.hot_loop,
        )

        hot_loop = self.hot_loop
        if self.log_level >= 2:
            logmsg = "TaskMgr.supervise() started {0} workers".format(self.worker_count)
            self.log.info(logmsg)
        self.workers = self.spawn_workers()

        ## Add work
        self.num_tasks = 0
        if not hot_loop:
            if self.log_level >= 2:
                logmsg = "TaskMgr.supervise() received {0} tasks".format(
                    len(self.work_todo)
                )
                self.log.info(logmsg)
            for task in self.work_todo:
                self.num_tasks += 1
                if self.log_level >= 2:
                    logmsg = "TaskMgr.supervise() queued task: {0}".format(task)
                    self.log.info(logmsg)
                self.queue_task(task)

        finished = False
        while not finished:
            try:
                if hot_loop:
                    # Calculate the adaptive loop delay
                    delay = self.calc_wait_time(stats.exec_times)
                    self.queue_tasks_from_controller(delay=delay)  # queue tasks
                    time.sleep(delay)

                r_msg = self.r_q.get_nowait()  # __ACK__ or __FINISHED__
                task = r_msg.get("task")
                w_id = r_msg.get("w_id")
                state = r_msg.get("state", "")
                if state == "__ACK__":
                    self.worker_assignments[w_id] = task
                    self.work_todo.remove(task)
                    if self.log_level >= 3:
                        self.log.debug("r_msg: {0}".format(r_msg))
                    if self.log_level >= 3:
                        self.log.debug("w_id={0} received task={1}".format(w_id, task))
                elif state == "__FINISHED__":
                    now = time.time()
                    task_exec_time = task.task_stop - task.task_start
                    task_queue_time = now - task.queue_time - task_exec_time
                    stats.exec_times.append(task_exec_time)
                    stats.queue_times.append(task_queue_time)

                    if self.log_level >= 1:
                        self.log.debug(
                            "TaskMgr.work_todo: {0} tasks left".format(
                                len(self.work_todo)
                            )
                        )
                    if self.log_level >= 3:
                        self.log.debug("TaskMgr.work_todo: {0}".format(self.work_todo))
                        self.log.debug("r_msg: {0}".format(r_msg))

                    if not hot_loop:
                        self.retval.add(task)  # Add result to retval
                        self.worker_assignments.pop(w_id)  # Delete the key
                        finished = self.is_finished()
                    else:
                        self.controller.to_q.put(task)  # Send to the controller
                        self.worker_assignments.pop(w_id)  # Delete the key

                elif state == "__ERROR__":

                    now = time.time()
                    task_exec_time = task.task_stop - task.task_start
                    task_queue_time = now - task.queue_time - task_exec_time
                    stats.exec_times.append(task_exec_time)
                    stats.queue_times.append(task_queue_time)

                    if self.log_level >= 1:
                        self.log.error("r_msg: {0}".format(r_msg))
                        self.log.error("".join(r_msg.get("error")))
                        self.log.debug(
                            "TaskMgr.work_todo: {0} tasks left".format(
                                len(self.work_todo)
                            )
                        )
                    if self.log_level >= 3:
                        self.log.debug("TaskMgr.work_todo: {0}".format(self.work_todo))

                    if not hot_loop:
                        if not self.resubmit_on_error:
                            # If task is in work_todo, delete it
                            for tt in self.work_todo:
                                if tt == task:
                                    self.work_todo.remove(task)  # Remove task...

                            try:
                                # Delete the worker assignment...
                                self.worker_assignments.pop(w_id)
                            except:
                                pass
                            self.retval.add(task)  # Add result to retval

                    self.respawn_dead_workers()

            except Empty:
                state = "__EMPTY__"
            except Exception as e:
                tb_str = "".join(tb.format_exception(*(sys.exc_info())))
                raise e(tb_str)

            if stats.log_time:
                if self.log_level >= 0:
                    self.log.info(stats.log_message)

            # Adaptive loop delay
            delay = self.calc_wait_time(stats.exec_times)
            time.sleep(delay)

            self.respawn_dead_workers()
            finished = self.is_finished()

        if not hot_loop:
            self.kill_workers()
            for w_id, p in self.workers.items():
                p.join()

            ## Log a final stats summary...
            if self.log_level > 0:
                self.log.info(stats.log_message)
            return self.retval
If not in a hot_loop, call supervise() to start the tasks