9

Bom, estou com uma certa dificuldade no uso de threads, preciso realizar o seguinte procedimento:

Tenho vários itens para serem comparados com uma variável, porém essa comparação leva um certo tempo, então gostaria de fazer várias comparações ao mesmo tempo e caso alguma comparação resulte verdadeira, parar todas as outras e seguir pra outra parte do código.

Tentei usar o join, mas acaba atrasando o programa, pois preciso aguardar comparações que se tornaram irrelevantes a partir do momento em que eu encontrei o que precisava. Tentei também sem o join, porém a execução de outras threads acabam atrapalhando o fluxo do meu programa.

Algo mais ou menos assim:

Para cada 10 itens da lista 
Use cada item em uma comparação distinta 
Se alguma das 10 comparações retornar como True, feche todas as outras comparações e vá para outra função 
Senão, pegue mais 10 itens da lista e refaça a comparação

De forma mais genérica fiz da seguinte forma:

palavras = ['palavra1', 'palavra2', 'palavra3', 'palavra4']
nth = 2
threads = []
def execute(palavra):
    print('\ntentando palavra'+palavra)
    time.sleep(5)
    print(palavra+'finalizada')
for nome in palavras:
    threads.append(nome)
    if len(threads)>= nth:
        for item in threads:
            _thread.start_new_thread(execute, (item,))
        threads = []

Entretanto, ele inicia continuamente as threads. Eu precisava manter somente 10, ou seja, ele pega 10 itens, compara e na medida que terminar uma comparação, ele pega mais um item pra ficar sempre em 10 comparações.

4
  • Já pensou em fazer uma espera ocupada dentro do execute? Commented 15/04/2015 às 16:41
  • Você quer que as threads sejam criadas conforme o número de itens da lista? Não pode ser 10 threads para todos os itens?
    – stderr
    Commented 15/04/2015 às 18:08
  • vc diz com um outro sleep @FelipeAvelar? @qmechanik pode der 10threads pra todos, o importante é fazer 10comparações por vez
    – x-x
    Commented 15/04/2015 às 19:40
  • @x-x pode ser um outro sleep, mas no programa principal, enquanto número de threads for igual a 10, espera, depois lance a thread... Commented 15/04/2015 às 22:53

2 Respostas 2

2

A saída próxima do ponto em que você está, no caso, é manter um "registro" de quantas threads você já tem disparadas, e só acrescentar novas threads quando ouver "espaço". O seu programa cria um número máximo de threads, mas não tem nenhum código para acrescentar novas threads (ou passar novos elementos para as threads já existentes) após atingir esse número máximo. A lógica para isso suando 'while', 'if' e uma ou duas variáveis para contar quantas threads estão ativas, e disparar mais caso o número de threads seja menor que o seu limite (no caso 10).

A solução "padrão" para esse tipo de problema é pouca coisa mais elegante, no entanto: ela passa pela criação de um conjunto fixo de threads - com o número de threads desejado - esse conjunto, na literatura é chamado de "threadpool" - na prática é uma coleção - que pode ser uma lista, em que cada elemento é uma das suas threads - que nesse contexto é chamada de "worker" (as threads trabalhadoras)

E nesse caso é usada uma estrutua de dados chamada "fila" - "Queue" que é alimentada por uma thread principal, e de onde cada worker thread puxa elementos. Dessa forma, uma worker thread pode puxar um novo elemento para processar assim que acabar o trabalho anterior, independente do que as demais estejam fazendo.

Em outras palavras: você coloca elementos na fila na thread principal - cada uma das worker threads criada previamente, fica num laço contínuo pegando o próximo elemento da fila e processando-o.

Você precisa de alguma outra forma de passar informação para as worker threads para dizer que o processamento acabou, e elas podem se encerrar. Tipicamente isso é feito colocando um objeto "Marker" na fila, e as threads param ao encontra-lo. No seu cenário no entanto, isso exigiria uma lógica a mais para ir colocando os elementos na fila aos poucos, de forma que o Marker não ficasse no final da fila (e você volta ao seu problema inicial) - então, para cenários mais simples: soluções mais simples: uma varíável global "COMPLETE" é usada,e setada por uma worker thread que encontra o resultado.

Note quem tanto na teoria de threads, como na implementação em linguagens de mais baixo nível, isso seria bem mais complicado: há condições de corrida para a variável global poder ser usada,q ue teriam que ser levadas em conta - no caso do Python, o GIL (Global Interpreter Lock) cuida disso para nós - e, as Queues são classes já existentes, utilizando internamente os locks necessários - de forma que é bem simples usa-las sem maiores preocupações.

(O preço que se paga por isso é justamente que se as threads forem intensivas em CPU num algoritmo em Python puro, o GIL não é liberado durante a execução do algoritmo, e seu ganho usando threads em comparação com um programa linear vai ser bem pequeno. As alternativas seriam: usar "multiprocessing" em vez de "threading" - isso põe cada worker num processo separado, e acaba com o problema do GIL (mas você vai precisar de outro mecanismo que não a variável global para sincronizar os Workers) ) - Ou, escrever a sua função execute em Cython, e usar a chamada disponível nesse super-conjunto do Python para liberar o GIL.

Aqui, o exemplo usando threading e Queue do Python3, com o seu cenário:

from threading import Thread
from queue import Queue
import random
import time

COMPLETE = False
class Worker(Thread):
    def __init__(self, queue, output):
        self.queue = queue
        self.output = output
        super(Worker, self).__init__()

    def run(self):
        while not COMPLETE:
            element = self.queue.get()
            self.execute(element)

    def execute(self, palavra):
        global COMPLETE
        print('\ntentando palavra'+palavra)
        time.sleep(1)
        print(palavra+' finalizada')
        if random.randint(0, 10) == 5:
            COMPLETE = True
            self.output.put(palavra)

def main(palavras, nworkers):
    queue = Queue()
    result_queue = Queue()
    threads = [Worker(queue, result_queue) for _ in range(nworkers)]
    for t in threads:
        t.start()
    for p in palavras:
        queue.put(p)
    result = result_queue.get()
    print ("O resultado final é:", result)


palavras = ['palavra_{}'.format(i) for i in range(50)]
main(palavras, nworkers = 10)

Para saber mais, veja a documentação do Queue: https://docs.python.org/3/library/queue.html (inclusive tem um exemplo parecido com esse lá)

1

Pra isso não seria melhor criar um executor e ir passando pra ele suas tarefas? Por exemplo, a seguir vou pegar uma lista de números até 1000 e retornar o fatorial. Vou chamar math.factorial em cada elemento da lista e colocar em um dicionário, executando somente 10 operações de cada vez.

from math import factorial
from pprint import pprint
from concurrent.futures import ThreadPoolExecutor 

lista = range(1000)

with ThreadPoolExecutor(max_workers=10) as executor:
    pprint({num: fat for num, fat in zip(lista, executor.map(factorial, lista))})

Basicamente o executor utiliza um map() para aplicar a função a cada elemento da lista.

Se quiser usar multiprocessing.Pool ao invés de concurrent.futures.ThreadPoolExecutor também dá certo, e daí você estaria fazendo verdadeiro paralelismo com vários processos ao invés de Threads.

Você deve fazer log-in para responder a esta pergunta.

Esta não é a resposta que você está procurando? Pesquise outras perguntas com a tag .