Multi Threading застрял - есть подозрение на ошибку в условной переменной

Question

Я тестирую программное обеспечение Ant Colony Optimization (ACO), которое работает с несколькими потоками (по 1 для каждого созданного муравья).

Каждая итерация ACO должна ждать завершения всех потоков, прежде чем разрешить запуск следующей итерации. Я делаю это с помощью "условия ()" из модуля потоков.

Поскольку муравьи совместно используют матрицу ферормона, чтение и запись в этой матрице могут быть заблокированы также из модуля потоков.

Теперь описание проблемы:

Я запускаю функцию и что-то печатаю на каждой итерации. Иногда, не всегда, кажется, что выполнение функции прекращается, то есть прекращается печать, то есть итерация никогда не заканчивается.

Я, честно говоря, не знаю, почему это происходит, и я был бы признателен за любой ответ, который мог бы привести меня в нужное русло. Если бы мне пришлось угадывать, я бы сказал, что переменная условия не называется должным образом, или что-то в этом роде. Однако я не уверен, и мне также кажется странным, что это случается иногда.

Ниже приведены соответствующие функции. ACO начинается с вызова функции start(). Это создает N потоков, которые после завершения вызывают update(). Эта функция обновления, после вызова N раз, вызывает notify, что позволяет start() продолжить процесс и, наконец, запустить следующую итерацию. Я также опубликовал метод запуска каждой темы.

Стоит упомянуть, что без действий демона ошибка практически не возникает. С действиями демона это происходит почти всегда (что я также нахожу странным). Наконец, ошибка не всегда возникает в одной и той же итерации.

    def start(self):
        self.ants = self.create_ants()
        self.iter_counter = 0

        while self.iter_counter < self.num_iterations:
            print "START ACQUIRED"
            self.cv.acquire()
            print "calling iteration"
            self.iteration()
            #CV wait until all ants (threads) finish and call update, which
            #calls notify(), and allow continuation
            while not self.iter_done:
                print "iter not complete, W8ING"
                self.cv.wait()
            print "global update "
            self.global_update_with_lock()
            print "START RELEASED"
            self.cv.release()

def update(self, ant):
    lock = Lock()
    lock.acquire()

    print "Update called by %s" % (ant.ID,)
    self.ant_counter += 1

    self.avg_path_cost += ant.path_cost

    # book-keeping
    if ant.path_cost < self.best_path_cost:
        self.best_path_cost = ant.path_cost
        self.best_path_mat = ant.path_mat
        self.best_path_vec = ant.path_vec
        self.last_best_path_iteration = self.iter_counter

    #all threads finished, call notify
    print "ant counter"
    print self.ant_counter
    if self.ant_counter == len(self.ants):
        print "ants finished"
        #THIS MIGHT CAUSE PROBLEMS (no need to notify if its no one waiting)
        self.best_cost_at_iter.append(self.best_path_cost)
        self.avg_path_cost /= len(self.ants)

        self.cv.acquire()
        self.iter_done = True
        self.cv.notify()
        self.cv.release()

    lock.release()

    # overide Thread's run()
    def run(self):
        graph = self.colony.graph
        while not self.end():
            # we need exclusive access to the graph
            graph.lock.acquire()
            new_node = self.state_transition_rule(self.curr_node)
            self.path_cost += graph.delta(self.curr_node, new_node)

            self.path_vec.append(new_node)
            self.path_mat[self.curr_node][new_node] = 1  #adjacency matrix representing path

            #print "Ant %s : %s, %s" % (self.ID, self.path_vec, self.path_cost,)

            self.local_updating_rule(self.curr_node, new_node)
            graph.lock.release()

            self.curr_node = new_node

        # close the tour
        self.path_vec.append(self.path_vec[0])

        #RUN LOCAL HEURISTIC
        if self.daemon == True:
            try:
                daemon_result =  twoOpt(self.path_vec, graph.delta_mat)
                d_path, d_adj = daemon_result['path_vec'], daemon_result['path_matrix']
                self.path_vec = d_path
                self.path_mat = d_adj
            except Exception, e:
                print "exception: " + str(e)
                traceback.print_exc()

        self.path_cost += graph.delta(self.path_vec[-2], self.path_vec[-1])
        # send our results to the colony
        self.colony.update(self)
        #print "Ant thread %s terminating." % (self.ID,)
        
        # allows thread to be restarted (calls Thread.__init__)
        self.__init__(self.ID, self.start_node, self.colony, self.daemon, self.Beta, self.Q0, self.Rho)

Решение проблемы: во- первых, я исправил ошибку в ожидании условных переменных в соответствии с комментариями здесь. Во-вторых, он иногда зависал, и это было связано с ошибкой в ​​обновлении счетчика потоков. Решением было изменить счетчик с int на массив с длиной num_threads, полной 0, где каждый поток обновляет свою позицию в списке. Когда все потоки заканчиваются, массив счетчиков должен быть все 1. В настоящее время это работает просто отлично.